← Back to Blog

If you've searched for how to build AI agents with the Claude API, you've probably hit tutorials that stop right before the part that actually matters — making multiple agents talk to each other and do real work. That's what this tutorial fixes.

We're going to build a production-ready multi-agent customer service system in Python using the Anthropic SDK. No toy examples, no pseudocode.

What You'll Build

You'll build a three-agent system that automatically handles inbound customer service requests from start to finish. A triage agent reads the request, decides urgency and category, then hands it off to either a resolution agent (for standard issues) or an escalation agent (for complex or high-priority cases).

Each agent has its own tools — ticket creation, knowledge base lookup, and escalation flagging — and they pass context to each other through a shared orchestrator loop. By the end, you'll have something you can actually drop into a real support workflow.

Prerequisites

  • Python 3.10 or higher installed
  • An Anthropic API key (get one at console.anthropic.com)
  • Basic familiarity with Python classes and async concepts
  • The anthropic and python-dotenv packages installed
  • A terminal and a code editor you're comfortable with
📦 Full Source Code
All the code snippets in this tutorial are designed to work together as a single system. Follow the steps in order and you'll have a complete, running multi-agent customer service pipeline by the end. Copy each block into the corresponding file as we go.

Step 1: Set Up Your Claude API Environment and Dependencies

First, install your dependencies. Open your terminal and run this — it takes about ten seconds.

terminal 
pip install anthropic python-dotenv

Next, create a .env file in your project root and add your API key. Never hard-code this in your Python files.

.env 
ANTHROPIC_API_KEY=your_api_key_here

Now create your main project file. We'll build on this throughout the tutorial.

customer_service_agents.py 
import os
import json
from dotenv import load_dotenv
import anthropic

load_dotenv()

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

That's your foundation. The client object is what every agent will use to talk to Claude. Keeping it at the module level means you only instantiate it once.

Step 2: Design Your Agent Roles (Triage, Resolution, Escalation)

Before writing code, get clear on what each agent is actually responsible for. This is where most multi-agent systems fall apart — agents that overlap too much or have no clear handoff logic.

Here's how we're splitting responsibility:

  • Triage Agent: Reads the raw customer message, classifies it (billing, technical, general), scores urgency (low/medium/high), and decides which agent handles it next.
  • Resolution Agent: Handles standard issues using the knowledge base. Creates a ticket and returns a response. This agent handles 80% of cases.
  • Escalation Agent: Takes over for high-urgency or unresolved issues. Flags the ticket for human review and generates a priority alert.

Each agent gets its own system prompt that defines its job. Here they are:

customer_service_agents.py (continued) 
TRIAGE_SYSTEM_PROMPT = """You are a triage agent for a customer service system.
Your job is to analyze incoming customer messages and determine:
1. The category: billing, technical, or general
2. The urgency level: low, medium, or high
3. Which agent should handle it: resolution or escalation

Use 'escalation' for anything marked high urgency or that involves account suspension,
data loss, legal threats, or repeated unresolved issues.
Always use the classify_ticket tool to record your decision."""

RESOLUTION_SYSTEM_PROMPT = """You are a resolution agent for a customer service system.
Your job is to resolve standard customer issues using the knowledge base.
Always search the knowledge base before responding.
Create a ticket for every interaction using the create_ticket tool.
Be concise, helpful, and solution-focused. If you cannot resolve the issue
with the available knowledge base information, recommend escalation."""

ESCALATION_SYSTEM_PROMPT = """You are an escalation agent for a customer service system.
Your job is to handle high-priority or complex issues that the resolution agent
could not solve. Use the escalate_ticket tool to flag the issue for human review.
Acknowledge the customer's frustration, explain that a specialist will follow up,
and provide a realistic timeframe. Always create a ticket first."""

Step 3: Define Tools and Function Calling for Each Agent

This is where Claude's tool use feature does the heavy lifting. Each agent gets a specific set of tools — Claude decides when and how to call them based on context.

We're defining three tools: classify_ticket for triage, knowledge_base_lookup for resolution, and escalate_ticket for escalation. We also give the resolution agent access to create_ticket.

