← Back to Blog

If you've been trying to figure out how to build AI agents that actually handle real customer support work — not just answer FAQs — you're in the right place. Most tutorials show you a single chatbot. This one shows you how to wire together multiple Claude agents that pass work between each other, look up tickets, search a knowledge base, and escalate to a human when needed.

I've built systems like this for businesses here in Southwest Florida — restaurants, car dealerships, healthcare offices — and the multi-agent pattern is what makes them actually useful in production. Let's build one from scratch.

What You'll Build

By the end of this tutorial, you'll have a working Python system with two Claude agents: an orchestrator that handles incoming customer messages and a specialist escalation agent that takes over when things get complicated. The system uses real tool calls — ticket lookup, FAQ search, and escalation routing — running in a proper agentic loop.

This isn't a toy demo. The patterns here are exactly what we use in production customer support deployments. You can run it locally today and extend it into a real product.

📦 Full Source Code
All the code you need is in the steps below, in the exact order you should write it. By Step 4 you'll have a fully working multi-agent system. Each snippet builds on the last — copy them in order and you'll have something running in under 30 minutes.

Prerequisites

  • Python 3.10 or higher installed
  • An Anthropic API key (get one at console.anthropic.com)
  • anthropic Python SDK installed (pip install anthropic)
  • Basic familiarity with Python classes and dictionaries
  • A .env file or environment variable set for ANTHROPIC_API_KEY

Step 1: Set Up Your Claude API Environment

First, let's get the environment right. I keep the API key in a .env file so it never ends up in version control by accident. Install the dependencies you need before writing any agent code.

terminal 
pip install anthropic python-dotenv

Now create your .env file in the project root:

.env 
ANTHROPIC_API_KEY=sk-ant-your-key-here

Then create your main project file and verify the connection works before building anything else. This saves you debugging time later.

verify_connection.py 
import os
from dotenv import load_dotenv
import anthropic

load_dotenv()

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

# Quick smoke test — make sure the key works before building the full system
message = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=64,
    messages=[{"role": "user", "content": "Reply with: API connection successful"}]
)

print(message.content[0].text)

You should see API connection successful printed to your terminal. If you get an authentication error, double-check your key in the .env file.

Step 2: Define Your Support Agent Tools

Tools are how Claude agents interact with the outside world. Think of them as functions Claude can decide to call when it needs information it doesn't have in context. We're building three tools: ticket_lookup, faq_search, and escalate_to_human.

I define tools in two parts: the JSON schema Claude reads to decide when to use them, and the Python functions that actually run when Claude calls them. Keep these close together in the same file so they stay in sync.

tools.py 
import os
from dotenv import load_dotenv

load_dotenv()

# --- Tool Schemas (Claude reads these to decide when and how to call each tool) ---

TOOL_DEFINITIONS = [
    {
        "name": "ticket_lookup",
        "description": (
            "Look up an existing customer support ticket by ticket ID. "
            "Use this when a customer references a ticket number or asks for a status update."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "ticket_id": {
                    "type": "string",
                    "description": "The support ticket ID, e.g. TKT-1042"
                }
            },
            "required": ["ticket_id"]
        }
    },
    {
        "name": "faq_search",
        "description": (
            "Search the knowledge base for answers to common customer questions. "
            "Use this before escalating — most questions have a documented answer."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The customer's question or topic to search for"
                }
            },
            "required": ["query"]
        }
    },
    {
        "name": "escalate_to_human",
        "description": (
            "Escalate this conversation to a human support agent. "
            "Use this when the issue is unresolved after checking tickets and FAQ, "
            "or when the customer is frustrated and requests a human."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "reason": {
                    "type": "string",
                    "description": "Brief explanation of why escalation is needed"
                },
                "priority": {
                    "type": "string",
                    "enum": ["low", "medium", "high", "urgent"],
                    "description": "Priority level for the human agent queue"
                },
                "customer_summary": {
                    "type": "string",
                    "description": "Summary of the conversation so the human agent has context"
                }
            },
            "required": ["reason", "priority", "customer_summary"]
        }
    }
]


