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

pip install anthropic python-dotenv

ANTHROPIC_API_KEY=sk-ant-your-key-here

import os
import json
from dotenv import load_dotenv
import anthropic

load_dotenv()

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

# Quick sanity check — remove after confirming it works
if __name__ == "__main__":
    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=64,
        messages=[{"role": "user", "content": "Say hello in one word."}]
    )
    print(response.content[0].text)

import os
import json
from dotenv import load_dotenv
import anthropic

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

# ── Tool schemas passed to Claude ──────────────────────────────────────────────

TOOLS = [
    {
        "name": "route_to_specialist",
        "description": (
            "Routes the customer's issue to the correct specialist team. "
            "Use this when you can identify the department that owns the problem."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "department": {
                    "type": "string",
                    "enum": ["billing", "technical_support", "sales", "returns", "general"],
                    "description": "The department best suited to handle this issue."
                },
                "priority": {
                    "type": "string",
                    "enum": ["low", "medium", "high"],
                    "description": "Issue priority based on urgency and customer sentiment."
                },
                "summary": {
                    "type": "string",
                    "description": "One-sentence summary of the customer's issue."
                }
            },
            "required": ["department", "priority", "summary"]
        }
    },
    {
        "name": "search_knowledge_base",
        "description": (
            "Searches the internal knowledge base for articles, FAQs, or procedures "
            "relevant to the customer's question. Use this before escalating."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "The search query derived from the customer's message."
                },
                "category": {
                    "type": "string",
                    "enum": ["billing", "product", "shipping", "returns", "account", "general"],
                    "description": "Category filter to narrow search results."
                }
            },
            "required": ["query"]
        }
    },
    {
        "name": "escalate_ticket",
        "description": (
            "Escalates the ticket to a human agent. Use this when the issue is complex, "
            "the customer is upset, or the knowledge base has no relevant answer."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "reason": {
                    "type": "string",
                    "description": "Why this ticket needs human intervention."
                },
                "urgency": {
                    "type": "string",
                    "enum": ["routine", "urgent", "critical"],
                    "description": "How quickly a human needs to respond."
                },
                "customer_sentiment": {
                    "type": "string",
                    "enum": ["neutral", "frustrated", "angry"],
                    "description": "The detected emotional state of the customer."
                }
            },
            "required": ["reason", "urgency", "customer_sentiment"]
        }
    }
]

# ── Python implementations of each tool ───────────────────────────────────────

def route_to_specialist(department: str, priority: str, summary: str) -> dict:
    """Simulates routing to a CRM or ticketing system like Zendesk or HubSpot."""
    ticket_id = f"TKT-{department[:3].upper()}-{hash(summary) % 10000:04d}"
    return {
        "status": "routed",
        "ticket_id": ticket_id,
        "department": department,
        "priority": priority,
        "message": f"Ticket {ticket_id} created and assigned to {department} team."
    }


def search_knowledge_base(query: str, category: str = "general") -> dict:
    """
    Simulates a knowledge base search. In production, replace this with
    a vector DB call (e.g., Pinecone, Weaviate) or an Elasticsearch query.
    """
    mock_articles = {
        "refund": {
            "title": "How to Request a Refund",
            "content": "Refunds are processed within 5-7 business days. Submit via account portal.",
            "url": "/help/refunds"
        },
        "password": {
            "title": "Reset Your Password",
            "content": "Click 'Forgot Password' on the login screen. Check your email for the link.",
            "url": "/help/password-reset"
        },
        "shipping": {
            "title": "Shipping Timelines",
            "content": "Standard shipping takes 3-5 days. Expedited is 1-2 days.",
            "url": "/help/shipping"
        }
    }
    # Simple keyword match — swap for real semantic search in production
    for keyword, article in mock_articles.items():
        if keyword in query.lower():
            return {"found": True, "article": article, "query": query}

    return {
        "found": False,
        "query": query,
        "message": "No matching articles found. Consider escalating to a human agent."
    }


def escalate_ticket(reason: str, urgency: str, customer_sentiment: str) -> dict:
    """Simulates pushing an escalation to a human agent queue."""
    escalation_id = f"ESC-{urgency[:3].upper()}-{hash(reason) % 9999:04d}"
    return {
        "status": "escalated",
        "escalation_id": escalation_id,
        "urgency": urgency,
        "customer_sentiment": customer_sentiment,
        "message": f"Escalation {escalation_id} queued. A human agent will respond shortly."
    }


