iii-sdk

Python SDK for the iii engine.

Install

pip install iii-sdk

Hello World

from iii import register_worker

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

def greet(data):
    return {"message": f"Hello, {data['name']}!"}

iii.register_function("hello::greet", greet)

result = iii.call({"function_id": "hello::greet", "payload": {"name": "world"}})
print(result)  # {"message": "Hello, world!"}

API

register_worker() creates the SDK instance and auto-connects to the engine. trigger and trigger_async are deprecated aliases of call and call_async.

Registering Functions

def create_order(data):
    return {"status_code": 201, "body": {"id": "123", "item": data["body"]["item"]}}

iii.register_function("orders::create", create_order)

The handler's return value is what the function's observers observe. Options: description, metadata, request_format, response_format (extracted from type hints when omitted), args_format (a JSON Schema or Pydantic model of the observers' args), private=True (only this worker may call it; its own calls pass the input and result by reference), observable=False (nobody may observe it), standby=True (take the function over when the worker that hosts it leaves) and local_dispatch.

Calling Functions

result = iii.call({"function_id": "orders::create", "payload": {"body": {"item": "widget"}}})

call() blocks the calling thread; await iii.call_async(...) works from any event loop. A call to a function this worker registered itself runs in-process once the engine confirmed it (no frames on the wire unless someone observes it): a sync handler on a thread of its own, an async one on the calling loop. Pass "dispatch": "remote" to go through the engine anyway, or register the function with local_dispatch=False.

Observing Functions

When orders::create runs, alerts::big-order runs:

iii.observe("orders::create", "alerts::big-order")

The target is a registered function; it receives the Observed envelope (input, output or error, caller, ...). The full form takes args (read by the worker that hosts the function), filters, on, capture and deliver, and a policy:

def big_order(envelope):
    notify(envelope["input"], envelope["output"])

iii.register_function("alerts::big-order", big_order)

observer = iii.observe(
    {"function_id": "orders::create", "filters": {"input.body.total": {"$gte": 500}}},
    "alerts::big-order",
    {"max_in_flight": 4},
)
observer.ready()            # or: await observer.ready_async()
# later
observer.dispose()

Hosting Observable Functions

The worker that hosts a function decides what observing it means with four hooks. Each one overrides something iii does by default:

on_observer_added and on_observer_removed can be set for one function or for every function the worker hosts; both levels run, the worker's first when an observer is added and last when it is removed. on_observer_match and on_observer_deliver are set per function.

The methods work as decorators and as plain calls, on the client by function id or on the FunctionRef that register_function returns. Set hooks by id before registering the function so they run for every observer.

from iii import Call, Observer, ObserveRejection

@iii.on_observer_added("cron::on-tick")
def start(obs: Observer) -> None:
    if not scheduler.is_valid(obs.args["expression"]):
        raise ObserveRejection("INVALID_CRON_EXPRESSION", "invalid expression")
    scheduler.retain(obs.args["expression"])

@iii.on_observer_removed("cron::on-tick")
def stop(obs: Observer) -> None:
    scheduler.release(obs.args["expression"])

# an observer of expression X gets the ticks of expression X
@iii.on_observer_match("cron::on-tick")
def same_expression(obs: Observer, call: Call) -> bool:
    return obs.args["expression"] == call.input["expression"]

iii.register_function(
    "cron::on-tick",
    lambda tick: tick,  # an event source returns the event it is called with
    args_format={"type": "object", "required": ["expression"],
                 "properties": {"expression": {"type": "string"}}},
    private=True,
)

# a tick is a call of the function
iii.call({"function_id": "cron::on-tick", "payload": {"expression": "*/5 * * * * *", "scheduled_time": now}})

on_observer_match takes a predicate (obs, call) -> bool, asked once per active observer per call, or observer= and call= key functions for a function with many observers. An observer wants a call when they share a key; the SDK computes an observer's keys once, when it is accepted, and keeps an index, so a call costs one lookup per key:

iii.on_observer_match(
    "database::on-row-changed",
    observer=lambda obs: f"{obs.args['table']}|{obs.args['op']}",  # what this observer listens for
    call=lambda call: f"{call.input['table']}|{call.input['op']}",  # what this call is
)

iii's delivery applies the on and filters of the observers that matched. The engine evaluates them for every call it routes: it holds the call's input, caller and metadata, and takes its output or error from this worker's answer. For those calls this worker runs only the match and deliver hooks, and its answer names the observers the match hook matched. The engine never sees the calls this worker makes to its own functions, so the SDK evaluates on and filters for those, through an index of the filters' $eq and $in conditions: with 1,000 observers that each filter on their own input.tenant, a local call evaluates one filter. obs.matches(call) evaluates one observer's on and filters in this worker.

iii.is_observed(function_id, call) tells a source whether a call would reach anyone (on a function matched by keys, by looking up the call's keys), and iii.observers(function_id) lists the active observers.

A function whose observers' answers matter, or whose calls each go to one observer, delivers its calls itself. Its on_observer_deliver receives the Call and the observers that matched it, calls through an observer, and returns what the caller gets back. iii delivers nothing for that function:

@iii.on_observer_deliver("http::on-request")
async def route(call, observers):
    obs = routes.match(call.input)
    if obs is None:
        return {"status_code": 404}
    if not obs.matches(call):
        return {"status_code": 422}
    return await iii.call_async({"observer": obs, "payload": call.input})

To send one observer something of its own, such as its initial state, call the function with "metadata": {"for_observer": obs.id} once the observer is accepted, and let the match hook give that call to that observer alone. iii delivers it like every call, so the target receives it in the shape it receives every call of the function:

import asyncio

from iii import CallAction, ObserveRejection


async def send_state(obs):
    try:
        await obs.accepted_async()
    except ObserveRejection:
        return  # a rejected observer gets nothing
    await iii.call_async({
        "function_id": "tmux::on-changed",
        "payload": state_for(obs.args),
        "metadata": {"for_observer": obs.id},
        "action": CallAction.Void(),
    })


@iii.on_observer_added("tmux::on-changed")
async def on_added(obs):
    asyncio.get_running_loop().create_task(send_state(obs))


@iii.on_observer_match("tmux::on-changed")
def wants(obs, call):
    for_observer = (call.metadata or {}).get("for_observer")
    if for_observer:
        return for_observer == obs.id
    return obs.args["session"] == call.input["session"]

With keys, file each observer under f"for_observer:{obs.id}" as well, and give such a call that one key.

Inside any handler, iii.current_invocation() returns the running invocation (function_id, invocation_id, caller, activation, observe_depth). Sync handlers and hooks run on a thread of their own started in a copy of the caller's context, so the current invocation and the OpenTelemetry context follow them.

A function that sets args_format, private=True, observable=False, an on_observer_match hook or an on_observer_deliver hook needs an engine that advertises observe.v2; against an older engine it stays unregistered, and the SDK logs UNSUPPORTED once for it.

Development

Install in development mode

pip install -e .

Type checking

mypy src

Linting

ruff check src

Resources

• Documentation
• iii Engine
• Examples