aga-runtime

Aga — durable execution for Python

Write typed functions that survive crashes, restarts, deploys, and long waits. One App owns declarations, connection, Run creation, and worker lifecycle. Aga records each durable boundary and reuses its committed result on recovery.

Source-breaking 0.4.2 release. App is now the only home for durable operations. Package-level join, sleep, signal, event, cancel, info, and deadline are removed rather than retained as aliases.

Calls made before a join can overlap in sticky and distributed workflows; caller waiting does not change placement. concurrency sizes separate bounded claim and sticky-function pools per worker (default four). Immediate awaits remain sequential. Checkpoints are written by the workflow thread; only committed results survive crashes. See the concurrent-task walkthrough. Server 0.2.1 or later can inspect measured sticky function times in the dashboard.

Install

Install the SDK with:

python -m pip install --upgrade "aga-runtime==0.4.2"

The distribution is aga-runtime; import it with the short, documented alias:

import aga_runtime as aga

App reads AGA_URL, AGA_TENANT, AGA_NAMESPACE, and AGA_API_KEY by default; constructor values override the connection location and scope.

Declare durable functions

Declarations belong to one App:

from dataclasses import dataclass

import aga_runtime as aga

app = aga.App("checkout")


@dataclass
class Order:
    id: str
    amount: int


@dataclass
class Receipt:
    id: str
    approved: bool


@app.step(retry=aga.RetryPolicy(max_attempts=5), pivot=True)
def charge(order: Order) -> Receipt:
    return payments.charge(order)


@app.workflow()
async def checkout(order: Order) -> Receipt:
    receipt = await charge(order)
    if order.amount > 10_000:
        await app.signal(
            aga.Approval("large-charge", evidence=receipt),
            timeout=24 * 60 * 60,
        )
    app.event("order.charged", {"order": order.id})
    return receipt


if __name__ == "__main__":
    app.serve()

Aga restores supported declared Python types at durable JSON boundaries, including dataclasses, containers, enums, dates, UUIDs, and Pydantic models.

Create and coordinate work

Calling a registered Step or Remote eagerly creates a typed Handle. Workflows have one creation form: pass the registered Workflow to app.start(...).

Every Step, Remote, root Run, and child Run uses the same Handle[T]. Inside a Workflow, await a Handle. Outside, a root Handle also supports blocking .result(timeout=...).

run = app.start(checkout.options(run_id=order.id), order)
receipt = run.result()


@app.workflow()
async def checkout_with_audit(order: Order) -> Receipt:
    audit = app.start(audit_order, order)  # owned child; auto-joined
    receipt = await charge(order)
    await audit
    return receipt

Only independent child work opts out of ownership:

app.start(audit_order.options(detached=True), order)

One App normally represents one worker service and may own many Workflow definitions. Runs have separate execution-local state; they share only the App's registration, connection, lifecycle, and bounded worker capacity. Import every module that registers definitions before calling app.serve().

join is the single multi-Handle combinator:

all_values = await app.join(*handles)
first_value = await app.join(*handles, count=1)
two_values = await app.join(*handles, count=2)

Sync, Async, and Async Distributed

These describe two separate choices:

• Sync versus Async is caller behavior. Sync waits for a root Handle's result; Async keeps the same Handle and lets the caller continue.
• Async versus Async Distributed is workflow placement. Default Async keeps ordinary Steps with the Workflow worker. Async Distributed independently dispatches and fences each Step.

Use Sync when a request or script needs the result now. Use Async for long work or responsive callers. Use Async Distributed when Steps need separate scaling, isolation, placement, or leases. There is no execution="sync"; waiting never changes placement.

# Sync caller
receipt = app.start(checkout, order).result()

# Async caller
run = app.start(checkout, order)
receipt = await run


# Distributed placement
@app.workflow(execution="async_distributed")
async def distributed_checkout(order: Order) -> Receipt:
    return await charge(order)

A caller timeout does not cancel the durable Run. Cancellation is explicit with app.cancel(run, reason).

Waiting, services, and schedules

sleep completes through engine time. signal completes through external input. Approval is an immutable request passed to signal, not a third wait verb; it selects governed, fail-closed behavior and replay-bound evidence.

A Remote accepts one required request object so its envelope maps to Python, Go, Java, and TypeScript without language-specific argument rules. The serving App registers a Step with the same method name and portable request schema.

Schedules are engine-owned root-Run creation and continue while workers are offline:

@app.schedule("0 6 * * *", input={"report": "kpi"}, revision=2)
@app.workflow(name="reports.daily")
async def reports_daily(request: dict[str, str]) -> Report:
    return await build_report(request["report"])

Migrating from 0.4.1

There is no compatibility alias. Migrate application code and tests together, then restart the 0.4.2 worker fleet as one coordinated change. Aga server 0.2.1 remains compatible; this SDK release does not change the wire or schema.

Documentation: overview · Python guide · source

Apache License 2.0. See LICENSE.