Quickstart

Get started with the III SDK for Python

Installation

First, install the III SDK:

pip install iii-sdk

Basic Example

Create a simple worker that registers a function and calls it:

import asyncio
from iii import III

async def greet(data):
    name = data.get("name", "World")
    return {"message": f"Hello, {name}!"}

iii = III("ws://localhost:49134")
iii.register_function("greet", greet)

async def main():
    await iii.connect()
    
    # Call the function
    result = await iii.call("greet", {"name": "Alice"})
    print(result)  # {"message": "Hello, Alice!"}
    
    # Keep running
    await asyncio.Event().wait()

if __name__ == "__main__":
    asyncio.run(main())

Function Registration

Register functions with descriptions and metadata:

from iii import III

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

async def process_order(data):
    order_id = data["order_id"]
    # Process the order...
    return {"status": "processed", "order_id": order_id}

iii.register_function(
    "orders.process",
    process_order,
    description="Process a customer order",
    metadata={"version": "1.0", "team": "orders"}
)

Calling Functions

Call remote functions with timeout control:

# Await response (default 30s timeout)
result = await iii.call("orders.process", {"order_id": "123"})

# Custom timeout
result = await iii.call("orders.process", {"order_id": "123"}, timeout=60.0)

# Fire-and-forget (no response)
iii.call_void("notifications.send", {"user_id": "456", "message": "Order shipped"})

Using Context and Logging

Access the execution context within functions:

from iii import get_context

async def process_payment(data):
    ctx = get_context()
    ctx.logger.info("Processing payment", data={"amount": data["amount"]})
    
    # Process payment...
    
    ctx.logger.info("Payment processed successfully")
    return {"status": "success"}

HTTP Triggers

Register HTTP triggers to expose functions as REST endpoints:

from iii import III, ApiRequest, ApiResponse

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

async def create_todo(data):
    req = ApiRequest(**data)
    title = req.body.get("title")
    
    # Save todo...
    
    return ApiResponse(
        status_code=201,
        body={"id": "123", "title": title}
    )

iii.register_function("api.todos.create", create_todo)

async def main():
    await iii.connect()
    
    # Register HTTP trigger
    iii.register_trigger(
        type="http",
        function_id="api.todos.create",
        config={
            "api_path": "/todos",
            "http_method": "POST"
        }
    )
    
    await asyncio.Event().wait()

asyncio.run(main())

Streaming Channels

Create channels for streaming data between workers:

async def producer(data):
    channel = await iii.create_channel()
    
    # Pass writer_ref to another function
    iii.call_void("consumer", {"reader": channel.reader_ref})
    
    # Write data
    await channel.writer.write(b"chunk1")
    await channel.writer.write(b"chunk2")
    await channel.writer.close_async()
    
    return {"status": "sent"}

async def consumer(data):
    reader = data["reader"]  # Automatically resolved to ChannelReader
    
    async for chunk in reader:
        print(f"Received: {chunk}")
    
    return {"status": "received"}

Connection Configuration

Configure the client with custom options:

from iii import III, InitOptions, ReconnectionConfig

options = InitOptions(
    worker_name="my-worker",
    invocation_timeout_ms=60000,  # 60 seconds
    reconnection_config=ReconnectionConfig(
        initial_delay_ms=2000,
        max_delay_ms=60000,
        backoff_multiplier=2.0,
        max_retries=-1  # Infinite retries
    )
)

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

Shutdown

Gracefully shutdown the connection:

async def main():
    iii = III("ws://localhost:49134")
    await iii.connect()
    
    try:
        # Your application logic
        await asyncio.Event().wait()
    except KeyboardInterrupt:
        await iii.shutdown()

asyncio.run(main())

Next Steps

Client API

Learn about the III class and connection options

Functions

Register and invoke functions

Channels

Stream data between workers

Telemetry

Enable OpenTelemetry tracing and metrics