File size: 3,753 Bytes
76962bf b1198f0 76962bf b1198f0 76962bf b1198f0 76962bf b1198f0 76962bf b1198f0 76962bf | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | """
MCP Server implementation for Medical CDM Tools.
This server exposes the TrendAnalyzer and FHIR tools to any MCP-compliant client.
"""
import sys
import json
import asyncio
from typing import Any, Dict, List
import src.tools.fhir_memory as fhir_tools
import src.tools.dietary_tools as dietary_tools
from langchain_core.tools import BaseTool
from src.agents.cdm_agents import TrendAnalyzer
from src.utils.logger import setup_logger
logger = setup_logger("MCPServer")
class MedicalMCPServer:
def __init__(self):
self.trend_analyzer = TrendAnalyzer()
self.tools = {}
for module in [fhir_tools, dietary_tools]:
for name in dir(module):
obj = getattr(module, name)
if isinstance(obj, BaseTool):
self.tools[obj.name] = obj
async def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
method = request.get("method")
params = request.get("params", {})
req_id = request.get("id")
logger.info(f"Received MCP request: method={method}, id={req_id}")
try:
if method == "list_tools":
result = self.list_tools()
elif method == "call_tool":
result = await self.call_tool(params.get("name"), params.get("arguments", {}))
else:
return {"jsonrpc": "2.0", "error": {"code": -32601, "message": "Method not found"}, "id": req_id}
return {"jsonrpc": "2.0", "result": result, "id": req_id}
except Exception as e:
return {"jsonrpc": "2.0", "error": {"code": -32603, "message": str(e)}, "id": req_id}
def list_tools(self) -> List[Dict[str, Any]]:
tool_list = [
{
"name": "analyze_health_trends",
"description": "Analyze FHIR observation trends for a patient.",
"inputSchema": {
"type": "object",
"properties": {
"patient_id": {"type": "string"}
},
"required": ["patient_id"]
}
}
]
for name, tool_obj in self.tools.items():
schema = {}
if tool_obj.args_schema:
try:
schema = tool_obj.args_schema.schema()
except AttributeError:
schema = tool_obj.args_schema.model_json_schema()
tool_list.append({
"name": name,
"description": tool_obj.description,
"inputSchema": schema
})
return tool_list
async def call_tool(self, name: str, args: Dict[str, Any]) -> Any:
logger.info(f"Calling MCP tool: {name}")
if name == "analyze_health_trends":
return await self.trend_analyzer.analyze_trends(args["patient_id"])
elif name in self.tools:
return self.tools[name].invoke(args)
else:
raise ValueError(f"Unknown tool: {name}")
async def main():
server = MedicalMCPServer()
# Simple stdio loop for MCP
while True:
line = await asyncio.get_event_loop().run_in_executor(None, sys.stdin.readline)
if not line:
break
try:
request = json.loads(line)
response = await server.handle_request(request)
print(json.dumps(response), flush=True)
except Exception as e:
print(json.dumps({"error": str(e)}), flush=True)
if __name__ == "__main__":
if len(sys.argv) > 1 and sys.argv[1] == "--serve":
asyncio.run(main())
else:
print("Medical MCP Server. Use --serve to start in stdio mode.")
|