How to Build a Multi-Agent Healthcare Scheduling System with Claude API

ANTHROPIC_API_KEY=your_anthropic_api_key_here

import os
import anthropic
from dotenv import load_dotenv

load_dotenv()

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

# Quick sanity check — if this prints a response, you're connected
message = client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=64,
    messages=[{"role": "user", "content": "Reply with: API connection successful."}]
)

print(message.content[0].text)

API connection successful.

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

load_dotenv()


class SchedulingAgent:
    """Handles appointment slot management using Claude API tool use."""

    def __init__(self):
        self.client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
        self.model = "claude-sonnet-4-5"
        self.conversation_history = []
        # Simulated calendar store — in production, connect to Google Calendar or similar
        self.calendar = self._initialize_calendar()

    def _initialize_calendar(self) -> dict:
        """Creates a mock calendar with available slots for the next 5 business days."""
        calendar = {}
        base = datetime(2026, 9, 15, 9, 0)  # Start from tutorial publish date
        for day_offset in range(5):
            date = base + timedelta(days=day_offset)
            date_str = date.strftime("%Y-%m-%d")
            calendar[date_str] = {
                "09:00": "available",
                "10:00": "available",
                "11:00": "booked",
                "14:00": "available",
                "15:00": "available",
                "16:00": "booked",
            }
        return calendar

    def get_tools(self) -> list[dict]:
        """Returns the tool definitions Claude uses to interact with the calendar."""
        return [
            {
                "name": "check_availability",
                "description": "Check available appointment slots for a given date.",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "date": {
                            "type": "string",
                            "description": "Date in YYYY-MM-DD format"
                        }
                    },
                    "required": ["date"]
                }
            },
            {
                "name": "book_appointment",
                "description": "Book an appointment slot for a patient.",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "date": {"type": "string", "description": "Date in YYYY-MM-DD format"},
                        "time": {"type": "string", "description": "Time in HH:MM format"},
                        "patient_name": {"type": "string", "description": "Full name of the patient"},
                        "reason": {"type": "string", "description": "Brief reason for the visit"}
                    },
                    "required": ["date", "time", "patient_name", "reason"]
                }
            },
            {
                "name": "cancel_appointment",
                "description": "Cancel an existing booked appointment.",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "date": {"type": "string"},
                        "time": {"type": "string"}
                    },
                    "required": ["date", "time"]
                }
            }
        ]

    def process_request(self, user_message: str) -> str:
        """Sends a message to Claude and handles tool use in a loop."""
        self.conversation_history.append({"role": "user", "content": user_message})

        system_prompt = """You are a healthcare scheduling assistant. Use the available tools 
        to check calendars, book appointments, and manage cancellations. Always confirm 
        the booking details clearly before finalizing."""

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

            # If Claude is done thinking and responding with text, return it
            if response.stop_reason == "end_turn":
                assistant_text = response.content[0].text
                self.conversation_history.append({"role": "assistant", "content": response.content})
                return assistant_text

            # Claude wants to use a tool — execute it and feed results back
            if response.stop_reason == "tool_use":
                self.conversation_history.append({"role": "assistant", "content": response.content})
                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)
                        })

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

    def _execute_tool(self, tool_name: str, tool_input: dict) -> Any:
        """Routes tool calls to the right calendar function."""
        if tool_name == "check_availability":
            return self._check_availability(tool_input["date"])
        elif tool_name == "book_appointment":
            return self._book_appointment(
                tool_input["date"],
                tool_input["time"],
                tool_input["patient_name"],
                tool_input["reason"]
            )
        elif tool_name == "cancel_appointment":
            return self._cancel_appointment(tool_input["date"], tool_input["time"])
        return {"error": f"Unknown tool: {tool_name}"}

    def _check_availability(self, date: str) -> dict:
        slots = self.calendar.get(date, {})
        available = [time for time, status in slots.items() if status == "available"]
        return {"date": date, "available_slots": available}

    def _book_appointment(self, date: str, time: str, patient_name: str, reason: str) -> dict:
        if date not in self.calendar:
            return {"success": False, "error": "Date not found in calendar"}
        if self.calendar[date].get(time) != "available":
            return {"success": False, "error": f"Slot {time} is not available"}
        self.calendar[date][time] = f"booked:{patient_name}"
        return {
            "success": True,
            "confirmation": f"Appointment booked for {patient_name} on {date} at {time} — Reason: {reason}"
        }

    def _cancel_appointment(self, date: str, time: str) -> dict:
        if date in self.calendar and time in self.calendar[date]:
            self.calendar[date][time] = "available"
            return {"success": True, "message": f"Slot {date} at {time} is now available"}
        return {"success": False, "error": "No appointment found at that date and time"}

