AGNT5 Python SDK
Build durable AI agents and workflows with automatic retries, checkpointing, and state management.
Documentation
See the online documentation for example applications, user guides, and detailed API reference.
Installation
pip install agnt5
Quick Start
from agnt5 import function, Context
@function
async def analyze_document(ctx: Context, document_id: str) -> dict:
"""Analyze document with automatic retries."""
ctx.logger.info(f"Analyzing document: {document_id}")
# Checkpoint expensive operations
embeddings = await ctx.step("embeddings", lambda: generate_embeddings(document_id))
# State management
ctx.set("embedding_id", embeddings["id"])
return {"document_id": document_id, "status": "completed"}
Functions automatically retry on failure. Use ctx.step() to checkpoint progress.
Client Authentication
The AGNT5 client supports two authentication methods:
Subdomain URLs (Recommended)
Access your deployment directly via its unique subdomain:
from agnt5 import Client
# Production environment
client = Client("https://myproject-prod.run.agnt5.com")
# Staging environment
client = Client("https://myproject-stg.run.agnt5.com")
# Run a workflow
result = client.run("my_workflow", {"input": "data"})
API Keys
For programmatic access, use service keys:
from agnt5 import Client
# Option 1: Pass api_key directly
client = Client(
gateway_url="https://api.agnt5.com",
api_key="agnt5_sk_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
)
# Option 2: Use AGNT5_API_KEY environment variable
# export AGNT5_API_KEY=agnt5_sk_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
client = Client(gateway_url="https://api.agnt5.com")
result = client.run("my_workflow", {"input": "data"})
See the Authentication Guide for more details.
Core Components
Functions
Stateless operations with automatic retry and checkpointing.
from agnt5 import function, RetryPolicy, BackoffPolicy, BackoffType
@function(
retries=RetryPolicy(max_attempts=5, initial_interval_ms=1000),
backoff=BackoffPolicy(type=BackoffType.EXPONENTIAL, multiplier=2.0)
)
async def generate_summary(ctx: Context, text: str) -> dict:
return await llm.generate(prompt=f"Summarize: {text}")
Entities
Stateful components where each instance maintains its own state.
from agnt5 import entity
@entity
class Conversation:
pass
@Conversation.method
async def add_message(ctx: Context, role: str, content: str) -> list:
messages = ctx.get("messages", [])
messages.append({"role": role, "content": content})
ctx.set("messages", messages)
return messages
# Each conversation has independent message history
convo = Conversation.instance("user-123")
await convo.add_message(ctx, role="user", content="Hello!")
Workflows
Orchestrate multi-step processes with parallel execution and coordination.
from agnt5 import workflow
@workflow
async def research_workflow(ctx: Context, topic: str) -> dict:
# Run searches in parallel
results = await ctx.parallel(
search_papers(ctx, topic),
search_web(ctx, topic)
)
# Wait for human review
reviewed = await ctx.signal("review_complete", timeout_ms=60000)
return {"topic": topic, "results": results, "reviewed": reviewed}
Tools
Functions callable by LLMs with automatic schema generation.
from agnt5 import tool
@tool
async def search_knowledge_base(ctx: Context, query: str, limit: int = 10) -> list:
"""Search the knowledge base using semantic search.
Args:
query: Search query
limit: Max results to return
"""
embeddings = await generate_embeddings(query)
return await vector_db.search(embeddings, limit=limit)
# Schema generated from type hints and docstring
schema = search_knowledge_base.get_schema()
Agents
LLM-powered agents with tool orchestration.
from agnt5 import Agent
agent = Agent(
name="research_assistant",
instructions="You are a research assistant. Use tools to find information.",
tools=[search_knowledge_base, search_papers, summarize_text],
model_name="gpt-4o"
)
result = await agent.run("Find recent papers on transformer architectures")
Context API
@function
async def example(ctx: Context, data: dict) -> dict:
# Metadata
ctx.run_id, ctx.attempt, ctx.logger
# State management
ctx.set("key", value)
ctx.get("key", default=None)
ctx.delete("key")
# Checkpointing
result = await ctx.step("step_name", lambda: expensive_op())
# Coordination
await ctx.sleep(5)
await ctx.timer(delay_ms=1000)
await ctx.signal("name", timeout_ms=5000)
# Parallel execution
await ctx.parallel(task1(), task2())
await ctx.gather(db=fetch_db(), api=fetch_api())
return {"result": result}
Error Handling
from agnt5.exceptions import RetryError, StateError
@function
async def robust_function(ctx: Context, data: str) -> dict:
try:
return await risky_operation(data)
except RetryError as e:
ctx.logger.error(f"Failed after {e.attempts} attempts")
return {"error": "max_retries_exceeded"}
Examples
See examples/ directory for working examples.
License
Apache 2.0
1 maintainer
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distributions
No source distribution files available for this release.See tutorial on generating distribution archives.
Built Distributions
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file agnt5-0.9.0-cp310-abi3-manylinux_2_39_x86_64.whl.
File metadata
- Download URL: agnt5-0.9.0-cp310-abi3-manylinux_2_39_x86_64.whl
- Upload date:
- Size: 17.4 MB
- Tags: CPython 3.10+, manylinux: glibc 2.39+ x86-64
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/6.1.0 CPython/3.13.14
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
68c0a1715e97519f56ccb1e2de41e43b19fd98613885ac58c80962eff072362b
|
|
| MD5 |
7d6df452d2c9547a751a24696d3f5545
|
|
| BLAKE2b-256 |
1d1a20fecd6680445aa7180e7cb3670b5b9bfd56394de0e6fef3f36dcc040f74
|
File details
Details for the file agnt5-0.9.0-cp310-abi3-manylinux_2_39_aarch64.whl.
File metadata
- Download URL: agnt5-0.9.0-cp310-abi3-manylinux_2_39_aarch64.whl
- Upload date:
- Size: 17.6 MB
- Tags: CPython 3.10+, manylinux: glibc 2.39+ ARM64
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/6.1.0 CPython/3.13.14
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
73c2cade5bda642b8d2048207a22a3db900bac2fbf1a156b392821534fb2bd22
|
|
| MD5 |
82e83d6ba9690022f1fb64b6f7b370b5
|
|
| BLAKE2b-256 |
05e7175e4458c93c1223e434c6fb3ac0cc1d0eb8ea0e3346dda2ec2953e688f6
|
File details
Details for the file agnt5-0.9.0-cp310-abi3-macosx_11_0_arm64.whl.
File metadata
- Download URL: agnt5-0.9.0-cp310-abi3-macosx_11_0_arm64.whl
- Upload date:
- Size: 15.9 MB
- Tags: CPython 3.10+, macOS 11.0+ ARM64
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/6.1.0 CPython/3.13.14
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
88a45dea3fe157147097b47c6b47f91483d14639821a5b3dbf2b57d8c8ed8bb4
|
|
| MD5 |
a06c78f5c925f92f072a5fb09e9513a5
|
|
| BLAKE2b-256 |
1ba856f1e39462957dc78501b9de6530e841d6931699ffe24f1e8a873688693a
|
File details
Details for the file agnt5-0.9.0-cp310-abi3-macosx_10_12_x86_64.whl.
File metadata
- Download URL: agnt5-0.9.0-cp310-abi3-macosx_10_12_x86_64.whl
- Upload date:
- Size: 16.4 MB
- Tags: CPython 3.10+, macOS 10.12+ x86-64
- Uploaded using Trusted Publishing? No
- Uploaded via:
twine/6.1.0 CPython/3.13.14
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
459ef1d24597523ff986f4adab92d80f8e746a1f26b81b2a1716d7c8343e5c67
|
|
| MD5 |
c2aa30e526fb8686dd4fb41b483d2d95
|
|
| BLAKE2b-256 |
5cfc4c67a77697cceb25266c15825abf4e78d956c79b9b7cbe87ef8653662916
|