customer_service_agents.py (continued) 
TRIAGE_TOOLS = [
    {
        "name": "classify_ticket",
        "description": "Classify a customer support ticket with category, urgency, and routing decision.",
        "input_schema": {
            "type": "object",
            "properties": {
                "category": {
                    "type": "string",
                    "enum": ["billing", "technical", "general"],
                    "description": "The category of the customer issue"
                },
                "urgency": {
                    "type": "string",
                    "enum": ["low", "medium", "high"],
                    "description": "Urgency level of the issue"
                },
                "route_to": {
                    "type": "string",
                    "enum": ["resolution", "escalation"],
                    "description": "Which agent should handle this ticket"
                },
                "summary": {
                    "type": "string",
                    "description": "A one-sentence summary of the customer issue"
                }
            },
            "required": ["category", "urgency", "route_to", "summary"]
        }
    }
]

RESOLUTION_TOOLS = [
    {
        "name": "knowledge_base_lookup",
        "description": "Search the knowledge base for articles relevant to the customer issue.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The search query to find relevant knowledge base articles"
                }
            },
            "required": ["query"]
        }
    },
    {
        "name": "create_ticket",
        "description": "Create a support ticket and return a ticket ID.",
        "input_schema": {
            "type": "object",
            "properties": {
                "customer_issue": {
                    "type": "string",
                    "description": "Description of the customer issue"
                },
                "resolution": {
                    "type": "string",
                    "description": "The resolution or response provided to the customer"
                },
                "status": {
                    "type": "string",
                    "enum": ["resolved", "pending", "escalated"],
                    "description": "Current status of the ticket"
                }
            },
            "required": ["customer_issue", "resolution", "status"]
        }
    }
]

ESCALATION_TOOLS = [
    {
        "name": "create_ticket",
        "description": "Create a support ticket and return a ticket ID.",
        "input_schema": {
            "type": "object",
            "properties": {
                "customer_issue": {
                    "type": "string",
                    "description": "Description of the customer issue"
                },
                "resolution": {
                    "type": "string",
                    "description": "The resolution or response provided to the customer"
                },
                "status": {
                    "type": "string",
                    "enum": ["resolved", "pending", "escalated"],
                    "description": "Current status of the ticket"
                }
            },
            "required": ["customer_issue", "resolution", "status"]
        }
    },
    {
        "name": "escalate_ticket",
        "description": "Flag a ticket for human specialist review with priority level.",
        "input_schema": {
            "type": "object",
            "properties": {
                "ticket_id": {
                    "type": "string",
                    "description": "The ID of the ticket to escalate"
                },
                "priority": {
                    "type": "string",
                    "enum": ["urgent", "high", "normal"],
                    "description": "Priority level for the human review queue"
                },
                "specialist_notes": {
                    "type": "string",
                    "description": "Notes for the human specialist reviewing this ticket"
                }
            },
            "required": ["ticket_id", "priority", "specialist_notes"]
        }
    }
]

Now we need the actual tool execution functions — the Python code that runs when Claude decides to call a tool. In production, these would connect to your real database or ticketing system. Here they simulate realistic responses.

customer_service_agents.py (continued) 
import random
import string

def generate_ticket_id():
    """Generate a random ticket ID like TKT-A3X7."""
    suffix = ''.join(random.choices(string.ascii_uppercase + string.digits, k=4))
    return f"TKT-{suffix}"

