Automating Critical Value SMS Routing for Lab Directors

Problem Statement

A panic-range potassium or troponin can sit unread in an EHR inbox for an hour while alert fatigue swallows the notification, and the laboratory still owns the CLIA obligation to reach a responsible clinician immediately. SMS is the most reliable deterministic side channel most laboratories already have, but a naive POST to an SMS gateway double-pages providers on retries, texts off-duty physicians when the on-call schedule rolls over, and leaves no defensible record that the message was ever sent. This page builds the routing handler that a lab director can defend to a CAP assessor: an asynchronous dispatcher that fires only against validated critical results, resolves the correct on-call recipient, guarantees single delivery under retry, and writes an immutable, hashed audit record for every attempt.

Prerequisites

• Python 3.11+ — the handler uses asyncio.TaskGroup and datetime.UTC.
• Pydantic v2 (pydantic>=2.6) for the typed ingress contract, and httpx (httpx>=0.27) as the async HTTP client for the SMS gateway.
• An upstream validation verdict. SMS routing must never execute against raw instrument output. The result must already carry a confirmed critical flag from the Clinical Result Validation & Rule Engine Architecture, having cleared the Reference Range Check Implementation, survived Delta Validation & Trend Analysis so a hemolysed or clotted spike is not paged as real, and been measured against panic limits owned by Threshold Tuning & Calibration.
• An on-call coverage source — an LDAP/Active Directory group or a scheduling API that maps the current wall-clock time to a reachable provider phone number.
• Regulatory baseline: CLIA §493.1291(g) (immediate notification of life-threatening results), CAP GEN.41350 (documented read-back), and HIPAA §164.312(b) minimum-necessary audit controls.

The routing layer is one stage of Critical Value Alert Routing; it owns delivery, deduplication, and evidence — never the clinical decision that a value is critical.

Step 1: Define the Ingress Contract as a Pydantic Model

The dispatcher accepts exactly one shape. A typed model rejects malformed payloads at the boundary instead of failing mid-transmission, and it normalises the E.164 phone number and the deterministic deduplication key before any network call. The key is derived from stable identity — accession_number, test_loinc, result_value, and the collection timestamp — never from wall-clock time, so a redelivered alert resolves to the same hash.

from __future__ import annotations

import hashlib
import re
from datetime import datetime
from pydantic import BaseModel, Field, computed_field, field_validator

E164 = re.compile(r"^\+[1-9]\d{7,14}$")


class CriticalAlert(BaseModel):
    """Validated panic-range result crossing into the routing stage."""

    accession_number: str = Field(min_length=1)
    patient_mrn: str = Field(min_length=1)
    test_loinc: str = Field(min_length=1)
    analyte: str
    result_value: float
    units: str
    critical_flag: bool
    collected_at: datetime
    recipient_phone: str | None = None  # resolved in Step 2

    @field_validator("recipient_phone")
    @classmethod
    def _check_e164(cls, v: str | None) -> str | None:
        if v is not None and not E164.match(v):
            raise ValueError(f"recipient_phone must be E.164, got {v!r}")
        return v

    @computed_field  # type: ignore[misc]
    @property
    def dedup_key(self) -> str:
        seed = f"{self.accession_number}:{self.test_loinc}:{self.result_value}:{self.collected_at.isoformat()}"
        return hashlib.sha256(seed.encode("utf-8")).hexdigest()

    @computed_field  # type: ignore[misc]
    @property
    def message_text(self) -> str:
        return (
            f"CRITICAL: {self.analyte} ({self.test_loinc}) = "
            f"{self.result_value} {self.units} on accession {self.accession_number}. "
            f"Read-back required."
        )

Constructing a CriticalAlert from an instrument payload that is missing result_value or carries a non-numeric result now raises a ValidationError at the edge, which the ingestion layer quarantines rather than transmitting.

Step 2: Resolve the On-Call Recipient Before Dispatch

The single largest source of misrouted critical values is a stale coverage matrix: an alert texted to a physician who went off shift at 07:00. Resolve the recipient against the live schedule at send time, keyed on the alert’s own timestamp, and fail closed — if coverage cannot be resolved, escalate to the laboratory director rather than guessing.

import httpx


class CoverageError(RuntimeError):
    """No reachable on-call provider could be resolved."""


async def resolve_recipient(
    alert: CriticalAlert,
    schedule_url: str,
    client: httpx.AsyncClient,
) -> str:
    """Return the E.164 number of the provider on call at the alert's timestamp."""
    resp = await client.get(
        f"{schedule_url}/on-call",
        params={"at": alert.collected_at.isoformat(), "service": alert.analyte},
        timeout=3.0,
    )
    resp.raise_for_status()
    provider = resp.json().get("phone")
    if not provider:
        raise CoverageError(f"No on-call provider for {alert.analyte} at {alert.collected_at}")
    return provider

Binding resolution to alert.collected_at rather than datetime.now() also keeps replayed or backfilled alerts routing to whoever was covering when the specimen was drawn, which is what the audit record must reflect.

Step 3: Dispatch Asynchronously with Idempotency and Backoff

The dispatcher posts to the SMS gateway under a bounded retry envelope. It sends the dedup_key as an idempotency header so a retried request never produces a second text, retries only transport-class failures (timeouts, 5xx, 429), and treats a 4xx as terminal because retrying a malformed request reproduces the same error.