# --- Tool Implementations (the actual Python that runs when Claude calls a tool) ---

# Simulated ticket database — swap this for a real DB call in production
MOCK_TICKETS = {
    "TKT-1042": {
        "status": "open",
        "issue": "Customer unable to log in after password reset",
        "created": "2026-07-29",
        "agent_assigned": None
    },
    "TKT-0991": {
        "status": "resolved",
        "issue": "Billing charge discrepancy on July invoice",
        "created": "2026-07-15",
        "agent_assigned": "Sarah M."
    },
    "TKT-1105": {
        "status": "pending",
        "issue": "Integration with third-party CRM not syncing",
        "created": "2026-07-31",
        "agent_assigned": "Dev Team"
    }
}

# Simulated FAQ knowledge base — in production this would hit a vector DB
MOCK_FAQ = {
    "password": "To reset your password, go to Settings > Security > Reset Password. You'll receive an email within 2 minutes.",
    "billing": "Billing questions can be resolved by visiting the Billing tab in your account dashboard. Invoices are generated on the 1st of each month.",
    "refund": "Refunds are processed within 5-7 business days. Contact billing@example.com with your order number.",
    "integration": "Our API supports REST and webhook integrations. Full documentation is at docs.example.com/api.",
    "cancel": "To cancel your subscription, go to Settings > Subscription > Cancel Plan. Your access continues until the end of the billing period.",
    "login": "If you can't log in, try clearing your browser cache or use the password reset flow at example.com/forgot-password."
}


def ticket_lookup(ticket_id: str) -> dict:
    """Look up a ticket by ID and return its details."""
    ticket = MOCK_TICKETS.get(ticket_id.upper())
    if ticket:
        return {"found": True, "ticket_id": ticket_id.upper(), **ticket}
    return {"found": False, "ticket_id": ticket_id, "message": "No ticket found with that ID."}


def faq_search(query: str) -> dict:
    """Search FAQ knowledge base using simple keyword matching."""
    query_lower = query.lower()
    matches = []

    for keyword, answer in MOCK_FAQ.items():
        if keyword in query_lower:
            matches.append({"topic": keyword, "answer": answer})

    if matches:
        return {"found": True, "results": matches}
    return {
        "found": False,
        "message": "No FAQ match found. Consider escalating to a human agent."
    }


def escalate_to_human(reason: str, priority: str, customer_summary: str) -> dict:
    """Trigger escalation and return a confirmation with queue position."""
    # In production this would create a ticket in Zendesk, Intercom, etc.
    return {
        "escalated": True,
        "queue_position": 3,
        "estimated_wait": "8-12 minutes",
        "priority": priority,
        "reason": reason,
        "message": (
            f"I've escalated your case to our support team with {priority} priority. "
            f"A human agent will reach you within 8-12 minutes."
        )
    }


# Dispatcher — maps tool names to their Python functions
TOOL_FUNCTIONS = {
    "ticket_lookup": ticket_lookup,
    "faq_search": faq_search,
    "escalate_to_human": escalate_to_human
}
💡 Note on Mock Data
The ticket database and FAQ here are dictionaries to keep this tutorial self-contained. In a real deployment, ticket_lookup would hit your CRM or helpdesk API, and faq_search would query a vector database like Pinecone or Weaviate. The agent code doesn't change — only the tool implementations do.

Step 3: Create the Lead Agent and Escalation Agent

Now we build the agents themselves. The lead agent handles every incoming message first — it's the one talking to the customer. The escalation agent only activates when the lead agent determines the issue needs deeper handling or a human.

Both agents are Claude, but they have different system prompts and different tool access. That's the whole trick with multi-agent systems — same model, different context and permissions.

agents.py 
import os
import json
from dotenv import load_dotenv
import anthropic
from tools import TOOL_DEFINITIONS, TOOL_FUNCTIONS

load_dotenv()