def execute_tool(tool_name: str, tool_input: dict) -> dict:
    """Central dispatcher — maps tool names to their Python functions."""
    if tool_name == "route_to_specialist":
        return route_to_specialist(**tool_input)
    elif tool_name == "search_knowledge_base":
        return search_knowledge_base(**tool_input)
    elif tool_name == "escalate_ticket":
        return escalate_ticket(**tool_input)
    else:
        return {"error": f"Unknown tool: {tool_name}"}

def run_triage_agent(customer_message: str) -> dict:
    """
    Reads a raw customer message and returns a structured classification.
    This agent doesn't use tools — it just reasons and returns JSON.
    """
    system_prompt = """You are a customer service triage agent. Your only job is to 
classify incoming customer messages. Respond with valid JSON only — no extra text.

Return this exact structure:
{
  "intent": "billing|technical|returns|sales|general",
  "sentiment": "neutral|frustrated|angry",
  "urgency": "low|medium|high",
  "summary": "one sentence describing the issue"
}"""

    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=256,
        system=system_prompt,
        messages=[
            {"role": "user", "content": customer_message}
        ]
    )

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

    # Strip markdown code fences if the model wraps the JSON
    if raw.startswith("```"):
        raw = raw.split("```")[1]
        if raw.startswith("json"):
            raw = raw[4:]

    return json.loads(raw)

def run_knowledge_agent(customer_message: str, triage_result: dict) -> dict:
    """
    Searches the knowledge base based on the triage classification.
    Returns a dict with the search result and a human-readable answer draft.
    """
    kb_tools = [t for t in TOOLS if t["name"] == "search_knowledge_base"]

    messages = [
        {
            "role": "user",
            "content": (
                f"Customer issue: {customer_message}\n\n"
                f"Triage classification: {json.dumps(triage_result)}\n\n"
                "Search the knowledge base for a relevant self-service answer. "
                "Use the customer's actual words in your search query."
            )
        }
    ]

    response = client.messages.create(
        model="claude-sonnet-4-5",
        max_tokens=512,
        tools=kb_tools,
        messages=messages
    )

    result = {"search_performed": False, "kb_result": None, "answer_draft": None}

    for block in response.content:
        if block.type == "tool_use" and block.name == "search_knowledge_base":
            result["search_performed"] = True
            result["kb_result"] = execute_tool("search_knowledge_base", block.input)

    # If a tool was called, ask Claude to draft a response based on what it found
    if result["search_performed"] and result["kb_result"]:
        messages.append({"role": "assistant", "content": response.content})
        messages.append({
            "role": "user",
            "content": [
                {
                    "type": "tool_result",
                    "tool_use_id": next(
                        b.id for b in response.content if b.type == "tool_use"
                    ),
                    "content": json.dumps(result["kb_result"])
                }
            ]
        })

        follow_up = client.messages.create(
            model="claude-sonnet-4-5",
            max_tokens=512,
            tools=kb_tools,
            messages=messages
        )
        result["answer_draft"] = follow_up.content[0].text if follow_up.content else None

    return result

