How to Build a Multi-Agent Customer Support System with Claude API

mkdir support-agent-system
cd support-agent-system
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate
pip install anthropic python-dotenv

ANTHROPIC_API_KEY=your_api_key_here

import os
import json
from dotenv import load_dotenv
import anthropic

# Load environment variables from .env file
load_dotenv()

# Initialize the Anthropic client using the key from your .env
client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

MODEL = "claude-sonnet-4-6"

if __name__ == "__main__":
    print("Anthropic SDK loaded. Client ready.")
    print(f"Using model: {MODEL}")

python support_system.py

# Expected output:
# Anthropic SDK loaded. Client ready.
# Using model: claude-sonnet-4-6

import json

# Tool definitions tell Claude what functions are available and what parameters they accept.
# Claude decides when and how to call them based on the conversation context.

SUPPORT_TOOLS = [
    {
        "name": "lookup_order_status",
        "description": (
            "Retrieves the current status of a customer order by order ID. "
            "Use this when the customer mentions an order number or asks where their order is."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "order_id": {
                    "type": "string",
                    "description": "The unique order identifier, e.g. ORD-10482"
                }
            },
            "required": ["order_id"]
        }
    },
    {
        "name": "get_billing_details",
        "description": (
            "Fetches billing information for a customer account including last charge, "
            "subscription status, and next billing date. Use when the customer asks about "
            "charges, invoices, or payment issues."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "account_id": {
                    "type": "string",
                    "description": "The customer account ID, e.g. ACC-7721"
                }
            },
            "required": ["account_id"]
        }
    },
    {
        "name": "escalate_ticket",
        "description": (
            "Logs and escalates a support ticket to the human support team when the issue "
            "is too complex for automated resolution, or when the customer explicitly requests "
            "a human agent."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "ticket_summary": {
                    "type": "string",
                    "description": "A brief summary of the customer issue for the human agent."
                },
                "priority": {
                    "type": "string",
                    "enum": ["low", "medium", "high", "urgent"],
                    "description": "The urgency level of the ticket."
                },
                "category": {
                    "type": "string",
                    "enum": ["billing", "technical", "shipping", "general"],
                    "description": "The category of the support issue."
                }
            },
            "required": ["ticket_summary", "priority", "category"]
        }
    }
]


def execute_tool(tool_name: str, tool_input: dict) -> str:
    """
    Simulates tool execution. In production, these would call real APIs,
    your database, or your CRM system.
    """
    if tool_name == "lookup_order_status":
        order_id = tool_input.get("order_id", "unknown")
        # Simulated order database response
        mock_orders = {
            "ORD-10482": {"status": "shipped", "carrier": "FedEx", "eta": "2026-10-10"},
            "ORD-99321": {"status": "processing", "carrier": None, "eta": "2026-10-12"},
        }
        order = mock_orders.get(order_id, {"status": "not_found", "carrier": None, "eta": None})
        return json.dumps({"order_id": order_id, **order})

    elif tool_name == "get_billing_details":
        account_id = tool_input.get("account_id", "unknown")
        # Simulated billing database response
        mock_accounts = {
            "ACC-7721": {
                "last_charge": "$49.00",
                "charge_date": "2026-09-15",
                "status": "active",
                "next_billing": "2026-10-15"
            }
        }
        account = mock_accounts.get(account_id, {"status": "not_found"})
        return json.dumps({"account_id": account_id, **account})

    elif tool_name == "escalate_ticket":
        summary = tool_input.get("ticket_summary")
        priority = tool_input.get("priority")
        category = tool_input.get("category")
        # Simulated ticket system response
        ticket_id = "TKT-" + str(abs(hash(summary)) % 90000 + 10000)
        return json.dumps({
            "ticket_id": ticket_id,
            "status": "escalated",
            "priority": priority,
            "category": category,
            "message": f"Ticket {ticket_id} has been logged and assigned to the support team."
        })

    return json.dumps({"error": f"Unknown tool: {tool_name}"})

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-6"

