Python SDK

Installation

Install the iii SDK via pip:

pip install iii-sdk

Quick Start

Here’s a simple example that creates a function and exposes it via HTTP:

from iii import III
import asyncio

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

async def add(data):
    return {"sum": data["a"] + data["b"]}

iii.register_function("math.add", add)

async def main():
    await iii.connect()

    iii.register_trigger(
        type="http",
        function_id="math.add",
        config={"api_path": "add", "http_method": "POST"}
    )

    # Keep running
    await asyncio.Event().wait()

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

Your function is now live at http://localhost:3111/add.

Connection

Initialize the SDK

Create an iii client instance:

from iii import III

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

Connect to the Engine

Establish a WebSocket connection:

import asyncio

async def main():
    await iii.connect()
    # Your code here
    await iii.disconnect()

asyncio.run(main())

Environment Variables

Configure the connection URL using environment variables:

import os
from iii import III

BRIDGE_URL = os.environ.get("III_BRIDGE_URL", "ws://127.0.0.1:49134")
iii = III(BRIDGE_URL)

Registering Functions

Basic Function Registration

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

iii.register_function("greet.user", greet)

Function with Type Hints

Use Python type hints for better code clarity:

from typing import Dict, Any

async def create_user(data: Dict[str, Any]) -> Dict[str, Any]:
    # Create user logic
    user_id = await save_user(data)
    return {"user_id": user_id, "created": True}

iii.register_function("user.create", create_user)

Synchronous vs Asynchronous

While the SDK is async-based, you can wrap synchronous code:

import asyncio

def sync_operation(data):
    # Synchronous operation
    return {"result": data["value"] * 2}

async def async_wrapper(data):
    # Run sync code in executor if needed
    result = await asyncio.to_thread(sync_operation, data)
    return result

iii.register_function("sync.operation", async_wrapper)

Error Handling

Handle errors in your functions:

async def divide(data):
    try:
        a = data["a"]
        b = data["b"]
        if b == 0:
            raise ValueError("Division by zero")
        return {"result": a / b}
    except KeyError as e:
        raise ValueError(f"Missing required field: {e}")

iii.register_function("math.divide", divide)

Registering Triggers

Triggers must be registered after connecting to the engine.

HTTP
Queue
Cron
Stream

Expose functions as HTTP endpoints:

await iii.connect()

iii.register_trigger(
    type="http",
    function_id="math.add",
    config={
        "api_path": "add",
        "http_method": "POST"
    }
)

Configuration options:

api_path: The URL path (e.g., ‘add’ → /add)
http_method: HTTP method (GET, POST, PUT, DELETE, PATCH)

Dynamic routes:

iii.register_trigger(
    type="http",
    function_id="user.get",
    config={
        "api_path": "users/:id",
        "http_method": "GET"
    }
)

Subscribe to queue topics:

iii.register_trigger(
    type="queue",
    function_id="email.send",
    config={"topic": "notifications"}
)

The function will be invoked whenever a message is published to the topic.

Schedule functions to run on a cron schedule:

iii.register_trigger(
    type="cron",
    function_id="cleanup.daily",
    config={"schedule": "0 2 * * *"}  # Every day at 2 AM
)

Uses standard cron syntax.

React to real-time state changes:

iii.register_trigger(
    type="stream",
    function_id="handle.update",
    config={"channel": "user-updates"}
)

Functions are invoked when data is streamed to the channel.

Invoking Functions

Invoke functions directly from your code:

# Invoke a function and wait for the result
result = await iii.invoke("math.add", {"a": 5, "b": 3})
print(result["sum"])  # 8

Fire-and-Forget Invocations

Invoke without waiting for a response:

# Send the invocation and continue immediately
iii.invoke_async("log.event", {"event": "user_login"})

Complete Example

Here’s a comprehensive example with multiple functions and triggers:

import asyncio
import logging
from iii import III

logging.basicConfig(level=logging.INFO)
log = logging.getLogger(__name__)

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

# In-memory storage
greetings = {}

async def greet_and_save(data):
    """Greet a user and save the greeting."""
    name = data.get("name", "World")
    greeting = f"Hello, {name}!"
    greetings[name] = greeting
    return {"greeting": greeting, "saved": True}

async def list_greetings(data):
    """List all saved greetings."""
    return {"greetings": greetings}

async def get_greeting(data):
    """Get a specific greeting by name."""
    name = data.get("name")
    greeting = greetings.get(name)
    if greeting:
        return {"greeting": greeting}
    return {"error": "Greeting not found"}

async def delete_greeting(data):
    """Delete a greeting."""
    name = data.get("name")
    if name in greetings:
        del greetings[name]
        return {"deleted": True}
    return {"error": "Greeting not found"}

# Register functions
iii.register_function("python.greet", greet_and_save)
iii.register_function("python.list_greetings", list_greetings)
iii.register_function("python.get_greeting", get_greeting)
iii.register_function("python.delete_greeting", delete_greeting)

