Client

III class for connecting to the III Engine

III

The main WebSocket client for communication with the III Engine.

from iii import III

iii = III("ws://localhost:49134")

Constructor

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

address · str · required

WebSocket URL of the III Engine (e.g., ws://localhost:49134)

options · InitOptions

Configuration options for the client

Methods

connect

Connect to the WebSocket server and initialize OpenTelemetry.

await iii.connect()

shutdown

Gracefully disconnect from the server and shut down OpenTelemetry.

await iii.shutdown()

get_connection_state

Get the current connection state.

state = iii.get_connection_state()
print(state)  # "connected", "connecting", "reconnecting", "disconnected", or "failed"

state · IIIConnectionState

One of: "connected", "connecting", "reconnecting", "disconnected", "failed"

on_connection_state_change

Register a callback for connection state changes.

def on_state_change(state):
    print(f"Connection state: {state}")

unsubscribe = iii.on_connection_state_change(on_state_change)

# Later: remove callback
unsubscribe()

callback · Callable[[IIIConnectionState], None] · required

Function called whenever connection state changes

unsubscribe · Callable[[], None]

Function to remove the callback

list_functions

List all registered functions from the engine.

functions = await iii.list_functions()
for func in functions:
    print(f"{func.function_id}: {func.description}")

functions · list[FunctionInfo]

List of function metadata objects

list_workers

List all connected workers from the engine.

workers = await iii.list_workers()
for worker in workers:
    print(f"{worker.name} ({worker.status}): {worker.function_count} functions")

workers · list[WorkerInfo]

List of worker metadata objects

on_functions_available

Subscribe to function availability events.

def on_functions(functions):
    print(f"Functions available: {[f.function_id for f in functions]}")

unsubscribe = iii.on_functions_available(on_functions)

callback · Callable[[list[FunctionInfo]], None] · required

Function called when functions become available

unsubscribe · Callable[[], None]

Function to remove the callback and clean up the trigger

Properties

worker_id

The worker ID assigned by the engine.

print(iii.worker_id)  # "worker-abc123" or None if not yet registered

worker_id · str | None

The worker ID, or None if not yet registered

InitOptions

Configuration options for the III client.

from iii import III, InitOptions, ReconnectionConfig

options = InitOptions(
    worker_name="my-worker",
    invocation_timeout_ms=60000,
    reconnection_config=ReconnectionConfig(
        initial_delay_ms=2000,
        max_delay_ms=60000
    )
)

iii = III("ws://localhost:49134", options)

worker_name · str

Custom name for this worker. Defaults to {hostname}:{pid}

enable_metrics_reporting · bool · default: true

Enable worker metrics reporting to the engine

invocation_timeout_ms · int · default: 30000

Default timeout for function invocations in milliseconds

reconnection_config · ReconnectionConfig

WebSocket reconnection behavior configuration

otel · dict[str, Any]

OpenTelemetry configuration dictionary (deprecated - use init_otel() instead)

telemetry · TelemetryOptions

Telemetry metadata to be reported to the engine

ReconnectionConfig

Configures automatic WebSocket reconnection behavior.

from iii import ReconnectionConfig

config = ReconnectionConfig(
    initial_delay_ms=1000,
    max_delay_ms=30000,
    backoff_multiplier=2.0,
    jitter_factor=0.3,
    max_retries=-1  # Infinite retries
)

initial_delay_ms · int · default: 1000

Starting delay in milliseconds before first retry

max_delay_ms · int · default: 30000

Maximum delay cap in milliseconds

backoff_multiplier · float

Exponential backoff multiplier for each retry

jitter_factor · float

Random jitter factor (0-1) to prevent thundering herd

max_retries · int

Maximum retry attempts. Set to -1 for infinite retries

FunctionRef

Reference to a registered function, returned by register_function().

ref = iii.register_function("my.function", handler)

print(ref.id)  # "my.function"

# Unregister the function
ref.unregister()

id · str

The function ID

unregister · Callable[[], None]

Function to unregister this function from the engine

Types

IIIConnectionState

Connection state literal type:

IIIConnectionState = Literal[
    "disconnected",
    "connecting", 
    "connected",
    "reconnecting",
    "failed"
]

ConnectionStateCallback

Callback type for connection state changes:

ConnectionStateCallback = Callable[[IIIConnectionState], None]

FunctionInfo

Metadata about a registered function:

class FunctionInfo(BaseModel):
    function_id: str
    description: str | None
    request_format: RegisterFunctionFormat | None
    response_format: RegisterFunctionFormat | None
    metadata: dict[str, Any] | None

WorkerInfo

Metadata about a connected worker:

class WorkerInfo(BaseModel):
    id: str
    name: str | None
    runtime: str | None  # "python"
    version: str | None  # SDK version
    os: str | None
    ip_address: str | None
    status: WorkerStatus  # "connected", "available", "busy", "disconnected"
    connected_at_ms: int
    function_count: int
    functions: list[str]
    active_invocations: int

Example: Full Configuration

import asyncio
from iii import III, InitOptions, ReconnectionConfig

options = InitOptions(
    worker_name="payment-processor-1",
    invocation_timeout_ms=60000,
    enable_metrics_reporting=True,
    reconnection_config=ReconnectionConfig(
        initial_delay_ms=2000,
        max_delay_ms=60000,
        backoff_multiplier=2.0,
        jitter_factor=0.3,
        max_retries=-1
    )
)

iii = III("ws://localhost:49134", options)

def on_state_change(state):
    print(f"Connection state: {state}")

async def main():
    iii.on_connection_state_change(on_state_change)
    await iii.connect()
    
    print(f"Worker ID: {iii.worker_id}")
    
    # Your application logic
    await asyncio.Event().wait()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        pass