AI Agent for Insurance Verification: Automating Coverage and Benefits Checks

Why Insurance Verification Is a Perfect AI Agent Use Case

Medical practices spend an average of 12 minutes per patient on manual insurance verification. Multiply that by 30 patients a day and you have a full-time employee doing nothing but calling payers, navigating phone trees, and entering data. An AI agent can verify eligibility in seconds through electronic data interchange (EDI) APIs, parse complex benefit structures, and calculate patient cost estimates before the visit.

The 270/271 Eligibility Transaction

Insurance verification in the US healthcare system uses the ANSI X12 270/271 transaction set. The 270 is the eligibility inquiry, and the 271 is the response. Most modern clearinghouses expose this as a REST API:

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 dataclasses import dataclass, field
from typing import Optional
from enum import Enum
import httpx

class ServiceType(Enum):
    MEDICAL = "1"
    SURGICAL = "2"
    CONSULTATION = "3"
    DIAGNOSTIC_XRAY = "4"
    DIAGNOSTIC_LAB = "5"
    MENTAL_HEALTH = "MH"
    CHIROPRACTIC = "33"
    DENTAL = "35"
    VISION = "47"
    PRESCRIPTION = "88"

@dataclass
class BenefitDetail:
    service_type: str
    coverage_level: str  # "individual" or "family"
    in_network: bool
    copay: Optional[float] = None
    coinsurance_pct: Optional[float] = None
    deductible: Optional[float] = None
    deductible_remaining: Optional[float] = None
    out_of_pocket_max: Optional[float] = None
    oop_remaining: Optional[float] = None
    requires_prior_auth: bool = False
    referral_required: bool = False

@dataclass
class EligibilityResponse:
    is_active: bool
    subscriber_name: str
    member_id: str
    group_number: str
    plan_name: str
    plan_begin_date: str
    benefits: list[BenefitDetail] = field(default_factory=list)
    raw_response: Optional[dict] = None
    error: Optional[str] = None

Building the Verification Agent

The agent wraps the clearinghouse API and adds intelligence — it retries on transient failures, caches recent verifications, and parses complex benefit structures:

from datetime import datetime, timedelta

class InsuranceVerificationAgent:
    CACHE_TTL_HOURS = 24

    def __init__(self, clearinghouse_url: str, api_key: str):
        self._client = httpx.AsyncClient(
            base_url=clearinghouse_url,
            headers={"Authorization": f"Bearer {api_key}"},
            timeout=30.0,
        )
        self._cache: dict[str, tuple[EligibilityResponse, datetime]] = {}

    async def verify(
        self,
        payer_id: str,
        member_id: str,
        dob: str,
        service_type: ServiceType,
        provider_npi: str,
    ) -> EligibilityResponse:
        cache_key = f"{payer_id}:{member_id}:{service_type.value}"
        cached = self._cache.get(cache_key)
        if cached and (datetime.utcnow() - cached[1]) < timedelta(hours=self.CACHE_TTL_HOURS):
            return cached[0]

        payload = {
            "payer_id": payer_id,
            "member_id": member_id,
            "date_of_birth": dob,
            "service_type_code": service_type.value,
            "provider_npi": provider_npi,
            "date_of_service": datetime.utcnow().strftime("%Y-%m-%d"),
        }

        try:
            response = await self._client.post("/api/v1/eligibility", json=payload)
            response.raise_for_status()
            result = self._parse_response(response.json())
            self._cache[cache_key] = (result, datetime.utcnow())
            return result
        except httpx.HTTPStatusError as e:
            return EligibilityResponse(
                is_active=False,
                subscriber_name="",
                member_id=member_id,
                group_number="",
                plan_name="",
                plan_begin_date="",
                error=f"HTTP {e.response.status_code}: {e.response.text[:200]}",
            )

    def _parse_response(self, data: dict) -> EligibilityResponse:
        benefits = []
        for b in data.get("benefits", []):
            benefits.append(BenefitDetail(
                service_type=b["service_type"],
                coverage_level=b.get("coverage_level", "individual"),
                in_network=b.get("in_network", True),
                copay=b.get("copay"),
                coinsurance_pct=b.get("coinsurance_pct"),
                deductible=b.get("deductible"),
                deductible_remaining=b.get("deductible_remaining"),
                out_of_pocket_max=b.get("out_of_pocket_max"),
                oop_remaining=b.get("oop_remaining"),
                requires_prior_auth=b.get("prior_auth_required", False),
                referral_required=b.get("referral_required", False),
            ))
        return EligibilityResponse(
            is_active=data["active"],
            subscriber_name=data["subscriber_name"],
            member_id=data["member_id"],
            group_number=data["group_number"],
            plan_name=data["plan_name"],
            plan_begin_date=data.get("plan_begin_date", ""),
            benefits=benefits,
            raw_response=data,
        )

