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.trigger({"function_id": "hello::greet", "payload": {"name": "world"}})
print(result)  # {"message": "Hello, world!"}

API

Operation	Signature	Description
Initialize	`register_worker(url, options?)`	Create an SDK instance and auto-connect
Register function	`iii.register_function(id, handler)`	Register a function that can be invoked by name
Invoke (await result)	`iii.trigger({"function_id": id, "payload": data})`	Invoke a function and wait for the result
Invoke (fire-and-forget)	`iii.trigger({"function_id": id, ..., "action": TriggerAction.Void()})`	Fire-and-forget
Invoke (enqueue)	`iii.trigger({"function_id": id, ..., "action": TriggerAction.Enqueue(queue="name")})`	Route invocation through a named queue
Observe	`iii.observe({"function_id": ..., "args": ..., "filters": ...}, target, options?)`	Invoke `target` on every matching event of a function
Current invocation	`iii.current_invocation()`	The invocation the running handler serves
Shutdown	`iii.shutdown()`	Disconnect and stop background thread

register_worker() creates the SDK instance and auto-connects to the engine.

Registering Functions

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

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

Invoking Functions

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

trigger() blocks the calling thread; await iii.trigger_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 the calling thread, 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

Every function is observable. The target is a registered function; it receives the Observed envelope (input, output or error, caller, ...).

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()

A function can hand its own events to the observers they belong to (owner delivery), and react when one is added or removed:

from iii import ObserveHooks, Observer

schedules: dict[str, set[Observer]] = {}  # expression -> observers

class CronHooks(ObserveHooks):
    delivery = "owner"
    args_format = {"type": "object", "required": ["expression"],
                   "properties": {"expression": {"type": "string"}}}

    def on_observer_added(self, obs):
        schedules.setdefault(obs.args["expression"], set()).add(obs)

    def on_observer_removed(self, obs):
        schedules[obs.args["expression"]].discard(obs)

def on_tick(tick):
    for obs in schedules.get(tick["expression"], ()):
        obs.emit(tick)  # the observer's filters are checked here

iii.register_function("cron::on-tick", on_tick, observe=CronHooks())

Inside any handler, iii.current_invocation() returns the running invocation (function_id, invocation_id, caller, activation, observe_depth). Sync handlers 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.

Trigger APIs (deprecated)

register_trigger and register_trigger_type keep working in 0.25 but are deprecated: they emit a DeprecationWarning and log a warning once per process. Use iii.observe(source, target) and register_function(id, handler, observe=ObserveHooks(...)) instead. See the migration guide.

Development

Install in development mode

pip install -e .

Type checking

mypy src

Linting

ruff check src

Resources

Metadata

Release files for iii-sdk 0.25.0

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for iii-sdk 0.25.0
File	Size	Uploaded	Provenance
iii_sdk-0.25.0.tar.gz	312.9 kB	Oct 22, 2026	Without provenance	Details

Built distribution (wheel)

Table of built distributions (wheels) for iii-sdk 0.25.0
File	Interpreter	ABI	Platform	Reset
iii_sdk-0.25.0-py3-none-any.whl 102.2 kB Oct 22, 2026 Without provenance	Python 3	none	any	Details

Total release size: 415.1 kB

Release files / iii_sdk-0.25.0.tar.gz

Download URL	iii_sdk-0.25.0.tar.gz
Size	312.9 kB
Tags	Source
SHA-256 checksum How to use checksums	`a69857b1e730dddd0ab9f288c7f6e63f255e66eb722a87de9d8692c56e6b8653`
BLAKE2b-256 checksum How to use checksums	`5e2155c193cf4654912c2d7c7a349aec0a46dddd0c8b293b1ab914e51abd68ce`
Upload date	Oct 22, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/7.0.0 CPython/3.13.16`

Release files / iii_sdk-0.25.0-py3-none-any.whl

Download URL	iii_sdk-0.25.0-py3-none-any.whl
Size	102.2 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`ba8943ddd48184464fc12535097897af6932777e97ccfd5bbe543ae554a6a0b5`
BLAKE2b-256 checksum How to use checksums	`641f19922a843924c47eb3f2c785bb1bdccd3a2dc09f79d9c58b61ef31c68b84`
Upload date	Oct 22, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/7.0.0 CPython/3.13.16`

Security for iii-sdk 0.25.0

Report project as malware

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

No provenance found

None of the files in this release were uploaded with provenance metadata.

A note on package security

PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.