# 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

# Python
pip install agentspan   # Python
# Typescript
npm install @agentspan-ai/sdk            # TypeScript

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 agentspan.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()

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
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()

from pydantic import BaseModel
from agentspan.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

agentspan credentials set GITHUB_TOKEN ghp_xxxxxxxxxxxx
agentspan credentials set SEARCH_API_KEY xxx-your-key

from agentspan.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"]}

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()

from agentspan.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"])

from agentspan.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()

from agentspan.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()

from agentspan.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()

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

from agentspan.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()

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

from agentspan.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()

from agentspan.agents import Agent, AgentRuntime
from agentspan.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()

from agentspan.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()

import time
from agentspan.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="openai/gpt-4o-mini",
    instructions="You are a helpful assistant.",
    callbacks=[TimingHandler(), LoggingHandler()],
)

with AgentRuntime() as runtime:
    result = runtime.run(agent, "Hello!")
    result.print_result()

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])

├── 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

agentspan server start     # Start the Agentspan server
agentspan server stop      # Stop the server
agentspan server logs      # View server logs
agentspan doctor           # Check system dependencies

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