How to Build a Multi-Agent Restaurant Ordering System with Claude API

import os
import anthropic

# Load your API key from the environment — never hardcode it
ANTHROPIC_API_KEY = os.environ.get("ANTHROPIC_API_KEY")

if not ANTHROPIC_API_KEY:
    raise ValueError("ANTHROPIC_API_KEY environment variable is not set.")

# Initialize the Anthropic client — you'll reuse this across all agents
client = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY)

# Quick sanity check to confirm the connection works
test_response = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=64,
    messages=[{"role": "user", "content": "Say 'API connected' and nothing else."}]
)

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

import os
import anthropic

client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
MODEL = "claude-sonnet-4-5"

# Each agent gets a focused system prompt — specificity matters here
ORDER_TAKER_PROMPT = """
You are a friendly order taker at Bella Napoli, an Italian restaurant in Naples, Florida.
Your job is to collect the customer's food and drink order clearly and completely.
Ask clarifying questions if an item is ambiguous. When you have a complete order, 
summarize it back to the customer for confirmation. Be warm, brief, and efficient.
Do not process payments or make reservations — direct those requests to the right place.
"""

MENU_CHECKER_PROMPT = """
You are the menu availability checker for Bella Napoli restaurant.
When given an order, use the check_menu tool to verify each item exists and is available.
Return a clear list of available items and flag anything that is out of stock or not on the menu.
Be precise. Do not guess — only report what the tool confirms.
"""

RESERVATION_HANDLER_PROMPT = """
You are the reservation agent for Bella Napoli restaurant.
Use the check_reservations tool to look up existing bookings and available time slots.
Use the process_order tool to confirm new reservations once a time is agreed upon.
Always confirm the date, time, party size, and guest name before finalizing anything.
"""

ORCHESTRATOR_PROMPT = """
You are the routing orchestrator for Bella Napoli restaurant's AI system.
Your only job is to read the customer's message and decide which agent should handle it.

Respond with ONLY one of these exact strings:
- ORDER_TAKER — if the customer wants to place a food or drink order
- MENU_CHECKER — if the customer is asking about menu items or availability
- RESERVATION_HANDLER — if the customer wants to book a table or check a reservation
- GENERAL — if none of the above apply (you handle these directly with a brief response)

Do not add explanation. Do not greet the customer. Just output the routing decision.
"""

def get_agent_response(system_prompt: str, user_message: str, tools: list = None) -> str:
    """Generic function to call any agent with a system prompt and optional tools."""
    kwargs = {
        "model": MODEL,
        "max_tokens": 1024,
        "system": system_prompt,
        "messages": [{"role": "user", "content": user_message}]
    }
    if tools:
        kwargs["tools"] = tools

    response = client.messages.create(**kwargs)

    # Handle both text responses and tool use responses
    for block in response.content:
        if block.type == "text":
            return block.text
        if block.type == "tool_use":
            return f"TOOL_CALL:{block.name}:{block.input}"

    return "No response generated."
      

import json
from datetime import datetime

# --- Mock data representing your restaurant's live systems ---

MENU_DATA = {
    "margherita pizza": {"available": True, "price": 16.00, "category": "pizza"},
    "pepperoni pizza": {"available": True, "price": 18.00, "category": "pizza"},
    "truffle pasta": {"available": False, "price": 24.00, "category": "pasta"},
    "caesar salad": {"available": True, "price": 12.00, "category": "salad"},
    "tiramisu": {"available": True, "price": 9.00, "category": "dessert"},
    "sparkling water": {"available": True, "price": 4.00, "category": "drinks"},
    "chianti": {"available": True, "price": 11.00, "category": "drinks"},
    "lobster ravioli": {"available": True, "price": 28.00, "category": "pasta"},
}

RESERVATIONS_DATA = {
    "2026-08-05": {
        "18:00": {"booked": True, "name": "Martinez", "party": 4},
        "18:30": {"booked": False},
        "19:00": {"booked": True, "name": "Thompson", "party": 2},
        "19:30": {"booked": False},
        "20:00": {"booked": False},
        "20:30": {"booked": True, "name": "Kowalski", "party": 6},
    }
}

ORDERS = []

# --- Tool implementation functions ---

def check_menu(item_name: str) -> dict:
    """Look up a menu item by name and return its availability and price."""
    item_key = item_name.lower().strip()
    if item_key in MENU_DATA:
        item = MENU_DATA[item_key]
        return {
            "found": True,
            "item": item_name,
            "available": item["available"],
            "price": item["price"],
            "category": item["category"]
        }
    return {
        "found": False,
        "item": item_name,
        "message": f"'{item_name}' is not on our menu."
    }

