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 durable runtime for AI agents, built for Conductor. Three pillars:

Long-running agents — Write an agent; it runs as long as it needs to. Minutes, hours, or until a human approves the next step. No timeout by default. If your worker process crashes, the server resumes from the last completed step when a new worker connects.

Dynamic agents (Plan-Execute) — The LLM decides what to do at runtime; Conductor locks it in and executes it deterministically. The planner emits a JSON plan once; the server compiles it into an immutable Conductor sub-workflow — no LLM randomness in orchestration, retries, or parallelism. Dynamic agents can call existing Conductor workflows as steps, bridging AI with your existing automation. → Strategy.PLAN_EXECUTE · works across Python, TypeScript, Java, C#

Event-driven agents — Trigger agents from cron schedules, Kafka topics, SQS queues, AMQP messages, webhooks, and database events. Agentspan runs on Conductor, so every event source Conductor supports is available to agents. Each trigger is a durable execution with full history. → deploy(agent, schedules=[Schedule(cron="0 0 9 MON-FRI")]) · Conductor event handlers

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

Build on Agentspan with the Conductor Agent SDK, available for Python, TypeScript/JavaScript, and C#/.NET:

# Python
pip install conductor-agent-sdk
TypeScript / JavaScript
npm install @conductor-oss/conductor-agent-sdk
C# / .NET
dotnet add package conductor-agent-sdk

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

# hello.py — run with: python hello.py
from conductor.ai.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, 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.

| | CrewAI | LangChain | AutoGen | OpenAI Agents | Agentspan | |---|---|---|---|---|---| | Execution model | In-memory | Checkpoints | In-memory | Client-side loop | Server-side executions | | Crash recovery | Manual replay | Checkpointer (Postgres) | None | None | Automatic resume | | Tool scaling | Single process | Single process | Distributed | Single process | Distributed workers (any language) | | Human approval | Stdin-blocking | interrupt() + checkpointer | Stdin-blocking | In-process | Durable pause (days, any machine) | | Orchestration API | Crew, Task, Agent, Flow | StateGraph, Node, Edge | AssistantAgent, GroupChat | Agent, Runner, Handoff | One class: Agent | | Pipeline syntax | YAML + Python | Graph builder API | Nested class hierarchy | Handoff chains | a >> b >> c | | Guardrails | Task guardrails | Middleware-based | Limited | Input/output/tool | Custom, regex, LLM — 4 failure modes | | Code execution | Docker sandbox | Community packages | Docker, Jupyter | Hosted interpreter | 4 built-in sandboxes | | MCP tools | Manual config | Manual config | Manual config | Manual config | Auto-discovered, server-side |

Code Examples

Agent with Tools

