Workers

How workers expand iii

Workers add capability to an iii system. Each one contributes functions the engine can route to. This page covers deploying and wiring workers into a project. Once connected, a worker exposes:

Functions, callable by function_id from anywhere in the system (see Using iii / Functions).
The same functions as event sources: any worker can observe them, so its own functions run each time they run (see Using iii / Observers).

For the full SDK surface each worker can use when interacting with iii, see the complete SDK reference by language: Node, Python, Rust, Go, or Browser.

Create a new worker

A worker is any process that installs an iii SDK, connects to an engine, and registers functions or observers. Create a normal TypeScript, JavaScript, Python, Rust, or Go project using that language’s package tools, then add the corresponding iii SDK and an entrypoint such as src/index.ts, src/main.py, or main.go. To let Compose start a local worker, add an iii.worker.yaml at the project root:

iii.worker.yaml

name: my-worker
description: One-line summary of what this worker does.
scripts:
  start: pnpm start

Declare the directory in worker-compose.yaml, or add it through a running daemon with iii call -n dev compose::add worker=./workers/my-worker. See Declare workers with Compose for path:// workers and Add a registry worker for registry packages.

The engine and SDK packages can have different patch versions within the same minor line. Keep the engine and SDKs on the same minor version, for example 0.11.x, unless a release note says otherwise.

Connecting to the engine

A worker connects to the engine over WebSocket. The convention is to set the engine URL via the III_URL environment variable, but it can also be passed explicitly to registerWorker. The connection string is the only coupling between a worker and the iii instance it joins, so the worker process can be deployed anywhere reachable on the network.

This connects with full trust, appropriate for workers you run. For an untrusted worker (a browser client or a third party’s), connect through the iii-worker-manager RBAC listener and gate it with an auth function instead. See Access Control and the iii-worker-manager worker page.

Node / TypeScript
Python
Rust
Go

import { registerWorker } from "iii-sdk";

const url = process.env.III_URL;
if (!url) throw new Error("III_URL must be set");
const worker = registerWorker(url, {
  workerName: "math-worker",
  workerDescription: "Evaluate math expressions over iii functions",
  namespace: "math", // scopes this worker's registrations
});

import os

from iii import InitOptions, register_worker

url = os.environ.get("III_URL")
if url is None:
    raise RuntimeError("III_URL must be set")

worker = register_worker(
    url,
    InitOptions(
        worker_name="math-worker",
        worker_description="Evaluate math expressions over iii functions",
        namespace="math",  # scopes this worker's registrations
    ),
)

use iii_sdk::runtime::WorkerMetadata;
use iii_sdk::{InitOptions, register_worker};

fn main() {
    let url = std::env::var("III_URL").expect("III_URL must be set");
    let _worker = register_worker(
        &url,
        InitOptions {
            metadata: Some(WorkerMetadata {
                name: "math-worker".into(),
                description: Some("Evaluate math expressions over iii functions".into()),
                ..Default::default()
            }),
            namespace: Some("math".into()), // scopes this worker's registrations
            ..Default::default()
        },
    );
}

package main

import (
    "os"

    iii "github.com/iii-hq/iii/sdk/packages/go/iii"
)

func main() {
    client := iii.RegisterWorker(os.Getenv("III_URL"),
        iii.WithName("math-worker"),
        iii.WithDescription("Evaluate math expressions over iii functions"),
        iii.WithNamespace("math"), // scopes this worker's registrations
    )

    defer client.Close()

    // Register functions on client: they are resent on each connection.
    select {
    }
}

namespace scopes everything this worker registers, so an identically-named worker or function id can coexist in another namespace. The SDK reads it from the option, then from the III_NAMESPACE environment variable, then falls back to default. With III_NAMESPACE set in the worker’s environment, these two are equivalent:

Node / TypeScript
Python
Rust
Go

import { registerWorker } from "iii-sdk";

const url = process.env.III_URL ?? "ws://127.0.0.1:49134";

registerWorker(url, { namespace: process.env.III_NAMESPACE });
registerWorker(url);

import os

from iii import InitOptions, register_worker

url = os.environ.get("III_URL", "ws://127.0.0.1:49134")

register_worker(url, InitOptions(namespace=os.environ.get("III_NAMESPACE")))
register_worker(url)

use iii_sdk::{InitOptions, register_worker};

fn main() {
    let url = "ws://127.0.0.1:49134";

    let explicit = register_worker(
        url,
        InitOptions {
            namespace: std::env::var("III_NAMESPACE").ok(),
            ..Default::default()
        },
    );
    let from_env = register_worker(url, InitOptions::default());

    drop((explicit, from_env));
}