def execute_tool(tool_name: str, tool_input: dict) -> str:
    """Execute a tool call and return a JSON string result."""

    if tool_name == "classify_ticket":
        return json.dumps({
            "status": "classified",
            "category": tool_input["category"],
            "urgency": tool_input["urgency"],
            "route_to": tool_input["route_to"],
            "summary": tool_input["summary"]
        })

    elif tool_name == "knowledge_base_lookup":
        # Simulated knowledge base articles based on query keywords
        query = tool_input["query"].lower()
        if "password" in query or "login" in query:
            return json.dumps({
                "articles": [
                    {"title": "Reset Your Password", "content": "Go to Settings > Security > Reset Password. You'll receive an email within 2 minutes."},
                    {"title": "Account Locked After Failed Attempts", "content": "After 5 failed login attempts, accounts are locked for 30 minutes automatically."}
                ]
            })
        elif "billing" in query or "charge" in query or "invoice" in query:
            return json.dumps({
                "articles": [
                    {"title": "Understanding Your Invoice", "content": "Invoices are generated on the 1st of each month. Charges reflect usage from the prior month."},
                    {"title": "Dispute a Charge", "content": "To dispute a charge, email [email protected] with your invoice number and reason."}
                ]
            })
        elif "slow" in query or "performance" in query or "speed" in query:
            return json.dumps({
                "articles": [
                    {"title": "Performance Troubleshooting", "content": "Clear your browser cache, disable extensions, and test on a different network. If issues persist, check our status page."},
                    {"title": "System Requirements", "content": "Minimum: 8GB RAM, modern browser (Chrome 90+, Firefox 88+, Safari 14+)."}
                ]
            })
        else:
            return json.dumps({
                "articles": [
                    {"title": "Contact Support", "content": "For issues not covered in our knowledge base, our support team is available Monday-Friday 9am-6pm EST."}
                ]
            })

    elif tool_name == "create_ticket":
        ticket_id = generate_ticket_id()
        return json.dumps({
            "ticket_id": ticket_id,
            "status": tool_input["status"],
            "message": f"Ticket {ticket_id} created successfully."
        })

    elif tool_name == "escalate_ticket":
        return json.dumps({
            "escalation_id": f"ESC-{tool_input['ticket_id']}",
            "priority": tool_input["priority"],
            "assigned_to": "Senior Support Specialist",
            "eta": "2-4 business hours",
            "message": f"Ticket {tool_input['ticket_id']} has been escalated with {tool_input['priority']} priority."
        })

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

Step 4: Implement the Orchestration Loop with Agent Coordination

This is the core of the whole system. The orchestrator handles the conversation loop for each agent, including tool use cycles, and then routes the result to the next agent based on what triage decided.

customer_service_agents.py (continued) 
class CustomerServiceOrchestrator:
    """
    Orchestrates a multi-agent customer service pipeline.
    Triage -> Resolution or Escalation, with full tool use support.
    """

    def __init__(self):
        self.client = client
        self.model = MODEL

    def run_agent(
        self,
        system_prompt: str,
        tools: list,
        messages: list,
        agent_name: str
    ) -> tuple[list, dict]:
        """
        Run a single agent loop until it stops calling tools.
        Returns the updated message history and the final text response.
        """
        print(f"\n{'='*50}")
        print(f"  AGENT: {agent_name}")
        print(f"{'='*50}")

        tool_results_context = {}

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

            # Add assistant response to message history
            messages.append({
                "role": "assistant",
                "content": response.content
            })

            # If Claude is done with tool calls, extract final text and return
            if response.stop_reason == "end_turn":
                final_text = ""
                for block in response.content:
                    if hasattr(block, "text"):
                        final_text = block.text
                        break
                print(f"\n[{agent_name}] Final response:\n{final_text}")
                return messages, {"text": final_text, "context": tool_results_context}

            # Handle tool use — Claude wants to call one or more tools
            if response.stop_reason == "tool_use":
                tool_results = []

                for block in response.content:
                    if block.type == "tool_use":
                        tool_name = block.name
                        tool_input = block.input
                        tool_use_id = block.id

                        print(f"\n[{agent_name}] Calling tool: {tool_name}")
                        print(f"  Input: {json.dumps(tool_input, indent=2)}")

                        result = execute_tool(tool_name, tool_input)
                        result_data = json.loads(result)

                        print(f"  Result: {json.dumps(result_data, indent=2)}")

                        # Store tool results for context passing between agents
                        tool_results_context[tool_name] = result_data

                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": tool_use_id,
                            "content": result
                        })

                # Add tool results back into the conversation so Claude can continue
                messages.append({
                    "role": "user",
                    "content": tool_results
                })

    def handle_request(self, customer_message: str) -> dict:
        """
        Main entry point. Takes a raw customer message and runs the full
        triage -> resolution/escalation pipeline.
        """
        print(f"\n{'#'*50}")
        print(f"  NEW REQUEST")
        print(f"  Customer: {customer_message}")
        print(f"{'#'*50}")

        # Step 1: Triage
        triage_messages = [
            {"role": "user", "content": f"Customer message: {customer_message}"}
        ]

        triage_messages, triage_result = self.run_agent(
            system_prompt=TRIAGE_SYSTEM_PROMPT,
            tools=TRIAGE_TOOLS,
            messages=triage_messages,
            agent_name="Triage Agent"
        )

        # Extract triage classification from tool context
        classification = triage_result["context"].get("classify_ticket", {})
        route_to = classification.get("route_to", "resolution")
        summary = classification.get("summary", customer_message)
        urgency = classification.get("urgency", "medium")
        category = classification.get("category", "general")

        print(f"\n[Orchestrator] Routing to: {route_to.upper()} AGENT")
        print(f"[Orchestrator] Category: {category} | Urgency: {urgency}")

        # Step 2: Route to the appropriate agent with triage context
        handoff_context = (
            f"Customer original message: {customer_message}\n"
            f"Triage summary: {summary}\n"
            f"Category: {category} | Urgency: {urgency}"
        )

        if route_to == "escalation":
            agent_messages = [
                {"role": "user", "content": handoff_context}
            ]
            agent_messages, agent_result = self.run_agent(
                system_prompt=ESCALATION_SYSTEM_PROMPT,
                tools=ESCALATION_TOOLS,
                messages=agent_messages,
                agent_name="Escalation Agent"
            )
        else:
            agent_messages = [
                {"role": "user", "content": handoff_context}
            ]
            agent_messages, agent_result = self.run_agent(
                system_prompt=RESOLUTION_SYSTEM_PROMPT,
                tools=RESOLUTION_TOOLS,
                messages=agent_messages,
                agent_name="Resolution Agent"
            )

        return {
            "customer_message": customer_message,
            "triage": classification,
            "routed_to": route_to,
            "final_response": agent_result["text"],
            "tool_context": agent_result["context"]
        }

