Building a Medical Appointment Voice Agent with OpenAI

Voice Agents in Healthcare

Healthcare organizations receive millions of phone calls daily for appointment scheduling, prescription refills, lab result inquiries, and insurance questions. Most of these calls follow predictable patterns that a well-designed voice agent can handle, freeing staff for complex clinical work.

This tutorial builds a medical appointment scheduling voice agent that handles patient verification, provider search, slot availability, and booking confirmation while respecting healthcare-specific compliance requirements.

HIPAA Compliance Considerations

Before writing any code, understand what HIPAA means for voice agents:

flowchart TD
    START["Building a Medical Appointment Voice Agent with O…"] --> A
    A["Voice Agents in Healthcare"]
    A --> B
    B["HIPAA Compliance Considerations"]
    B --> C
    C["Patient Verification Tools"]
    C --> D
    D["Agent Definition"]
    D --> E
    E["Voice Pipeline Setup"]
    E --> F
    F["FastAPI Server with Call Recording"]
    F --> G
    G["Testing the Medical Agent"]
    G --> H
    H["Key Considerations for Healthcare Voice…"]
    H --> DONE["Key Takeaways"]
    style START fill:#4f46e5,stroke:#4338ca,color:#fff
    style DONE fill:#059669,stroke:#047857,color:#fff

Protected Health Information (PHI): Any individually identifiable health information. This includes patient names, dates of birth, medical record numbers, appointment details, and diagnosis information.

Key requirements for voice agents:

• All data in transit must be encrypted (TLS/WSS which OpenAI's API already uses)
• Audio recordings containing PHI must be stored in HIPAA-compliant storage
• Access to PHI must be logged and auditable
• Business Associate Agreements (BAAs) must be in place with all vendors processing PHI
• Minimum necessary principle: only access the PHI needed for the task

# compliance.py
import logging
from datetime import datetime

class HIPAAComplianceLogger:
    """Log all PHI access events for audit purposes."""

    def __init__(self):
        self.logger = logging.getLogger("hipaa_audit")
        handler = logging.FileHandler("/var/log/hipaa_audit.log")
        handler.setFormatter(logging.Formatter(
            "%(asctime)s | %(message)s"
        ))
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)

    def log_phi_access(
        self,
        session_id: str,
        action: str,
        data_type: str,
        patient_id: str | None = None,
    ):
        self.logger.info(
            f"session={session_id} | action={action} | "
            f"data_type={data_type} | patient_id={patient_id}"
        )

    def log_phi_disclosure(
        self,
        session_id: str,
        disclosed_to: str,
        data_type: str,
        reason: str,
    ):
        self.logger.info(
            f"session={session_id} | DISCLOSURE | "
            f"to={disclosed_to} | data_type={data_type} | reason={reason}"
        )

audit = HIPAAComplianceLogger()

Patient Verification Tools

Before accessing any patient data, the agent must verify the caller's identity. A typical two-factor verification uses date of birth and one other identifier.

# tools.py
from agents import function_tool
import httpx
import os

EHR_BASE_URL = "http://ehr-api:8000"
EHR_API_KEY = os.environ["EHR_API_KEY"]
API_HEADERS = {"Authorization": f"Bearer {EHR_API_KEY}"}

@function_tool
async def verify_patient(
    date_of_birth: str,
    last_name: str,
    phone_number: str,
) -> str:
    """Verify a patient identity using date of birth, last name,
    and phone number. Must be called before accessing any patient data."""
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            f"{EHR_BASE_URL}/api/patients/verify",
            json={
                "date_of_birth": date_of_birth,
                "last_name": last_name,
                "phone_number": phone_number,
            },
            headers=API_HEADERS,
        )

    if resp.status_code == 404:
        return "Patient not found. Please verify the information and try again."
    if resp.status_code == 401:
        return "Verification failed. The information does not match our records."

    data = resp.json()
    audit.log_phi_access(
        session_id="current",
        action="verify_patient",
        data_type="identity",
        patient_id=data["patient_id"],
    )
    return (
        f"Patient verified: {data['first_name']} {data['last_name']}. "
        f"Patient ID: {data['patient_id']}. "
        f"You may now access their appointment information."
    )

@function_tool
async def search_providers(
    specialty: str,
    location: str | None = None,
) -> str:
    """Search for available healthcare providers by specialty and location."""
    async with httpx.AsyncClient() as client:
        params = {"specialty": specialty}
        if location:
            params["location"] = location
        resp = await client.get(
            f"{EHR_BASE_URL}/api/providers/search",
            params=params,
            headers=API_HEADERS,
        )
        providers = resp.json()["providers"]

    if not providers:
        return f"No {specialty} providers found. Try a different specialty or location."

    lines = []
    for p in providers[:5]:
        lines.append(
            f"Dr. {p['name']} - {p['specialty']} at {p['location']}. "
            f"Next available: {p['next_available']}"
        )
    return "Available providers:\n" + "\n".join(lines)

