edit on github↗

Example: OpenAI Agents SDK — customer support

Use case: A multi-agent customer support system using the OpenAI Agents SDK’s native handoff feature — a triage agent routes tickets to billing, technical, or account specialists. Wrapped with Agentspan in one line.

What Agentspan adds to the OpenAI Agents SDK

The OpenAI Agents SDK handles your agent definitions, handoffs, and tool routing. Agentspan adds a production execution layer without changing any of that:

  • Crash recovery — if your process dies during a multi-step resolution, Agentspan resumes when a worker reconnects
  • Full handoff trace — every handoff between agents is a logged step, visible in the UI at http://localhost:6767
  • Human approval on tools — add approval_required=True to any tool to pause execution for human sign-off
  • Execution history — every ticket run is stored with inputs, outputs, and timing

Your agent definitions, handoff configurations, and tool implementations stay exactly as written.

Before: plain OpenAI Agents SDK

Standard code using the OpenAI Agents SDK. The native handoff pattern works well, but runs have no history, no crash recovery, and no human-in-the-loop.

from agents import Agent, Runner, function_tool, handoff

# ── Tools ────────────────────────────────────────────────────────────────────

@function_tool
def get_account(customer_id: str) -> dict:
    """Look up a customer's account: plan, billing status, usage."""
    return accounts_db.find({"id": customer_id})

@function_tool
def get_invoice(invoice_id: str) -> dict:
    """Fetch an invoice by ID."""
    return billing_api.get_invoice(invoice_id)

@function_tool
def process_refund(invoice_id: str, reason: str) -> dict:
    """Issue a full refund for an invoice."""
    return billing_api.refund(invoice_id, reason=reason)

@function_tool
def get_ticket_history(customer_id: str) -> list[dict]:
    """Get the last 5 support tickets for a customer."""
    return tickets_db.find({"customer_id": customer_id}).limit(5)

@function_tool
def reset_password(customer_id: str) -> dict:
    """Send a password reset email to the customer."""
    return auth_api.send_reset(customer_id)

@function_tool
def check_service_status(region: str) -> dict:
    """Check current service health for a region."""
    return status_api.get(region)

@function_tool
def escalate_to_human(ticket_id: str, reason: str, priority: str) -> dict:
    """Escalate this ticket to a human agent."""
    return queue_api.escalate(ticket_id, reason=reason, priority=priority)

# ── Specialist agents ─────────────────────────────────────────────────────────

billing_agent = Agent(
    name="billing_specialist",
    model="gpt-4o",
    instructions="""You handle billing questions: invoices, charges, refunds, plan changes.
    Always look up the account first. Process refunds only for clear billing errors.
    For amounts over $200, escalate to a human.""",
    tools=[get_account, get_invoice, process_refund, escalate_to_human],
)

technical_agent = Agent(
    name="technical_specialist",
    model="gpt-4o",
    instructions="""You handle technical issues: login problems, service outages, API errors.
    Always check service status first. If the issue is account-specific, look up the account.
    Reset passwords only after verifying the customer's identity.""",
    tools=[get_account, check_service_status, reset_password, escalate_to_human],
)

account_agent = Agent(
    name="account_specialist",
    model="gpt-4o",
    instructions="""You handle account changes: upgrades, downgrades, cancellations, data exports.
    Always look up the account and ticket history before making changes.
    For cancellations, attempt retention first.""",
    tools=[get_account, get_ticket_history, escalate_to_human],
)

# ── Triage agent with handoffs ────────────────────────────────────────────────

triage_agent = Agent(
    name="support_triage",
    model="gpt-4o-mini",  # fast, cheap — just routes
    instructions="""You are a support triage agent. Understand the customer's issue
    and hand off to the right specialist immediately.

    - Billing, charges, invoices, refunds → billing_specialist
    - Login, outages, API errors, technical issues → technical_specialist
    - Plan changes, cancellations, account settings → account_specialist""",
    handoffs=[
        handoff(billing_agent),
        handoff(technical_agent),
        handoff(account_agent),
    ],
)

# ── Run (plain OpenAI Agents SDK — no durability) ─────────────────────────────

result = Runner.run_sync(
    triage_agent,
    "Hi, I was charged $99 twice last month (invoice INV-8821). Can I get a refund?",
)
print(result.final_output)

After: wrapped with Agentspan

Pass the OpenAI Agents SDK agent directly to runtime.run(). Agentspan auto-detects it — no integration module needed.

from agentspan.agents import AgentRuntime

# was: result = Runner.run_sync(triage_agent, message)
with AgentRuntime() as runtime:
    result = runtime.run(triage_agent, message)

print(result.output)
print(f"Run ID: {result.execution_id}")

What you gain

Full handoff trace — every handoff between agents is a logged step. Open http://localhost:6767 to see exactly which specialist handled the ticket and what they did.

Crash recovery — if your process dies during a complex multi-step billing resolution, Agentspan resumes when a worker reconnects. The customer’s ticket isn’t dropped.

Run history — every execution is stored with inputs, outputs, token usage, and timing.

Async variant

from agentspan.agents import run_async

async def handle_ticket(message: str):
    result = await run_async(triage_agent, message)
    return result.output

asyncio.run(handle_ticket("I was charged twice last month"))

Fire-and-forget for slow tickets

from agentspan.agents import start

# Returns immediately
handle = start(triage_agent, customer_message)
print(f"Ticket queued: {handle.execution_id}")

# Get result later
result = handle.stream().get_result()
print(result.output)

Stream events as they happen

from agentspan.agents import stream

for event in stream(triage_agent, customer_message):
    if event.type == "handoff":
        print(f"  → routed to {event.target}")
    elif event.type == "tool_call":
        print(f"  → {event.tool_name}({event.args})")
    elif event.type == "done":
        print(f"\n{event.output}")

Adding human approval for large refunds

Use Agentspan’s @tool decorator with approval_required=True on any sensitive tool:

from agentspan.agents import tool as agentspan_tool

@agentspan_tool(approval_required=True)
def process_refund(invoice_id: str, reason: str) -> dict:
    """Issue a full refund. Requires human approval."""
    return billing_api.refund(invoice_id, reason=reason)

# Use it in your OpenAI Agents SDK agent as normal
billing_agent = Agent(
    name="billing_specialist",
    tools=[get_account, get_invoice, process_refund, escalate_to_human],
    ...
)

Testing

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

result = mock_run(
    triage_agent,
    "I was charged twice last month",
    events=[
        MockEvent.handoff("billing_specialist"),
        MockEvent.done("I've looked into your account. A refund has been initiated."),
    ]
)
expect(result).completed()
Previous ← LangGraph — code review bot Next Google ADK — research assistant →