class CustomerServiceOrchestrator:
    """
    Coordinates triage, knowledge lookup, and final resolution.
    Holds conversation history so the orchestrator can handle follow-ups.
    """

    def __init__(self):
        self.conversation_history = []
        self.model = "claude-sonnet-4-5"
        self.system_prompt = """You are the orchestrating agent for a customer service system. 
You receive pre-processed triage data and knowledge base results, then decide the best action.

Your decision process:
1. If the knowledge base found a relevant article AND the issue is not urgent/angry → return the answer.
2. If no KB article was found OR sentiment is frustrated/angry → escalate_ticket.
3. Always route_to_specialist so the right team has a record, regardless of outcome.

Be decisive. Do not ask clarifying questions. Take action with the tools available."""

    def process(self, customer_message: str, triage: dict, kb_data: dict) -> dict:
        """
        Main orchestration method. Takes the outputs of the triage and knowledge
        agents, runs tool calls, and returns a final resolution summary.
        """
        user_content = (
            f"Customer message: {customer_message}\n\n"
            f"Triage result: {json.dumps(triage, indent=2)}\n\n"
            f"Knowledge base result: {json.dumps(kb_data, indent=2)}\n\n"
            "Make a decision and take the appropriate actions using your tools. "
            "Then provide a final customer-facing response."
        )

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

        tool_results_log = []

        # Agentic loop — keeps going until Claude stops calling tools
        while True:
            response = client.messages.create(
                model=self.model,
                max_tokens=1024,
                system=self.system_prompt,
                tools=TOOLS,
                messages=self.conversation_history
            )

            self.conversation_history.append(
                {"role": "assistant", "content": response.content}
            )

            # If Claude is done reasoning and has a final answer
            if response.stop_reason == "end_turn":
                final_text = next(
                    (b.text for b in response.content if hasattr(b, "text")), ""
                )
                return {
                    "final_response": final_text,
                    "tool_calls": tool_results_log,
                    "conversation_turns": len(self.conversation_history)
                }

            # Process any tool calls Claude made
            if response.stop_reason == "tool_use":
                tool_result_blocks = []

                for block in response.content:
                    if block.type != "tool_use":
                        continue

                    tool_output = execute_tool(block.name, block.input)
                    tool_results_log.append({
                        "tool": block.name,
                        "input": block.input,
                        "output": tool_output
                    })

                    tool_result_blocks.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": json.dumps(tool_output)
                    })

                # Feed tool results back so Claude can continue reasoning
                self.conversation_history.append({
                    "role": "user",
                    "content": tool_result_blocks
                })

            else:
                # Unexpected stop reason — bail out cleanly
                break

        return {
            "final_response": "Unable to process request.",
            "tool_calls": tool_results_log,
            "conversation_turns": len(self.conversation_history)
        }

def handle_customer_message(customer_message: str, verbose: bool = True) -> dict:
    """
    Full pipeline: triage → knowledge base → orchestrate → resolve.
    Set verbose=True to see each agent's reasoning as it runs.
    """
    print(f"\n{'='*60}")
    print(f"CUSTOMER MESSAGE: {customer_message}")
    print(f"{'='*60}\n")

    # Stage 1: Triage
    if verbose:
        print("🔍 [TRIAGE AGENT] Classifying message...")
    triage = run_triage_agent(customer_message)
    if verbose:
        print(f"   Intent: {triage.get('intent')} | "
              f"Sentiment: {triage.get('sentiment')} | "
              f"Urgency: {triage.get('urgency')}")

    # Stage 2: Knowledge base lookup
    if verbose:
        print("\n📚 [KNOWLEDGE AGENT] Searching knowledge base...")
    kb_data = run_knowledge_agent(customer_message, triage)
    if verbose:
        found = kb_data.get("kb_result", {}).get("found", False)
        print(f"   KB article found: {found}")
        if found:
            title = kb_data["kb_result"]["article"]["title"]
            print(f"   Article: {title}")

    # Stage 3: Orchestration
    if verbose:
        print("\n🤖 [ORCHESTRATOR AGENT] Making routing decision...")
    orchestrator = CustomerServiceOrchestrator()
    result = orchestrator.process(customer_message, triage, kb_data)

    if verbose:
        print(f"\n   Tools used: {[t['tool'] for t in result['tool_calls']]}")
        print(f"   Conversation turns: {result['conversation_turns']}")

    print(f"\n{'─'*60}")
    print("✅ FINAL RESPONSE TO CUSTOMER:")
    print(f"{'─'*60}")
    print(result["final_response"])
    print(f"{'─'*60}\n")

    return {
        "triage": triage,
        "knowledge_base": kb_data,
        "resolution": result
    }


def run_interactive_loop():
    """Simple CLI for testing the system with real messages."""
    print("\n🚀 Naples AI — Multi-Agent Customer Service System")
    print("Type your customer message and press Enter. Type 'quit' to exit.\n")

    while True:
        message = input("Customer: ").strip()
        if message.lower() in ("quit", "exit", "q"):
            print("Shutting down. Goodbye!")
            break
        if not message:
            continue
        handle_customer_message(message)


if __name__ == "__main__":
    # Run a demo with three different customer scenarios
    test_messages = [
        "I want a refund for my order from last week. It never arrived.",
        "I can't log into my account. I forgot my password.",
        "THIS IS UNACCEPTABLE. I've been waiting 3 weeks and no one responds to my emails!"
    ]

    for msg in test_messages:
        handle_customer_message(msg)
        print("\n")