FHIR Progressive Discovery MCP Server

""" Interactive test for the FHIR MCP Server Simulates how an MCP client would interact with the server """ import json from fhir_mcp_server import mcp def print_section(title): """Print a formatted section header""" print("\n" + "=" * 70) print(f" {title}") print("=" * 70) def print_result(result): """Pretty print a result""" print(json.dumps(result, indent=2)) def demo_progressive_discovery(): """Demonstrate the progressive discovery flow""" print_section("PROGRESSIVE DISCOVERY DEMO") # Step 1: User has intent print("\n📋 Scenario: User wants to work with patient clinical data") print(" LLM calls: discover_resource_groups(intent='patient clinical data')") result = mcp.call_tool("discover_resource_groups", {"intent": "patient clinical data"}) print("\n📊 Result:") print_result(result) # Step 2: Explore a group print("\n\n📋 User wants to explore the 'clinical' group") print(" LLM calls: explore_resource_group(group_id='clinical')") result = mcp.call_tool("explore_resource_group", {"group_id": "clinical"}) print("\n📊 Result:") print_result(result) # Step 3: Get specific resource schema print("\n\n📋 User wants to create an Observation") print(" LLM calls: get_resource_schema(resource_type='Observation')") result = mcp.call_tool("get_resource_schema", {"resource_type": "Observation"}) print("\n📊 Result (truncated for readability):") if "schema" in result: schema_summary = { "resource_type": result["resource_type"], "fhir_version": result["fhir_version"], "required_fields": result["required_fields"], "properties_count": len(result["schema"].get("properties", {})), } print_result(schema_summary) def demo_create_patient(): """Demonstrate creating a patient resource""" print_section("CREATE PATIENT RESOURCE DEMO") print("\n📋 Creating a new Patient resource") patient_data = { "resourceType": "Patient", "active": True, "name": [{"use": "official", "family": "Smith", "given": ["Jane", "Marie"]}], "gender": "female", "birthDate": "1985-07-15", "telecom": [{"system": "phone", "value": "555-0123", "use": "mobile"}], "address": [ { "use": "home", "line": ["123 Main St"], "city": "Springfield", "state": "IL", "postalCode": "62701", "country": "US", } ], } print("\n📝 Patient data:") print_result(patient_data) print("\n🔧 Calling: create_resource(resource_type='Patient', resource_data=...)") result = mcp.call_tool( "create_resource", {"resource_type": "Patient", "resource_data": patient_data} ) print("\n📊 Result:") print_result(result) return result.get("resource") def demo_create_observation(patient_resource): """Demonstrate creating an observation""" print_section("CREATE OBSERVATION RESOURCE DEMO") print("\n📋 Creating a Blood Pressure observation") observation_data = { "resourceType": "Observation", "status": "final", "category": [ { "coding": [ { "system": "http://terminology.hl7.org/CodeSystem/observation-category", "code": "vital-signs", "display": "Vital Signs", } ] } ], "code": { "coding": [ {"system": "http://loinc.org", "code": "85354-9", "display": "Blood pressure panel"} ] }, "subject": { "reference": f"Patient/{patient_resource.get('id', '123')}" if patient_resource else "Patient/123" }, "effectiveDateTime": "2024-01-15T10:30:00Z", "component": [ { "code": { "coding": [ { "system": "http://loinc.org", "code": "8480-6", "display": "Systolic blood pressure", } ] }, "valueQuantity": { "value": 120, "unit": "mmHg", "system": "http://unitsofmeasure.org", "code": "mm[Hg]", }, }, { "code": { "coding": [ { "system": "http://loinc.org", "code": "8462-4", "display": "Diastolic blood pressure", } ] }, "valueQuantity": { "value": 80, "unit": "mmHg", "system": "http://unitsofmeasure.org", "code": "mm[Hg]", }, }, ], } print("\n📝 Observation data:") print_result(observation_data) print("\n🔧 Calling: create_resource(resource_type='Observation', resource_data=...)") result = mcp.call_tool( "create_resource", {"resource_type": "Observation", "resource_data": observation_data} ) print("\n📊 Result:") print_result(result) return result.get("resource") def demo_format_conversion(): """Demonstrate format conversion""" print_section("FORMAT CONVERSION DEMO") print("\n📋 Converting a simple Patient resource to XML") patient_json = json.dumps( { "resourceType": "Patient", "id": "example-xml", "gender": "male", "birthDate": "1990-01-01", "name": [{"family": "Doe", "given": ["John"]}], } ) print("\n🔧 Calling: convert_between_formats(input_format='json', output_format='xml')") result = mcp.call_tool( "convert_between_formats", {"resource_data": patient_json, "input_format": "json", "output_format": "xml"}, ) print("\n📊 XML Output:") if result.get("success"): print(result["output"]) else: print(f"Error: {result.get('error')}") def demo_search_validation(): """Demonstrate search parameter validation""" print_section("SEARCH PARAMETER VALIDATION DEMO") print("\n📋 Validating search parameters for Patient search") print(" Search params: name=Smith, birthdate=1985-07-15, gender=female") result = mcp.call_tool( "search_resources", { "resource_type": "Patient", "search_params": { "name": "Smith", "birthdate": "1985-07-15", "gender": "female", "invalid_param": "test", # This should be flagged as invalid }, }, ) print("\n📊 Result:") print_result(result) def demo_list_all_groups(): """Show all available resource groups""" print_section("ALL RESOURCE GROUPS") print("\n📋 Listing all available FHIR resource groups") result = mcp.call_tool("list_all_resource_groups", {}) print("\n📊 Result:") print_result(result) def demo_version_info(): """Show FHIR version information""" print_section("FHIR VERSION INFO") result = mcp.call_tool("get_fhir_version_info", {}) print("\n📊 Result:") print_result(result) def main(): """Run all demos""" print("\n" + "🎯" * 35) print(" FHIR PROGRESSIVE DISCOVERY MCP SERVER") print(" Interactive Test & Demo") print("🎯" * 35) try: # Demo 1: Progressive Discovery demo_progressive_discovery() # Demo 2: Create Patient patient = demo_create_patient() # Demo 3: Create Observation linked to Patient demo_create_observation(patient) # Demo 4: Format Conversion demo_format_conversion() # Demo 5: Search Validation demo_search_validation() # Demo 6: List All Groups demo_list_all_groups() # Demo 7: Version Info demo_version_info() print("\n" + "=" * 70) print("✅ ALL DEMOS COMPLETED SUCCESSFULLY!") print("=" * 70) print("\n💡 Key Takeaways:") print(" 1. Progressive discovery reduces context window usage") print(" 2. Resources are grouped by clinical domain") print(" 3. All FHIR resources are validated with Pydantic") print(" 4. Format conversion supports JSON, XML, and YAML") print(" 5. Ready for MCP client integration!") except Exception as e: print(f"\n❌ Demo failed: {e}") import traceback traceback.print_exc() if __name__ == "__main__": main()