The orchestration platform for AI agents and workflows.
Durable execution, evals, and observability — unified in one SDK.
No more stitching together a workflow engine, an eval framework, and an observability stack.
from agnt5 import durable, Context @durable('triage_ticket', retries=5)async def triage(ctx: Context, ticket: dict): # LLM call — automatically retried and checkpointeddraft = await ctx.step('draft', lambda: ctx.llm.chat(messages=[{'role': 'user', 'content': ticket['text']}])) # Pause for human approval — can wait days, survives deploysdecision = await ctx.signal.wait('agent_approval', timeout='24h')if decision != 'approve':return 'held' # Send — exactly once, even after crashesawait ctx.step('send', lambda: ctx.email.send(to=ticket['user'], body=draft.text))return 'sent'Why AGNT5
Build, ship, and trust your AI agents.
Durable Runtime
Crash recovery built into the runtime. Each step is checkpointed — if the process crashes after extraction, it skips straight to classification on restart.
@durable('process_document', retries=3)
async def process(ctx: Context, doc: dict):
# Each step is checkpointed — if the process crashes
# after extraction, it skips straight to classification
text = await ctx.step('extract', lambda: ocr.extract(doc['url']))
label = await ctx.step('classify', lambda: ctx.llm.chat(
messages=[{'role': 'user', 'content': text}]
))
await ctx.step('store', lambda: db.insert(doc['id'], label))Deployments
Ship with a single command. AGNT5 Cloud handles scaling, rollbacks, and zero-downtime deploys. Or self-host on your own infra — same runtime, same guarantees.
$ agnt5 deploy
Deploying triage_ticket...
Building image ............ done
Pushing ................... done
Activating ................ done
Deployment live:
https://app.agnt5.dev/triage_ticket
Workflows: triage_ticket, process_document
Runtime: durable (PostgreSQL)
Replicas: auto (2-10)Evals in the runtime
Catch regressions before they reach users. Evals run on every execution — not just in CI. Define quality gates inline, right next to the code they protect.
@durable('support_agent')
async def support(ctx: Context, ticket: dict):
response = await ctx.step('respond', lambda: ctx.llm.chat(
messages=[{'role': 'user', 'content': ticket['text']}]
))
# Eval runs on every execution — not just in CI
await ctx.eval('quality', response.text, [
eval.toxicity(threshold=0.1),
eval.relevance(query=ticket['text'], threshold=0.8),
eval.latency(max_ms=3000),
])Observability built in
Traces, metrics, and logs — no instrumentation layer. Every step emits structured telemetry automatically. Debug in dev, monitor in production.
@durable('research_agent')
async def research(ctx: Context, query: str):
# Every step emits structured traces and metrics
# automatically — no decorators, no SDK calls
sources = await ctx.step('search', lambda: web.search(query))
summaries = []
for src in sources:
s = await ctx.step(f'summarize_{src.id}',
lambda: ctx.llm.chat(messages=[
{'role': 'user', 'content': src.text}
])
)
summaries.append(s)
return await ctx.step('synthesize',
lambda: ctx.llm.chat(messages=[
{'role': 'user', 'content': '\n'.join(summaries)}
])
)Platform
Built on a reliable foundation.
From filesystem to runtime, every layer is engineered for durability and developer experience.
Learn moreRust core, polyglot SDKs
The runtime is written in Rust for speed and safety. Language SDKs are thin bindings — Python today, TypeScript and Go coming — so you get native performance without leaving your language.
Event-sourced durability
Every step writes to an append-only event log. On crash, the runtime replays the log to reconstruct state — no manual checkpointing, no custom serialization.
Local dev = production
agnt5 dev
runs the full runtime locally with SQLite. Same execution model, same
guarantees. No Docker, no infra setup.
Deploy anywhere
Ship to AGNT5 Cloud with a single command, or self-host on your own infrastructure. PostgreSQL backend, horizontal scaling, multi-tenant isolation when you need it.