class SupportAgent:
    """
    Base class for a Claude-powered support agent.
    Handles the core message-sending and tool-execution logic.
    """

    def __init__(self, name: str, system_prompt: str, tools: list, max_tokens: int = 1024):
        self.name = name
        self.system_prompt = system_prompt
        self.tools = tools
        self.max_tokens = max_tokens
        self.client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

    def send_message(self, messages: list) -> tuple[str, list]:
        """
        Send a message to Claude and handle any tool calls it makes.
        Returns the final text response and an updated messages list.
        """
        response = self.client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=self.max_tokens,
            system=self.system_prompt,
            tools=self.tools,
            messages=messages
        )

        # Keep looping as long as Claude wants to use tools
        while 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"  [{self.name}] Calling tool: {tool_name}({json.dumps(tool_input)})")

                    # Execute the tool and capture its output
                    tool_fn = TOOL_FUNCTIONS.get(tool_name)
                    if tool_fn:
                        result = tool_fn(**tool_input)
                    else:
                        result = {"error": f"Unknown tool: {tool_name}"}

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

            # Append the assistant's tool-use turn and the tool results to history
            messages = messages + [
                {"role": "assistant", "content": response.content},
                {"role": "user", "content": tool_results}
            ]

            # Let Claude respond again now that it has the tool results
            response = self.client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=self.max_tokens,
                system=self.system_prompt,
                tools=self.tools,
                messages=messages
            )

        # Extract the final text response
        final_text = ""
        for block in response.content:
            if hasattr(block, "text"):
                final_text += block.text

        # Add the final assistant message to history
        messages = messages + [{"role": "assistant", "content": final_text}]

        return final_text, messages


class LeadSupportAgent(SupportAgent):
    """
    The front-line agent. Handles all initial customer contact,
    looks up tickets, searches FAQ, and decides when to escalate.
    """

    def __init__(self):
        system_prompt = """You are a friendly and efficient customer support agent for a SaaS company.

Your job is to resolve customer issues as quickly as possible using the tools available to you.

Follow this process:
1. Understand what the customer needs.
2. If they mention a ticket number, use ticket_lookup immediately.
3. For general questions, use faq_search to find the answer.
4. If you can resolve the issue with available information, do so clearly and concisely.
5. If the issue is unresolved after checking tickets and FAQ — or if the customer is upset and asks for a human — use escalate_to_human.

Always be concise, warm, and helpful. Don't make the customer repeat themselves."""

        super().__init__(
            name="LeadAgent",
            system_prompt=system_prompt,
            tools=TOOL_DEFINITIONS
        )


class EscalationAgent(SupportAgent):
    """
    Specialist agent that activates when the lead agent determines
    escalation is needed. Prepares a handoff summary for the human queue.
    """

    def __init__(self):
        system_prompt = """You are a senior support specialist reviewing escalated customer cases.

Your job is to:
1. Review the conversation context you've been given.
2. Use ticket_lookup if there's a ticket ID mentioned to get the latest status.
3. Use escalate_to_human with a clear reason, appropriate priority, and a thorough customer_summary.
4. Tell the customer exactly what to expect next — wait time, who will contact them, and how.

Be empathetic. The customer is being escalated because something didn't get resolved.
Acknowledge that, then give them a clear path forward."""

        # Escalation agent only gets the tools it needs
        escalation_tools = [
            t for t in TOOL_DEFINITIONS
            if t["name"] in ["ticket_lookup", "escalate_to_human"]
        ]

        super().__init__(
            name="EscalationAgent",
            system_prompt=system_prompt,
            tools=escalation_tools
        )

Step 4: Build the Orchestration Loop

The orchestration loop is the part that makes this a multi-agent system instead of just a chatbot. It decides which agent handles each turn and when to hand off between them. This is where the real value of the pattern shows up.

I keep the orchestrator as a separate class with its own logic for detecting when escalation happened inside a lead agent response. That keeps the routing rules clean and easy to modify later.

orchestrator.py 
import os
from dotenv import load_dotenv
from agents import LeadSupportAgent, EscalationAgent

load_dotenv()


class SupportOrchestrator:
    """
    Orchestrates conversation flow between the LeadSupportAgent
    and the EscalationAgent. Manages shared conversation history
    and handles agent handoffs.
    """

    def __init__(self):
        self.lead_agent = LeadSupportAgent()
        self.escalation_agent = EscalationAgent()
        self.conversation_history = []
        self.escalated = False

    def _escalation_triggered(self, response_text: str) -> bool:
        """
        Detect if the lead agent's response indicates an escalation happened.
        We check for keywords that appear in the escalate_to_human tool result.
        """
        escalation_signals = [
            "escalated your case",
            "human agent will reach",
            "escalated to our support team",
            "a human agent"
        ]
        response_lower = response_text.lower()
        return any(signal in response_lower for signal in escalation_signals)

    def handle_message(self, user_message: str) -> str:
        """
        Route an incoming customer message to the right agent
        and return the agent's response as a string.
        """
        print(f"\n[Customer] {user_message}")

        # Add the new user message to the shared conversation history
        self.conversation_history.append({
            "role": "user",
            "content": user_message
        })

        if self.escalated:
            # Once escalated, all subsequent messages go to the escalation agent
            print("  [Orchestrator] Routing to EscalationAgent (already escalated)")
            response_text, self.conversation_history = self.escalation_agent.send_message(
                self.conversation_history
            )
        else:
            # Lead agent handles the message first
            print("  [Orchestrator] Routing to LeadAgent")
            response_text, self.conversation_history = self.lead_agent.send_message(
                self.conversation_history
            )

            # Check if the lead agent triggered an escalation
            if self._escalation_triggered(response_text):
                self.escalated = True
                print("  [Orchestrator] Escalation detected — switching to EscalationAgent")

                # Hand off to the escalation agent with full conversation context
                escalation_prompt = (
                    "The lead agent has initiated an escalation. "
                    "Review the conversation above and complete the escalation process "
                    "with a clear handoff to the human queue."
                )

                self.conversation_history.append({
                    "role": "user",
                    "content": escalation_prompt
                })

                response_text, self.conversation_history = self.escalation_agent.send_message(
                    self.conversation_history
                )

        print(f"\n[Agent] {response_text}\n")
        return response_text

    def run_interactive(self):
        """Run a simple command-line loop for testing the system interactively."""
        print("=" * 60)
        print("Naples AI — Multi-Agent Customer Support System")
        print("Type 'quit' to exit")
        print("=" * 60)

        while True:
            user_input = input("\nYou: ").strip()
            if user_input.lower() in ("quit", "exit", "q"):
                print("Session ended.")
                break
            if not user_input:
                continue
            self.handle_message(user_input)


# --- Demo: run a scripted conversation to show the full system in action ---

def run_demo():
    """
    Runs a pre-scripted multi-turn conversation to demonstrate
    ticket lookup, FAQ search, and escalation in sequence.
    """
    orchestrator = SupportOrchestrator()

    demo_messages = [
        "Hi, I'm having trouble logging into my account after resetting my password.",
        "I tried that already. Can you check ticket TKT-1042? I submitted it yesterday.",
        "That ticket has been open for two days with no update. I need to speak to a human right now."
    ]

    print("\n" + "=" * 60)
    print("DEMO: Multi-Agent Customer Support Conversation")
    print("=" * 60)

    for message in demo_messages:
        orchestrator.handle_message(message)
        print("-" * 60)


if __name__ == "__main__":
    run_demo()

Run it with python orchestrator.py and you'll see the full conversation play out in your terminal with routing logs showing which agent handled each turn.

Example Multi-Turn Conversation Output

Here's what the output actually looks like when you run the demo. The routing logs in brackets show exactly what's happening under the hood.

terminal output 
============================================================
DEMO: Multi-Agent Customer Support Conversation
============================================================

[Customer] Hi, I'm having trouble logging into my account after resetting my password.
  [Orchestrator] Routing to LeadAgent
  [LeadAgent] Calling tool: faq_search({"query": "trouble logging in after password reset"})

[Agent] I can help with that! After a password reset, sometimes the browser caches
the old session. Here's what to try:

1. Clear your browser cache and cookies completely
2. Close and reopen your browser
3. Go to example.com/forgot-password and reset your password again

You should receive the reset email within 2 minutes. Let me know if that works!

------------------------------------------------------------

[Customer]