from conductor.ai.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],     instructi, )
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 conductor.ai.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=[getweather], outputtype=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 GITHUBTOKEN ghpxxxxxxxxxxxx
agentspan credentials set SEARCHAPIKEY 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 conductor.ai.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=["SEARCHAPIKEY"])
def search(query: str) -> dict:
    """Search using API key."""
    key = getcredential("SEARCHAPI_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 conductor.ai.agents import httptool, mcptool
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 = mcptool(serverurl="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 conductor.ai.agents import Agent, AgentRuntime, tool
@tool def checkbalance(accountid: str) -> dict:     """Check account balance."""     return {"accountid": accountid, "balance": 5432.10}
billing = Agent(name="billing", model="openai/gpt-4o",                 instructi, tools=[check_balance]) technical = Agent(name="technical", model="openai/gpt-4o",                   instructi)
support = Agent(     name="support", model="openai/gpt-4o",     instructi,     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 conductor.ai.agents import Agent, AgentRuntime
researcher = Agent(name="researcher", model="openai/gpt-4o",                    instructi) writer = Agent(name="writer", model="openai/gpt-4o",                instructi) editor = Agent(name="editor", model="openai/gpt-4o",                instructi)
pipeline = researcher >> writer >> editor
with AgentRuntime() as runtime:     result = runtime.run(pipeline, "AI agents in software development")     result.print_result()

Parallel Agents

from conductor.ai.agents import Agent, AgentRuntime
market = Agent(name="market", model="openai/gpt-4o",                instructi) risk = Agent(name="risk", model="openai/gpt-4o",              instructi)
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 conductor.ai.agents import Agent, AgentRuntime, tool
@tool(approval_required=True) def transferfunds(fromacct: 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 conductor.ai.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(wordlimit, onfail=OnFail.RETRY)], )
with AgentRuntime() as runtime:     result = runtime.run(agent, "Explain quantum computing.")     result.print_result()

Streaming

from conductor.ai.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 "toolcall":       print(f"Calling {event.toolname}...")             case "thinking":        print(f"Thinking: {event.content}")             case "guardrailpass":  print(f"Guardrail passed: {event.guardrailname}")             case "guardrailfail":  print(f"Guardrail failed: {event.guardrailname}")             case "done":            print(f"\n{event.output}")

Server-Side Tools (No Workers Needed)

from conductor.ai.agents import Agent, AgentRuntime, apitool, httptool, 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)
weatherapi = httptool(
    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 = mcptool(serverurl="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 conductor.ai.agents import Agent, AgentRuntime
from conductor.ai.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()],     instructi, )
with AgentRuntime() as runtime:     result = runtime.run(agent, "Calculate the first 20 Fibonacci numbers.")     result.print_result()

Shared State (Tool Context)

from conductor.ai.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=[additem, getitems],     instructi, )
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 conductor.ai.agents import Agent, AgentRuntime, CallbackHandler
class TimingHandler(CallbackHandler):     def onagentstart(self, **kwargs):         self.t0 = time.time()     def onagentend(self, **kwargs):         print(f"Took {time.time() - self.t0:.2f}s")
class LoggingHandler(CallbackHandler):     def onmodelstart(self, , messages=None, *kwargs):         print(f"Sending {len(messages or [])} messages")     def onmodelend(self, , llm_result=None, *kwargs):         print(f"LLM responded: {(llm_result or '')[:80]}")
agent = Agent(     name="my_agent",     model="anthropic/claude-sonnet-4-6",     instructi,     callbacks=[TimingHandler(), LoggingHandler()], )
with AgentRuntime() as runtime:     result = runtime.run(agent, "Hello!")     result.print_result()

Six hook positions: onagentstart, onagentend, onmodelstart, onmodelend, ontoolstart, ontoolend.

Execution order: onagentstart → (onmodelstart → LLM → onmodelend)* → onagentend

Multi-Agent Strategies

| Strategy | Description | |---|---| | handoff (default) | LLM chooses which sub-agent handles the request | | sequential | Sub-agents run in order, output feeds forward (>> operator) | | parallel | All sub-agents run concurrently, results aggregated | | router | Router agent or function selects the sub-agent | | round_robin | Agents take turns in a fixed rotation | | swarm | Condition-based handoffs between agents | | random | Random sub-agent selection each turn | | manual | Human selects which agent speaks each turn |

Examples

180+ runnable examples covering every feature across 5 frameworks:

