#!/usr/bin/env python3 """ MCP Server for Coherence Gateway. This server implements the Model Context Protocol (MCP) to allow Claude Desktop to interact with the Coherence Gateway and execute OpenAPI tools. NOTE: This version uses manual JSON-RPC handling to avoid stdio issues with the MCP library. """ import asyncio import json import logging import os import sys from collections.abc import Sequence from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import ( EmbeddedResource, ImageContent, TextContent, Tool, ) try: from .gateway_client import GatewayClient except ImportError: # Handle case when running directly from gateway_client import GatewayClient # Configure logging function to be called after MCP_MODE is set def setup_logging(): """Setup logging based on MCP_MODE environment variable.""" import tempfile handlers = [] # Always log to a temp file for debugging log_file = tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.log') handlers.append(logging.FileHandler(log_file.name)) # Only use stderr if not in stdio mode (to avoid interfering with MCP protocol) if os.getenv("MCP_MODE") != "stdio": handlers.append(logging.StreamHandler(sys.stderr)) logging.basicConfig( level=os.getenv("LOG_LEVEL", "INFO"), format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=handlers, force=True # Reconfigure if already configured ) return log_file.name # We'll call setup_logging() in main() after setting MCP_MODE logger = logging.getLogger(__name__) # Configuration GATEWAY_URL = os.getenv("GATEWAY_URL", "http://localhost:8000") JWT_SECRET = os.getenv("JWT_SECRET", "55c121aa48dc1b79e3c04c805aabd2fe7646253ab6edd1c61894f8146ce1ed17") class CoherenceMCPServer: """MCP Server implementation for Coherence Gateway.""" def __init__(self): self.gateway_client = GatewayClient(GATEWAY_URL, JWT_SECRET) self.tools = {} async def load_tools(self): """Load MCP tools from gateway discovery endpoint.""" try: # Use httpx for async requests import httpx async with httpx.AsyncClient() as client: response = await client.get( f"{GATEWAY_URL}/mcp/tools/discovery", timeout=10.0 ) if response.status_code == 200: data = response.json() if data.get("tools"): for tool in data["tools"]: # Convert from discovery format to MCP format self.tools[tool["name"]] = { "name": tool["name"], "description": tool.get("description", ""), "input_schema": tool.get("inputSchema", {}) } logger.info(f"Loaded {len(self.tools)} tools from gateway discovery") else: logger.warning("No tools found in gateway discovery response") else: logger.error(f"Failed to fetch tools from gateway: {response.status_code}") except Exception as e: logger.error(f"Error loading tools from gateway: {e}") # Continue without tools rather than crash async def serve() -> None: """Run the MCP server.""" # Initialize server components coherence = CoherenceMCPServer() server = Server("coherence-mcp") logger.info("Starting Coherence MCP Server") logger.info(f"Gateway URL: {GATEWAY_URL}") # Load tools from gateway await coherence.load_tools() @server.list_tools() async def list_tools() -> list[Tool]: """List all available tools.""" # Tools should already be loaded during startup mcp_tools = [] for tool_name, tool_data in coherence.tools.items(): # Convert our tool format to MCP Tool format mcp_tool = Tool( name=tool_name, description=tool_data.get("description", ""), inputSchema=tool_data.get("input_schema", {}), ) mcp_tools.append(mcp_tool) # Don't log during stdio mode to avoid interfering with protocol if os.getenv("MCP_MODE") != "stdio": logger.info(f"Listing {len(mcp_tools)} tools") return mcp_tools @server.call_tool() async def call_tool( name: str, arguments: dict ) -> Sequence[TextContent | ImageContent | EmbeddedResource]: """Execute a tool via the gateway.""" if os.getenv("MCP_MODE") != "stdio": logger.info(f"Executing tool: {name} with args: {arguments}") # Validate tool exists if name not in coherence.tools: error_msg = f"Unknown tool: {name}" if os.getenv("MCP_MODE") != "stdio": logger.error(error_msg) return [TextContent(type="text", text=error_msg)] try: # Execute via gateway result = await coherence.gateway_client.execute_tool(name, arguments) # Format response if result.get("error"): response_text = f"Error: {result['error']}" else: # Extract data from nested response structure data = result if "output" in data: data = data["output"] if isinstance(data, dict) and "data" in data: data = data["data"] # Format based on data type if isinstance(data, list): # List of items (e.g., pets) response_text = json.dumps(data, indent=2) elif isinstance(data, dict): # Single item or structured data response_text = json.dumps(data, indent=2) else: response_text = str(data) if os.getenv("MCP_MODE") != "stdio": logger.info(f"Tool execution completed: {name}") return [TextContent(type="text", text=response_text)] except Exception as e: error_msg = f"Failed to execute tool {name}: {str(e)}" if os.getenv("MCP_MODE") != "stdio": logger.error(error_msg, exc_info=True) return [TextContent(type="text", text=error_msg)] # Create initialization options options = server.create_initialization_options() # Run the server using stdio transport async with stdio_server() as (read_stream, write_stream): await server.run(read_stream, write_stream, options) def main(): """Main entry point.""" # Set stdio mode to suppress stderr logging os.environ["MCP_MODE"] = "stdio" # Setup logging after MCP_MODE is set log_file = setup_logging() try: # Log the temp file location for debugging logger.info(f"MCP Server starting - logs in {log_file}") asyncio.run(serve()) except KeyboardInterrupt: logger.info("Server stopped by user") sys.exit(0) except Exception as e: logger.error(f"Server error: {e}", exc_info=True) sys.exit(1) if __name__ == "__main__": main()