from scheduling_agent import SchedulingAgent

agent = SchedulingAgent()

# Test checking availability
response = agent.process_request("What slots are available on 2026-09-15?")
print("--- Availability Check ---")
print(response)

# Test booking an appointment
response = agent.process_request(
    "Book the 9:00 AM slot on 2026-09-15 for Maria Santos. She needs a flu vaccine."
)
print("\n--- Booking Confirmation ---")
print(response)

--- Availability Check ---
On September 15, 2026, the following slots are available: 9:00 AM, 10:00 AM, 2:00 PM, and 3:00 PM.

--- Booking Confirmation ---
I've booked your appointment! Here are the details:
- Patient: Maria Santos
- Date: September 15, 2026
- Time: 9:00 AM
- Reason: Flu vaccine

Is there anything else you need?

import os
import json
import anthropic
from dotenv import load_dotenv
from typing import TypedDict

load_dotenv()


class QualificationResult(TypedDict):
    approved: bool
    urgency: str          # "routine", "priority", "emergency"
    visit_type: str       # "general", "specialist", "urgent_care"
    reason: str
    patient_name: str
    chief_complaint: str


class PatientQualificationAgent:
    """Screens patients and determines appropriate care routing before scheduling."""

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

    def get_qualification_tools(self) -> list[dict]:
        return [
            {
                "name": "assess_urgency",
                "description": "Evaluates patient symptoms and determines urgency level.",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "symptoms": {
                            "type": "array",
                            "items": {"type": "string"},
                            "description": "List of symptoms the patient reported"
                        },
                        "duration": {
                            "type": "string",
                            "description": "How long the patient has had these symptoms"
                        },
                        "severity": {
                            "type": "integer",
                            "description": "Patient-reported severity on a scale of 1 to 10"
                        }
                    },
                    "required": ["symptoms", "duration", "severity"]
                }
            },
            {
                "name": "determine_visit_type",
                "description": "Decides what type of appointment the patient needs.",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "chief_complaint": {"type": "string"},
                        "urgency_level": {"type": "string"}
                    },
                    "required": ["chief_complaint", "urgency_level"]
                }
            },
            {
                "name": "submit_qualification",
                "description": "Submits the final qualification decision for a patient.",
                "input_schema": {
                    "type": "object",
                    "properties": {
                        "patient_name": {"type": "string"},
                        "approved": {"type": "boolean"},
                        "urgency": {"type": "string"},
                        "visit_type": {"type": "string"},
                        "chief_complaint": {"type": "string"},
                        "reason": {"type": "string"}
                    },
                    "required": ["patient_name", "approved", "urgency", "visit_type", "chief_complaint", "reason"]
                }
            }
        ]

    def _execute_qualification_tool(self, tool_name: str, tool_input: dict) -> dict:
        """Runs the qualification logic — in production, these could call external APIs."""
        if tool_name == "assess_urgency":
            symptoms = tool_input.get("symptoms", [])
            severity = tool_input.get("severity", 0)
            # Flag high-risk symptoms immediately
            emergency_symptoms = ["chest pain", "difficulty breathing", "stroke symptoms", "severe bleeding"]
            is_emergency = any(s.lower() in emergency_symptoms for s in symptoms) or severity >= 9
            is_priority = severity >= 6 or len(symptoms) > 3

            if is_emergency:
                return {"urgency_level": "emergency", "recommendation": "Direct to ER immediately"}
            elif is_priority:
                return {"urgency_level": "priority", "recommendation": "Same-day or next-day appointment"}
            else:
                return {"urgency_level": "routine", "recommendation": "Standard scheduling within 1 week"}

        elif tool_name == "determine_visit_type":
            complaint = tool_input.get("chief_complaint", "").lower()
            urgency = tool_input.get("urgency_level", "routine")

            if urgency == "emergency":
                return {"visit_type": "urgent_care", "notes": "Redirect to ER or urgent care facility"}
            elif any(word in complaint for word in ["specialist", "cardio", "ortho", "neuro"]):
                return {"visit_type": "specialist", "notes": "Requires specialist referral"}
            else:
                return {"visit_type": "general", "notes": "General practitioner appointment"}

        elif tool_name == "submit_qualification":
            # Store or forward the result — here we just return it
            return {"status": "submitted", "qualification": tool_input}

        return {"error": f"Unknown tool: {tool_name}"}

    def qualify_patient(self, patient_info: str) -> QualificationResult:
        """Runs the patient through qualification and returns a structured result."""
        system_prompt = """You are a medical triage assistant. Assess the patient's information 
        using the available tools. Always call assess_urgency first, then determine_visit_type, 
        then submit_qualification. Never approve emergency patients for standard scheduling."""

        messages = [{"role": "user", "content": patient_info}]
        final_result = None

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

            if response.stop_reason == "end_turn":
                break

            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":
                        result = self._execute_qualification_tool(block.name, block.input)
                        # Capture the final qualification submission
                        if block.name == "submit_qualification":
                            q = block.input
                            final_result = QualificationResult(
                                approved=q["approved"],
                                urgency=q["urgency"],
                                visit_type=q["visit_type"],
                                reason=q["reason"],
                                patient_name=q["patient_name"],
                                chief_complaint=q["chief_complaint"]
                            )
                        tool_results.append({
                            "type": "tool_result",
                            "tool_use_id": block.id,
                            "content": json.dumps(result)
                        })

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

        # Safe fallback if submit_qualification was never called
        if final_result is None:
            final_result = QualificationResult(
                approved=False,
                urgency="unknown",
                visit_type="general",
                reason="Qualification process did not complete",
                patient_name="Unknown",
                chief_complaint="Unknown"
            )

        return final_result

