# agentspan — Machine-Readable API Reference # Version: 0.1.0 | Updated: 2026-04-04 # Canonical source: https://agentspan.ai/llms.txt # Facts verified by live testing on 2026-03-17 unless marked [UNVERIFIED] ## Identity name: agentspan tagline: Durable runtime for AI agents. Your code runs in your process. Execution state lives on the server. license: MIT github: https://github.com/agentspan-ai/agentspan skills_repo: https://github.com/agentspan-ai/agentspan-skills pypi: agentspan npm: @agentspan-ai/agentspan built_by: The team at Orkes (orkes.io) python_requires: ">=3.10" ## Install # Python SDK + CLI (recommended for Python developers — gives you everything) pip install agentspan # Installs the Python SDK AND a CLI shim that downloads the Go binary on first use. # After pip install, `agentspan server start`, `agentspan doctor`, all commands work. # CLI only (no Python SDK — for users who only need the CLI) npm install -g @agentspan-ai/agentspan # Downloads the same Go binary eagerly at install time (not lazily on first use). # Both pip and npm deliver the same Go binary from S3. pip is sufficient for Python devs. # Version: 0.1.0 # AI coding agent skills (full SDK knowledge in your agent context) # Claude Code (plugin marketplace): # /plugin marketplace add agentspan-ai/agentspan-skills # /plugin install agentspan@agentspan-skills # All other agents (Claude Code, Codex, Gemini CLI, Cursor, Windsurf, Cline, Copilot, Aider): # curl -sSL https://raw.githubusercontent.com/agentspan-ai/agentspan-skills/main/install.sh | bash -s -- --all ## Imports from agentspan.agents import Agent, tool, run, start, stream, configure, AgentRuntime, AgentHandle from agentspan.agents import run_async, start_async, stream_async, shutdown, plan from agentspan.agents import Strategy, Status, FinishReason, TokenUsage from agentspan.agents import TextMentionTermination, MaxMessageTermination, StopMessageTermination, TokenUsageTermination from agentspan.agents import CallbackHandler, ConversationMemory, SemanticMemory from agentspan.agents import http_tool, mcp_tool, agent_tool, search_tool, index_tool from agentspan.agents import image_tool, audio_tool, video_tool, pdf_tool, api_tool, human_tool from agentspan.agents.testing import mock_run, MockEvent, expect ## configure() # Must be called before run/start/stream if not using default local server. # Both forms work — the SDK auto-appends /api if missing (config.py __post_init__). configure(server_url="http://localhost:6767/api") # explicit — preferred configure(server_url="http://localhost:6767") # also works — SDK appends /api configure(server_url="https://your-server.example.com/api") # remote server configure(server_url=..., auth_key="key", auth_secret="secret") # with auth # Default local server port: 6767 (NOT 8080) shutdown() # stop singleton workers — call on exit when using module-level API ## Core API # Blocking — waits for completion result = run(agent, "prompt") # → AgentResult # Fire-and-forget — returns immediately handle = start(agent, "prompt") # → AgentHandle # Streaming — returns iterable of AgentEvent stream_obj = stream(agent, "prompt") # → AgentStream (iterable of AgentEvent) # Async variants — same signatures, add await result = await run_async(agent, "prompt") handle = await start_async(agent, "prompt") s = await stream_async(agent, "prompt") async for event in s: # correct async streaming pattern ... # WRONG: async with stream_async(...) — not supported # Dry-run compile (no execution) plan(agent) # → prints plan, no network call needed # AgentRuntime context manager (enables token_usage, explicit server binding) with AgentRuntime(server_url="...", auth_key="...", auth_secret="...") as runtime: result = run(agent, "prompt") # token_usage populated here runtime.serve(agent, blocking=False) # register workers (required before AgentHandle reconnect) ## Agent class parameters Agent( name="my-agent", # required; str; used as workflow name model="openai/gpt-4o", # required; str; provider/model format tools=[...], # list of tool functions or built-in tool objects instructions="...", # system prompt agents=[...], # sub-agents for multi-agent strategies strategy=Strategy.HANDOFF, # default is HANDOFF (not SEQUENTIAL) router=classifier_agent, # only for strategy=Strategy.ROUTER handoffs=[...], # only for strategy=Strategy.SWARM output_type=MyModel, # Pydantic model class; result.output becomes an instance max_turns=25, # default 25 (not 20) timeout_seconds=0, # default 0 = no timeout (not 3600) memory=ConversationMemory(), guardrails=[...], ) ## AgentResult fields result.output # dict {'result': str, 'finishReason': str} for plain agents # Pydantic model instance when Agent(output_type=...) is set result.output['result'] # the text output (only when no output_type is set) result.status # Status enum: COMPLETED | FAILED | TERMINATED | TIMED_OUT # String comparisons work: result.status == "COMPLETED" result.finish_reason # FinishReason enum: STOP | LENGTH | TOOL_CALLS | ERROR | CANCELLED | TIMEOUT | GUARDRAIL | REJECTED result.token_usage # TokenUsage or None — ONLY populated via AgentRuntime context manager # Always None when using module-level run() result.token_usage.prompt_tokens # int result.token_usage.completion_tokens # int result.token_usage.total_tokens # int result.sub_results # dict of per-agent AgentResult — PARALLEL strategy only result.execution_id # str or None [NOT workflow_id — that field does not exist] result.events # list[AgentEvent] result.is_success # bool property result.is_failed # bool property result.is_rejected # bool property result.print_result() # convenience print ## AgentHandle fields and methods handle.execution_id # str [NOT handle.workflow_id — that field does not exist] handle.get_status() # → AgentStatus handle.stream().get_result() # wait for result [NOT handle.wait() — that method does not exist] handle.approve() # resume after HITL approval checkpoint handle.reject(reason) # reject at HITL approval checkpoint handle.send(message) # send message to a waiting agent handle.pause() # pause workflow handle.resume() # resume paused workflow handle.cancel(reason) # cancel workflow # Reconnect to an existing run by execution_id # CRITICAL: must call runtime.serve(agent, blocking=False) BEFORE creating AgentHandle # when the agent has @tool functions — otherwise tools hang runtime = AgentRuntime(...) runtime.serve(agent, blocking=False) handle = AgentHandle(execution_id="", runtime=runtime) ## @tool decorator @tool def my_tool(input: str) -> str: """Description shown to LLM.""" return result # With options @tool( approval_required=True, # agent pauses; requires handle.approve() to resume description="...", # override docstring name="...", # override function name ) def my_tool(input: str) -> str: ... # ToolContext — inject by adding a parameter typed ToolContext from agentspan.agents import ToolContext @tool def my_tool(query: str, ctx: ToolContext) -> str: ctx.execution_id # str [NOT ctx.workflow_id] ctx.agent_name # str ctx.task_id # str return result ## Built-in tools (server-side, no worker code needed) # http_tool — make HTTP requests http_tool( url="https://api.example.com/endpoint", name="call_api", description="...", method="POST", # GET | POST | PUT | DELETE | PATCH headers={...}, body={...}, ) # api_tool — wraps an OpenAPI/REST endpoint api_tool( url="https://api.example.com", name="my_api", description="...", ) # mcp_tool — connect to an MCP server mcp_tool( server_url="...", name="my_mcp", description="...", ) # search_tool — web/knowledge search search_tool(name="search", description="...") # index_tool — semantic indexing index_tool(name="index", description="...") # image_tool — image generation image_tool(name="gen_image", description="...", llm_provider="...", model="...") # audio_tool — audio generation audio_tool(name="gen_audio", description="...", llm_provider="...", model="...") # video_tool — video generation video_tool(name="gen_video", description="...", llm_provider="...", model="...") # pdf_tool — PDF generation from markdown pdf_tool(name="gen_pdf", description="...") # human_tool — human-in-the-loop input collection human_tool(name="ask_human", description="...") # agent_tool — wrap another agent as a tool agent_tool(agent=other_agent, name="...", description="...") ## Multi-agent strategies (all 8 verified working) # Default strategy is HANDOFF (not SEQUENTIAL) # 1. Sequential — pipe operator or explicit pipeline = researcher >> writer >> publisher # equivalent: Agent(name="pipeline", agents=[researcher, writer, publisher], strategy=Strategy.SEQUENTIAL) # 2. Parallel — all sub-agents run concurrently; result.sub_results available Agent(name="parallel", agents=[agent_a, agent_b], strategy=Strategy.PARALLEL) # 3. Handoff — default; LLM decides which sub-agent to call Agent(name="x", agents=[agent_a, agent_b], strategy=Strategy.HANDOFF) # NOT: handoff([a, b]) — that function does not exist # 4. Router — classifier agent routes to one sub-agent Agent(name="x", agents=[agent_a, agent_b], strategy=Strategy.ROUTER, router=classifier_agent) # 5. Swarm — agents hand off based on text triggers from agentspan.agents import OnTextMention Agent(name="x", agents=[...], strategy=Strategy.SWARM, handoffs=[OnTextMention(...)]) # 6. Round Robin — cycles through agents in order Agent(name="x", agents=[...], strategy=Strategy.ROUND_ROBIN) # 7. Random — randomly selects a sub-agent Agent(name="x", agents=[...], strategy=Strategy.RANDOM) # 8. Manual — caller controls agent selection Agent(name="x", agents=[...], strategy=Strategy.MANUAL) ## Memory # ConversationMemory memory = ConversationMemory(max_messages=50) # only valid param: max_messages (NOT max_turns) memory.messages # list of dicts with 'role' and 'message' keys # NOT: memory.entries (doesn't exist) # NOT: ConversationMemory(max_turns=..., include_tool_calls=..., summarize_after=...) # SemanticMemory memory = SemanticMemory(store=None, max_results=5, session_id=None) # NOT: SemanticMemory(namespace=..., embedding_model=...) memory.add(content) # store content [NOT memory.store()] memory.search(query) # → List[str] ## Guardrails from agentspan.agents import Guardrail, RegexGuardrail Guardrail(func=my_fn, ...) # func= keyword [NOT fn=] RegexGuardrail(...) # no flags= param ## HITL (Human-in-the-Loop) # Agent pauses when LLM calls a tool decorated with approval_required=True @tool(approval_required=True) def sensitive_action(params: str) -> str: ... handle = start(agent, "prompt") status = handle.get_status() if status.is_waiting: handle.approve() # resume # or: handle.reject("reason") # reject ## Testing from agentspan.agents.testing import mock_run, MockEvent, expect result = mock_run(agent, "prompt", events=[ MockEvent.tool_call("tool_name", {"arg": "value"}), MockEvent.tool_result("tool_name", "result string"), MockEvent.done("final output"), ]) # All MockEvent constructors MockEvent.tool_call(name, args) # simulate LLM calling a tool MockEvent.tool_result(name, result) # simulate tool returning a result MockEvent.done(output) # simulate agent finishing MockEvent.waiting(msg) # simulate pausing for HITL [NOT MockEvent.human_approval()] MockEvent.handoff(from_agent, to_agent) # simulate agent handoff MockEvent.thinking(content) # simulate reasoning step MockEvent.message(content) # simulate agent message # Expectations expect(result).completed() expect(result).used_tool("tool_name") # [NOT .tool_used()] # Accessing tool calls on result result.tool_calls # list[dict] result.tool_calls[0]['name'] # access as dict [NOT result.tool_calls[0].name] ## Integration patterns (all verified working) # LangGraph — pass compiled StateGraph directly to run() from langgraph.graph import StateGraph graph = StateGraph(...) # ... build graph ... app = graph.compile() result = run(app, "prompt") # OpenAI Agents SDK — pass Agent directly to run() from agents import Agent as OAIAgent oai_agent = OAIAgent(name="...", model="...", instructions="...") result = run(oai_agent, "prompt") # Google ADK — pass pipeline directly to run() from google.adk.agents import SequentialAgent pipeline = SequentialAgent(...) result = run(pipeline, "prompt") # No agentspan.integrations module exists — pass foreign agents directly to run() ## Termination conditions TextMentionTermination(text="DONE") MaxMessageTermination(max_messages=10) StopMessageTermination() TokenUsageTermination(max_tokens=4000) ## Things that do NOT exist (common AI coding mistakes) parallel(a, b) # not a function — use Strategy.PARALLEL handoff([a, b]) # not a function — use Strategy.HANDOFF agentspan.integrations.* # no such module handle.wait() # use handle.stream().get_result() result.workflow_id # use result.execution_id handle.workflow_id # use handle.execution_id context.workflow_id # use context.execution_id in ToolContext result.turns # doesn't exist result.duration_ms # doesn't exist expect(result).tool_used("name") # it's .used_tool("name") MockEvent.human_approval() # use MockEvent.waiting() EventType.WAITING_FOR_HUMAN # use EventType.WAITING PromptTemplate (local dev) # requires server-side pre-creation; unusable with local dev server ConversationMemory(max_turns=...) # only max_messages is valid ConversationMemory(include_tool_calls=...) # doesn't exist ConversationMemory(summarize_after=...) # doesn't exist memory.entries # use memory.messages SemanticMemory(namespace=...) # param doesn't exist SemanticMemory(embedding_model=...) # param doesn't exist memory.store() # use memory.add() Guardrail(fn=my_fn) # use func= not fn= RegexGuardrail(flags=...) # no flags param configure(providers={...}) # providers param doesn't exist async with stream_async(...) # not supported; use s = await stream_async(...); async for event in s runtime.register_workers() # use runtime.serve(agent, blocking=False) handle.select_agent() # doesn't exist ## CLI commands [UNVERIFIED — verify exact commands before relying on them] # Install CLI npm install -g @agentspan-ai/agentspan # Placeholder — CLI reference: https://agentspan.ai/docs/cli ## Documentation https://agentspan.ai/docs/quickstart https://agentspan.ai/docs/concepts/agents https://agentspan.ai/docs/concepts/tools https://agentspan.ai/docs/concepts/multi-agent https://agentspan.ai/docs/concepts/guardrails https://agentspan.ai/docs/concepts/memory https://agentspan.ai/docs/concepts/streaming https://agentspan.ai/docs/concepts/testing https://agentspan.ai/docs/deployment https://agentspan.ai/docs/self-hosting https://agentspan.ai/docs/cli https://agentspan.ai/docs/providers ## AI Coding Agent Skills Install agentspan skills to get full SDK knowledge, CLI commands, examples, and multi-agent patterns directly in your coding agent. # Claude Code (plugin marketplace): # /plugin marketplace add agentspan-ai/agentspan-skills # /plugin install agentspan@agentspan-skills # # All other agents (Claude Code, Codex, Gemini CLI, Cursor, Windsurf, Cline, Copilot, Aider): # curl -sSL https://raw.githubusercontent.com/agentspan-ai/agentspan-skills/main/install.sh | bash -s -- --all Skills repository: https://github.com/agentspan-ai/agentspan-skills