agentspan

AI agents that don't die when your process does.

Docs • Quickstart • 52+ Examples • Discord • API Reference

Agentspan is a distributed, durable runtime for running AI agents that survive crashes, scale across machines, and pause for human approval for days — not minutes.

Agentspan is the execution layer, not the replacement. Use native Agentspan agents, or bring LangGraph, the OpenAI Agents SDK, or Google ADK — pass your existing agent to runtime.run() and it gains crash recovery, human-in-the-loop pauses, and full execution history. Your definitions stay unchanged.

from agentspan.agents import Agent, AgentRuntime, tool

@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"72F and sunny in {city}"

agent = Agent(name="weatherbot", model="openai/gpt-4o", tools=[get_weather])

with AgentRuntime() as runtime:
    result = runtime.run(agent, "What's the weather in NYC?")
    result.print_result()

Why Agentspan?

Other frameworks give you a Python library. Agentspan gives you a production runtime.

Your agent code compiles to a durable, server-side execution. The server manages execution, retries, scaling, and state — so your agents keep running even when your process doesn't.

What makes it different

1. True durable execution — Not checkpoints. Not client-side loops. Your agent compiles to a server-side execution that the Agentspan server executes independently of your process. Deploy new code, restart your machine, kill the process — the agent keeps running. When it finishes, poll for the result from anywhere. This is the same execution model that powers mission-critical systems at scale.

2. Cross-process agent access — Every running agent has an execution ID. Any process, on any machine, can use that ID to check status, stream events, approve or reject tool calls, pause, resume, or cancel the agent. No graph rebuilding, no checkpointer setup — just the ID and a runtime connection. LangGraph requires re-instantiating the graph and checkpointer; CrewAI and AutoGen have no cross-process access at all.

3. Distributed workers in any language — Tools don't run inside your agent process. They execute as distributed tasks that workers pick up. Write workers in Python, Java, Go, or any language. Scale each tool independently. Load-balance automatically. Your agent process just submits work — the server and workers handle the rest.

4. One primitive — No Crew, Task, StateGraph, Node, or AssistantAgent. Everything is an Agent. Single agents, multi-agent teams, nested hierarchies — one class.

5. The >> operator — Compose pipelines with Python syntax: researcher >> writer >> editor. No YAML, no graph builders.

6. Real human-in-the-loop — @tool(approval_required=True) pauses the execution durably on the server. No process stays alive waiting. Approve from any machine, any process, days later.

7. Production guardrails — Custom functions, regex patterns, or LLM judges. Four failure modes: retry, raise, fix, or escalate to human. Guardrails are durable tasks, not post-processing — they survive execution restarts.

8. Server-side tools — HTTP endpoints and MCP servers execute as server-side tasks. No worker process needed. MCP tools are auto-discovered at compile time.

9. Code execution sandboxes — Local subprocess, Docker containers, Jupyter kernels, or serverless functions. Four options, built in.

10. Full observability — OpenTelemetry spans, Prometheus metrics, visual execution UI, execution history, and token/cost tracking — all built in.

11. Framework agnostic — Use Google ADK, Langchain, OpenAI, CrewAI etc to write agents, run on Agentspan's durable execution runtime.

Quickstart

Install

uv venv && source .venv/bin/activate
uv pip install agentspan

Start the Server

The SDK auto-starts the server when needed, but you can also start it manually (recommended):

# Set the API key for your LLM provider:
export OPENAI_API_KEY=sk-...          # For OpenAI models (gpt-4o, gpt-4o-mini, etc.)
# export ANTHROPIC_API_KEY=sk-ant-... # For Anthropic models (claude-sonnet, etc.)
# export GOOGLE_API_KEY=...           # For Google models (gemini, etc.)

agentspan server start   # Start the Agentspan server
agentspan server stop    # Stop the server
agentspan server logs    # View server logs

export AGENTSPAN_SERVER_URL=http://localhost:6767/api

cp .env.example .env
# Edit .env with your server URL and API keys

Hello World

from agentspan.agents import Agent, AgentRuntime

agent = Agent(name="hello", model="openai/gpt-4o")

with AgentRuntime() as runtime:
    result = runtime.run(agent, "Say hello and tell me a fun fact.")
    result.print_result()

Add Tools

from agentspan.agents import Agent, AgentRuntime, tool

@tool
def get_weather(city: str) -> dict:
    """Get current weather for a city."""
    return {"city": city, "temp": 72, "condition": "Sunny"}

@tool
def calculate(expression: str) -> dict:
    """Evaluate a math expression."""
    return {"result": eval(expression)}

agent = Agent(
    name="assistant",
    model="openai/gpt-4o",
    tools=[get_weather, calculate],
    instructions="You are a helpful assistant.",
)

