How to Build a Multi-Agent Scheduling Assistant with Claude API (Step-by-Step)

import os
import anthropic

# Pull the API key from your environment — never hardcode this
api_key = os.environ.get("ANTHROPIC_API_KEY")

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

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

# Quick connectivity test
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)

API connected

import os
import json
import anthropic
from datetime import datetime, timedelta
from typing import Any

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

class SchedulingAgent:
    """
    A specialized agent responsible for calendar operations.
    It processes tool calls from Claude and executes them against
    our simulated calendar backend.
    """

    def __init__(self):
        self.model = "claude-sonnet-4-6"
        self.conversation_history = []
        self.system_prompt = (
            "You are a scheduling assistant. Your job is to check calendar availability, "
            "book appointments, and send confirmations. Always use the tools provided to "
            "complete scheduling tasks. Be concise and confirm actions clearly."
        )
        # Simulated in-memory calendar — in production this would be Google Calendar, etc.
        self.calendar = {}

    def run(self, user_message: str) -> str:
        """Process a scheduling request and return the result."""
        self.conversation_history.append({
            "role": "user",
            "content": user_message
        })

        # Initial Claude call with tools available
        response = client.messages.create(
            model=self.model,
            max_tokens=1024,
            system=self.system_prompt,
            tools=self._get_tools(),
            messages=self.conversation_history
        )

        # Run the agentic loop until Claude stops calling tools
        while response.stop_reason == "tool_use":
            tool_results = []

            for block in response.content:
                if block.type == "tool_use":
                    result = self._execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": json.dumps(result)
                    })

            # Append assistant message and tool results to history
            self.conversation_history.append({
                "role": "assistant",
                "content": response.content
            })
            self.conversation_history.append({
                "role": "user",
                "content": tool_results
            })

            # Call Claude again with the tool results
            response = client.messages.create(
                model=self.model,
                max_tokens=1024,
                system=self.system_prompt,
                tools=self._get_tools(),
                messages=self.conversation_history
            )

        # Extract the final text response
        final_text = ""
        for block in response.content:
            if hasattr(block, "text"):
                final_text += block.text

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

        return final_text

    def _execute_tool(self, tool_name: str, tool_input: dict) -> Any:
        """Route tool calls to the correct handler method."""
        if tool_name == "check_availability":
            return self._check_availability(
                tool_input["date"],
                tool_input["time_slot"]
            )
        elif tool_name == "book_appointment":
            return self._book_appointment(
                tool_input["date"],
                tool_input["time_slot"],
                tool_input["attendee_name"],
                tool_input["meeting_purpose"]
            )
        elif tool_name == "send_confirmation":
            return self._send_confirmation(
                tool_input["attendee_name"],
                tool_input["attendee_email"],
                tool_input["date"],
                tool_input["time_slot"],
                tool_input["meeting_purpose"]
            )
        else:
            return {"error": f"Unknown tool: {tool_name}"}

    def _check_availability(self, date: str, time_slot: str) -> dict:
        """Check if a time slot is open on the calendar."""
        slot_key = f"{date}_{time_slot}"
        is_available = slot_key not in self.calendar
        return {
            "available": is_available,
            "date": date,
            "time_slot": time_slot,
            "message": f"{'Available' if is_available else 'Already booked'}: {date} at {time_slot}"
        }

    def _book_appointment(
        self,
        date: str,
        time_slot: str,
        attendee_name: str,
        meeting_purpose: str
    ) -> dict:
        """Book an appointment if the slot is open."""
        slot_key = f"{date}_{time_slot}"
        if slot_key in self.calendar:
            return {
                "success": False,
                "message": f"Slot {date} at {time_slot} is already booked."
            }

        # Write the booking to our in-memory calendar
        self.calendar[slot_key] = {
            "attendee": attendee_name,
            "purpose": meeting_purpose,
            "booked_at": datetime.now().isoformat()
        }
        return {
            "success": True,
            "booking_id": f"BK-{abs(hash(slot_key)) % 100000:05d}",
            "date": date,
            "time_slot": time_slot,
            "attendee": attendee_name,
            "message": f"Appointment booked for {attendee_name} on {date} at {time_slot}."
        }

    def _send_confirmation(
        self,
        attendee_name: str,
        attendee_email: str,
        date: str,
        time_slot: str,
        meeting_purpose: str
    ) -> dict:
        """Simulate sending a confirmation email."""
        # In production, this calls SendGrid, Mailgun, etc.
        confirmation_number = f"CONF-{abs(hash(attendee_email + date)) % 99999:05d}"
        print(f"\n[EMAIL SENT] To: {attendee_email}")
        print(f"  Subject: Your appointment on {date} at {time_slot} is confirmed")
        print(f"  Confirmation #: {confirmation_number}\n")
        return {
            "sent": True,
            "confirmation_number": confirmation_number,
            "recipient": attendee_email,
            "message": f"Confirmation email sent to {attendee_email}."
        }

    def _get_tools(self) -> list:
        """Return the tool definitions for this agent."""
        return _get_scheduling_tools()

