{"id":271,"date":"2026-01-16T14:42:52","date_gmt":"2026-01-16T06:42:52","guid":{"rendered":"https:\/\/connectword.dpdns.org\/?p=271"},"modified":"2026-01-16T14:42:52","modified_gmt":"2026-01-16T06:42:52","slug":"how-to-build-a-safe-autonomous-prior-authorization-agent-for-healthcare-revenue-cycle-management-with-human-in-the-loop-controls","status":"publish","type":"post","link":"https:\/\/connectword.dpdns.org\/?p=271","title":{"rendered":"How to Build a Safe, Autonomous Prior Authorization Agent for Healthcare Revenue Cycle Management with Human-in-the-Loop Controls"},"content":{"rendered":"

In this tutorial, we demonstrate how an autonomous, agentic AI system can simulate the end-to-end prior authorization workflow within healthcare Revenue Cycle Management (RCM). We show how an agent continuously monitors incoming surgery orders, gathers the required clinical documentation, submits prior authorization requests to payer systems, tracks their status, and intelligently responds to denials through automated analysis and appeals. We design the system to act conservatively and responsibly, escalating to a human reviewer when uncertainty crosses a defined threshold. While the implementation uses mocked EHR and payer portals for clarity and safety, we intentionally mirror real-world healthcare workflows to make the logic transferable to production environments. Also, we emphasize that it is strictly a technical simulation and not a substitute for clinical judgment, payer policy interpretation, or regulatory compliance. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n

\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
!pip -q install \"pydantic>=2.0.0\" \"httpx>=0.27.0\"\n\n\nimport os, time, json, random, hashlib\nfrom typing import List, Dict, Optional, Any\nfrom enum import Enum\nfrom datetime import datetime, timedelta\nfrom pydantic import BaseModel, Field<\/code><\/pre>\n<\/div>\n<\/div>\n
We set up the execution environment and installed the minimal dependencies required to run the tutorial. We configure optional OpenAI usage in a safe, fail-open manner so the system continues to work even without external models. We ensure the foundation is lightweight, reproducible, and suitable for healthcare simulations. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
USE_OPENAI = False\nOPENAI_AVAILABLE = False\n\n\ntry:\n   from getpass import getpass\n   if not os.environ.get(\"OPENAI_API_KEY\"):\n       pass\n   if os.environ.get(\"OPENAI_API_KEY\"):\n       USE_OPENAI = True\nexcept Exception:\n   USE_OPENAI = False\n\n\nif USE_OPENAI:\n   try:\n       !pip -q install openai\n       from openai import OpenAI\n       client = OpenAI()\n       OPENAI_AVAILABLE = True\n   except Exception:\n       OPENAI_AVAILABLE = False\n       USE_OPENAI = False<\/code><\/pre>\n<\/div>\n<\/div>\n
We define strongly typed domain models for patients, surgical orders, clinical documents, and authorization decisions. We use explicit enums and schemas to mirror real healthcare RCM structures while avoiding ambiguity. We enforce clarity and validation to reduce downstream errors in automated decision-making. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
class DocType(str, Enum):\n   H_AND_P = \"history_and_physical\"\n   LABS = \"labs\"\n   IMAGING = \"imaging\"\n   MED_LIST = \"medication_list\"\n   CONSENT = \"consent\"\n   PRIOR_TX = \"prior_treatments\"\n   CLINICAL_NOTE = \"clinical_note\"\n\n\nclass SurgeryType(str, Enum):\n   KNEE_ARTHROPLASTY = \"knee_arthroplasty\"\n   SPINE_FUSION = \"spine_fusion\"\n   CATARACT = \"cataract\"\n   BARIATRIC = \"bariatric_surgery\"\n\n\nclass InsurancePlan(str, Enum):\n   PAYER_ALPHA = \"PayerAlpha\"\n   PAYER_BETA = \"PayerBeta\"\n   PAYER_GAMMA = \"PayerGamma\"\n\n\nclass Patient(BaseModel):\n   patient_id: str\n   name: str\n   dob: str\n   member_id: str\n   plan: InsurancePlan\n\n\nclass SurgeryOrder(BaseModel):\n   order_id: str\n   patient: Patient\n   surgery_type: SurgeryType\n   scheduled_date: str\n   ordering_provider_npi: str\n   diagnosis_codes: List[str] = Field(default_factory=list)\n   created_at: str\n\n\nclass ClinicalDocument(BaseModel):\n   doc_id: str\n   doc_type: DocType\n   created_at: str\n   content: str\n   source: str\n\n\nclass PriorAuthRequest(BaseModel):\n   request_id: str\n   order: SurgeryOrder\n   submitted_at: Optional[str] = None\n   docs_attached: List[ClinicalDocument] = Field(default_factory=list)\n   payload: Dict[str, Any] = Field(default_factory=dict)\n\n\nclass AuthStatus(str, Enum):\n   DRAFT = \"draft\"\n   SUBMITTED = \"submitted\"\n   IN_REVIEW = \"in_review\"\n   APPROVED = \"approved\"\n   DENIED = \"denied\"\n   NEEDS_INFO = \"needs_info\"\n   APPEALED = \"appealed\"\n\n\nclass DenialReason(str, Enum):\n   MISSING_DOCS = \"missing_docs\"\n   MEDICAL_NECESSITY = \"medical_necessity\"\n   MEMBER_INELIGIBLE = \"member_ineligible\"\n   DUPLICATE = \"duplicate\"\n   CODING_ISSUE = \"coding_issue\"\n   OTHER = \"other\"\n\n\nclass PayerResponse(BaseModel):\n   status: AuthStatus\n   payer_ref: str\n   message: str\n   denial_reason: Optional[DenialReason] = None\n   missing_docs: List[DocType] = Field(default_factory=list)\n   confidence: float = 0.9\n\n\nclass AgentDecision(BaseModel):\n   action: str\n   missing_docs: List[DocType] = Field(default_factory=list)\n   rationale: str = \"\"\n   uncertainty: float = 0.0\n   next_wait_seconds: int = 0\n   appeal_text: Optional[str] = None<\/code><\/pre>\n<\/div>\n<\/div>\n
We simulate an EHR system that emits surgery orders and stores clinical documentation. We intentionally model incomplete charts to reflect real-world documentation gaps that often drive prior authorization denials. We show how an agent can retrieve and augment patient records in a controlled manner. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
def _now_iso() -> str:\n   return datetime.utcnow().replace(microsecond=0).isoformat() + \"Z\"\n\n\ndef _stable_id(prefix: str, seed: str) -> str:\n   h = hashlib.sha256(seed.encode(\"utf-8\")).hexdigest()[:10]\n   return f\"{prefix}_{h}\"\n\n\nclass MockEHR:\n   def __init__(self):\n       self.orders_queue: List[SurgeryOrder] = []\n       self.patient_docs: Dict[str, List[ClinicalDocument]] = {}\n\n\n   def seed_data(self, n_orders: int = 5):\n       random.seed(7)\n\n\n       def make_patient(i: int) -> Patient:\n           pid = f\"PT{i:04d}\"\n           plan = random.choice(list(InsurancePlan))\n           return Patient(\n               patient_id=pid,\n               name=f\"Patient {i}\",\n               dob=\"1980-01-01\",\n               member_id=f\"M{i:08d}\",\n               plan=plan,\n           )\n\n\n       def docs_for_order(patient: Patient, surgery: SurgeryType) -> List[ClinicalDocument]:\n           base = [\n               ClinicalDocument(\n                   doc_id=_stable_id(\"DOC\", patient.patient_id + \"H&P\"),\n                   doc_type=DocType.H_AND_P,\n                   created_at=_now_iso(),\n                   content=\"H&P: Relevant history, exam findings, and surgical indication.\",\n                   source=\"EHR\",\n               ),\n               ClinicalDocument(\n                   doc_id=_stable_id(\"DOC\", patient.patient_id + \"NOTE\"),\n                   doc_type=DocType.CLINICAL_NOTE,\n                   created_at=_now_iso(),\n                   content=\"Clinical note: Symptoms, conservative management attempted, clinician assessment.\",\n                   source=\"EHR\",\n               ),\n               ClinicalDocument(\n                   doc_id=_stable_id(\"DOC\", patient.patient_id + \"MEDS\"),\n                   doc_type=DocType.MED_LIST,\n                   created_at=_now_iso(),\n                   content=\"Medication list: Current meds, allergies, contraindications.\",\n                   source=\"EHR\",\n               ),\n           ]\n\n\n           maybe = []\n           if surgery in [SurgeryType.KNEE_ARTHROPLASTY, SurgeryType.SPINE_FUSION, SurgeryType.BARIATRIC]:\n               maybe.append(\n                   ClinicalDocument(\n                       doc_id=_stable_id(\"DOC\", patient.patient_id + \"LABS\"),\n                       doc_type=DocType.LABS,\n                       created_at=_now_iso(),\n                       content=\"Labs: CBC\/CMP within last 30 days.\",\n                       source=\"LabSystem\",\n                   )\n               )\n\n\n           if surgery in [SurgeryType.SPINE_FUSION, SurgeryType.KNEE_ARTHROPLASTY]:\n               maybe.append(\n                   ClinicalDocument(\n                       doc_id=_stable_id(\"DOC\", patient.patient_id + \"IMG\"),\n                       doc_type=DocType.IMAGING,\n                       created_at=_now_iso(),\n                       content=\"Imaging: MRI\/X-ray report supporting diagnosis and severity.\",\n                       source=\"Radiology\",\n                   )\n               )\n\n\n           final = base + [d for d in maybe if random.random() > 0.35]\n\n\n           if random.random() > 0.6:\n               final.append(\n                   ClinicalDocument(\n                       doc_id=_stable_id(\"DOC\", patient.patient_id + \"PRIOR_TX\"),\n                       doc_type=DocType.PRIOR_TX,\n                       created_at=_now_iso(),\n                       content=\"Prior treatments: PT, meds, injections tried over 6+ weeks.\",\n                       source=\"EHR\",\n                   )\n               )\n\n\n           if random.random() > 0.5:\n               final.append(\n                   ClinicalDocument(\n                       doc_id=_stable_id(\"DOC\", patient.patient_id + \"CONSENT\"),\n                       doc_type=DocType.CONSENT,\n                       created_at=_now_iso(),\n                       content=\"Consent: Signed procedure consent and risk disclosure.\",\n                       source=\"EHR\",\n                   )\n               )\n\n\n           return final\n\n\n       for i in range(1, n_orders + 1):\n           patient = make_patient(i)\n           surgery = random.choice(list(SurgeryType))\n           order = SurgeryOrder(\n               order_id=_stable_id(\"ORD\", patient.patient_id + surgery.value),\n               patient=patient,\n               surgery_type=surgery,\n               scheduled_date=(datetime.utcnow().date() + timedelta(days=random.randint(3, 21))).isoformat(),\n               ordering_provider_npi=str(random.randint(1000000000, 1999999999)),\n               diagnosis_codes=[\"M17.11\", \"M54.5\"] if surgery != SurgeryType.CATARACT else [\"H25.9\"],\n               created_at=_now_iso(),\n           )\n           self.orders_queue.append(order)\n           self.patient_docs[patient.patient_id] = docs_for_order(patient, surgery)\n\n\n   def poll_new_surgery_orders(self, max_n: int = 1) -> List[SurgeryOrder]:\n       pulled = self.orders_queue[:max_n]\n       self.orders_queue = self.orders_queue[max_n:]\n       return pulled\n\n\n   def get_patient_documents(self, patient_id: str) -> List[ClinicalDocument]:\n       return list(self.patient_docs.get(patient_id, []))\n\n\n   def fetch_additional_docs(self, patient_id: str, needed: List[DocType]) -> List[ClinicalDocument]:\n       generated = []\n       for dt in needed:\n           generated.append(\n               ClinicalDocument(\n                   doc_id=_stable_id(\"DOC\", patient_id + dt.value + str(time.time())),\n                   doc_type=dt,\n                   created_at=_now_iso(),\n                   content=f\"Auto-collected document for {dt.value}: extracted and formatted per payer policy.\",\n                   source=\"AutoCollector\",\n               )\n           )\n       self.patient_docs.setdefault(patient_id, []).extend(generated)\n       return generated\n\n\nclass MockPayerPortal:\n   def __init__(self):\n       self.db: Dict[str, Dict[str, Any]] = {}\n       random.seed(11)\n\n\n   def required_docs_policy(self, plan: InsurancePlan, surgery: SurgeryType) -> List[DocType]:\n       base = [DocType.H_AND_P, DocType.CLINICAL_NOTE, DocType.MED_LIST]\n       if surgery in [SurgeryType.SPINE_FUSION, SurgeryType.KNEE_ARTHROPLASTY]:\n           base += [DocType.IMAGING, DocType.LABS, DocType.PRIOR_TX]\n       if surgery == SurgeryType.BARIATRIC:\n           base += [DocType.LABS, DocType.PRIOR_TX]\n       if plan in [InsurancePlan.PAYER_BETA, InsurancePlan.PAYER_GAMMA]:\n           base += [DocType.CONSENT]\n       return sorted(list(set(base)), key=lambda x: x.value)\n\n\n   def submit(self, pa: PriorAuthRequest) -> PayerResponse:\n       payer_ref = _stable_id(\"PAYREF\", pa.request_id + _now_iso())\n       docs_present = {d.doc_type for d in pa.docs_attached}\n       required = self.required_docs_policy(pa.order.patient.plan, pa.order.surgery_type)\n       missing = [d for d in required if d not in docs_present]\n\n\n       self.db[payer_ref] = {\n           \"status\": AuthStatus.SUBMITTED,\n           \"order_id\": pa.order.order_id,\n           \"plan\": pa.order.patient.plan,\n           \"surgery\": pa.order.surgery_type,\n           \"missing\": missing,\n           \"polls\": 0,\n           \"submitted_at\": _now_iso(),\n           \"denial_reason\": None,\n       }\n\n\n       msg = \"Submission received. Case queued for review.\"\n       if missing:\n           msg += \" Initial validation indicates incomplete documentation.\"\n       return PayerResponse(status=AuthStatus.SUBMITTED, payer_ref=payer_ref, message=msg)\n\n\n   def check_status(self, payer_ref: str) -> PayerResponse:\n       if payer_ref not in self.db:\n           return PayerResponse(\n               status=AuthStatus.DENIED,\n               payer_ref=payer_ref,\n               message=\"Case not found (possible payer system error).\",\n               denial_reason=DenialReason.OTHER,\n               confidence=0.4,\n           )\n\n\n       case = self.db[payer_ref]\n       case[\"polls\"] += 1\n\n\n       if case[\"status\"] == AuthStatus.SUBMITTED and case[\"polls\"] >= 1:\n           case[\"status\"] = AuthStatus.IN_REVIEW\n\n\n       if case[\"status\"] == AuthStatus.IN_REVIEW and case[\"polls\"] >= 3:\n           if case[\"missing\"]:\n               case[\"status\"] = AuthStatus.DENIED\n               case[\"denial_reason\"] = DenialReason.MISSING_DOCS\n           else:\n               roll = random.random()\n               if roll < 0.10:\n                   case[\"status\"] = AuthStatus.DENIED\n                   case[\"denial_reason\"] = DenialReason.CODING_ISSUE\n               elif roll < 0.18:\n                   case[\"status\"] = AuthStatus.DENIED\n                   case[\"denial_reason\"] = DenialReason.MEDICAL_NECESSITY\n               else:\n                   case[\"status\"] = AuthStatus.APPROVED\n\n\n       if case[\"status\"] == AuthStatus.DENIED:\n           dr = case[\"denial_reason\"] or DenialReason.OTHER\n           missing = case[\"missing\"] if dr == DenialReason.MISSING_DOCS else []\n           conf = 0.9 if dr != DenialReason.OTHER else 0.55\n           return PayerResponse(\n               status=AuthStatus.DENIED,\n               payer_ref=payer_ref,\n               message=f\"Denied. Reason={dr.value}.\",\n               denial_reason=dr,\n               missing_docs=missing,\n               confidence=conf,\n           )\n\n\n       if case[\"status\"] == AuthStatus.APPROVED:\n           return PayerResponse(\n               status=AuthStatus.APPROVED,\n               payer_ref=payer_ref,\n               message=\"Approved. Authorization issued.\",\n               confidence=0.95,\n           )\n\n\n       return PayerResponse(\n           status=case[\"status\"],\n           payer_ref=payer_ref,\n           message=f\"Status={case['status'].value}. Polls={case['polls']}.\",\n           confidence=0.9,\n       )\n\n\n   def file_appeal(self, payer_ref: str, appeal_text: str, attached_docs: List[ClinicalDocument]) -> PayerResponse:\n       if payer_ref not in self.db:\n           return PayerResponse(\n               status=AuthStatus.DENIED,\n               payer_ref=payer_ref,\n               message=\"Appeal failed: case not found.\",\n               denial_reason=DenialReason.OTHER,\n               confidence=0.4,\n           )\n\n\n       case = self.db[payer_ref]\n       docs_present = {d.doc_type for d in attached_docs}\n       still_missing = [d for d in case[\"missing\"] if d not in docs_present]\n       case[\"missing\"] = still_missing\n       case[\"status\"] = AuthStatus.APPEALED\n       case[\"polls\"] = 0\n\n\n       msg = \"Appeal submitted and queued for review.\"\n       if still_missing:\n           msg += f\" Warning: still missing {', '.join([d.value for d in still_missing])}.\"\n       return PayerResponse(status=AuthStatus.APPEALED, payer_ref=payer_ref, message=msg, confidence=0.9)<\/code><\/pre>\n<\/div>\n<\/div>\n
We model payer-side behavior, including documentation policies, review timelines, and denial logic. We encode simplified but realistic payer rules to demonstrate how policy-driven automation works in practice. We expose predictable failure modes that the agent must respond to safely. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
def required_docs_for_order(payer: MockPayerPortal, order: SurgeryOrder) -> List[DocType]:\n   return payer.required_docs_policy(order.patient.plan, order.surgery_type)\n\n\ndef attach_best_docs(ehr_docs: List[ClinicalDocument], required: List[DocType]) -> List[ClinicalDocument]:\n   by_type: Dict[DocType, List[ClinicalDocument]] = {}\n   for d in ehr_docs:\n       by_type.setdefault(d.doc_type, []).append(d)\n   attached = []\n   for dt in required:\n       if dt in by_type:\n           attached.append(by_type[dt][-1])\n   return attached\n\n\ndef compute_uncertainty(payer_resp: PayerResponse, missing_docs: List[DocType], llm_used: bool) -> float:\n   base = 0.15\n   if payer_resp.denial_reason in [DenialReason.OTHER]:\n       base += 0.35\n   if payer_resp.denial_reason in [DenialReason.MEDICAL_NECESSITY]:\n       base += 0.25\n   if payer_resp.denial_reason in [DenialReason.CODING_ISSUE]:\n       base += 0.20\n   if missing_docs:\n       base += 0.10\n   if llm_used:\n       base -= 0.05\n   return max(0.0, min(1.0, base + (1 - payer_resp.confidence) * 0.6))\n\n\ndef rule_based_denial_analysis(order: SurgeryOrder, payer_resp: PayerResponse) -> Dict[str, Any]:\n   rec = {\"missing_docs\": [], \"rationale\": \"\", \"appeal_text\": \"\"}\n   if payer_resp.denial_reason == DenialReason.MISSING_DOCS:\n       rec[\"missing_docs\"] = payer_resp.missing_docs\n       rec[\"rationale\"] = \"Denial indicates incomplete documentation per payer policy. Collect and resubmit as appeal.\"\n       rec[\"appeal_text\"] = (\n           f\"Appeal for prior authorization ({payer_resp.payer_ref})n\"\n           f\"Patient: {order.patient.name} ({order.patient.member_id})n\"\n           f\"Procedure: {order.surgery_type.value}n\"\n           f\"Reason for appeal: Missing documentation has now been attached. Please re-review.n\"\n       )\n   elif payer_resp.denial_reason == DenialReason.CODING_ISSUE:\n       rec[\"rationale\"] = \"Potential coding mismatch. Verify diagnosis\/procedure codes and include supporting note.\"\n       rec[\"appeal_text\"] = (\n           f\"Appeal ({payer_resp.payer_ref}): Requesting reconsideration.n\"\n           f\"Attached: Updated clinical note clarifying diagnosis and indication; please re-review coding alignment.n\"\n       )\n   elif payer_resp.denial_reason == DenialReason.MEDICAL_NECESSITY:\n       rec[\"rationale\"] = \"Medical necessity denial. Add prior treatments timeline, imaging severity, and functional impact.\"\n       rec[\"appeal_text\"] = (\n           f\"Appeal ({payer_resp.payer_ref}): Medical necessity reconsideration.n\"\n           f\"Attached: Prior conservative therapies, imaging, and clinician attestation of functional limitation.n\"\n       )\n   else:\n       rec[\"rationale\"] = \"Unclear denial. Escalate if payer message lacks actionable details.\"\n       rec[\"appeal_text\"] = (\n           f\"Appeal ({payer_resp.payer_ref}): Requesting clarification and reconsideration.n\"\n           f\"Please provide specific criteria not met; attached full clinical packet.n\"\n       )\n   return rec\n\n\ndef llm_denial_analysis_and_appeal(order: SurgeryOrder, payer_resp: PayerResponse, docs: List[ClinicalDocument]) -> Dict[str, Any]:\n   if not OPENAI_AVAILABLE:\n       return rule_based_denial_analysis(order, payer_resp)\n\n\n   doc_summary = [{\"doc_type\": d.doc_type.value, \"source\": d.source, \"created_at\": d.created_at} for d in docs]\n   prompt = {\n       \"role\": \"user\",\n       \"content\": (\n           \"You are an RCM prior authorization specialist agent.n\"\n           \"Given the order, attached docs, and payer denial response, do three things:n\"\n           \"1) Identify what documentation is missing or what needs clarification.n\"\n           \"2) Recommend next steps.n\"\n           \"3) Draft a concise appeal letter.nn\"\n           f\"ORDER:n{order.model_dump_json(indent=2)}nn\"\n           f\"PAYER_RESPONSE:n{payer_resp.model_dump_json(indent=2)}nn\"\n           f\"ATTACHED_DOCS_METADATA:n{json.dumps(doc_summary, indent=2)}nn\"\n           \"Return STRICT JSON with keys: missing_docs (list of strings), rationale (string), appeal_text (string).\"\n       )\n   }\n\n\n   try:\n       resp = client.chat.completions.create(\n           model=\"gpt-4o-mini\",\n           messages=[prompt],\n           temperature=0.2,\n       )\n       text = resp.choices[0].message.content.strip()\n       data = json.loads(text)\n       missing = []\n       for x in data.get(\"missing_docs\", []):\n           try:\n               missing.append(DocType(x))\n           except Exception:\n               pass\n       return {\n           \"missing_docs\": missing,\n           \"rationale\": data.get(\"rationale\", \"\"),\n           \"appeal_text\": data.get(\"appeal_text\", \"\"),\n       }\n   except Exception:\n       return rule_based_denial_analysis(order, payer_resp)\n\n\nclass PriorAuthAgent:\n   def __init__(self, ehr: MockEHR, payer: MockPayerPortal, uncertainty_threshold: float = 0.55):\n       self.ehr = ehr\n       self.payer = payer\n       self.uncertainty_threshold = uncertainty_threshold\n       self.audit_log: List[Dict[str, Any]] = []\n\n\n   def log(self, event: str, payload: Dict[str, Any]):\n       self.audit_log.append({\"ts\": _now_iso(), \"event\": event, **payload})\n\n\n   def build_prior_auth_request(self, order: SurgeryOrder) -> PriorAuthRequest:\n       required = required_docs_for_order(self.payer, order)\n       docs = self.ehr.get_patient_documents(order.patient.patient_id)\n       attached = attach_best_docs(docs, required)\n\n\n       req = PriorAuthRequest(\n           request_id=_stable_id(\"PA\", order.order_id + order.patient.member_id),\n           order=order,\n           docs_attached=attached,\n           payload={\n               \"member_id\": order.patient.member_id,\n               \"plan\": order.patient.plan.value,\n               \"procedure\": order.surgery_type.value,\n               \"diagnosis_codes\": order.diagnosis_codes,\n               \"scheduled_date\": order.scheduled_date,\n               \"provider_npi\": order.ordering_provider_npi,\n               \"attached_doc_types\": [d.doc_type.value for d in attached],\n           }\n       )\n       self.log(\"pa_request_built\", {\"order_id\": order.order_id, \"required_docs\": [d.value for d in required], \"attached\": req.payload[\"attached_doc_types\"]})\n       return req\n\n\n   def submit_and_monitor(self, pa: PriorAuthRequest, max_polls: int = 7) -> Dict[str, Any]:\n       pa.submitted_at = _now_iso()\n       submit_resp = self.payer.submit(pa)\n       self.log(\"submitted\", {\"request_id\": pa.request_id, \"payer_ref\": submit_resp.payer_ref, \"message\": submit_resp.message})\n\n\n       payer_ref = submit_resp.payer_ref\n\n\n       for _ in range(max_polls):\n           time.sleep(0.25)\n           status = self.payer.check_status(payer_ref)\n           self.log(\"status_polled\", {\"payer_ref\": payer_ref, \"status\": status.status.value, \"message\": status.message})\n\n\n           if status.status == AuthStatus.APPROVED:\n               return {\"final_status\": \"APPROVED\", \"payer_ref\": payer_ref, \"details\": status.model_dump()}\n\n\n           if status.status == AuthStatus.DENIED:\n               decision = self.handle_denial(pa, payer_ref, status)\n               if decision.action == \"escalate\":\n                   return {\n                       \"final_status\": \"ESCALATED_TO_HUMAN\",\n                       \"payer_ref\": payer_ref,\n                       \"decision\": decision.model_dump(),\n                       \"details\": status.model_dump(),\n                   }\n               if decision.action == \"appeal\":\n                   appeal_docs = pa.docs_attached[:]\n                   appeal_resp = self.payer.file_appeal(payer_ref, decision.appeal_text or \"\", appeal_docs)\n                   self.log(\"appeal_filed\", {\"payer_ref\": payer_ref, \"message\": appeal_resp.message})\n\n\n                   for _ in range(max_polls):\n                       time.sleep(0.25)\n                       post = self.payer.check_status(payer_ref)\n                       self.log(\"post_appeal_polled\", {\"payer_ref\": payer_ref, \"status\": post.status.value, \"message\": post.message})\n                       if post.status == AuthStatus.APPROVED:\n                           return {\"final_status\": \"APPROVED_AFTER_APPEAL\", \"payer_ref\": payer_ref, \"details\": post.model_dump()}\n                       if post.status == AuthStatus.DENIED:\n                           return {\"final_status\": \"DENIED_AFTER_APPEAL\", \"payer_ref\": payer_ref, \"details\": post.model_dump(), \"decision\": decision.model_dump()}\n\n\n                   return {\"final_status\": \"APPEAL_PENDING\", \"payer_ref\": payer_ref, \"decision\": decision.model_dump()}\n\n\n               return {\"final_status\": \"DENIED_NO_ACTION\", \"payer_ref\": payer_ref, \"decision\": decision.model_dump(), \"details\": status.model_dump()}\n\n\n       return {\"final_status\": \"PENDING_TIMEOUT\", \"payer_ref\": payer_ref}\n\n\n   def handle_denial(self, pa: PriorAuthRequest, payer_ref: str, denial_resp: PayerResponse) -> AgentDecision:\n       order = pa.order\n       analysis = llm_denial_analysis_and_appeal(order, denial_resp, pa.docs_attached) if (USE_OPENAI and OPENAI_AVAILABLE) else rule_based_denial_analysis(order, denial_resp)\n       missing_docs: List[DocType] = analysis.get(\"missing_docs\", [])\n       rationale: str = analysis.get(\"rationale\", \"\")\n       appeal_text: str = analysis.get(\"appeal_text\", \"\")\n\n\n       if denial_resp.denial_reason == DenialReason.MISSING_DOCS and denial_resp.missing_docs:\n           missing_docs = denial_resp.missing_docs\n\n\n       if missing_docs:\n           new_docs = self.ehr.fetch_additional_docs(order.patient.patient_id, missing_docs)\n           pa.docs_attached.extend(new_docs)\n           self.log(\"missing_docs_collected\", {\"payer_ref\": payer_ref, \"collected\": [d.doc_type.value for d in new_docs]})\n\n\n       uncertainty = compute_uncertainty(denial_resp, missing_docs, llm_used=(USE_OPENAI and OPENAI_AVAILABLE))\n       self.log(\"denial_analyzed\", {\"payer_ref\": payer_ref, \"denial_reason\": (denial_resp.denial_reason.value if denial_resp.denial_reason else None),\n                                   \"uncertainty\": uncertainty, \"missing_docs\": [d.value for d in missing_docs]})\n\n\n       if uncertainty >= self.uncertainty_threshold:\n           return AgentDecision(\n               action=\"escalate\",\n               missing_docs=missing_docs,\n               rationale=f\"{rationale} Escalating due to high uncertainty ({uncertainty:.2f}) >= threshold ({self.uncertainty_threshold:.2f}).\",\n               uncertainty=uncertainty,\n               next_wait_seconds=0,\n           )\n\n\n       if not appeal_text:\n           analysis2 = rule_based_denial_analysis(order, denial_resp)\n           appeal_text = analysis2.get(\"appeal_text\", \"\")\n\n\n       attached_types = sorted(list({d.doc_type.value for d in pa.docs_attached}))\n       appeal_text = (\n           appeal_text.strip()\n           + \"nnAttached documents:n- \"\n           + \"n- \".join(attached_types)\n           + \"nnRequested outcome: Reconsideration and authorization issuance.n\"\n       )\n\n\n       return AgentDecision(\n           action=\"appeal\",\n           missing_docs=missing_docs,\n           rationale=f\"{rationale} Proceeding autonomously (uncertainty {uncertainty:.2f} < threshold {self.uncertainty_threshold:.2f}).\",\n           uncertainty=uncertainty,\n           appeal_text=appeal_text,\n           next_wait_seconds=1,\n       )<\/code><\/pre>\n<\/div>\n<\/div>\n
We implement the core intelligence layer that attaches documents, analyzes denials, and estimates uncertainty. We demonstrate how rule-based logic and optional LLM reasoning can coexist without compromising determinism. We explicitly gate automation decisions to maintain safety in a healthcare context. Check out the\u00a0FULL CODES here<\/a><\/strong>.<\/p>\n
\n
\n
\n
\n
<\/div>\n
<\/div>\n
<\/div>\n<\/div>\n
Copy Code<\/span>Copied<\/span>Use a different Browser<\/span><\/a><\/div>\n<\/div>\n
ehr = MockEHR()\nehr.seed_data(n_orders=6)\n\n\npayer = MockPayerPortal()\nagent = PriorAuthAgent(ehr, payer, uncertainty_threshold=0.55)\n\n\nresults = []\nprint(\"=== Starting Autonomous Prior Authorization Agent Demo ===\")\nprint(f\"OpenAI enabled: {USE_OPENAI and OPENAI_AVAILABLE}n\")\n\n\nwhile True:\n   new_orders = ehr.poll_new_surgery_orders(max_n=1)\n   if not new_orders:\n       break\n\n\n   order = new_orders[0]\n   print(f\"n--- New Surgery Order Detected ---\")\n   print(f\"Order: {order.order_id} | Patient: {order.patient.patient_id} | Plan: {order.patient.plan.value} | Surgery: {order.surgery_type.value}\")\n\n\n   pa = agent.build_prior_auth_request(order)\n   outcome = agent.submit_and_monitor(pa, max_polls=7)\n   results.append({\"order_id\": order.order_id, \"patient_id\": order.patient.patient_id, **outcome})\n\n\n   print(f\"Outcome: {outcome['final_status']} | PayerRef: {outcome.get('payer_ref')}\")\n\n\nprint(\"n=== Summary ===\")\nstatus_counts = {}\nfor r in results:\n   status_counts[r[\"final_status\"]] = status_counts.get(r[\"final_status\"], 0) + 1\nprint(\"Final status counts:\", status_counts)\n\n\nprint(\"nSample result (first case):\")\nprint(json.dumps(results[0], indent=2))\n\n\nprint(\"n=== Audit Log (last ~12 events) ===\")\nfor row in agent.audit_log[-12:]:\n   print(json.dumps(row, indent=2))\n\n\nprint(\n   \"nHardening checklist (high level):n\"\n   \"- Swap mocks for real EHR + payer integrations (FHIR\/HL7, payer APIs\/portal automations)n\"\n   \"- Add PHI governance (tokenization, least-privilege access, encrypted logging, retention controls)n\"\n   \"- Add deterministic policy engine + calibrated uncertainty modeln\"\n   \"- Add human-in-the-loop UI with SLA timers, retries\/backoff, idempotency keysn\"\n   \"- Add evidence packing (policy citations, structured attachments, templates)n\"\n)\n<\/code><\/pre>\n<\/div>\n<\/div>\n
We orchestrate the full end-to-end workflow and generate operational summaries and audit logs. We track outcomes, escalation events, and system behavior to support transparency and compliance. We emphasize observability and traceability as essential requirements for healthcare AI systems.<\/p>\n
In conclusion, we illustrated how agentic AI can meaningfully reduce administrative friction in healthcare RCM by automating repetitive, rules-driven prior authorization tasks while preserving human oversight for ambiguous or high-risk decisions. We showed that combining deterministic policy logic, uncertainty estimation, and optional LLM-assisted reasoning enables a balanced approach that aligns with healthcare\u2019s safety-critical nature. This work should be viewed as an architectural and educational reference rather than a deployable medical system; any real-world implementation must adhere to HIPAA and regional data protection laws, incorporate de-identification and access controls, undergo clinical and compliance review, and be validated against payer-specific policies.<\/p>\n
\n
Check out the\u00a0FULL CODES here<\/a><\/strong>.\u00a0Also,\u00a0feel free to follow us on\u00a0Twitter<\/mark><\/a><\/strong>\u00a0and don\u2019t forget to join our\u00a0100k+ ML SubReddit<\/a><\/strong>\u00a0and Subscribe to\u00a0our Newsletter<\/a><\/strong>. Wait! are you on telegram?\u00a0now you can join us on telegram as well.<\/a><\/strong><\/p>\n
The post How to Build a Safe, Autonomous Prior Authorization Agent for Healthcare Revenue Cycle Management with Human-in-the-Loop Controls<\/a> appeared first on MarkTechPost<\/a>.<\/p>","protected":false},"excerpt":{"rendered":"
In this tutorial, we demonstra…<\/p>\n","protected":false},"author":1,"featured_media":29,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1],"tags":[],"class_list":["post-271","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-uncategorized"],"_links":{"self":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/271","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=271"}],"version-history":[{"count":0,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/posts\/271\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=\/wp\/v2\/media\/29"}],"wp:attachment":[{"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=271"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=271"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/connectword.dpdns.org\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=271"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}