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

Primitives comparison

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

Note: Set your OPENAI_API_KEY environment variable before running.

Next Steps

Getting Started

• Quickstart - Installation, first worker, and local development setup
• Worker Runtime - Configure and deploy workers

Core Primitives

• Functions - Stateless operations with retries
• Entities - Stateful components with unique keys
• Workflows - Multi-step orchestration patterns
• Context API - Orchestration, state, AI, and observability APIs

Agent Development Kit (ADK)

• Agents - Autonomous LLM-driven systems
• Sessions - Conversation containers and multi-agent coordination
• Tools - Callable capabilities that extend agent abilities
• Memory - Long-term knowledge storage with semantic search

Examples

• Examples - Practical usage examples

Get up and running with the AGNT5 Python SDK in minutes. This guide covers installation, your first worker, and local development setup.

Installation

System Requirements

• Python 3.8 or higher
• pip or uv package manager
• Docker (for local development server)

Install from PyPI

pip install agnt5

Development Installation

For development or contributing to the SDK:

git clone https://github.com/agnt5/agnt5
cd agnt5/sdk/sdk-python
pip install -e .

Verify Installation

import agnt5
print(agnt5.__version__)

First Worker

Create a simple worker with a greeting function:

import asyncio
from agnt5 import Worker, function

@function()
def greet(name: str) -> str:
    """Greet a user by name."""
    return f"Hello, {name}!"

@function("math_add")
def add_numbers(a: int, b: int) -> int:
    """Add two numbers together."""
    return a + b

async def main():
    worker = Worker(service_name="hello-service")
    await worker.run()

if __name__ == "__main__":
    asyncio.run(main())

Local Development Setup

Start the AGNT5 Development Server

The development server runs all AGNT5 services locally in Docker:

# Install the AGNT5 CLI first
brew install agnt5/tap/agnt5  # macOS
# or
curl -LsSf https://agnt5.com/cli.sh | bash  # Linux

# Start development environment
agnt5 dev up

git clone https://github.com/agnt5/agnt5
cd agnt5
just dev-up

Run Your Worker

In a new terminal, run your worker:

python worker.py

You should see output like:

INFO:agnt5.worker:Starting worker for service: hello-service
INFO:agnt5.worker:Registered function: greet
INFO:agnt5.worker:Registered function: math_add
INFO:agnt5.worker:Worker running, waiting for tasks...

Test Your Functions

Using HTTP API

Test your functions using the Gateway HTTP API:

# Test the greet function
curl -X POST http://localhost:8080/call \
  -H "Content-Type: application/json" \
  -d '{
    "serviceName": "hello-service",
    "handlerName": "greet",
    "inputData": "QWxpY2U="
  }'

The inputData is base64-encoded JSON. For "Alice", the base64 is "QWxpY2U=".

Using Python Client

import asyncio
import json
import base64
from agnt5 import Client

async def test_functions():
    client = Client("http://localhost:8080")

    # Test greet function
    name = "Alice"
    input_data = base64.b64encode(json.dumps(name).encode()).decode()

    result = await client.call(
        service_name="hello-service",
        handler_name="greet",
        input_data=input_data
    )

    print(f"Greeting result: {result}")

    # Test add function
    numbers = {"a": 5, "b": 3}
    input_data = base64.b64encode(json.dumps(numbers).encode()).decode()

    result = await client.call(
        service_name="hello-service",
        handler_name="math_add",
        input_data=input_data
    )

    print(f"Addition result: {result}")

asyncio.run(test_functions())

ASGI Integration

For web applications, use the ASGI runtime:

from agnt5 import Worker, function

@function()
def api_handler(request: dict) -> dict:
    return {
        "message": "Hello from AGNT5!",
        "received": request
    }

# Create ASGI application
app = Worker("web-service", runtime="asgi")
app.enable_cors()  # Enable CORS for browser access

Run with uvicorn:

pip install uvicorn
uvicorn asgi_app:app --reload --port 8000

Test the ASGI endpoints:

# Health check
curl http://localhost:8000/health

# List available functions
curl http://localhost:8000/functions

# Call a function
curl -X POST http://localhost:8000/invoke/api_handler \
  -H "Content-Type: application/json" \
  -d '{"test": "data"}'

Configuration

Environment Variables

Configure your worker using environment variables:

export AGNT5_COORDINATOR_ENDPOINT=http://localhost:9091
export AGNT5_SERVICE_NAME=my-service
export AGNT5_LOG_LEVEL=DEBUG

python worker.py

Configuration in Code

import logging
from agnt5 import Worker
from agnt5.logging import install_opentelemetry_logging

# Configure logging
logging.basicConfig(level=logging.INFO)
install_opentelemetry_logging()

# Create worker with custom configuration
worker = Worker(
    service_name="configured-service",
    service_version="1.2.0",
    coordinator_endpoint="http://localhost:9091",
    runtime="standalone"
)

Error Handling

Handle errors gracefully in your functions:

from agnt5 import function
import logging

logger = logging.getLogger(__name__)

@function()
def safe_divide(a: float, b: float) -> dict:
    try:
        if b == 0:
            return {"error": "Division by zero", "result": None}

        result = a / b
        logger.info(f"Division successful: {a} / {b} = {result}")

        return {"result": result, "error": None}

    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        return {"error": str(e), "result": None}

Development Tips

Hot Reload

During development, restart your worker when code changes:

import asyncio
import sys
from pathlib import Path
from agnt5 import Worker, function

# Add auto-reload during development
if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\nWorker stopped")
        sys.exit(0)

Debugging

Enable debug logging to see detailed execution information:

import logging
logging.basicConfig(level=logging.DEBUG)

from agnt5.logging import install_opentelemetry_logging
install_opentelemetry_logging(level=logging.DEBUG)

Testing Functions

Test your functions locally without the full platform:

from agnt5.decorators import execute_component

# Test function directly
result = execute_component("greet", b'{"name": "Alice"}')
print(result)

Next Steps

Core Primitives

• Functions - Stateless operations with retries
• Entities - Stateful components with unique keys
• Workflows - Multi-step orchestration
• Context API - Full API reference

Agent Development Kit

• Agents - Autonomous LLM-driven systems
• Tools - Extend agent capabilities
• Sessions - Conversation management
• Memory - Long-term knowledge storage

Configuration

• Worker Runtime - Configure and deploy workers
• Examples - Real-world usage patterns

Functions are the core building blocks of AGNT5 applications. Use the @function decorator to register Python callables as invokable components that can be discovered and executed by the platform.

