Agentspan

Agentspan: Durable AI Agent Workflow Orchestration

Expert AI Agent for Agentspan Workflows Create, run, schedule, monitor, and debug durable AI agent workflows using the Agentspan Python SDK and CLI.

What I Do

I am an expert Agentspan workflow architect and developer. I help you build reliable, scalable, and fault-tolerant AI agent systems using Agentspan's durable execution runtime.

Agentspan guarantees your agent workflows complete successfully, even through:

• Worker process crashes and restarts
• Server failures and deployments
• Long-running operations (hours/days/weeks)
• Human-in-the-loop approval delays
• Network partitions and timeouts

Core Capabilities

When to Use Agentspan

Quick Start

# Install
pip install agentspan

# Verify setup
agentspan doctor

# Set API key
export OPENAI_API_KEY=sk-...

# Start the server (downloads JAR on first run)
agentspan server start
# UI at http://localhost:6767

# Run your first agent
python hello.py

# hello.py
from agentspan.agents import Agent, AgentRuntime, tool

@tool
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"72°F 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()

Architecture Overview

┌─────────────────────────────────────────────────────┐
│                  Agentspan Server                     │
│         (port 6767, SQLite or PostgreSQL)            │
│                                                      │
│  ┌──────────┐  ┌──────────┐  ┌──────────────────┐  │
│  │ Workflow  │  │ Task     │  │ Execution History │  │
│  │ Engine    │  │ Queue    │  │ & State Store     │  │
│  └──────────┘  └──────────┘  └──────────────────┘  │
│         │             │                │              │
└─────────┼─────────────┼────────────────┼───────────┘
          │             │                │
          ▼             ▼                ▼
┌──────────────────────────────────────────────────────┐
│               Python Worker Process                   │
│                                                       │
│  ┌─────────┐  ┌─────────┐  ┌─────────┐             │
│  │ @tool   │  │ http_   │  │ mcp_    │             │
│  │ Python  │  │ tool()  │  │ tool()  │             │
│  │ funcs   │  │ (server)│  │ (server)│             │
│  └─────────┘  └─────────┘  └─────────┘             │
│                                                       │
│  AgentRuntime manages the worker lifecycle            │
└──────────────────────────────────────────────────────┘

Key principle: @tool functions run in your Python worker process (full access to code, libraries, local state). http_tool(), api_tool(), mcp_tool() run server-side (no code to write).

CLI Reference

See references/cli-quick-reference.md for the full CLI command reference.

# Server management
agentspan server start          # Start server (downloads JAR on first run)
agentspan server stop           # Stop server
agentspan server logs            # View server logs

# Credentials
agentspan credentials set KEY value    # Store encrypted credential
agentspan credentials list             # List credential keys
agentspan credentials delete KEY       # Delete a credential

# Agent operations
agentspan agent status <exec-id>       # Check execution status
agentspan agent list                    # List registered agents
agentspan agent run --name my_agent "prompt"  # Run and stream
agentspan agent execution --since 1h    # View execution history

# Diagnostics
agentspan doctor                       # Check system dependencies

# Configuration
agentspan configure --url https://server.example.com

Agent Definition

from agentspan.agents import Agent, AgentRuntime

agent = Agent(
    name="my_agent",                    # Unique name (becomes workflow name)
    model="openai/gpt-4o",              # Provider/model format
    instructions="You are helpful.",    # System prompt (str or callable)
    tools=[...],                        # @tool functions or ToolDefs
    agents=[...],                        # Sub-agents for multi-agent
    strategy="handoff",                  # Multi-agent strategy
    router=None,                         # Router agent or function
    output_type=None,                    # Pydantic BaseModel for structured output
    guardrails=[...],                    # Input/output validation
    memory=None,                         # ConversationMemory or SemanticMemory
    dependencies={...},                  # Injected into ToolContext
    max_turns=25,                        # Max agent loop iterations
    max_tokens=None,                     # LLM max tokens
    temperature=None,                    # LLM temperature
    stop_when=None,                      # Early termination condition
    metadata=None,                       # Arbitrary metadata
)

Tools

