How to Build a Multi-Agent Customer Support System with Claude API (Step-by-Step)

python -m venv support-agents-env
source support-agents-env/bin/activate   # On Windows: support-agents-env\Scripts\activate
pip install anthropic python-dotenv

ANTHROPIC_API_KEY=your_api_key_here

import os
from dotenv import load_dotenv
import anthropic

load_dotenv()

client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

response = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=64,
    messages=[{"role": "user", "content": "Say: connection successful"}]
)

print(response.content[0].text)

import os
import json
from dotenv import load_dotenv
import anthropic

load_dotenv()

client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

MODEL = "claude-sonnet-4-5"


class SupportAgent:
    """Base class for all support agents in the multi-agent system."""

    def __init__(self, name: str, system_prompt: str, tools: list):
        self.name = name
        self.system_prompt = system_prompt
        self.tools = tools

    def run(self, messages: list, tool_handler=None) -> dict:
        """
        Sends messages to Claude, handles tool use in a loop,
        and returns the final text response plus any tool results.
        """
        response = client.messages.create(
            model=MODEL,
            max_tokens=1024,
            system=self.system_prompt,
            tools=self.tools,
            messages=messages
        )

        tool_results = []

        # Keep looping while Claude wants to use tools
        while response.stop_reason == "tool_use":
            tool_uses = [b for b in response.content if b.type == "tool_use"]
            results_for_claude = []

            for tool_use in tool_uses:
                tool_name = tool_use.name
                tool_input = tool_use.input

                # Delegate actual tool execution to the caller
                if tool_handler:
                    result = tool_handler(tool_name, tool_input)
                else:
                    result = {"error": f"No handler for tool: {tool_name}"}

                tool_results.append({"tool": tool_name, "result": result})
                results_for_claude.append({
                    "type": "tool_result",
                    "tool_use_id": tool_use.id,
                    "content": json.dumps(result)
                })

            # Feed tool results back to Claude and continue
            messages = messages + [
                {"role": "assistant", "content": response.content},
                {"role": "user", "content": results_for_claude}
            ]

            response = client.messages.create(
                model=MODEL,
                max_tokens=1024,
                system=self.system_prompt,
                tools=self.tools,
                messages=messages
            )

        final_text = ""
        for block in response.content:
            if hasattr(block, "text"):
                final_text = block.text
                break

        return {
            "agent": self.name,
            "response": final_text,
            "tool_results": tool_results,
            "stop_reason": response.stop_reason
        }

import os
from dotenv import load_dotenv
from agents import SupportAgent

load_dotenv()

TRIAGE_SYSTEM_PROMPT = """You are a customer support triage agent. Your only job is to:
1. Analyze the customer's message
2. Look up their ticket if a ticket ID is provided
3. Classify the issue as one of: billing, technical, account, general
4. Determine urgency: low, medium, high, critical
5. Output a JSON object with keys: category, urgency, summary, recommended_agent

Always use the lookup_ticket tool when a ticket ID is mentioned.
Respond only with the JSON object. No extra commentary."""

TRIAGE_TOOLS = [
    {
        "name": "lookup_ticket",
        "description": "Looks up a support ticket by ID and returns its details.",
        "input_schema": {
            "type": "object",
            "properties": {
                "ticket_id": {
                    "type": "string",
                    "description": "The support ticket ID, e.g. TKT-1042"
                }
            },
            "required": ["ticket_id"]
        }
    }
]


def create_triage_agent() -> SupportAgent:
    return SupportAgent(
        name="TriageAgent",
        system_prompt=TRIAGE_SYSTEM_PROMPT,
        tools=TRIAGE_TOOLS
    )

import os
from dotenv import load_dotenv
from agents import SupportAgent

load_dotenv()

RESOLUTION_SYSTEM_PROMPT = """You are a customer support resolution agent. Your job is to:
1. Read the triage summary passed to you as context
2. Query the knowledge base for relevant solutions
3. Provide a clear, friendly resolution to the customer's issue
4. If you cannot resolve the issue confidently, call the escalate tool

Be concise. Use numbered steps for instructions. Always confirm you've solved
the issue at the end. If the knowledge base returns no useful results, escalate."""