package main

import (
    "os"

    iii "github.com/iii-hq/iii/sdk/packages/go/iii"
)

func main() {
    url := os.Getenv("III_URL")

    // WithNamespace refuses a blank name, so the first form needs III_NAMESPACE set.
    iii.RegisterWorker(url, iii.WithNamespace(os.Getenv("III_NAMESPACE")))
    iii.RegisterWorker(url)
}

Omit the option to serve many tenants from one worker package: set III_NAMESPACE per deployment, and each deployment of the same image registers in its own namespace.

For calling across namespaces and handling a rejected registration, see Use namespaces.

The browser SDK has no environment to read, so it takes the option only.

Worker lifecycle

States

Workers transition through a small set of states after connecting: connected → available / busy → disconnected. connected means the worker has joined the engine’s registry. available and busy describe whether the worker is currently handling invocations. disconnected is the terminal state when the WebSocket closes. The engine tracks these transitions and surfaces them to other workers and tooling through its discovery functions, so the rest of the system can react. The WebSocket handshake that precedes connected is not one of these states.

Inspecting the live registry

To see what’s currently connected to the engine, invoke one of the engine::*::list functions to get the current state of the registry. Each returns a list:

Function	What it returns
`engine::workers::list`	Every connected worker with metrics.
`engine::functions::list`	Every registered function, with its `args_format`, `private`, `observable` and `observer_count`. Filterable by `include_internal`.
`engine::observers::list`	Every observer, with its observed function, target, and state.

List workers, functions and observers

Node / TypeScript
Python
Rust
Go

import { type Observed, registerWorker } from "iii-sdk";

const worker = registerWorker(process.env.III_URL ?? "ws://127.0.0.1:49134");

// engine::workers::list, pass { worker_id: "<uuid>" } to look up one worker
const { workers } = await worker.call({
  function_id: "engine::workers::list",
  payload: {},
});

// engine::functions::list
const { functions } = await worker.call({
  function_id: "engine::functions::list",
  payload: { include_internal: false },
});

// engine::observers::list
const { observers } = await worker.call({
  function_id: "engine::observers::list",
  payload: {},
});

console.log(workers, functions, observers);

import os

from iii import register_worker

worker = register_worker(os.environ.get("III_URL", "ws://127.0.0.1:49134"))

# engine::workers::list, pass {"worker_id": "<uuid>"} to look up one worker
workers = worker.call({
    "function_id": "engine::workers::list",
    "payload": {},
})["workers"]

# engine::functions::list
functions = worker.call({
    "function_id": "engine::functions::list",
    "payload": {"include_internal": False},
})["functions"]

# engine::observers::list
observers = worker.call({
    "function_id": "engine::observers::list",
    "payload": {},
})["observers"]

print(workers, functions, observers)

use iii_sdk::protocol::CallRequest;
use iii_sdk::{InitOptions, register_worker};
use serde_json::json;

#[tokio::main]
async fn main() -> Result<(), iii_sdk::Error> {
    let worker = register_worker("ws://127.0.0.1:49134", InitOptions::default());

    // engine::workers::list, pass json!({ "worker_id": "<uuid>" }) to look up one worker
    let workers = worker
        .call(CallRequest::new("engine::workers::list", json!({})))
        .await?;

    // engine::functions::list
    let functions = worker
        .call(CallRequest::new(
            "engine::functions::list",
            json!({ "include_internal": false }),
        ))
        .await?;

    // engine::observers::list
    let observers = worker
        .call(CallRequest::new("engine::observers::list", json!({})))
        .await?;

    println!("{workers}\n{functions}\n{observers}");
    Ok(())
}

package main

import (
    "context"
    "encoding/json"
    "log"
    "os"

    iii "github.com/iii-hq/iii/sdk/packages/go/iii"
)

func main() {
    client := iii.RegisterWorker(os.Getenv("III_URL"))
    ctx := context.Background()

    // engine::workers::list, pass {"worker_id":"<uuid>"} to look up one worker
    workers, err := client.Call(ctx, iii.CallRequest{
        FunctionID: "engine::workers::list",
        Payload:    json.RawMessage(`{}`),
    })
    if err != nil {
        log.Fatal(err)
    }

    // engine::functions::list
    functions, err := client.Call(ctx, iii.CallRequest{
        FunctionID: "engine::functions::list",
        Payload:    json.RawMessage(`{"include_internal":false}`),
    })
    if err != nil {
        log.Fatal(err)
    }

    // engine::observers::list
    observers, err := client.Call(ctx, iii.CallRequest{
        FunctionID: "engine::observers::list",
        Payload:    json.RawMessage(`{}`),
    })
    if err != nil {
        log.Fatal(err)
    }

    log.Printf("%s\n%s\n%s", workers, functions, observers)
}

Handling worker disconnects

When a worker’s WebSocket closes, the engine cleans up after it automatically. Its functions leave the live registry, observers of those functions go pending until a worker registers them again, and any in-flight invocations of those functions are cancelled. The worker’s own observers go suspended for observers.observer_grace_ms (30 s by default): if it reconnects in that time it keeps them, and otherwise they are disposed. When it reconnects later, its SDK observes again, under the same id, each of them that has neither max_fires nor expires_in_ms (see Observer states).

In flight requests

In flight requests will get a invocation_stopped error, catch these errors and treat them like a cancellation. Retrying will fail until the worker that owns this function reconnects.

Handle a stopped invocation

Node / TypeScript
Python
Rust
Go

import { type Observed, registerWorker } from "iii-sdk";
import { InvocationError } from "iii-sdk/errors";

type Worker = ReturnType<typeof registerWorker>;

async function addNumbers(worker: Worker): Promise<void> {
  try {
    const result = await worker.call({
      function_id: "math::add",
      payload: { a: 1, b: 2 },
    });
    console.log(result);
  } catch (err) {
    if (err instanceof InvocationError && err.code === "invocation_stopped") {
      // The worker hosting math::add disconnected mid-invocation. Observe
      // `engine::on-functions-changed` (see "Observe registry changes" below)
      // to know when to retry.
      return;
    }
    throw err;
  }
}

from iii.errors import InvocationError


def add_numbers(worker) -> None:
    try:
        result = worker.call({
            "function_id": "math::add",
            "payload": {"a": 1, "b": 2},
        })
    except InvocationError as err:
        if err.code == "invocation_stopped":
            # The worker hosting math::add disconnected mid-invocation. Observe
            # `engine::on-functions-changed` (see "Observe registry changes" below)
            # to know when to retry.
            return
        raise
    print(result)

use iii_sdk::protocol::CallRequest;
use iii_sdk::{Error, IIIClient};
use serde_json::json;

async fn add_numbers(worker: &IIIClient) -> Result<(), Error> {
    match worker
        .call(CallRequest::new("math::add", json!({ "a": 1, "b": 2 })))
        .await
    {
        Ok(value) => println!("{value}"),
        Err(Error::Remote { code, .. }) if code == "invocation_stopped" => {
            // The worker hosting math::add disconnected mid-invocation. Observe
            // `engine::on-functions-changed` (see "Observe registry changes" below)
            // to know when to retry.
        }
        Err(err) => return Err(err),
    }
    Ok(())
}

package mathworker

import (
    "context"
    "encoding/json"
    "errors"
    "log"

    iii "github.com/iii-hq/iii/sdk/packages/go/iii"
)

func addNumbers(ctx context.Context, client *iii.Client) error {
    result, err := client.Call(ctx, iii.CallRequest{
        FunctionID: "math::add",
        Payload:    json.RawMessage(`{"a":1,"b":2}`),
    })
    var ie *iii.InvocationError
    if errors.As(err, &ie) && ie.Code == "invocation_stopped" {
        // The worker hosting math::add disconnected mid-invocation. Observe
        // `engine::on-functions-changed` (see "Observe registry changes" below)
        // to know when to retry.
        return nil
    }
    if err != nil {
        return err
    }
    log.Printf("math::add returned %s", result)
    return nil
}

Observe registry changes

The engine hosts three functions that run when the registry changes. Observe them to react to topology changes as they happen, for example to continue work when a worker comes back online.

Function	Runs when	Event (`input`)
`engine::on-workers-changed`	A worker connects, disconnects, or updates its metadata.	`{ event, worker_id }`
`engine::on-functions-changed`	A function is registered or unregistered.	`{ event, functions }`, the full list after the change
`engine::on-observers-changed`	An observer changes state (for example `pending` to `active`, or removed).	`{ observer_id, function_id, target, state, reason? }`

event on engine::on-workers-changed is worker_connected, worker_disconnected, or worker_metadata_updated. With capture: ["input"], the target receives the event as the payload’s input. See Engine-hosted event sources.

React to registry changes

Node / TypeScript
Python
Rust
Go

import { type Observed, registerWorker } from "iii-sdk";

const worker = registerWorker(process.env.III_URL ?? "ws://127.0.0.1:49134");