import os
import logging
import anthropic
from dotenv import load_dotenv
from scheduling_agent import SchedulingAgent
from qualification_agent import PatientQualificationAgent, QualificationResult

load_dotenv()

# Set up logging so you can see what each agent is doing
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)


class HealthcareSchedulingOrchestrator:
    """
    Multi-agent orchestrator that coordinates patient qualification
    and appointment scheduling using two specialized Claude agents.
    """

    def __init__(self):
        self.qualification_agent = PatientQualificationAgent()
        self.scheduling_agent = SchedulingAgent()
        self.max_retries = 3

    def handle_patient_request(self, patient_info: str, scheduling_request: str) -> dict:
        """
        Main entry point. Takes raw patient info and a scheduling request,
        runs qualification, then schedules if approved.
        """
        logger.info("Starting patient intake process")
        result = {
            "status": "pending",
            "qualification": None,
            "scheduling": None,
            "error": None
        }

        # --- Phase 1: Qualify the patient ---
        try:
            logger.info("Running patient qualification...")
            qualification = self._run_with_retry(
                lambda: self.qualification_agent.qualify_patient(patient_info),
                step="qualification"
            )
            result["qualification"] = qualification
            logger.info(f"Qualification complete — urgency: {qualification['urgency']}, approved: {qualification['approved']}")

        except Exception as e:
            logger.error(f"Qualification failed: {e}")
            result["status"] = "failed"
            result["error"] = f"Qualification error: {str(e)}"
            return result

        # --- Phase 2: Route based on urgency ---
        if qualification["urgency"] == "emergency":
            result["status"] = "emergency_redirect"
            result["scheduling"] = {
                "message": "Patient flagged as emergency. Please call 911 or proceed to the nearest ER immediately.",
                "booked": False
            }
            logger.warning(f"Emergency patient detected: {qualification['patient_name']}")
            return result

        if not qualification["approved"]:
            result["status"] = "not_approved"
            result["scheduling"] = {
                "message": f"Patient not approved for scheduling. Reason: {qualification['reason']}",
                "booked": False
            }
            return result

        # --- Phase 3: Schedule the approved patient ---
        try:
            logger.info("Handing off to scheduling agent...")
            # Build a rich scheduling prompt from qualification data
            enriched_request = (
                f"{scheduling_request} "
                f"Patient name: {qualification['patient_name']}. "
                f"Visit type: {qualification['visit_type']}. "
                f"Reason for visit: {qualification['chief_complaint']}. "