Python

The Python SDK, orbit-lang-sdk, runs ORBIT task handlers written in Python. It uses only the standard library, supports plain and async def handlers, runs a configurable number of tasks in parallel on threads, and handles registration, long polling, heartbeats, cancellation, result delivery through runtime outages, and graceful shutdown on SIGINT and SIGTERM. It also includes a small Client for starting workflows and waiting for their results. Read How workers work first for the concepts this page relies on.

Install

pip install orbit-lang-sdk

The package requires Python 3.9 or later and has no dependencies. The distribution is named orbit-lang-sdk; the import name is orbit_sdk.

A minimal worker

This worker implements text.normalize, the Python step of examples/polyglot.orbit:

task "text.normalize" { version "1.0.0" input { text: String } output { text: String } }

step normalize { worker "text.normalize" using python with { text: input.text } }

import re

from orbit_sdk import TaskError, Worker

worker = Worker("text-python")


@worker.task("text.normalize", version="1.0.0")
def normalize(input, ctx):
    text = input["text"]
    if not text.strip():
        raise TaskError("EMPTY_TEXT", "text must not be blank", retryable=False)
    ctx.log("normalizing", chars=len(text))
    text = re.sub(r"[^\w\s']", " ", text.lower())
    return {"text": re.sub(r"\s+", " ", text).strip()}


if __name__ == "__main__":
    worker.run()

Run it against a local runtime. The workflow has four steps in four languages, so start the other three example workers from the repository root as well:

orbit server &
python3 text_worker.py &
node examples/workers/polyglot/text_worker.ts &
(cd examples/workers/polyglot/go && go run .) &
cargo run -p orbit-worker --example text_worker &

orbit run examples/polyglot.orbit \
  --input '{"text": "The quick brown fox jumps over the lazy dog. The dog sleeps."}'

With blank text the handler raises a non-retryable error, so the step fails on its first attempt, the other steps are cancelled, and orbit run reports:

FAILED 4ea9f162-...  DocumentInsights@1.0.0
  compensation: none
  error: EMPTY_TEXT step `normalize` failed: text must not be blank

Registering tasks

Register each task with the @worker.task decorator or with worker.register. The decorator returns your function unchanged, so you can still call it directly in tests.

@worker.task("payments.charge", version="1.0.0")
def charge(input, ctx):
    ...

def refund(input, ctx):
    ...

worker.register("payments.refund", refund, version="1.0.0")

The name and version must match the task declaration in your .orbit files exactly. A worker registers each task name once; registering the same name again raises ValueError. Register every task before calling run().

Handlers and results

A handler takes two arguments, the task input as a dict and a Context, and returns a dict that becomes the step’s output:

@worker.task("inventory.reserve", version="1.0.0")
def reserve(input, ctx):
    return {"reservation_id": f"res-{input['order_id']}"}

• When the runtime sends no input, input is {}.
• Returning None reports {}.
• Returning anything other than a dict fails the attempt with INVALID_RESULT, which is not retried.
• The runtime validates the output against the contract’s output schema; a mismatch fails the step with OUTPUT_CONTRACT_VIOLATION.

async def handlers are supported. The SDK runs each one to completion with asyncio.run on the poll thread that received the task:

@worker.task("text.fetch", version="1.0.0")
async def fetch(input, ctx):
    body = await download(input["url"])
    return {"bytes": len(body)}

Raising task errors

Raise TaskError to fail an attempt with a structured error:

from orbit_sdk import TaskError

@worker.task("payments.charge", version="1.0.0")
def charge(input, ctx):
    if input["amount"] > 5000:
        raise TaskError("CARD_DECLINED", "limit exceeded", retryable=False,
                        details={"limit": 5000})
    ...

Any other exception is reported as a retryable UNHANDLED_EXCEPTION. The message is <ExceptionType>: <text>, and details.traceback holds the last frames of the traceback. The SDK also logs the traceback at ERROR level.

Context

The second handler argument describes the attempt being executed.

The context does not expose the contract version or timeout_ms.

Cancellation and timeouts

When an attempt times out, or its workflow is cancelled or fails, the runtime lists the attempt in the next heartbeat response and the SDK sets ctx.cancelled. The SDK does not interrupt your handler and does not act on deadline_ms by itself, so check the flag between units of work:

@worker.task("reports.render", version="1.0.0")
def render(input, ctx):
    pages = []
    for section in input["sections"]:
        if ctx.is_cancelled:
            raise TaskError("CANCELLED", "stopped by the runtime")
        pages.append(render_section(section, timeout=(ctx.remaining_ms() or 30000) / 1000))
    return {"pages": len(pages)}

ctx.cancelled.wait(seconds) is a cancellable sleep. Heartbeats run on a separate daemon thread at the interval the runtime requests, so a long-running handler keeps its session alive.

Concurrency

Worker(concurrency=N) starts N poll threads (default 4). Each thread runs one task at a time, so N is the maximum number of tasks this process runs at once. Synchronous handlers run on the poll thread; make them thread-safe if they share state. For CPU-bound work, run several worker processes rather than raising concurrency, because of the global interpreter lock.

