[!IMPORTANT]

Agentspan is now part of Orkes Conductor

As of August 17, 2026, Agentspan has merged into Orkes Conductor and this repository is archived (read-only).

• New users: start with the Conductor agent quickstart
• Existing users: your agents keep running — see the migration guide (for most code: one install line + one import path)
• Final release: v0.4.4 — release assets remain downloadable; pip install agentspan stays installable (final: 0.2.1)
• Issues & contributions: conductor-oss/conductor

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 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 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.

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],
    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 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=[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 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=["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 conductor.ai.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 conductor.ai.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 conductor.ai.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 conductor.ai.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 conductor.ai.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 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(word_limit, on_fail=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 "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 conductor.ai.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 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()],
    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 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=[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 conductor.ai.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="anthropic/claude-sonnet-4-6",
    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