agentspan

name	agentspan
description	Comprehensive Agentspan durable workflow orchestration skill for creating, running, scheduling, monitoring, and debugging AI agent workflows using the Agentspan Python SDK and CLI. Covers agents, tools, multi-agent strategies, guardrails, memory, streaming, human-in-the-loop, testing, and production deployment.

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

Capability	Description
Create	Design and implement agents, tools, multi-agent strategies, guardrails, and memory systems
Run	Execute agent workflows locally or against a remote server
Schedule	Set up recurring agent runs and batch processing
Monitor	Stream events, query status, inspect execution history
Debug	Troubleshoot failed executions, replay events, analyze errors
Test	Mock agent behavior, record/replay, pytest integration

When to Use Agentspan

Use Case	Use Agentspan	Use Dagu	Use Temporal
AI agent orchestration	✅ Perfect fit	❌ Not designed for agents	⚠️ Overkill, no agent primitives
Multi-agent pipelines (research→write→edit)	✅ Built-in strategies	⚠️ Possible but manual	⚠️ No agent primitives
Human-in-the-loop approval flows	✅ Built-in HITL	⚠️ Manual approval gates	✅ Signals/queries
LLM tool-calling workflows	✅ First-class @tool support	❌ No LLM integration	❌ No LLM integration
Cron-style task scheduling	✅ Agent + CLI	✅ Perfect fit	✅ Built-in
Simple shell script orchestration	❌ Overkill	✅ Perfect fit	❌ Overkill
Durable business process (multi-day)	⚠️ Can work	❌ Not designed for it	✅ Designed for it
Batch document processing	✅ Parallel agents	✅ Sequential steps	⚠️ Overkill

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

Strategy	Description	Key Use
handoff	LLM chooses which sub-agent handles the request	Customer support, routing
sequential	Sub-agents run in order, output feeds forward	Research→Write→Edit pipelines
parallel	All sub-agents run concurrently	Multi-perspective analysis
router	Dedicated classifier selects sub-agent	Cost-optimized routing
swarm	Condition-based handoffs between agents	Complex multi-step support
round_robin	Agents take turns in fixed rotation	Debates, discussions
random	Random sub-agent each turn	Diverse output ensembles
manual	Human selects which agent runs next	Human-directed workflows

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.",
    on_fail=OnFail.RETRY,
    max_retries=2,
)

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

OnFail Modes

Mode	Behavior
OnFail.RETRY	Append feedback message and re-run the LLM (up to max_retries)
OnFail.RAISE	Fail the execution immediately
OnFail.FIX	Replace output with GuardrailResult.fixed_output
OnFail.HUMAN	Pause for human review (creates a WaitTask)

Input vs Output Guardrails

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

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

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

Variable	Default	Description
SERVER_PORT	6767	Server port
SPRING_PROFILES_ACTIVE	default (SQLite)	Set to postgres for PostgreSQL
SPRING_DATASOURCE_URL	jdbc:sqlite:agent-runtime.db	Database URL
AGENTSPAN_SERVER_URL	http://localhost:6767	Server URL for SDK/workers
AGENTSPAN_AUTH_KEY	—	Auth key (required for non-localhost)
AGENTSPAN_AUTH_SECRET	—	Auth secret (required for non-localhost)

LLM Providers

Set environment variables for the providers you need:

Provider	Env Var	Model Prefix
OpenAI	OPENAI_API_KEY	openai/
Anthropic	ANTHROPIC_API_KEY	anthropic/
Google Gemini	GEMINI_API_KEY + GOOGLE_CLOUD_PROJECT	google_gemini/
Azure OpenAI	AZURE_OPENAI_API_KEY + endpoint + deployment	azure_openai/
AWS Bedrock	AWS_ACCESS_KEY_ID + AWS_SECRET_ACCESS_KEY	aws_bedrock/
Mistral	MISTRAL_API_KEY	mistral/
Ollama	OLLAMA_BASE_URL	ollama/
DeepSeek	DEEPSEEK_API_KEY	deepseek/
Grok/xAI	XAI_API_KEY	grok/
Cohere	COHERE_API_KEY	cohere/

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

Feature	Agentspan	LangGraph	CrewAI	AutoGen
Durable execution	✅ Server-managed state	❌ In-process only	❌ In-process	❌ In-process
Crash recovery	✅ Automatic	❌ Lost on crash	❌ Lost on crash	❌ Lost on crash
Server UI	✅ Built-in	❌	❌	❌
Human-in-the-loop	✅ First-class	⚠️ Manual	⚠️ Manual	⚠️ Manual
Testing	✅ mock_run, record/replay	❌	❌	❌
Multi-agent strategies	✅ 8 built-in	⚠️ Manual graph	✅ Built-in	✅ Built-in
LLM provider support	✅ 15+	✅ Via LangChain	✅	✅
Code execution	✅ Docker/Jupyter/Local	❌	❌	⚠️ Limited
Credential management	✅ Server-side encrypted	❌	❌	❌

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

Issue	Solution
Server won't start	Run agentspan doctor. Ensure Java 11+ is installed.
Worker can't connect	Check AGENTSPAN_SERVER_URL. Default: http://localhost:6767
LLM API errors	Verify API keys: agentspan doctor. Check rate limits.
Tool timeout	Increase timeout_seconds in @tool(). Check network connectivity.
Agent runs forever	Set max_turns lower. Add stop_when condition.
Guardrail loops	Set max_retries on guardrails. Use OnFail.RAISE for critical checks.
Worker process dies	Execution continues on server. Reconnect with AgentHandle(workflow_id=...).
Port 6767 in use	Set SERVER_PORT environment variable or kill existing process.

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