Documentation Index
Fetch the complete documentation index at: https://mintlify.com/iii-hq/sdk/llms.txt
Use this file to discover all available pages before exploring further.
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)
WebSocket URL of the III Engine (e.g., ws://localhost:49134)
Configuration options for the client
Methods
connect
Connect to the WebSocket server and initialize OpenTelemetry.
shutdown
Gracefully disconnect from the server and shut down OpenTelemetry.
get_connection_state
Get the current connection state.
state = iii.get_connection_state()
print(state) # "connected", "connecting", "reconnecting", "disconnected", or "failed"
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
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}")
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")
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
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
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)
Custom name for this worker. Defaults to {hostname}:{pid}
Enable worker metrics reporting to the engine
Default timeout for function invocations in milliseconds
WebSocket reconnection behavior configuration
OpenTelemetry configuration dictionary (deprecated - use init_otel() instead)
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
)
Starting delay in milliseconds before first retry
Maximum delay cap in milliseconds
Exponential backoff multiplier for each retry
Random jitter factor (0-1) to prevent thundering herd
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()
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