← Back to Blog

If you've ever watched your support team drown in repetitive tickets — password resets, order status questions, basic troubleshooting — you already know the problem this tutorial solves. By the end of this guide, you'll have a working Claude API multi-agent system that automatically routes incoming support tickets, queries a knowledge base, and resolves common issues without a human touching them. This is the same pattern we use at Naples AI when we build autonomous customer service agents for local businesses across Southwest Florida.

📦 Full Source Code
The complete, working code for this Claude API multi-agent system is spread across the steps below. Each step builds on the last, so by Step 4 you'll have everything you need to run it end to end. Copy each block in order and you're good to go.

What You'll Build

You're building a multi-agent customer support system in Python using the Anthropic SDK. It includes a ticket-routing agent, a knowledge-base lookup tool, and an orchestrator that decides which agent handles which request. The system accepts a raw customer message, classifies it, pulls relevant answers from a knowledge base, and returns a resolved response — autonomously.

Prerequisites

  • Python 3.10 or higher installed
  • An Anthropic API key (get one at console.anthropic.com)
  • Basic familiarity with Python classes and functions
  • anthropic Python SDK installed (pip install anthropic)
  • A terminal and a code editor (VS Code works great)

Step 1: Set Up Your Claude API Environment and Dependencies

First things first — let's get the environment wired up. You only need the Anthropic SDK for this project; no LangChain required unless you want to extend it later.

Create a project folder, set up a virtual environment, and install the SDK:

terminal 
mkdir claude-support-agents
cd claude-support-agents
python -m venv venv
source venv/bin/activate   # Windows: venv\Scripts\activate
pip install anthropic python-dotenv

Now create a .env file in your project root to store your API key safely. Never hardcode credentials in source files.

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

Create your main file and confirm the SDK loads correctly before writing any agent logic:

verify_setup.py 
import os
from dotenv import load_dotenv
import anthropic

load_dotenv()

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

# Quick ping to confirm the API key is valid
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=64,
    messages=[{"role": "user", "content": "Say 'API connected' and nothing else."}]
)

print(response.content[0].text)
# Expected output: API connected

Run python verify_setup.py and you should see API connected printed in your terminal. If you get an auth error, double-check the key in your .env file.

Step 2: Define Tool Functions for Ticket Routing and Knowledge Lookup

This is where the real meat of a Claude API multi-agent system lives. Tools are functions Claude can call during its reasoning loop — think of them as capabilities you're handing to the model. We'll define two: one to classify and route a ticket, and one to search a knowledge base.

Claude's tool-use feature works by passing a tools list to the API. The model decides when to call a tool and what arguments to pass. You handle the actual execution on your end and return the result.

tools.py 
import json

# ── Fake knowledge base (replace with your real DB or vector store) ──────────
KNOWLEDGE_BASE = {
    "password_reset": "To reset your password, visit /account/reset and enter your email. You'll receive a link within 2 minutes.",
    "order_status": "You can track your order at /orders using your order ID. Orders ship within 1-2 business days.",
    "billing_issue": "For billing questions, log in and go to /billing. Charges appear within 3-5 business days.",
    "technical_error": "Try clearing your browser cache and cookies first. If the issue persists, disable browser extensions.",
    "refund_request": "Refunds are processed within 5-7 business days back to your original payment method.",
    "account_locked": "Accounts lock after 5 failed login attempts. Wait 30 minutes or contact support for an immediate unlock.",
    "general": "Our support team is available Monday–Friday, 9am–6pm ET. Average response time is under 2 hours."
}