| Example | Description | |---|---| | 01basic_agent.py | Hello world | | 02_tools.py | Multiple tools with approval | | 02asimple_tools.py | Two tools, LLM picks the right one | | 02bmultistep_tools.py | Chained lookups and calculations | | 03structured_output.py | Pydantic output types | | 04httpandmcp_tools.py | Server-side HTTP and MCP tools | | 04mcp_weather.py | MCP server tools (live weather) | | 05_handoffs.py | Agent delegation | | 06sequential_pipeline.py | agent >> agent >> agent | | 07parallel_agents.py | Fan-out / fan-in | | 08router_agent.py | LLM routing to specialists | | 09humaninthe_loop.py | Approval patterns | | 09bhitlwith_feedback.py | Custom feedback (respond API) | | 09chitl_streaming.py | Streaming + HITL approval | | 10_guardrails.py | Output validation + retry | | 11_streaming.py | Real-time events | | 12long_running.py | Fire-and-forget with polling | | 13hierarchical_agents.py | Nested agent teams | | 14existing_workers.py | Existing workers as tools | | 15agent_discussion.py | Round-robin debate | | 16random_strategy.py | Random agent selection | | 17swarm_orchestration.py | Swarm with handoff conditions | | 18manual_selection.py | Human picks which agent speaks | | 19composable_termination.py | Composable termination conditions | | 20constrained_transitions.py | Restricted agent transitions | | 21regex_guardrails.py | RegexGuardrail (block/allow) | | 22llm_guardrails.py | LLMGuardrail (AI judge) | | 23token_tracking.py | Token usage and cost tracking | | 24code_execution.py | Code execution sandboxes | | 25semantic_memory.py | Long-term memory with retrieval | | 26opentelemetry_tracing.py | OpenTelemetry spans | | 28gptassistant_agent.py | OpenAI Assistants API wrapper | | 29agent_introductions.py | Agents introduce themselves | | 30multimodal_agent.py | Vision model analysis | | 31tool_guardrails.py | Pre-execution tool validation | | 32human_guardrail.py | Human review on guardrail failure | | 33external_workers.py | Workers in other services | | 33singleturn_tool.py | Single-turn tool call | | 34prompt_templates.py | Server-side prompt templates | | 35standalone_guardrails.py | Guardrails without agents | | 36simpleagent_guardrails.py | Guardrails on simple agents | | 37fix_guardrail.py | Auto-correct with | | 38tech_trends.py | Tech trends research | | 39localcode_execution.py | Local code sandbox | | 39adockercode_execution.py | Docker-sandboxed execution | | 39bjupytercode_execution.py | Jupyter kernel execution | | 39cserverlesscode_execution.py | Serverless execution | | 40mediageneration_agent.py | Image/audio/video generation | | 41sequentialpipeline_tools.py | Pipeline with per-stage tools | | 42security_testing.py | Security testing pipeline | | 43datasecurity_pipeline.py | Data redaction pipeline | | 44safety_guardrails.py | PII detection and sanitization | | 45agent_tool.py | Agent as a callable tool | | 46transfer_control.py | Restricted handoff transitions | | 47_callbacks.py | Lifecycle hooks | | 48_planner.py | Planning before execution | | 49include_contents.py | Context control for sub-agents | | 50thinking_config.py | Extended reasoning | | 51shared_state.py | Shared state via ToolContext | | 52nested_strategies.py | Nested parallel + sequential | | 53agentlifecycle_callbacks.py | Agent-level before/after hooks | | 54softwarebug_assistant.py | Software debugging agent | | 55ml_engineering.py | ML engineering assistant | | 56rag_agent.py | Retrieval-augmented generation | | 57plandry_run.py | Plan execution preview | | 58scatter_gather.py | Massive parallel map-reduce | | 59coding_agent.py | Code generation agent | | 60githubcoding_agent.py | GitHub integration for coding | | 61githubcodingagent_chained.py | Chained GitHub operations | | 62clitool_guardrails.py | CLI tool input validation | | 63_deploy.py | Agent deployment | | 64swarmwith_tools.py | Swarm + tool orchestration | | 65parallelwith_tools.py | Parallel agents with tools | | 66handoffto_parallel.py | Handoff to parallel execution | | 67routerto_sequential.py | Router to sequential pipeline | | 68context_condensation.py | Auto-condense long conversations | | 70cesupport_agent.py | Full support agent with Zendesk, JIRA, HubSpot | | 71api_tool.py | Auto-discover tools from OpenAPI/Swagger/Postman |

Framework Examples:

| Framework | Count | Location | |---|---|---| | OpenAI Agents SDK | 10 examples | Handoffs, guardrails, streaming, multi-model | | Google ADK | 35 examples | Full ADK compatibility, all agent types | | LangChain | 25 examples | ReAct, memory, document analysis | | LangGraph | 44 examples | StateGraph, human-in-the-loop, subgraphs |

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

| Environment | Guide | |---|---| | Local (dev) | agentspan server start — zero config, SQLite | | Single server | Docker / Docker Compose | | Production | Kubernetes + Helm |

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