Noukai Client

Noukai (Sync Client)

Main synchronous client for executing flows.

Constructor

class Noukai:
    def __init__(
        self,
        *,
        api_key: str | None = None,
        org: str | None = None,
        project: str | None = None,
        env: str | None = None,
        timeout: int = 300_000,
        max_retries: int = 1,
        onLog: Callable[[LogEvent], None] | None = None,
        logPayloads: bool = False,
        signal: AbortSignal | None = None
    ) -> None: ...

Parameters:

• api_key (str, optional): API key starting with nk_. Falls back to NOUKAI_API_KEY env var.
• org (str, optional): Default organization name for flow lookups. Must be paired with project. If set, client.flow("slug") uses the defaults; if omitted, client.flow("org/project/slug") is required.
• project (str, optional): Default project name for flow lookups. Must be paired with org. If set, client.flow("slug") uses the defaults; if omitted, client.flow("org/project/slug") is required.
• env (str, optional): Environment for base URL resolution. Accepts "dev"/"development" (→ http://localhost:8080/api/v1) or "production" (default → https://api.noukai.xyz/api/v1). Falls back to NOUKAI_ENV env var.
• timeout (int, optional): Request timeout in milliseconds. Default: 300000 (5 minutes).
• max_retries (int, optional): Number of retries on 5xx errors. Default: 1.
• onLog (callable, optional): Logging hook that receives LogEvent objects.
• logPayloads (bool, optional): If True, include request/response bodies in log events. Default: False.

Raises:

• AuthenticationError: If no API key is provided or the prefix is invalid.
• ValueError: If only one of org or project is provided (both required together).

Example:

from noukai_sdk import Noukai
 
# Production (default) with org/project defaults
with Noukai(org="acme", project="spelling") as client:
    flow = client.flow("grade-3")
 
# Using explicit API key
with Noukai(api_key="nk_...", org="acme", project="spelling") as client:
    flow = client.flow("grade-3")
 
# Local development — connects to http://localhost:8080/api/v1
with Noukai(env="dev", org="acme", project="spelling") as client:
    flow = client.flow("grade-3")
 
# Without defaults, use fully-qualified slug
with Noukai() as client:
    flow = client.flow("acme/spelling/grade-3")

Methods

flow(identifier) -> Flow

Get a flow proxy for executing a specific flow.

Parameters:

• identifier (str | dict): Flow identifier in one of three forms:
  • Single-segment slug (e.g., "grade-3") — only valid if the client was constructed with org and project defaults
  • Three-segment path (e.g., "org/project/slug") — fully-qualified, works regardless of defaults
  • Dictionary (e.g., {"org": "acme", "project": "spelling", "slug": "grade-3"}) — kwargs form, overrides defaults if provided

Returns: Flow proxy object.

Raises:

• ValueError: If a single-segment slug is given without client-level defaults, or if the identifier format is invalid.

Example:

# With defaults set, use single-segment slug (recommended)
client = Noukai(org="acme", project="spelling")
flow = client.flow("grade-3")
 
# Override defaults with fully-qualified path
flow = client.flow("acme/spelling/grade-3")
 
# Override defaults with kwargs
flow = client.flow(org="acme", project="spelling", slug="grade-3")
 
# Without defaults, use fully-qualified
client = Noukai()
flow = client.flow("acme/spelling/grade-3")

close() -> None

Close the client and clean up resources. Called automatically with context manager.

Example:

client = Noukai()
try:
    result = client.flow("...").execute(message="...")
finally:
    client.close()

Properties

default_org (read-only)

Returns the organization name passed to the constructor, or None if not set.

client = Noukai(org="acme", project="spelling")
print(client.default_org)  # "acme"

default_project (read-only)

Returns the project name passed to the constructor, or None if not set.

client = Noukai(org="acme", project="spelling")
print(client.default_project)  # "spelling"

Context Manager

Use with to automatically close the client:

with Noukai(org="acme", project="spelling") as client:
    result = client.flow("grade-3").execute(message="hello")

AsyncNoukai (Async Client)

Asynchronous client for non-blocking flow execution. Same API as Noukai but with async/await.

Constructor

Same as Noukai, but used with async with:

async with AsyncNoukai() as client:
    result = await client.flow("acme/spelling/grade-3").execute(message="hello")

Methods

async flow(identifier) -> AsyncFlow

Same as sync Flow.flow() but returns AsyncFlow.

async aclose() -> None

Async equivalent of close().

Properties

Same as Noukai:

• default_org (read-only): Organization name from constructor, or None
• default_project (read-only): Project name from constructor, or None

Context Manager

async with AsyncNoukai(org="acme", project="spelling") as client:
    result = await client.flow("grade-3").execute(message="...")
# Resources cleaned up automatically

LogEvent (Type)

Structured logging event passed to onLog hook.

Attributes:

• phase (str): One of "request", "response", "retry"
• method (str): HTTP method (GET, POST, etc.)
• path (str): Request path
• attempt (int): Attempt number (1 for first try, 2+ for retries)
• statusCode (int, optional): HTTP status code (only on response)
• requestId (str, optional): Request ID from response headers
• requestBody (any, optional): Request body (if logPayloads=True)
• responseBody (any, optional): Response body (if logPayloads=True)

Example:

def my_logger(event):
    print(f"[{event.phase}] {event.method} {event.path} (attempt {event.attempt})")
    if event.statusCode:
        print(f"  Status: {event.statusCode}")
 
client = Noukai(onLog=my_logger)

FlowIdentifier (Type)

Flow identifier can be a string or dict:

FlowIdentifier = str | { org: str, project: str, slug: str }

String format: "org/project/slug" Dict format: {"org": "acme", "project": "spelling", "slug": "grade-3"}

Both refer to the same flow.