Customer Support Resolution Agent.

from anthropic import Anthropic

client = Anthropic()

def build_system_prompt(case_facts: dict) -> str:
    return f"""You are a customer support agent for ACME.

CASE_FACTS (immutable; re-read every turn):
- customer_id: {case_facts['customer_id']}
- order_id: {case_facts['order_id']}
- refund_amount: ${case_facts['amount']}
- policy_cap: ${case_facts['cap']}

Constraints:
- Verify customer before ANY account-modifying call.
- Refunds above policy_cap MUST escalate (a hook enforces this).
- Branch on stop_reason. Never on response text."""

tools = [
    {
        "name": "verify_customer",
        "description": (
            "Look up a customer by customer_id and confirm they are active.\n"
            "Use this BEFORE any other tool that mentions the customer.\n"
            "Edge cases: returns 'not_found' if customer_id is missing.\n"
            "Always run before lookup_order or process_refund."
        ),
        "input_schema": {"type": "object", "properties": {
            "customer_id": {"type": "string"}
        }, "required": ["customer_id"]},
    },
    # ... lookup_order, process_refund, escalate_to_human
]

# .claude/hooks/refund_policy.py
import sys, json, os

POLICY_CAP = float(os.environ.get("REFUND_CAP", "500"))

def main():
    payload = json.loads(sys.stdin.read())
    if payload["tool_name"] != "process_refund":
        sys.exit(0)  # not our concern, allow
    amount = payload["tool_input"].get("amount", 0)
    if amount > POLICY_CAP:
        print(f"refund ${amount} exceeds cap ${POLICY_CAP}, escalate", file=sys.stderr)
        sys.exit(2)  # DENY
    sys.exit(0)  # allow

if __name__ == "__main__":
    main()

def run_agent_loop(user_msg: str, case_facts: dict, max_iter: int = 15):
    messages = [{"role": "user", "content": user_msg}]
    for _ in range(max_iter):
        resp = client.messages.create(
            model="claude-sonnet-4.5",
            max_tokens=4096,
            system=build_system_prompt(case_facts),
            tools=tools,
            messages=messages,
        )
        if resp.stop_reason == "end_turn":
            return extract_text(resp)
        if resp.stop_reason == "tool_use":
            tool_uses = [b for b in resp.content if b.type == "tool_use"]
            results = [execute_tool(t) for t in tool_uses]
            messages.append({"role": "assistant", "content": resp.content})
            messages.append({"role": "user", "content": results})
            continue
        if resp.stop_reason == "max_tokens":
            return {"status": "partial", "text": extract_text(resp)}
    return {"status": "iteration_cap"}

def escalate(case_facts: dict, reason: str, partial: dict) -> dict:
    return {
        "customer_id": case_facts["customer_id"],
        "order_id": case_facts["order_id"],
        "intent": partial.get("intent", "unknown"),
        "partial_status": partial.get("last_action"),
        "blocker": reason,
        "recommended_action": derive_recommendation(reason),
        "evidence": [partial.get("last_tool_result")],
    }

def post_response_gates(response: str, agent_confidence: float):
    sentiment = sentiment_score(response)
    if sentiment == "distressed" and agent_confidence < 0.7:
        return {"action": "escalate", "reason": "low_confidence_distressed"}
    if agent_confidence < 0.5:
        return {"action": "escalate", "reason": "low_confidence"}
    return {"action": "send"}

resp = client.messages.create(
    model="claude-sonnet-4.5",
    max_tokens=4096,
    system=[
        {
            "type": "text",
            "text": build_system_prompt(case_facts),
            "cache_control": {"type": "ephemeral"},
        },
    ],
    tools=tools,  # tools also auto-cached when stable
    messages=messages,
)

def audit_log(case_facts: dict, agent_path: list, elapsed_ms: int, csat: int | None):
    db.audit.insert({
        "ts": datetime.utcnow(),
        "customer_id": case_facts["customer_id"],
        "order_id": case_facts["order_id"],
        "tool_calls": [c["name"] for c in agent_path if c["type"] == "tool_use"],
        "stop_reasons": [c["stop_reason"] for c in agent_path],
        "elapsed_ms": elapsed_ms,
        "csat": csat,
    })