ROUTER_SYSTEM_PROMPT = """
You are a customer support routing agent. Your only job is to classify incoming 
support messages and determine which specialist should handle them.

You must respond with a valid JSON object — nothing else. No explanation, no extra text.

Use this format exactly:
{
  "category": "billing" | "technical" | "shipping" | "general",
  "priority": "low" | "medium" | "high" | "urgent",
  "confidence": 0.0 to 1.0,
  "reason": "one sentence explanation"
}

Classification rules:
- billing: questions about charges, invoices, subscriptions, refunds, payments
- technical: bugs, errors, login issues, integration problems, feature malfunctions  
- shipping: order tracking, delivery, damaged goods, returns
- general: account questions, product info, feedback, anything else
- Priority is urgent only if the customer explicitly says their business is down or data is lost.
"""


def route_ticket(customer_message: str) -> dict:
    """
    Sends the customer message to the routing agent and returns a classification dict.
    """
    response = client.messages.create(
        model=MODEL,
        max_tokens=256,
        system=ROUTER_SYSTEM_PROMPT,
        messages=[
            {"role": "user", "content": customer_message}
        ]
    )

    raw_text = response.content[0].text.strip()

    try:
        # Claude should return clean JSON per the system prompt instructions
        classification = json.loads(raw_text)
    except json.JSONDecodeError:
        # Fallback if the model adds any extra text unexpectedly
        classification = {
            "category": "general",
            "priority": "medium",
            "confidence": 0.5,
            "reason": "Could not parse routing response, defaulted to general."
        }

    return classification


if __name__ == "__main__":
    test_message = "I was charged twice for my subscription this month and I need a refund."
    result = route_ticket(test_message)
    print("Routing result:")
    print(json.dumps(result, indent=2))

python router.py

# Expected output:
# Routing result:
# {
#   "category": "billing",
#   "priority": "high",
#   "confidence": 0.97,
#   "reason": "Customer reports a duplicate charge and is requesting a refund."
# }

import os
import json
from dotenv import load_dotenv
import anthropic
from tools import SUPPORT_TOOLS, execute_tool

load_dotenv()

client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
MODEL = "claude-sonnet-4-6"

# Specialist system prompts — each agent has a focused persona and scope
AGENT_PROMPTS = {
    "billing": """
You are a billing support specialist. You help customers with charges, invoices, 
subscriptions, and refunds. You have access to billing lookup tools.
Be empathetic, clear, and always confirm the resolution before closing.
If you cannot resolve the issue, escalate using the escalate_ticket tool.
""",
    "technical": """
You are a technical support engineer. You diagnose software bugs, login problems,
integration errors, and feature issues. You have access to diagnostic tools.
Ask clarifying questions if needed. Provide step-by-step solutions.
If the issue requires engineering team access, escalate using the escalate_ticket tool.
""",
    "shipping": """
You are a shipping and fulfillment specialist. You help customers track orders,
handle delivery problems, and process returns. You have access to order lookup tools.
Always look up the order ID before responding with status information.
If the order is unresolvable, escalate using the escalate_ticket tool.
""",
    "general": """
You are a general customer support agent. You handle account questions, product 
information requests, and general feedback. Be friendly and helpful.
If the issue is outside your scope, escalate using the escalate_ticket tool.
"""
}


def run_specialist_agent(category: str, customer_message: str) -> str:
    """
    Runs the appropriate specialist agent with an agentic tool-use loop.
    The loop continues until Claude stops calling tools and gives a final response.
    """
    system_prompt = AGENT_PROMPTS.get(category, AGENT_PROMPTS["general"])

    messages = [
        {"role": "user", "content": customer_message}
    ]

    # Agentic loop: keep going until Claude returns stop_reason = "end_turn"
    while True:
        response = client.messages.create(
            model=MODEL,
            max_tokens=1024,
            system=system_prompt,
            tools=SUPPORT_TOOLS,
            messages=messages
        )

        # If Claude is done (no more tool calls), return the final text response
        if response.stop_reason == "end_turn":
            for block in response.content:
                if hasattr(block, "text"):
                    return block.text
            return "I'm sorry, I wasn't able to generate a response."

        # If Claude wants to use a tool, execute it and feed the result back
        if response.stop_reason == "tool_use":
            # Add Claude's response (with tool calls) to the message history
            messages.append({"role": "assistant", "content": response.content})

            # Process each tool call Claude requested
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    tool_result = execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": tool_result
                    })

            # Feed the tool results back as the next user message
            messages.append({"role": "user", "content": tool_results})

        else:
            # Unexpected stop reason — break to avoid infinite loop
            break

    return "An unexpected error occurred while processing your request."


