AGNT5 Python SDK

Build reliable AI agents and durable workflows in Python. AGNT5 checkpoints each workflow step, so a run that fails partway resumes from its last completed step instead of starting over. The SDK provides typed functions and workflows, retries, streaming, tool calling, multi-agent handoffs, human-in-the-loop (HITL) approvals, MCP, evals, and runtime observability.

Documentation · Starter templates · Website

Requirements

• Python 3.11 or newer
• An AGNT5 runtime for deployed execution

Installation

pip install agnt5

Quick start

Define a function. Type hints are used to derive its input and output schema.

from agnt5 import FunctionContext, function


@function(retries=3, backoff="exponential")
async def greet(ctx: FunctionContext, name: str) -> dict[str, str]:
    ctx.logger.info("greeting user", extra={"name": name})
    return {"message": f"Hello, {name}!"}

Register application components with a worker for runtime-backed execution:

import asyncio

from agnt5 import Worker


async def main() -> None:
    worker = Worker(service_name="hello-python")
    await worker.run()


asyncio.run(main())

Decorated functions, workflows, agents, tools, and scorers are registered when their modules are imported. See examples/app.py for a complete application entrypoint.

Invoke a deployed component

from agnt5 import Client

client = Client(
    gateway_url="https://gw.agnt5.com",
    api_key="agnt5_sk_...",
    deployment_id="deployment-id",
)

result = client.run("greet", {"name": "Ada"})
print(result)

Client also supports asynchronous submission, status and result polling, streaming events, batches, cancellation, workflow resume, chat, and evaluation. Configuration can be supplied explicitly or through AGNT5_GATEWAY_URL, AGNT5_API_KEY, and AGNT5_DEPLOYMENT_ID.

Durable steps

Use @step for an operation inside a durable workflow. Its handler starts after the runtime acknowledges admission, and its result becomes available after the runtime acknowledges completion. Those two activation calls also record the step's lifecycle; the SDK adds no separate function checkpoints.

from agnt5 import StepContext, WorkflowContext, step, workflow


@step(retries=3, backoff="exponential", timeout_ms=5000)
async def prepare_search(ctx: StepContext, topic: str) -> dict[str, str]:
    return {"query": topic.strip()}


@workflow
async def research(ctx: WorkflowContext, topic: str) -> dict[str, str]:
    return await ctx.step(prepare_search, topic, key="search-query")

Register the workflow with a worker; @step handlers are workflow-local and are not standalone components. Invoke them through ctx.step(...). Direct calls and execution without durable activation support raise an error rather than running without durability. A first parameter named ctx receives a StepContext; it can be omitted when unused. Sync handlers run in an executor thread.

Use explicit keys for parallel or repeated work. Completed steps replay their committed result without executing the handler again. Retries reuse the same activation identity with runtime-owned attempts. External side effects still need idempotency: ctx.activation.idempotency_key is available to handlers. Existing @function handlers remain supported with their function checkpoints. ctx.run(...) is deprecated; replace it with ctx.step(...) using the same arguments and keys. It remains functional and emits a DeprecationWarning.

Core APIs

Functions called inside workflow steps honor their declared retry and backoff policy. With negotiated durable activations, each attempt is admitted and recorded by the runtime; compatibility execution applies the same attempt budget locally. Completed steps replay their recorded output. Make external side effects idempotent so an interrupted attempt can safely run again.

In async workflows, use await ctx.state.set_async(key, value) and await ctx.state.delete_async(key). These await the runtime's durable acknowledgment while allowing other workflows to run. Reads remain local via ctx.state.get(key). The existing synchronous set and delete methods remain supported, but block their calling thread while waiting for persistence.

The shared Rust runtime foundation lives in agnt5dev/sdk-core. Vendor sandbox adapters live in agnt5dev/sdk-integrations.

Event triggers

Use event(...) or webhook(...) in a workflow's triggers declarations. Filtering, input mapping, batching, and delays are not supported yet. Leave filter_expression, input_mapping, batch_window_ms, and delay_expression unset; workflow registration rejects nonempty expressions and nonzero batch windows with an error naming the option.

Examples and documentation

• examples/ contains functions, workflows, agents, tools, streaming, HITL, MCP, chat, and evaluation examples.
• docs/ contains SDK-specific guides.
• AGNT5 documentation covers platform concepts and deployment.

Response wait

Run and stream calls wait up to 5 minutes by default. Set the per-call wait to any value from zero to 24 hours. Zero returns a pending receipt immediately after acceptance. This controls response waiting, not the workflow execution deadline: accepted work continues when the wait expires or the client disconnects.

result = client.run("process_order", order, component_type="workflow", wait_timeout=60)
for event in client.stream_events("process_order", order, component_type="workflow", wait_timeout=60):
    print(event.event_type, event.run_id)

wait_timeout uses seconds and also applies to async run and stream_events. Run calls return 202 pending receipts directly, without additional polling. Event streams emit stream.wait_expired when the wait expires, or stream.detached for a 202 receipt. Use the run ID to read status/results. Chunk-only stream raises RunError with the run ID when waiting ends.

The default HTTP timeout allows at least the wait plus 10 seconds, or the client timeout if longer. Pass timeout=75 to set it explicitly. Python's HTTP timeout limits individual network operations; wait_timeout bounds the gateway wait.

Structured assertions

structured_assertions is a reserved built-in scorer, so workers need no user registration. The local helper uses the same SDK-core implementation as the runtime:

from agnt5.eval import ScorerInput, structured_assertions

result = structured_assertions(
    ScorerInput(output=[1, 2, 3], expected={"expected_length": 3}),
    {"assertions": [
        {"name": "unique_ids", "expr": "unique(output_json)"},
        {"name": "count", "expr": "size(output_json) == expected.expected_length"},
    ]},
)

The score is the fraction of assertions that pass; score_threshold defaults to

1. Configuration and input errors always fail. See the SDK-core contract for supported expressions and execution limits. Requires the matching native extension.

Serverless signing

Pass signing_secret to agnt5.serverless.serve() or ServerlessApp. It can be a string or a resolver that reads your provider's environment. Invokes with no resolved secret return HTTP 503 (WORKERLESS_SIGNING_SECRET_REQUIRED) before running user code. Missing or invalid signatures with a configured secret return 401. The manifest remains available for discovery.

For local development only, serve(allow_unsigned=True) permits unsigned invokes when no secret resolves and logs a warning at startup. It still verifies requests when a secret is configured. Missing static secrets warn at startup; an empty request-time resolver warns on its first failed invoke. AGNT5 activation and dispatch still require a signing secret, regardless of this SDK option.

Development

uv sync --all-groups
uv run ruff check src tests
uv run pytest tests/unit -q
uv build --sdist

Development builds use a sibling checkout of sdk-core.

Contributing

See CONTRIBUTING.md. Report security issues according to SECURITY.md.

License

Licensed under the Apache License 2.0.

Declared steps keep managed waits such as ctx.sleep() and ctx.wait_for_user() at workflow boundaries. Call them between steps. For a timed-out synchronous handler, failure and retries wait until its executor thread exits; a timeout cannot stop a Python thread.