Manual prior authorization costs clinics hours per request and blocks patient care. AI agents prior authorization pipelines cut that by programmatically pulling clinical data, evaluating medical necessity, and submitting structured requests to payers. This guide shows how to build one end to end with standard healthcare APIs and an LLM-based extraction layer.
Step 1: Map the workflow and integration surfaces
Before writing code, pin down the three systems your agent must touch:
- EHR / source of truth – Most modern EHRs expose FHIR R4 REST endpoints (
/Patient,/Condition,/MedicationRequest,/DocumentReference). If you’re dealing with legacy records, you may only have CCDA documents or PDF visit notes. - Payer interface – The CMS Prior Authorization Support (PAS) FHIR IG defines a standard
Bundlecontaining aClaimand supporting resources. Some payers still use X12 278 EDI or a proprietary JSON API. Use a sandbox endpoint for development. - Policy knowledge – Medical necessity rules live in payer-specific PDFs, LCD/NCD articles, or internal criteria sets.
Your agent is an orchestrator: it pulls from (1), reasons against (3), and posts to (2). Keep every call idempotent and traceable.
# config.py - centralize endpoints
EHR_BASE = "https://ehr.example.org/fhir/R4"
PAYER_BASE = "https://api.payer.example.com/pas" # sandbox
PAYER_TOKEN = "swap_with_oauth_flow"
Step 2: Extract structured data from clinical documents
EHR FHIR endpoints are clean, but referral letters and encounter notes are not. Use an LLM to map free text to FHIR resources. An OpenAI-compatible client works; point it at any gateway. If you need resilient model access, an endpoint like n4n.ai gives automatic fallback when a provider is rate-limited and per-token usage metering, which matters when you process thousands of notes nightly.
from openai import OpenAI
import os, json
client = OpenAI(
base_url=os.environ["LLM_BASE_URL"], # e.g. https://api.n4n.ai/v1
api_key=os.environ["LLM_KEY"],
)
SYSTEM = """You are a clinical data extractor. Output ONLY valid JSON matching
this schema: {patient: {id, name, birthDate}, conditions: [icd10],
medications: [rxnorm], encounter_date: str}."""
def extract_note(note_text: str) -> dict:
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": SYSTEM},
{"role": "user", "content": note_text}
],
temperature=0,
)
return json.loads(resp.choices[0].message.content)
Call this on each DocumentReference retrieved from the EHR. Store the result keyed by patient.id.
Handling confidence
The model may hallucinate an ICD-10 code. Reject any extraction where the model’s logprob for the code field is below a threshold, or where the code isn’t in your local ICD-10 CM vocabulary.
import requests
def validate_icd10(code: str) -> bool:
# local vocabulary or WHO API
r = requests.get(f"https://icd10api.example.org/validate/{code}")
return r.status_code == 200
Step 3: Evaluate medical necessity against payer policy
AI agents prior authorization succeed only when they attach the right evidence. Load the payer’s policy into a vector store, retrieve the relevant section, and ask the model to produce a rationale plus a binary verdict.
def check_policy(patient_data: dict, policy_id: str) -> dict:
# retrieve policy chunk
chunk = retrieve_policy_chunk(policy_id, patient_data["conditions"])
prompt = f"""Policy: {chunk}
Patient: {json.dumps(patient_data)}
Does this meet medical necessity? Reply JSON: {{verdict: 'meet'|'not_meet', reason: str}}"""
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"},
)
return json.loads(resp.choices[0].message.content)
If verdict is not_meet, route to a human reviewer instead of submitting. This prevents auto-denials that damage provider-payer relations.
Step 4: Build the Prior Authorization Submission bundle
The CMS PAS IG uses a Bundle of type collection. Below is a minimal Claim resource inside that bundle. Use the extracted data to fill fields.
{
"resourceType": "Bundle",
"type": "collection",
"entry": [
{
"resource": {
"resourceType": "Claim",
"status": "active",
"use": "preauthorization",
"patient": { "reference": "Patient/123" },
"created": "2024-05-01",
"provider": { "reference": "Organization/abc" },
"insurance": [{ "coverage": { "reference": "Coverage/999" } }],
"item": [
{
"productOrService": { "coding": [{
"system": "http://www.ama-assn.org/go/cpt", "code": "99213"
}]},
"supportingInfoSequence": [1]
}
],
"supportingInfo": [
{ "sequence": 1, "category": { "text": "diagnosis" },
"code": { "coding": [{
"system": "http://hl7.org/fhir/sid/icd-10-cm", "code": "I10"
}]}}
]
}
}
]
}
Generate this programmatically:
def build_bundle(patient_data, policy_check):
return {
"resourceType": "Bundle",
"type": "collection",
"entry": [{
"resource": {
"resourceType": "Claim",
"status": "active",
"use": "preauthorization",
"patient": {"reference": f"Patient/{patient_data['patient']['id']}"},
"created": patient_data["encounter_date"],
"item": [{
"productOrService": {"coding": [{
"system": "http://www.ama-assn.org/go/cpt",
"code": patient_data["cpt"]
}]},
"supportingInfoSequence": [1]
}],
"supportingInfo": [{
"sequence": 1,
"category": {"text": "diagnosis"},
"code": {"coding": [{
"system": "http://hl7.org/fhir/sid/icd-10-cm",
"code": patient_data["conditions"][0]
}]}
}]
}
}]
}
Step 5: Submit and poll for decision
Post the bundle to the payer’s PAS endpoint. Use OAuth2 client credentials; never embed tokens in code.
import requests
def submit_pa(bundle: dict) -> str:
r = requests.post(
f"{PAYER_BASE}/Claim",
json=bundle,
headers={"Authorization": f"Bearer {PAYER_TOKEN}"},
timeout=30,
)
r.raise_for_status()
return r.headers["Location"] # operation URL for status
def poll_status(location: str) -> dict:
for _ in range(10):
r = requests.get(location, headers={"Authorization": f"Bearer {PAYER_TOKEN}"})
data = r.json()
if data.get("outcome") in ("approved", "denied"):
return data
time.sleep(60)
raise TimeoutError("PA decision not returned in polling window")
AI agents prior authorization must tolerate async payers. Some return a task resource; poll the Task.status until completed.
Step 6: Add audit trail and human review
Healthcare requires traceability. Log every LLM call, extracted field, policy chunk, and HTTP exchange to an immutable store.
import logging
logging.basicConfig(filename="pa_agent.log", level=logging.INFO)
def log_step(step: str, payload: dict):
logging.info({"step": step, "payload": payload})
If the policy check returns not_meet or extraction confidence is low, create a Task in your internal system for a nurse to review. The agent should not auto-submit in those cases.
Verify success
Run the pipeline against a payer sandbox with a known test patient:
extract_note()returns JSON with valid ICD-10 codes (validate viavalidate_icd10).check_policy()returnsmeetfor a criterion you pre-loaded.submit_pa()returns a 201 and aLocationheader.poll_status()returnsapprovedwithin 10 minutes.- Logs show each step with timestamps and model usage.
If any step fails, the agent should surface a structured error to the EHR inbox rather than silently dropping the request. That’s the difference between a demo and a system a clinic will actually run.
Building AI agents prior authorization this way replaces fax machines with API calls and gives providers a single async queue. The hard parts are policy mapping and exception handling, not the LLM calls—spend your engineering effort there.