Credentials and TLS

The worker reads its credentials from the environment:

The server and token arguments of Worker override the first three. The Worker constructor has no TLS arguments; it builds its Client from the environment. To configure TLS in code, replace worker.client before calling run(), and keep the timeout above the poll wait:

from orbit_sdk import Client, Worker

worker = Worker("payments-1")
worker.client = Client(
    server="https://orbit.internal:7233",
    ca_cert="ca.pem",
    client_cert="payments-1.pem",
    client_key="payments-1.key",
    timeout=worker.poll_wait_ms / 1000 + 15,
)

With a per-worker identity from a workers file, set the worker id to the identity’s id and the token to the identity’s token.

Logging

The SDK logs to the standard logging logger named orbit. Worker.run() calls logging.basicConfig with the level from ORBIT_LOG_LEVEL (default INFO); because basicConfig does nothing when logging is already configured, your own configuration takes precedence when you set it up before calling run(). At INFO the SDK logs registration, each task it executes, task failures, cancellations and ignored results. Runtime warnings, such as an unused contract version, are logged at WARNING.

Shutdown

worker.run() blocks until the process receives SIGINT or SIGTERM, or until worker.stop() is called from another thread. It then stops polling, lets running handlers finish and deliver their results, unregisters, and returns. A poll that is already waiting is not interrupted, so shutdown can take up to poll_wait_ms (20 seconds by default); lower it for faster shutdown.

run() installs signal handlers only when called from the main thread. Pass install_signal_handlers=False to keep your own. worker.start() runs the worker on a background daemon thread without signal handlers and returns the thread.

Starting workflows from Python

Client calls the runtime’s HTTP API. Starting executions requires the admin token or the token of a client identity granted start for the workflow. Pass it explicitly when the environment also holds a worker token; a client token is passed exactly like the admin token:

import os

from orbit_sdk import Client

client = Client(token=os.environ["ORBIT_ADMIN_TOKEN"])
execution_id = client.start("DocumentInsights", {"text": "Hello world"},
                            execution_id="doc-42")
result = client.wait(execution_id, timeout_ms=60000)
print(result["status"], result.get("output"))

Starting with an execution_id that already exists returns the existing execution instead of creating a duplicate.

Signals and queries

signal sends a typed signal declared with signal <name> { ... } in the workflow, and query evaluates a declared query. When authentication is enabled, both need the admin token, or a client token granted signal or query for the execution’s workflow, and wait needs read. A refused request raises ApiError with status 403 and code FORBIDDEN.

import os

from orbit_sdk import ApiError, Client

client = Client(token=os.environ["ORBIT_ADMIN_TOKEN"])

print(client.query("order-A-100", "status"))
# {'needs_approval': True, 'approved': None, 'reviewer': None, 'charged': False}

seq = client.signal("order-A-100", "review", {"approved": True, "reviewer": "dana"})

try:
    client.signal("order-A-100", "review", {"approved": "yes", "reviewer": "dana"})
except ApiError as e:
    print(e.code)  # SIGNAL_PAYLOAD_INVALID

signal returns the history sequence number of the recorded SignalReceived event. The runtime validates the payload and buffers the signal until an await step takes it. See Signals and queries.

API reference

Worker

Worker(worker_id=None, server=None, token=None, concurrency=4,
       runtime="python", poll_wait_ms=20000)

TaskError

TaskError(code, message="", retryable=True, details=None)

An Exception subclass with the attributes code, message, retryable and details. See Raising task errors.

Context

See Context. The SDK creates it; you do not construct one.

Client

Client(server=None, token=None, timeout=70.0, ca_cert=None,
       client_cert=None, client_key=None)

ApiError

Raised by Client for an error response or a connection failure. Attributes: status (int, 0 for a connection failure), code (the runtime’s error code, CONNECTION, or HTTP_ERROR), message, and the property is_connection_error.

Testing

Handlers are plain functions, so unit-test them without a runtime. A SimpleNamespace can stand in for the context:

from types import SimpleNamespace

from text_worker import normalize

def test_normalize():
    ctx = SimpleNamespace(log=lambda *a, **k: None, is_cancelled=False)
    assert normalize({"text": "Hello,  World!"}, ctx) == {"text": "hello world"}

For an integration test, start orbit server with a temporary database, deploy your workflow, run the worker on a background thread with worker.start(), and drive executions with Client:

worker = Worker("test-worker", poll_wait_ms=1000)
...
thread = worker.start()
client = Client(token="admin-token")
execution_id = client.start("DocumentInsights", {"text": "a b a"})
assert client.wait(execution_id)["status"] == "COMPLETED"
worker.stop()
thread.join()

A short poll_wait_ms makes stop() return quickly. The SDK’s own tests in sdks/python/tests use a scripted fake runtime built on http.server, which is a useful model for testing retries and session loss.

See also

• How workers work
• Worker protocol
• Task contracts
• Retries and timeouts
• Idempotency
• Security
• Polyglot workflows