with AgentRuntime() as runtime:
    result = runtime.run(agent, "What's the weather in NYC? Also, what's 42 * 17?")
    result.print_result()

Structured Output

from pydantic import BaseModel
from agentspan.agents import Agent, AgentRuntime, tool

class WeatherReport(BaseModel):
    city: str
    temperature: float
    condition: str
    recommendation: str

@tool
def get_weather(city: str) -> dict:
    """Get weather data for a city."""
    return {"city": city, "temp_f": 72, "condition": "Sunny", "humidity": 45}

agent = Agent(name="reporter", model="openai/gpt-4o", tools=[get_weather], output_type=WeatherReport)

with AgentRuntime() as runtime:
    result = runtime.run(agent, "What's the weather in NYC?")
    report: WeatherReport = result.output  # Fully typed
    print(f"{report.city}: {report.temperature}F, {report.condition}")

Multi-Agent Handoffs

from agentspan.agents import Agent, AgentRuntime, tool

@tool
def check_balance(account_id: str) -> dict:
    """Check account balance."""
    return {"account_id": account_id, "balance": 5432.10}

billing = Agent(name="billing", model="openai/gpt-4o",
                instructions="Handle billing inquiries.", tools=[check_balance])
technical = Agent(name="technical", model="openai/gpt-4o",
                  instructions="Handle technical issues.")

support = Agent(
    name="support", model="openai/gpt-4o",
    instructions="Route customer requests to the right team.",
    agents=[billing, technical],
    strategy="handoff",
)

with AgentRuntime() as runtime:
    result = runtime.run(support, "What's the balance on account ACC-123?")
    result.print_result()

Pipeline Composition

from agentspan.agents import Agent, AgentRuntime

researcher = Agent(name="researcher", model="openai/gpt-4o",
                   instructions="Research the topic and provide key facts.")
writer = Agent(name="writer", model="openai/gpt-4o",
               instructions="Write an engaging article from the research.")
editor = Agent(name="editor", model="openai/gpt-4o",
               instructions="Polish the article for publication.")

pipeline = researcher >> writer >> editor

with AgentRuntime() as runtime:
    result = runtime.run(pipeline, "AI agents in software development")
    result.print_result()

Parallel Agents

from agentspan.agents import Agent, AgentRuntime

market = Agent(name="market", model="openai/gpt-4o",
               instructions="Analyze market size, growth, key players.")
risk = Agent(name="risk", model="openai/gpt-4o",
             instructions="Analyze regulatory, technical, competitive risks.")

analysis = Agent(name="analysis", model="openai/gpt-4o",
                 agents=[market, risk], strategy="parallel")

with AgentRuntime() as runtime:
    result = runtime.run(analysis, "Launching an AI healthcare tool in the US")
    result.print_result()

Human-in-the-Loop (Durable)

from agentspan.agents import Agent, AgentRuntime, tool

@tool(approval_required=True)
def transfer_funds(from_acct: str, to_acct: str, amount: float) -> dict:
    """Transfer funds. Requires human approval."""
    return {"status": "completed", "amount": amount}

agent = Agent(name="banker", model="openai/gpt-4o", tools=[transfer_funds])

with AgentRuntime() as runtime:
    handle = runtime.start(agent, "Transfer $5000 from checking to savings")
    # Execution pauses at transfer_funds...

    # Days later, from any process, any machine:
    status = handle.get_status()
    if status.is_waiting:
        handle.approve()   # Or: handle.reject("Amount too high")

Guardrails

from agentspan.agents import Agent, AgentRuntime, Guardrail, GuardrailResult, OnFail, guardrail

@guardrail
def word_limit(content: str) -> GuardrailResult:
    """Keep responses concise."""
    if len(content.split()) > 500:
        return GuardrailResult(passed=False, message="Too long. Be more concise.")
    return GuardrailResult(passed=True)

agent = Agent(
    name="concise_bot", model="openai/gpt-4o",
    guardrails=[Guardrail(word_limit, on_fail=OnFail.RETRY)],
)

with AgentRuntime() as runtime:
    result = runtime.run(agent, "Explain quantum computing.")
    result.print_result()

Streaming

from agentspan.agents import Agent, AgentRuntime

agent = Agent(name="writer", model="openai/gpt-4o")

with AgentRuntime() as runtime:
    for event in runtime.stream(agent, "Write a haiku about Python"):
        match event.type:
            case "tool_call":       print(f"Calling {event.tool_name}...")
            case "thinking":        print(f"Thinking: {event.content}")
            case "guardrail_pass":  print(f"Guardrail passed: {event.guardrail_name}")
            case "guardrail_fail":  print(f"Guardrail failed: {event.guardrail_name}")
            case "done":            print(f"\n{event.output}")