@tool — Custom Python Functions

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(name="custom_name", approval_required=True, timeout_seconds=60)
def dangerous_action(target: str) -> dict:
    """Do something that requires human approval."""
    return {"done": True}

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

ToolContext — Shared State and Dependencies

from agentspan.agents import tool, ToolContext

@tool
def query_database(query: str, context: ToolContext) -> dict:
    """Run a database query."""
    db = context.dependencies["db"]
    user_id = context.dependencies["user_id"]
    return db.execute(query, user=user_id)

agent = Agent(
    name="bot", model="openai/gpt-4o",
    tools=[query_database],
    dependencies={"db": my_database, "user_id": "u-123"},
)

http_tool — HTTP Endpoints (Server-Side)

from agentspan.agents import http_tool

weather = http_tool(
    name="get_weather",
    description="Get weather for a city",
    url="https://api.weather.com/v1/current",
    method="GET",
    headers={"Authorization": "Bearer ${WEATHER_KEY}"},
    input_schema={
        "type": "object",
        "properties": {"city": {"type": "string"}},
        "required": ["city"],
    },
    credentials=["WEATHER_KEY"],
)

api_tool — OpenAPI Auto-Discovery (Server-Side)

from agentspan.agents import api_tool

stripe = api_tool(
    url="https://api.stripe.com/openapi.json",
    headers={"Authorization": "Bearer ${STRIPE_KEY}"},
    credentials=["STRIPE_KEY"],
    max_tools=20,  # LLM auto-filters to most relevant
)

mcp_tool — MCP Servers (Server-Side)

from agentspan.agents import mcp_tool

github = mcp_tool(
    server_url="http://localhost:3001/mcp",
    name="github",
    description="GitHub operations",
)

Credential Management

# Store credentials (encrypted at rest with AES-256-GCM)
agentspan credentials set GITHUB_TOKEN ghp_xxxxxxxxxxxx
agentspan credentials set STRIPE_API_KEY sk_live_xxxxxxxx

# Option A: Isolated subprocess (credentials as env vars)
@tool(credentials=["GITHUB_TOKEN"])
def list_repos(username: str) -> dict:
    import os
    token = os.environ["GITHUB_TOKEN"]  # Auto-injected

# Option B: In-process (use get_credential)
@tool(isolated=False, credentials=["SEARCH_API_KEY"])
def search(query: str) -> dict:
    from agentspan.agents import get_credential
    key = get_credential("SEARCH_API_KEY")

Multi-Agent Strategies

Sequential Pipeline (a >> b >> c)

researcher = Agent(name="researcher", model="openai/gpt-4o",
                   instructions="Research the topic thoroughly.")
writer = Agent(name="writer", model="openai/gpt-4o",
               instructions="Write an 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 2025")

Parallel

market = Agent(name="market", model="openai/gpt-4o",
               instructions="Analyze market size and growth.")
risk = Agent(name="risk", model="openai/gpt-4o",
             instructions="Analyze risks.")
financial = Agent(name="financial", model="openai/gpt-4o",
                  instructions="Analyze financial projections.")

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

with AgentRuntime() as runtime:
    result = runtime.run(analysis, "Launching an AI healthcare tool")
    print(result.sub_results["market"])
    print(result.sub_results["risk"])
    print(result.sub_results["financial"])

Handoff (Default)

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

Router

# Router agent classifies and dispatches
classifier = Agent(
    name="classifier",
    model="openai/gpt-4o-mini",
    instructions="Classify: billing, technical, or general. Reply with just the category.",
)

support = Agent(
    name="support",
    model="openai/gpt-4o",
    agents=[billing, technical, general],
    strategy="router",
    router=classifier,
)

# Or use a Python function as router
def route(prompt: str) -> str:
    if "bill" in prompt.lower():
        return "billing"
    elif "error" in prompt.lower():
        return "technical"
    return "general"

support = Agent(name="support", agents=[billing, technical, general],
                strategy="router", router=route)

Swarm (Condition-Based Handoffs)

from agentspan.agents import Agent, AgentRuntime, Strategy
from agentspan.agents import TextMentionTermination

triage = Agent(name="triage", model="openai/gpt-4o",
               instructions="Triage requests. Say BILLING or TECH.")