def check_reservations(date: str, party_size: int) -> dict:
    """Return available time slots for a given date and party size."""
    if date not in RESERVATIONS_DATA:
        return {"date": date, "available_slots": ["18:00", "18:30", "19:00", "19:30", "20:00", "20:30"]}

    slots = RESERVATIONS_DATA[date]
    available = [time for time, info in slots.items() if not info.get("booked", False)]

    return {
        "date": date,
        "party_size": party_size,
        "available_slots": available,
        "message": f"Found {len(available)} available slots on {date}."
    }

def process_order(order_items: list, customer_name: str, order_type: str = "dine-in") -> dict:
    """Process and store an order or reservation confirmation."""
    order_id = f"ORD-{len(ORDERS) + 1001}"
    total = 0.0
    confirmed_items = []

    for item in order_items:
        item_key = item.lower().strip()
        if item_key in MENU_DATA and MENU_DATA[item_key]["available"]:
            confirmed_items.append(item)
            total += MENU_DATA[item_key]["price"]

    order_record = {
        "order_id": order_id,
        "customer_name": customer_name,
        "order_type": order_type,
        "items": confirmed_items,
        "total": round(total, 2),
        "timestamp": datetime.now().isoformat(),
        "status": "confirmed"
    }

    ORDERS.append(order_record)

    return {
        "success": True,
        "order_id": order_id,
        "confirmed_items": confirmed_items,
        "total": round(total, 2),
        "message": f"Order {order_id} confirmed for {customer_name}. Total: ${total:.2f}"
    }

# --- Claude tool schema definitions ---
# These tell Claude exactly how to call each function

TOOL_DEFINITIONS = [
    {
        "name": "check_menu",
        "description": "Check if a specific menu item is available and get its price. Use this before confirming any order.",
        "input_schema": {
            "type": "object",
            "properties": {
                "item_name": {
                    "type": "string",
                    "description": "The exact name of the menu item to check, e.g. 'margherita pizza'"
                }
            },
            "required": ["item_name"]
        }
    },
    {
        "name": "check_reservations",
        "description": "Look up available reservation time slots for a specific date and party size.",
        "input_schema": {
            "type": "object",
            "properties": {
                "date": {
                    "type": "string",
                    "description": "The date for the reservation in YYYY-MM-DD format"
                },
                "party_size": {
                    "type": "integer",
                    "description": "The number of guests in the party"
                }
            },
            "required": ["date", "party_size"]
        }
    },
    {
        "name": "process_order",
        "description": "Confirm and store a completed food order or reservation after items are verified.",
        "input_schema": {
            "type": "object",
            "properties": {
                "order_items": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "List of menu item names to include in the order"
                },
                "customer_name": {
                    "type": "string",
                    "description": "The customer's name for the order"
                },
                "order_type": {
                    "type": "string",
                    "enum": ["dine-in", "takeout", "reservation"],
                    "description": "Type of order being processed"
                }
            },
            "required": ["order_items", "customer_name"]
        }
    }
]
      

import os
import json
import anthropic
from agents import (
    client, MODEL,
    ORDER_TAKER_PROMPT,
    MENU_CHECKER_PROMPT,
    RESERVATION_HANDLER_PROMPT,
    ORCHESTRATOR_PROMPT,
    get_agent_response
)
from tools import (
    check_menu,
    check_reservations,
    process_order,
    TOOL_DEFINITIONS
)

def route_request(user_message: str) -> str:
    """Use the orchestrator to decide which agent handles the message."""
    response = client.messages.create(
        model=MODEL,
        max_tokens=32,
        system=ORCHESTRATOR_PROMPT,
        messages=[{"role": "user", "content": user_message}]
    )
    return response.content[0].text.strip()

def execute_tool_call(tool_name: str, tool_input: dict) -> str:
    """Execute the appropriate tool function and return the result as a JSON string."""
    if tool_name == "check_menu":
        result = check_menu(**tool_input)
    elif tool_name == "check_reservations":
        result = check_reservations(**tool_input)
    elif tool_name == "process_order":
        result = process_order(**tool_input)
    else:
        result = {"error": f"Unknown tool: {tool_name}"}

    return json.dumps(result)

