FHIR Progressive Discovery MCP Server

""" MCP Client Test - Simulates an actual MCP client connecting to the server This uses the MCP SDK to interact with the server properly """ import asyncio import json from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def test_mcp_connection(): """Test actual MCP connection to the server""" print("🔌 Connecting to FHIR MCP Server...") # Server parameters server_params = StdioServerParameters(command="python3", args=["fhir_mcp_server.py"]) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: # Initialize the connection await session.initialize() print("✅ Connected to MCP Server\n") # List available tools print("📋 Listing available tools...") tools = await session.list_tools() print(f"\n🔧 Found {len(tools.tools)} tools:") for tool in tools.tools: print(f" - {tool.name}: {tool.description[:80]}...") # Test 1: Discover resource groups print("\n" + "=" * 70) print("TEST 1: Progressive Discovery") print("=" * 70) result = await session.call_tool( "discover_resource_groups", arguments={"intent": "patient clinical care"} ) print("\n📊 Discovered Groups:") for content in result.content: if hasattr(content, "text"): data = json.loads(content.text) for group in data.get("discovered_groups", [])[:3]: print( f" • {group['name']}: {group['resource_count']} resources (score: {group['relevance_score']})" ) # Test 2: Explore a group print("\n" + "=" * 70) print("TEST 2: Explore Clinical Group") print("=" * 70) result = await session.call_tool( "explore_resource_group", arguments={"group_id": "clinical"} ) print("\n📊 Clinical Resources:") for content in result.content: if hasattr(content, "text"): data = json.loads(content.text) print(f" Resources: {', '.join(data.get('resources', []))}") # Test 3: Create a Patient print("\n" + "=" * 70) print("TEST 3: Create Patient Resource") print("=" * 70) patient_data = { "resourceType": "Patient", "active": True, "name": [{"use": "official", "family": "TestPatient", "given": ["MCP", "Test"]}], "gender": "other", "birthDate": "2000-01-01", } result = await session.call_tool( "create_resource", arguments={"resource_type": "Patient", "resource_data": patient_data}, ) print("\n📊 Created Patient:") for content in result.content: if hasattr(content, "text"): data = json.loads(content.text) if data.get("success"): print(f" ✅ Success! Resource ID: {data.get('id')}") print(f" Resource Type: {data.get('resource_type')}") else: print(f" ❌ Failed: {data.get('error')}") # Test 4: Get Resource Schema print("\n" + "=" * 70) print("TEST 4: Get Observation Schema") print("=" * 70) result = await session.call_tool( "get_resource_schema", arguments={"resource_type": "Observation"} ) print("\n📊 Schema Info:") for content in result.content: if hasattr(content, "text"): data = json.loads(content.text) print(f" Resource: {data.get('resource_type')}") print(f" FHIR Version: {data.get('fhir_version')}") print( f" Properties: {len(data.get('schema', {}).get('properties', {}))} fields" ) # Test 5: Version Info print("\n" + "=" * 70) print("TEST 5: FHIR Version Info") print("=" * 70) result = await session.call_tool("get_fhir_version_info", arguments={}) print("\n📊 Version Info:") for content in result.content: if hasattr(content, "text"): data = json.loads(content.text) print(f" Default: {data.get('default_version')}") print(f" Supported: {', '.join(data.get('supported_versions', []))}") print(f" Features: {len(data.get('features', []))} available") print("\n" + "=" * 70) print("✅ ALL MCP CLIENT TESTS PASSED!") print("=" * 70) async def main(): """Main entry point""" print("\n" + "🎯" * 35) print(" FHIR MCP SERVER - MCP CLIENT TEST") print("🎯" * 35 + "\n") try: await test_mcp_connection() print("\n💡 What this proves:") print(" ✓ MCP server starts correctly") print(" ✓ All tools are properly registered") print(" ✓ Progressive discovery works via MCP protocol") print(" ✓ Resources can be created and validated") print(" ✓ Ready for Claude Desktop integration!") except Exception as e: print(f"\n❌ Test failed: {e}") import traceback traceback.print_exc() if __name__ == "__main__": asyncio.run(main())