agnt5

AGNT5 Python SDK

Build reliable AI agents and durable workflows in Python. The AGNT5 SDK provides typed components, workflow checkpoints, retries, streaming, tools, human-in-the-loop coordination, evaluation, and runtime observability.

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.

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.

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.

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.