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

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

⭐ If you find Agentspan useful, give us a star — it helps others find the project!

https://github.com/user-attachments/assets/dd4b720d-d11c-42e8-93a6-875c5a740fd8

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

Quickstart (60 seconds)

# macOS / Linux
curl -fsSL https://raw.githubusercontent.com/agentspan-ai/agentspan/main/cli/install.sh | sh

# Windows (PowerShell)
irm https://raw.githubusercontent.com/agentspan-ai/agentspan/main/cli/install.ps1 | iex

Install SDKs

# Python
pip install agentspan

# TypeScript / JavaScript
npm install @agentspan-ai/sdk

# C# / .NET
dotnet add package Agentspan

export OPENAI_API_KEY=sk-...   # or any supported provider
agentspan server start         # runs on localhost:6767 with UI

# hello.py — run with: python hello.py
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()

Open http://localhost:6767 to see the visual execution UI.

# npm
npm install -g @agentspan-ai/agentspan

# Windows — CMD / double-click
curl -fsSL https://raw.githubusercontent.com/agentspan-ai/agentspan/main/cli/install.bat -o install.bat && install.bat

# From source
cd cli && go build -o agentspan .

# Verify setup
agentspan doctor

Why Agentspan?

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.

Code Examples

Agent with 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

Credential Management

Store API keys and secrets once on the server. Tools resolve them automatically at runtime — no .env files, no hardcoded keys, no secrets in git.

Step 1: Store credentials on the server

agentspan credentials set GITHUB_TOKEN ghp_xxxxxxxxxxxx
agentspan credentials set SEARCH_API_KEY xxx-your-key

Credentials are encrypted at rest (AES-256-GCM). List them with agentspan credentials list.

Step 2: Declare which credentials a tool needs

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

# Default: tool runs in isolated subprocess with credentials as env vars
@tool(credentials=["GITHUB_TOKEN"])
def list_repos(username: str) -> dict:
    """List GitHub repos."""
    import os
    token = os.environ["GITHUB_TOKEN"]  # Auto-injected by the runtime
    return {"repos": ["repo1", "repo2"]}

# Alternative: access credentials in-process (no subprocess)
@tool(isolated=False, credentials=["SEARCH_API_KEY"])
def search(query: str) -> dict:
    """Search using API key."""
    key = get_credential("SEARCH_API_KEY")  # Resolve from server at runtime
    return {"results": ["result1"]}

Step 3: Run — credentials resolve automatically

agent = Agent(
    name="github_helper",
    model="openai/gpt-4o",
    tools=[list_repos, search],
    credentials=["GITHUB_TOKEN"],  # Agent-level credentials (shared with all tools)
)

with AgentRuntime() as runtime:
    result = runtime.run(agent, "List my GitHub repos and search for AI papers")
    result.print_result()

Credentials work with every tool type:

from agentspan.agents import http_tool, mcp_tool

# HTTP tools: server substitutes ${NAME} in headers at runtime
api = http_tool(
    name="weather_api", description="Get weather data",
    url="https://api.weather.com/v1/current",
    headers={"Authorization": "Bearer ${WEATHER_KEY}"},
    credentials=["WEATHER_KEY"],
)

# MCP tools: credentials passed to MCP server connection
github = mcp_tool(server_url="http://localhost:3001/mcp", credentials=["GITHUB_TOKEN"])

No credentials leave the server unencrypted. Workers resolve them via scoped execution tokens that expire with the execution. See the 11 credential examples (16_*.py through 16k_*.py) for every mode: isolated subprocess, in-process, CLI tools, HTTP headers, MCP, and framework passthrough.

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")

# 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, api_tool, http_tool, mcp_tool

# Point to any OpenAPI/Swagger spec — all endpoints auto-discovered
stripe = api_tool(
    url="https://api.stripe.com/openapi.json",
    headers={"Authorization": "Bearer ${STRIPE_KEY}"},
    credentials=["STRIPE_KEY"],
    max_tools=20,  # LLM auto-filters 300+ ops to top 20 most relevant
)

# Single HTTP endpoint (manual definition)
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"}}},
)

# MCP server tools (auto-discovered)
github = mcp_tool(server_url="http://localhost:6767/mcp")

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

with AgentRuntime() as runtime:
    result = runtime.run(agent, "Create a Stripe customer for alice@example.com")
    result.print_result()

Three ways to connect APIs — all server-side, no workers needed:

• api_tool() — point to an OpenAPI/Swagger/Postman spec, all endpoints auto-discovered
• http_tool() — define a single HTTP endpoint manually
• mcp_tool() — connect to an MCP server, tools auto-discovered

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

180+ runnable examples covering every feature across 5 frameworks:

Framework Examples:

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])

Deployment

Full deployment guide → deployment/README.md

Project Structure

├── cli/                  # Go CLI (agentspan server start/stop/logs)
├── server/               # Java runtime server (Spring Boot + Conductor)
│   └── src/
├── deployment/
│   ├── k8s/              # Kubernetes manifests
│   ├── helm/             # Helm chart
│   └── docker-compose/   # Compose stack (single node)
├── ui/                   # React execution UI (served at localhost:6767)
├── sdk/
│   ├── python/           # Python SDK
│   │   ├── src/agentspan/agents/
│   │   ├── examples/     # 70+ progressive examples
│   │   └── validation/   # Multi-model validation framework
│   └── typescript/       # TypeScript SDK
│       ├── src/
│       └── examples/
└── docs/                 # Consolidated documentation
    ├── sdk-design/       # Multi-language SDK design specs
    ├── python-sdk/       # Python SDK reference docs
    └── server/           # Server documentation

CLI Reference

agentspan server start     # Start the Agentspan server
agentspan server stop      # Stop the server
agentspan server logs      # View server logs
agentspan doctor           # Check system dependencies

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