> ## 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.

# Client

> III class for connecting to the III Engine

## III

The main WebSocket client for communication with the III Engine.

```python theme={null}
from iii import III

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

### Constructor

```python theme={null}
III(address: str, options: InitOptions | None = None)
```

<ParamField path="address" type="str" required>
  WebSocket URL of the III Engine (e.g., `ws://localhost:49134`)
</ParamField>

<ParamField path="options" type="InitOptions">
  Configuration options for the client
</ParamField>

### Methods

#### connect

Connect to the WebSocket server and initialize OpenTelemetry.

```python theme={null}
await iii.connect()
```

#### shutdown

Gracefully disconnect from the server and shut down OpenTelemetry.

```python theme={null}
await iii.shutdown()
```

#### get\_connection\_state

Get the current connection state.

```python theme={null}
state = iii.get_connection_state()
print(state)  # "connected", "connecting", "reconnecting", "disconnected", or "failed"
```

<ResponseField name="state" type="IIIConnectionState">
  One of: `"connected"`, `"connecting"`, `"reconnecting"`, `"disconnected"`, `"failed"`
</ResponseField>

#### on\_connection\_state\_change

Register a callback for connection state changes.

```python theme={null}
def on_state_change(state):
    print(f"Connection state: {state}")

unsubscribe = iii.on_connection_state_change(on_state_change)

# Later: remove callback
unsubscribe()
```

<ParamField path="callback" type="Callable[[IIIConnectionState], None]" required>
  Function called whenever connection state changes
</ParamField>

<ResponseField name="unsubscribe" type="Callable[[], None]">
  Function to remove the callback
</ResponseField>

#### list\_functions

List all registered functions from the engine.

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

<ResponseField name="functions" type="list[FunctionInfo]">
  List of function metadata objects
</ResponseField>

#### list\_workers

List all connected workers from the engine.

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

<ResponseField name="workers" type="list[WorkerInfo]">
  List of worker metadata objects
</ResponseField>

#### on\_functions\_available

Subscribe to function availability events.

```python theme={null}
def on_functions(functions):
    print(f"Functions available: {[f.function_id for f in functions]}")

unsubscribe = iii.on_functions_available(on_functions)
```

<ParamField path="callback" type="Callable[[list[FunctionInfo]], None]" required>
  Function called when functions become available
</ParamField>

<ResponseField name="unsubscribe" type="Callable[[], None]">
  Function to remove the callback and clean up the trigger
</ResponseField>

### Properties

#### worker\_id

The worker ID assigned by the engine.

```python theme={null}
print(iii.worker_id)  # "worker-abc123" or None if not yet registered
```

<ResponseField name="worker_id" type="str | None">
  The worker ID, or None if not yet registered
</ResponseField>

## InitOptions

Configuration options for the III client.

```python theme={null}
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)
```

<ParamField path="worker_name" type="str">
  Custom name for this worker. Defaults to `{hostname}:{pid}`
</ParamField>

<ParamField path="enable_metrics_reporting" type="bool" default={true}>
  Enable worker metrics reporting to the engine
</ParamField>

<ParamField path="invocation_timeout_ms" type="int" default={30000}>
  Default timeout for function invocations in milliseconds
</ParamField>

<ParamField path="reconnection_config" type="ReconnectionConfig">
  WebSocket reconnection behavior configuration
</ParamField>

<ParamField path="otel" type="dict[str, Any]">
  OpenTelemetry configuration dictionary (deprecated - use `init_otel()` instead)
</ParamField>

<ParamField path="telemetry" type="TelemetryOptions">
  Telemetry metadata to be reported to the engine
</ParamField>

## ReconnectionConfig

Configures automatic WebSocket reconnection behavior.

```python theme={null}
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
)
```

<ParamField path="initial_delay_ms" type="int" default={1000}>
  Starting delay in milliseconds before first retry
</ParamField>

<ParamField path="max_delay_ms" type="int" default={30000}>
  Maximum delay cap in milliseconds
</ParamField>

<ParamField path="backoff_multiplier" type="float" default={2.0}>
  Exponential backoff multiplier for each retry
</ParamField>

<ParamField path="jitter_factor" type="float" default={0.3}>
  Random jitter factor (0-1) to prevent thundering herd
</ParamField>

<ParamField path="max_retries" type="int" default={-1}>
  Maximum retry attempts. Set to `-1` for infinite retries
</ParamField>

## FunctionRef

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

```python theme={null}
ref = iii.register_function("my.function", handler)

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

# Unregister the function
ref.unregister()
```

<ResponseField name="id" type="str">
  The function ID
</ResponseField>

<ResponseField name="unregister" type="Callable[[], None]">
  Function to unregister this function from the engine
</ResponseField>

## Types

### IIIConnectionState

Connection state literal type:

```python theme={null}
IIIConnectionState = Literal[
    "disconnected",
    "connecting", 
    "connected",
    "reconnecting",
    "failed"
]
```

### ConnectionStateCallback

Callback type for connection state changes:

```python theme={null}
ConnectionStateCallback = Callable[[IIIConnectionState], None]
```

### FunctionInfo

Metadata about a registered function:

```python theme={null}
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:

```python theme={null}
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

```python theme={null}
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
```