Epic Patient API MCP Server

"""Epic Patient API MCP Server.""" import os import sys import json from datetime import datetime from dotenv import load_dotenv from mcp.server import Server from mcp.types import Tool, TextContent from .data import PatientDataLoader from .tools import ( handle_list_allergies, handle_get_allergy_details, handle_list_medications, handle_get_medication_details, handle_list_conditions, handle_get_condition_details, handle_list_clinical_notes, handle_get_clinical_note_details, handle_get_attachment, handle_list_labs, handle_get_lab_details, handle_list_vitals, handle_get_vital_details, handle_list_procedures, handle_get_procedure_details, handle_search, ) # Load environment variables load_dotenv() # Initialize server and data loader app = Server("epic-patient-api") loader = PatientDataLoader() def log_request(tool_name: str, arguments: dict, result_preview: str): """Log tool requests in a compact format to stderr.""" timestamp = datetime.now().strftime("%H:%M:%S") patient_id = arguments.get("patient_id", "N/A") # Estimate tokens: ~4 chars per token for English text estimated_tokens = len(result_preview) // 4 # Create a compact argument display extra_args = {k: v for k, v in arguments.items() if k != "patient_id"} args_str = f" {extra_args}" if extra_args else "" # Check if result is an error try: result_data = json.loads(result_preview) if "error" in result_data: status = "❌" else: status = "✓" except: status = "✓" # Log to stderr so it doesn't interfere with MCP stdio protocol print(f"{status} [{timestamp}] {tool_name}({patient_id}{args_str}) → ~{estimated_tokens:,} tokens", file=sys.stderr, flush=True) @app.list_tools() async def list_tools() -> list[Tool]: """List available MCP tools.""" return [ # Allergy tools Tool( name="list_patient_allergies", description="List patient allergies with IDs, allergens, severity, and dates (summary only)", inputSchema={ "type": "object", "properties": { "patient_id": { "type": "string", "description": "Patient identifier (e.g., 'patient_001')" } }, "required": ["patient_id"] } ), Tool( name="get_allergy_details", description="Get detailed information for a specific allergy including reaction", inputSchema={ "type": "object", "properties": { "patient_id": { "type": "string", "description": "Patient identifier (e.g., 'patient_001')" }, "allergy_id": { "type": "string", "description": "Allergy identifier from list_patient_allergies" } }, "required": ["patient_id", "allergy_id"] } ), # Medication tools Tool( name="list_patient_medications", description="List patient medications with IDs, names, status, and date ranges (summary only)", inputSchema={ "type": "object", "properties": { "patient_id": { "type": "string", "description": "Patient identifier (e.g., 'patient_001')" }, "status": { "type": "string", "description": "Optional: Filter by medication status", "enum": ["active", "discontinued", "completed"] } }, "required": ["patient_id"] } ), Tool( name="get_medication_details", description="Get detailed information for a specific medication including instructions and prescriber", inputSchema={ "type": "object", "properties": { "patient_id": { "type": "string", "description": "Patient identifier (e.g., 'patient_001')" }, "medication_id": { "type": "string", "description": "Medication identifier from list_patient_medications" } }, "required": ["patient_id", "medication_id"] } ), # Condition tools Tool( name="list_patient_conditions", description="List patient conditions with IDs, names, status, onset dates, and severity (summary only)", inputSchema={ "type": "object", "properties": { "patient_id": { "type": "string", "description": "Patient identifier (e.g., 'patient_001')" } }, "required": ["patient_id"] } ), Tool( name="get_condition_details", description="Get detailed information for a specific condition including notes and resolved date", inputSchema={ "type": "object", "properties": { "patient_id": { "type": "string", "description": "Patient identifier (e.g., 'patient_001')" }, "condition_id": { "type": "string", "description": "Condition identifier from list_patient_conditions" } }, "required": ["patient_id", "condition_id"] } ), # Clinical note tools Tool( name="list_clinical_notes", description="List clinical notes with IDs, dates, types, providers, and summaries (content not included)", inputSchema={ "type": "object", "properties": { "patient_id": { "type": "string", "description": "Patient identifier (e.g., 'patient_001')" }, "note_type": { "type": "string", "description": "Optional: Filter by note type (e.g., 'Progress Note', 'Discharge Summary')" } }, "required": ["patient_id"] } ), Tool( name="get_clinical_note_details", description="Get full content and attachments for a specific clinical note", inputSchema={ "type": "object", "properties": { "patient_id": { "type": "string", "description": "Patient identifier (e.g., 'patient_001')" }, "note_id": { "type": "string", "description": "Note identifier from list_clinical_notes" } }, "required": ["patient_id", "note_id"] } ), Tool( name="get_note_attachment", description="Retrieve the content of a specific attachment from a clinical note", inputSchema={ "type": "object", "properties": { "patient_id": { "type": "string", "description": "Patient identifier (e.g., 'patient_001')" }, "attachment_id": { "type": "string", "description": "Attachment identifier (e.g., 'att-1')" } }, "required": ["patient_id", "attachment_id"] } ), # Lab tools Tool( name="list_lab_results", description="List lab results with IDs, test names, dates, and ordering providers (results not included)", inputSchema={ "type": "object", "properties": { "patient_id": { "type": "string", "description": "Patient identifier (e.g., 'patient_001')" } }, "required": ["patient_id"] } ), Tool( name="get_lab_result_details", description="Get detailed component values, units, and reference ranges for a specific lab result", inputSchema={ "type": "object", "properties": { "patient_id": { "type": "string", "description": "Patient identifier (e.g., 'patient_001')" }, "lab_id": { "type": "string", "description": "Lab result identifier from list_lab_results" } }, "required": ["patient_id", "lab_id"] } ), # Vitals tools Tool( name="list_vitals", description="List vital signs measurements with IDs and dates (values not included)", inputSchema={ "type": "object", "properties": { "patient_id": { "type": "string", "description": "Patient identifier (e.g., 'patient_001')" } }, "required": ["patient_id"] } ), Tool( name="get_vital_details", description="Get detailed vital signs including blood pressure, heart rate, temperature, weight, etc.", inputSchema={ "type": "object", "properties": { "patient_id": { "type": "string", "description": "Patient identifier (e.g., 'patient_001')" }, "vital_id": { "type": "string", "description": "Vital signs identifier from list_vitals" } }, "required": ["patient_id", "vital_id"] } ), # Procedure tools Tool( name="list_procedures", description="List procedures with IDs, names, dates, and providers (details not included)", inputSchema={ "type": "object", "properties": { "patient_id": { "type": "string", "description": "Patient identifier (e.g., 'patient_001')" } }, "required": ["patient_id"] } ), Tool( name="get_procedure_details", description="Get detailed information for a specific procedure including indication and outcome", inputSchema={ "type": "object", "properties": { "patient_id": { "type": "string", "description": "Patient identifier (e.g., 'patient_001')" }, "procedure_id": { "type": "string", "description": "Procedure identifier from list_procedures" } }, "required": ["patient_id", "procedure_id"] } ), # Search tool Tool( name="search_patient_data", description="Natural language search across all patient data using AI. Use this for complex queries that require understanding context across multiple data types.", inputSchema={ "type": "object", "properties": { "patient_id": { "type": "string", "description": "Patient identifier (e.g., 'patient_001')" }, "query": { "type": "string", "description": "Natural language query about the patient's medical records" } }, "required": ["patient_id", "query"] } ), ] @app.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: """Handle tool calls.""" patient_id = arguments.get("patient_id") # Allergy tools if name == "list_patient_allergies": result = handle_list_allergies(patient_id, loader) elif name == "get_allergy_details": allergy_id = arguments.get("allergy_id") result = handle_get_allergy_details(patient_id, allergy_id, loader) # Medication tools elif name == "list_patient_medications": status = arguments.get("status") result = handle_list_medications(patient_id, loader, status=status) elif name == "get_medication_details": medication_id = arguments.get("medication_id") result = handle_get_medication_details(patient_id, medication_id, loader) # Condition tools elif name == "list_patient_conditions": result = handle_list_conditions(patient_id, loader) elif name == "get_condition_details": condition_id = arguments.get("condition_id") result = handle_get_condition_details(patient_id, condition_id, loader) # Clinical note tools elif name == "list_clinical_notes": note_type = arguments.get("note_type") result = handle_list_clinical_notes(patient_id, loader, note_type=note_type) elif name == "get_clinical_note_details": note_id = arguments.get("note_id") result = handle_get_clinical_note_details(patient_id, note_id, loader) elif name == "get_note_attachment": attachment_id = arguments.get("attachment_id") result = handle_get_attachment(patient_id, attachment_id, loader) # Lab tools elif name == "list_lab_results": result = handle_list_labs(patient_id, loader) elif name == "get_lab_result_details": lab_id = arguments.get("lab_id") result = handle_get_lab_details(patient_id, lab_id, loader) # Vitals tools elif name == "list_vitals": result = handle_list_vitals(patient_id, loader) elif name == "get_vital_details": vital_id = arguments.get("vital_id") result = handle_get_vital_details(patient_id, vital_id, loader) # Procedure tools elif name == "list_procedures": result = handle_list_procedures(patient_id, loader) elif name == "get_procedure_details": procedure_id = arguments.get("procedure_id") result = handle_get_procedure_details(patient_id, procedure_id, loader) # Search tool elif name == "search_patient_data": query = arguments.get("query") result = handle_search(patient_id, query, loader) else: result = f'{{"error": "Unknown tool: {name}"}}' # Log the request log_request(name, arguments, result) return [TextContent(type="text", text=result)] async def main(): """Run the MCP server with SSE transport.""" import uvicorn from starlette.applications import Starlette from starlette.routing import Route, Mount from starlette.responses import Response from mcp.server.sse import SseServerTransport # Print startup banner print("=" * 60) print("🏥 Epic Patient API MCP Server") print("=" * 60) llm_provider = os.getenv("LLM_PROVIDER", "gemini") print(f"LLM Provider: {llm_provider}") print(f"Available patients: patient_001, patient_002, patient_003") port = int(os.getenv("PORT", "8000")) print(f"Server URL: http://localhost:{port}/sse") print("=" * 60, flush=True) # Create SSE transport sse = SseServerTransport("/messages/") async def handle_sse(request): """Handle SSE endpoint.""" async with sse.connect_sse( request.scope, request.receive, request._send ) as streams: await app.run( streams[0], streams[1], app.create_initialization_options() ) return Response() # Create Starlette app web_app = Starlette( debug=True, routes=[ Route("/sse", endpoint=handle_sse, methods=["GET"]), Mount("/messages/", app=sse.handle_post_message), ] ) config = uvicorn.Config( web_app, host="0.0.0.0", port=port, log_level="warning", access_log=False, reload=True, reload_dirs=[os.path.join(os.path.dirname(__file__), "..", "..")] ) server = uvicorn.Server(config) await server.serve() if __name__ == "__main__": import asyncio asyncio.run(main())