worker.registerFunction(
  "math::refresh-workers",
  async ({ input }: Observed<{ event: string; worker_id: string }>) => {
    if (input?.event === "worker_connected") {
      // The worker joined the registry; its functions are callable now.
    }
  },
);
worker.observe(
  { function_id: "engine::on-workers-changed", capture: ["input"] },
  "math::refresh-workers",
);

worker.registerFunction(
  "math::refresh-functions",
  async ({ input }: Observed<{ event: string; functions: { function_id: string }[] }>) => {
    // `functions` is the full snapshot after the change.
    const ids = (input?.functions ?? []).map((f) => f.function_id);
    console.log(ids);
  },
);
worker.observe(
  { function_id: "engine::on-functions-changed", capture: ["input"] },
  "math::refresh-functions",
);

import os

from iii import register_worker

worker = register_worker(os.environ.get("III_URL", "ws://127.0.0.1:49134"))


async def refresh_workers(observed: dict) -> None:
    if observed["input"]["event"] == "worker_connected":
        # The worker joined the registry; its functions are callable now.
        pass


worker.register_function("math::refresh-workers", refresh_workers)
worker.observe(
    {"function_id": "engine::on-workers-changed", "capture": ["input"]},
    "math::refresh-workers",
)


async def refresh_functions(observed: dict) -> None:
    # `functions` is the full snapshot after the change.
    ids = [f["function_id"] for f in observed["input"].get("functions", [])]
    print(ids)


worker.register_function("math::refresh-functions", refresh_functions)
worker.observe(
    {"function_id": "engine::on-functions-changed", "capture": ["input"]},
    "math::refresh-functions",
)

use iii_sdk::{Capture, IIIClient, Observed, RegisterFunction, Source};
use schemars::JsonSchema;
use serde::Deserialize;
use serde_json::Value;

#[derive(Deserialize, JsonSchema)]
struct WorkersChanged {
    event: String,
    worker_id: String,
}

#[derive(Deserialize, JsonSchema)]
struct FunctionsChanged {
    event: String,
    functions: Vec<Value>,
}

async fn wire(worker: &IIIClient) -> Result<(), iii_sdk::Error> {
    worker.register_function(
        "math::refresh-workers",
        RegisterFunction::new_async(|observed: Observed<WorkersChanged>| async move {
            if observed.input.is_some_and(|input| input.event == "worker_connected") {
                // The worker joined the registry; its functions are callable now.
            }
            Ok::<_, iii_sdk::Error>(())
        }),
    );
    worker.observe(
        Source::function("engine::on-workers-changed").capture([Capture::Input]),
        "math::refresh-workers",
    );

    worker.register_function(
        "math::refresh-functions",
        RegisterFunction::new_async(|observed: Observed<FunctionsChanged>| async move {
            // `functions` is the full snapshot after the change.
            let count = observed.input.map_or(0, |input| input.functions.len());
            println!("{count} functions");
            Ok::<_, iii_sdk::Error>(())
        }),
    );
    worker.observe(
        Source::function("engine::on-functions-changed").capture([Capture::Input]),
        "math::refresh-functions",
    );

    Ok(())
}

package main

import (
    "context"
    "encoding/json"
    "log"
    "os"

    iii "github.com/iii-hq/iii/sdk/packages/go/iii"
)

func main() {
    client := iii.RegisterWorker(os.Getenv("III_URL"))

    if _, err := client.RegisterFunction("math::refresh-workers", func(ctx context.Context, data json.RawMessage) (any, error) {
        var observed struct {
            Input struct {
                Event    string `json:"event"`
                WorkerID string `json:"worker_id"`
            } `json:"input"`
        }
        if err := json.Unmarshal(data, &observed); err != nil {
            return nil, err
        }
        event := observed.Input
        if event.Event == "worker_connected" {
            // The worker joined the registry; its functions are callable now.
            log.Printf("worker %s connected", event.WorkerID)
        }
        return nil, nil
    }); err != nil {
        log.Fatal(err)
    }
    if _, err := client.Observe(iii.Source{
        FunctionID: "engine::on-workers-changed",
        Capture:    []iii.Capture{iii.CaptureInput},
    }, "math::refresh-workers"); err != nil {
        log.Fatal(err)
    }

    if _, err := client.RegisterFunction("math::refresh-functions", func(ctx context.Context, data json.RawMessage) (any, error) {
        var observed struct {
            Input struct {
                Event     string `json:"event"`
                Functions []struct {
                    FunctionID string `json:"function_id"`
                } `json:"functions"`
            } `json:"input"`
        }
        if err := json.Unmarshal(data, &observed); err != nil {
            return nil, err
        }
        event := observed.Input
        // `functions` is the full snapshot after the change.
        ids := make([]string, 0, len(event.Functions))
        for _, f := range event.Functions {
            ids = append(ids, f.FunctionID)
        }
        log.Printf("functions: %v", ids)
        return nil, nil
    }); err != nil {
        log.Fatal(err)
    }
    if _, err := client.Observe(iii.Source{
        FunctionID: "engine::on-functions-changed",
        Capture:    []iii.Capture{iii.CaptureInput},
    }, "math::refresh-functions"); err != nil {
        log.Fatal(err)
    }
}

