After-Hours Healthcare AI: Building an Emergency Triage and Callback Agent

The After-Hours Challenge

When a medical practice closes for the evening, patient calls do not stop. Answering services take messages, but they lack clinical context. On-call providers get woken up for non-urgent issues while truly urgent cases wait in a queue. An after-hours AI agent solves this by triaging urgency in real time, routing true emergencies to 911, scheduling prioritized callbacks for the on-call provider, and creating documentation that feeds directly into the next-day workflow.

Urgency Detection Engine

The first and most critical job of the after-hours agent is determining whether a caller needs immediate emergency intervention:

from dataclasses import dataclass
from enum import Enum
from typing import Optional
import re

class UrgencyLevel(Enum):
    EMERGENCY = "emergency"       # Call 911 immediately
    URGENT = "urgent"             # On-call provider callback within 30 minutes
    SEMI_URGENT = "semi_urgent"   # Callback within 2 hours
    NON_URGENT = "non_urgent"     # Next business day follow-up
    INFORMATIONAL = "info"        # Self-service answer sufficient

@dataclass
class UrgencyAssessment:
    level: UrgencyLevel
    confidence: float
    matched_keywords: list[str]
    reasoning: str
    override_to_human: bool = False

class UrgencyDetector:
    EMERGENCY_PATTERNS = [
        (r"chests+pain", "chest pain"),
        (r"can.?ts+breathe", "difficulty breathing"),
        (r"difficultys+breathing", "difficulty breathing"),
        (r"severes+bleeding", "severe bleeding"),
        (r"unconscious", "unconsciousness"),
        (r"seizure", "seizure"),
        (r"stroke", "possible stroke"),
        (r"suicid", "suicidal ideation"),
        (r"overdos", "possible overdose"),
        (r"anaphyla", "anaphylaxis"),
    ]

    URGENT_PATTERNS = [
        (r"fever.*10[2-9]|fever.*1[1-9]d", "high fever"),
        (r"vomiting.*blood", "vomiting blood"),
        (r"severes+pain", "severe pain"),
        (r"heads+injur", "head injury"),
        (r"brokens+bone|fracture", "possible fracture"),
        (r"allergics+reaction", "allergic reaction"),
    ]

    def assess(self, patient_message: str) -> UrgencyAssessment:
        message_lower = patient_message.lower()
        matched = []

        # Check emergency patterns first
        for pattern, label in self.EMERGENCY_PATTERNS:
            if re.search(pattern, message_lower):
                matched.append(label)

        if matched:
            return UrgencyAssessment(
                level=UrgencyLevel.EMERGENCY,
                confidence=0.95,
                matched_keywords=matched,
                reasoning=f"Emergency indicators detected: {', '.join(matched)}",
            )

        # Check urgent patterns
        for pattern, label in self.URGENT_PATTERNS:
            if re.search(pattern, message_lower):
                matched.append(label)

        if matched:
            return UrgencyAssessment(
                level=UrgencyLevel.URGENT,
                confidence=0.85,
                matched_keywords=matched,
                reasoning=f"Urgent indicators detected: {', '.join(matched)}",
            )

        return UrgencyAssessment(
            level=UrgencyLevel.NON_URGENT,
            confidence=0.7,
            matched_keywords=[],
            reasoning="No urgent or emergency indicators detected.",
        )

Emergency Routing

When the detector identifies an emergency, the agent does not attempt to manage it — it routes immediately:

@dataclass
class RoutingDecision:
    action: str
    message_to_patient: str
    internal_notes: str
    notify_on_call: bool = False

