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
| API | Purpose |
|---|---|
@function |
Typed, retryable units of work |
@workflow |
Durable multi-step orchestration and checkpointing |
Agent and @agent |
Model and tool orchestration |
@tool |
Typed tools with generated schemas |
ctx.state |
Durable workflow and component state |
ctx.memory |
Conversation and application memory |
Client / AsyncClient |
Invoke and observe deployed components |
Worker |
Register components and serve runtime dispatch |
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
- 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.
1 maintainer
Metadata
1 maintainer
Release files for agnt5 0.14.4
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| agnt5-0.14.4.tar.gz | 949.5 kB | Details |
Built distributions (wheels)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| agnt5-0.14.4-cp310-abi3-manylinux_2_28_x86_64.whl | CPython 3.10 | abi3 | Linux glibc 2.28+ x86-64 | Details |
| agnt5-0.14.4-cp310-abi3-manylinux_2_28_aarch64.whl | CPython 3.10 | abi3 | Linux glibc 2.28+ ARM64 | Details |
| agnt5-0.14.4-cp310-abi3-macosx_11_0_arm64.whl | CPython 3.10 | abi3 | macOS 11.0+ ARM64 | Details |
Total release size: 55.5 MB
Release files / agnt5-0.14.4.tar.gz
| Download URL | agnt5-0.14.4.tar.gz |
|---|---|
| Size | 949.5 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
f7f210fb671ab450296c549d594564823cb7280194ddeeb7e61bbe6a72da4d2a
|
|
BLAKE2b-256 checksum How to use checksums |
2fa011bbec7aab1ee0240845bf6f5732e1a75a567598e736fe06c15afd2c87c6
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Release files / agnt5-0.14.4-cp310-abi3-manylinux_2_28_x86_64.whl
| Download URL | agnt5-0.14.4-cp310-abi3-manylinux_2_28_x86_64.whl |
|---|---|
| Size | 18.4 MB |
| Tags | CPython 3.10 Linux glibc 2.28+ x86-64 abi3 |
|
SHA-256 checksum How to use checksums |
980cfd9bef2b407477296744c14a7af0743c7e1f675205f2a92227b7e7516a5f
|
|
BLAKE2b-256 checksum How to use checksums |
b51a4086dabd9f3f70f85f48bc0612ab4fe932e235291ddf3dc25cddc528f074
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Release files / agnt5-0.14.4-cp310-abi3-manylinux_2_28_aarch64.whl
| Download URL | agnt5-0.14.4-cp310-abi3-manylinux_2_28_aarch64.whl |
|---|---|
| Size | 19.4 MB |
| Tags | CPython 3.10 Linux glibc 2.28+ ARM64 abi3 |
|
SHA-256 checksum How to use checksums |
d170770fbfff895c1d3ddf4c3d697f03b5a3c513499555b10e625674848aa3f1
|
|
BLAKE2b-256 checksum How to use checksums |
aeb6090271fc509dbe1455095b9d8f83f924e0e3668ae978ae417c74daa9c2e3
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Release files / agnt5-0.14.4-cp310-abi3-macosx_11_0_arm64.whl
| Download URL | agnt5-0.14.4-cp310-abi3-macosx_11_0_arm64.whl |
|---|---|
| Size | 16.7 MB |
| Tags | CPython 3.10 abi3 macOS 11.0+ ARM64 |
|
SHA-256 checksum How to use checksums |
edcb13ca0c4e45ec8f653ea5fb083189c8e3e8c43971a19e6e6127924e3071de
|
|
BLAKE2b-256 checksum How to use checksums |
b9863f25501e094542e8fe75db06710f526f8c60a8fe6cded36263e2b779e376
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