Worker manifest

iii.worker.yaml is the manifest at the worker’s root that tells Compose how to start a local or bundled worker. A local Compose container can replace the start command with its own scripts.run, and add scripts.pre_run or scripts.post_run lifecycle hooks. The container cannot change the manifest’s name or runtime.base_image; those stay as the manifest declares them.

name: math-worker
description: Evaluate math expressions over iii functions.
runtime:
  # Base OCI image used as the worker rootfs. Override to pin a version.
  base_image: docker.io/iiidev/python:latest
scripts:
  start: "watchfiles 'python src/math_worker.py'"

description is an optional one-line, human/LLM-readable summary of what the worker does. scripts.start launches the worker. Here, watchfiles reloads it whenever you edit a source file. runtime.base_image selects the OCI image used for a bundled worker’s root filesystem. The manifest is metadata about starting the worker. See Worker manifest for the field reference.

If a worker isn’t starting correctly, check its manifest, the Compose daemon output, and iii call -n dev compose::status.

iii does not distinguish a Compose-managed process from a manually run process that uses the iii SDK. Both connect to the engine and register the same way.

Shutting down a worker

Call the SDK’s shutdown to close the WebSocket cleanly. The engine then applies the cleanup described in Handling worker disconnects at once, instead of waiting for the dropped socket. Without shutdown, an abrupt process exit reaches the same state once the engine notices the dropped socket. Graceful shutdown makes it deterministic and faster.

Node / TypeScript
Python
Rust
Go

import { registerWorker } from "iii-sdk";

const worker = registerWorker(process.env.III_URL ?? "ws://127.0.0.1:49134");

process.on("SIGTERM", async () => {
  await worker.shutdown();
  process.exit(0);
});

import os
import signal

from iii import InitOptions, register_worker

worker = register_worker(
    os.environ.get("III_URL", "ws://127.0.0.1:49134"),
    InitOptions(worker_name="math-worker"),
)


def _on_term(*_):
    worker.shutdown()
    raise SystemExit(0)


signal.signal(signal.SIGTERM, _on_term)

use iii_sdk::{InitOptions, register_worker};
use tokio::signal;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let worker = register_worker("ws://127.0.0.1:49134", InitOptions::default());

    // Rust threads do not keep the process alive on their own; await this
    // before `main` returns so the connection thread exits cleanly.
    signal::unix::signal(signal::unix::SignalKind::terminate())?.recv().await;
    worker.shutdown_async().await;
    Ok(())
}

package main

import (
    "context"
    "log"
    "os"
    "os/signal"
    "syscall"
    "time"

    iii "github.com/iii-hq/iii/sdk/packages/go/iii"
)

func main() {
    client := iii.RegisterWorker(os.Getenv("III_URL"))

    // Block until SIGTERM, then shut the worker down gracefully.
    ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM)
    defer stop()
    <-ctx.Done()

    shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()
    if err := client.Shutdown(shutdownCtx); err != nil {
        log.Printf("shutdown cut short: %v", err)
    }
}

Shutdown is useful for one-shot and ephemeral workers. Kubernetes Jobs, serverless containers, and scheduled scripts connect like any other worker, do their work, and call shutdown().

Getting Started

Using iii

Understanding iii

Creating Workers

Upgrading

How workers expand iii

Create a new worker

Connecting to the engine

Worker lifecycle

States

Inspecting the live registry

Handling worker disconnects

In flight requests

Observe registry changes

Worker manifest

Shutting down a worker

​How workers expand iii

​Create a new worker

​Connecting to the engine

​Worker lifecycle

​States

​Inspecting the live registry

​Handling worker disconnects

​In flight requests

​Observe registry changes

​Worker manifest

​Shutting down a worker

How workers expand iii

Create a new worker

Connecting to the engine

Worker lifecycle

States

Inspecting the live registry

Handling worker disconnects

In flight requests

Observe registry changes

Worker manifest

Shutting down a worker