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

mkdir restaurant-agent-system
cd restaurant-agent-system
pip install anthropic python-dotenv

ANTHROPIC_API_KEY=sk-ant-your-key-here

import os
from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()

def get_client() -> Anthropic:
    """Return a configured Anthropic client using the env API key."""
    api_key = os.environ.get("ANTHROPIC_API_KEY")
    if not api_key:
        raise ValueError("ANTHROPIC_API_KEY environment variable is not set.")
    return Anthropic(api_key=api_key)

# Shared model identifier used across all agents
MODEL = "claude-sonnet-4-5"

from typing import Any

# Tool definition for processing food and drink orders
ORDER_PROCESSOR_TOOL: dict[str, Any] = {
    "name": "order_processor",
    "description": (
        "Process a customer's food or drink order. Use this tool when the customer "
        "wants to order menu items. It validates the items against the menu, calculates "
        "the total, and returns an order confirmation with an order ID."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "items": {
                "type": "array",
                "description": "List of menu items the customer wants to order.",
                "items": {
                    "type": "object",
                    "properties": {
                        "name": {
                            "type": "string",
                            "description": "The name of the menu item."
                        },
                        "quantity": {
                            "type": "integer",
                            "description": "How many of this item the customer wants.",
                            "minimum": 1
                        },
                        "special_instructions": {
                            "type": "string",
                            "description": "Any special requests for this item (optional)."
                        }
                    },
                    "required": ["name", "quantity"]
                }
            },
            "table_number": {
                "type": "integer",
                "description": "The table number associated with this order."
            },
            "customer_name": {
                "type": "string",
                "description": "The name of the customer placing the order."
            }
        },
        "required": ["items", "table_number", "customer_name"]
    }
}

# Tool definition for checking and booking table reservations
RESERVATION_CHECKER_TOOL: dict[str, Any] = {
    "name": "reservation_checker",
    "description": (
        "Check availability and book a table reservation. Use this tool when the customer "
        "wants to make a reservation, check if a time slot is available, or confirm an "
        "existing booking. Returns availability status and a confirmation number if booked."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "date": {
                "type": "string",
                "description": "The date for the reservation in YYYY-MM-DD format."
            },
            "time": {
                "type": "string",
                "description": "The requested time in HH:MM 24-hour format (e.g., 19:30)."
            },
            "party_size": {
                "type": "integer",
                "description": "Number of people in the party.",
                "minimum": 1,
                "maximum": 20
            },
            "customer_name": {
                "type": "string",
                "description": "Full name of the customer making the reservation."
            },
            "customer_phone": {
                "type": "string",
                "description": "Phone number for the reservation confirmation."
            },
            "special_requests": {
                "type": "string",
                "description": "Any special accommodations (e.g., wheelchair access, birthday setup)."
            }
        },
        "required": ["date", "time", "party_size", "customer_name", "customer_phone"]
    }
}

# Bundle all tools together for easy import by agents
ALL_TOOLS = [ORDER_PROCESSOR_TOOL, RESERVATION_CHECKER_TOOL]

import json
import random
import string
from typing import Any
from client import get_client, MODEL
from tools import ORDER_PROCESSOR_TOOL

# A simple in-memory menu. In production, this would hit your POS database.
MENU: dict[str, float] = {
    "margherita pizza": 14.99,
    "pepperoni pizza": 16.99,
    "caesar salad": 10.99,
    "house salad": 9.99,
    "grilled salmon": 24.99,
    "chicken parmesan": 19.99,
    "spaghetti carbonara": 17.99,
    "tiramisu": 7.99,
    "cheesecake": 6.99,
    "sparkling water": 3.99,
    "house wine": 8.99,
    "craft beer": 6.99,
}

ORDER_AGENT_SYSTEM_PROMPT = """You are an order-taking specialist for Bella Napoli, 
an Italian restaurant. Your only job is to help customers place food and drink orders.

When a customer wants to order something:
1. Confirm the items they want and ask for any special instructions if unclear.
2. Use the order_processor tool to submit the order.
3. Tell the customer their order total and confirmation number once it's placed.

If the customer asks about anything other than ordering food or drinks, politely 
let them know you handle orders only, and suggest they speak with the main assistant.
Only call order_processor when you have all the required information."""


def execute_order_processor(tool_input: dict[str, Any]) -> dict[str, Any]:
    """
    Simulate the backend order processing logic.
    In production, this would call your POS system API.
    """
    items = tool_input.get("items", [])
    table_number = tool_input.get("table_number")
    customer_name = tool_input.get("customer_name")

    order_total = 0.0
    validated_items = []
    unknown_items = []

    for item in items:
        item_name = item["name"].lower()
        quantity = item["quantity"]
        special = item.get("special_instructions", "")

        if item_name in MENU:
            item_price = MENU[item_name] * quantity
            order_total += item_price
            validated_items.append({
                "name": item["name"],
                "quantity": quantity,
                "unit_price": MENU[item_name],
                "line_total": round(item_price, 2),
                "special_instructions": special
            })
        else:
            unknown_items.append(item["name"])

    if unknown_items:
        return {
            "success": False,
            "error": f"The following items are not on our menu: {', '.join(unknown_items)}. "
                     f"Please check the menu and try again."
        }

    # Generate a simple order ID
    order_id = "ORD-" + "".join(random.choices(string.ascii_uppercase + string.digits, k=6))
    tax = round(order_total * 0.07, 2)
    total_with_tax = round(order_total + tax, 2)

    return {
        "success": True,
        "order_id": order_id,
        "table_number": table_number,
        "customer_name": customer_name,
        "items": validated_items,
        "subtotal": round(order_total, 2),
        "tax": tax,
        "total": total_with_tax,
        "estimated_wait_minutes": random.randint(15, 30),
        "status": "confirmed"
    }


class OrderAgent:
    """Specialist agent that handles all food and drink order processing."""

    def __init__(self):
        self.client = get_client()
        self.tools = [ORDER_PROCESSOR_TOOL]

    def process(self, customer_message: str, context: dict[str, Any] = None) -> str:
        """
        Process a customer order request and return the agent's response.
        context can include table_number and customer_name from the orchestrator.
        """
        context = context or {}

        # Inject context into the message so the agent has what it needs
        enriched_message = customer_message
        if context.get("table_number"):
            enriched_message += f"\n[Context: Customer is at table {context['table_number']}]"
        if context.get("customer_name"):
            enriched_message += f"\n[Context: Customer name is {context['customer_name']}]"

        messages = [{"role": "user", "content": enriched_message}]

        # Agentic loop: keep going until the model stops using tools
        while True:
            response = self.client.messages.create(
                model=MODEL,
                max_tokens=1024,
                system=ORDER_AGENT_SYSTEM_PROMPT,
                tools=self.tools,
                messages=messages
            )

            # If the model is done, return its final text response
            if response.stop_reason == "end_turn":
                for block in response.content:
                    if hasattr(block, "text"):
                        return block.text
                return "Your order has been processed."

            # Handle tool use blocks in the response
            if response.stop_reason == "tool_use":
                # Add the assistant's response (including tool call) to history
                messages.append({"role": "assistant", "content": response.content})

                tool_results = []
                for block in response.content:
                    if block.type == "tool_use":
                        # Execute the actual tool function
                        if block.name == "order_processor":
                            result = execute_order_processor(block.input)
                        else:
                            result = {"error": f"Unknown tool: {block.name}"}

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

                # Feed results back to the model so it can continue
                messages.append({"role": "user", "content": tool_results})

            else:
                # Unexpected stop reason, bail out gracefully
                return "I encountered an issue processing your order. Please try again."

import json
import random
import string
from datetime import datetime
from typing import Any
from client import get_client, MODEL
from tools import RESERVATION_CHECKER_TOOL

RESERVATION_AGENT_SYSTEM_PROMPT = """You are a reservation specialist for Bella Napoli 
Italian restaurant. Your job is to help customers check availability and book tables.

Restaurant hours: Tuesday through Sunday, 5:00 PM to 10:00 PM. Closed Mondays.
Maximum party size: 20 people. Reservations are held for 15 minutes past booking time.

When a customer wants a reservation:
1. Gather the date, time, party size, name, and phone number.
2. Use the reservation_checker tool to check availability and book the table.
3. Confirm the booking details clearly, including the confirmation number.

If the requested time is outside business hours or the restaurant is closed, let 
the customer know and suggest alternatives. Only use reservation_checker when you 
have all required information from the customer."""

# Simulated reservations database (in production, this hits a real DB)
EXISTING_RESERVATIONS: dict[str, list[str]] = {
    "2026-11-15": ["18:00", "18:30", "19:00", "20:30"],
    "2026-11-16": ["17:00", "19:30", "20:00"],
    "2026-11-20": ["18:00", "18:30", "19:00", "19:30", "20:00", "20:30", "21:00"],
}


def check_availability(date: str, time: str) -> bool:
    """Check if a time slot is available. Returns True if the slot is open."""
    booked_times = EXISTING_RESERVATIONS.get(date, [])
    return time not in booked_times


def suggest_alternative_times(date: str, requested_time: str) -> list[str]:
    """Find up to 3 open time slots near the requested time on the same date."""
    all_slots = [
        "17:00", "17:30", "18:00", "18:30", "19:00",
        "19:30", "20:00", "20:30", "21:00", "21:30"
    ]
    booked_times = EXISTING_RESERVATIONS.get(date, [])
    available = [s for s in all_slots if s not in booked_times]

    # Sort alternatives by proximity to the requested time
    def time_diff(slot: str) -> int:
        req_h, req_m = map(int, requested_time.split(":"))
        slot_h, slot_m = map(int, slot.split(":"))
        return abs((slot_h * 60 + slot_m) - (req_h * 60 + req_m))

    return sorted(available, key=time_diff)[:3]


def execute_reservation_checker(tool_input: dict[str, Any]) -> dict[str, Any]:
    """
    Simulate reservation booking logic.
    In production, this would read from and write to your reservations database.
    """
    date_str = tool_input.get("date")
    time_str = tool_input.get("time")
    party_size = tool_input.get("party_size")
    customer_name = tool_input.get("customer_name")
    customer_phone = tool_input.get("customer_phone")
    special_requests = tool_input.get("special_requests", "")

    # Validate the date isn't a Monday (restaurant is closed)
    try:
        reservation_date = datetime.strptime(date_str, "%Y-%m-%d")
        if reservation_date.weekday() == 0:  # Monday is weekday 0
            return {
                "success": False,
                "error": "Bella Napoli is closed on Mondays. We're open Tuesday through Sunday."
            }
    except ValueError:
        return {
            "success": False,
            "error": f"Invalid date format: {date_str}. Please use YYYY-MM-DD format."
        }

    # Validate time is within business hours (17:00 to 21:30 last seating)
    try:
        hour, minute = map(int, time_str.split(":"))
        time_as_minutes = hour * 60 + minute
        if time_as_minutes < 17 * 60 or time_as_minutes > 21 * 60 + 30:
            return {
                "success": False,
                "error": "We seat guests between 5:00 PM and 9:30 PM. Please choose a time in that range."
            }
    except ValueError:
        return {
            "success": False,
            "error": f"Invalid time format: {time_str}. Please use HH:MM format."
        }

    # Check availability
    if not check_availability(date_str, time_str):
        alternatives = suggest_alternative_times(date_str, time_str)
        return {
            "success": False,
            "available": False,
            "error": f"The {time_str} slot on {date_str} is fully booked.",
            "alternative_times": alternatives,
            "message": f"We have openings at: {', '.join(alternatives)}. Would any of those work?"
        }

    # Book the reservation by adding it to our simulated database
    if date_str not in EXISTING_RESERVATIONS:
        EXISTING_RESERVATIONS[date_str] = []
    EXISTING_RESERVATIONS[date_str].append(time_str)

    confirmation_number = "RES-" + "".join(random.choices(string.digits, k=8))

    return {
        "success": True,
        "available": True,
        "confirmation_number": confirmation_number,
        "customer_name": customer_name,
        "customer_phone": customer_phone,
        "date": date_str,
        "time": time_str,
        "party_size": party_size,
        "special_requests": special_requests,
        "deposit_required": party_size > 8,
        "deposit_amount": 25.00 * party_size if party_size > 8 else 0,
        "message": "Reservation confirmed. You'll receive a text confirmation shortly."
    }


class ReservationAgent:
    """Specialist agent that handles all table reservation requests."""

    def __init__(self):
        self.client = get_client()
        self.tools = [RESERVATION_CHECKER_TOOL]

    def process(self, customer_message: str) -> str:
        """Process a reservation request and return the agent's response."""
        messages = [{"role": "user", "content": customer_message}]

        while True:
            response = self.client.messages.create(
                model=MODEL,
                max_tokens=1024,
                system=RESERVATION_AGENT_SYSTEM_PROMPT,
                tools=self.tools,
                messages=messages
            )

            if response.stop_reason == "end_turn":
                for block in response.content:
                    if hasattr(block, "text"):
                        return block.text
                return "Your reservation request has been handled."

            if response.stop_reason == "tool_use":
                messages.append({"role": "assistant", "content": response.content})

                tool_results = []
                for block in response.content:
                    if block.type == "tool_use":
                        if block.name == "reservation_checker":
                            result = execute_reservation_checker(block.input)
                        else:
                            result = {"error": f"Unknown tool: {block.name}"}

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

                messages.append({"role": "user", "content": tool_results})

            else:
                return "I had trouble processing your reservation request. Please try again."