Basic Usage

Simple Function

from agnt5 import function

@function()
def greet_user(name: str) -> str:
    """Greet a user by name."""
    return f"Hello, {name}!"

Named Function

Override the registered name:

@function("math.add")
def add_numbers(a: int, b: int) -> int:
    """Add two numbers together."""
    return a + b

Function with Context

Access execution metadata through the context parameter:

from agnt5 import function
from agnt5.components import ExecutionContext

@function()
def context_aware(ctx: ExecutionContext, data: dict) -> dict:
    """Process data with execution context."""
    return {
        "invocation_id": ctx.invocation_id,
        "service_name": ctx.metadata.get("service_name"),
        "processed_data": data,
        "component_type": ctx.component_type.value
    }

Decorator Parameters

function(name=None)

# Uses function name "process_data"
@function()
def process_data(data: dict) -> dict:
    return data

# Uses custom name "data_processor"
@function("data_processor")
def process_data(data: dict) -> dict:
    return data

Handler Signatures

AGNT5 supports flexible function signatures to accommodate different use cases.

Without Context

For simple stateless functions:

@function()
def calculate_tax(amount: float, rate: float) -> float:
    return amount * rate

@function()
def format_message(template: str, **kwargs) -> str:
    return template.format(**kwargs)

With Context

When you need access to invocation metadata:

@function()
def audit_handler(ctx: ExecutionContext, action: str, data: dict) -> dict:
    """Handler that logs audit information."""
    import logging

    logger = logging.getLogger(__name__)
    logger.info(f"Audit: {action} from {ctx.invocation_id}")

    return {
        "action": action,
        "invocation_id": ctx.invocation_id,
        "data": data,
        "timestamp": time.time()
    }

Async Functions

AGNT5 supports both synchronous and asynchronous functions:

Async Handler

import asyncio
from agnt5 import function

@function()
async def async_processor(data: dict) -> dict:
    """Async processing with I/O operations."""
    # Simulate async I/O
    await asyncio.sleep(0.1)

    # Async HTTP request example
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://api.example.com/process",
            json=data
        )
        external_result = response.json()

    return {
        "original": data,
        "external": external_result,
        "processed_at": datetime.utcnow().isoformat()
    }

Async with Context

@function()
async def async_context_handler(ctx: ExecutionContext, query: str) -> dict:
    """Async handler with context access."""

    # Use context for correlation
    correlation_id = ctx.metadata.get("correlation_id", ctx.invocation_id)

    # Async database query
    result = await database.execute(
        "SELECT * FROM items WHERE name LIKE %s",
        f"%{query}%"
    )

    return {
        "correlation_id": correlation_id,
        "query": query,
        "results": [dict(row) for row in result]
    }

Streaming Functions

For functions that need to return multiple responses over time:

Basic Streaming

from agnt5 import function

@function(streaming=True)
async def stream_data(count: int):
    """Stream multiple data chunks."""
    for i in range(count):
        yield {
            "chunk": i,
            "data": f"Data chunk {i}",
            "timestamp": time.time()
        }
        await asyncio.sleep(0.1)  # Simulate processing time

Streaming with Context

@function(streaming=True)
async def stream_with_context(ctx: ExecutionContext, query: str):
    """Stream search results progressively."""
    search_id = ctx.invocation_id

    # Stream results as they're found
    async for result in search_engine.stream_search(query):
        yield {
            "search_id": search_id,
            "result": result,
            "timestamp": time.time()
        }

Function Metadata

The SDK automatically captures and provides metadata about registered functions:

Inspecting Functions

from agnt5.decorators import get_registered_functions, get_function_metadata

# Get all registered functions
functions = get_registered_functions()
print(f"Registered functions: {list(functions.keys())}")

# Get metadata for a specific function
@function()
def sample_function(name: str, age: int = 25) -> dict:
    return {"name": name, "age": age}

metadata = get_function_metadata(sample_function)
print(metadata)

Output:

{
    "name": "sample_function",
    "type": "function",
    "parameters": [
        {"name": "name", "type": "str", "required": True},
        {"name": "age", "type": "int", "required": False, "default": 25}
    ],
    "return_type": "dict"
}

Runtime Annotations

The decorator adds runtime annotations to functions:

@function()
def annotated_function(data: str) -> str:
    return data.upper()

# Check annotations
print(annotated_function._agnt5_handler_name)  # "annotated_function"
print(annotated_function._agnt5_is_function)   # True

Error Handling

Basic Error Handling

@function()
def safe_divider(a: float, b: float) -> dict:
    """Safely divide two numbers."""
    try:
        if b == 0:
            return {"error": "Division by zero", "result": None}

        result = a / b
        return {"result": result, "error": None}

    except Exception as e:
        return {"error": str(e), "result": None}

Context-Aware Error Handling

import logging
from agnt5 import function
from agnt5.components import ExecutionContext

@function()
def robust_handler(ctx: ExecutionContext, data: dict) -> dict:
    """Handler with comprehensive error handling."""
    logger = logging.getLogger(__name__)

    try:
        # Log the invocation
        logger.info(f"Processing invocation {ctx.invocation_id}")

        # Validate input
        if not isinstance(data, dict):
            raise ValueError("Input must be a dictionary")

        required_fields = ["id", "name"]
        missing_fields = [field for field in required_fields if field not in data]
        if missing_fields:
            raise ValueError(f"Missing required fields: {missing_fields}")

        # Process data
        result = {
            "processed": True,
            "id": data["id"],
            "name": data["name"].upper(),
            "invocation_id": ctx.invocation_id
        }

        logger.info(f"Successfully processed {data['id']}")
        return result

    except ValueError as e:
        logger.warning(f"Validation error in {ctx.invocation_id}: {e}")
        return {"error": f"Validation error: {e}", "result": None}

    except Exception as e:
        logger.error(f"Unexpected error in {ctx.invocation_id}: {e}")
        return {"error": "Internal error", "result": None}

Type Annotations

Use Python type hints for better documentation and validation:

Basic Types

from typing import Dict, List, Optional, Union

@function()
def typed_handler(
    name: str,
    age: int,
    tags: List[str],
    metadata: Optional[Dict[str, any]] = None
) -> Dict[str, Union[str, int, List[str]]]:
    """Handler with comprehensive type annotations."""
    return {
        "name": name,
        "age": age,
        "tags": tags,
        "has_metadata": metadata is not None
    }

Pydantic Models

For complex data validation:

