Customer Message
       ↓
[Classifier] → intent, severity, category
       ↓
[Knowledge Base Tool] → relevant articles
       ↓
[Response Generator] → draft reply
       ↓
[Escalation Check] → route to human if needed
       ↓
Customer Response or Human Handoff

bashexport ANTHROPIC_API_KEY="sk-ant-..."

python# knowledge_base.py

KNOWLEDGE_BASE = {
    "refund_policy": """
        Refunds are available within 30 days of purchase for any reason.
        After 30 days, refunds require manager approval.
        To request a refund, customers must provide their order number.
        Refunds are processed within 5-7 business days.
    """,
    "account_locked": """
        Accounts are temporarily locked after 5 failed login attempts.
        Lockouts last 30 minutes automatically.
        For immediate unlock, customers must verify their email.
        Persistent lockouts may indicate a security issue and should be escalated.
    """,
    "subscription_plans": """
        Starter: $19/month — 5 projects, 10GB storage
        Pro: $49/month — unlimited projects, 100GB storage, priority support
        Enterprise: Custom pricing — SSO, dedicated account manager, SLA guarantee
        Plan changes take effect on the next billing cycle.
    """,
    "api_rate_limits": """
        Free tier: 100 requests/hour
        Pro tier: 10,000 requests/hour
        Enterprise: Custom limits
        Rate limit errors return HTTP 429. Use exponential backoff.
    """
}

def lookup_knowledge_base(query: str) -> str:
    """Simple keyword-based lookup. Replace with vector search in production."""
    query_lower = query.lower()
    
    results = []
    if any(word in query_lower for word in ["refund", "money back", "cancel"]):
        results.append(KNOWLEDGE_BASE["refund_policy"])
    if any(word in query_lower for word in ["locked", "login", "password", "access"]):
        results.append(KNOWLEDGE_BASE["account_locked"])
    if any(word in query_lower for word in ["plan", "pricing", "subscription", "upgrade"]):
        results.append(KNOWLEDGE_BASE["subscription_plans"])
    if any(word in query_lower for word in ["api", "rate limit", "429", "requests"]):
        results.append(KNOWLEDGE_BASE["api_rate_limits"])
    
    return "\n\n".join(results) if results else "No matching articles found."

python# tools.py

SUPPORT_TOOLS = [
    {
        "name": "lookup_knowledge_base",
        "description": "Search the product knowledge base for relevant help articles and policies. Use this before generating any response to ensure accuracy.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The customer's question or topic to search for"
                }
            },
            "required": ["query"]
        }
    },
    {
        "name": "create_support_ticket",
        "description": "Create a support ticket in the ticketing system. Use when the issue requires follow-up or cannot be resolved immediately.",
        "input_schema": {
            "type": "object",
            "properties": {
                "category": {
                    "type": "string",
                    "enum": ["billing", "technical", "account", "general"],
                    "description": "The category of the support issue"
                },
                "severity": {
                    "type": "string",
                    "enum": ["low", "medium", "high", "critical"],
                    "description": "Severity based on business impact"
                },
                "summary": {
                    "type": "string",
                    "description": "One-sentence summary of the issue"
                }
            },
            "required": ["category", "severity", "summary"]
        }
    },
    {
        "name": "escalate_to_human",
        "description": "Escalate the conversation to a human support agent. Use for: billing disputes over $200, security incidents, legal threats, or situations where you cannot resolve the issue.",
        "input_schema": {
            "type": "object",
            "properties": {
                "reason": {
                    "type": "string",
                    "description": "Why this needs human attention"
                },
                "priority": {
                    "type": "string",
                    "enum": ["normal", "urgent"],
                    "description": "normal for standard escalations, urgent for security or legal issues"
                }
            },
            "required": ["reason", "priority"]
        }
    }
]

python# agent.py

import anthropic
import json
from knowledge_base import lookup_knowledge_base
from tools import SUPPORT_TOOLS

client = anthropic.Anthropic()

SYSTEM_PROMPT = """You are a helpful customer support agent for Acme SaaS.

Your responsibilities:
1. ALWAYS search the knowledge base first before answering any question
2. Be empathetic and professional in all responses
3. Never promise what you cannot deliver
4. Escalate to a human agent when: billing disputes > $200, security incidents, legal threats, or issues you cannot resolve
5. Create a support ticket for any issue that requires follow-up

When you cannot find an answer in the knowledge base, say so honestly rather than guessing.

Tone: Friendly, concise, solution-focused. Avoid corporate jargon."""