Server-Side Tools (No Workers Needed)

from agentspan.agents import Agent, AgentRuntime, http_tool, mcp_tool

weather_api = http_tool(
    name="get_weather", description="Get weather for a city",
    url="https://api.weather.com/v1/current", method="GET",
    input_schema={"type": "object", "properties": {"city": {"type": "string"}}},
)

github = mcp_tool(server_url="http://localhost:6767/mcp")  # Auto-discovered

agent = Agent(name="assistant", model="openai/gpt-4o", tools=[weather_api, github])

with AgentRuntime() as runtime:
    result = runtime.run(agent, "What's the weather in NYC?")
    result.print_result()

Code Execution

from agentspan.agents import Agent, AgentRuntime
from agentspan.agents.code_executor import DockerCodeExecutor

executor = DockerCodeExecutor(image="python:3.12-slim", timeout=30)
agent = Agent(
    name="coder", model="openai/gpt-4o",
    tools=[executor.as_tool()],
    instructions="Write and execute Python code to solve problems.",
)

with AgentRuntime() as runtime:
    result = runtime.run(agent, "Calculate the first 20 Fibonacci numbers.")
    result.print_result()

Shared State (Tool Context)

from agentspan.agents import Agent, AgentRuntime, tool, ToolContext

@tool
def add_item(item: str, context: ToolContext) -> str:
    """Add an item to the shared list."""
    items = context.state.get("items", [])
    items.append(item)
    context.state["items"] = items
    return f"Added '{item}'. List now has {len(items)} items."

@tool
def get_items(context: ToolContext) -> str:
    """Get all items from the shared list."""
    items = context.state.get("items", [])
    return f"Items: {', '.join(items)}" if items else "No items yet."

agent = Agent(
    name="list_manager", model="openai/gpt-4o",
    tools=[add_item, get_items],
    instructions="Manage a shared list of items.",
)

with AgentRuntime() as runtime:
    result = runtime.run(agent, "Add apples, bananas, and cherries, then show the list.")
    result.print_result()

Agent Lifecycle Callbacks

Hook into agent, model, and tool lifecycle events with CallbackHandler classes. Multiple handlers chain per-position in list order — each one handles a single concern:

import time
from agentspan.agents import Agent, AgentRuntime, CallbackHandler

class TimingHandler(CallbackHandler):
    def on_agent_start(self, **kwargs):
        self.t0 = time.time()
    def on_agent_end(self, **kwargs):
        print(f"Took {time.time() - self.t0:.2f}s")

class LoggingHandler(CallbackHandler):
    def on_model_start(self, *, messages=None, **kwargs):
        print(f"Sending {len(messages or [])} messages")
    def on_model_end(self, *, llm_result=None, **kwargs):
        print(f"LLM responded: {(llm_result or '')[:80]}")

agent = Agent(
    name="my_agent",
    model="openai/gpt-4o-mini",
    instructions="You are a helpful assistant.",
    callbacks=[TimingHandler(), LoggingHandler()],
)

with AgentRuntime() as runtime:
    result = runtime.run(agent, "Hello!")
    result.print_result()

Six hook positions: on_agent_start, on_agent_end, on_model_start, on_model_end, on_tool_start, on_tool_end.

Execution order: on_agent_start → (on_model_start → LLM → on_model_end)* → on_agent_end

Multi-Agent Strategies

Examples

Runnable examples covering every feature:

Google ADK Compatibility

Drop-in compatibility with the Google ADK API, backed by durable execution. 32 examples included.

from google.adk.agents import Agent, SequentialAgent

researcher = Agent(name="researcher", model="gemini-2.0-flash",
                   instruction="Research the topic.", tools=[search])
writer = Agent(name="writer", model="gemini-2.0-flash",
               instruction="Write an article from the research.")

pipeline = SequentialAgent(name="pipeline", sub_agents=[researcher, writer])

Community

We're building Agentspan in the open and would love your help.

• Discord — Ask questions, share what you're building, get help
• GitHub Issues — Bug reports and feature requests
• Contributing Guide — How to contribute code, docs, and examples

Contributing

git clone https://github.com/agentspan-ai/agentspan.git
cd agentspan/sdk/python
uv venv && source .venv/bin/activate
uv pip install -e ".[dev]"
pytest

We welcome PRs of all sizes — from typo fixes to new examples to core features.

Spread the Word

If Agentspan is useful to you, help others find it:

• Star this repo — it helps more than you think
• Share on LinkedIn — tell your network
• Share on X/Twitter — spread the word
• Share on Reddit — post in r/MachineLearning or r/LocalLLaMA

API Reference

See API Reference for the complete API reference and architecture guide.

License

MIT