Python SDKs — Ablative docs

Install

Use Python 3.10 or later and a virtual environment. Replace the checkout path below with your authorized copy. The worker installation builds its generated protocol package from that checkout.

python3 -m venv .venv
. .venv/bin/activate
python -m pip install -e /path/to/aion/sdks/python/aion-client
python -m pip install -e /path/to/aion/sdks/python/aion-worker

Start and read a workflow

Deploy the first workflow before running this caller. The local example uses an unauthenticated loopback server. Set auth and TLS for a remote deployment.

import asyncio
from aion_client import Client

async def main() -> None:
    client = await Client.connect(
        "127.0.0.1:50051",
        tls=False,
        namespace="default",
        stream_endpoint="ws://127.0.0.1:8080/events/stream",
    )
    async with client:
        started = await client.start(
            "greet_visitor",
            {"name": "Ada", "enthusiastic": True},
            idempotency_key="greeting-ada-001",
        )
        handle = started.handle
        print(handle.workflow_id, handle.run_id)
        print(await handle.describe(include_history=True))

asyncio.run(main())

Client.connect: gRPC endpoint; optional auth bearer, TLSConfig or tls=False for local development, namespace, subject and stream_endpoint. Close with async with or await close().
start: Typed JSON input or raw bytes with content_type. Supports idempotency_key, routing_key, task_queue and display_name. Returns StartOutcome.handle and display_name_not_applied.
signal / query: Signal sends JSON or raw payload to a named signal. Query decodes a named live query’s result. Current query validates then drops payload/raw arguments, so use no-argument queries through this client.
cancel / retire_workloop: Cancel targets a workflow/run and records a reason. retire_workloop requests the declared retirement body.
reopen / pause / resume: Return RunOutcome with run_id and status. Pause holds new activity dispatch durably; resume releases that hold.
list / describe / read_history: List requires sort and limit; replay next_cursor under identical filters. Describe can include history. read_history returns events, next_from_seq and head_seq.
WorkflowHandle: Binds client, namespace, workflow_id and optional run_id. latest_run() and specific_run(id) select a generation. Signal, query, cancel, describe, pause and resume operate on that handle.
subscribe: handle.subscribe(from_seq=1) returns an async event stream. Configure stream_endpoint. Reconnection resumes at last delivered per-workflow sequence plus one.

page = await client.list(
    sort={"field": "started_at", "direction": "desc"},
    limit=20,
)
print(page["items"], page.get("next_cursor"))
history = await client.read_history(
    handle.workflow_id, from_seq=1, limit=100,
)
print(history.events, history.next_from_seq, history.head_seq)

Reusing an idempotency key with the same start returns the original run. A different workflow, input or routing choice raises a conflict. A requested display-name change on replay is reported separately.

Serve an activity

This complete handler receives one record and returns another. Deploy a workflow whose action name and contract agree with it; the greeting tutorial itself uses pure workflow steps.

from dataclasses import dataclass
from aion_worker import activity

@dataclass
class Greeting:
    name: str

@dataclass
class Greeted:
    message: str

@activity(name="examples.greet")
def greet(input: Greeting) -> Greeted:
    return Greeted(message=f"Hello, {input.name}!")

python -m aion_worker greet.py --check
python -m aion_worker greet.py \
  --endpoint 127.0.0.1:50051 \
  --task-queue default --identity python-worker-1 --concurrency 4 \
  --reconnect-initial-backoff 0.1 --reconnect-max-backoff 5 \
  --reconnect-max-attempts 10

activity: Decorator registers a named sync or async handler. Typed dataclass input decodes the action’s named JSON fields. Return annotations derive output schemas.
WorkerConfig: Requires endpoint, task_queue, identity, max_concurrency and ReconnectConfig. namespaces defaults to ("default",); node defaults to the hostname. Set subject and transport credentials for the deployment.
ReconnectConfig: Set initial_backoff_seconds, max_backoff_seconds and max_attempts. Optional connect_timeout_seconds and registration_timeout_seconds bound establishment.
Worker: Uses the decorated registry by default. Worker(config, registry=...) or Worker(config, activities=...) selects handlers explicitly. await run(shutdown=event) serves until shutdown or a terminal connection failure.
ActivityContext: Exposes workflow_id, run_id, activity_id, attempt and idempotency_key. heartbeat(detail) reports progress. is_cancelled() reads the flag; await cancelled() waits for it.
Failures: Raise RetryableError, TerminalError or PolicyRefusedError to classify failure. Input decoding and output encoding failures name the type boundary.
Contracts: All advertised actions need concrete schemas on a contract-bearing queue. An Any return is refused locally. Return records as dataclasses so their fields are advertised and encoded.

Handle errors

Caller errors include InvalidArgument, AlreadyExists, NotFound, Unauthenticated, NamespaceDenied, QueryFailed, QueryTimeout, Unavailable and ServerError. Handle the named condition at the application boundary. A query deadline bounds the client wait.

The implementation lives in the two packages’ public modules under sdks/python. Their client, handle, context and worker modules define the operations above. Read worker admission before commissioning a queue.