p4_handoff.py

"""Pattern 4: handoffs. Control AND the conversation move between specialists; a router in code decides who's next."""

import asyncio
import re

from claude_agent_sdk import ClaudeAgentOptions, ResultMessage, query
from demo_util import PROJECT

SPECIALISTS = {
    "triage": dict(tools=[], prompt=(
        "You are front-desk triage for a software team. Acknowledge the user in one sentence, then hand off to the right "
        "specialist. If the message has BOTH a billing and a technical part, hand off to billing first. "
        "End your reply with exactly one line: HANDOFF:billing, HANDOFF:tech, or HANDOFF:done.")),
    "billing": dict(tools=[], prompt=(
        "You are the billing specialist. Resolve the billing part of the conversation so far in two sentences "
        "(you may promise a refund within 5 business days). If a technical issue remains unresolved in the conversation, "
        "end with HANDOFF:tech. Otherwise end with HANDOFF:done.")),
    "tech": dict(allowed_tools=["Read", "Grep", "Glob"], prompt=(
        "You are tech support with read access to the code in this directory. Diagnose the technical issue from the "
        "conversation so far by reading src/, name the file:line and a one-sentence fix. End with HANDOFF:done.")),
}


def next_agent(text: str) -> str | None:
    m = re.search(r"HANDOFF:(\w+)", text or "")
    return m.group(1) if m else None


async def handoff_loop(user_message: str, max_hops: int = 4):
    current, session_id, total = "triage", None, 0.0
    for hop in range(max_hops):                                                   # (1) bounded
        spec = SPECIALISTS[current]
        opts = ClaudeAgentOptions(
            cwd=PROJECT, max_turns=6, setting_sources=[],
            tools=spec.get("tools"), allowed_tools=spec.get("allowed_tools", []),
            system_prompt=spec["prompt"],                                          # (2) the specialist's instructions
            resume=session_id, fork_session=session_id is not None,                # (3) the conversation travels with the handoff
        )
        prompt = user_message if hop == 0 else (
            f"[System: this conversation was handed off to the {current} specialist. "
            f"Your instructions: {spec['prompt']} Continue from the conversation so far.]")   # (4) re-inject: resumed sessions keep their recorded prompt
        text = None
        async for m in query(prompt=prompt, options=opts):
            if isinstance(m, ResultMessage):
                session_id, text, total = m.session_id, m.result, total + (m.total_cost_usd or 0)
        print(f"\n[{current}] (session {session_id[:8]}...)\n{text}")
        target = next_agent(text)
        if target in (None, "done"):
            print(f"\n-- conversation closed by {current} after {hop + 1} hop(s), ${total:.4f}")
            return
        current = target                                                           # (5) the handoff
    print("\n-- escalated to a human: too many handoffs")


asyncio.run(handoff_loop(
    "Hi, I was charged twice for my subscription last month. Also, looking up my account by name on the user page "
    "crashes the app, I think the lookup code is broken."))