Building a Referral Coordination Agent: Specialist Matching and Appointment Facilitation

The Referral Coordination Problem

When a general dentist refers a patient to a specialist — an endodontist for a root canal, an oral surgeon for an extraction, or a periodontist for gum treatment — a complex coordination chain begins. The referring office must find an appropriate specialist, verify the specialist accepts the patient's insurance, transfer clinical records, and schedule the appointment. Each step involves phone calls, faxes, and manual tracking. Studies show that 25 to 50 percent of referrals are never completed, meaning patients fall through the cracks.

A referral coordination agent automates this entire workflow, ensuring every referral reaches its destination.

Referral Data Model

from dataclasses import dataclass, field
from datetime import date, datetime
from typing import Optional
from enum import Enum
import uuid

class ReferralStatus(Enum):
    CREATED = "created"
    SPECIALIST_MATCHED = "specialist_matched"
    APPOINTMENT_SCHEDULED = "appointment_scheduled"
    RECORDS_SENT = "records_sent"
    COMPLETED = "completed"
    PATIENT_DECLINED = "patient_declined"
    EXPIRED = "expired"

class Specialty(Enum):
    ENDODONTICS = "endodontics"
    ORAL_SURGERY = "oral_surgery"
    PERIODONTICS = "periodontics"
    ORTHODONTICS = "orthodontics"
    PROSTHODONTICS = "prosthodontics"
    PEDIATRIC = "pediatric_dentistry"
    PATHOLOGY = "oral_pathology"

@dataclass
class Referral:
    id: str = field(
        default_factory=lambda: str(uuid.uuid4())
    )
    patient_id: str = ""
    referring_provider_id: str = ""
    specialty_needed: Specialty = Specialty.ENDODONTICS
    reason: str = ""
    urgency: str = "routine"  # routine, urgent, emergency
    tooth_numbers: list[int] = field(default_factory=list)
    clinical_notes: str = ""
    matched_specialist_id: Optional[str] = None
    appointment_date: Optional[datetime] = None
    status: ReferralStatus = ReferralStatus.CREATED
    created_at: datetime = field(
        default_factory=datetime.utcnow
    )
    insurance_payer_id: Optional[str] = None

@dataclass
class Specialist:
    id: str
    name: str
    specialty: Specialty
    practice_name: str
    phone: str
    fax: str
    email: str
    address: str
    accepted_insurances: list[str]
    npi: str
    average_wait_days: int
    distance_miles: float = 0.0
    rating: float = 0.0
    accepts_emergency: bool = False

Specialist Matching Engine

The matching engine finds the best specialist based on multiple criteria: specialty, insurance acceptance, distance, availability, and patient preferences.

flowchart LR
    CALLER(["Patient or Caregiver"])
    subgraph TEL["Telephony"]
        SIP["Twilio SIP and PSTN"]
    end
    subgraph BRAIN["Healthcare AI Agent"]
        STT["Streaming STT<br/>Deepgram or Whisper"]
        NLU{"Intent and<br/>Entity Extraction"}
        TOOLS["Tool Calls"]
        TTS["Streaming TTS<br/>ElevenLabs or Rime"]
    end
    subgraph DATA["Live Data Plane"]
        CRM[("CRM and Notes")]
        CAL[("Calendar and<br/>Schedule")]
        KB[("Knowledge Base<br/>and Policies")]
    end
    subgraph OUT["Outcomes"]
        O1(["Appointment booked"])
        O2(["Prescription refill request"])
        O3(["Triage to clinician"])
    end
    CALLER --> SIP --> STT --> NLU
    NLU -->|Lookup| TOOLS
    TOOLS <--> CRM
    TOOLS <--> CAL
    TOOLS <--> KB
    NLU --> TTS --> SIP --> CALLER
    NLU -->|Resolved| O1
    NLU -->|Schedule| O2
    NLU -->|Escalate| O3
    style CALLER fill:#f1f5f9,stroke:#64748b,color:#0f172a
    style NLU fill:#4f46e5,stroke:#4338ca,color:#fff
    style O1 fill:#059669,stroke:#047857,color:#fff
    style O2 fill:#0ea5e9,stroke:#0369a1,color:#fff
    style O3 fill:#f59e0b,stroke:#d97706,color:#1f2937

from typing import Optional

class SpecialistMatcher:
    def __init__(self, db):
        self.db = db

    async def find_matches(
        self, referral: Referral,
        patient_lat: float, patient_lng: float,
        max_distance_miles: float = 25.0,
        limit: int = 5,
    ) -> list[Specialist]:
        rows = await self.db.fetch("""
            SELECT s.*,
                earth_distance(
                    ll_to_earth(s.latitude, s.longitude),
                    ll_to_earth($3, $4)
                ) / 1609.34 AS distance_miles
            FROM specialists s
            JOIN specialist_insurances si
                ON si.specialist_id = s.id
            WHERE s.specialty = $1
              AND si.payer_id = $2
              AND s.accepting_new_patients = true
              AND earth_distance(
                  ll_to_earth(s.latitude, s.longitude),
                  ll_to_earth($3, $4)
              ) / 1609.34 <= $5
            ORDER BY
                CASE WHEN $6 = 'emergency'
                     AND s.accepts_emergency
                     THEN 0 ELSE 1 END,
                s.average_wait_days ASC,
                distance_miles ASC
            LIMIT $7
        """,
            referral.specialty_needed.value,
            referral.insurance_payer_id,
            patient_lat, patient_lng,
            max_distance_miles,
            referral.urgency,
            limit,
        )

        return [
            Specialist(
                id=r["id"],
                name=r["name"],
                specialty=Specialty(r["specialty"]),
                practice_name=r["practice_name"],
                phone=r["phone"],
                fax=r["fax"],
                email=r["email"],
                address=r["address"],
                accepted_insurances=[],
                npi=r["npi"],
                average_wait_days=r["average_wait_days"],
                distance_miles=round(r["distance_miles"], 1),
                rating=r.get("rating", 0),
                accepts_emergency=r["accepts_emergency"],
            )
            for r in rows
        ]

    def rank_matches(
        self, specialists: list[Specialist],
        urgency: str,
    ) -> list[Specialist]:
        def score(s: Specialist) -> float:
            distance_score = max(0, 25 - s.distance_miles) / 25
            wait_score = max(0, 30 - s.average_wait_days) / 30
            rating_score = s.rating / 5.0

            if urgency == "emergency":
                return wait_score * 0.6 + distance_score * 0.3 + rating_score * 0.1
            elif urgency == "urgent":
                return wait_score * 0.4 + distance_score * 0.3 + rating_score * 0.3
            else:
                return distance_score * 0.3 + wait_score * 0.3 + rating_score * 0.4

        return sorted(specialists, key=score, reverse=True)

Availability Checking and Appointment Scheduling

Once a specialist is selected, the agent checks their availability and books the appointment through the specialist's scheduling system.

class ReferralScheduler:
    def __init__(self, db):
        self.db = db

    async def check_specialist_availability(
        self, specialist_id: str,
        preferred_date: date,
        search_days: int = 14,
    ) -> list[dict]:
        rows = await self.db.fetch("""
            SELECT schedule_date, start_time, end_time,
                   slot_duration_minutes
            FROM specialist_availability
            WHERE specialist_id = $1
              AND schedule_date BETWEEN $2
                  AND ($2 + $3 * INTERVAL '1 day')
              AND slots_remaining > 0
            ORDER BY schedule_date, start_time
        """, specialist_id, preferred_date, search_days)

        return [
            {
                "date": r["schedule_date"],
                "start": r["start_time"],
                "end": r["end_time"],
            }
            for r in rows
        ]

    async def book_referral_appointment(
        self, referral: Referral,
        specialist: Specialist,
        appointment_datetime: datetime,
    ) -> dict:
        await self.db.execute("""
            UPDATE referrals
            SET matched_specialist_id = $2,
                appointment_date = $3,
                status = 'appointment_scheduled'
            WHERE id = $1
        """, referral.id, specialist.id,
             appointment_datetime)

        await self.db.execute("""
            INSERT INTO referral_appointments
                (referral_id, specialist_id, patient_id,
                 appointment_time, status)
            VALUES ($1, $2, $3, $4, 'scheduled')
        """, referral.id, specialist.id,
             referral.patient_id, appointment_datetime)

        return {
            "specialist": specialist.name,
            "practice": specialist.practice_name,
            "address": specialist.address,
            "phone": specialist.phone,
            "appointment": appointment_datetime.isoformat(),
        }

Clinical Document Transfer