async def main():
    # Connect to the engine
    log.info("Connecting to iii engine...")
    await iii.connect()

    # Register HTTP triggers
    iii.register_trigger(
        type="http",
        function_id="python.greet",
        config={"api_path": "greet", "http_method": "GET"}
    )

    iii.register_trigger(
        type="http",
        function_id="python.list_greetings",
        config={"api_path": "greetings", "http_method": "GET"}
    )

    iii.register_trigger(
        type="http",
        function_id="python.get_greeting",
        config={"api_path": "greetings/:name", "http_method": "GET"}
    )

    iii.register_trigger(
        type="http",
        function_id="python.delete_greeting",
        config={"api_path": "greetings/:name", "http_method": "DELETE"}
    )

    log.info("=" * 60)
    log.info("API Endpoints:")
    log.info("  GET    /greet?name=X      - Greet and save")
    log.info("  GET    /greetings         - List all greetings")
    log.info("  GET    /greetings/:name   - Get specific greeting")
    log.info("  DELETE /greetings/:name   - Delete greeting")
    log.info("=" * 60)

    # Keep running
    log.info("Worker running. Press Ctrl+C to exit...")
    try:
        await asyncio.Event().wait()
    except KeyboardInterrupt:
        log.info("Shutting down...")
    finally:
        await iii.disconnect()

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

API Reference

`III(url: str)`

Create an iii SDK instance. Parameters:

url: WebSocket URL of the iii engine (default: ws://localhost:49134)

Returns: SDK instance

`register_function(function_id: str, handler: Callable)`

function_id: Unique function identifier
handler: Async function that receives data dict and returns result dict

`register_trigger(type: str, function_id: str, config: dict)`

type: Trigger type (http, queue, cron, stream)
function_id: ID of the function to invoke
config: Trigger-specific configuration dict

`async connect()`

Connect to the iii engine.

`async disconnect()`

Disconnect from the iii engine.

`async invoke(function_id: str, data: dict)`

Invoke a function and wait for the result. Parameters:

function_id: ID of the function to invoke
data: Input data dictionary

Returns: Function result

Best Practices

Use async/await consistently

The SDK is async-first. Always use async def for function handlers and await for operations.

Handle connection lifecycle

Always call await iii.connect() before registering triggers and await iii.disconnect() when shutting down.

Use structured logging

Use Python’s logging module for better observability.

Type hints for clarity

Use type hints to make your code more maintainable and catch errors early.

Graceful shutdown

Handle KeyboardInterrupt and other signals to disconnect cleanly.

Next Steps

Node.js SDK

Learn about the Node.js SDK

Rust SDK

Learn about the Rust SDK

HTTP Module

Deep dive into HTTP triggers

Queue Module

Learn about queue-based triggers

Get Started

Core Concepts

Modules

SDKs

Deployment

Advanced

Python SDK

Installation

Quick Start

Connection

Initialize the SDK

Connect to the Engine

Environment Variables

Registering Functions

Basic Function Registration

Function with Type Hints

Synchronous vs Asynchronous

Error Handling

Registering Triggers

Invoking Functions

Fire-and-Forget Invocations

Complete Example

API Reference

`III(url: str)`

`register_function(function_id: str, handler: Callable)`

`register_trigger(type: str, function_id: str, config: dict)`

`async connect()`

`async disconnect()`

`async invoke(function_id: str, data: dict)`

Best Practices

Next Steps

Node.js SDK

Rust SDK

HTTP Module

Queue Module

Build docs developers (and LLMs) love

Get Started

Core Concepts

Modules

SDKs

Deployment

Advanced

Documentation Index

​Installation

​Quick Start

​Connection

​Initialize the SDK

​Connect to the Engine

​Environment Variables

​Registering Functions

​Basic Function Registration

​Function with Type Hints

​Synchronous vs Asynchronous

​Error Handling

​Registering Triggers

​Invoking Functions

​Fire-and-Forget Invocations

​Complete Example

​API Reference

​III(url: str)

​register_function(function_id: str, handler: Callable)

​register_trigger(type: str, function_id: str, config: dict)

​async connect()

​async disconnect()

​async invoke(function_id: str, data: dict)

​Best Practices

​Next Steps

Node.js SDK

Rust SDK

HTTP Module

Queue Module

Build docs developers (and LLMs) love

Installation

Quick Start

Connection

Initialize the SDK

Connect to the Engine

Environment Variables

Registering Functions

Basic Function Registration

Function with Type Hints

Synchronous vs Asynchronous

Error Handling

Registering Triggers

Invoking Functions

Fire-and-Forget Invocations

Complete Example

API Reference

`III(url: str)`

`register_function(function_id: str, handler: Callable)`

`register_trigger(type: str, function_id: str, config: dict)`

`async connect()`

`async disconnect()`

`async invoke(function_id: str, data: dict)`

Best Practices

Next Steps