RESOLUTION_TOOLS = [
    {
        "name": "query_knowledge_base",
        "description": "Searches the internal knowledge base for solutions to customer issues.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The search query describing the customer's problem"
                },
                "category": {
                    "type": "string",
                    "description": "Issue category: billing, technical, account, or general"
                }
            },
            "required": ["query", "category"]
        }
    },
    {
        "name": "escalate",
        "description": "Escalates the ticket to a human agent when the issue cannot be resolved automatically.",
        "input_schema": {
            "type": "object",
            "properties": {
                "reason": {
                    "type": "string",
                    "description": "Why this ticket needs human attention"
                },
                "priority": {
                    "type": "string",
                    "description": "Escalation priority: normal or urgent"
                }
            },
            "required": ["reason", "priority"]
        }
    }
]


def create_resolution_agent() -> SupportAgent:
    return SupportAgent(
        name="ResolutionAgent",
        system_prompt=RESOLUTION_SYSTEM_PROMPT,
        tools=RESOLUTION_TOOLS
    )

import json
from datetime import datetime

# Simulated ticket database
TICKET_DATABASE = {
    "TKT-1042": {
        "id": "TKT-1042",
        "customer": "Maria Gonzalez",
        "email": "[email protected]",
        "subject": "Cannot log in after password reset",
        "status": "open",
        "created": "2026-11-07T14:22:00",
        "account_type": "pro"
    },
    "TKT-1098": {
        "id": "TKT-1098",
        "customer": "James Whitfield",
        "email": "[email protected]",
        "subject": "Charged twice for November subscription",
        "status": "open",
        "created": "2026-11-08T09:10:00",
        "account_type": "enterprise"
    }
}

# Simulated knowledge base articles
KNOWLEDGE_BASE = {
    "technical": [
        {
            "title": "Password Reset Not Working",
            "solution": (
                "1. Clear your browser cache and cookies.\n"
                "2. Try the reset link in an incognito window.\n"
                "3. Make sure you're using the link within 24 hours of receiving it.\n"
                "4. If the issue persists, request a new reset email."
            )
        },
        {
            "title": "Two-Factor Authentication Lockout",
            "solution": (
                "1. Use one of your backup codes from account setup.\n"
                "2. If you don't have backup codes, contact support with photo ID."
            )
        }
    ],
    "billing": [
        {
            "title": "Duplicate Charge Resolution",
            "solution": (
                "1. We'll verify the duplicate charge on our end.\n"
                "2. Refunds are processed within 3-5 business days.\n"
                "3. You'll receive a confirmation email once the refund is initiated."
            )
        }
    ],
    "account": [
        {
            "title": "Account Access Issues",
            "solution": (
                "1. Verify your email address is correct.\n"
                "2. Check your spam folder for verification emails.\n"
                "3. Try logging in from a different device."
            )
        }
    ],
    "general": []
}


def lookup_ticket(ticket_id: str) -> dict:
    """Fetches ticket details from the simulated database."""
    ticket = TICKET_DATABASE.get(ticket_id)
    if ticket:
        return {"found": True, "ticket": ticket}
    return {"found": False, "message": f"No ticket found with ID {ticket_id}"}


def query_knowledge_base(query: str, category: str) -> dict:
    """Searches knowledge base articles by category."""
    articles = KNOWLEDGE_BASE.get(category, [])
    if not articles:
        return {"results": [], "message": "No articles found for this category."}

    # Simple keyword match (in production, use vector search)
    query_lower = query.lower()
    matched = [
        a for a in articles
        if any(word in a["title"].lower() for word in query_lower.split())
    ]

    if not matched:
        matched = articles  # Fall back to all articles in category

    return {"results": matched, "count": len(matched)}


def escalate(reason: str, priority: str) -> dict:
    """Logs an escalation and returns a ticket reference for the human queue."""
    escalation_id = f"ESC-{datetime.now().strftime('%H%M%S')}"
    return {
        "escalation_id": escalation_id,
        "status": "queued",
        "priority": priority,
        "reason": reason,
        "message": f"Ticket escalated. A human agent will follow up within {'1 hour' if priority == 'urgent' else '4 hours'}."
    }


def handle_tool_call(tool_name: str, tool_input: dict) -> dict:
    """
    Central dispatcher. Routes tool calls from any agent
    to the correct Python function.
    """
    if tool_name == "lookup_ticket":
        return lookup_ticket(tool_input["ticket_id"])
    elif tool_name == "query_knowledge_base":
        return query_knowledge_base(tool_input["query"], tool_input["category"])
    elif tool_name == "escalate":
        return escalate(tool_input["reason"], tool_input["priority"])
    else:
        return {"error": f"Unknown tool: {tool_name}"}

import json
import os
from dotenv import load_dotenv

from triage_agent import create_triage_agent
from resolution_agent import create_resolution_agent
from tools import handle_tool_call

load_dotenv()