def handle_with_agent(agent_prompt: str, user_message: str, use_tools: bool = False) -> str:
    """
    Run a full agent turn, including tool use resolution if needed.
    This handles the back-and-forth between Claude and the tool executor.
    """
    messages = [{"role": "user", "content": user_message}]
    tools = TOOL_DEFINITIONS if use_tools else []

    while True:
        kwargs = {
            "model": MODEL,
            "max_tokens": 1024,
            "system": agent_prompt,
            "messages": messages
        }
        if tools:
            kwargs["tools"] = tools

        response = client.messages.create(**kwargs)

        # If Claude is done (no tool calls), return the final text
        if response.stop_reason == "end_turn":
            for block in response.content:
                if hasattr(block, "text"):
                    return block.text
            return "Task completed."

        # If Claude wants to use a tool, execute it and feed the result back
        if response.stop_reason == "tool_use":
            # Add Claude's response (including tool call) to the conversation
            messages.append({"role": "assistant", "content": response.content})

            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    tool_result = execute_tool_call(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": tool_result
                    })

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

        # Fallback for unexpected stop reasons
        return "An unexpected issue occurred. Please try again."

def process_customer_message(user_message: str) -> dict:
    """
    Main entry point. Route the message to the right agent and return the response.
    Returns a dict with the routing decision and the agent's response.
    """
    print(f"\n[CUSTOMER]: {user_message}")

    # Step 1: Orchestrator decides which agent gets the message
    route = route_request(user_message)
    print(f"[ORCHESTRATOR]: Routing to → {route}")

    # Step 2: Hand off to the appropriate agent
    if route == "ORDER_TAKER":
        response = handle_with_agent(ORDER_TAKER_PROMPT, user_message, use_tools=False)
        agent_used = "Order Taker"

    elif route == "MENU_CHECKER":
        response = handle_with_agent(MENU_CHECKER_PROMPT, user_message, use_tools=True)
        agent_used = "Menu Checker"

    elif route == "RESERVATION_HANDLER":
        response = handle_with_agent(RESERVATION_HANDLER_PROMPT, user_message, use_tools=True)
        agent_used = "Reservation Handler"

    else:
        # GENERAL — orchestrator handles simple questions directly
        response = handle_with_agent(
            "You are a helpful assistant for Bella Napoli restaurant in Naples, Florida. Answer briefly and warmly.",
            user_message,
            use_tools=False
        )
        agent_used = "General Assistant"

    print(f"[{agent_used.upper()}]: {response}\n")

    return {
        "route": route,
        "agent_used": agent_used,
        "response": response
    }
      

import os
import anthropic
from orchestrator import process_customer_message, handle_with_agent
from agents import RESERVATION_HANDLER_PROMPT

client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
MODEL = "claude-sonnet-4-5"

def stream_agent_response(system_prompt: str, user_message: str) -> str:
    """
    Stream a response from a specific agent in real time.
    Useful for longer responses like order summaries.
    """
    full_response = ""
    print("[STREAMING RESPONSE]: ", end="", flush=True)

    with client.messages.stream(
        model=MODEL,
        max_tokens=512,
        system=system_prompt,
        messages=[{"role": "user", "content": user_message}]
    ) as stream:
        for text_chunk in stream.text_stream:
            print(text_chunk, end="", flush=True)
            full_response += text_chunk

    print()  # Newline after streaming finishes
    return full_response

def run_restaurant_session():
    """
    Interactive multi-turn session loop.
    Each message is independently routed — this simulates a real customer interaction.
    """
    print("=" * 60)
    print("  Bella Napoli AI Ordering System — Powered by Claude API")
    print("  Built by Naples AI | naplesai.agency")
    print("=" * 60)
    print("Type your message below. Type 'quit' to exit.\n")

    while True:
        user_input = input("You: ").strip()

        if not user_input:
            continue

        if user_input.lower() in ("quit", "exit", "q"):
            print("\nThanks for visiting Bella Napoli! See you soon. 🍕")
            break

        # Route and respond
        result = process_customer_message(user_input)
        print(f"\nBella Napoli AI: {result['response']}\n")
        print("-" * 60)

def run_demo_sequence():
    """
    Run a scripted demo showing all three agents in action.
    This is useful for testing and for showing clients what the system can do.
    """
    demo_messages = [
        "What are your hours tonight?",
        "Do you have truffle pasta available?",
        "I'd like to order a margherita pizza and a caesar salad for Maria Gonzalez.",
        "I need to book a table for 3 people on August 5th.",
        "Can you check if the lobster ravioli is on the menu?"
    ]

    print("\n" + "=" * 60)
    print("  DEMO MODE — Scripted Multi-Agent Interaction")
    print("=" * 60 + "\n")

    for message in demo_messages:
        result = process_customer_message(message)
        print(f"  → Agent Used: {result['agent_used']}")
        print(f"  → Response: {result['response'][:120]}...")
        print()

if __name__ == "__main__":
    import sys
    if len(sys.argv) > 1 and sys.argv[1] == "demo":
        run_demo_sequence()
    else:
        run_restaurant_session()