class EmergencyRouter:
    def route(self, assessment: UrgencyAssessment, patient_phone: str) -> RoutingDecision:
        if assessment.level == UrgencyLevel.EMERGENCY:
            return RoutingDecision(
                action="transfer_to_911",
                message_to_patient=(
                    "Based on what you have described, this may be a medical emergency. "
                    "Please call 911 or go to your nearest emergency room immediately. "
                    "I am also alerting your on-call provider."
                ),
                internal_notes=(
                    f"EMERGENCY routing triggered. Keywords: {assessment.matched_keywords}. "
                    f"Patient phone: {patient_phone}. Agent directed to call 911."
                ),
                notify_on_call=True,
            )
        if assessment.level == UrgencyLevel.URGENT:
            return RoutingDecision(
                action="priority_callback",
                message_to_patient=(
                    "I understand this is concerning. I am marking this as urgent and "
                    "your on-call provider will call you back within 30 minutes. "
                    "If your condition worsens before then, please call 911."
                ),
                internal_notes=f"Urgent callback requested. Keywords: {assessment.matched_keywords}.",
                notify_on_call=True,
            )
        return RoutingDecision(
            action="standard_callback",
            message_to_patient=(
                "Thank you for calling. I have logged your message and someone "
                "from our office will follow up with you on the next business day."
            ),
            internal_notes="Non-urgent after-hours message. Queued for next business day.",
        )

Callback Queue Management

The on-call provider needs a prioritized queue, not a flat list of messages:

from datetime import datetime
import heapq

@dataclass
class CallbackRequest:
    patient_id: str
    patient_phone: str
    urgency: UrgencyLevel
    summary: str
    created_at: datetime
    callback_by: datetime
    completed: bool = False

    def __lt__(self, other: "CallbackRequest") -> bool:
        priority = {UrgencyLevel.URGENT: 0, UrgencyLevel.SEMI_URGENT: 1, UrgencyLevel.NON_URGENT: 2}
        if priority.get(self.urgency, 3) != priority.get(other.urgency, 3):
            return priority.get(self.urgency, 3) < priority.get(other.urgency, 3)
        return self.created_at < other.created_at

class CallbackQueue:
    def __init__(self):
        self._queue: list[CallbackRequest] = []

    def add(self, request: CallbackRequest) -> None:
        heapq.heappush(self._queue, request)

    def get_next(self) -> Optional[CallbackRequest]:
        while self._queue:
            request = heapq.heappop(self._queue)
            if not request.completed:
                return request
        return None

    def get_pending_count(self) -> dict[str, int]:
        counts: dict[str, int] = {}
        for req in self._queue:
            if not req.completed:
                level = req.urgency.value
                counts[level] = counts.get(level, 0) + 1
        return counts

Interaction Documentation

Every after-hours interaction must produce a structured note for continuity of care:

@dataclass
class AfterHoursNote:
    patient_id: str
    timestamp: datetime
    urgency: UrgencyLevel
    patient_reported_symptoms: str
    agent_assessment: str
    action_taken: str
    callback_status: str
    follow_up_needed: bool

    def to_emr_note(self) -> str:
        return (
            f"AFTER-HOURS CONTACT - {self.timestamp.strftime('%Y-%m-%d %H:%M')}
"
            f"Urgency: {self.urgency.value.upper()}
"
            f"Patient Report: {self.patient_reported_symptoms}
"
            f"Assessment: {self.agent_assessment}
"
            f"Action Taken: {self.action_taken}
"
            f"Callback Status: {self.callback_status}
"
            f"Follow-up Needed: {'Yes' if self.follow_up_needed else 'No'}"
        )

FAQ

How does the after-hours agent handle repeat callers?

The agent checks the callback queue for existing requests from the same patient. If a patient calls back about the same issue, the agent escalates the urgency level rather than creating a duplicate entry. If they call about a new issue, a new callback request is created with its own priority.

What if the on-call provider does not respond to the callback notification?

The agent implements an escalation chain. If the primary on-call provider does not acknowledge the notification within a configurable window (typically 10 minutes for urgent cases), the agent automatically notifies the backup provider. If neither responds, it alerts practice administration.

Can the agent handle prescription refill requests after hours?

For non-controlled substances with existing prescriptions, the agent can log the refill request for next-business-day processing. It should never authorize or promise a refill — it simply captures the medication name, pharmacy, and patient details for the clinical staff to act on during business hours.

---

#HealthcareAI #AfterHoursCare #EmergencyTriage #CallbackScheduling #Python #AgenticAI #LearnAI #AIEngineering