class SupportOrchestrator:
    """
    Orchestrates the full multi-agent support pipeline:
    Triage -> Resolution (or Escalation).
    """

    def __init__(self):
        self.triage_agent = create_triage_agent()
        self.resolution_agent = create_resolution_agent()

    def run(self, customer_message: str, ticket_id: str = None) -> dict:
        """
        Main entry point. Takes a raw customer message and optional ticket ID,
        runs the full agent pipeline, and returns the final response.
        """
        print(f"\n{'='*60}")
        print(f"ORCHESTRATOR: New support request received")
        print(f"Message: {customer_message}")
        if ticket_id:
            print(f"Ticket ID: {ticket_id}")
        print(f"{'='*60}\n")

        # Build the initial message for triage
        triage_input = customer_message
        if ticket_id:
            triage_input = f"Ticket ID: {ticket_id}\n\nCustomer message: {customer_message}"

        # Stage 1: Triage
        print("[Stage 1] Running TriageAgent...")
        triage_messages = [{"role": "user", "content": triage_input}]

        triage_result = self.triage_agent.run(
            messages=triage_messages,
            tool_handler=handle_tool_call
        )

        print(f"[TriageAgent] Raw response: {triage_result['response']}")

        # Parse triage JSON output
        triage_data = self._parse_triage_output(triage_result["response"])
        print(f"[TriageAgent] Parsed classification: {json.dumps(triage_data, indent=2)}")

        # Stage 2: Resolution
        print("\n[Stage 2] Running ResolutionAgent...")

        # Build context-rich message for resolution agent
        resolution_context = (
            f"TRIAGE SUMMARY:\n"
            f"Category: {triage_data.get('category', 'general')}\n"
            f"Urgency: {triage_data.get('urgency', 'low')}\n"
            f"Summary: {triage_data.get('summary', customer_message)}\n\n"
            f"ORIGINAL CUSTOMER MESSAGE:\n{customer_message}"
        )

        # Include ticket data if triage found it
        if triage_result.get("tool_results"):
            for tr in triage_result["tool_results"]:
                if tr["tool"] == "lookup_ticket" and tr["result"].get("found"):
                    ticket_info = tr["result"]["ticket"]
                    resolution_context += (
                        f"\n\nTICKET DETAILS:\n"
                        f"Customer: {ticket_info['customer']}\n"
                        f"Account type: {ticket_info['account_type']}\n"
                        f"Subject: {ticket_info['subject']}"
                    )

        resolution_messages = [{"role": "user", "content": resolution_context}]

        resolution_result = self.resolution_agent.run(
            messages=resolution_messages,
            tool_handler=handle_tool_call
        )

        print(f"[ResolutionAgent] Response: {resolution_result['response']}")

        # Check if the resolution agent escalated
        escalation_info = None
        for tr in resolution_result.get("tool_results", []):
            if tr["tool"] == "escalate":
                escalation_info = tr["result"]

        return {
            "triage": triage_data,
            "resolution": resolution_result["response"],
            "escalated": escalation_info is not None,
            "escalation_details": escalation_info,
            "tools_used": (
                [t["tool"] for t in triage_result.get("tool_results", [])] +
                [t["tool"] for t in resolution_result.get("tool_results", [])]
            )
        }

    def _parse_triage_output(self, raw_response: str) -> dict:
        """
        Safely parses the triage agent JSON output.
        Falls back to a default if parsing fails.
        """
        try:
            # Strip markdown code fences if Claude added them
            clean = raw_response.strip()
            if clean.startswith("```"):
                lines = clean.split("\n")
                clean = "\n".join(lines[1:-1])
            return json.loads(clean)
        except json.JSONDecodeError:
            # Fallback if the model returns unexpected format
            return {
                "category": "general",
                "urgency": "medium",
                "summary": raw_response[:200],
                "recommended_agent": "resolution"
            }

import json
from orchestrator import SupportOrchestrator


def print_final_response(result: dict):
    """Pretty-prints the final orchestrator result for the demo."""
    print("\n" + "="*60)
    print("FINAL SYSTEM OUTPUT")
    print("="*60)
    print(f"\nTriage Classification:")
    print(f"  Category : {result['triage'].get('category', 'N/A')}")
    print(f"  Urgency  : {result['triage'].get('urgency', 'N/A')}")
    print(f"  Summary  : {result['triage'].get('summary', 'N/A')}")
    print(f"\nTools Used: {', '.join(result['tools_used']) if result['tools_used'] else 'None'}")
    print(f"\nEscalated: {'Yes' if result['escalated'] else 'No'}")

    if result["escalated"] and result["escal