Example: Support ticket triage

Use case: Customer support automation. Incoming tickets are classified, simple issues are resolved automatically, and anything involving money or account changes requires a human to approve before the agent acts.

Full code

from agentspan.agents import Agent, tool, start
from pydantic import BaseModel
from enum import Enum

# ── Data types ────────────────────────────────────────────────────────────────

class TicketCategory(str, Enum):
    BILLING    = "billing"
    TECHNICAL  = "technical"
    ACCOUNT    = "account"
    GENERAL    = "general"

class Resolution(BaseModel):
    category: TicketCategory
    action_taken: str
    response_to_customer: str
    requires_followup: bool

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

@tool
def lookup_customer(email: str) -> dict:
    """Fetch customer record: plan, billing status, open tickets, account age."""
    return db.customers.find_one({"email": email})

@tool
def lookup_ticket_history(customer_id: str) -> list[dict]:
    """Fetch the last 10 support tickets for this customer."""
    return db.tickets.find({"customer_id": customer_id}).sort("created_at", -1).limit(10)

@tool
def send_reply(customer_id: str, message: str) -> dict:
    """Send a reply to the customer and mark the ticket as resolved."""
    return support_api.send(customer_id=customer_id, message=message, status="resolved")

@tool(approval_required=True)
def issue_refund(customer_id: str, amount_usd: float, reason: str) -> dict:
    """Issue a refund to the customer. Requires human approval."""
    return billing_api.refund(customer_id=customer_id, amount=amount_usd, reason=reason)

@tool(approval_required=True)
def suspend_account(customer_id: str, reason: str) -> dict:
    """Suspend a customer account. Requires human approval."""
    return accounts_api.suspend(customer_id=customer_id, reason=reason)

@tool(approval_required=True)
def apply_credit(customer_id: str, amount_usd: float, note: str) -> dict:
    """Apply account credit. Requires human approval for amounts over $50."""
    return billing_api.credit(customer_id=customer_id, amount=amount_usd, note=note)

# ── Agent ─────────────────────────────────────────────────────────────────────

support_agent = Agent(
    name="support_agent",
    model="anthropic/claude-sonnet-4-6",
    output_type=Resolution,
    tools=[
        lookup_customer,
        lookup_ticket_history,
        send_reply,
        issue_refund,
        suspend_account,
        apply_credit,
    ],
    instructions="""You are a support agent for a SaaS product.

When a ticket arrives:
1. Look up the customer's account and ticket history
2. Diagnose the issue based on context
3. For general / technical questions: resolve directly with send_reply
4. For billing actions (refunds, credits): use the appropriate tool — these will pause
   for human review before executing
5. Return a Resolution with what happened

Always be clear and empathetic in your response_to_customer.
Never invent facts about the customer's account.""",
)

# ── Handle a ticket ───────────────────────────────────────────────────────────

def handle_ticket(ticket_id: str, customer_email: str, message: str):
    prompt = f"""
Ticket ID: {ticket_id}
Customer email: {customer_email}
Message: {message}
"""
    handle = start(support_agent, prompt)

    for event in handle.stream():
        if event.type == "waiting":
            # Route to your approval queue — Slack, internal tool, email, etc.
            notify_approval_queue(
                ticket_id=ticket_id,
                workflow_id=handle.execution_id,
                tool=event.tool_name,
                args=event.args,
                preview=f"Agent wants to {event.tool_name} for {customer_email}",
            )
            # handle.approve() or handle.reject() will be called by the reviewer
            break

    return handle.execution_id

# ── Reviewer approves or rejects ──────────────────────────────────────────────

def reviewer_approve(workflow_id: str, runtime: AgentRuntime):
    from agentspan.agents import AgentHandle
    # runtime must have serve(agent, blocking=False) called before this
    handle = AgentHandle(execution_id=workflow_id, runtime=runtime)
    handle.approve()

def reviewer_reject(workflow_id: str, runtime: AgentRuntime, reason: str):
    from agentspan.agents import AgentHandle
    handle = AgentHandle(execution_id=workflow_id, runtime=runtime)
    handle.reject(reason)
    # Agent receives the rejection and can try an alternative action

What this demonstrates

Human-in-the-loop with approval_required=True — tools decorated this way pause the agent when called. The agent’s state is held on the server indefinitely — no timeout, no polling loop needed on your side. The reviewer can approve or reject hours later.

Reconnecting to a paused agent — AgentHandle(execution_id=...) lets any process (a webhook, a background job, a CLI command) resume a paused execution. The execution ID is stable and stored.

Different tools with different risk levels — lookup_customer and send_reply run immediately. issue_refund, suspend_account, and apply_credit all require human review. This is expressed in the tool definition, not in orchestration logic.

Variations

Approve from the CLI

agentspan agent respond exec-a1b2c3 --approve
agentspan agent respond exec-a1b2c3 --reject "Amount too large, escalate to finance"

Auto-approve small refunds

Override the approval gate based on args:

def reviewer_route(workflow_id: str, tool: str, args: dict, runtime: AgentRuntime):
    if tool == "issue_refund" and args["amount_usd"] <= 25:
        AgentHandle(execution_id=workflow_id, runtime=runtime).approve()
    else:
        notify_approval_queue(workflow_id, tool, args)

Query ticket history

Open http://localhost:6767 to browse execution history. You can filter by agent name, status, and date range in the UI.

Testing

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

# Test the happy path: general question, resolved without approval
result = mock_run(
    support_agent,
    "Ticket: billing@acme.com — How do I export my data?",
    events=[
        MockEvent.tool_call("lookup_customer", {"email": "billing@acme.com"}),
        MockEvent.tool_result("lookup_customer", {"id": "cust_123", "plan": "pro"}),
        MockEvent.tool_call("send_reply", {"customer_id": "cust_123", "message": "..."}),
        MockEvent.tool_result("send_reply", {"status": "sent"}),
        MockEvent.done('{"category": "general", "action_taken": "Sent export instructions", ...}'),
    ]
)

expect(result).completed().used_tool("send_reply")

# Test the refund path: confirm approval tool is called and agent waits
result = mock_run(
    support_agent,
    "Ticket: billing@acme.com — I was charged twice this month, please refund",
    events=[
        MockEvent.tool_call("lookup_customer", {"email": "billing@acme.com"}),
        MockEvent.tool_result("lookup_customer", {"id": "cust_123", "plan": "pro"}),
        MockEvent.tool_call("issue_refund", {"customer_id": "cust_123", "amount_usd": 99.0, "reason": "duplicate charge"}),
        MockEvent.waiting("Waiting for human approval on issue_refund"),
        MockEvent.done('{"category": "billing", "action_taken": "Refund pending approval", ...}'),
    ]
)

expect(result).completed().used_tool("issue_refund")