# ── Tool schemas passed to Claude ─────────────────────────────────────────────
TOOL_DEFINITIONS = [
    {
        "name": "route_ticket",
        "description": (
            "Classifies an incoming customer support message into a category "
            "and assigns a priority level. Call this first for every new ticket."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "category": {
                    "type": "string",
                    "enum": [
                        "password_reset", "order_status", "billing_issue",
                        "technical_error", "refund_request", "account_locked", "general"
                    ],
                    "description": "The category that best matches the customer's issue."
                },
                "priority": {
                    "type": "string",
                    "enum": ["low", "medium", "high", "urgent"],
                    "description": "Priority based on urgency and business impact."
                },
                "summary": {
                    "type": "string",
                    "description": "One sentence summarizing the customer's issue."
                }
            },
            "required": ["category", "priority", "summary"]
        }
    },
    {
        "name": "lookup_knowledge_base",
        "description": (
            "Retrieves the relevant help article or resolution script for a "
            "given support category. Call this after routing to get answer content."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "category": {
                    "type": "string",
                    "enum": [
                        "password_reset", "order_status", "billing_issue",
                        "technical_error", "refund_request", "account_locked", "general"
                    ],
                    "description": "The ticket category to look up in the knowledge base."
                }
            },
            "required": ["category"]
        }
    }
]


def route_ticket(category: str, priority: str, summary: str) -> dict:
    """Simulates writing the ticket to a routing system."""
    ticket_id = f"TKT-{abs(hash(summary)) % 100000:05d}"
    return {
        "ticket_id": ticket_id,
        "category": category,
        "priority": priority,
        "summary": summary,
        "status": "routed"
    }


def lookup_knowledge_base(category: str) -> dict:
    """Returns the knowledge base article for the given category."""
    article = KNOWLEDGE_BASE.get(category, KNOWLEDGE_BASE["general"])
    return {
        "category": category,
        "resolution": article,
        "found": category in KNOWLEDGE_BASE
    }


def execute_tool(tool_name: str, tool_input: dict) -> str:
    """Routes tool calls to the correct Python function and returns JSON string."""
    if tool_name == "route_ticket":
        result = route_ticket(**tool_input)
    elif tool_name == "lookup_knowledge_base":
        result = lookup_knowledge_base(**tool_input)
    else:
        result = {"error": f"Unknown tool: {tool_name}"}
    return json.dumps(result)
💡 Tip: In a real production system, lookup_knowledge_base would call a vector database like Pinecone or pgvector. The fake dict here is just to keep the tutorial self-contained and runnable without any extra accounts.

Step 3: Create the Multi-Agent Orchestrator Class

The orchestrator is the brain of the system. It holds the conversation state, sends messages to Claude, and decides what to do when Claude calls a tool. Everything routes through this one class.

I structured this as a class so you can instantiate one orchestrator per incoming ticket and keep conversation history isolated. That's important when you're running multiple tickets in parallel.

orchestrator.py 
import os
import json
from dotenv import load_dotenv
import anthropic
from tools import TOOL_DEFINITIONS, execute_tool

load_dotenv()

SYSTEM_PROMPT = """You are a customer support AI agent for a SaaS company.

Your job is to:
1. Call route_ticket to classify and prioritize every incoming message.
2. Call lookup_knowledge_base to retrieve the right resolution.
3. Respond to the customer with a clear, friendly, and complete answer.

Always call both tools before writing your final response.
Be concise. Use plain language. Never make up information not in the knowledge base."""


class SupportOrchestrator:
    """
    Manages the full lifecycle of a single customer support ticket.
    One instance = one ticket conversation.
    """

    def __init__(self):
        self.client = anthropic.Anthropic(
            api_key=os.environ.get("ANTHROPIC_API_KEY")
        )
        self.model = "claude-sonnet-4-6"
        self.conversation_history = []
        self.tool_call_log = []       # Stores every tool call for auditing
        self.ticket_metadata = {}     # Populated after route_ticket runs

    def add_user_message(self, message: str) -> None:
        """Appends a user message to the conversation history."""
        self.conversation_history.append({
            "role": "user",
            "content": message
        })

    def _handle_tool_use(self, tool_use_block) -> dict:
        """
        Executes the tool Claude requested and returns a tool_result block.
        This is the bridge between Claude's decision and your Python functions.
        """
        tool_name = tool_use_block.name
        tool_input = tool_use_block.input
        tool_use_id = tool_use_block.id

        print(f"  [Tool Call] {tool_name}({json.dumps(tool_input, indent=2)})")

        # Run the actual Python function
        result_json = execute_tool(tool_name, tool_input)
        result_dict = json.loads(result_json)

        # Save routing metadata when the routing tool fires
        if tool_name == "route_ticket":
            self.ticket_metadata = result_dict

        # Log every tool call for audit trails
        self.tool_call_log.append({
            "tool": tool_name,
            "input": tool_input,
            "output": result_dict
        })

        print(f"  [Tool Result] {result_json}")

        return {
            "type": "tool_result",
            "tool_use_id": tool_use_id,
            "content": result_json
        }

    def call_claude(self) -> anthropic.types.Message:
        """Sends the current conversation history to Claude and returns the response."""
        return self.client.messages.create(
            model=self.model,
            max_tokens=1024,
            system=SYSTEM_PROMPT,
            tools=TOOL_DEFINITIONS,
            messages=self.conversation_history
        )