def handle_tool_call(tool_name: str, tool_input: dict) -> str:
    """Execute the tool and return the result as a string."""
    
    if tool_name == "lookup_knowledge_base":
        result = lookup_knowledge_base(tool_input["query"])
        return result
    
    elif tool_name == "create_support_ticket":
        # In production, this would call your ticketing API (Zendesk, Linear, etc.)
        ticket_id = f"TKT-{hash(tool_input['summary']) % 10000:04d}"
        return f"Ticket created: {ticket_id} | Category: {tool_input['category']} | Severity: {tool_input['severity']}"
    
    elif tool_name == "escalate_to_human":
        # In production, this would notify your support team via Slack/PagerDuty
        return f"Escalated to human agent | Priority: {tool_input['priority']} | Reason: {tool_input['reason']}"
    
    return "Tool execution failed"


def run_support_agent(customer_message: str, conversation_history: list = None) -> dict:
    """
    Run the support agent for a single customer message.
    Returns: {"response": str, "actions_taken": list, "escalated": bool}
    """
    
    if conversation_history is None:
        conversation_history = []
    
    # Add customer message to history
    messages = conversation_history + [
        {"role": "user", "content": customer_message}
    ]
    
    actions_taken = []
    escalated = False
    
    # Tool-use loop
    while True:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            system=SYSTEM_PROMPT,
            tools=SUPPORT_TOOLS,
            messages=messages
        )
        
        # If Claude wants to use tools
        if response.stop_reason == "tool_use":
            # Add Claude's response to messages
            messages.append({"role": "assistant", "content": response.content})
            
            # Process each tool call
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    tool_name = block.name
                    tool_input = block.input
                    
                    print(f"  → Calling tool: {tool_name}")
                    result = handle_tool_call(tool_name, tool_input)
                    
                    actions_taken.append({
                        "tool": tool_name,
                        "input": tool_input,
                        "result": result
                    })
                    
                    if tool_name == "escalate_to_human":
                        escalated = True
                    
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result
                    })
            
            # Feed tool results back to Claude
            messages.append({"role": "user", "content": tool_results})
        
        # Claude has a final response
        elif response.stop_reason == "end_turn":
            final_text = ""
            for block in response.content:
                if hasattr(block, "text"):
                    final_text += block.text
            
            return {
                "response": final_text,
                "actions_taken": actions_taken,
                "escalated": escalated
            }
        
        else:
            return {
                "response": "I encountered an unexpected error. Please try again.",
                "actions_taken": actions_taken,
                "escalated": False
            }

python# main.py

from agent import run_support_agent

test_cases = [
    "I've been charged twice for my subscription this month and I want a refund immediately.",
    "My account is locked and I have a demo in 10 minutes, please help!",
    "What's the difference between the Pro and Enterprise plans?",
    "I'm hitting rate limits on your API even though I'm on the Pro plan."
]

for message in test_cases:
    print(f"\n{'='*60}")
    print(f"Customer: {message}")
    print(f"{'='*60}")
    
    result = run_support_agent(message)
    
    print(f"\nAgent Response:\n{result['response']}")
    
    if result['actions_taken']:
        print(f"\nActions taken:")
        for action in result['actions_taken']:
            print(f"  - {action['tool']}: {action['result']}")
    
    if result['escalated']:
        print(f"\n⚠️  ESCALATED TO HUMAN AGENT")

==============================
Customer: I've been charged twice for my subscription this month and I want a refund immediately.
==============================

  → Calling tool: lookup_knowledge_base
  → Calling tool: create_support_ticket
  → Calling tool: escalate_to_human

Agent Response:
I completely understand your frustration — being charged twice is not acceptable, and I want to get this resolved for you right away.

I've looked up our refund policy and created a support ticket (TKT-2847) for your case. Because this involves a billing dispute, I've also escalated this to our billing team who will contact you within 2 hours.

In the meantime, could you share your order or transaction ID? This will help our billing team process your refund faster. Per our policy, refunds are processed within 5-7 business days once approved.

Actions taken:
  - lookup_knowledge_base: [refund policy content]
  - create_support_ticket: Ticket created: TKT-2847 | Category: billing | Severity: high
  - escalate_to_human: Escalated to human agent | Priority: urgent

pythondef support_session():
    """Run an interactive support session with conversation memory."""
    history = []
    
    print("Support Agent: Hello! How can I help you today?")
    
    while True:
        user_input = input("\nYou: ").strip()
        if user_input.lower() in ["exit", "quit", "bye"]:
            print("Support Agent: Thank you for contacting us. Have a great day!")
            break
        
        result = run_support_agent(user_input, history)
        print(f"\nSupport Agent: {result['response']}")
        
        # Update history for next turn
        history.append({"role": "user", "content": user_input})
        history.append({"role": "assistant", "content": result['response']})
        
        if result['escalated']:
            print("\n[Session transferred to human agent]")
            break

if __name__ == "__main__":
    support_session()