from pydantic import BaseModel, Field
from typing import Optional
from agnt5 import function

class UserRequest(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    email: str = Field(..., pattern=r'^[\w\.-]+@[\w\.-]+\.\w+$')
    age: Optional[int] = Field(None, ge=0, le=150)

class UserResponse(BaseModel):
    id: str
    name: str
    email: str
    age: Optional[int]
    created_at: str

@function()
def create_user(request: UserRequest) -> UserResponse:
    """Create a user with validation."""
    user_id = generate_user_id()

    return UserResponse(
        id=user_id,
        name=request.name,
        email=request.email,
        age=request.age,
        created_at=datetime.utcnow().isoformat()
    )

Testing Functions

Direct Testing

Test functions directly without the full platform:

import pytest
from agnt5.decorators import execute_component
from agnt5.components import ExecutionContext, ComponentType

def test_greet_function():
    # Test with execute_component
    result = execute_component(
        "greet_user",
        b'{"name": "Alice"}',
        context=None
    )

    # Result is JSON bytes
    import json
    parsed = json.loads(result.decode())
    assert parsed == "Hello, Alice!"

def test_context_function():
    # Create mock context
    ctx = ExecutionContext(
        invocation_id="test-123",
        component_type=ComponentType.FUNCTION
    )

    # Test directly
    result = context_aware(ctx, {"test": "data"})
    assert result["invocation_id"] == "test-123"

Async Testing

import pytest
import asyncio

@pytest.mark.asyncio
async def test_async_function():
    result = await async_processor({"test": "data"})
    assert "original" in result
    assert "processed_at" in result

Mock Context Testing

from unittest.mock import Mock

def test_with_mock_context():
    # Create mock context
    mock_ctx = Mock(spec=ExecutionContext)
    mock_ctx.invocation_id = "mock-123"
    mock_ctx.component_type = ComponentType.FUNCTION
    mock_ctx.metadata = {"service_name": "test-service"}

    # Test function
    result = context_aware(mock_ctx, {"test": "data"})
    assert result["invocation_id"] == "mock-123"
    assert result["service_name"] == "test-service"

Function Registry

Registry Management

from agnt5.decorators import (
    get_registered_functions,
    clear_registry,
    get_function_metadata
)

# Get all registered functions
functions = get_registered_functions()

# Clear registry (useful for testing)
clear_registry()

# Re-register functions
@function()
def new_function(data: str) -> str:
    return data.upper()

# Inspect metadata
metadata = get_function_metadata(new_function)

Custom Registration

For advanced use cases, register functions manually:

from agnt5.decorators import register_function

def my_handler(data: str) -> str:
    return data.lower()

# Manual registration
register_function("custom_handler", my_handler)

Best Practices

Function Design

1. Keep functions focused - Each function should have a single responsibility
2. Use type hints - Improve documentation and enable validation
3. Handle errors gracefully - Return error information rather than raising exceptions
4. Log appropriately - Use structured logging for debugging and monitoring

Performance

1. Minimize imports - Import only what you need
2. Use async for I/O - Async functions for database queries and API calls
3. Cache expensive operations - Use local caching for repeated computations
4. Batch operations - Process multiple items together when possible

Testing

1. Test functions directly - Unit test without the platform
2. Mock external dependencies - Use mocks for databases, APIs, etc.
3. Test error conditions - Ensure error handling works correctly
4. Use fixtures - Share common test data and setup

Next Steps

• Workflows - Multi-step orchestration patterns
• Worker Runtime - Configure and deploy workers
• API Reference - Complete decorator API reference
• Examples - Real-world function examples

Entities are stateful components identified by unique keys. Use entities to model AI agents with conversation history, workflow orchestrators, or any business object that maintains state across interactions.

Key Characteristics

• Unique Key - Each instance identified by a unique key (e.g., agent-conv-123)
• Private State - Built-in key-value storage per instance
• Single-Writer - Automatic consistency - only one write operation per key at a time
• Durable - State survives crashes and restarts
• Scalable - Different keys execute in parallel

Basic Usage

Creating an Entity

from agnt5 import entity

# Create entity type
agent = entity("ConversationAgent")

# Write method (exclusive access per key)
@agent.write
async def send_message(ctx, message: str) -> dict:
    history = await ctx.get("history", [])
    history.append({"role": "user", "content": message})

    response = await call_llm(history)
    history.append({"role": "assistant", "content": response})

    ctx.set("history", history)
    return {"response": response}

# Shared method (read-only, concurrent)
@agent.shared
async def get_history(ctx) -> list:
    return await ctx.get("history", [])

Calling Entities

Call entity methods from functions:

from agnt5 import function

@function
async def chat(ctx, conv_id: str, msg: str):
    # Call entity method with unique key
    return await ctx.entity("ConversationAgent", conv_id).send_message(msg)

Entity API

Core Methods

State Operations

@agent.write
async def process(ctx, data: dict) -> dict:
    # Get with default
    history = await ctx.get("history", [])
    count = await ctx.get("count", 0)

    return {"history": history, "count": count}

@agent.write
async def update_state(ctx, new_data: dict) -> dict:
    # Set values
    ctx.set("last_update", datetime.now().isoformat())
    ctx.set("data", new_data)
    ctx.set("version", 2)

    return {"status": "updated"}

@agent.write
async def clear_cache(ctx) -> dict:
    # Delete keys
    ctx.delete("cached_results")
    ctx.delete("temporary_data")

    return {"status": "cleared"}

Common Patterns

Conversational AI Agent

agent = entity("ChatAgent")

@agent.write
async def send_message(ctx, message: str) -> dict:
    """Handle conversational turns with LLM."""
    history = await ctx.get("history", [])
    history.append({"role": "user", "content": message})

    # Generate response
    response = await ctx.llm.generate(
        prompt=history,
        model="gpt-4"
    )
    history.append({"role": "assistant", "content": response.text})

    # Keep last 20 messages
    if len(history) > 20:
        history = history[-20:]

    ctx.set("history", history)
    return {"response": response.text}

@agent.shared
async def get_history(ctx) -> list:
    """Get conversation history (read-only)."""
    return await ctx.get("history", [])

@agent.shared
async def get_message_count(ctx) -> int:
    """Get total message count."""
    history = await ctx.get("history", [])
    return len(history)

Usage:

@function
async def chat_endpoint(ctx, conversation_id: str, message: str):
    # Call entity with unique conversation ID
    return await ctx.entity("ChatAgent", conversation_id).send_message(message)

Research Agent

research_agent = entity("ResearchAgent")

@research_agent.write
async def start_research(ctx, topic: str) -> dict:
    """Initialize research task."""
    ctx.set("topic", topic)
    ctx.set("findings", [])
    ctx.set("status", "in_progress")
    return {"status": "started", "topic": topic}

@research_agent.write
async def add_finding(ctx, finding: str, source: str) -> dict:
    """Add research finding."""
    findings = await ctx.get("findings", [])
    findings.append({
        "content": finding,
        "source": source,
        "timestamp": datetime.now().isoformat()
    })
    ctx.set("findings", findings)
    return {"count": len(findings)}

@research_agent.write
async def synthesize(ctx) -> dict:
    """Generate summary from findings."""
    findings = await ctx.get("findings", [])
    topic = await ctx.get("topic")

    # Use LLM to synthesize
    summary = await ctx.llm.generate(
        prompt=f"Synthesize these findings about {topic}: {findings}",
        model="gpt-4"
    )

    ctx.set("summary", summary.text)
    ctx.set("status", "completed")
    return {"summary": summary.text}

@research_agent.shared
async def get_progress(ctx) -> dict:
    """Check research progress."""
    return {
        "status": await ctx.get("status"),
        "topic": await ctx.get("topic"),
        "findings_count": len(await ctx.get("findings", []))
    }

Workflow Orchestrator

workflow = entity("WorkflowOrchestrator")

@workflow.write
async def start(ctx, steps: list) -> dict:
    """Start workflow execution."""
    ctx.set("steps", steps)
    ctx.set("current_step", 0)
    ctx.set("results", [])
    ctx.set("status", "running")
    return {"status": "started", "total_steps": len(steps)}

@workflow.write
async def complete_step(ctx, result: dict) -> dict:
    """Mark step as complete and store result."""
    results = await ctx.get("results", [])
    results.append(result)
    ctx.set("results", results)

    current = len(results)
    ctx.set("current_step", current)

    # Check if workflow is done
    steps = await ctx.get("steps", [])
    if current >= len(steps):
        ctx.set("status", "completed")

    return {"completed": current, "total": len(steps)}

@workflow.shared
async def get_progress(ctx) -> dict:
    """Get workflow progress."""
    return {
        "current_step": await ctx.get("current_step", 0),
        "total_steps": len(await ctx.get("steps", [])),
        "status": await ctx.get("status", "unknown")
    }

Consistency & Concurrency

Single-Writer Per Key

Only one write operation per entity key executes at a time:

# Same key = serial execution (consistency guaranteed)
await ctx.entity("agent", "conv-1").send_message("msg1")  # Runs first
await ctx.entity("agent", "conv-1").send_message("msg2")  # Runs second

# No race conditions, no lost updates

Parallel Execution Across Keys

Different entity keys execute in parallel:

# Different keys = parallel execution (scales horizontally)
await ctx.entity("agent", "conv-1").send_message(msg)  # Parallel
await ctx.entity("agent", "conv-2").send_message(msg)  # Parallel
await ctx.entity("agent", "conv-3").send_message(msg)  # Parallel

Shared Methods for Reads

Use @entity.shared for read-only operations that can run concurrently:

# Multiple shared calls can run in parallel for same key
@agent.shared
async def get_history(ctx) -> list:
    return await ctx.get("history", [])

# These execute concurrently
await ctx.entity("agent", "conv-1").get_history()  # Concurrent
await ctx.entity("agent", "conv-1").get_history()  # Concurrent

Best Practices

1. Choose Stable, Meaningful Keys

Use unique, stable identifiers for entity keys:

# Descriptive and stable
"agent-conv-{conversation_id}"
"workflow-{run_id}"
"user-{user_id}"
"research-{task_id}"

# Not descriptive
"abc123"

# Changes every time
"user-{timestamp}"

# Too generic
"agent-1"

2. Design for Concurrency

Choose key granularity for optimal parallelism:

# ✓ Good - One entity per conversation
await ctx.entity("ChatAgent", f"conv-{conv_id}").send_message(msg)

# ✗ Bad - Single global entity (serializes everything)
await ctx.entity("ChatAgent", "global").send_message(msg)

3. Use Shared for Read Operations

Enable concurrent reads with @entity.shared:

# Write methods - exclusive access
@agent.write
async def update_state(ctx, data: dict):
    ctx.set("state", data)

# Read methods - concurrent access
@agent.shared
async def get_state(ctx) -> dict:
    return await ctx.get("state", {})

4. Keep State Minimal

Store only what you need:

# ✓ Good - Essential state only
ctx.set("history", recent_messages[-20:])
ctx.set("summary", summary_text)

# ✗ Avoid - Excessive state
ctx.set("full_transcript", all_messages)  # Could be huge
ctx.set("raw_responses", all_llm_responses)  # Redundant

Entity Use Cases

Functions vs Entities

When to use Functions:

• Stateless operations
• Independent requests
• Data transformations
• API integrations

When to use Entities:

• Stateful AI agents with memory
• Workflow orchestration
• User sessions and context
• Any state that needs consistency

Next Steps

• Context API - Entity state operations and APIs
• Functions - Stateless operations
• Workflows - Multi-step orchestration
• Agent Component - AI agents built on entities

Workflows enable durable, multi-step orchestration with automatic recovery, state persistence, and complex dependency management. Built on the AGNT5 orchestration plane for exactly-once execution guarantees.

Basic Workflow

Simple Sequential Workflow

from agnt5 import workflow, task_step
from agnt5.workflows import FlowDefinition

@workflow()
def data_pipeline() -> FlowDefinition:
    return FlowDefinition([
        task_step(
            name="extract",
            service_name="etl-service",
            handler_name="extract_data"
        ),
        task_step(
            name="transform",
            service_name="etl-service",
            handler_name="transform_data",
            dependencies=["extract"]
        ),
        task_step(
            name="load",
            service_name="etl-service",
            handler_name="load_data",
            dependencies=["transform"]
        )
    ])

Parallel Execution

Steps without dependencies execute in parallel:

@workflow()
def parallel_processing() -> FlowDefinition:
    return FlowDefinition([
        # These three steps run in parallel
        task_step(
            name="process_a",
            service_name="service",
            handler_name="process_type_a"
        ),
        task_step(
            name="process_b",
            service_name="service",
            handler_name="process_type_b"
        ),
        task_step(
            name="process_c",
            service_name="service",
            handler_name="process_type_c"
        ),
        # This step waits for all three to complete
        task_step(
            name="merge_results",
            service_name="service",
            handler_name="merge_results",
            dependencies=["process_a", "process_b", "process_c"]
        )
    ])

Step Types

Task Steps

Execute function handlers on services:

from agnt5.workflows import task_step

# Basic task step
step = task_step(
    name="unique_step_name",
    service_name="my-service",
    handler_name="my_handler"
)

# Task with dependencies and input data
step = task_step(
    name="dependent_step",
    service_name="my-service",
    handler_name="process_data",
    dependencies=["previous_step"],
    input_data={"config": "production", "batch_size": 100}
)

# Task with object keys (Phase 2)
step = task_step(
    name="object_step",
    service_name="object-service",
    handler_name="update_state",
    dependencies=["init_step"],
    object_keys=["user:123", "cart:456"]
)

Wait Signal Steps

Pause execution until external signals:

from agnt5.workflows import wait_signal_step

# Basic signal wait
step = wait_signal_step(
    name="wait_for_approval",
    signal_name="approval_granted",
    dependencies=["review_step"]
)

# Signal wait with timeout
step = wait_signal_step(
    name="wait_with_timeout",
    signal_name="user_action",
    dependencies=["prompt_user"],
    timeout_ms=300000,  # 5 minutes
    on_timeout="timeout_handler_step"
)

Wait Timer Steps

Scheduled delays and cron-based execution:

from agnt5.workflows import wait_timer_step

# Fixed delay
step = wait_timer_step(
    name="delay_before_retry",
    timer_key="retry_delay",
    delay_ms=30000,  # 30 seconds
    dependencies=["failed_step"]
)

# Cron schedule
step = wait_timer_step(
    name="nightly_batch",
    timer_key="nightly",
    cron_expr="0 2 * * *",  # 2 AM daily
    dependencies=["prep_step"]
)

# Timer with retries
step = wait_timer_step(
    name="retry_with_backoff",
    timer_key="exponential_backoff",
    delay_ms=5000,
    max_retries=3,
    dependencies=["error_step"]
)

Workflow Examples

ETL Pipeline

@workflow()
def nightly_etl() -> FlowDefinition:
    return FlowDefinition([
        # Start with data validation
        task_step(
            name="validate_sources",
            service_name="etl-service",
            handler_name="validate_data_sources"
        ),

        # Extract from multiple sources in parallel
        task_step(
            name="extract_database",
            service_name="etl-service",
            handler_name="extract_from_database",
            dependencies=["validate_sources"]
        ),
        task_step(
            name="extract_api",
            service_name="etl-service",
            handler_name="extract_from_api",
            dependencies=["validate_sources"]
        ),
        task_step(
            name="extract_files",
            service_name="etl-service",
            handler_name="extract_from_files",
            dependencies=["validate_sources"]
        ),

        # Wait for all extractions to complete
        task_step(
            name="merge_extracted_data",
            service_name="etl-service",
            handler_name="merge_data",
            dependencies=["extract_database", "extract_api", "extract_files"]
        ),

        # Transform data
        task_step(
            name="clean_data",
            service_name="etl-service",
            handler_name="clean_and_normalize",
            dependencies=["merge_extracted_data"]
        ),
        task_step(
            name="enrich_data",
            service_name="etl-service",
            handler_name="enrich_with_metadata",
            dependencies=["clean_data"]
        ),

        # Wait for maintenance window
        wait_timer_step(
            name="wait_for_maintenance_window",
            timer_key="maintenance",
            cron_expr="0 3 * * *",  # 3 AM
            dependencies=["enrich_data"]
        ),

        # Load data
        task_step(
            name="load_to_warehouse",
            service_name="etl-service",
            handler_name="load_data_warehouse",
            dependencies=["wait_for_maintenance_window"]
        ),

        # Generate reports
        task_step(
            name="generate_reports",
            service_name="reporting-service",
            handler_name="generate_daily_reports",
            dependencies=["load_to_warehouse"]
        )
    ])

Approval Workflow

@workflow()
def document_approval() -> FlowDefinition:
    return FlowDefinition([
        # Submit document for review
        task_step(
            name="submit_document",
            service_name="doc-service",
            handler_name="submit_for_review",
            input_data={"priority": "normal"}
        ),

        # Notify reviewers
        task_step(
            name="notify_reviewers",
            service_name="notification-service",
            handler_name="send_review_notifications",
            dependencies=["submit_document"]
        ),

        # Wait for approval (with timeout)
        wait_signal_step(
            name="wait_for_approval",
            signal_name="document_approved",
            dependencies=["notify_reviewers"],
            timeout_ms=172800000,  # 48 hours
            on_timeout="escalate_approval"
        ),

        # Escalation path
        task_step(
            name="escalate_approval",
            service_name="doc-service",
            handler_name="escalate_to_manager",
            # No dependencies - triggered by timeout
        ),

        # Wait for escalated approval
        wait_signal_step(
            name="wait_escalated_approval",
            signal_name="escalated_approval",
            dependencies=["escalate_approval"],
            timeout_ms=86400000  # 24 hours
        ),

        # Publish approved document
        task_step(
            name="publish_document",
            service_name="doc-service",
            handler_name="publish_approved_document",
            dependencies=["wait_for_approval", "wait_escalated_approval"]
        )
    ])

Workflow Registration

Using the Decorator

from agnt5 import workflow
from agnt5.workflows import get_registered_workflows

@workflow()
def my_workflow() -> FlowDefinition:
    return FlowDefinition([...])

@workflow("custom_name")
def workflow_with_custom_name() -> FlowDefinition:
    return FlowDefinition([...])

# Inspect registered workflows
workflows = get_registered_workflows()
print(f"Registered workflows: {list(workflows.keys())}")

Manual Registration

from agnt5.workflows import register_workflow

def create_workflow_definition() -> FlowDefinition:
    return FlowDefinition([
        task_step("step1", service_name="service", handler_name="handler1"),
        task_step("step2", service_name="service", handler_name="handler2",
                 dependencies=["step1"])
    ])

# Manual registration
register_workflow("manual_workflow", create_workflow_definition())

Workflow Data Classes

FlowDefinition

Container for workflow steps with serialization:

from agnt5.workflows import FlowDefinition

# Create definition
flow = FlowDefinition([
    task_step("step1", service_name="svc", handler_name="h1"),
    task_step("step2", service_name="svc", handler_name="h2", dependencies=["step1"])
])

# Serialize to dictionary
flow_dict = flow.to_dict()

# Serialize to JSON string
flow_json = flow.to_json()

WorkflowStep

Individual step configuration:

from agnt5.workflows import WorkflowStep, StepType

# Manual step creation (usually use helper functions instead)
step = WorkflowStep(
    name="custom_step",
    step_type=StepType.TASK,
    service_name="my-service",
    handler_name="my_handler",
    dependencies=["previous_step"],
    input_data={"key": "value"}
)

Configuration Classes

from agnt5.workflows import SignalConfig, TimerConfig

# Signal configuration
signal_config = SignalConfig(
    name="approval_signal",
    timeout_ms=3600000,  # 1 hour
    on_timeout="timeout_step"
)

# Timer configuration
timer_config = TimerConfig(
    key="batch_timer",
    delay_ms=60000,  # 1 minute
    max_retries=5
)

# Cron timer configuration
cron_timer = TimerConfig(
    key="daily_job",
    cron_expr="0 0 * * *",  # Daily at midnight
    max_retries=3
)

Validation and Error Handling

Workflow Validation

AGNT5 validates workflows during registration:

from agnt5.workflows import register_workflow, FlowDefinition, task_step

# This will raise ValueError: Missing dependencies
try:
    invalid_flow = FlowDefinition([
        task_step("step2", service_name="svc", handler_name="h2",
                 dependencies=["step1"]),  # step1 doesn't exist
        task_step("step3", service_name="svc", handler_name="h3",
                 dependencies=["step2"])
    ])
    register_workflow("invalid", invalid_flow)
except ValueError as e:
    print(f"Validation error: {e}")

Validation Rules

• At least one step must be defined
• Step names must be unique within the workflow
• Dependencies must reference existing steps
• Dependencies must appear earlier in the definition (causal order)
• Required fields must be populated based on step type

Testing Workflows

Workflow Definition Testing

import pytest
from agnt5.workflows import FlowDefinition, task_step, get_registered_workflows

def test_workflow_definition():
    # Test workflow structure
    flow = FlowDefinition([
        task_step("extract", service_name="etl", handler_name="extract"),
        task_step("transform", service_name="etl", handler_name="transform",
                 dependencies=["extract"])
    ])

    # Verify serialization
    flow_dict = flow.to_dict()
    assert len(flow_dict["steps"]) == 2
    assert flow_dict["steps"][1]["dependencies"] == ["extract"]

def test_workflow_registration():
    @workflow()
    def test_workflow() -> FlowDefinition:
        return FlowDefinition([
            task_step("test_step", service_name="test", handler_name="test")
        ])

    # Verify registration
    workflows = get_registered_workflows()
    assert "test_workflow" in workflows

Integration Testing

Test workflows with a local development environment:

import asyncio
from agnt5 import Client

async def test_workflow_execution():
    client = Client("http://localhost:8080")

    # Trigger workflow
    result = await client.start_workflow(
        workflow_name="data_pipeline",
        input_data={"source": "test_data"}
    )

    workflow_id = result["workflow_id"]

    # Poll for completion
    while True:
        status = await client.get_workflow_status(workflow_id)
        if status["state"] in ["completed", "failed"]:
            break
        await asyncio.sleep(1)

    assert status["state"] == "completed"

Best Practices

Design Patterns

1. Idempotent Steps - Design steps to be safely retryable
2. Small Steps - Break complex operations into smaller, focused steps
3. Clear Dependencies - Make step relationships explicit
4. Meaningful Names - Use descriptive names for steps and workflows
5. Error Handling - Plan for failure and recovery scenarios

Performance

1. Parallel Execution - Remove unnecessary dependencies to enable parallelism
2. Batch Operations - Group related operations into single steps
3. Resource Management - Consider resource usage when designing workflows
4. State Minimization - Keep workflow state as small as possible

Monitoring

1. Structured Logging - Add logging to workflow steps
2. Progress Tracking - Use meaningful step names and descriptions
3. Metrics Collection - Track workflow success rates and durations
4. Error Alerting - Set up alerts for workflow failures

Next Steps

• Worker Runtime - Configure and deploy workers
• API Reference - Complete workflows API reference
• Examples - Real-world workflow patterns

The Context (ctx) is the execution environment provided to all AGNT5 components. It provides APIs for orchestration, state management, LLM interactions, coordination, and observability.

Core Capabilities

• Orchestration - Execute tasks, spawn functions, parallel execution
• State Management - Get/set/delete state for entities
• Coordination - Signals, timers, human approvals
• AI Integration - LLM calls, tool registration
• Observability - Logging, metrics, tracing

Orchestration APIs

Task Execution

Execute functions and wait for results (workflows only):

@workflow
async def process_workflow(ctx):
    # Execute a task
    result = await ctx.task(
        service_name="analytics",
        handler_name="process_data",
        input={"dataset": "users"}
    )

    return result

Parallel Execution

Run multiple tasks concurrently:

# Returns list of results in order
results = await ctx.parallel(
    ctx.task("service1", "handler1"),
    ctx.task("service2", "handler2"),
    ctx.task("service3", "handler3")
)

# Access results by index
result1 = results[0]
result2 = results[1]

# Returns dict with named results
results = await ctx.gather(
    db=ctx.task("analytics", "analyze_db"),
    api=ctx.task("analytics", "analyze_api"),
    cache=ctx.task("analytics", "analyze_cache")
)

# Access results by name
db_result = results["db"]
api_result = results["api"]

Async Invocation

Spawn child functions without waiting:

@function
async def batch_processor(ctx, items: list):
    # Spawn child invocations
    handles = []
    for item in items:
        handle = ctx.spawn(process_item, item, key=f"item-{item['id']}")
        handles.append(handle)

    # Continue other work...

    # Wait for results later if needed
    results = [await h.result() for h in handles]
    return {"processed": len(results)}

Checkpointing

Checkpoint expensive operations (functions only):

@function
async def process_pipeline(ctx, data_id: str):
    # Each step is checkpointed
    raw = await ctx.step("extract", lambda: extract_data(data_id))
    cleaned = await ctx.step("clean", lambda: clean_data(raw))
    result = await ctx.step("analyze", lambda: analyze(cleaned))

    # If crash occurs, resumes from last completed step
    return result

State Management (Entities)

Get, Set, Delete

Manage entity state:

@entity.write
async def update_profile(ctx, name: str, age: int):
    # Get with default
    profile = await ctx.get("profile", {})

    # Update profile
    profile.update({"name": name, "age": age})

    # Set state
    ctx.set("profile", profile)
    ctx.set("last_updated", datetime.now().isoformat())

    # Delete temporary data
    ctx.delete("temp_cache")

    return {"status": "updated"}

Entity Method Calls

Call entity methods from functions:

@function
async def chat(ctx, conversation_id: str, message: str):
    # Call entity method
    response = await ctx.entity(
        "ChatAgent",
        conversation_id
    ).send_message(message)

    return response

Coordination APIs

Signals

Wait for external events:

@workflow
async def approval_workflow(ctx, document_id: str):
    # Submit for review
    await ctx.task("docs", "submit_review", input={"id": document_id})

    # Wait for approval signal (24 hours timeout)
    approval = await ctx.signal(
        "manager_approved",
        timeout_ms=86400000,
        default={"approved": False}
    )

    if approval["approved"]:
        await ctx.task("docs", "publish", input={"id": document_id})
        return {"status": "published"}
    else:
        return {"status": "rejected"}

@function
async def approve_document(ctx, workflow_id: str):
    # Send approval signal to waiting workflow
    await ctx.signal.emit(
        "manager_approved",
        target_workflow_id=workflow_id,
        payload={"approved": True, "approver": "manager@example.com"}
    )

    return {"status": "signal_sent"}

Timers & Sleep

Add delays and scheduled execution:

@workflow
async def scheduled_job(ctx):
    # Wait 5 seconds
    await ctx.timer(delay_ms=5000)

    # Or use sleep (alternative syntax)
    await ctx.sleep(30)  # 30 seconds

    # Wait until specific time (cron)
    await ctx.timer(cron="0 0 * * *")  # Daily at midnight

    return {"status": "completed"}

Human-in-the-Loop

Request human approval:

@workflow
async def deployment_workflow(ctx, version: str):
    # Run tests
    test_results = await ctx.task("ci", "run_tests", input={"version": version})

    if test_results["passed"]:
        # Request human approval for production
        approval = await ctx.human.approval(
            "deploy_production",
            payload={"version": version, "tests": test_results},
            timeout=timedelta(minutes=30),
            required_roles=["admin", "devops"]
        )

        if approval.decision == "approved":
            await ctx.task("deploy", "to_production", input={"version": version})
            return {"status": "deployed"}

    return {"status": "cancelled"}

AI Integration

LLM Generation

Generate text or structured responses:

@function
async def summarize(ctx, text: str):
    response = await ctx.llm.generate(
        prompt=f"Summarize this text: {text}",
        model="gpt-4o-mini"
    )

    return {"summary": response.text}

@function
async def chat_response(ctx, messages: list):
    response = await ctx.llm.generate(
        prompt=[
            {"role": "system", "content": "You are a helpful assistant"},
            *messages
        ],
        model="gpt-4"
    )

    return {"response": response.text}

@function
async def extract_info(ctx, text: str):
    # JSON Schema-constrained output
    response = await ctx.llm.generate(
        prompt=f"Extract information: {text}",
        schema={
            "type": "object",
            "properties": {
                "name": {"type": "string"},
                "age": {"type": "integer"},
                "email": {"type": "string"}
            },
            "required": ["name"]
        },
        model="gpt-4o-mini"
    )

    return response.object  # Parsed JSON

Streaming

Stream responses for real-time output:

@function
async def stream_story(ctx, topic: str):
    # Stream text generation
    async for chunk in await ctx.llm.stream(
        prompt=f"Write a story about {topic}",
        model="gpt-4o"
    ):
        if chunk.text:
            yield chunk.text  # Stream to client

Tool Registration

Register tools for LLM use:

@function
async def agent_with_tools(ctx, query: str):
    # Register search tool
    search_tool = ctx.tools.register(
        "web_search",
        handler=perform_search,
        description="Search the web for information",
        schema={
            "type": "object",
            "properties": {
                "query": {"type": "string"},
                "max_results": {"type": "integer"}
            }
        }
    )

    # Generate with tool
    response = await ctx.llm.generate(
        prompt=query,
        tools=[search_tool],
        model="gpt-4o"
    )

    return response

Observability

Logging

Structured logging with context:

@function
async def tracked_operation(ctx, data: dict):
    logger = ctx.log()

    logger.info("Processing started", extra={"data_size": len(data)})

    try:
        result = process(data)
        logger.info("Processing completed", extra={"result_size": len(result)})
        return result
    except Exception as e:
        logger.error("Processing failed", exc_info=True)
        raise

Metrics

Record custom metrics:

@function
async def monitored_function(ctx, request: dict):
    metrics = ctx.metrics()

    # Increment counter
    metrics.increment("requests.count", service="api")

    # Record timing
    start = time.time()
    result = await process_request(request)
    duration = (time.time() - start) * 1000

    metrics.observe("latency.ms", duration, endpoint="/api/process")

    return result

Distributed Tracing

Create spans for tracing:

@function
async def traced_operation(ctx, data: dict):
    # Create span for external API call
    with ctx.trace_span().start("external_api_call", service="payments"):
        result = await call_payment_api(data)

    # Create span for database operation
    with ctx.trace_span().start("database_query", service="postgres"):
        await save_to_db(result)

    return result

Configuration & Secrets

Secrets

Access secrets securely:

@function
async def api_call(ctx, endpoint: str):
    # Get API key from secrets
    api_key = ctx.secrets().get("openai_api_key")
    db_password = ctx.secrets().get("database_password")

    # Use secrets in API calls
    response = await make_request(endpoint, api_key=api_key)
    return response

Configuration

Feature flags and config:

@function
async def feature_gated_handler(ctx, data: dict):
    config = ctx.config()

    # Check feature flag
    if config.get("new_feature_enabled", default=False):
        return await new_implementation(data)
    else:
        return await legacy_implementation(data)

    # A/B testing variant
    variant = config.variant("experiment_group", default="control")
    if variant == "treatment":
        return await experimental_flow(data)

Request Headers

Access incoming headers:

@function
async def header_aware(ctx, data: dict):
    headers = ctx.headers()

    user_agent = headers.get("user-agent", "unknown")
    correlation_id = headers.get("x-correlation-id")

    logger = ctx.log()
    logger.info(f"Request from {user_agent}", extra={"correlation_id": correlation_id})

    return {"processed": True}

Context Properties

Access execution metadata:

@function
async def introspective(ctx, data: dict):
    return {
        "run_id": ctx.run_id,              # Workflow/run identifier
        "step_id": ctx.step_id,            # Current step identifier
        "attempt": ctx.attempt,            # Retry attempt number
        "component_type": ctx.component_type,  # "function", "entity", "workflow"
        "object_id": ctx.object_id,        # Entity key (for entities)
        "method_name": ctx.method_name,    # Entity method name (for entities)
        "processed": data
    }

API Reference

Orchestration

State (Entities)

Coordination

AI Integration

Observability

Configuration

Common Patterns

Parallel with Error Handling

@workflow
async def robust_workflow(ctx):
    results = await ctx.gather(
        task1=ctx.task("svc", "task1"),
        task2=ctx.task("svc", "task2")
    )

    if results["task1"] and results["task2"]:
        return {"status": "success", "results": results}
    else:
        return {"status": "partial_failure"}

Conditional Signal Waiting

@workflow
async def conditional_approval(ctx, needs_approval: bool):
    if needs_approval:
        approval = await ctx.signal("approval_signal", timeout_ms=60000)
        if not approval.get("approved"):
            return {"status": "rejected"}

    # Proceed with operation
    result = await ctx.task("service", "operation")
    return {"status": "completed", "result": result}

LLM with Tool Execution

@function
async def agent_handler(ctx, query: str):
    # Register tools
    search = ctx.tools.register("search", handler=search_web, ...)
    calc = ctx.tools.register("calculator", handler=calculate, ...)

    # Generate with tools
    response = await ctx.llm.generate(
        prompt=query,
        tools=[search, calc],
        model="gpt-4o"
    )

    # Execute tool calls if needed
    if response.tool_calls:
        for tool_call in response.tool_calls:
            handler = ctx.tools.handler(tool_call.name)
            await handler(**tool_call.arguments)

    return response

Next Steps

• Functions - Using context in functions
• Entity - Using context in entities
• Workflows - Using context in workflows
• Agent - AI integration with context

Agents are autonomous LLM-driven systems that reason, plan, and execute tasks using tools. They orchestrate complex multi-step workflows by breaking down problems, selecting appropriate tools, and iterating until complete.

Key Characteristics

• LLM-Powered - Driven by language models for reasoning and decision-making
• Tool Orchestration - Automatically selects and executes appropriate tools
• Memory Integration - Maintains long-term knowledge across conversations
• Session Aware - Uses sessions for conversation context
• Streaming Support - Real-time event streaming for responsive UX
• Durable - Built on AGNT5 primitives for automatic fault tolerance

Basic Usage

Simple Agent

from agnt5 import Agent, LanguageModel

lm = LanguageModel()

agent = Agent(
    name="assistant",
    model=lm,
    instructions="You are a helpful coding assistant."
)

# Run agent
result = await agent.run("Explain recursion")
print(result.output)

Agent with Tools

from agnt5 import Agent, tool, LanguageModel

@tool.function(auto_schema=True)
def search_docs(query: str, language: str = "python") -> List[Dict]:
    """Search programming language documentation."""
    # Implementation
    return search_results

@tool.function(auto_schema=True)
def run_code(code: str, language: str = "python") -> Dict[str, str]:
    """Execute code and return output."""
    # Implementation
    return {"output": result}

lm = LanguageModel()
agent = Agent(
    name="coding_assistant",
    model=lm,
    instructions="""You are a coding assistant.
    Use search_docs to find API references.
    Use run_code to test code examples.""",
    tools=[search_docs, run_code]
)

result = await agent.run("How do I read a file in Python? Show me an example.")

Agent with Session and Memory

from agnt5 import Agent, Session, Memory, LanguageModel

# Create session
session = Session(
    id="tutoring-session-789",
    user_id="student-123",
    metadata={"subject": "mathematics"}
)

# Create memory
memory = Memory(service=VectorMemoryService())
await memory.store("student_level", "Advanced calculus")

# Create agent
lm = LanguageModel()
agent = Agent(
    name="math_tutor",
    model=lm,
    instructions="You are a patient math tutor. Adapt to student's level.",
    tools=[solve_equation_tool, plot_function_tool],
    session=session,
    memory=memory
)

result = await agent.run("Help me understand limits")

Agent Configuration

Parameters

Instructions

Write clear, actionable instructions:

agent = Agent(
    name="code_reviewer",
    model=lm,
    instructions="""You are an expert code reviewer specializing in Python.

    Review process:
    1. Analyze code for complexity, duplication, style
    2. Check for security vulnerabilities
    3. Suggest improvements with code examples
    4. Prioritize: security > correctness > performance > style

    Be constructive and explain your reasoning."""
)

agent = Agent(
    name="helper",
    model=lm,
    instructions="Help the user with stuff."  # Too vague
)

Streaming Agents

Stream events in real-time:

async for event in agent.stream("Analyze this dataset", session=session):
    match event.type:
        case "thinking":
            print(f"🤔 {event.content}")
        case "tool_call":
            print(f"🔧 Calling {event.tool_name}({event.arguments})")
        case "tool_result":
            print(f"✓ Result: {event.result}")
        case "response":
            print(f"💬 {event.content}")
        case "error":
            print(f"❌ Error: {event.error}")

Agent Planning

Preview execution plan before running:

# Get plan without executing
plan = agent.plan("Analyze competitor pricing strategies")

print(f"Estimated steps: {len(plan.steps)}")
for step in plan.steps:
    print(f"- {step.type}: {step.description}")
    if step.tool:
        print(f"  Tool: {step.tool.name}")

# Review and execute if approved
if user_approves(plan):
    result = await agent.run("Analyze competitor pricing strategies")

Common Patterns

Research Agent

from agnt5 import Agent, Session, Memory, tool

@tool.function(auto_schema=True)
def search_academic(query: str, year_from: int = 2020) -> List[Dict]:
    """Search academic papers."""
    pass

@tool.function(auto_schema=True)
def extract_insights(paper_text: str) -> Dict[str, List[str]]:
    """Extract key insights from paper."""
    pass

# Create research agent
session = Session(id="research-ai-safety-001", user_id="researcher-123")
memory = Memory(service=VectorMemoryService())

lm = LanguageModel()
research_agent = Agent(
    name="research_agent",
    model=lm,
    instructions="""You are a research assistant specializing in AI safety.

    Research process:
    1. Search for relevant recent papers
    2. Extract key insights from each paper
    3. Identify common themes and gaps
    4. Synthesize findings into comprehensive summary""",
    tools=[search_academic, extract_insights],
    session=session,
    memory=memory
)

result = await research_agent.run(
    "Survey the current state of AI alignment research"
)

# Store findings in memory
await memory.ingest_from_session(session, strategy="smart")