Patient Cost Estimation

With benefit data in hand, the agent can estimate what the patient will owe:

@dataclass
class CostEstimate:
    estimated_total: float
    insurance_pays: float
    patient_pays: float
    breakdown: dict[str, float]
    notes: list[str]

class CostEstimator:
    def estimate(
        self, procedure_cost: float, benefit: BenefitDetail
    ) -> CostEstimate:
        notes = []
        patient_pays = 0.0

        # Apply deductible first
        deductible_applies = 0.0
        if benefit.deductible_remaining and benefit.deductible_remaining > 0:
            deductible_applies = min(procedure_cost, benefit.deductible_remaining)
            patient_pays += deductible_applies
            notes.append(f"Deductible applies: ${deductible_applies:.2f}")

        remaining = procedure_cost - deductible_applies

        # Apply copay or coinsurance
        if benefit.copay is not None:
            patient_pays += benefit.copay
            notes.append(f"Copay: ${benefit.copay:.2f}")
        elif benefit.coinsurance_pct is not None:
            coinsurance = remaining * (benefit.coinsurance_pct / 100)
            patient_pays += coinsurance
            notes.append(f"Coinsurance ({benefit.coinsurance_pct}%): ${coinsurance:.2f}")

        # Cap at out-of-pocket max
        if benefit.oop_remaining is not None:
            patient_pays = min(patient_pays, benefit.oop_remaining)

        insurance_pays = procedure_cost - patient_pays

        return CostEstimate(
            estimated_total=procedure_cost,
            insurance_pays=insurance_pays,
            patient_pays=patient_pays,
            breakdown={
                "deductible": deductible_applies,
                "copay": benefit.copay or 0,
                "coinsurance": patient_pays - deductible_applies - (benefit.copay or 0),
            },
            notes=notes,
        )

Prior Authorization Workflow

When a benefit check reveals prior authorization is required, the agent initiates the request:

class PriorAuthAgent:
    async def check_and_initiate(
        self, benefit: BenefitDetail, procedure_code: str, clinical_notes: str
    ) -> dict:
        if not benefit.requires_prior_auth:
            return {"required": False}

        auth_request = {
            "procedure_code": procedure_code,
            "service_type": benefit.service_type,
            "clinical_justification": clinical_notes,
            "status": "submitted",
        }
        # In production, this would submit to the payer's prior auth portal
        return {"required": True, "request": auth_request, "estimated_turnaround": "2-5 business days"}

FAQ

How does the agent handle patients with multiple insurance plans?

The agent verifies each plan independently, determines coordination of benefits order (primary vs. secondary), and calculates cost estimates by applying the primary insurance first, then running the remaining balance through the secondary plan. The order is determined by standard COB rules, which the agent encodes.

What happens when the clearinghouse returns incomplete benefit data?

Common with smaller payers. The agent flags which fields are missing, provides partial estimates with clear disclaimers, and queues the case for manual verification by billing staff. It never presents incomplete data as a definitive cost estimate.

How often should eligibility be re-verified?

Best practice is to verify at scheduling, again 2 to 3 days before the appointment, and once more at check-in. Insurance status can change at any time due to job changes, plan cancellations, or retroactive terminations. The agent's caching layer handles this by using a 24-hour TTL.

---

#HealthcareAI #InsuranceVerification #PriorAuthorization #RevenueCycle #Python #AgenticAI #LearnAI #AIEngineering