) (1).png)
What You'll Build
If you've been searching for a real build AI agents tutorial — not a toy demo — this is it. You're going to build a production-ready multi-agent healthcare scheduling system in Python that handles patient intake, checks appointment availability, books slots, and sends follow-up notifications automatically.
The system uses four agents coordinated by a central orchestrator: an intake agent that collects patient info, a scheduling agent that checks and books slots, a notification agent that sends confirmations, and an orchestrator that decides which agent handles what. All of it runs on the Claude API with the Anthropic SDK.
By the end you'll have roughly 300 lines of working Python that you can actually deploy — not just paste into a notebook and forget.
📋 Full Source Code Note: The complete working code is built step by step in the sections below. Each step adds a real, runnable piece to the system. By Step 6 you'll have the full file. Copy each block in order and you'll have a working multi-agent system.
Prerequisites
- Python 3.10 or higher installed
- An Anthropic API key (get one at console.anthropic.com)
- Basic comfort with Python classes and functions
anthropicSDK installed:pip install anthropic- Optional but helpful: familiarity with tool use / function calling in LLMs
Step 1: Set Up Your Claude API Keys and Environment
First, install the Anthropic SDK and set your API key as an environment variable. Never hard-code keys in source files — it's a bad habit that will eventually bite you.
terminalpip install anthropic export ANTHROPIC_API_KEY="sk-ant-your-key-here"
Now create your project file and add the base imports and configuration. This block also defines a simple in-memory appointment store we'll use as a stand-in for a real database.
healthcare_agents.pyimport os
import json
import anthropic
from datetime import datetime, timedelta
from typing import Any
# Initialize the Anthropic client — reads ANTHROPIC_API_KEY from environment
client = anthropic.Anthropic()
MODEL = "claude-sonnet-4-6"
# Simple in-memory store simulating a database of appointments and patients
appointment_store: dict[str, Any] = {
"appointments": [],
"availability": {
"2026-06-22": ["09:00", "10:00", "11:00", "14:00", "15:00"],
"2026-06-23": ["09:00", "10:30", "13:00", "16:00"],
"2026-06-24": ["08:00", "11:00", "14:30", "15:30"],
},
"patients": []
}
Step 2: Create the Patient Intake Agent
The intake agent's job is to collect structured patient information from a conversation. It uses Claude to extract name, date of birth, reason for visit, and insurance info from natural language input.
Notice that the system prompt is tight and specific. Vague prompts produce vague agents — give it a clear role and a clear output format.
healthcare_agents.py (continued)class PatientIntakeAgent:
"""Collects and validates patient information before scheduling."""
SYSTEM_PROMPT = """You are a patient intake specialist at a medical clinic.
Your job is to collect the following information from the patient in a friendly,
professional tone:
1. Full name
2. Date of birth (MM/DD/YYYY)
3. Reason for visit
4. Insurance provider (or 'self-pay')
Once you have all four pieces of information, respond with a JSON block in this
exact format and nothing else after it:
INTAKE_COMPLETE:
{"name": "...", "dob": "...", "reason": "...", "insurance": "..."}"""
def __init__(self):
self.conversation_history: list[dict] = []
def process(self, user_message: str) -> tuple[str, dict | None]:
"""
Process a user message and return (response_text, patient_data).
patient_data is None until intake is complete.
"""
self.conversation_history.append({
"role": "user",
"content": user_message
})
response = client.messages.create(
model=MODEL,
max_tokens=1024,
system=self.SYSTEM_PROMPT,
messages=self.conversation_history
)
reply = response.content[0].text
self.conversation_history.append({
"role": "assistant",
"content": reply
})
# Check if intake is complete and extract structured data
patient_data = None
if "INTAKE_COMPLETE:" in reply:
try:
json_part = reply.split("INTAKE_COMPLETE:")[1].strip()
patient_data = json.loads(json_part)
appointment_store["patients"].append(patient_data)
except (json.JSONDecodeError, IndexError):
pass # Return None if parsing fails; orchestrator will retry
return reply, patient_data
Step 3: Create the Appointment Scheduling Agent
The scheduling agent uses Claude's tool use feature to check real availability and book appointments. This is where multi-agent systems get interesting — the agent decides when to call a tool versus when to just reply.
Tool use with the Anthropic SDK follows a loop pattern: Claude responds, you check if it wants to call a tool, you run the tool, feed the result back, and repeat until Claude gives a final text response.
healthcare_agents.py (continued)class AppointmentSchedulingAgent:
"""Checks availability and books appointments using tool calls."""
SYSTEM_PROMPT = """You are a medical appointment scheduler.
Use the available tools to check open appointment slots and book them for patients.
Always confirm the appointment details before finalizing.
Be concise and professional."""
def __init__(self):
self.tools = [
{
"name": "get_availability",
"description": "Get 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": "schedule_appointment",
"description": "Book an appointment slot for a patient.",
"input_schema": {
"type": "object",
"properties": {
"patient_name": {
"type": "string",
"description": "Full name of the patient"
},
"date": {
"type": "string",
"description": "Appointment date in YYYY-MM-DD format"
},
"time": {
"type": "string",
"description": "Appointment time in HH:MM format"
},
"reason": {
"type": "string",
"description": "Reason for the visit"
}
},
"required": ["patient_name", "date", "time", "reason"]
}
}
]
def _run_tool(self, tool_name: str, tool_input: dict) -> str:
"""Execute a tool call and return the result as a string."""
if tool_name == "get_availability":
date = tool_input["date"]
slots = appointment_store["availability"].get(date, [])
if slots:
return json.dumps({"date": date, "available_slots": slots})
return json.dumps({"date": date, "available_slots": [], "message": "No availability on this date."})
if tool_name == "schedule_appointment":
date = tool_input["date"]
time = tool_input["time"]
# Remove the booked slot from availability
if date in appointment_store["availability"]:
appointment_store["availability"][date] = [
s for s in appointment_store["availability"][date] if s != time
]
appointment = {
"id": f"APT-{len(appointment_store['appointments']) + 1001}",
"patient_name": tool_input["patient_name"],
"date": date,
"time": time,
"reason": tool_input["reason"],
"status": "confirmed"
}
appointment_store["appointments"].append(appointment)
return json.dumps({"success": True, "appointment": appointment})
return json.dumps({"error": f"Unknown tool: {tool_name}"})
def process(self, patient_data: dict, preferred_date: str) -> tuple[str, dict | None]:
"""
Run the scheduling agent loop and return (response_text, appointment).
Uses the tool-use loop: call Claude, check for tool use, run tools, repeat.
"""
messages = [{
"role": "user",
"content": (
f"Please schedule an appointment for {patient_data['name']}. "
f"Reason for visit: {patient_data['reason']}. "
f"Preferred date: {preferred_date}. "
f"Check availability and book the earliest available slot."
)
}]
booked_appointment = None
final_response = ""
# Agent run loop — keeps going until Claude stops calling tools
while True:
response = client.messages.create(
model=MODEL,
max_tokens=1024,
system=self.SYSTEM_PROMPT,
tools=self.tools,
messages=messages
)
# If Claude is done, grab the text and exit loop
if response.stop_reason == "end_turn":
for block in response.content:
if hasattr(block, "text"):
final_response = block.text
break
# If Claude wants to use a tool, run it and feed the result back
if response.stop_reason == "tool_use":
# Add Claude's response (with tool call) to message history
messages.append({"role": "assistant", "content": response.content})
tool_results = []
for block in response.content:
if block.type == "tool_use":
tool_result = self._run_tool(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": tool_result
})
# Capture booked appointment data for the orchestrator
if block.name == "schedule_appointment":
result_data = json.loads(tool_result)
if result_data.get("success"):
booked_appointment = result_data["appointment"]
# Feed tool results back to Claude
messages.append({"role": "user", "content": tool_results})
else:
# Unexpected stop reason — break to avoid infinite loop
break
return final_response, booked_appointment
⚠️ Tool Use Loop Pattern: The
while True loop with a break on end_turn is the correct pattern for Anthropic tool use. Don't skip it — if you just call client.messages.create once, you'll get a half-finished response whenever Claude decides to use a tool.
Step 4: Create the Follow-Up Notification Agent
This agent generates and "sends" appointment confirmation messages. In a real system you'd plug in Twilio or SendGrid here. For now it formats the message and logs it — easy to swap out later.
healthcare_agents.py (continued)class FollowUpNotificationAgent:
"""Generates and sends appointment confirmation notifications."""
SYSTEM_PROMPT = """You are a medical office communications specialist.
Write clear, friendly appointment confirmation messages.
Include: patient name, appointment date and time, reason for visit,
appointment ID, and clinic instructions (arrive 10 minutes early, bring insurance card).
Keep it under 150 words."""
def __init__(self):
self.sent_notifications: list[dict] = []
def _send_notification(self, patient_data: dict, appointment: dict, message: str) -> dict:
"""Simulate sending a notification — replace with real SMS/email integration."""
notification = {
"timestamp": datetime.now().isoformat(),
"patient_name": patient_data["name"],
"appointment_id": appointment["id"],
"channel": "sms", # Would be dynamic in production
"message": message,
"status": "sent"
}
self.sent_notifications.append(notification)
return notification
def process(self, patient_data: dict, appointment: dict) -> tuple[str, dict]:
"""Generate a confirmation message and send the notification."""
response = client.messages.create(
model=MODEL,
max_tokens=512,
system=self.SYSTEM_PROMPT,
messages=[{
"role": "user",
"content": (
f"Write a confirmation message for:\n"
f"Patient: {patient_data['name']}\n"
f"Appointment ID: {appointment['id']}\n"
f"Date: {appointment['date']} at {appointment['time']}\n"
f"Reason: {appointment['reason']}"
)
}]
)
message_text = response.content[0].text
notification = self._send_notification(patient_data, appointment, message_text)
return message_text, notification
Step 5: Build the Orchestrator Agent
The orchestrator is what makes this a real multi-agent system. It decides which agent to call, passes data between them, and manages the overall workflow state. Think of it as the project manager that never sleeps.
This one also uses Claude to make routing decisions — it's not just a hard-coded if/else chain. That means it can handle edge cases and unexpected input more gracefully.
healthcare_agents.py (continued)class HealthcareOrchestrator:
"""
Central orchestrator that routes between specialized agents.
Manages workflow state and coordinates intake -> scheduling -> notification.
"""
SYSTEM_PROMPT = """You are a healthcare workflow coordinator.
Your job is to determine the next step in the patient scheduling workflow.
Given the current state, respond with exactly one of these routing decisions:
- ROUTE:INTAKE — patient info is incomplete, continue intake
- ROUTE:SCHEDULE — intake is complete, proceed to scheduling
- ROUTE:NOTIFY — appointment is booked, send confirmation
- ROUTE:COMPLETE — workflow is done
Only respond with the ROUTE: command. No other text."""
def __init__(self):
self.intake_agent = PatientIntakeAgent()
self.scheduling_agent = AppointmentSchedulingAgent()
self.notification_agent = FollowUpNotificationAgent()
self.state = {
"patient_data": None,
"appointment": None,
"notification": None,
"step": "intake"
}
def _decide_route(self) -> str:
"""Ask Claude to determine the next workflow step based on current state."""
state_summary = (
f"Patient data collected: {self.state['patient_data'] is not None}\n"
f"Appointment booked: {self.state['appointment'] is not None}\n"
f"Notification sent: {self.state['notification'] is not None}"
)
response = client.messages.create(
model=MODEL,
max_tokens=64,
system=self.SYSTEM_PROMPT,
messages=[{"role": "user", "content": state_summary}]
)
return response.content[0].text.strip()
def run(self, initial_message: str, preferred_date: str = "2026-06-22") -> dict:
"""
Run the full multi-agent workflow from intake to notification.
Returns a summary of what each agent did.
"""
print("\n" + "="*60)
print("HEALTHCARE SCHEDULING ORCHESTRATOR")
print("="*60)
workflow_log = []
# --- Step 1: Patient Intake ---
print("\n[INTAKE AGENT] Starting patient intake...")
current_message = initial_message
patient_data = None
# Intake loop — keep going until we have complete patient data
for _ in range(6): # Max 6 turns to prevent infinite loops
reply, patient_data = self.intake_agent.process(current_message)
print(f"\n Intake Agent: {reply[:200]}...")
workflow_log.append({"agent": "intake", "response": reply})
if patient_data:
self.state["patient_data"] = patient_data
print(f"\n ✓ Patient data collected: {patient_data}")
break
# Simulate follow-up answers for the demo
current_message = "Maria Santos, 03/15/1985, annual checkup, BlueCross BlueShield"
# --- Step 2: Route Decision ---
route = self._decide_route()
print(f"\n[ORCHESTRATOR] Routing decision: {route}")
# --- Step 3: Appointment Scheduling ---
if "SCHEDULE" in route and self.state["patient_data"]:
print("\n[SCHEDULING AGENT] Checking availability and booking...")
sched_response, appointment = self.scheduling_agent.process(
self.state["patient_data"],
preferred_date
)
print(f"\n Scheduling Agent: {sched_response[:200]}...")
if appointment:
self.state["appointment"] = appointment
print(f"\n ✓ Appointment booked: {appointment}")
workflow_log.append({"agent": "scheduling", "response": sched_response, "appointment": appointment})
# --- Step 4: Re-route and Notify ---
route = self._decide_route()
print(f"\n[ORCHESTRATOR] Routing decision: {route}")
if "NOTIFY" in route and self.state["appointment"]:
print("\n[NOTIFICATION AGENT] Sending confirmation...")
notif_message, notification = self.notification_agent.process(
self.state["patient_data"],
self.state["appointment"]
)
self.state["notification"] = notification
print(f"\n Notification Agent: {notif_message}")
print(f"\n ✓ Notification sent: {notification['status']}")
workflow_log.append({"agent": "notification", "message": notif_message})
# --- Step 5: Final route check ---
route = self._decide_route()
print(f"\n[ORCHESTRATOR] Final status: {route}")
print("\n" + "="*60)
return {
"patient": self.state["patient_data"],
"appointment": self.state["appointment"],
"notification_status": self.state["notification"]["status"] if self.state["notification"] else None,
"workflow_log": workflow_log
}
Step 6: Implement Tool Definitions and Run the System
Now wire everything together with a main entry point. This is the block that actually runs the system — call it with a patient's opening message and a preferred date and watch all three agents coordinate in sequence.
healthcare_agents.py (continued)def main():
"""Entry point — runs the full multi-agent scheduling workflow."""
# Simulate a patient's opening message to the system
initial_patient_message = (
"Hi, I'd like to schedule an appointment. "
"My name is Maria Santos, date of birth March 15, 1985. "
"I need an annual checkup and my insurance is BlueCross BlueShield."
)
orchestrator = HealthcareOrchestrator()
result = orchestrator.run(
initial_message=initial_patient_message,
preferred_date="2026-06-22"
)
print("\n📋 FINAL WORKFLOW SUMMARY")
print("-" * 40)
print(f"Patient: {result['patient']['name']}")
print(f"Appointment ID: {result['appointment']['id']}")
print(f"Date & Time: {result['appointment']['date']} at {result['appointment']['time']}")
print(f"Notification: {result['notification_status']}")
if __name__ == "__main__":
main()
Example Conversation Output
Here's what you'll actually see in your terminal when you run this. The three agents hand off to each other cleanly, with the orchestrator making routing decisions in between.
terminal output============================================================
HEALTHCARE SCHEDULING ORCHESTRATOR
============================================================
[INTAKE AGENT] Starting patient intake...
Intake Agent: Thank you for reaching out! I have most of your information.
Let me confirm: Maria Santos, DOB 03/15/1985, annual checkup, BlueCross
BlueShield.
INTAKE_COMPLETE:
{"name": "Maria Santos", "dob": "03/15/1985", "reason": "annual checkup",
"insurance": "BlueCross BlueShield"}...
✓ Patient data collected: {'name': 'Maria Santos', 'dob': '03/15/1985',
'reason': 'annual checkup', 'insurance': 'BlueCross BlueShield'}
[ORCHESTRATOR] Routing decision: ROUTE: