test_updated_patient_50.py•5.3 kB
#!/usr/bin/env python3
"""
Re-test Patient 50's updated DeviceRequest with FDA Verification Agent
"""
import asyncio
import json
import sys
import requests
from pathlib import Path
# Add parent directory to path
sys.path.append(str(Path(__file__).parent))
from agents.fda_verification_agent import FDAVerificationAgent
# FHIR server URL
FHIR_BASE = "https://hapi-fhir-davinci-data.apps.cluster-sdzgj.sdzgj.sandbox319.opentlc.com/fhir"
async def test_updated_device_request():
"""Test the updated DeviceRequest with real FDA data."""
print("=" * 80)
print("Testing Updated Patient 50 DeviceRequest")
print("=" * 80)
print("Patient ID: aud-patient-00050")
print("DeviceRequest ID: device-request-hearing-aid-00050")
print("=" * 80)
# Fetch the updated DeviceRequest from FHIR server
print("\nFetching updated DeviceRequest from FHIR server...")
url = f"{FHIR_BASE}/DeviceRequest/device-request-hearing-aid-00050"
response = requests.get(url)
if response.status_code != 200:
print(f"❌ Failed to fetch DeviceRequest: {response.status_code}")
return
device_request = response.json()
print("✅ Retrieved updated DeviceRequest")
# Display key information
print("\n📋 DeviceRequest Details:")
print(f" Device: {device_request['codeCodeableConcept']['text']}")
# Show FDA parameters
if 'parameter' in device_request:
print("\n FDA Information:")
for param in device_request['parameter']:
code = param.get('code', {}).get('coding', [{}])[0].get('display', 'Unknown')
if 'valueCodeableConcept' in param:
value = param['valueCodeableConcept'].get('text', 'N/A')
if code == "FDA 510(k) Number":
print(f" • {code}: {value} ✅")
elif 'valueString' in param:
value = param['valueString']
if code == "Manufacturer":
print(f" • {code}: {value}")
elif 'valueBoolean' in param:
value = "Yes" if param['valueBoolean'] else "No"
if code == "Over-The-Counter Status":
print(f" • OTC Status: {value}")
# Initialize FDA agent
import os
api_key = os.getenv("LLAMA_API_KEY", "af78aa99d7ae5bb58f19380ee007277c")
base_url = os.getenv("LLAMA_BASE_URL",
"https://llama-4-scout-17b-16e-w4a16-maas-apicast-production.apps.prod.rhoai.rh-aiservices-bu.com:443/v1")
mcp_url = os.getenv("FDA_MCP_URL", "http://localhost:8090/mcp/")
print("\n🤖 Running FDA Verification Agent...")
print("-" * 60)
agent = FDAVerificationAgent(
mcp_url=mcp_url,
api_key=api_key,
base_url=base_url
)
try:
# Analyze with agent
result = await agent.analyze_device_request(device_request)
# Display results
print(f"\n📊 PA Decision: {result.decision.value.upper()}")
if result.decision.value == "approved":
print(" ✅✅✅ DEVICE APPROVED! ✅✅✅")
print(f"\n💭 Reasoning: {result.reasoning}")
if result.fda_verification:
fda = result.fda_verification
print(f"\n🔍 FDA Verification Results:")
print(f" Device Name: {fda.device_name}")
print(f" FDA Status: {fda.fda_status}")
print(f" FDA Approved/Cleared: {'✅ Yes' if fda.is_fda_approved else '❌ No'}")
if fda.fda_number:
print(f" FDA Number: {fda.fda_number}")
if fda.manufacturer:
print(f" Manufacturer: {fda.manufacturer}")
if fda.device_class:
print(f" Device Class: {fda.device_class}")
if fda.is_otc:
print(f" OTC Status: ✅ Over-the-counter")
print(f" Confidence: {fda.confidence:.0%}")
print(f" Verification Method: {fda.verification_method}")
if result.requirements_met:
print(f"\n✓ Policy Requirements:")
all_met = True
for req, met in result.requirements_met.items():
status = "✅" if met else "❌"
print(f" {status} {req}")
if not met:
all_met = False
if all_met:
print("\n 🎉 ALL REQUIREMENTS MET!")
if result.recommendations:
print(f"\n💡 Recommendations:")
for rec in result.recommendations:
print(f" • {rec}")
print(f"\n🎯 Overall Confidence: {result.confidence:.0%}")
# Final verdict
print("\n" + "=" * 80)
if result.decision.value == "approved":
print("🎉 SUCCESS! The Lexie Lumen hearing aid (K223137) is FDA APPROVED!")
print(" The agent correctly verified the real FDA 510(k) clearance.")
else:
print(f"⚠️ Decision was {result.decision.value.upper()}, not approved as expected.")
print(" Check the reasoning above for details.")
print("=" * 80)
except Exception as e:
print(f"\n❌ Error: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
asyncio.run(test_updated_device_request())