billing = Agent(name="billing", model="openai/gpt-4o",
                instructions="Handle billing.")
technical = Agent(name="technical", model="openai/gpt-4o",
                  instructions="Handle technical issues.")

team = Agent(
    name="support_team",
    model="openai/gpt-4o",
    agents=[triage, billing, technical],
    strategy=Strategy.SWARM,
    handoffs=[
        TextMentionTermination("BILLING", target="billing"),
        TextMentionTermination("TECH", target="technical"),
    ],
)

Strategy Summary

Guardrails

from agentspan.agents import (
    Agent, Guardrail, GuardrailResult, guardrail,
    OnFail, Position, RegexGuardrail, LLMGuardrail,
)

# Custom guardrail function
@guardrail
def no_pii(content: str) -> GuardrailResult:
    """Reject responses containing email addresses."""
    import re
    if re.search(r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}", content):
        return GuardrailResult(passed=False, message="Response contains PII (email). Remove it.")
    return GuardrailResult(passed=True)

# Regex guardrail
no_ssn = RegexGuardrail(
    patterns=[r"\b\d{3}-\d{2}-\d{4}\b"],
    name="no_ssn",
    message="Do not include SSNs.",
)

# LLM-as-judge guardrail
factual_check = LLMGuardrail(
    model="openai/gpt-4o-mini",
    policy="Is this response factually accurate? Reply YES or NO.",
   
    max_retries=2,
)

agent = Agent(
    name="safe_bot",
    model="openai/gpt-4o",
    guardrails=[
        Guardrail(no_pii, max_retries=3),
        no_ssn,
        factual_check,
    ],
)

OnFail Modes

Input vs Output Guardrails