if __name__ == "__main__":
    # Quick test of the billing agent
    test_message = (
        "Hi, I was charged $49 on my account ACC-7721 but I don't recognize this charge. "
        "Can you tell me what it was for?"
    )
    print("Running billing specialist agent...\n")
    result = run_specialist_agent("billing", test_message)
    print("Agent response:")
    print(result)

python agents.py

# Expected output:
# Running billing specialist agent...
#
# Agent response:
# Hi there! I looked up your account ACC-7721 and can see the details clearly.
#
# Here's what I found:
# - Last charge: $49.00 on September 15, 2026
# - Subscription status: Active
# - Next billing date: October 15, 2026
#
# The $49.00 charge is your monthly subscription fee. Your account is currently 
# active and in good standing. If you'd like to review your plan details or 
# discuss a change to your subscription, I'm happy to help with that!
#
# Is there anything else I can assist you with?

import os
import json
import time
from dotenv import load_dotenv
import anthropic
from router import route_ticket
from agents import run_specialist_agent

load_dotenv()

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


class SupportOrchestrator:
    """
    Main orchestrator for the multi-agent customer support system.
    Coordinates routing and specialist agents to handle support tickets end-to-end.
    """

    def __init__(self):
        self.processed_count = 0
        self.session_log = []

    def handle_ticket(self, customer_message: str, customer_id: str = "unknown") -> dict:
        """
        Full pipeline: classify → route → respond.
        Returns a structured result dict with the response and metadata.
        """
        print(f"\n{'='*60}")
        print(f"Processing ticket from customer: {customer_id}")
        print(f"Message: {customer_message[:80]}{'...' if len(customer_message) > 80 else ''}")
        print(f"{'='*60}")

        start_time = time.time()

        # Step 1: Route the ticket to determine category and priority
        print("\n[Router] Classifying ticket...")
        routing = route_ticket(customer_message)
        print(f"[Router] Category: {routing['category']} | Priority: {routing['priority']} | Confidence: {routing['confidence']}")

        # Step 2: Hand off to the appropriate specialist agent
        print(f"\n[Agent] Dispatching to {routing['category']} specialist...")
        agent_response = run_specialist_agent(routing["category"], customer_message)

        elapsed = round(time.time() - start_time, 2)
        self.processed_count += 1

        # Build the structured result
        result = {
            "ticket_id": f"TKT-{self.processed_count:05d}",
            "customer_id": customer_id,
            "category": routing["category"],
            "priority": routing["priority"],
            "routing_confidence": routing["confidence"],
            "routing_reason": routing["reason"],
            "response": agent_response,
            "processing_time_seconds": elapsed
        }

        # Log this session for review
        self.session_log.append(result)

        print(f"\n[Response] Delivered in {elapsed}s")
        print(f"\n{'-'*60}")
        print("CUSTOMER RESPONSE:")
        print(f"{'-'*60}")
        print(agent_response)

        return result

    def process_batch(self, tickets: list) -> list:
        """
        Processes a list of ticket dicts, each with 'message' and 'customer_id' keys.
        Returns a list of result dicts.
        """
        results = []
        for ticket in tickets:
            result = self.handle_ticket(
                customer_message=ticket["message"],
                customer_id=ticket.get("customer_id", "unknown")
            )
            results.append(result)
        return results

    def print_session_summary(self):
        """Prints a summary of all tickets processed in this session."""
        print(f"\n{'='*60}")
        print("SESSION SUMMARY")
        print(f"{'='*60}")
        print(f"Total tickets processed: {self.processed_count}")

        if self.session_log:
            categories = {}
            priorities = {}
            total_time = 0

            for entry in self.session_log:
                cat = entry["category"]
                pri = entry["priority"]
                categories[cat]