Skip to main content
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 (worker_name, plus an optional worker_description, a one-line summary surfaced in engine::workers::list / engine::workers::info) 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

Register a callable function on this worker. 
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

Every invocation failure raises IIIInvocationError. Inspect its code attribute to distinguish categories:
codeWhen raised
"FORBIDDEN"RBAC denied the invocation.
"TIMEOUT"The invocation timed out.
otherAny other engine-side rejection.
IIIInvocationError 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. Import it from iii_observability; it is no longer re-exported from iii. 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.