Step 4: Build the Agent Run Loop with Message Handling

This is the part most tutorials skip over — the actual agentic loop. Claude doesn't always finish in one call. When it needs to use a tool, it stops and waits for your code to execute the tool and return the result. You keep looping until Claude's stop_reason is end_turn.

Add the following method to your SupportOrchestrator class, then add the process_ticket wrapper and the main runner below it:

orchestrator.py (continued — add to the class, then add the runner) 
    def run_agent_loop(self) -> str:
        """
        The core agentic loop. Keeps calling Claude and handling tool use
        until Claude signals it's done with stop_reason == 'end_turn'.
        Returns Claude's final text response.
        """
        max_iterations = 10   # Safety cap to prevent runaway loops
        iteration = 0

        while iteration < max_iterations:
            iteration += 1
            print(f"\n--- Agent Loop Iteration {iteration} ---")

            response = self.call_claude()
            print(f"Stop reason: {response.stop_reason}")

            # Collect tool results if Claude requested any tools
            tool_results = []
            final_text = ""

            for block in response.content:
                if block.type == "tool_use":
                    tool_result = self._handle_tool_use(block)
                    tool_results.append(tool_result)
                elif block.type == "text":
                    final_text = block.text

            # If Claude called tools, add its response + results to history and loop
            if response.stop_reason == "tool_use":
                # Append Claude's assistant message (with tool_use blocks)
                self.conversation_history.append({
                    "role": "assistant",
                    "content": response.content
                })
                # Append all tool results as a user message
                self.conversation_history.append({
                    "role": "user",
                    "content": tool_results
                })
                continue   # Go back to the top and call Claude again

            # Claude is done — return its final text answer
            if response.stop_reason == "end_turn":
                return final_text

        return "Error: agent loop exceeded maximum iterations."

    def process_ticket(self, customer_message: str) -> dict:
        """
        Public method: takes a raw customer message and returns a full result dict.
        This is what you'd call from your API endpoint or webhook handler.
        """
        print(f"\n{'='*60}")
        print(f"INCOMING TICKET: {customer_message}")
        print(f"{'='*60}")

        self.add_user_message(customer_message)
        final_response = self.run_agent_loop()

        return {
            "customer_message": customer_message,
            "ticket_metadata": self.ticket_metadata,
            "final_response": final_response,
            "tool_call_log": self.tool_call_log
        }


# ── Runner — demonstrates the full system with three sample tickets ───────────
if __name__ == "__main__":
    test_tickets = [
        "Hi, I forgot my password and can't log in. I've tried three times already.",
        "Where is my order? I placed it four days ago and haven't heard anything.",
        "I was charged twice for my subscription this month and need a refund."
    ]

    for message in test_tickets:
        orchestrator = SupportOrchestrator()   # Fresh instance per ticket
        result = orchestrator.process_ticket(message)

        print(f"\n{'='*60}")
        print("FINAL RESULT")
        print(f"{'='*60}")
        print(f"Ticket ID  : {result['ticket_metadata'].get('ticket_id', 'N/A')}")
        print(f"Category   : {result['ticket_metadata'].get('category', 'N/A')}")
        print(f"Priority   : {result['ticket_metadata'].get('priority', 'N/A')}")
        print(f"Response   :\n{result['final_response']}")
        print(f"{'='*60}\n")

