"""FDA Device Verification Tools for Prior Authorization Workflow"""
import json
import aiohttp
from typing import Dict, List, Optional, Any
from dataclasses import dataclass, asdict
from enum import Enum
from fastmcp import Context
from core.app import mcp
import logging
logger = logging.getLogger(__name__)
# FDA API Base URL
FDA_API_BASE = "https://api.fda.gov"
# FHIR Server Base URL
FHIR_BASE = "https://hapi-fhir-davinci-data.apps.cluster-sdzgj.sdzgj.sandbox319.opentlc.com"
class FDAStatus(Enum):
CLEARED = "cleared" # 510(k) clearance
APPROVED = "approved" # PMA approval
REGISTERED = "registered" # Only registration/listing
NOT_FOUND = "not_found"
UNKNOWN = "unknown"
@dataclass
class DeviceInfo:
"""Structured device information extracted from DeviceRequest"""
device_name: Optional[str] = None
manufacturer: Optional[str] = None
model_number: Optional[str] = None
udi: Optional[str] = None
k_number: Optional[str] = None
pma_number: Optional[str] = None
device_type: Optional[str] = None
codes: List[Dict[str, str]] = None
@dataclass
class FDAVerificationResult:
"""Structured FDA verification result for PA decision-making"""
status: FDAStatus
device_name: str
manufacturer: str
fda_number: Optional[str] = None # K-number or PMA number
decision_date: Optional[str] = None
product_code: Optional[str] = None
device_class: Optional[str] = None
is_otc: bool = False
confidence_score: float = 0.0
match_method: str = "none"
notes: List[str] = None
raw_data: Optional[Dict] = None
# Brand to Manufacturer mapping for common medical devices
BRAND_MANUFACTURER_MAP = {
"jabra": "GN Hearing A/S",
"lexie": "Bose Corporation",
"sony": "WSAUD A/S",
"eargo": "Eargo, Inc.",
"audicus": "Audicus, Inc.",
"lively": "Best Sound Technology, Inc.",
"bose": "Bose Corporation",
"phonak": "Sonova AG",
"oticon": "Demant A/S",
"widex": "WS Audiology A/S",
"signia": "WS Audiology A/S",
"starkey": "Starkey Laboratories, Inc.",
"resound": "GN Hearing A/S"
}
async def make_fda_request(endpoint: str, params: Dict[str, str]) -> Optional[Dict]:
"""Make async request to FDA API"""
try:
async with aiohttp.ClientSession() as session:
url = f"{FDA_API_BASE}{endpoint}"
async with session.get(url, params=params) as response:
if response.status == 200:
return await response.json()
else:
logger.warning(f"FDA API returned status {response.status} for {url}")
return None
except Exception as e:
logger.error(f"FDA API request failed: {e}")
return None
@mcp.tool
async def analyze_device_request(device_request: Dict[str, Any]) -> DeviceInfo:
"""
Extract device information from a FHIR DeviceRequest resource.
This tool parses a FHIR DeviceRequest to identify device details that can be used
for FDA verification, including device names, codes, identifiers, and manufacturer info.
Args:
device_request: FHIR DeviceRequest resource as a dictionary
Returns:
Structured DeviceInfo with extracted details
"""
info = DeviceInfo()
# Extract from codeCodeableConcept if present
if "codeCodeableConcept" in device_request:
code_concept = device_request["codeCodeableConcept"]
if "text" in code_concept:
info.device_name = code_concept["text"]
if "coding" in code_concept:
info.codes = []
for coding in code_concept["coding"]:
info.codes.append({
"system": coding.get("system", ""),
"code": coding.get("code", ""),
"display": coding.get("display", "")
})
# Extract from codeReference if present (reference to Device resource)
if "codeReference" in device_request:
# This would typically involve fetching the Device resource
# For now, extract what we can from the reference
ref = device_request["codeReference"]
if "display" in ref:
if not info.device_name:
info.device_name = ref["display"]
# Extract UDI if present in extension or contained Device
if "extension" in device_request:
for ext in device_request["extension"]:
if "udi" in ext.get("url", "").lower():
info.udi = ext.get("valueString")
# Look for manufacturer in performer or contained resources
if "performer" in device_request:
performer = device_request["performer"]
if "display" in performer:
info.manufacturer = performer["display"]
# Parse notes for additional identifiers (K-numbers, model numbers)
if "note" in device_request:
for note in device_request["note"]:
text = note.get("text", "").upper()
# Look for K-number pattern
if "K" in text and any(c.isdigit() for c in text):
import re
k_match = re.search(r'K\d{6}', text)
if k_match:
info.k_number = k_match.group()
# Look for PMA pattern
if "P" in text and any(c.isdigit() for c in text):
import re
p_match = re.search(r'P\d{6}', text)
if p_match:
info.pma_number = p_match.group()
return asdict(info)
@mcp.tool
async def search_fda_by_identifier(
k_number: Optional[str] = None,
pma_number: Optional[str] = None,
udi: Optional[str] = None
) -> Optional[FDAVerificationResult]:
"""
Search FDA database using precise device identifiers.
This is the most reliable search method when specific FDA identifiers are known.
Searches 510(k), PMA, or UDI databases based on provided identifiers.
Args:
k_number: FDA 510(k) number (e.g., "K213424")
pma_number: FDA PMA number (e.g., "P160043")
udi: FDA Unique Device Identifier
Returns:
FDAVerificationResult with device status, or None if not found
"""
if k_number:
# Search 510(k) database
params = {"search": f'k_number:"{k_number}"', "limit": "1"}
response = await make_fda_request("/device/510k.json", params)
if response and response.get("results"):
record = response["results"][0]
# Check for OTC classification
product_code = record.get("product_code", "")
is_otc = False
notes = [f"FDA 510(k) Clearance: {record.get('decision_description', '')}"]
# Check primary product code
if product_code in ["QUF", "QUG", "QUH", "SCR"]:
is_otc = True
notes.append("Device classified as OTC (Over-The-Counter)")
# Also check subsequent product codes in openfda
openfda = record.get("openfda", {})
if "subsequent_product_code" in openfda:
codes = openfda["subsequent_product_code"]
if any(code in ["QUF", "QUG", "QUH", "SCR"] for code in codes):
is_otc = True
if "Device classified as OTC (Over-The-Counter)" not in notes:
notes.append("Device classified as OTC (Over-The-Counter)")
return asdict(FDAVerificationResult(
status=FDAStatus.CLEARED,
device_name=record.get("device_name", ""),
manufacturer=record.get("applicant", ""),
fda_number=k_number,
decision_date=record.get("decision_date", ""),
product_code=product_code,
device_class=record.get("openfda", {}).get("device_class", ""),
is_otc=is_otc,
confidence_score=1.0,
match_method="k_number_exact",
notes=notes,
raw_data=record
))
if pma_number:
# Search PMA database
params = {"search": f'pma_number:"{pma_number}"', "limit": "1"}
response = await make_fda_request("/device/pma.json", params)
if response and response.get("results"):
record = response["results"][0]
return asdict(FDAVerificationResult(
status=FDAStatus.APPROVED,
device_name=record.get("trade_name", ""),
manufacturer=record.get("applicant", ""),
fda_number=pma_number,
decision_date=record.get("decision_date", ""),
product_code=record.get("product_code", ""),
device_class="3", # PMA devices are Class III
confidence_score=1.0,
match_method="pma_number_exact",
notes=["FDA PMA Approval (Class III device)"],
raw_data=record
))
if udi:
# Search UDI database
params = {"search": f'udi:"{udi}"', "limit": "1"}
response = await make_fda_request("/device/udi.json", params)
if response and response.get("results"):
record = response["results"][0]
# UDI records may reference 510(k) or PMA numbers
clearance_number = record.get("clearance_number")
if clearance_number:
if clearance_number.startswith("K"):
return await search_fda_by_identifier(k_number=clearance_number)
elif clearance_number.startswith("P"):
return await search_fda_by_identifier(pma_number=clearance_number)
# Return basic registration info if no clearance found
return asdict(FDAVerificationResult(
status=FDAStatus.REGISTERED,
device_name=record.get("brand_name", ""),
manufacturer=record.get("company_name", ""),
confidence_score=0.7,
match_method="udi_exact",
notes=["Device found in FDA UDI database but clearance status unclear"],
raw_data=record
))
return None
@mcp.tool
async def search_fda_by_device_info(
device_name: Optional[str] = None,
manufacturer: Optional[str] = None,
product_code: Optional[str] = None,
fuzzy_search: bool = True
) -> List[FDAVerificationResult]:
"""
Search FDA database using device name, manufacturer, or product code.
Handles brand name to manufacturer mapping and fuzzy matching.
Returns multiple candidates when exact match isn't found.
Args:
device_name: Device or brand name
manufacturer: Manufacturer or applicant name
product_code: FDA product classification code (e.g., "QUH" for OTC hearing aids)
fuzzy_search: Enable brand mapping and fuzzy matching
Returns:
List of potential FDAVerificationResult matches, sorted by confidence
"""
results = []
# Handle brand to manufacturer mapping
if fuzzy_search and device_name and not manufacturer:
brand_lower = device_name.lower()
for brand, mfr in BRAND_MANUFACTURER_MAP.items():
if brand in brand_lower:
manufacturer = mfr
logger.info(f"Mapped brand '{device_name}' to manufacturer '{manufacturer}'")
break
# Build search query
search_parts = []
if device_name:
# Clean device name for search
clean_name = device_name.replace('"', '').replace("'", "")
search_parts.append(f'device_name:"{clean_name}"')
if manufacturer:
search_parts.append(f'applicant:"{manufacturer}"')
if product_code:
search_parts.append(f'product_code:"{product_code}"')
if not search_parts:
return []
# Search 510(k) database first (most common)
search_query = " AND ".join(search_parts) if len(search_parts) > 1 else search_parts[0]
params = {"search": search_query, "limit": "10"}
response = await make_fda_request("/device/510k.json", params)
if response and response.get("results"):
for record in response["results"]:
confidence = calculate_confidence(record, device_name, manufacturer)
result = FDAVerificationResult(
status=FDAStatus.CLEARED,
device_name=record.get("device_name", ""),
manufacturer=record.get("applicant", ""),
fda_number=record.get("k_number", ""),
decision_date=record.get("decision_date", ""),
product_code=record.get("product_code", ""),
device_class=record.get("openfda", {}).get("device_class", ""),
confidence_score=confidence,
match_method="device_info_search",
notes=[],
raw_data=record
)
# Check for OTC classification
openfda = record.get("openfda", {})
if "subsequent_product_code" in openfda:
codes = openfda["subsequent_product_code"]
if any(code in ["QUF", "QUG", "QUH", "SCR"] for code in codes):
result.is_otc = True
result.notes.append("Device classified as OTC (Over-The-Counter)")
results.append(asdict(result))
# If no results or low confidence, try PMA database
if not results or all(r["confidence_score"] < 0.5 for r in results):
params = {"search": search_query, "limit": "5"}
response = await make_fda_request("/device/pma.json", params)
if response and response.get("results"):
for record in response["results"]:
confidence = calculate_confidence(record, device_name, manufacturer, name_field="trade_name")
results.append(asdict(FDAVerificationResult(
status=FDAStatus.APPROVED,
device_name=record.get("trade_name", ""),
manufacturer=record.get("applicant", ""),
fda_number=record.get("pma_number", ""),
decision_date=record.get("decision_date", ""),
product_code=record.get("product_code", ""),
device_class="3",
confidence_score=confidence,
match_method="device_info_search_pma",
notes=["FDA PMA Approval (Class III device)"],
raw_data=record
)))
# Sort by confidence score
results.sort(key=lambda x: x["confidence_score"], reverse=True)
return results[:5] # Return top 5 matches
def calculate_confidence(record: Dict, device_name: Optional[str],
manufacturer: Optional[str], name_field: str = "device_name") -> float:
"""Calculate confidence score for a match"""
confidence = 0.0
if device_name and name_field in record:
record_name = record[name_field].lower()
search_name = device_name.lower()
if search_name == record_name:
confidence += 0.5
elif search_name in record_name or record_name in search_name:
confidence += 0.3
else:
# Check for partial word matches
search_words = set(search_name.split())
record_words = set(record_name.split())
if search_words & record_words:
confidence += 0.2
if manufacturer and "applicant" in record:
record_mfr = record["applicant"].lower()
search_mfr = manufacturer.lower()
if search_mfr == record_mfr:
confidence += 0.5
elif search_mfr in record_mfr or record_mfr in search_mfr:
confidence += 0.3
return min(confidence, 1.0)
@mcp.tool
async def verify_fda_status(
fda_results: Optional[List[Dict[str, Any]]] = None,
required_class: Optional[str] = None,
require_otc: bool = False
) -> Dict[str, Any]:
"""
Verify FDA status and return structured decision for PA workflow.
Analyzes FDA search results and returns a decision with reasoning
suitable for prior authorization determination.
Args:
fda_results: List of FDA verification results from search tools (can be None or empty)
required_class: Required device class ("1", "2", or "3")
require_otc: Whether device must be OTC for approval
Returns:
Structured decision with approval status, reasoning, and recommendations
"""
# Handle None or empty results
if not fda_results or fda_results is None:
return {
"approved_for_pa": False,
"reason": "No FDA clearance or approval found",
"recommendations": [
"Verify device name and manufacturer",
"Check if device requires FDA clearance",
"Request specific FDA 510(k) or PMA number from provider"
]
}
# Use highest confidence result
best_result = fda_results[0]
# Check basic FDA status
if best_result["status"] not in [FDAStatus.CLEARED.value, FDAStatus.APPROVED.value]:
return {
"approved_for_pa": False,
"reason": f"Device has FDA status '{best_result['status']}' - clearance or approval required",
"fda_number": best_result.get("fda_number"),
"recommendations": ["Device must have FDA 510(k) clearance or PMA approval"]
}
# Check confidence
if best_result["confidence_score"] < 0.3:
return {
"approved_for_pa": False,
"reason": "Low confidence match - unable to verify exact device",
"fda_number": best_result.get("fda_number"),
"confidence": best_result["confidence_score"],
"recommendations": ["Request specific device model number or FDA submission number"]
}
# Check device class requirement
if required_class and best_result.get("device_class") != required_class:
return {
"approved_for_pa": False,
"reason": f"Device is Class {best_result.get('device_class')} - Class {required_class} required",
"fda_number": best_result.get("fda_number"),
"recommendations": [f"Policy requires Class {required_class} device"]
}
# Check OTC requirement
if require_otc and not best_result.get("is_otc"):
return {
"approved_for_pa": False,
"reason": "Device is not classified as OTC - OTC device required",
"fda_number": best_result.get("fda_number"),
"recommendations": ["Policy requires Over-The-Counter (OTC) device"]
}
# Device meets requirements
return {
"approved_for_pa": True,
"reason": f"Device has valid FDA {best_result['status']}",
"fda_number": best_result.get("fda_number"),
"device_name": best_result.get("device_name"),
"manufacturer": best_result.get("manufacturer"),
"decision_date": best_result.get("decision_date"),
"device_class": best_result.get("device_class"),
"is_otc": best_result.get("is_otc"),
"confidence": best_result["confidence_score"],
"match_method": best_result.get("match_method"),
"notes": best_result.get("notes", [])
}
@mcp.tool
async def get_device_classification(product_code: str) -> Dict[str, Any]:
"""
Get detailed FDA device classification information for a product code.
Retrieves regulation details, device class, and regulatory requirements
for a specific FDA product classification code.
Args:
product_code: FDA product classification code (e.g., "QUH")
Returns:
Classification details including regulation number, device name, and requirements
"""
params = {"search": f'product_code:"{product_code}"', "limit": "1"}
response = await make_fda_request("/device/classification.json", params)
if response and response.get("results"):
record = response["results"][0]
# Determine if it's an OTC product code
otc_codes = ["QUF", "QUG", "QUH", "SCR"]
is_otc = product_code in otc_codes
return {
"product_code": product_code,
"device_name": record.get("device_name", ""),
"device_class": record.get("device_class", ""),
"regulation_number": record.get("regulation_number", ""),
"medical_specialty": record.get("medical_specialty_description", ""),
"review_panel": record.get("review_panel", ""),
"is_otc": is_otc,
"implant_flag": record.get("implant_flag", "N") == "Y",
"life_sustain_support_flag": record.get("life_sustain_support_flag", "N") == "Y",
"definition": record.get("definition", ""),
"physical_state": record.get("physical_state", ""),
"technical_method": record.get("technical_method", ""),
"target_area": record.get("target_area", "")
}
return {
"product_code": product_code,
"error": "Product code not found in FDA classification database"
}
@mcp.tool
async def get_device_requests_from_fhir(patient_id: str) -> List[Dict[str, Any]]:
"""
Fetch DeviceRequest resources for a patient from FHIR server.
Retrieves all DeviceRequest resources for a specific patient and extracts
comprehensive device information including FDA details from parameters.
Args:
patient_id: FHIR patient ID
Returns:
List of DeviceRequest resources with extracted device information
"""
try:
async with aiohttp.ClientSession() as session:
url = f"{FHIR_BASE}/fhir/DeviceRequest"
params = {"patient": patient_id, "_sort": "-date"}
async with session.get(url, params=params) as response:
if response.status != 200:
logger.error(f"Failed to fetch DeviceRequests: {response.status}")
return []
bundle = await response.json()
device_requests = []
for entry in bundle.get("entry", []):
resource = entry.get("resource", {})
# Extract comprehensive device information
device_info = {
"id": resource.get("id"),
"status": resource.get("status"),
"authored_on": resource.get("authoredOn"),
"device_name": None,
"manufacturer": None,
"fda_510k": None,
"is_otc": None,
"product_code": None,
"notes": []
}
# Extract device name from codeCodeableConcept
if "codeCodeableConcept" in resource:
code = resource["codeCodeableConcept"]
# Prioritize text over coding display
device_info["device_name"] = code.get("text") or (
code.get("coding", [{}])[0].get("display") if code.get("coding") else None
)
# Extract manufacturer from performer
if "performer" in resource:
device_info["manufacturer"] = resource["performer"].get("display")
# Extract FDA and OTC info from parameters
for param in resource.get("parameter", []):
param_code = param.get("code", {}).get("coding", [{}])[0].get("code", "")
if param_code == "fda-510k":
# Extract FDA 510k number
value_concept = param.get("valueCodeableConcept", {})
device_info["fda_510k"] = value_concept.get("text") or (
value_concept.get("coding", [{}])[0].get("code") if value_concept.get("coding") else None
)
elif param_code == "otc-status":
device_info["is_otc"] = param.get("valueBoolean")
elif param_code == "product-code":
value_concept = param.get("valueCodeableConcept", {})
device_info["product_code"] = value_concept.get("text") or (
value_concept.get("coding", [{}])[0].get("code") if value_concept.get("coding") else None
)
# Extract notes for additional context
for note in resource.get("note", []):
note_text = note.get("text", "")
device_info["notes"].append(note_text)
# Try to extract FDA info from notes if not in parameters
if not device_info["fda_510k"] and "K" in note_text:
# Look for K-number pattern (e.g., K223137)
import re
k_match = re.search(r'K\d{6}', note_text)
if k_match:
device_info["fda_510k"] = k_match.group()
# Extract manufacturer from notes if not found
if not device_info["manufacturer"] and "manufacturer" in note_text.lower():
# Try to extract manufacturer name from notes
if "Lexie" in note_text or "hearX" in note_text:
device_info["manufacturer"] = "hearX SA (Pty) Ltd."
device_requests.append(device_info)
return device_requests
except Exception as e:
logger.error(f"Failed to fetch DeviceRequests from FHIR: {e}")
return []
@mcp.tool
async def select_best_device_request(device_requests: List[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
"""
Select the best DeviceRequest from a list based on information completeness.
Prioritizes:
1. Requests with FDA 510k numbers
2. Requests with specific device names (not generic)
3. Requests with manufacturer info
4. Most recent requests
Args:
device_requests: List of device request dictionaries
Returns:
The most complete device request or None if empty
"""
if not device_requests:
return None
def score_device_request(req: Dict) -> int:
"""Score a device request based on information completeness."""
score = 0
# FDA 510k is most important for PA
if req.get("fda_510k"):
score += 100
# Specific device name vs generic
device_name = req.get("device_name", "").lower()
if device_name:
# Specific brands get higher scores
specific_brands = ["lexie", "phonak", "widex", "oticon", "resound", "signia", "starkey"]
if any(brand in device_name for brand in specific_brands):
score += 50
elif device_name != "hearing aid" and device_name != "digital hearing aid":
score += 30
else:
score += 10
# Manufacturer info
if req.get("manufacturer"):
score += 20
# OTC status specified
if req.get("is_otc") is not None:
score += 10
# Has notes with details
if req.get("notes"):
score += 5 * len(req.get("notes", []))
return score
# Sort by score (highest first)
sorted_requests = sorted(device_requests, key=score_device_request, reverse=True)
return sorted_requests[0] if sorted_requests else None