def _get_scheduling_tools() -> list:
    """
    Define the tools available to the scheduling agent.
    These follow the Anthropic tool use specification exactly.
    """
    return [
        {
            "name": "check_availability",
            "description": (
                "Check whether a specific date and time slot is available on the calendar. "
                "Always call this before attempting to book an appointment."
            ),
            "input_schema": {
                "type": "object",
                "properties": {
                    "date": {
                        "type": "string",
                        "description": "The date to check in YYYY-MM-DD format (e.g., 2026-09-15)"
                    },
                    "time_slot": {
                        "type": "string",
                        "description": "The time slot to check (e.g., '2:00 PM', '10:30 AM')"
                    }
                },
                "required": ["date", "time_slot"]
            }
        },
        {
            "name": "book_appointment",
            "description": (
                "Book an appointment on the calendar after confirming availability. "
                "Returns a booking ID on success."
            ),
            "input_schema": {
                "type": "object",
                "properties": {
                    "date": {
                        "type": "string",
                        "description": "The appointment date in YYYY-MM-DD format"
                    },
                    "time_slot": {
                        "type": "string",
                        "description": "The time slot for the appointment (e.g., '2:00 PM')"
                    },
                    "attendee_name": {
                        "type": "string",
                        "description": "Full name of the person booking the appointment"
                    },
                    "meeting_purpose": {
                        "type": "string",
                        "description": "Brief description of what the meeting is about"
                    }
                },
                "required": ["date", "time_slot", "attendee_name", "meeting_purpose"]
            }
        },
        {
            "name": "send_confirmation",
            "description": (
                "Send a confirmation email to the attendee after successfully booking. "
                "Always call this after a successful book_appointment."
            ),
            "input_schema": {
                "type": "object",
                "properties": {
                    "attendee_name": {
                        "type": "string",
                        "description": "Full name of the attendee"
                    },
                    "attendee_email": {
                        "type": "string",
                        "description": "Email address to send the confirmation to"
                    },
                    "date": {
                        "type": "string",
                        "description": "The booked date in YYYY-MM-DD format"
                    },
                    "time_slot": {
                        "type": "string",
                        "description": "The booked time slot"
                    },
                    "meeting_purpose": {
                        "type": "string",
                        "description": "What the meeting is about"
                    }
                },
                "required": [
                    "attendee_name",
                    "attendee_email",
                    "date",
                    "time_slot",
                    "meeting_purpose"
                ]
            }
        }
    ]

import os
import json
import anthropic

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

class OrchestratorAgent:
    """
    The top-level agent that interprets user intent and delegates
    scheduling tasks to the SchedulingAgent worker.
    """

    def __init__(self, scheduling_agent):
        self.model = "claude-sonnet-4-6"
        # Hold a reference to the worker agent
        self.scheduling_agent = scheduling_agent
        self.system_prompt = (
            "You are a scheduling orchestrator. Your job is to understand what the user "
            "wants to schedule, extract all necessary details, and produce a clear, structured "
            "task description for the scheduling agent. "
            "If any required information is missing (date, time, name, email, or purpose), "
            "ask the user for it before proceeding. "
            "When you have everything you need, respond with a JSON object in this exact format:\n"
            "{\n"
            '  "action": "schedule",\n'
            '  "date": "YYYY-MM-DD",\n'
            '  "time_slot": "H:MM AM/PM",\n'
            '  "attendee_name": "Full Name",\n'
            '  "attendee_email": "[email protected]",\n'
            '  "meeting_purpose": "Brief description"\n'
            "}\n"
            "If the user's request is unclear or off-topic, respond normally in plain English."
        )
        self.conversation_history = []

    def process(self, user_input: str) -> str:
        """
        Take a raw user message, interpret it, and either ask
        a clarifying question or delegate to the SchedulingAgent.
        """
        self.conversation_history.append({
            "role": "user",
            "content": user_input
        })

        response = client.messages.create(
            model=self.model,
            max_tokens=512,
            system=self.system_prompt,
            messages=self.conversation_history
        )

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

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

        # Try to detect whether the orchestrator produced a scheduling JSON
        parsed = self._try_parse_scheduling_json(orchestrator_reply)

        if parsed:
            # Delegate to the scheduling agent with a structured task
            task = (
                f"Please complete the following scheduling task:\n"
                f"- Check availability for {parsed['date']} at {parsed['time_slot']}\n"
                f"- If available, book an appointment for {parsed['attendee_name']} "
                f"for: {parsed['meeting_purpose']}\n"
                f"- Then send a confirmation to {parsed['attendee_email']}"
            )
            result = self.scheduling_agent.run(task)
            return result
        else:
            # Orchestrator is asking a follow-up question
            return orchestrator_reply

    def _try_parse_scheduling_json(self, text: str) -> dict | None:
        """
        Attempt to extract a JSON scheduling object from the orchestrator's reply.
        Returns None if the text isn't a scheduling instruction.
        """
        try:
            # Handle cases where Claude wraps JSON in markdown code fences
            if "```" in text:
                start = text.find("{")
                end = text.rfind("}") + 1
                if start != -1 and end > start:
                    text = text[start:end]

            data = json.loads(text)
            # Validate the required fields are present
            required = [
                "action", "date", "time_slot",
                "attendee_name", "attendee_email", "meeting_purpose"
            ]
            if all(k in data for k in required) and data.get("action") == "schedule":
                return data
        except (json.JSONDecodeError, ValueError):
            pass

        return None

import os
import anthropic
from scheduling_agent import SchedulingAgent
from orchestrator import OrchestratorAgent

def run_scheduler():
    """
    Main entry point. Initializes both agents and runs
    an interactive scheduling session in the terminal.
    """
    print("=" * 55)
    print("  Naples AI — Multi-Agent Scheduling Assistant")
    print("  Powered by Claude claude-sonnet-4-6")
    print("=" * 55)
    print("Type your scheduling request below.")
    print("Type 'quit' or 'exit' to stop.\n")

    # Instantiate agents — SchedulingAgent is injected into Orchestrator
    scheduling_agent = SchedulingAgent()
    orchestrator = OrchestratorAgent(scheduling_agent)

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

            if not user_input:
                continue

            if user_input.lower() in ["quit", "exit", "q"]:
                print("\nScheduler session ended. Goodbye!")
                break

            print("\nAssistant: ", end="", flush=True)

            try:
                response = orchestrator.process(user_input)
                print(response)

            except anthropic.APIStatusError as e:
                # Handles 4xx/5xx errors from the Anthropic API
                print(f"\n[API Error {e.status_code}]: {e.message}")
                print("Please try again or check your API key.")

            except anthropic.RateLimitError:
                print("\n[Rate Limit]: Too many requests. Wait a moment and try again.")

            except anthropic.APIConnectionError:
                print("\n[Connection Error]: Can't reach Anthropic API. Check your internet.")

            except Exception as e:
                # Catch-all for unexpected errors — log in production
                print(f"\n[Unexpected Error]: {type(e).__name__}: {e}")

            print()  # Blank line between exchanges

        except KeyboardInterrupt:
            print("\n\nInterrupted. Goodbye!")
            break


if __name__ == "__main__":
    run_scheduler()

You: Book a meeting for Maria Gonzalez on September 15 2026 at 2pm.
     Her email is [email protected]. It's for a project kickoff.