@function_tool
async def get_available_slots(
    provider_id: str,
    date: str,
    appointment_type: str = "general",
) -> str:
    """Get available appointment slots for a provider on a given date."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"{EHR_BASE_URL}/api/providers/{provider_id}/slots",
            params={"date": date, "type": appointment_type},
            headers=API_HEADERS,
        )
        slots = resp.json()["slots"]

    if not slots:
        return f"No available slots on {date}. Would you like to check another date?"

    slot_list = ", ".join(
        f"{s['time']} ({s['duration_min']} min)" for s in slots
    )
    return f"Available slots on {date}: {slot_list}"

@function_tool
async def book_appointment(
    patient_id: str,
    provider_id: str,
    slot_id: str,
    reason: str,
) -> str:
    """Book an appointment for a verified patient."""
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            f"{EHR_BASE_URL}/api/appointments",
            json={
                "patient_id": patient_id,
                "provider_id": provider_id,
                "slot_id": slot_id,
                "reason": reason,
            },
            headers=API_HEADERS,
        )

    if resp.status_code == 409:
        return "That slot is no longer available. Please choose another time."

    data = resp.json()
    audit.log_phi_access(
        session_id="current",
        action="book_appointment",
        data_type="appointment",
        patient_id=patient_id,
    )
    return (
        f"Appointment confirmed. Confirmation number: {data['confirmation_id']}. "
        f"Date: {data['date']} at {data['time']} with Dr. {data['provider_name']}. "
        f"Please arrive 15 minutes early."
    )

@function_tool
async def get_patient_appointments(patient_id: str) -> str:
    """Get upcoming appointments for a verified patient."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"{EHR_BASE_URL}/api/patients/{patient_id}/appointments",
            params={"upcoming": True},
            headers=API_HEADERS,
        )
        appointments = resp.json()["appointments"]

    audit.log_phi_access(
        session_id="current",
        action="get_appointments",
        data_type="appointment_list",
        patient_id=patient_id,
    )

    if not appointments:
        return "No upcoming appointments found."

    lines = []
    for appt in appointments[:5]:
        lines.append(
            f"{appt['date']} at {appt['time']} - "
            f"Dr. {appt['provider_name']} ({appt['type']}). "
            f"Confirmation: {appt['confirmation_id']}"
        )
    return "Upcoming appointments:\n" + "\n".join(lines)

@function_tool
async def cancel_appointment(
    patient_id: str,
    confirmation_id: str,
    reason: str,
) -> str:
    """Cancel an existing appointment. Requires patient verification first."""
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            f"{EHR_BASE_URL}/api/appointments/cancel",
            json={
                "patient_id": patient_id,
                "confirmation_id": confirmation_id,
                "reason": reason,
            },
            headers=API_HEADERS,
        )

    if resp.status_code == 404:
        return "Appointment not found. Please verify the confirmation number."
    if resp.status_code == 400:
        return f"Cannot cancel: {resp.json()['detail']}"

    audit.log_phi_access(
        session_id="current",
        action="cancel_appointment",
        data_type="appointment",
        patient_id=patient_id,
    )
    return "Appointment cancelled successfully. Would you like to reschedule?"

Agent Definition

The medical appointment agent has strict instructions about verification and privacy.

flowchart TD
    CENTER(("Core Concepts"))
    CENTER --> N0["All data in transit must be encrypted T…"]
    CENTER --> N1["Audio recordings containing PHI must be…"]
    CENTER --> N2["Access to PHI must be logged and audita…"]
    CENTER --> N3["Business Associate Agreements BAAs must…"]
    CENTER --> N4["Minimum necessary principle: only acces…"]
    CENTER --> N5["Audit trails: Log every PHI access with…"]
    style CENTER fill:#4f46e5,stroke:#4338ca,color:#fff

# medical_agent.py
from agents import Agent
from tools import (
    verify_patient, search_providers, get_available_slots,
    book_appointment, get_patient_appointments, cancel_appointment,
)

medical_appointment_agent = Agent(
    name="Medical Appointment Agent",
    instructions="""You are a medical appointment scheduling assistant for
City Health Medical Group. You help patients schedule, view, and cancel
appointments over the phone.

CRITICAL RULES:
1. ALWAYS verify the patient identity before accessing ANY patient data.
   Ask for their date of birth, last name, and confirm their phone number.
2. NEVER read back sensitive medical information unprompted. Only confirm
   what the patient themselves state.
3. If the patient asks about test results, diagnoses, or treatment plans,
   tell them you can only help with scheduling and they need to speak
   with their care team.
4. Keep responses concise and clear for voice conversation.
5. Always confirm appointment details before booking.
6. At the end of the call, summarize what was done.

CONVERSATION FLOW:
1. Greet the patient
2. Ask how you can help (schedule, reschedule, cancel, check appointments)
3. Verify identity (date of birth + last name + phone number)
4. Handle their request
5. Confirm and summarize
6. Ask if there is anything else

VOICE GUIDELINES:
- Speak dates as "March fourteenth" not "03/14"
- Spell out confirmation numbers letter by letter
- Pause briefly after important information to let it register""",
    tools=[
        verify_patient,
        search_providers,
        get_available_slots,
        book_appointment,
        get_patient_appointments,
        cancel_appointment,
    ],
)

