← Back to Blog

What You'll Build

By the end of this tutorial, you'll have a working multi-agent healthcare scheduling system built in Python using the Anthropic SDK and Claude Sonnet. The system uses three specialized agents — a coordinator, a scheduler, and a patient intake verifier — that communicate with each other and query a mock database using Claude's tool use feature. This is exactly the kind of architecture we build at Naples AI for real healthcare clients who need to automate appointment booking, reduce front-desk load, and eliminate manual data entry errors.

📦 Full Source Code
The complete, working code for this project is built up section by section below. Every snippet is self-contained with its imports shown. Copy each step in order and you'll have a fully functional multi-agent pipeline by Step 6.

Prerequisites

  • Python 3.10 or higher installed
  • An Anthropic API key (get one at console.anthropic.com)
  • Basic familiarity with Python classes and functions
  • pip installed so you can add the Anthropic SDK
  • A terminal and a code editor (VS Code works great)

Step 1: Set Up Claude API and Install Dependencies

First, install the Anthropic SDK. That's the only external dependency you need for this project — everything else is in the Python standard library.

terminal
pip install anthropic

Next, set your API key as an environment variable. Never hardcode it in your source files.

terminal
export ANTHROPIC_API_KEY="sk-ant-your-key-here"

Now create your project folder and the main file you'll be working in:

terminal
mkdir healthcare_scheduler
cd healthcare_scheduler
touch main.py

Create a second file called database.py — that's where we'll put the mock patient and slot data the agents will query. We'll build that out in Step 5.

Step 2: Create the Appointment Scheduler Agent

The scheduler agent's only job is to find available time slots and confirm bookings. It doesn't know about the patient's insurance or intake status — that's the intake agent's problem. Keeping these concerns separate is what makes the system easy to debug and extend later.

scheduler_agent.py
import anthropic
import os

class SchedulerAgent:
    """Handles appointment availability and booking confirmations."""

    def __init__(self):
        self.client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
        self.model = "claude-sonnet-4-5"
        self.system_prompt = (
            "You are a healthcare appointment scheduling assistant. "
            "Your job is to find available appointment slots and confirm bookings. "
            "Always respond in a friendly, professional tone. "
            "If a slot is unavailable, suggest the next two available times. "
            "Never book an appointment without first confirming the patient's name and reason for visit."
        )

    def find_available_slots(self, date: str, provider: str, available_slots: list) -> str:
        """Ask Claude to format and present available slots to the user."""
        prompt = (
            f"A patient is looking for an appointment with Dr. {provider} on {date}. "
            f"The following slots are available: {available_slots}. "
            "Present these options clearly and ask the patient which time works best."
        )

        message = self.client.messages.create(
            model=self.model,
            max_tokens=512,
            system=self.system_prompt,
            messages=[{"role": "user", "content": prompt}]
        )

        return message.content[0].text

    def confirm_booking(self, patient_name: str, provider: str, slot: str, reason: str) -> str:
        """Generate a booking confirmation message."""
        prompt = (
            f"Confirm the following appointment booking:\n"
            f"Patient: {patient_name}\n"
            f"Provider: Dr. {provider}\n"
            f"Time slot: {slot}\n"
            f"Reason for visit: {reason}\n"
            "Write a short, friendly confirmation message the patient would receive via text or email."
        )

        message = self.client.messages.create(
            model=self.model,
            max_tokens=256,
            system=self.system_prompt,
            messages=[{"role": "user", "content": prompt}]
        )

        return message.content[0].text

Notice the system prompt is specific about what this agent does and doesn't do. That specificity is what keeps the agent on task when it gets edge-case inputs from real users.

Step 3: Create the Patient Intake Verification Agent

The intake agent checks whether a patient has completed their paperwork and whether their insurance is on file. In a real deployment, this would hit an EHR API — here we'll wire it to our mock database in Step 5. The agent returns a structured status so the coordinator knows whether to proceed with scheduling.

intake_agent.py
import anthropic
import os