# Input guardrail — validates user prompt before LLM call
Guardrail(no_jailbreak, position=Position.INPUT,

# Output guardrail — validates LLM response (default)
Guardrail(no_pii, position=Position.OUTPUT,

Memory

ConversationMemory — Chat History

from agentspan.agents import Agent, AgentRuntime, ConversationMemory

memory = ConversationMemory(max_messages=100)

agent = Agent(name="assistant", model="openai/gpt-4o",
              instructions="You are a helpful assistant.", memory=memory)

with AgentRuntime() as runtime:
    result = runtime.run(agent, "My name is Alice.")
    memory.add_user_message("My name is Alice.")
    memory.add_assistant_message(result.output['result'])

    result2 = runtime.run(agent, "What's my name?")
    # "Your name is Alice."

SemanticMemory — Long-Term Knowledge

from agentspan.agents import Agent, AgentRuntime, tool
from agentspan.agents.semantic_memory import SemanticMemory

memory = SemanticMemory(max_results=3)
memory.add("Customer prefers email communication.")
memory.add("Account is on the Enterprise plan since March 2021.")

@tool
def get_context(query: str) -> str:
    """Retrieve relevant context from memory."""
    return memory.get_context(query)

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

Human-in-the-Loop (HITL)

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 process_transfer(from_acct, to_acct, 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")
    for event in handle.stream():
        if event.type == "waiting":
            # Human approves or rejects
            handle.approve()  # or handle.reject("Too risky")

CLI approval:

agentspan agent respond <execution-id> --approve
agentspan agent respond <execution-id> --deny --reason "Amount too large"

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 "thinking":      print(f"Thinking: {event.content}")
            case "tool_call":     print(f"Calling {event.tool_name}({event.args})")
            case "tool_result":   print(f"Result: {event.result}")
            case "handoff":       print(f"Delegating to {event.target}")
            case "waiting":       print("Waiting for human approval...")
            case "guardrail_pass": print(f"Guardrail passed: {event.guardrail_name}")
            case "guardrail_fail": print(f"Guardrail failed: {event.guardrail_name}")
            case "message":        print(f"Message: {event.content}")
            case "error":          print(f"Error: {event.content}")
            case "done":           print(f"\nFinal: {event.output}")

Testing

from agentspan.agents import Agent, tool
from agentspan.agents.testing import mock_run, MockEvent, expect

@tool
def search_web(query: str) -> str:
    """Search the web."""
    return "results"

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

result = mock_run(
    agent,
    "What is agentspan?",
    events=[
        MockEvent.thinking("I should search for information about agentspan."),
        MockEvent.tool_call("search_web", {"query": "agentspan Python agent runtime"}),
        MockEvent.tool_result("search_web", "Agentspan is an open source Python runtime for AI agents."),
        MockEvent.done("Agentspan is an open source Python runtime for building AI agents."),
    ]
)

expect(result).completed().output_contains("Agentspan").used_tool("search_web")

Record and Replay

from agentspan.agents.testing import record, replay

# Record a real execution (calls LLM)
recording = record(agent, "What's the capital of France?")
recording.save("tests/fixtures/capital_query.json")

# Replay it deterministically (no LLM, no server)
result = replay("tests/fixtures/capital_query.json")
expect(result).completed().output_contains("Paris")

Deployment

Local Development (SQLite — Zero Setup)

agentspan server start
# Server at http://localhost:6767
# Data stored in agent-runtime.db

Production (PostgreSQL + Docker Compose)

cd deployment/docker-compose
cp .env.example .env
# Set OPENAI_API_KEY in .env
docker compose up -d

Kubernetes

# Helm chart available in deployment/helm/
# K8s manifests in deployment/k8s/

Configuration

LLM Providers

Set environment variables for the providers you need:

Common Patterns

See references/patterns.md for detailed pattern examples including:

• Research pipeline (sequential)
• Support ticket triage (handoff/router)
• Batch document processing (parallel)
• Crash and resume (durable execution)
• Human-in-the-loop approval
• Guardrail patterns
• Memory-augmented agents

Key Differences from Alternatives

Best Practices

1. Always set max_turns — Prevent runaway agents. Default is 25, lower for simple tasks.
2. Use guardrails for production — Input and output validation prevents safety issues.
3. Store credentials on the server — Never hardcode API keys. Use agentspan credentials set.
4. Test with mock_run — Write deterministic tests before deploying to production.
5. Use AgentRuntime context manager — Ensures worker processes are properly cleaned up.
6. Set stop_when for long pipelines — Custom termination conditions prevent unnecessary LLM calls.
7. Use http_tool and api_tool for external APIs — No worker process needed, runs server-side.
8. Use SemanticMemory for cross-session context — Better than stuffing conversation history.
9. Monitor with agentspan agent execution — Track failed executions in production.
10. Use approval_required=True for destructive tools — Always require human approval for irreversible actions.

Troubleshooting

References

• Official Docs: https://agentspan.ai/docs/
• GitHub: https://github.com/agentspan-ai/agentspan
• Examples: https://agentspan.ai/examples/
• PyPI: https://pypi.org/project/agentspan/
• Discord: https://discord.com/invite/ajcA66JcKq

Using Agentspan from RTerm (the agentspan-bridge plugin)

If you run RTerm / neuralOS (the AI-native terminal & ops platform), you don't have to drive Agentspan by hand — the agentspan-bridge plugin (v2.9.9+) wires it into RTerm's agent so any RTerm-managed host can launch durable Agentspan executions.

• Configure: RTerm Settings → AgentSpan → agentspan.serverUrl (default http://localhost:6767) + optional agentspan.authSecretRef (a vault key holding AGENTSPAN_AUTH_KEY/AGENTSPAN_AUTH_SECRET — never inline).
• RTerm agent tools: agentspan_health, agentspan_run (an AgentConfig or a registered Conductor workflow name → executionId), agentspan_status, agentspan_approve (HITL respond), agentspan_list, agentspan_stop.
• Trigger + panel: agentspan_execution_failed (fires on FAILED/TERMINATED/TIMED_OUT) and an agentspan-executions live dashboard feed.
• Why pair them: RTerm gives you the terminal/fleet/governance surface (SSH/WinRM, playbooks, MOP approval, audit, SRE observability); Agentspan gives those operations durability (resume-from-step on crash), plan-execute determinism, and Kafka/SQS/AMQP event triggers. Ask the RTerm agent: "run the disk-cleanup SOP as a durable agent on Agentspan."

See the rterm-gateway skill for the RPC surface and the rterm-backend/neuralos skills for backend setup.