AGNT5 Python SDK

Build AI agents and durable workflows with the AGNT5 Python SDK

Build AI agents and reliable workflows with automatic recovery. AGNT5 combines agent orchestration and fault-tolerant execution in one lightweight framework.

Primitives comparison

Attribute	Function	Entity	Workflow	Agent	Tool
What	Stateless operation with retries	Stateful component with unique key	Multi-step orchestrated process	LLM with instructions and tools	Python function LLMs can call
State	None	Isolated per entity key	Isolated per workflow instance	Conversation history via Entity	None
Durability	Automatic retries, checkpointing	Persistent state across runs	Checkpointed steps, resume on failure	Context preserved in Entity	Runs within agent context
Best For	Document analysis, embeddings generation, LLM API calls	AI chat sessions, agent memory, conversation history	RAG pipelines, content generation with review, AI evals	Customer support, research assistants, code review	Vector search, knowledge base queries, API integrations

Key Features

Automatic recovery from failures with configurable retry policies
Checkpointing resumes from exact failure point
Multi-agent coordination via handoffs and composition
Python-native - decorators, async/await, type hints
Multi-provider - OpenAI, Anthropic, Groq, Azure, Bedrock, OpenRouter
Built-in tracing for debugging and monitoring

Installation

pip install agnt5

Quick example

from agnt5 import Agent, workflow, tool, Context, WorkflowContext

# Define a tool for the agent
@tool(auto_schema=True)
async def search_docs(ctx: Context, query: str) -> str:
    """Search documentation for answers."""
    # Your search logic here
    return f"Found documentation about: {query}"

# Create an AI agent with tools
agent = Agent(
    name="assistant",
    model="openai/gpt-4o-mini",
    instructions="You are a helpful assistant. Search docs when needed.",
    tools=[search_docs]
)

# Create a durable workflow that orchestrates the agent
@workflow
async def process_question(ctx: WorkflowContext, question: str) -> dict:
    """Durable workflow for processing questions."""

    # Step 1: Get answer from agent (checkpointed)
    answer = await ctx.step("get_answer", agent.run(question))

    # Step 2: Store result (checkpointed)
    await ctx.step("store", save_answer(question, answer))

    return {"question": question, "answer": answer}

# If this crashes after step 1, it resumes from step 2 on restart