Python SDK

This page is a hand-authored snapshot of the planned public surface. The final reference will be generated from the SDK source.

Installation

pip install iii-sdk

Imported as iii.

Common methods

`register_worker`

Connect a worker to a running iii engine and return its handle.

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

address is the engine’s SDK WebSocket URL. options configures worker identity and reconnection. The returned III instance carries every method below.

The SDK’s OpenTelemetry hookup is wired through options as well; the export and rollup side is owned by iii-observability.

`register_function`

def register_function(
    self,
    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,
) -> FunctionRef: ...

request_format / response_format accept either a JSON Schema dict or a RegisterFunctionFormat helper. They are stored alongside the function for the iii console and the agent-readable skills.

`register_trigger`

Bind a registered function to a configured trigger instance.

def register_trigger(self, trigger: RegisterTriggerInput | dict[str, Any]) -> Trigger: ...

Drop the trigger with trigger.unregister() on the returned handle. There is no top-level unregister_trigger method.

`register_trigger_type`

Declare a new trigger type that this worker advertises.

def register_trigger_type(
    self,
    trigger_type: RegisterTriggerTypeInput | dict[str, Any],
    handler: TriggerHandler[Any],
) -> TriggerTypeRef[Any, Any]: ...

`unregister_trigger_type`

Remove a previously registered trigger type.

def unregister_trigger_type(
    self,
    trigger_type: RegisterTriggerTypeInput | dict[str, Any],
) -> None: ...

`trigger` / `trigger_async`

Invoke a registered function. trigger is the synchronous entry point (runs the async machinery on the SDK’s internal loop); trigger_async is the awaitable form for callers inside asyncio.

def trigger(self, request: dict[str, Any] | TriggerRequest) -> Any: ...
async def trigger_async(self, request: dict[str, Any] | TriggerRequest) -> Any: ...

Both return the function’s value for synchronous invocations, an EnqueueResult for TriggerAction.Enqueue actions, and None for TriggerAction.Void.

`shutdown` / `shutdown_async`

Disconnect from the engine and release resources. Use shutdown_async from asyncio contexts.

def shutdown(self) -> None: ...
async def shutdown_async(self) -> None: ...

Trigger actions

TriggerAction is a factory class with two static helpers; TriggerActionEnqueue and TriggerActionVoid are the concrete return shapes.

TriggerAction.Void()                  # fire-and-forget; returns TriggerActionVoid()
TriggerAction.Enqueue(queue="math")   # route through iii-queue; returns TriggerActionEnqueue(...)

queue is a keyword-only argument on Enqueue.

Error types

The base class is IIIInvocationError. The SDK’s wire-error decoder maps engine error codes to two known subclasses; everything else stays on the base class:

Class	When raised
`IIIInvocationError`	Any engine-side invocation error.
`IIIForbiddenError`	`code == "FORBIDDEN"` (RBAC).
`IIITimeoutError`	`code == "TIMEOUT"`.

All three are exported. The base class has code, message, function_id, and stacktrace attributes.

Channels

ChannelReader and ChannelWriter wrap the engine’s stream WebSockets. StreamChannelRef identifies a channel:

class StreamChannelRef(BaseModel):
    channel_id: str
    access_key: str
    direction: Literal["read", "write"]

Both classes are constructed with the engine’s WS base URL and a StreamChannelRef.

Logger

Logger exposes info, warn, error, and debug, each accepting a message and an optional data dict. The output integrates with the SDK’s OpenTelemetry setup; see iii-observability for the export side.

Info types

FunctionInfo. function_id, optional description, optional request_format / response_format, optional metadata.
TriggerInfo. id, trigger_type, function_id, optional config / metadata.

WorkerInfo exists in iii_types but isn’t currently re-exported from the package root; import it from iii.iii_types when needed. WorkerMetadata is not part of this SDK.

`MessageType`

A runtime enum naming every wire frame the SDK exchanges with the engine. Used internally by middleware; rarely needed by callers.

`RegisterFunctionFormat`

The Python-only helper for declaring a function’s request or response schema in a structured way. Accepts either a JSON Schema dict directly or constructed values; both forms reach register_function.

Connection state

IIIConnectionState is the literal-type alias "disconnected" | "connecting" | "connected" | "reconnecting" | "failed". It is defined in iii.iii_constants but not re-exported from the package root; treat the connection as established once register_worker returns.

Engine protocol

Language SDKs

Python SDK

Installation

Common methods

`register_worker`

`register_function`

`register_trigger`

`register_trigger_type`

`unregister_trigger_type`

`trigger` / `trigger_async`

`shutdown` / `shutdown_async`

Trigger actions

Error types

Channels

Logger

Info types

`MessageType`

`RegisterFunctionFormat`

Connection state

​Installation

​Common methods

​register_worker

​register_function

​register_trigger

​register_trigger_type

​unregister_trigger_type

​trigger / trigger_async

​shutdown / shutdown_async

​Trigger actions

​Error types

​Channels

​Logger

​Info types

​MessageType

​RegisterFunctionFormat

​Connection state

Installation

Common methods

`register_worker`

`register_function`

`register_trigger`

`register_trigger_type`

`unregister_trigger_type`

`trigger` / `trigger_async`

`shutdown` / `shutdown_async`

Trigger actions

Error types

Channels

Logger

Info types

`MessageType`

`RegisterFunctionFormat`

Connection state