Voice Pipeline Setup

# voice_pipeline.py
from agents.voice import VoicePipeline, SingleAgentVoiceWorkflow
from medical_agent import medical_appointment_agent

pipeline = VoicePipeline(
    workflow=SingleAgentVoiceWorkflow(medical_appointment_agent),
    config={
        "model": "gpt-4o-realtime",
        "voice": "nova",
        "turn_detection": {
            "type": "server_vad",
            "threshold": 0.5,
            "silence_duration_ms": 800,
        },
    },
)

FastAPI Server with Call Recording

Healthcare regulations often require call recording. We record the audio while streaming it through the pipeline.

# main.py
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from voice_pipeline import pipeline
from agents.voice import StreamedAudioInput
from compliance import audit
import wave
import io
import uuid
import asyncio
import boto3
from datetime import datetime

app = FastAPI(title="Medical Appointment Voice Agent")
s3_client = boto3.client("s3")
RECORDING_BUCKET = "hipaa-compliant-recordings"

@app.websocket("/ws/medical-voice")
async def medical_voice(websocket: WebSocket):
    await websocket.accept()
    session_id = str(uuid.uuid4())
    audio_frames: list[bytes] = []

    audit.log_phi_access(
        session_id=session_id,
        action="call_started",
        data_type="voice_session",
    )

    audio_input = StreamedAudioInput()

    async def receive_and_record():
        try:
            while True:
                data = await websocket.receive_bytes()
                audio_frames.append(data)
                audio_input.add_audio(data)
        except WebSocketDisconnect:
            audio_input.close()

    async def send_responses():
        result = await pipeline.run(audio_input)
        async for event in result.stream():
            if event.type == "voice_stream_event_audio":
                audio_frames.append(event.data)
                await websocket.send_bytes(event.data)

    try:
        await asyncio.gather(receive_and_record(), send_responses())
    except Exception:
        pass
    finally:
        # Save recording to HIPAA-compliant storage
        if audio_frames:
            recording_key = (
                f"recordings/{datetime.utcnow().strftime('%Y/%m/%d')}"
                f"/{session_id}.wav"
            )
            wav_buffer = io.BytesIO()
            with wave.open(wav_buffer, "wb") as wf:
                wf.setnchannels(1)
                wf.setsampwidth(2)
                wf.setframerate(24000)
                wf.writeframes(b"".join(audio_frames))

            s3_client.put_object(
                Bucket=RECORDING_BUCKET,
                Key=recording_key,
                Body=wav_buffer.getvalue(),
                ServerSideEncryption="aws:kms",
            )

        audit.log_phi_access(
            session_id=session_id,
            action="call_ended",
            data_type="voice_session",
        )

Testing the Medical Agent

# test_medical_agent.py
import pytest
from agents import Runner
from medical_agent import medical_appointment_agent

@pytest.mark.asyncio
async def test_agent_requires_verification():
    """The agent should ask for verification before accessing data."""
    result = await Runner.run(
        medical_appointment_agent,
        input="I want to see my upcoming appointments",
    )
    output = result.final_output.lower()
    assert any(
        phrase in output
        for phrase in ["date of birth", "verify", "last name"]
    )

@pytest.mark.asyncio
async def test_agent_refuses_medical_info():
    """The agent should not discuss test results or diagnoses."""
    result = await Runner.run(
        medical_appointment_agent,
        input="What were my blood test results from last week?",
    )
    output = result.final_output.lower()
    assert any(
        phrase in output
        for phrase in ["care team", "cannot", "scheduling", "speak with"]
    )

@pytest.mark.asyncio
async def test_appointment_booking_flow():
    """Test the full booking flow with mocked tools."""
    from unittest.mock import patch

    with patch("tools.verify_patient") as mock_verify:
        mock_verify.return_value = (
            "Patient verified: John Smith. Patient ID: P12345."
        )
        result = await Runner.run(
            medical_appointment_agent,
            input=(
                "I need to schedule an appointment. "
                "My name is Smith, date of birth January 15, 1985."
            ),
        )
        assert "verified" in result.final_output.lower() or mock_verify.called

Key Considerations for Healthcare Voice Agents

1. BAAs: Ensure you have Business Associate Agreements with OpenAI and all infrastructure providers before processing real PHI
2. Data residency: Confirm that audio and transcripts are processed in regions that comply with your data residency requirements
3. Audit trails: Log every PHI access with timestamps, session IDs, and action types
4. Minimum necessary: Only retrieve the data needed for the specific task
5. Fallback to human: Always offer a path to a human operator for complex or sensitive situations
6. Training data: Never use real patient conversations for model fine-tuning without proper de-identification

Sources:

• https://www.hhs.gov/hipaa/for-professionals/privacy/index.html
• https://platform.openai.com/docs/guides/realtime
• https://openai.github.io/openai-agents-python/