class IntakeAgent:
    """Verifies patient intake status and insurance information."""

    def __init__(self):
        self.client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
        self.model = "claude-sonnet-4-5"
        self.system_prompt = (
            "You are a patient intake verification assistant for a medical office. "
            "Your job is to check whether a patient has completed intake forms and has valid insurance on file. "
            "Be concise and factual. Return a clear status: COMPLETE, INCOMPLETE, or PENDING. "
            "If something is missing, list exactly what needs to be resolved before the appointment can be booked."
        )

    def verify_patient(self, patient_data: dict) -> dict:
        """
        Verify intake completeness for a given patient.
        Returns a dict with 'status' and 'message' keys.
        """
        prompt = (
            f"Review this patient record and determine their intake status:\n"
            f"Name: {patient_data.get('name', 'Unknown')}\n"
            f"Date of birth: {patient_data.get('dob', 'Not provided')}\n"
            f"Insurance on file: {patient_data.get('insurance', 'None')}\n"
            f"Intake forms completed: {patient_data.get('forms_complete', False)}\n"
            f"Prior visit: {patient_data.get('prior_visit', 'No')}\n"
            "Based on this information, return a status (COMPLETE, INCOMPLETE, or PENDING) "
            "and a brief explanation of what is or isn't ready."
        )

        message = self.client.messages.create(
            model=self.model,
            max_tokens=256,
            system=self.system_prompt,
            messages=[{"role": "user", "content": prompt}]
        )

        response_text = message.content[0].text

        # Determine status by scanning the response text
        if "COMPLETE" in response_text and "INCOMPLETE" not in response_text:
            status = "COMPLETE"
        elif "INCOMPLETE" in response_text:
            status = "INCOMPLETE"
        else:
            status = "PENDING"

        return {"status": status, "message": response_text}

    def generate_intake_instructions(self, missing_items: list) -> str:
        """Tell the patient what they need to complete before their appointment."""
        prompt = (
            f"A patient needs to complete these items before their appointment can be confirmed:\n"
            f"{chr(10).join(f'- {item}' for item in missing_items)}\n"
            "Write a friendly but clear message explaining what they need to do and how to do it."
        )

        message = self.client.messages.create(
            model=self.model,
            max_tokens=256,
            system=self.system_prompt,
            messages=[{"role": "user", "content": prompt}]
        )

        return message.content[0].text

Step 4: Build the Coordinator Agent That Routes Requests

This is the brain of the system. The coordinator receives the initial request, decides which agent to call and in what order, and assembles the final response. It uses Claude's tool use feature so the model can call our database query functions directly during its reasoning loop.

The coordinator doesn't know the business logic of scheduling or intake — it just knows which tools exist and how to route the results. That separation is what makes this architecture scale.

coordinator_agent.py
import anthropic
import os
import json
from scheduler_agent import SchedulerAgent
from intake_agent import IntakeAgent
from database import get_patient_record, get_available_slots

class CoordinatorAgent:
    """
    Routes scheduling requests between the SchedulerAgent and IntakeAgent.
    Uses Claude tool use to query the mock database before delegating tasks.
    """

    def __init__(self):
        self.client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
        self.model = "claude-sonnet-4-5"
        self.scheduler = SchedulerAgent()
        self.intake = IntakeAgent()

        self.system_prompt = (
            "You are a healthcare scheduling coordinator. "
            "When a patient requests an appointment, you must:\n"
            "1. Look up their patient record using the get_patient_record tool.\n"
            "2. Check available slots using the get_available_slots tool.\n"
            "3. Based on the results, decide whether to proceed with booking or flag intake issues.\n"
            "Always use the tools before making any scheduling decisions. "
            "Be concise in your reasoning steps."
        )

        # Tool definitions Claude will use to call our database functions
        self.tools = [
            {
                "name": "get_patient_record",
                "description": "Retrieves a patient's record including intake status and insurance from the database.",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "patient_name": {
                            "type": "string",
                            "description": "The full name of the patient."
                        }
                    },
                    "required": ["patient_name"]
                }
            },
            {
                "name": "get_available_slots",
                "description": "Returns available appointment slots for a given provider and date.",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "provider": {
                            "type": "string",
                            "description": "The doctor's last name."
                        },
                        "date": {
                            "type": "string",
                            "description": "The requested appointment date in YYYY-MM-DD format."
                        }
                    },
                    "required": ["provider", "date"]
                }
            }
        ]

    def _process_tool_call(self, tool_name: str, tool_input: dict) -> str:
        """Execute the appropriate database function based on the tool Claude selected."""
        if tool_name == "get_patient_record":
            result = get_patient_record(tool_input["patient_name"])
            return json.dumps(result)
        elif tool_name == "get_available_slots":
            result = get_available_slots(tool_input["provider"], tool_input["date"])
            return json.dumps(result)
        else:
            return json.dumps({"error": f"Unknown tool: {tool_name}"})

    def handle_request(self, user_message: str) -> str:
        """
        Main entry point. Runs the agentic tool-use loop until Claude
        stops calling tools and returns a final text response.
        """
        messages = [{"role": "user", "content": user_message}]

        print(f"\n[Coordinator] Received request: {user_message}")

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

            print(f"[Coordinator] Stop reason: {response.stop_reason}")

            # If Claude is done reasoning and has a final answer, break out
            if response.stop_reason == "end_turn":
                final_text = next(
                    (block.text for block in response.content if hasattr(block, "text")),
                    "I was unable to process your request."
                )
                break

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

                tool_results = []
                for block in response.content:
                    if block.type == "tool_use":
                        print(f"[Coordinator] Calling tool: {block.name} with {block.input}")
                        result = self._process_tool_call(block.name, block.input)
                        print(f"[Coordinator] Tool result: {result}")

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

                # Feed all tool results back to Claude in one user turn
                messages.append({"role": "user", "content": tool_results})

            else:
                # Unexpected stop reason — bail out safely
                final_text = "An unexpected error occurred during coordination."
                break

        # Now delegate to the specialist agents based on what Claude found
        return self._delegate_to_agents(final_text, messages)

    def _delegate_to_agents(self, coordinator_summary: str, messages: list) -> str:
        """
        Parse the coordinator's findings and hand off to the right agent.
        In a production system this would be more structured (e.g., JSON output).
        """
        # Extract tool results from message history for downstream agents
        patient_data = {}
        available_slots = []
        provider = ""
        date = ""

        for msg in messages:
            if isinstance(msg["content"], list):
                for block in msg["content"]:
                    if isinstance(block, dict) and block.get("type") == "tool_result":
                        try:
                            data = json.loads(block["content"])
                            if "name" in data:
                                # This is a patient record
                                patient_data = data
                            elif "slots" in data:
                                # This is a slot list
                                available_slots = data["slots"]
                                provider = data.get("provider", "")
                                date = data.get("date", "")
                        except (json.JSONDecodeError, KeyError):
                            pass

        # Run intake verification if we have patient data
        if patient_data:
            intake_result = self.intake.verify_patient(patient_data)
            print(f"[IntakeAgent] Status: {intake_result['status']}")

            if intake_result["status"] == "INCOMPLETE":
                missing = ["Complete intake forms", "Provide insurance information"]
                instructions = self.intake.generate_intake_instructions(missing)
                return f"Intake issue detected:\n\n{instructions}"

        # If intake is clear and we have slots, confirm the booking
        if available_slots and patient_data:
            patient_name = patient_data.get("name", "Patient")
            slot = available_slots[0]  # Pick the first available slot
            reason = "General consultation"

            confirmation = self.scheduler.confirm_booking(patient_name, provider, slot, reason)
            return confirmation

        # Fall back to the coordinator's plain-language summary
        return coordinator_summary

Step 5: Implement Tool Definitions for Database Queries

This is the mock database the coordinator's tools will query. In production, you'd swap these functions out for real API calls to an EHR like Epic or a scheduling platform like Acuity. But the interface stays exactly the same — that's the beauty of this design.

database.py
from typing import Optional

# Mock patient records — in production, replace with EHR API calls
PATIENT_RECORDS = {
    "Maria Gonzalez": {
        "name": "Maria Gonzalez",
        "dob": "1985-03-14",
        "insurance": "BlueCross PPO",
        "forms_complete": True,
        "prior_visit": "2025-11-20"
    },
    "James Thornton": {
        "name": "James Thornton",
        "dob": "1972-07-08",
        "insurance": None,
        "forms_complete": False,
        "prior_visit": "No"
    },
    "Linda Park": {
        "name": "Linda Park",
        "dob": "1990-01-22",
        "insurance": "Aetna HMO",
        "forms_complete": True,
        "prior_visit": "2026-01-05"
    }
}

# Mock appointment slots — keyed by provider last name and date
AVAILABLE_SLOTS = {
    ("Martinez", "2026-09-22"): {
        "provider": "Martinez",
        "date": "2026-09-22",
        "slots": ["9:00 AM", "10:30 AM", "2:00 PM", "3:30 PM"]
    },
    ("Patel", "2026-09-22"): {
        "provider": "Patel",
        "date": "2026-09-22",
        "slots": ["8:00 AM", "11:00 AM", "1:00 PM"]
    },
    ("Martinez", "2026-09-23"): {
        "provider": "Martinez",
        "date": "2026-09-23",
        "slots": ["9:30 AM", "4:00 PM"]
    }
}

def get_patient_record(patient_name: str) -> dict:
    """
    Look up a patient record by name.
    Returns an error dict if not found so the agent can handle it gracefully.
    """
    record = PATIENT_RECORDS.get(patient_name)
    if record:
        return record
    return {"error": f"No record found for patient: {patient_name}"}

def get_available_slots(provider: str, date: str) -> dict:
    """
    Look up available appointment slots for a provider on a given date.
    Returns an error dict if no slots exist for that combination.
    """
    key = (provider, date)
    slots = AVAILABLE_SLOTS.get(key)
    if slots:
        return slots
    return {
        "error": f"No available slots for Dr. {provider} on {date}",
        "provider": provider,
        "date": date,
        "slots": []
    }

Step 6: Run the Multi-Agent Pipeline

Now we wire everything together in main.py. This script runs three test scenarios — a patient with complete intake, a patient with missing intake, and a request for an unknown patient. That way you can see the full decision tree in action.

main.py
import os
from coordinator_agent import CoordinatorAgent

def run_demo():
    """Run three scheduling scenarios to demonstrate the multi-agent pipeline."""

    coordinator = CoordinatorAgent()

    test_cases = [
        # Scenario 1: Patient with complete intake requesting an appointment
        "Hi, I'm Maria Gonzalez and I'd like to book an appointment with Dr. Martinez on September 22, 2026.",

        # Scenario 2: Patient with incomplete intake
        "I'm James Thornton. Can I schedule a visit with Dr. Patel on September 22, 2026?",

        # Scenario 3: Unknown patient not in the database
        "My name is Robert Chen and I need an appointment with Dr. Martinez on September 22, 2026."
    ]

    for i, request in enumerate(test_cases, 1):
        print("\n" + "="*60)
        print(f"SCENARIO {i}")
        print("="*60)
        print(f"Patient request: {request}")
        print("-"*60)

        result = coordinator.handle_request(request)

        print("\n[FINAL RESPONSE TO PATIENT]")
        print(result)
        print()

if __name__ == "__main__":
    # Make sure the API key is set before we do anything
    if not os.environ.get("ANTHROPIC_API_KEY"):
        raise EnvironmentError("ANTHROPIC_API_KEY environment variable is not set.")
    run_demo()

Run it with:

terminal
python main.py

Here's what the output looks like for Scenario 1 (Maria Gonzalez, complete intake):

sample output — scenario 1
============================================================
SCENARIO 1
============================================================
Patient request: Hi, I'm Maria Gonzalez and I'd like to book an appointment with Dr. Martinez on September 22, 2026.
------------------------------------------------------------

[Coordinator] Received request: Hi, I'm Maria Gonzalez...
[Coordinator] Stop reason: tool_use
[Coordinator] Calling tool: get_patient_record with {'patient_name': 'Maria Gonzalez'}
[Coordinator] Tool result: {"name": "Maria Gonzalez", "dob": "1985-03-14", "insurance": "BlueCross PPO", "forms_complete": true, "prior_visit": "2025-11-20"}
[Coordinator] Calling tool: get_available_slots with {'provider': 'Martinez', 'date': '2026-09-22'}
[Coordinator] Tool result: