) (1).png)
What You'll Build
By the end of this tutorial, you'll have a working multi-agent customer service system that automatically classifies incoming support tickets, routes them to the right handler, searches a knowledge base for answers, and escalates to a human when needed. It runs entirely in Python using the Anthropic SDK with Claude claude-sonnet-4-6 as the reasoning engine. This is the same architecture we use at Naples AI when building customer support automation for local businesses — it's production-ready, not just a demo.
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
anthropicandpython-dotenvpackages installable via pip- A terminal and a code editor (VS Code works great)
The complete working code is built step-by-step throughout this tutorial. Every snippet below is production-ready and connects together into one final system. Copy each section in order and you'll have a fully functional multi-agent customer service bot by the end.
Step 1: Set Up Your Claude API Environment and Dependencies
First, let's get your environment wired up. Create a new project folder and install the two packages you need.
terminalmkdir multi-agent-support cd multi-agent-support pip install anthropic python-dotenv
Next, create a .env file in your project root and add your API key. Never hardcode keys directly in your source files.
ANTHROPIC_API_KEY=sk-ant-your-key-here
Now create the main file. The import block below is what every snippet in this tutorial assumes is at the top of your agent.py file.
import os
import json
import anthropic
from dotenv import load_dotenv
from typing import Optional
load_dotenv()
client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
That's it for setup. The anthropic client object is what we'll pass around throughout the system — one client, reused everywhere.
Step 2: Design Your Agent System Architecture
Before writing more code, it helps to understand what we're actually building. This system has three logical agents working together: a Router Agent that classifies the ticket type, a Resolution Agent that tries to answer it using a knowledge base, and an Escalation Agent that hands off to a human when the bot can't resolve it.
The key insight here is that all three agents share the same Claude model — what makes them different is their system prompts and the tools available to them. Claude uses tools to take actions, not just generate text.
Think of tools as Claude's hands. The model can reason about what to do, but it can only act on the world through tools you define. In our system, the three main tools are:
ticket_lookup, knowledge_base_search, and escalate_to_human.
Here's the data model we'll use to track a support ticket through the system:
agent.py (continued)from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class SupportTicket:
ticket_id: str
customer_name: str
issue: str
category: Optional[str] = None # set by the router agent
status: str = "open" # open | resolved | escalated
resolution: Optional[str] = None
created_at: str = field(default_factory=lambda: datetime.now().isoformat())
# Simulated ticket database for the demo
TICKET_DB = {
"TKT-001": SupportTicket(
ticket_id="TKT-001",
customer_name="Maria Santos",
issue="I was charged twice for my subscription last month.",
),
"TKT-002": SupportTicket(
ticket_id="TKT-002",
customer_name="James Holloway",
issue="The app crashes every time I try to upload a file larger than 10MB.",
),
"TKT-003": SupportTicket(
ticket_id="TKT-003",
customer_name="Linda Park",
issue="How do I export my data to CSV format?",
),
}
Step 3: Define Tool Schemas for Ticket Escalation and Knowledge Base Lookup
Claude uses tool definitions to understand what actions are available and what arguments each action expects. These schemas are just dictionaries — Claude reads them at runtime and decides when to call each one.
Let's define all three tools. The schemas follow Anthropic's tool-use format exactly:
agent.py (continued)TOOLS = [
{
"name": "ticket_lookup",
"description": (
"Retrieve the full details of a support ticket by its ID. "
"Use this to get the customer name, issue description, and current status."
),
"input_schema": {
"type": "object",
"properties": {
"ticket_id": {
"type": "string",
"description": "The unique ticket identifier, e.g. TKT-001",
}
},
"required": ["ticket_id"],
},
},
{
"name": "knowledge_base_search",
"description": (
"Search the internal knowledge base for articles that match a customer issue. "
"Returns the most relevant article content. Use this before escalating."
),
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "A short search phrase describing the customer's problem",
}
},
"required": ["query"],
},
},
{
"name": "escalate_to_human",
"description": (
"Escalate this ticket to a human support agent. "
"Use this when the issue involves billing disputes, account security, "
"bugs that cannot be resolved with documentation, or when the customer "
"is very frustrated. Always include a reason."
),
"input_schema": {
"type": "object",
"properties": {
"ticket_id": {
"type": "string",
"description": "The ticket ID to escalate",
},
"reason": {
"type": "string",
"description": "A brief explanation of why this ticket needs a human",
},
"priority": {
"type": "string",
"enum": ["low", "medium", "high", "urgent"],
"description": "The urgency level for the human agent queue",
},
},
"required": ["ticket_id", "reason", "priority"],
},
},
]
Now let's implement the actual Python functions that get called when Claude uses each tool. These are your real business logic functions:
agent.py (continued)# Simulated knowledge base articles
KNOWLEDGE_BASE = {
"csv export": (
"To export your data to CSV: Go to Settings > Data Management > Export. "
"Select 'CSV' as the format, choose your date range, and click Download. "
"Large exports may take up to 5 minutes to generate."
),
"file upload limit": (
"The standard plan supports file uploads up to 10MB. "
"To upload larger files, upgrade to the Pro plan which supports up to 500MB. "
"Alternatively, compress your file using ZIP before uploading."
),
"duplicate charge": (
"Duplicate charges can occur during payment processing errors. "
"Our billing team reviews all duplicate charge reports within 24 hours. "
"Refunds are processed back to the original payment method within 3-5 business days."
),
"password reset": (
"Visit the login page and click 'Forgot Password'. "
"Enter your email address and you'll receive a reset link within 2 minutes. "
"Check your spam folder if you don't see it."
),
}
def ticket_lookup(ticket_id: str) -> dict:
"""Fetch ticket data from the simulated database."""
ticket = TICKET_DB.get(ticket_id)
if not ticket:
return {"error": f"No ticket found with ID {ticket_id}"}
return {
"ticket_id": ticket.ticket_id,
"customer_name": ticket.customer_name,
"issue": ticket.issue,
"category": ticket.category,
"status": ticket.status,
"created_at": ticket.created_at,
}
def knowledge_base_search(query: str) -> dict:
"""Simple keyword search over the knowledge base articles."""
query_lower = query.lower()
results = []
for keyword, content in KNOWLEDGE_BASE.items():
# Check if any word in the keyword phrase appears in the query
if any(word in query_lower for word in keyword.split()):
results.append({"keyword": keyword, "content": content})
if results:
return {"found": True, "articles": results}
return {"found": False, "message": "No relevant articles found in the knowledge base."}
def escalate_to_human(ticket_id: str, reason: str, priority: str) -> dict:
"""Mark the ticket as escalated and log the handoff details."""
ticket = TICKET_DB.get(ticket_id)
if ticket:
ticket.status = "escalated" # Update the in-memory ticket record
return {
"success": True,
"ticket_id": ticket_id,
"escalated_to": "human_support_queue",
"priority": priority,
"reason": reason,
"message": (
f"Ticket {ticket_id} has been escalated with {priority} priority. "
"A human agent will respond within the next business hour."
),
}
def dispatch_tool(tool_name: str, tool_input: dict) -> str:
"""Route tool calls from Claude to the correct Python function."""
if tool_name == "ticket_lookup":
result = ticket_lookup(**tool_input)
elif tool_name == "knowledge_base_search":
result = knowledge_base_search(**tool_input)
elif tool_name == "escalate_to_human":
result = escalate_to_human(**tool_input)
else:
result = {"error": f"Unknown tool: {tool_name}"}
return json.dumps(result) # Claude expects tool results as strings
Step 4: Build the Main Agent Orchestrator with Prompt Engineering
This is the core of the system. The CustomerServiceOrchestrator class holds the system prompt, manages which tools are available, and sends messages to Claude. The system prompt is where most of your agent behavior lives — it's not magic, it's just very specific instructions.
SYSTEM_PROMPT = """You are a customer service AI agent for a SaaS software company.
Your job is to resolve support tickets efficiently and accurately.
Follow this process for every ticket:
1. Use the ticket_lookup tool to retrieve the full ticket details.
2. Analyze the issue and determine if it can be resolved using the knowledge base.
3. Use knowledge_base_search to find relevant help articles.
4. If you find a relevant article, provide the solution clearly to the customer.
5. If the issue involves billing disputes, account security, or a confirmed software bug
that documentation cannot fix, use escalate_to_human immediately.
6. Always be empathetic, concise, and professional in your responses.
Rules:
- Never guess at solutions. If you're not sure, escalate.
- Always search the knowledge base before escalating.
- When escalating, always set the correct priority:
- urgent: account hacked or complete service outage
- high: billing errors or data loss
- medium: bugs affecting core features
- low: general how-to questions you couldn't answer
- End every response by summarizing what action was taken.
"""
class CustomerServiceOrchestrator:
def __init__(self):
self.client = client
self.model = "claude-sonnet-4-6"
self.system_prompt = SYSTEM_PROMPT
self.tools = TOOLS
self.max_iterations = 10 # Safety cap to prevent infinite loops
def process_ticket(self, ticket_id: str) -> str:
"""
Entry point: takes a ticket ID, runs the agent loop,
and returns the final response string.
"""
print(f"\n{'='*60}")
print(f"Processing ticket: {ticket_id}")
print(f"{'='*60}")
# Start the conversation with the agent's task
initial_message = f"Please handle support ticket {ticket_id}."
conversation_history = [
{"role": "user", "content": initial_message}
]
final_response = self._run_agent_loop(conversation_history)
return final_response
def _run_agent_loop(self, conversation_history: list) -> str:
"""
The core agentic loop. Keeps running until Claude stops
calling tools and returns a final text response.
"""
iteration = 0
while iteration < self.max_iterations:
iteration += 1
print(f"\n[Agent Loop - Iteration {iteration}]")
# Call Claude with current conversation history
response = self.client.messages.create(
model=self.model,
max_tokens=2048,
system=self.system_prompt,
tools=self.tools,
messages=conversation_history,
)
print(f"Stop reason: {response.stop_reason}")
# If Claude is done using tools, return the final text
if response.stop_reason == "end_turn":
for block in response.content:
if hasattr(block, "text"):
return block.text
return "No text response generated."
# If Claude wants to use tools, handle them
if response.stop_reason == "tool_use":
# Add Claude's response (including tool calls) to history
conversation_history.append({
"role": "assistant",
"content": response.content,
})
# Process every tool call Claude made in this turn
tool_results = []
for block in response.content:
if block.type == "tool_use":
print(f" → Tool called: {block.name}")
print(f" Input: {json.dumps(block.input, indent=2)}")
result = dispatch_tool(block.name, block.input)
print(f" Result: {result}")
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result,
})
# Feed tool results back to Claude as the next user turn
conversation_history.append({
"role": "user",
"content": tool_results,
})
return "Agent reached maximum iteration limit without resolving the ticket."
Step 5: Implement the Run Loop and Conversation State Management
The run loop is already wired into the orchestrator above, but let's add the entry point that ties everything together and lets you process multiple tickets in sequence. This is what you'd call from a web server, a queue processor, or a cron job in production.
agent.py (continued)def run_demo():
"""Process all tickets in the demo database and print results."""
orchestrator = CustomerServiceOrchestrator()
for ticket_id in TICKET_DB.keys():
response = orchestrator.process_ticket(ticket_id)
print(f"\n{'─'*60}")
print(f"FINAL AGENT RESPONSE for {ticket_id}:")
print(f"{'─'*60}")
print(response)
print()
if __name__ == "__main__":
run_demo()
Run it with python agent.py and you'll see the agent work through each ticket live in your terminal. The conversation state is maintained inside each call to process_ticket — every tool call and result gets appended to conversation_history so Claude always has the full context.
In this demo, conversation history resets between tickets. In a real production system, you'd persist the history to a database (like Redis or Postgres) so a customer can pick up a conversation where they left off across sessions.
Step 6: Add Error Handling and Fallback Mechanisms
Production systems fail. APIs time out, tool functions raise exceptions, and Claude occasionally returns unexpected responses. Wrapping your key functions with proper error handling prevents one bad ticket from crashing the whole system.
agent.py (add this updated dispatch_tool and process_ticket)import time
def dispatch_tool_safe(tool_name: str, tool_input: dict) -> str:
"""
Safe wrapper around dispatch_tool with error catching and retry logic.
Returns a JSON string in all cases — never raises to the agent loop.
"""
max_retries = 2
for attempt in range(max_retries + 1):
try:
return dispatch_tool(tool_name, tool_input)
except KeyError as e:
# Missing required argument in tool_input
return json.dumps({"error": f"Missing required argument: {str(e)}"})
except Exception as e:
if attempt < max_retries:
print(f" ⚠ Tool error (attempt {attempt + 1}): {e}. Retrying...")
time.sleep(1)
else:
return json.dumps({
"error": f"Tool '{tool_name}' failed after {max_retries + 1} attempts: {str(e)}"
})
def process_ticket_safe(orchestrator: CustomerServiceOrchestrator, ticket_id: str) -> str:
"""
Wraps process_ticket with top-level exception handling.
Falls back to a canned escalation message if Claude API is unreachable.
"""
try:
return orchestrator.process_ticket(ticket_id)
except anthropic.APIConnectionError:
return (
f"⚠ Could not connect to the Claude API while processing {ticket_id}. "
"The ticket has been flagged for manual review."
)
except anthropic.RateLimitError:
return (
f"⚠ API rate limit hit while processing {ticket_id}. "
"Ticket queued for retry in 60 seconds."
)
except anthropic.APIStatusError as e:
return (
f"⚠ API error {e.status_code} while processing {ticket_id}: {e.message}"
)
Replace the dispatch_tool calls inside _run_agent_loop with dispatch_tool_safe and you've got a resilient system that handles failures gracefully instead of crashing.
How It Works: Agent Decision Flow and Tool Usage
Here's what actually happens when you run the system against a ticket like TKT-002 (the file upload crash). Claude receives the instruction, then the loop begins:
Iteration 1: Claude calls ticket_lookup("TKT-002") to read the issue. The tool returns James Holloway's crash report about 10MB file uploads.
Iteration 2: Claude calls knowledge_base_search("file upload crash 10MB"). The knowledge base returns an article explaining the 10MB limit on the standard plan and how to upgrade or compress files.
Iteration 3: Claude sees a valid resolution in the article. It returns a final text message — no more tool calls — and the loop exits.
For TKT-001 (the duplicate charge), Claude will search the knowledge base, find the billing article, but still call escalate_to_human with priority: "high" because the system prompt explicitly instructs it to escalate billing disputes even when docs are available. That's the power of clear prompt engineering — the model follows your business rules.
Here's a sample of what the terminal output actually looks like when you run the demo:
sample output============================================================
Processing ticket: TKT-001
============================================================
[Agent Loop - Iteration 1]
Stop reason: tool_use
→ Tool called: ticket_lookup
Input: {
"ticket_id": "TKT-001"
}
Result: {"ticket_id": "TKT-001", "customer_name": "Maria Santos", "issue": "I was charged twice for my subscription last month.", "category": null,