import asyncio
from typing import Any


class DispatchResult(BaseModel):
    delivered: bool
    gateway_id: str | None = None
    attempts: int
    final_status: int | None = None


async def dispatch_sms(
    alert: CriticalAlert,
    gateway_url: str,
    api_key: str,
    client: httpx.AsyncClient,
    max_retries: int = 3,
) -> DispatchResult:
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Idempotency-Key": alert.dedup_key,
        "X-Request-Source": "LIMS-Validation-Pipeline",
    }
    body: dict[str, Any] = {
        "to": alert.recipient_phone,
        "message": alert.message_text,
        "metadata": {"accession": alert.accession_number, "loinc": alert.test_loinc},
    }

    last_status: int | None = None
    for attempt in range(1, max_retries + 1):
        try:
            resp = await client.post(
                f"{gateway_url}/v1/messages", headers=headers, json=body, timeout=5.0
            )
            last_status = resp.status_code
            if resp.status_code < 400:
                return DispatchResult(
                    delivered=True,
                    gateway_id=resp.json().get("id"),
                    attempts=attempt,
                    final_status=resp.status_code,
                )
            if 400 <= resp.status_code < 500 and resp.status_code != 429:
                break  # terminal client error — do not retry
        except (httpx.TimeoutException, httpx.TransportError):
            pass  # transport failure — fall through to backoff

        await asyncio.sleep(min(2 ** attempt, 10) + attempt * 0.1)  # capped backoff + jitter

    return DispatchResult(delivered=False, attempts=attempt, final_status=last_status)

Step 4: Write an Immutable, Hashed Audit Record

Every transmission attempt — successful or not — produces one append-only record. The patient MRN is hashed so the audit sink honours HIPAA minimum-necessary, while the dedup_key and gateway id make each row independently reconcilable against the carrier’s delivery receipts. This is the same immutable substrate described in implementing HIPAA-compliant audit trails in LIMS.

import json
import logging
from datetime import UTC

logger = logging.getLogger("critical_value_router")


def write_audit(alert: CriticalAlert, result: DispatchResult) -> dict[str, Any]:
    record = {
        "event": "critical_value_sms_dispatch",
        "dedup_key": alert.dedup_key,
        "mrn_hash": hashlib.sha256(alert.patient_mrn.encode()).hexdigest(),
        "loinc": alert.test_loinc,
        "recipient_phone": alert.recipient_phone,
        "delivered": result.delivered,
        "attempts": result.attempts,
        "gateway_id": result.gateway_id,
        "final_status": result.final_status,
        "logged_at": datetime.now(UTC).isoformat(),
    }
    logger.info(json.dumps(record))  # ship to an append-only / WORM sink
    return record


async def route_critical_value(
    alert: CriticalAlert, gateway_url: str, schedule_url: str, api_key: str
) -> DispatchResult:
    if not alert.critical_flag:
        raise ValueError("route_critical_value called on a non-critical result")
    async with httpx.AsyncClient() as client:
        alert.recipient_phone = await resolve_recipient(alert, schedule_url, client)
        result = await dispatch_sms(alert, gateway_url, api_key, client)
    write_audit(alert, result)
    return result

Verification & Testing

Confirm correct behaviour before wiring the handler to a live gateway:

• Deduplication is stable. Build two CriticalAlert instances from the same accession, LOINC, value, and collected_at; assert their dedup_key values are identical, then mutate result_value by 0.1 and assert the key changes.
• Retry semantics. Point dispatch_sms at a mock returning 503, 503, 200; expect delivered=True and attempts == 3. Return 400; expect delivered=False and attempts == 1 (no retry on a terminal client error).
• Fail-closed coverage. Stub the schedule endpoint to return an empty phone; assert resolve_recipient raises CoverageError and no POST to the gateway is made.
• Audit completeness. Assert write_audit emits a record on both the delivered and the exhausted-retry path, and that mrn_hash never contains the raw MRN substring.

Expected success log line, one JSON object per attempt:

{"event": "critical_value_sms_dispatch", "dedup_key": "a1b2...", "delivered": true, "attempts": 1, "final_status": 200, ...}

Compliance Note

This implementation satisfies CLIA §493.1291(g), which requires the laboratory to immediately alert the responsible clinician when a life-threatening result falls in its critical range. The hashed, append-only audit record — capturing recipient, timestamp, attempt count, and gateway delivery id for every transmission — provides the documentary evidence that CAP GEN.41350 read-back and 21 CFR Part 11.10(e) demand: a reviewable record that cannot be altered after the fact. Configure retention to the longer of your state medical-board requirement and your accreditor’s, typically 7–10 years, on WORM-compliant storage.

Troubleshooting

Related

• Critical Value Alert Routing — the routing stage’s ingress/egress contract, acknowledgement state machine, and escalation model this SMS handler plugs into.
• Delta Validation & Trend Analysis — filters spurious spikes so only genuine critical values reach this dispatcher.
• Implementing delta checks for electrolyte panels in Python — the sibling implementation whose exceeded=True output feeds a critical alert into this routing path.
• Threshold Tuning & Calibration — governs the panic boundaries that decide whether SMS routing fires at all.
• Implementing HIPAA-compliant audit trails in LIMS — the immutable, hashed audit substrate each dispatch attempt writes to.

Part of: Critical Value Alert Routing.