Step 5: Test with Real Customer Service Scenarios

Now let's put it all together and run three test scenarios — a standard technical issue, a billing question, and a high-urgency escalation case.

customer_service_agents.py (continued) 
def main():
    orchestrator = CustomerServiceOrchestrator()

    test_scenarios = [
        "I can't log into my account. I've tried resetting my password twice and it's still not working.",
        "I was charged twice for my subscription this month and I need a refund immediately.",
        "Your platform has been down for 6 hours and I'm losing $10,000 per hour in sales. "
        "I'm going to take legal action if this isn't fixed in the next 30 minutes."
    ]

    results = []
    for i, scenario in enumerate(test_scenarios, 1):
        print(f"\n\n{'*'*60}")
        print(f"  SCENARIO {i} of {len(test_scenarios)}")
        print(f"{'*'*60}")
        result = orchestrator.handle_request(scenario)
        results.append(result)

        print(f"\n[FINAL SUMMARY - Scenario {i}]")
        print(f"  Routed to:      {result['routed_to'].upper()} AGENT")
        print(f"  Category:       {result['triage'].get('category', 'N/A')}")
        print(f"  Urgency:        {result['triage'].get('urgency', 'N/A')}")
        print(f"  Response:       {result['final_response'][:120]}...")

    return results

if __name__ == "__main__":
    main()

Run it with python customer_service_agents.py and you'll see output like this:

sample output 
**************************************************
  SCENARIO 1 of 3
**************************************************

##################################################
  NEW REQUEST
  Customer: I can't log into my account. I've tried resetting my password twice and it's still not working.
##################################################

==================================================
  AGENT: Triage Agent
==================================================

[Triage Agent] Calling tool: classify_ticket
  Input: {
    "category": "technical",
    "urgency": "medium",
    "route_to": "resolution",
    "summary": "Customer unable to log in despite two password reset attempts."
  }
  Result: {
    "status": "classified",
    "category": "technical",
    "urgency": "medium",
    "route_to": "resolution",
    "summary": "Customer