test_fda_mcp.py•8.31 kB
#!/usr/bin/env python3
"""Test the FDA MCP server using the official FastMCP client"""
import asyncio
import json
from fastmcp.client import Client
from typing import Any, Dict
FDA_MCP_URL = "https://fda-mcp-davinci-mcp.apps.cluster-sdzgj.sdzgj.sandbox319.opentlc.com/mcp/"
async def test_fda_mcp():
"""Test the FDA MCP server tools"""
print("=" * 60)
print("FDA MCP Server Test")
print("=" * 60)
print(f"Server: {FDA_MCP_URL}\n")
try:
async with Client(FDA_MCP_URL) as client:
# List available tools
print("📋 Available Tools:")
print("-" * 40)
tools = await client.list_tools()
for tool in tools:
print(f" • {tool.name}")
if hasattr(tool, 'description') and tool.description:
# Truncate long descriptions
desc = tool.description[:100] + "..." if len(tool.description) > 100 else tool.description
print(f" {desc}")
print(f"\n✅ Total tools: {len(tools)}")
# Test 1: Search for Jabra Enhance Plus by K-number
print("\n" + "=" * 60)
print("Test 1: Search by K-number (Jabra Enhance Plus)")
print("-" * 40)
result = await client.call_tool(
"search_fda_by_identifier",
{"k_number": "K213424"}
)
if result:
print(f"✅ Device Found!")
# Result is a CallToolResult, need to access the content
data = result.content if hasattr(result, 'content') else result
if isinstance(data, dict):
print(f" Name: {data.get('device_name')}")
print(f" Manufacturer: {data.get('manufacturer')}")
print(f" Status: {data.get('status')}")
print(f" FDA Number: {data.get('fda_number')}")
print(f" Decision Date: {data.get('decision_date')}")
print(f" Confidence: {data.get('confidence_score')}")
else:
print(f" Result: {data}")
else:
print("❌ Device not found")
# Test 2: Search by device name with fuzzy matching
print("\n" + "=" * 60)
print("Test 2: Fuzzy search by device name")
print("-" * 40)
results = await client.call_tool(
"search_fda_by_device_info",
{
"device_name": "Jabra hearing aid",
"fuzzy_search": True
}
)
if results:
data = results.content if hasattr(results, 'content') else results
if isinstance(data, list) and data:
print(f"✅ Found {len(data)} potential matches:")
for i, res in enumerate(data[:3], 1):
print(f"\n Match {i}:")
print(f" Name: {res.get('device_name')}")
print(f" Manufacturer: {res.get('manufacturer')}")
print(f" FDA Number: {res.get('fda_number')}")
print(f" Confidence: {res.get('confidence_score')}")
else:
print(f" Result: {data}")
else:
print("❌ No matches found")
# Test 3: Get classification for OTC hearing aid code
print("\n" + "=" * 60)
print("Test 3: Get device classification (QUH - OTC hearing aid)")
print("-" * 40)
classification = await client.call_tool(
"get_device_classification",
{"product_code": "QUH"}
)
if classification:
data = classification.content if hasattr(classification, 'content') else classification
if isinstance(data, dict) and "error" not in data:
print(f"✅ Classification Found:")
print(f" Product Code: {data.get('product_code')}")
print(f" Device Name: {data.get('device_name')}")
print(f" Device Class: {data.get('device_class')}")
print(f" Regulation: {data.get('regulation_number')}")
print(f" Is OTC: {data.get('is_otc')}")
print(f" Medical Specialty: {data.get('medical_specialty')}")
else:
print(f"❌ Classification error: {data.get('error') if isinstance(data, dict) else data}")
else:
print(f"❌ No classification found")
# Test 4: Verify FDA status for PA decision
print("\n" + "=" * 60)
print("Test 4: Verify FDA status for PA workflow")
print("-" * 40)
# First get the device data
device_data = await client.call_tool(
"search_fda_by_identifier",
{"k_number": "K213424"}
)
if device_data:
device_result = device_data.content if hasattr(device_data, 'content') else device_data
# Verify for PA
pa_decision = await client.call_tool(
"verify_fda_status",
{
"fda_results": [device_result],
"require_otc": True
}
)
if pa_decision:
data = pa_decision.content if hasattr(pa_decision, 'content') else pa_decision
if isinstance(data, dict):
print(f"PA Decision:")
print(f" Approved: {data.get('approved_for_pa')}")
print(f" Reason: {data.get('reason')}")
if data.get('recommendations'):
print(f" Recommendations:")
for rec in data['recommendations']:
print(f" • {rec}")
if data.get('notes'):
print(f" Notes:")
for note in data['notes']:
print(f" • {note}")
else:
print(f" Result: {data}")
# Test 5: Test analyze_device_request with sample FHIR DeviceRequest
print("\n" + "=" * 60)
print("Test 5: Analyze FHIR DeviceRequest")
print("-" * 40)
sample_device_request = {
"resourceType": "DeviceRequest",
"codeCodeableConcept": {
"text": "Jabra Enhance Plus hearing aid",
"coding": [{
"system": "http://snomed.info/sct",
"code": "706230002",
"display": "Hearing aid"
}]
},
"note": [{
"text": "Patient requires OTC hearing aid. FDA 510(k) number K213424"
}],
"performer": {
"display": "GN Hearing A/S"
}
}
device_info = await client.call_tool(
"analyze_device_request",
{"device_request": sample_device_request}
)
if device_info:
data = device_info.content if hasattr(device_info, 'content') else device_info
if isinstance(data, dict):
print(f"Extracted Device Info:")
print(f" Device Name: {data.get('device_name')}")
print(f" Manufacturer: {data.get('manufacturer')}")
print(f" K-Number: {data.get('k_number')}")
if data.get('codes'):
print(f" Codes: {len(data['codes'])} code(s) found")
else:
print(f" Result: {data}")
print("\n" + "=" * 60)
print("✅ All tests completed successfully!")
print("=" * 60)
except Exception as e:
print(f"\n❌ Error: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
asyncio.run(test_fda_mcp())