Python SDK

Installation

pip install iii-sdk

Initialization

register_worker

Register the worker with a iii instance, returns a connected worker client. Blocks up to 30 seconds for the WebSocket connection to be established. If the engine is not reachable in time, a warning is logged and the client is returned anyway; it keeps retrying in the background and flushes registrations once connected. Use add_connection_state_listener to observe the actual transition. Signature

register_worker(address: str | None = None, options: InitOptions | None = None) -> III

Parameters
Example

str | None

WebSocket URL of the III engine (e.g. ws://localhost:49134). When omitted, resolves from III_URL and then DEFAULT_ENGINE_URL.

InitOptions | None

Optional configuration for worker name, timeouts, reconnection, and OTel.

Show `InitOptions` fields

bool

Enable worker metrics via OpenTelemetry. Default True.

dict[str, str] | None

Custom HTTP headers sent during the WebSocket handshake, on the main connection and on streaming channels. Authorization: Bearer $III_TOKEN is added unless you set Authorization here.

int

Default timeout for worker.call() invocations in milliseconds. Default 30000.

str | None

Namespace this worker belongs to. Falls back to the III_NAMESPACE env var; when neither is set the engine applies default. It scopes more than the registration: the worker and its functions register here, and everything the worker does afterwards follows it: call resolves its target here and observe resolves its source and target here, unless the call names another namespace.

OtelConfig | dict[str, Any] | None

OpenTelemetry configuration. Enabled by default. Set {'enabled': False} or env OTEL_ENABLED=false to disable.

ReconnectionConfig | None

WebSocket reconnection behavior.

TelemetryOptions | None

Internal worker metadata reported to the engine.

str | None

One-line, human/LLM-readable summary of what this worker does. Surfaces in engine::workers::list / engine::workers::info.

str | None

Display name for this worker. A non-empty III_WORKER_NAME overrides it. Defaults to hostname:pid.

from iii import register_worker, InitOptions
worker = register_worker()  # address from III_URL
other = register_worker('ws://localhost:49134', InitOptions(worker_name='my-worker'))

Methods

register_function

Register a function with the engine. Pass a handler for local execution, or an HttpInvocationConfig for HTTP-invoked functions (Lambda, Cloudflare Workers, etc.). Handlers can be synchronous or asynchronous. Sync handlers are automatically wrapped with run_in_executor so they do not block the event loop. Each handler receives a data argument containing the call’s payload, and may optionally accept a second metadata argument carrying per-invocation metadata (e.g. def handler(data, metadata=None) or def handler(data, *, metadata=None)). Metadata is only forwarded to handlers that declare a parameter literally named metadata, so existing handlers, including ones with unrelated extra parameters, *args, or **kwargs, keep working unchanged. The handler’s return value is what the function’s observers observe. An event source’s handler returns the event it is called with. What this worker does when the function is observed is set with the hook methods (on_observer_added, on_observer_removed, on_observation_match, on_observation_deliver) on the client by id or on the returned FunctionRef. Set on_observer_added and on_observer_removed by id before registering so they run for every observer: the engine may announce an observer before a hook set on the FunctionRef is in place. request_format and response_format are auto-extracted from the handler’s type hints when omitted or passed as None (the default). To opt out of auto-extraction, pass an explicit schema (RegisterFunctionFormat or dict). This behavior is Python-specific; the Node SDK relies on explicit schemas because TypeScript types are erased at runtime. Signature

register_function(function_id: str, handler_or_invocation: RemoteFunctionHandler | HttpInvocationConfig, *, description: str | None = None, metadata: dict[str, Any] | None = None, request_format: RegisterFunctionFormat | dict[str, Any] | None = None, response_format: RegisterFunctionFormat | dict[str, Any] | None = None, args_format: Any = None, private: bool = False, observable: bool = True, local_dispatch: bool = True) -> FunctionRef

Parameters
Example

str

required

Unique string identifier for the function.

RemoteFunctionHandler | HttpInvocationConfig

required

A callable handler or HttpInvocationConfig. Callable handlers receive data (the call’s payload) as the first argument and may optionally accept metadata (per-invocation metadata) as a second argument; they may return a value.

str | None

Human-readable description of what the function does.

dict[str, Any] | None

Arbitrary metadata attached to the function.

RegisterFunctionFormat | dict[str, Any] | None

Schema describing expected input. When None (default), auto-extracted from the handler’s first-parameter type hint. Pass an explicit schema to override; there is no way to register with no schema when the handler is typed.

Show `RegisterFunctionFormat` fields

list[RegisterFunctionFormat] | None

Nested fields for object types.

str | None

Human-readable description of the parameter.

RegisterFunctionFormat | None

Item schema for array types.

str

required

Parameter name.

bool

Whether the parameter is required.

str

required

Type string (string, number, boolean, object, array, null, map).

RegisterFunctionFormat | dict[str, Any] | None

Schema describing the handler’s return value, what observers observe. Same auto-extraction semantics as request_format.

Any

JSON Schema (or a Pydantic model) of the observers’ args; the engine validates them (INVALID_ARGS). A function without it takes no args.

bool

Only this worker may call the function; others may observe it. This worker’s own calls pass the input and the result by reference: the handler and hooks treat both as read-only.

bool

False: nobody may observe the function (NOT_OBSERVABLE), for functions whose payloads carry credentials.

bool

Calls this worker makes to its own function run in-process once the engine confirmed it (default True). Like a remote invocation, each local call to a sync handler runs on a thread of its own (one per call, not pooled), so the caller can stop waiting at its timeout. A sync handler that blocks for long should be async, or set False so its calls go through the engine (also for a handler that relies on engine-side effects).

def greet(data):
    return {'message': f"Hello, {data['name']}!"}
fn = worker.register_function("greet", greet, description="Greets a user")
fn.unregister()
from pydantic import BaseModel
class GreetInput(BaseModel):
    name: str
class GreetOutput(BaseModel):
    message: str
async def greet(data: GreetInput) -> GreetOutput:
    return GreetOutput(message=f"Hello, {data.name}!")
fn = worker.register_function("greet", greet, description="Greets a user")
on_tick = worker.register_function(
    "cron::on-tick", lambda tick: tick, args_format=CronArgs, private=True
)

call

Call a function and return its result. The routing behavior and return type depend on the action field:

No action: synchronous, waits for the function to return.
CallAction.Enqueue(...): async via named queue, returns a dict with messageReceiptId.
CallAction.Void(): fire-and-forget, returns None.

A function this worker registered with a handler, once the engine confirmed it, runs in-process (local dispatch): a sync handler on a thread of its own (the caller waits up to the timeout), an async one on the SDK loop. Its observers are still delivered to, and only matched calls reach the engine. dispatch="remote" (or action=Enqueue) always goes through the engine. Whatever a local handler raises, BaseExceptions such as KeyboardInterrupt and SystemExit included, reaches the caller as InvocationError("invocation_failed") with the original as its __cause__ (for a Void call nothing reaches the caller: the error is logged). metadata is any JSON value; the handler receives it beside the payload, and None is no metadata. A request with observer (an Observer of a function this worker hosts) calls through it: the engine runs the observer’s target with its policy (max_fires, expiry, max_in_flight), metadata sidecar and loop guard, and the call resolves to the target’s result. It takes no metadata: the target’s metadata argument is the observer’s registration metadata, and a request that names metadata raises ValueError before anything is sent. The payload goes as given: filters are applied where the worker chooses observers: iii’s delivery, or obs.matches(call) in an on_observation_deliver. A host also calls through an observer to send it something of its own, such as its initial state once it is accepted (see Observer.accepted). Signature

call(request: dict[str, Any] | CallRequest) -> Any

Parameters
Example

dict[str, Any] | CallRequest

required

A CallRequest or dict with function_id (or observer), payload, and optional action / timeout_ms / metadata / namespace / dispatch ("auto", the default, or "remote").

Hide `CallRequest` fields

CallActionEnqueue | CallActionVoid | None

Sets how the call is routed. Omit for a synchronous request/response. Specify for a specific routing scheme (e.g. CallAction.Enqueue(...), CallAction.Void()).

Literal['auto', 'remote'] | None

auto (default) runs a function this worker hosts in-process once the engine confirmed it; remote always goes through the engine.

str | None

ID of the function to call.

Any | None

Per-invocation metadata (any JSON value) supplied to the called handler beside the payload. None is no metadata.

str | None

Target namespace for routing. Omit to inherit this worker’s; say default to reach the engine’s from a namespaced worker.

Any

An Observer of a function this worker hosts: the call runs its target, with its policy and metadata sidecar. Takes no function_id, namespace, metadata or enqueue action.

Any

Input data passed to the function.

int | None

Override the default invocation timeout, in milliseconds.

result = worker.call({'function_id': 'greet', 'payload': {'name': 'World'}})
worker.call({'function_id': 'notify', 'payload': {}, 'action': CallAction.Void()})
worker.call({'observer': obs, 'payload': request})  # in a deliver hook

observe

Observes a function: when source runs, target runs. Every call of the source function that passes source["filters"] invokes target, a registered function, with the Observed payload: the top-level fields plus what source["capture"] names that applies to the outcome (input, output or error, caller, metadata). capture also decides which outcomes fire the target. The worker that hosts the source can decide otherwise in its on_observation_deliver. Returns synchronously, also before the connection is up; the observer is replayed with the same id on every reconnect. The handle holds its id until its dispose() is called or the engine disposes it; the id is free again at once after that. Errors (INVALID_ARGS, INVALID_FILTER, FORBIDDEN, NOT_OBSERVABLE, the rejection of the source’s worker, UNSUPPORTED against an engine without observe) show up as state error and make ready() raise ObserveRejection. An error answer to the handle’s observe or to its replay after a reconnect is the engine’s refusal and the handle’s final answer: the engine holds no observer for it, it is not replayed, it holds its id until its dispose(), and that dispose() sends nothing. An observer that moves to error after its answer stays held: it is replayed, and its dispose() sends an unobserve. Signature

observe(source: str | ObserveSource | dict[str, Any], target: Target, options: ObserverOptions | dict[str, Any] | None = None) -> ObserverHandle

Parameters
Example

str | ObserveSource | dict[str, Any]

required

What is observed: a function id, or function_id, namespace, args (read by the worker that hosts the function, validated against its args_format), filters, and capture.

Show `ObserveSource` fields

str

required

The observed function.

dict[str, Any]

For the worker that hosts the function, which reads them in its hooks; validated against the function’s args_format. A function without args_format takes none.

list[Capture]

What the target’s payload carries, and with it which outcomes fire the target. Default ["input", "output", "caller"]. input and output fire on success, error fires on error; caller and metadata only add to the payload, so a capture must name at least one of input, output, error (INVALID_ARGS otherwise). [] fires on success with the top-level fields only. The default fires on success, ["error", "caller"] on errors, ["input", "error"] on both.

Filter

Which calls reach the target; {} (or absent) = every call.

str

Defaults to this worker’s (engine::* and configuration::* resolve to default).

Target

required

The function to invoke: its id, or {"function_id": ..., "namespace": ...}. Register it first; until then the observer stays pending.

ObserverOptions | dict[str, Any] | None

id, max_fires, expires_in_ms, max_in_flight, metadata (the activation’s metadata sidecar).

Show `ObserverOptions` fields

int

Disposed after this long.

str

Default: a random UUID, stable across replays.

int

Disposed after this many activations.

int

Activations running at once (the engine queues the rest).

dict[str, Any]

Delivered as the metadata sidecar of every activation.

iii.observe("orders::create", "alerts::big-order")
iii.register_function("alerts::payment-failed", notify)
failures = iii.observe(
    {"function_id": "payments::charge", "capture": ["error", "caller"],
     "filters": {"input.amount": {"$gte": 500}}},
    "alerts::payment-failed",
    {"max_in_flight": 4},
)
failures.ready()            # or: await failures.ready_async()
failures.dispose()

trigger

Deprecated since 0.26.0. Use call.

Call a function: call under its 0.25 name. Signature

trigger(request: dict[str, Any] | CallRequest) -> Any

Parameters

dict[str, Any] | CallRequest

required

Show `CallRequest` fields

CallActionEnqueue | CallActionVoid | None

Sets how the call is routed. Omit for a synchronous request/response. Specify for a specific routing scheme (e.g. CallAction.Enqueue(...), CallAction.Void()).

Literal['auto', 'remote'] | None

auto (default) runs a function this worker hosts in-process once the engine confirmed it; remote always goes through the engine.

str | None

ID of the function to call.

Any | None

Per-invocation metadata (any JSON value) supplied to the called handler beside the payload. None is no metadata.

str | None

Target namespace for routing. Omit to inherit this worker’s; say default to reach the engine’s from a namespaced worker.

Any

An Observer of a function this worker hosts: the call runs its target, with its policy and metadata sidecar. Takes no function_id, namespace, metadata or enqueue action.

Any

Input data passed to the function.

int | None

Override the default invocation timeout, in milliseconds.

add_connection_state_listener

Subscribe to connection-state transitions. The handler is fired immediately with the current state (on the caller’s thread), then once per transition. Transitions fire on the SDK’s background event-loop thread: keep handlers fast and do not call sync SDK methods from them (they would raise RuntimeError). Treat calls as state notifications, not edges. A state may rarely be observed twice around subscription. Registering the same handler twice fires it twice. Returns an idempotent unsubscribe function that removes only its own registration. Signature

add_connection_state_listener(handler: ConnectionStateCallback) -> Callable[[], None]

Parameters
Example

ConnectionStateCallback

required

unsubscribe = worker.add_connection_state_listener(
    lambda state: print(f"engine link: {state}")
)

call_async

Call a function and return its result (on any event loop). The routing behavior and return type depend on the action field:

No action: synchronous, waits for the function to return.
CallAction.Enqueue(...): async via named queue, returns a dict with messageReceiptId.
CallAction.Void(): fire-and-forget, returns None.

A function this worker registered with a handler, once the engine confirmed it, runs in-process on the calling loop (a sync handler on a thread of its own); dispatch="remote" (or action=Enqueue) always goes through the engine. A request with observer calls through it, as for call. Signature

async (request: dict[str, Any] | CallRequest) -> Any

Parameters
Example

dict[str, Any] | CallRequest

required

A CallRequest or dict with function_id (or observer), payload, and optional action / timeout_ms / metadata / namespace / dispatch ("auto" or "remote").

Show `CallRequest` fields

CallActionEnqueue | CallActionVoid | None

Sets how the call is routed. Omit for a synchronous request/response. Specify for a specific routing scheme (e.g. CallAction.Enqueue(...), CallAction.Void()).

Literal['auto', 'remote'] | None

auto (default) runs a function this worker hosts in-process once the engine confirmed it; remote always goes through the engine.

str | None

ID of the function to call.

Any | None

Per-invocation metadata (any JSON value) supplied to the called handler beside the payload. None is no metadata.

str | None

Target namespace for routing. Omit to inherit this worker’s; say default to reach the engine’s from a namespaced worker.

Any

An Observer of a function this worker hosts: the call runs its target, with its policy and metadata sidecar. Takes no function_id, namespace, metadata or enqueue action.

Any

Input data passed to the function.

int | None

Override the default invocation timeout, in milliseconds.

result = await worker.call_async({'function_id': 'greet', 'payload': {'name': 'World'}})
await worker.call_async(
    {'observer': obs, 'payload': state, 'action': CallAction.Void()}
)

connect_async

Connect to the III Engine via WebSocket. Initializes OpenTelemetry (if configured), attaches the event loop, and establishes the WebSocket connection. This is called automatically during construction; use it only if you need to reconnect manually from an async context. Signature

async () -> None

current_invocation

The invocation the calling code runs in (in every handler, remote or local), or None outside any handler. Signature

current_invocation() -> InvocationContext | None

Example

def create_order(order):
    inv = iii.current_invocation()  # function_id, invocation_id, caller, observe_depth
    return {"id": inv.invocation_id if inv else None}

get_address

Return the engine address this worker resolved to. The explicit register_worker argument, else III_URL, else DEFAULT_ENGINE_URL. Mirrors the Rust SDK’s address() and the Node SDK’s getAddress(). Signature

get_address() -> str

get_connection_state

Return the current WebSocket connection state. Signature

get_connection_state() -> IIIConnectionState

Example

worker = register_worker("ws://localhost:49134")
if worker.get_connection_state() != "connected":
    print("engine not reachable yet")

is_observed

Whether a call of a function this worker hosts would reach anyone, so a source can skip building an event nobody wants. Without call: whether the function has any active observer (one load). With call (a Call or a partial one, {"input": ...}) on a function matched by keys (on_observation_match): whether an observer is filed under one of the call’s keys. On any other function the call is ignored. Signature

is_observed(function_id: str, call: CallLike | None = None) -> bool

Parameters
Example

str

required

CallLike | None

if iii.is_observed("database::on-row-changed", {"input": change}):
    iii.call({"function_id": "database::on-row-changed", "payload": change})

observers

The active observers of a function this worker hosts, as it sees them. Signature

observers(function_id: str) -> list[Observer]

Parameters
Example

str

required

for obs in iii.observers("cron::on-tick"):
    print(obs.id, obs.args)

on_observation_deliver

Delivers each call of one function itself: hook(call, observers) receives the Call and the observers that matched it (on_observation_match, or all of them), as a snapshot taken when the call started, and what it returns (or raises) is what the caller gets back. iii delivers nothing for that function: the hook calls through observers (iii.call({"observer": obs, ...})) and applies their capture and filters with obs.matches(call). Works as a decorator (@iii.on_observation_deliver("fn::id")) and returns the hook; None restores iii’s delivery. The hook runs inside the function’s invocation, so the calls it makes have that invocation as their caller. A sync hook runs on a thread of its own. A function with a deliver hook needs an engine with observe.v2. Signature

on_observation_deliver(function_id: str, hook: Any = ...) -> Any

Parameters
Example

str

required

Any

@iii.on_observation_deliver("http::on-request")
async def route(call, observers):
    obs = routes.match(call.input)
    if obs is None:
        return {"status_code": 404}
    if not obs.matches(call):
        return {"status_code": 422}
    return await iii.call_async({"observer": obs, "payload": call.input})

on_observation_match

Sets which observers want each call of one function. By default every observer wants every call; iii’s delivery then applies their capture and filters, and an on_observation_deliver receives only the observers that matched. As a predicate, hook(obs, call) -> bool is asked once per observation: for every call, once for each active observer, before its filters; a predicate that raises counts as no match for that observer. As keys, observer= and call= key functions each return a string or strings, and an observer wants a call when they share a key: the SDK computes an observer’s keys once, when it is accepted, and files it under each, so a call costs one lookup per key whatever the number of observers. An observer whose key function raises is rejected (OBSERVE_REJECTED); a call whose key function raises matches no observer. A match hook that raises is logged once per function. Wildcards are keys too: file an observer of every table under "primary|*|insert" and give a change the keys "primary|orders|insert", "primary|orders|*", "primary|*|insert" and "primary|*|*". Works as a decorator for a predicate (@iii.on_observation_match("fn::id")) and returns the hook (a MatchKeys for the keys form); None removes it, and every observer gets every call again. A match hook set while the function has active observers is applied to them on the SDK loop, which rebuilds the index from them. Predicates and key functions are sync: they run on every call. A function with a match hook needs an engine with observe.v2: against an older one it stays unregistered, and UNSUPPORTED is logged once. Signature

on_observation_match(function_id: str, hook: Any = ..., *, observer: Any = None, call: Any = None) -> Any

Parameters
Example

str

required

Any

iii.on_observation_match(
    "cron::on-tick", lambda obs, call: obs.args["expression"] == call.input["expression"]
)
iii.on_observation_match(
    "database::on-row-changed",
    observer=lambda obs: f"{obs.args['table']}|{obs.args['op']}",
    call=lambda call: f"{call.input['table']}|{call.input['op']}",
)

on_observer_added

Sets what runs when an observer is added, before it is accepted: for every function this worker hosts (on_observer_added(hook)), or for one (on_observer_added("fn::id", hook)). Works as a decorator, @iii.on_observer_added or @iii.on_observer_added("fn::id"), and returns the hook; None removes it. The worker’s hook runs, then the function’s; the observer is accepted (acked, then active) once both resolved. A hook that raises rejects the observer with the ObserveRejection it raised, or OBSERVE_REJECTED; one still running after 30 s rejects it with OWNER_TIMEOUT. Then the on_observer_removed of each level whose hook already resolved runs. Hooks may be sync (on a thread of their own) or async, and the hooks of one function run one at a time. Hooks are stored by function id, so they can be set before the function is registered (which makes them run for every observer) and stay in place across unregister(). A hook set while the function already has active observers runs for each of them at once (a raise rejects that observer) and logs one warning. A second after the connection is registered, and after the last hook set by id, a warning names each hook id this worker has not registered. Signature

on_observer_added(function_id_or_hook: Any, hook: Any = ...) -> Any

Parameters
Example

Any

required

Any

@iii.on_observer_added            # every function this worker hosts
async def retain(obs):
    await events.retain()
@iii.on_observer_added("cron::on-tick")
def start(obs):
    scheduler.retain(obs.args["expression"])

on_observer_removed

Sets what runs when an accepted observer goes (on unobserve, on disposal, after a late rejection): for every function this worker hosts, or for one. The same forms as on_observer_added. It runs exactly once for every accepted observer, the function’s hook and then the worker’s, and never for an observer that was not accepted. An error it raises is logged; the observer is removed either way. Signature

on_observer_removed(function_id_or_hook: Any, hook: Any = ...) -> Any

Parameters
Example

Any

required

Any

iii.on_observer_removed(lambda obs: events.release())
@iii.on_observer_removed("cron::on-tick")
def stop(obs):
    scheduler.release(obs.args["expression"])

register_trigger

Removed in 0.26.0: raises an error that points to observe. Signature

register_trigger(_args: Any = (), _kwargs: Any = {}) -> NoReturn

Parameters

Any

register_trigger_type

Removed in 0.26.0: raises an error that points to register_function and the observer hooks. Signature

register_trigger_type(_args: Any = (), _kwargs: Any = {}) -> NoReturn

Parameters

Any

shutdown

Gracefully shut down the client, releasing all resources. Cancels any pending reconnection attempts, rejects all in-flight invocations with an error, closes the WebSocket connection, and stops the background event-loop thread. After this call the instance must not be reused. Signature

shutdown() -> None

Example

worker = register_worker('ws://localhost:49134')
# ... do work ...
worker.shutdown()

shutdown_async

async () -> None

Example

worker = register_worker('ws://localhost:49134')
# ... do work ...
await worker.shutdown_async()

trigger_async

Deprecated since 0.26.0. Use call_async.

Call a function: call_async under its 0.25 name. Signature

async (request: dict[str, Any] | CallRequest) -> Any

Parameters

dict[str, Any] | CallRequest

required

Show `CallRequest` fields

CallActionEnqueue | CallActionVoid | None

Sets how the call is routed. Omit for a synchronous request/response. Specify for a specific routing scheme (e.g. CallAction.Enqueue(...), CallAction.Void()).

Literal['auto', 'remote'] | None

auto (default) runs a function this worker hosts in-process once the engine confirmed it; remote always goes through the engine.

str | None

ID of the function to call.

Any | None

Per-invocation metadata (any JSON value) supplied to the called handler beside the payload. None is no metadata.

str | None

Target namespace for routing. Omit to inherit this worker’s; say default to reach the engine’s from a namespaced worker.

Any

An Observer of a function this worker hosts: the call runs its target, with its policy and metadata sidecar. Takes no function_id, namespace, metadata or enqueue action.

Any

Input data passed to the function.

int | None

Override the default invocation timeout, in milliseconds.

unregister_trigger_type

Removed in 0.26.0: raises an error that points to FunctionRef.unregister. Signature

unregister_trigger_type(_args: Any = (), _kwargs: Any = {}) -> NoReturn

Parameters

Any

Types

iii

Call · CallAction · CallActionEnqueue · CallActionVoid · Caller · EnqueueResult · InitOptions · InvocationContext · MatchKeys · MiddlewareFunctionInput · Observed · Observer · ObserverHandle · ObserverOptions · ObserveSource · StreamRequest · StreamResponse · TelemetryOptions

Call

One call of a function, as its observers observe it and its on_observation_match and on_observation_deliver hooks receive it.

Name	Type	Required	Description
`caller`	`dict[str, Any] \| None`	No	Who made the call: `\{worker_id?, worker_name?, function_id?, invocation_id?, parent?\}`; `None` outside any handler.
`error`	`dict[str, Any] \| None`	No	What the handler raised, `\{code, message, stacktrace?\}` (`None` when it returned).
`function_id`	`str`	Yes	The called function.
`input`	`InputT`	Yes	What the function was called with, as the caller passed it (channel refs as refs, not readers and writers).
`invocation_id`	`str`	Yes	This call’s id.
`metadata`	`Any`	No	The call’s metadata sidecar.
`output`	`OutputT \| None`	No	The handler’s return value: what observers observe (`None` when it raised).
`status`	`Literal['success', 'error']`	Yes	`"success"` or `"error"`.

CallAction

Factory for creating call actions used with call().

Name	Type	Required	Description
`Enqueue`	(*, queue: str) -> `CallActionEnqueue`	Yes	Route the invocation through a named queue for async processing.
`Void`	() -> `CallActionVoid`	Yes	Fire-and-forget routing. No response is returned.

CallActionEnqueue

Routes the call through a named queue for async processing. Requires the queue worker in worker-compose.yaml and a matching entry under that worker’s queue_configs. Without it the call rejects with enqueue_error (no queue provider).

Name	Type	Required	Description
`queue`	`str`	Yes	Name of the target queue.
`type`	`Literal['enqueue']`	No	Always `'enqueue'`.

CallActionVoid

Fire-and-forget routing. No response is returned.

Name	Type	Required	Description
`type`	`Literal['void']`	No	Always `'void'`.

Caller

Who started an invocation: the worker (worker_id / worker_name, stamped by the engine), and the function and invocation the call was made from (absent for top-level calls). parent is that handler’s own caller, one level only.

Name	Type	Required	Description
`function_id`	`str`	No	-
`invocation_id`	`str`	No	-
`worker_id`	`str`	No	-
`worker_name`	`str`	No	-
`parent`	`CallerParent`	No	-

EnqueueResult

Result returned when a function is called with CallAction.Enqueue.

Name	Type	Required	Description
`messageReceiptId`	`str`	Yes	Unique receipt ID for the enqueued message.

InitOptions

Configuration options passed to register_worker.

Name	Type	Required	Description
`enable_metrics_reporting`	`bool`	No	Enable worker metrics via OpenTelemetry. Default `True`.
`headers`	`dict[str, str] \| None`	No	Custom HTTP headers sent during the WebSocket handshake, on the main connection and on streaming channels. `Authorization: Bearer $III_TOKEN` is added unless you set `Authorization` here.
`invocation_timeout_ms`	`int`	No	Default timeout for `worker.call()` invocations in milliseconds. Default `30000`.
`namespace`	`str \| None`	No	Namespace this worker belongs to. Falls back to the `III_NAMESPACE` env var; when neither is set the engine applies `default`. It scopes more than the registration: the worker and its functions register here, and everything the worker does afterwards follows it: `call` resolves its target here and `observe` resolves its source and target here, unless the call names another namespace.
`otel`	`OtelConfig \| dict[str, Any] \| None`	No	OpenTelemetry configuration. Enabled by default. Set `\{'enabled': False\}` or env `OTEL_ENABLED=false` to disable.
`reconnection_config`	`ReconnectionConfig \| None`	No	WebSocket reconnection behavior.
`telemetry`	`TelemetryOptions` \| None	No	Internal worker metadata reported to the engine.
`worker_description`	`str \| None`	No	One-line, human/LLM-readable summary of what this worker does. Surfaces in `engine::workers::list` / `engine::workers::info`.
`worker_name`	`str \| None`	No	Display name for this worker. A non-empty `III_WORKER_NAME` overrides it. Defaults to `hostname:pid`.

InvocationContext

The invocation the current code runs in, in every handler, remote or local. Returned by iii.current_invocation().

Name	Type	Required	Description
`activation`	`dict[str, Any] \| None`	No	`\{observer_id, activation_id?\}` when this invocation is the activation of an observer’s target.
`caller`	`dict[str, Any] \| None`	No	Who started it: `\{worker_id, worker_name, function_id?, invocation_id?, parent?\}`; `None` when unknown.
`function_id`	`str`	Yes	The function whose handler is running.
`invocation_id`	`str`	Yes	This invocation’s id.
`observe_depth`	`int`	No	Depth of the causal chain in observe hops (0 outside any activation).

MatchKeys

on_observation_match as keys: an observer wants a call when they share a key, and the SDK keeps an index instead of asking every observer.

Name	Type	Required	Description
`call`	Callable[[`Call`[Any, Any]], Key]	Yes	What a call is, computed once per call; a raise matches no observer.
`observer`	Callable[[`Observer`], Key]	Yes	What an observer listens for, computed once when it is accepted; an observer whose keys raise is rejected (`OBSERVE_REJECTED`).

MiddlewareFunctionInput

Input passed to the RBAC middleware function on every function invocation through the RBAC port.

Name	Type	Required	Description
`action`	`CallActionEnqueue` \| `CallActionVoid` \| None	No	Routing action, if any.
`context`	`dict[str, Any]`	Yes	Auth context returned by the auth function for this session.
`function_id`	`str`	Yes	ID of the function being invoked.
`namespace`	`str \| None`	No	Target namespace the invoke addressed; forward the call here to stay in the caller’s namespace. Absent -> the engine’s default namespace.
`payload`	`dict[str, Any]`	Yes	Payload sent by the caller.

Observed

What an observer’s target receives when iii delivers: the top-level fields, plus whichever of the observer’s capture apply to the outcome. Fields not captured, or that do not apply, are absent.

Name	Type	Required	Description
`activation_id`	`str`	No	-
`caller`	`Caller`	No	-
`duration_ms`	`float`	No	-
`error`	`dict[str, Any]`	No	-
`function_id`	`str`	No	-
`input`	`Any`	No	-
`invocation_id`	`str`	No	-
`metadata`	`dict[str, Any]`	No	-
`namespace`	`str`	No	-
`observer_id`	`str`	No	-
`output`	`Any`	No	-
`started_at_ms`	`int`	No	-
`status`	`Literal['success', 'error']`	No	-

Observer

An observer as the worker that hosts its function sees it (on_observer_added, observers(), a deliver hook’s observers): what was asked for. Never who observes, the observer’s metadata or its policy. The object stays the same for the observer’s whole lifetime, so a worker keeps it in its own index (route table, consumers, …).

Name	Type	Required	Description
`accepted`	`(timeout: float \| None = None) -> None`	Yes	Blocks until the SDK has acked this observer: a call made afterwards reaches an active observer. Use it to send a new observer its initial state, outside its `on_observer_added` (which runs before the ack): a call through it (`\{"observer": obs, "payload": ...\}`), which runs its target alone with the payload as given. Use `accepted_async` in async code.
`accepted_async`	`async () -> None`	Yes	Waits until the SDK has acked this observer (on any event loop).
`args`	`dict[str, Any]`	No	The observer’s args, validated against `args_format` by the engine.
`capture`	`list[str]`	No	What the target’s payload carries when iii delivers, and with it the outcomes that fire it (see `ObserveSource`).
`filters`	`dict[str, Any]`	No	The observer’s filters, normalized; iii’s delivery and `matches` apply them.
`fires_on`	`list[str] \| None`	No	-
`function_id`	`str`	No	The observed function (one this worker hosts).
`id`	`str`	No	The observer’s id.
`matches`	`(call: CallLike) -> bool`	Yes	Whether `call` passes this observer’s capture and filters: what iii’s delivery checks, for a deliver hook that calls targets itself.
`reject`	`(code: str, message: str) -> None`	Yes	Late rejection of an active observer (e.g. after a config reload).
`target`	`str`	No	The target function id.
`to_dict`	`() -> dict[str, Any]`	Yes	-

ObserverHandle

What iii.observe returns, synchronously. The observer is replayed with the same id on every reconnect until it is disposed. An error answer to its observe or to a replay is the engine’s refusal and is final: the engine holds nothing for it, and it is not replayed. State listeners are called in order, on a callback thread of the SDK (never on the event loop that reads the engine’s frames), so they may call the sync API.

Name	Type	Required	Description
`dispose`	`() -> None`	Yes	Removes the observer (`unobserve`). It is not replayed on reconnect.
`dispose_async`	`async () -> None`	Yes	The async form of `dispose` (it does not wait for the engine).
`id`	`Any`	No	The observer’s id (stable across replays).
`on_state_change`	`(listener: StateListener) -> Callable[[], None]`	Yes	Calls `listener(state, reason)` on every state change, in order, on the SDK’s callback thread. Returns the unsubscribe function.
`ready`	`(timeout: float \| None = None) -> None`	Yes	Blocks until the observer is `active`. Use `ready_async` in async code.
`ready_async`	`async () -> None`	Yes	Waits until the observer is `active` (on any event loop).
`reason`	`ObserveReason \| None`	Yes	Why the observer is in its state (`error`, `suspended`, `disposed`).
`state`	`ObserverState`	Yes	`pending`, `active`, `suspended`, `error` or `disposed`.

ObserverOptions

The observer itself: its id, policy and metadata (the third argument of iii.observe).

Name	Type	Required	Description
`expires_in_ms`	`int`	No	Disposed after this long.
`id`	`str`	No	Default: a random UUID, stable across replays.
`max_fires`	`int`	No	Disposed after this many activations.
`max_in_flight`	`int`	No	Activations running at once (the engine queues the rest).
`metadata`	`dict[str, Any]`	No	Delivered as the metadata sidecar of every activation.

ObserveSource

The left side of iii.observe: what is observed.

Name	Type	Required	Description
`function_id`	`str`	Yes	The observed function.
`args`	`dict[str, Any]`	No	For the worker that hosts the function, which reads them in its hooks; validated against the function’s `args_format`. A function without `args_format` takes none.
`capture`	`list[Capture]`	No	What the target’s payload carries, and with it which outcomes fire the target. Default `["input", "output", "caller"]`. `input` and `output` fire on success, `error` fires on error; `caller` and `metadata` only add to the payload, so a `capture` must name at least one of `input`, `output`, `error` (`INVALID_ARGS` otherwise). `[]` fires on success with the top-level fields only. The default fires on success, `["error", "caller"]` on errors, `["input", "error"]` on both.
`filters`	`Filter`	No	Which calls reach the target; `\{\}` (or absent) = every call.
`namespace`	`str`	No	Defaults to this worker’s (`engine::` and `configuration::` resolve to `default`).

StreamRequest

Incoming streaming request received by a function that serves a streaming HTTP route.

Name	Type	Required	Description
`body`	`Any`	Yes	-
`headers`	`dict[str, str \| list[str]]`	Yes	-
`method`	`str`	Yes	-
`path_params`	`dict[str, str]`	Yes	-
`query_params`	`dict[str, str \| list[str]]`	Yes	-
`request_body`	`ChannelReader`	Yes	-

StreamResponse

Streaming response built on top of a ChannelWriter.

Name	Type	Required	Description
`close`	`() -> None`	Yes	-
`headers`	`async (headers: dict[str, str]) -> None`	Yes	-
`status`	`async (status_code: int) -> None`	Yes	-
`stream`	`WritableStream`	Yes	-
`writer`	`ChannelWriter`	Yes	-

TelemetryOptions

Worker metadata reported to the engine (language, framework, project).

Name	Type	Required	Description
`framework`	`str \| None`	No	Framework name, if applicable.
`language`	`str \| None`	No	Programming language of the worker.
`project_name`	`str \| None`	No	Name of the project this worker belongs to.

iii.channel

Channel · ChannelReader · ChannelWriter · StreamChannelRef

Channel

A streaming channel pair for worker-to-worker data transfer.

Name	Type	Required	Description
`reader`	`ChannelReader`	Yes	-
`reader_ref`	`StreamChannelRef`	Yes	-
`writer`	`ChannelWriter`	Yes	-
`writer_ref`	`StreamChannelRef`	Yes	-

ChannelReader

WebSocket-backed reader for streaming binary data and text messages.

Name	Type	Required	Description
`close_async`	`async () -> None`	Yes	-
`on_message`	`(callback: Callable[[str], Any]) -> None`	Yes	-
`read_all`	`async () -> bytes`	Yes	Read the entire stream into a single bytes object.
`stream`	`Any`	No	-

ChannelWriter

WebSocket-backed writer for streaming binary data and text messages.

Name	Type	Required	Description
`close`	`() -> None`	Yes	End the stream with a clean close frame. Queued from a running loop; a sync caller waits for it.
`close_async`	`async () -> None`	Yes	-
`send_message`	`(msg: str) -> None`	Yes	Text message. Queued from a running loop; a sync caller waits for it.
`send_message_async`	`async (msg: str) -> None`	Yes	-
`stream`	`Any`	No	-
`write`	`async (data: bytes) -> None`	Yes	-

StreamChannelRef

Reference to a streaming channel for worker-to-worker data transfer.

Name	Type	Required	Description
`access_key`	`str`	Yes	Secret key for authenticating channel access.
`channel_id`	`str`	Yes	Unique channel identifier.
`direction`	`Literal['read', 'write']`	Yes	Channel direction (`read` or `write`).

iii.engine

EngineFunctions · EngineSources

EngineFunctions

Engine function ids for internal operations (parity with the Node SDK).

Name	Type	Required	Description
`INFO_FUNCTIONS`	`Final[str]`	No	-
`INFO_WORKERS`	`Final[str]`	No	-
`LIST_FUNCTIONS`	`Final[str]`	No	-
`LIST_WORKERS`	`Final[str]`	No	-
`REGISTER_WORKER`	`Final[str]`	No	-

EngineSources

Ids of the engine-hosted functions that are called once per event: observe them with iii.observe(EngineSources.FUNCTIONS_CHANGED, target) (parity with the Node SDK).

Name	Type	Required	Description
`CONFIGURATION_UPDATED`	`Final[str]`	No	-
`FUNCTIONS_CHANGED`	`Final[str]`	No	-
`LOG`	`Final[str]`	No	-
`OBSERVERS_CHANGED`	`Final[str]`	No	-
`SPAN`	`Final[str]`	No	-
`WORKERS_CHANGED`	`Final[str]`	No	-

iii.errors

InvocationError · ObserveRejection · RegistrationRejectedError

InvocationError

Raised when an invocation dispatched by the SDK fails. Inspect err.code to react to a specific category (e.g. 'FORBIDDEN' for RBAC denials, 'TIMEOUT' for timeouts). Catch this class to handle every rejection. except Exception continues to work because InvocationError inherits from Exception. Attributes are read-only after construction. stacktrace is the engine-side trace when the remote handler raised; it may include internal file paths and should not be surfaced to end users. str(err) intentionally never includes the stacktrace.

Name	Type	Required	Description
`code`	`Any`	No	-
`function_id`	`Any`	No	-
`invocation_id`	`Any`	No	-
`message`	`Any`	No	-
`stacktrace`	`Any`	No	-

ObserveRejection

Rejects an observer with a code of its own. Raise it from an on_observer_added hook, at either level (HTTP_ROUTE_CONFLICT, INVALID_CRON_EXPRESSION, …); any other exception rejects the observer with OBSERVE_REJECTED. Raised from an on_observation_deliver hook, its code is the error the caller gets. Also what ObserverHandle.ready() and Observer.accepted() raise when the observer turns error or disposed, with the engine’s or the hook’s code (INVALID_ARGS, FORBIDDEN, UNSUPPORTED, …).

Name	Type	Required	Description
`code`	`Any`	No	-
`message`	`Any`	No	-

RegistrationRejectedError

Raised when the engine rejects this worker’s registration. The engine pushes a registrationrejected message and closes the connection on a registration collision (e.g. another live worker already owns (namespace, worker_name)). This is fatal: the SDK does not reconnect. Inspect the attributes to identify the conflict.

Name	Type	Required	Description
`code`	`Any`	No	-
`namespace`	`Any`	No	-
`owner_worker_id`	`Any`	No	-
`worker_name`	`Any`	No	-

iii.protocol

CallRequest · MessageType · RegisterFunctionFormat · RegisterFunctionInput · RegisterFunctionMessage · SDK_FEATURES

CallRequest

Request object for call() / call_async(). Name the function to call with function_id, or call through an observer with observer: the call runs the observer’s target, and only the worker that hosts the observed function may make it.

Name	Type	Required	Description
`action`	`CallActionEnqueue` \| `CallActionVoid` \| None	No	Sets how the call is routed. Omit for a synchronous request/response. Specify for a specific routing scheme (e.g. `CallAction.Enqueue(...)`, `CallAction.Void()`).
`dispatch`	`Literal['auto', 'remote'] \| None`	No	`auto` (default) runs a function this worker hosts in-process once the engine confirmed it; `remote` always goes through the engine.
`function_id`	`str \| None`	No	ID of the function to call.
`metadata`	`Any \| None`	No	Per-invocation metadata (any JSON value) supplied to the called handler beside the payload. `None` is no metadata.
`namespace`	`str \| None`	No	Target namespace for routing. Omit to inherit this worker’s; say `default` to reach the engine’s from a namespaced worker.
`observer`	`Any`	No	An `Observer` of a function this worker hosts: the call runs its target, with its policy and metadata sidecar. Takes no `function_id`, `namespace`, `metadata` or enqueue action.
`payload`	`Any`	No	Input data passed to the function.
`timeout_ms`	`int \| None`	No	Override the default invocation timeout, in milliseconds.

MessageType

Message types for iii communication.

Name	Type	Required	Description
`FUNCTION_REGISTERED`	`Any`	No	-
`INVOCATIONS_OBSERVED`	`Any`	No	-
`INVOCATION_RESULT`	`Any`	No	-
`INVOKE_FUNCTION`	`Any`	No	-
`LOCAL_INVOCATION_STATS`	`Any`	No	-
`OBSERVE`	`Any`	No	-
`OBSERVER_ACK`	`Any`	No	-
`OBSERVER_ADDED`	`Any`	No	-
`OBSERVER_REMOVED`	`Any`	No	-
`OBSERVER_STATE_CHANGED`	`Any`	No	-
`OBSERVE_RESULT`	`Any`	No	-
`REATTACH`	`Any`	No	-
`REGISTER_FUNCTION`	`Any`	No	-
`REGISTER_SERVICE`	`Any`	No	-
`REGISTRATION_REJECTED`	`Any`	No	-
`UNOBSERVE`	`Any`	No	-
`UNREGISTER_FUNCTION`	`Any`	No	-
`WORKER_REGISTERED`	`Any`	No	-

RegisterFunctionFormat

Format definition for function parameters.

Name	Type	Required	Description
`body`	list[`RegisterFunctionFormat`] \| None	No	Nested fields for object types.
`description`	`str \| None`	No	Human-readable description of the parameter.
`items`	`RegisterFunctionFormat` \| None	No	Item schema for array types.
`name`	`str`	Yes	Parameter name.
`required`	`bool`	No	Whether the parameter is required.
`type`	`str`	Yes	Type string (`string`, `number`, `boolean`, `object`, `array`, `null`, `map`).

RegisterFunctionInput

Input for registering a function, matches Node.js RegisterFunctionInput.

Name	Type	Required	Description
`description`	`str \| None`	No	Human-readable description.
`id`	`str`	Yes	Unique function identifier.
`invocation`	`HttpInvocationConfig \| None`	No	HTTP invocation config for externally hosted functions.
`metadata`	`Any \| None`	No	Arbitrary metadata attached to the function.
`request_format`	`RegisterFunctionFormat` \| dict[str, Any] \| None	No	Schema describing expected input.
`response_format`	`RegisterFunctionFormat` \| dict[str, Any] \| None	No	Schema describing expected output.

RegisterFunctionMessage

Name	Type	Required	Description
`args_format`	`Any \| None`	No	-
`description`	`str \| None`	No	-
`id`	`str`	Yes	-
`invocation`	`HttpInvocationConfig \| None`	No	-
`message_type`	`MessageType`	No	-
`metadata`	`Any \| None`	No	-
`observable`	`bool \| None`	No	-
`private`	`bool \| None`	No	-
`request_format`	`RegisterFunctionFormat` \| dict[str, Any] \| None	No	-
`response_format`	`RegisterFunctionFormat` \| dict[str, Any] \| None	No	-

SDK_FEATURES

SDK_FEATURES = tuple[str, ...]

iii.runtime

FunctionRef

FunctionRef

A registered function: the client’s hook methods with its id filled in, its observers, and unregister(). Each hook method works as a decorator and as a plain call, returns the hook, and removes the hook when given None. Hooks are stored by function id, so they stay in place across unregister() and a later registration of the same id. The SDK loop runs on a thread of its own, so the engine may announce an observer before a hook set here is in place: on_observer_added then runs for that observer after it became active, and a warning says so. For on_observer_added and on_observer_removed to run for every observer, set them on the client by id before register_function.

Name	Type	Required	Description
`id`	`Any`	No	The function’s id.
`is_observed`	`(call: CallLike \| None = None) -> bool`	Yes	Whether a call would reach anyone (see `III.is_observed`).
`observers`	() -> list[`Observer`]	Yes	The function’s active observers, as this worker sees them.
`on_observation_deliver`	`(hook: DeliverHook \| None) -> DeliverHook \| None`	Yes	This function’s `on_observation_deliver` (see `III.on_observation_deliver`).
`on_observation_match`	`(hook: Any = ..., *, observer: Any = None, call: Any = None) -> Any`	Yes	This function’s `on_observation_match`: a predicate `(obs, call) -> bool`, or `observer=` and `call=` key functions (see `III.on_observation_match`).
`on_observer_added`	`(hook: AddedHook \| None) -> AddedHook \| None`	Yes	This function’s `on_observer_added` (see `III.on_observer_added`).
`on_observer_removed`	`(hook: RemovedHook \| None) -> RemovedHook \| None`	Yes	This function’s `on_observer_removed` (see `III.on_observer_removed`).
`unregister`	`() -> None`	Yes	Removes the function from the engine; `on_observer_removed` runs for each of its accepted observers.

iii.state

IState · StateDeleteInput · StateDeleteResult · StateEventData · StateEventType · StateGetInput · StateListInput · StateSetInput · StateSetResult · StateUpdateInput · StateUpdateResult

IState

Abstract interface for state management operations.

Name	Type	Required	Description
`delete`	async (input: `StateDeleteInput`) -> `StateDeleteResult`	Yes	Delete a state value.
`get`	async (input: `StateGetInput`) -> TData \| None	Yes	Retrieve a value by scope and key.
`list`	async (input: `StateListInput`) -> list[TData]	Yes	List all values in a scope.
`set`	async (input: `StateSetInput`) -> `StateSetResult`[TData] \| None	Yes	Set (create or overwrite) a state value.
`update`	async (input: `StateUpdateInput`) -> `StateUpdateResult`[TData] \| None	Yes	Apply atomic update operations to a state value.

StateDeleteInput

Input for deleting a state value.

Name	Type	Required	Description
`key`	`str`	Yes	Key within the scope.
`scope`	`str`	Yes	State scope (namespace).

StateDeleteResult

Result of a state delete operation.

Name	Type	Required	Description
`old_value`	`Any \| None`	No	Previous value (if it existed).

StateEventData

Payload for state change events.

Name	Type	Required	Description
`event_type`	`StateEventType`	Yes	Type of state change.
`key`	`str`	Yes	Key within the scope.
`new_value`	`TData \| None`	No	New value (for create/update events).
`old_value`	`TData \| None`	No	Previous value (for update/delete events).
`scope`	`str`	Yes	State scope (namespace).
`type`	`str`	No	Event category (always `state`).

StateEventType

Types of state change events.

Name	Type	Required	Description
`CREATED`	`Any`	No	-
`DELETED`	`Any`	No	-
`UPDATED`	`Any`	No	-

StateGetInput

Input for retrieving a state value.

Name	Type	Required	Description
`key`	`str`	Yes	Key within the scope.
`scope`	`str`	Yes	State scope (namespace).

StateListInput

Input for listing all values in a state scope.

Name	Type	Required	Description
`scope`	`str`	Yes	State scope (namespace).

StateSetInput

Input for setting a state value.

Name	Type	Required	Description
`key`	`str`	Yes	Key within the scope.
`scope`	`str`	Yes	State scope (namespace).
`value`	`Any`	Yes	Value to store.

StateSetResult

Result of a state set operation.

Name	Type	Required	Description
`new_value`	`TData`	Yes	New value that was stored.
`old_value`	`TData \| None`	No	Previous value (if it existed).

StateUpdateInput

Input for atomically updating a state value.

Name	Type	Required	Description
`key`	`str`	Yes	Key within the scope.
`ops`	`list[UpdateOp]`	Yes	Ordered list of update operations to apply atomically.
`scope`	`str`	Yes	State scope (namespace).

StateUpdateResult

Result of a state update operation.

Name	Type	Required	Description
`new_value`	`TData`	Yes	New value after the update.
`old_value`	`TData \| None`	No	Previous value (if it existed).

iii.stream

IStream · StreamDeleteInput · StreamDeleteResult · StreamGetInput · StreamListGroupsInput · StreamListInput · StreamSetInput · StreamSetResult · StreamUpdateInput · StreamUpdateResult

IStream

Abstract interface for stream operations. .. deprecated:: IStream is deprecated (iii-stream) and will be removed in an upcoming release. Behavior is unchanged for now. Migration guide: https://iii.dev/docs/upgrading/migrate-from-streams Defining a subclass emits a FutureWarning attributed to the line of the class statement (once per subclass definition, never per request). Importing this module does not warn.

Name	Type	Required	Description
`delete`	async (input: `StreamDeleteInput`) -> `StreamDeleteResult`	Yes	Delete an item from the stream.
`get`	async (input: `StreamGetInput`) -> TData \| None	Yes	Get an item from the stream.
`list`	async (input: `StreamListInput`) -> list[TData]	Yes	Get all items in a group.
`list_groups`	async (input: `StreamListGroupsInput`) -> List[str]	Yes	List all groups in the stream.
`set`	async (input: `StreamSetInput`) -> `StreamSetResult`[TData] \| None	Yes	Set an item in the stream.
`update`	async (input: `StreamUpdateInput`) -> `StreamUpdateResult`[TData] \| None	Yes	Apply atomic update operations to a stream item.

StreamDeleteInput

Input for stream delete operation. .. deprecated:: StreamDeleteInput is deprecated (iii-stream) and will be removed in an upcoming release. Behavior is unchanged for now. Migration guide: https://iii.dev/docs/upgrading/migrate-from-streams

Name	Type	Required	Description
`group_id`	`str`	Yes	Group identifier.
`item_id`	`str`	Yes	Item identifier.
`stream_name`	`str`	Yes	Name of the stream.

StreamDeleteResult

Result of stream delete operation.

Name	Type	Required	Description
`old_value`	`Any \| None`	No	Previous value (if it existed).

StreamGetInput

Input for stream get operation. .. deprecated:: StreamGetInput is deprecated (iii-stream) and will be removed in an upcoming release. Behavior is unchanged for now. Migration guide: https://iii.dev/docs/upgrading/migrate-from-streams

Name	Type	Required	Description
`group_id`	`str`	Yes	Group identifier.
`item_id`	`str`	Yes	Item identifier.
`stream_name`	`str`	Yes	Name of the stream.

StreamListGroupsInput

Input for stream list groups operation. .. deprecated:: StreamListGroupsInput is deprecated (iii-stream) and will be removed in an upcoming release. Behavior is unchanged for now. Migration guide: https://iii.dev/docs/upgrading/migrate-from-streams

Name	Type	Required	Description
`stream_name`	`str`	Yes	Name of the stream.

StreamListInput

Input for stream list operation. .. deprecated:: StreamListInput is deprecated (iii-stream) and will be removed in an upcoming release. Behavior is unchanged for now. Migration guide: https://iii.dev/docs/upgrading/migrate-from-streams

Name	Type	Required	Description
`group_id`	`str`	Yes	Group identifier.
`stream_name`	`str`	Yes	Name of the stream.

StreamSetInput

Input for stream set operation. .. deprecated:: StreamSetInput is deprecated (iii-stream) and will be removed in an upcoming release. Behavior is unchanged for now. Migration guide: https://iii.dev/docs/upgrading/migrate-from-streams

Name	Type	Required	Description
`data`	`Any`	Yes	Data to store.
`group_id`	`str`	Yes	Group identifier.
`item_id`	`str`	Yes	Item identifier.
`stream_name`	`str`	Yes	Name of the stream.

StreamSetResult

Result of stream set operation.

Name	Type	Required	Description
`new_value`	`TData`	Yes	New value that was stored.
`old_value`	`TData \| None`	No	Previous value (if it existed).

StreamUpdateInput

Input for stream update operation. .. deprecated:: StreamUpdateInput is deprecated (iii-stream) and will be removed in an upcoming release. Behavior is unchanged for now. Migration guide: https://iii.dev/docs/upgrading/migrate-from-streams

Name	Type	Required	Description
`group_id`	`str`	Yes	Group identifier.
`item_id`	`str`	Yes	Item identifier.
`ops`	`list['UpdateOp']`	Yes	Ordered list of update operations to apply atomically.
`stream_name`	`str`	Yes	Name of the stream.

StreamUpdateResult

Result of stream update operation.

Name	Type	Required	Description
`errors`	`list[UpdateOpError]`	No	Per-op errors. Emitted by `merge` and `append` for validation rejections (path depth/size, value depth, or a `__proto__` / `constructor` / `prototype` segment or top-level key) and by `append` for the `append.type_mismatch` and `append.target_not_object` surfaces. Successfully applied ops are still reflected in `new_value`. The field is omitted from the JSON wire when empty.
`new_value`	`TData`	Yes	New value after the update.
`old_value`	`TData \| None`	No	Previous value (if it existed).

​Installation

​Initialization

​register_worker

​Methods

​register_function

​call

​observe

​trigger

​Parameters

​add_connection_state_listener

​call_async

​connect_async

​current_invocation

​Example

​get_address

​get_connection_state

​Example

​is_observed

​observers

​on_observation_deliver

​on_observation_match

​on_observer_added

​on_observer_removed

​register_trigger

​Parameters

​register_trigger_type

​Parameters

​shutdown

​Example

​shutdown_async

​Example

​trigger_async

​Parameters

​unregister_trigger_type

​Parameters

​Types

​iii

​Call

​CallAction

​CallActionEnqueue

​CallActionVoid

​Caller

​EnqueueResult

​InitOptions

​InvocationContext

​MatchKeys

​MiddlewareFunctionInput

​Observed

​Observer

​ObserverHandle

​ObserverOptions

​ObserveSource

​StreamRequest

​StreamResponse

​TelemetryOptions

​iii.channel

​Channel

​ChannelReader

​ChannelWriter

​StreamChannelRef

​iii.engine

​EngineFunctions

​EngineSources

​iii.errors

​InvocationError

​ObserveRejection

​RegistrationRejectedError

​iii.protocol

​CallRequest

​MessageType

​RegisterFunctionFormat

​RegisterFunctionInput

​RegisterFunctionMessage

​SDK_FEATURES

​iii.runtime

​FunctionRef

​iii.state

​IState

​StateDeleteInput

​StateDeleteResult

Installation

Initialization

register_worker

Methods

register_function

call

observe

trigger

Parameters

add_connection_state_listener

call_async

connect_async

current_invocation

Example

get_address

get_connection_state

Example

is_observed

observers

on_observation_deliver

on_observation_match

on_observer_added

on_observer_removed

register_trigger

Parameters

register_trigger_type

Parameters

shutdown

Example

shutdown_async

Example

trigger_async

Parameters

unregister_trigger_type

Parameters

Types

iii

Call

CallAction

CallActionEnqueue

CallActionVoid

Caller

EnqueueResult

InitOptions

InvocationContext

MatchKeys

MiddlewareFunctionInput

Observed

Observer

ObserverHandle

ObserverOptions

ObserveSource

StreamRequest

StreamResponse

TelemetryOptions

iii.channel

Channel

ChannelReader

ChannelWriter

StreamChannelRef

iii.engine

EngineFunctions

EngineSources

iii.errors

InvocationError

ObserveRejection

RegistrationRejectedError

iii.protocol

CallRequest

MessageType

RegisterFunctionFormat

RegisterFunctionInput

RegisterFunctionMessage

SDK_FEATURES

iii.runtime

FunctionRef

iii.state

IState

StateDeleteInput

StateDeleteResult