"""
FDA Device Verification Agent for Prior Authorization Workflow
This agent analyzes FHIR DeviceRequest resources to determine if a device
is FDA approved/cleared and meets prior authorization requirements.
"""
import json
import logging
from typing import Dict, List, Any, Optional, Tuple
from dataclasses import dataclass, asdict
from enum import Enum
import asyncio
from datetime import datetime
from openai import AsyncOpenAI
from fastmcp.client import Client
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class PADecision(Enum):
"""Prior Authorization decision types"""
APPROVED = "approved"
DENIED = "denied"
PENDING_INFO = "pending_information"
MANUAL_REVIEW = "manual_review"
@dataclass
class FDAVerificationResult:
"""Result of FDA verification for a device"""
device_name: str
is_fda_approved: bool
fda_status: str # cleared, approved, registered, not_found
fda_number: Optional[str] = None
manufacturer: Optional[str] = None
device_class: Optional[str] = None
is_otc: bool = False
confidence: float = 0.0
verification_method: str = "unknown"
notes: List[str] = None
@dataclass
class PARecommendation:
"""Prior Authorization recommendation from the agent"""
decision: PADecision
fda_verification: Optional[FDAVerificationResult] = None
reasoning: str = ""
requirements_met: Dict[str, bool] = None
missing_information: List[str] = None
recommendations: List[str] = None
confidence: float = 0.0
timestamp: str = ""
class FDAVerificationAgent:
"""
Agent that verifies FDA approval status of medical devices
for prior authorization workflows.
"""
def __init__(
self,
mcp_url: str = "http://localhost:8090/mcp/",
api_key: str = None,
base_url: str = None,
model: str = "llama-4-scout-17b-16e-w4a16"
):
"""
Initialize the FDA Verification Agent.
Args:
mcp_url: URL of the FDA MCP server
api_key: API key for LLM
base_url: Base URL for LLM API
model: Model name to use
"""
self.mcp_url = mcp_url
self.model = model
# Initialize OpenAI client for Llama
self.llm = AsyncOpenAI(
api_key=api_key,
base_url=base_url
)
logger.info(f"FDA Verification Agent initialized with MCP: {mcp_url}")
async def analyze_device_request(self, device_request: Dict[str, Any]) -> PARecommendation:
"""
Main entry point: Analyze a FHIR DeviceRequest and provide PA recommendation.
Args:
device_request: FHIR DeviceRequest resource
Returns:
PARecommendation with decision and supporting information
"""
logger.info("Starting FDA verification analysis for DeviceRequest")
try:
# Step 1: Extract device information using LLM
device_info = await self._extract_device_info_with_llm(device_request)
# Step 2: Query FDA MCP server for verification
fda_result = await self._verify_fda_status(device_info)
# Step 3: Make PA decision based on FDA status and requirements
recommendation = await self._make_pa_decision(device_request, fda_result)
return recommendation
except Exception as e:
logger.error(f"Error in FDA verification: {e}")
return PARecommendation(
decision=PADecision.MANUAL_REVIEW,
reasoning=f"Error during FDA verification: {str(e)}",
recommendations=["Manual review required due to processing error"],
confidence=0.0,
timestamp=datetime.now().isoformat()
)
async def _extract_device_info_with_llm(self, device_request: Dict[str, Any]) -> Dict[str, Any]:
"""
Use LLM to intelligently extract device information from DeviceRequest.
"""
prompt = f"""You are analyzing a FHIR DeviceRequest for FDA verification.
Extract the following information from this DeviceRequest:
DeviceRequest:
{json.dumps(device_request, indent=2)}
Extract and return a JSON object with:
{{
"device_name": "extracted device name or description",
"manufacturer": "manufacturer if mentioned",
"model_number": "model number if present",
"device_type": "type of device (e.g., hearing aid, pacemaker)",
"k_number": "FDA 510(k) number if present (format: K######)",
"pma_number": "FDA PMA number if present (format: P######)",
"notes": ["any relevant notes or additional information"]
}}
Focus on:
1. The codeCodeableConcept text and display fields
2. Any notes that might contain FDA numbers
3. Performer field for manufacturer information
4. Any extensions with device details
Return ONLY the JSON object, no other text.
"""
try:
response = await self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "You are a medical device expert extracting information from FHIR resources."},
{"role": "user", "content": prompt}
],
temperature=0.1,
max_tokens=500
)
content = response.choices[0].message.content.strip()
# Clean up the response if needed
if "```json" in content:
content = content.split("```json")[1].split("```")[0].strip()
elif "```" in content:
content = content.split("```")[1].split("```")[0].strip()
device_info = json.loads(content)
logger.info(f"Extracted device info: {device_info}")
return device_info
except Exception as e:
logger.error(f"LLM extraction failed: {e}")
# Fallback to basic extraction
return self._basic_extraction(device_request)
def _basic_extraction(self, device_request: Dict[str, Any]) -> Dict[str, Any]:
"""Fallback method for basic extraction without LLM."""
device_info = {
"device_name": None,
"manufacturer": None,
"model_number": None,
"device_type": None,
"k_number": None,
"pma_number": None,
"notes": []
}
# Extract from codeCodeableConcept
if "codeCodeableConcept" in device_request:
code_concept = device_request["codeCodeableConcept"]
if "text" in code_concept:
device_info["device_name"] = code_concept["text"]
if "coding" in code_concept:
for coding in code_concept["coding"]:
if "display" in coding:
device_info["device_type"] = coding["display"]
# Extract from performer
if "performer" in device_request:
performer = device_request["performer"]
if isinstance(performer, dict) and "display" in performer:
device_info["manufacturer"] = performer["display"]
# Look for FDA numbers in notes
if "note" in device_request:
import re
for note in device_request["note"]:
text = note.get("text", "")
# Look for K-number
k_match = re.search(r'K\d{6}', text.upper())
if k_match:
device_info["k_number"] = k_match.group()
# Look for PMA number
p_match = re.search(r'P\d{6}', text.upper())
if p_match:
device_info["pma_number"] = p_match.group()
device_info["notes"].append(text)
return device_info
async def _verify_fda_status(self, device_info: Dict[str, Any]) -> FDAVerificationResult:
"""
Query FDA MCP server to verify device status.
"""
async with Client(self.mcp_url) as client:
# Try precise search first if we have FDA numbers
if device_info.get("k_number"):
logger.info(f"Searching by K-number: {device_info['k_number']}")
result = await client.call_tool(
"search_fda_by_identifier",
{"k_number": device_info["k_number"]}
)
if result:
return self._parse_fda_result(result, "k_number")
if device_info.get("pma_number"):
logger.info(f"Searching by PMA number: {device_info['pma_number']}")
result = await client.call_tool(
"search_fda_by_identifier",
{"pma_number": device_info["pma_number"]}
)
if result:
return self._parse_fda_result(result, "pma_number")
# Try fuzzy search by device name and manufacturer
if device_info.get("device_name"):
logger.info(f"Searching by device info: {device_info['device_name']}")
results = await client.call_tool(
"search_fda_by_device_info",
{
"device_name": device_info.get("device_name"),
"manufacturer": device_info.get("manufacturer"),
"fuzzy_search": True
}
)
if results:
return self._parse_fda_results_list(results)
# No FDA record found
return FDAVerificationResult(
device_name=device_info.get("device_name", "Unknown Device"),
is_fda_approved=False,
fda_status="not_found",
confidence=0.0,
verification_method="no_match",
notes=["No FDA clearance or approval found for this device"]
)
def _parse_fda_result(self, result: Any, method: str) -> FDAVerificationResult:
"""Parse single FDA result from MCP."""
try:
# Extract JSON from TextContent
if hasattr(result, 'content'):
for content in result.content:
if hasattr(content, 'text'):
data = json.loads(content.text)
is_approved = data.get("status") in ["cleared", "approved"]
return FDAVerificationResult(
device_name=data.get("device_name", ""),
is_fda_approved=is_approved,
fda_status=data.get("status", "unknown"),
fda_number=data.get("fda_number"),
manufacturer=data.get("manufacturer"),
device_class=data.get("device_class"),
is_otc=data.get("is_otc", False),
confidence=data.get("confidence_score", 1.0),
verification_method=method,
notes=data.get("notes", [])
)
except Exception as e:
logger.error(f"Error parsing FDA result: {e}")
return FDAVerificationResult(
device_name="Parse Error",
is_fda_approved=False,
fda_status="error",
confidence=0.0,
verification_method=method,
notes=[f"Error parsing FDA response: {str(e)}"]
)
def _parse_fda_results_list(self, results: Any) -> FDAVerificationResult:
"""Parse list of FDA results and return best match."""
try:
if hasattr(results, 'content'):
for content in results.content:
if hasattr(content, 'text'):
data_list = json.loads(content.text)
if not data_list:
break
# Use highest confidence result
best = max(data_list, key=lambda x: x.get("confidence_score", 0))
if best.get("confidence_score", 0) < 0.3:
return FDAVerificationResult(
device_name=best.get("device_name", "Unknown"),
is_fda_approved=False,
fda_status="low_confidence",
confidence=best.get("confidence_score", 0),
verification_method="fuzzy_search",
notes=["Low confidence match - manual review recommended"]
)
is_approved = best.get("status") in ["cleared", "approved"]
return FDAVerificationResult(
device_name=best.get("device_name", ""),
is_fda_approved=is_approved,
fda_status=best.get("status", "unknown"),
fda_number=best.get("fda_number"),
manufacturer=best.get("manufacturer"),
device_class=best.get("device_class"),
is_otc=best.get("is_otc", False),
confidence=best.get("confidence_score", 0),
verification_method="fuzzy_search",
notes=best.get("notes", [])
)
except Exception as e:
logger.error(f"Error parsing FDA results list: {e}")
return FDAVerificationResult(
device_name="Unknown",
is_fda_approved=False,
fda_status="not_found",
confidence=0.0,
verification_method="fuzzy_search",
notes=["No matching FDA records found"]
)
async def _make_pa_decision(
self,
device_request: Dict[str, Any],
fda_result: FDAVerificationResult
) -> PARecommendation:
"""
Make PA decision based on FDA status and use LLM for reasoning.
"""
# Build context for LLM decision
context = {
"device_request_summary": self._summarize_device_request(device_request),
"fda_verification": asdict(fda_result),
"policy_requirements": {
"mandatory": {
"fda_clearance_required": True,
"device_class_2_or_lower": True
},
"preferred": {
"otc_device": True
}
}
}
prompt = f"""You are a Prior Authorization specialist reviewing a medical device request.
Your job is to APPROVE devices that meet all MANDATORY requirements.
Device Request Summary:
{json.dumps(context['device_request_summary'], indent=2)}
FDA Verification Results:
{json.dumps(context['fda_verification'], indent=2)}
Policy Requirements:
{json.dumps(context['policy_requirements'], indent=2)}
DECISION RULES:
1. APPROVE if ALL mandatory requirements are met:
- Device has FDA clearance or approval (status = "cleared" or "approved")
- Device is Class 2 or lower (device_class = "1" or "2")
- Documentation is complete (confidence >= 0.5)
2. DENY if ANY mandatory requirement is NOT met
3. PENDING_INFORMATION if verification confidence is low (< 0.5)
4. MANUAL_REVIEW only for exceptional cases
IMPORTANT:
- OTC status is PREFERRED but NOT required for approval
- If a device meets all mandatory requirements, it MUST be approved
- Focus on FDA clearance/approval as the primary criterion
The device "{context['fda_verification']['device_name']}" has:
- FDA Status: {context['fda_verification']['fda_status']}
- Device Class: {context['fda_verification']['device_class']}
- Is OTC: {context['fda_verification']['is_otc']}
- Confidence: {context['fda_verification']['confidence']}
Return a JSON object with:
{{
"decision": "approved|denied|pending_information|manual_review",
"reasoning": "detailed explanation focusing on mandatory requirements",
"requirements_met": {{
"fda_clearance": true/false,
"device_class_acceptable": true/false,
"documentation_complete": true/false
}},
"missing_information": ["list of missing items if any"],
"recommendations": ["list of recommendations"],
"confidence": 0.0-1.0
}}
Return ONLY the JSON object.
"""
try:
response = await self.llm.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "You are a Prior Authorization specialist. Your primary goal is to APPROVE devices that meet mandatory FDA requirements. OTC status is a preference, not a requirement."},
{"role": "user", "content": prompt}
],
temperature=0.1,
max_tokens=800
)
content = response.choices[0].message.content.strip()
if "```json" in content:
content = content.split("```json")[1].split("```")[0].strip()
elif "```" in content:
content = content.split("```")[1].split("```")[0].strip()
decision_data = json.loads(content)
return PARecommendation(
decision=PADecision(decision_data["decision"]),
fda_verification=fda_result,
reasoning=decision_data["reasoning"],
requirements_met=decision_data.get("requirements_met", {}),
missing_information=decision_data.get("missing_information", []),
recommendations=decision_data.get("recommendations", []),
confidence=decision_data.get("confidence", 0.5),
timestamp=datetime.now().isoformat()
)
except Exception as e:
logger.error(f"Error in LLM decision making: {e}")
# Fallback to rule-based decision
return self._rule_based_decision(fda_result)
def _rule_based_decision(self, fda_result: FDAVerificationResult) -> PARecommendation:
"""Fallback rule-based decision when LLM fails."""
requirements_met = {
"fda_clearance": fda_result.is_fda_approved,
"device_class_acceptable": fda_result.device_class in ["1", "2", None],
"documentation_complete": fda_result.confidence > 0.5
}
if fda_result.is_fda_approved and all(requirements_met.values()):
decision = PADecision.APPROVED
reasoning = f"Device '{fda_result.device_name}' has FDA {fda_result.fda_status} status"
elif fda_result.fda_status == "not_found":
decision = PADecision.PENDING_INFO
reasoning = "Unable to verify FDA status - additional information needed"
elif not fda_result.is_fda_approved:
decision = PADecision.DENIED
reasoning = f"Device does not have required FDA clearance/approval"
else:
decision = PADecision.MANUAL_REVIEW
reasoning = "Manual review required for final determination"
return PARecommendation(
decision=decision,
fda_verification=fda_result,
reasoning=reasoning,
requirements_met=requirements_met,
missing_information=["FDA number", "Manufacturer details"] if fda_result.fda_status == "not_found" else [],
recommendations=self._generate_recommendations(fda_result),
confidence=fda_result.confidence,
timestamp=datetime.now().isoformat()
)
def _summarize_device_request(self, device_request: Dict[str, Any]) -> Dict[str, Any]:
"""Create a summary of key DeviceRequest fields."""
summary = {
"resource_type": device_request.get("resourceType", "DeviceRequest"),
"status": device_request.get("status", "unknown")
}
if "codeCodeableConcept" in device_request:
code = device_request["codeCodeableConcept"]
summary["device_description"] = code.get("text", "")
if "coding" in code and code["coding"]:
summary["device_code"] = code["coding"][0].get("code", "")
summary["device_system"] = code["coding"][0].get("system", "")
if "subject" in device_request:
summary["patient"] = device_request["subject"].get("reference", "")
if "authoredOn" in device_request:
summary["request_date"] = device_request["authoredOn"]
if "reasonCode" in device_request:
reasons = []
for reason in device_request["reasonCode"]:
if "text" in reason:
reasons.append(reason["text"])
summary["clinical_reasons"] = reasons
return summary
def _generate_recommendations(self, fda_result: FDAVerificationResult) -> List[str]:
"""Generate recommendations based on FDA verification results."""
recommendations = []
if fda_result.fda_status == "not_found":
recommendations.append("Request FDA 510(k) or PMA number from provider")
recommendations.append("Verify exact device model and manufacturer")
elif fda_result.confidence < 0.5:
recommendations.append("Confirm device details with provider due to low confidence match")
elif fda_result.is_fda_approved:
if fda_result.is_otc:
recommendations.append("Device is available over-the-counter")
if fda_result.device_class == "3":
recommendations.append("Class III device - ensure clinical justification is documented")
return recommendations
# Convenience function for testing
async def verify_device(
device_request: Dict[str, Any],
mcp_url: str = "http://localhost:8090/mcp/",
api_key: str = None,
base_url: str = None
) -> PARecommendation:
"""
Convenience function to verify a device.
Args:
device_request: FHIR DeviceRequest resource
mcp_url: FDA MCP server URL
api_key: LLM API key
base_url: LLM base URL
Returns:
PARecommendation with decision and details
"""
agent = FDAVerificationAgent(
mcp_url=mcp_url,
api_key=api_key,
base_url=base_url
)
return await agent.analyze_device_request(device_request)