The agent packages and sends relevant clinical documents — x-rays, treatment notes, medical history — to the specialist's office.

class DocumentTransfer:
    def __init__(self, db, fax_client, secure_email):
        self.db = db
        self.fax = fax_client
        self.secure_email = secure_email

    async def prepare_referral_packet(
        self, referral: Referral,
    ) -> dict:
        documents = await self.db.fetch("""
            SELECT d.id, d.doc_type, d.file_path,
                   d.created_at
            FROM patient_documents d
            WHERE d.patient_id = $1
              AND (
                  d.doc_type IN (
                      'xray', 'periapical', 'panoramic',
                      'cbct'
                  )
                  OR d.created_at > CURRENT_DATE
                      - INTERVAL '90 days'
              )
            ORDER BY d.created_at DESC
        """, referral.patient_id)

        medical_history = await self.db.fetchrow("""
            SELECT allergies, medications, conditions,
                   blood_pressure, medical_alerts
            FROM patient_medical_history
            WHERE patient_id = $1
        """, referral.patient_id)

        return {
            "referral_id": referral.id,
            "clinical_notes": referral.clinical_notes,
            "reason": referral.reason,
            "tooth_numbers": referral.tooth_numbers,
            "documents": [
                {
                    "type": d["doc_type"],
                    "path": d["file_path"],
                }
                for d in documents
            ],
            "medical_history": dict(medical_history)
                if medical_history else {},
        }

    async def send_to_specialist(
        self, specialist: Specialist,
        packet: dict,
        method: str = "secure_email",
    ) -> bool:
        if method == "fax":
            pdf = await self._generate_referral_pdf(packet)
            result = await self.fax.send(
                specialist.fax, pdf
            )
        else:
            result = await self.secure_email.send(
                to=specialist.email,
                subject=(
                    f"Referral: Patient "
                    f"{packet['referral_id']}"
                ),
                attachments=packet["documents"],
                body=self._format_referral_letter(packet),
            )

        await self.db.execute("""
            UPDATE referrals
            SET status = 'records_sent'
            WHERE id = $1
        """, packet["referral_id"])

        return result.get("success", False)

Referral Completion Tracking

The agent monitors whether referred patients actually complete their specialist visit, closing the loop for the referring provider.

class ReferralTracker:
    def __init__(self, db, notification_service):
        self.db = db
        self.notify = notification_service

    async def check_completion_status(self):
        pending = await self.db.fetch("""
            SELECT r.*, p.first_name, p.phone,
                   s.name AS specialist_name
            FROM referrals r
            JOIN patients p ON p.id = r.patient_id
            JOIN specialists s
                ON s.id = r.matched_specialist_id
            WHERE r.status IN (
                'appointment_scheduled', 'records_sent'
            )
            AND r.appointment_date < CURRENT_TIMESTAMP
        """)

        for ref in pending:
            days_past = (
                datetime.utcnow() - ref["appointment_date"]
            ).days

            if days_past > 7:
                await self.notify.send_to_provider(
                    provider_id=ref["referring_provider_id"],
                    message=(
                        f"Referral for {ref['first_name']} "
                        f"to {ref['specialist_name']} may be "
                        f"incomplete. Appointment was "
                        f"{days_past} days ago."
                    ),
                )

FAQ

How does the agent handle patients who want to choose their own specialist instead of using the recommended match?

The agent presents the ranked specialist options as suggestions, not requirements. If the patient names a specific specialist, the agent looks them up in the database, verifies they accept the patient's insurance, and proceeds with that choice. If the specialist is not in the system, the agent adds their information and still handles record transfer and appointment coordination.

What happens when no specialist within range accepts the patient's insurance?

The agent expands the search radius in increments and also checks for specialists who offer sliding-scale fees or payment plans for out-of-network patients. It presents the options transparently — showing both in-network options farther away and closer out-of-network options with estimated costs — so the patient and referring provider can make an informed decision.

How does the agent get the specialist's availability if they use a different scheduling system?

The agent supports multiple integration methods. For specialists on the same practice management software, it queries availability directly. For external practices, it uses standardized APIs where available or falls back to faxing a referral request with preferred dates. The specialist's office confirms the appointment, and the agent updates the referral status automatically.

---

#ReferralManagement #SpecialistMatching #HealthcareAI #CareCoordination #Python #AgenticAI #LearnAI #AIEngineering