Run the full system with python orchestrator.py. Here's what the output looks like for the first ticket:

sample output 
============================================================
INCOMING TICKET: Hi, I forgot my password and can't log in. I've tried three times already.
============================================================

--- Agent Loop Iteration 1 ---
Stop reason: tool_use
  [Tool Call] route_ticket({
  "category": "password_reset",
  "priority": "medium",
  "summary": "Customer cannot log in after forgetting their password and has made three failed attempts."
})
  [Tool Result] {"ticket_id": "TKT-47821", "category": "password_reset", "priority": "medium", "summary": "Customer cannot log in...", "status": "routed"}
  [Tool Call] lookup_knowledge_base({
  "category": "password_reset"
})
  [Tool Result] {"category": "password_reset", "resolution": "To reset your password, visit /account/reset and enter your email. You'll receive a link within 2 minutes.", "found": true}

--- Agent Loop Iteration 2 ---
Stop reason: end_turn

============================================================
FINAL RESULT
============================================================
Ticket ID  : TKT-47821
Category   : password_reset
Priority   : medium
Response   :
Hi there! No worries — resetting your password is quick and easy.

Head to **/account/reset** and enter your email address. You'll receive a reset link within 2 minutes. Click the link and you'll be back in your account right away.

If you don't see the email, check your spam folder. Let me know if you need anything else!
============================================================

How It Works

Claude doesn't call your Python functions directly — it returns a structured response saying "I want to call this tool with these arguments." Your orchestrator reads that response, runs the function, and hands the result back. Then Claude continues from there.

The loop in run_agent_loop handles this back-and-forth automatically. Each iteration either processes tool calls and feeds results back in, or it hits end_turn and breaks out with Claude's final answer. The conversation history grows with each pass, so Claude always has full context.

The reason we instantiate a fresh SupportOrchestrator per ticket is isolation — each ticket gets its own history, its own tool log, and its own metadata. That makes it easy to log, audit, and debug individual tickets without one conversation bleeding into another.

Common Errors and Fixes

Error 1: anthropic.AuthenticationError — Invalid API Key

anthropic.AuthenticationError: Error code: 401 - {'type': 'error', 'error': {'type': 'authentication_error', 'message': 'invalid x-api-key'}}

This almost always means your .env file isn't loading or the key has a typo. Make sure load_dotenv() is called before you instantiate the Anthropic client, and verify your key starts with sk-ant-. Also confirm you're running the script from the same directory as your .env file.

Error 2: ValidationError on Tool Input Schema

anthropic.BadRequestError: Error code: 400 - {'type': 'error', 'error': {'type': 'invalid_request_error', 'message': "tools.0.input_schema: 'required' is required"}}

The Anthropic SDK requires every tool definition to have a required array inside input_schema, even if it's empty. Check your TOOL_DEFINITIONS list and make sure every tool has "required": [...] set. Missing or misspelled keys in the schema cause this same error.

Error 3: Agent Loop Never Terminates

# No error message — the script just hangs or runs forever
# You'll see "--- Agent Loop Iteration N ---" printing endlessly

This happens when tool results aren't appended to the conversation history in the right format, so Claude keeps requesting tools without ever reaching end_turn. The fix is making sure your tool_results list uses the exact {"type": "tool_result", "tool_use_id": ..., "content": ...} structure. The max_iterations cap in the loop above prevents this from running forever while you debug.

Next Steps

This system is a solid foundation. Here are four directions worth exploring once you have it running:

  • Swap the fake knowledge base for a real vector store. Connect Pinecone or pgvector and replace lookup_knowledge_base with a semantic similarity search. This makes the system handle any question, not just the categories you pre-defined.
  • Add a human escalation tool. Build a third tool called escalate_to_human that fires when priority is urgent or when Claude can't find a resolution. It can trigger a Slack message or create a Zendesk ticket via API.
  • Run tickets in parallel with asyncio. Wrap process_ticket in an async version and use asyncio.gather to process a batch of tickets simultaneously. This is how you handle high-volume queues without waiting for each one to finish.
  • Add memory across sessions. Store ticket_metadata and conversation history in a database keyed by