Python SDK for the iii engine. Use when building workers, registering functions, or invoking triggers in Python.
The async Python SDK for connecting workers to the iii engine.
Full API reference: https://iii.dev/docs/api-reference/sdk-python
pip install iii-sdk
| Export | Purpose |
|---|---|
register_worker(address, options?) | Connect to the engine, returns the client |
InitOptions(worker_name, otel?) | Connection configuration |
register_function(id, handler) | Register an async function handler |
register_trigger(type, function_id, config) | Bind a trigger to a function |
trigger(request) | Invoke a function synchronously |
trigger_async(request) | Invoke a function asynchronously |
get_context() | Access logger and trace context inside handlers |
ApiRequest / ApiResponse | HTTP request/response types (pydantic) |
IStream | Interface for custom stream implementations |
on_functions_available(callback) | Listen for function discovery |
on_connection_state_change(callback) | Monitor connection state |
register_trigger(type, fn_id, config, metadata?) | Bind a trigger with optional metadata |
register_worker() returns a synchronous client; handlers are asyncApiResponse uses camelCase statusCode (pydantic alias), not status_codewhile True: await asyncio.sleep(60) to keep the event loop aliveasyncio.to_thread() for CPU-heavy sync work inside handlerstrigger_async(request) and a synchronous trigger(request). Use trigger_async inside async handlers, and trigger in synchronous scripts or threads where blocking behavior is desired.# Async invocation (non-blocking, typical inside handlers)
result = await iii.trigger_async({
"function_id": "greet",
"payload": {"name": "World"}
})
# Sync invocation (blocks the current thread, useful in sync contexts)
result = iii.trigger({
"function_id": "greet",
"payload": {"name": "World"}
})
iii-functions-and-triggersiii-http-middlewareiii-node-sdkiii-rust-sdkiii-browser-sdkiii-python-sdk in the iii engine.