AACT Clinical Trials MCP Server

  • src
  • mcp_server_aact
import logging from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationOptions import mcp.server.stdio from mcp_server_aact.database import AACTDatabase from mcp_server_aact.handlers import MCPHandlers from mcp.types import LoggingLevel, EmptyResult import json from pathlib import Path logger = logging.getLogger('mcp_aact_server') logger.setLevel(logging.DEBUG) class AACTServer(Server): def __init__(self): super().__init__("aact-manager") self.db = AACTDatabase() # Load the schema resource schema_path = Path(__file__).parent / "resources" / "database_schema.json" with open(schema_path) as f: self.schema = json.load(f) # Pass schema to handlers self.handlers = MCPHandlers(self.db, self.schema) self._register_handlers() # Set up logging handler that sends to MCP client self.log_handler = MCPLogHandler(self) logger.addHandler(self.log_handler) def _register_handlers(self): @self.list_resources() async def handle_list_resources(): return await self.handlers.handle_list_resources() @self.read_resource() async def handle_read_resource(uri): return await self.handlers.handle_read_resource(uri) @self.list_prompts() async def handle_list_prompts(): return await self.handlers.handle_list_prompts() @self.get_prompt() async def handle_get_prompt(name, arguments): return await self.handlers.handle_get_prompt(name, arguments) @self.list_tools() async def handle_list_tools(): return await self.handlers.handle_list_tools() @self.call_tool() async def handle_call_tool(name, arguments): return await self.handlers.handle_call_tool(name, arguments) @self.set_logging_level() async def handle_set_logging_level(level: LoggingLevel) -> EmptyResult: """Handle requests to change the logging level""" logger.info(f"Setting logging level to {level}") logging.getLogger('mcp_aact_server').setLevel(level.upper()) return EmptyResult() class MCPLogHandler(logging.Handler): def __init__(self, server): super().__init__() self.server = server def emit(self, record): try: msg = self.format(record) # Just log to standard logger instead of trying to send to session logger.log(record.levelno, msg) except Exception: self.handleError(record) async def main(): try: server = AACTServer() async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_name="aact", server_version="0.1.0", capabilities=server.get_capabilities( notification_options=NotificationOptions(), experimental_capabilities={}, ), ), ) except Exception as e: logger.error(f"Server error: {str(e)}", exc_info=True) raise