Spec3 MCP Server

"""HTTP server version of the MCP server for network access from Windows host.""" import sys import logging import os import uvicorn from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse # Add the src directory to Python path for relative imports sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..')) from spec3_mcp_server.server import create_server # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) logger = logging.getLogger(__name__) # Create FastAPI app app = FastAPI(title="Spec3 MCP Server HTTP", version="0.1.0") # Add CORS middleware for cross-origin requests app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # Initialize MCP server mcp_server = create_server() @app.get("/health") async def health_check(): """Health check endpoint.""" return {"status": "healthy", "service": "spec3-mcp-server"} @app.post("/") async def mcp_endpoint(request: Request): """ Main MCP endpoint for HTTP-based connections. This allows Claude Desktop on Windows to connect via HTTP to WSL. """ try: body = await request.json() logger.info(f"Received MCP request: {body}") # Process the MCP request using FastMCP # Note: This is a simplified HTTP wrapper - FastMCP is designed for stdio method = body.get("method", "") if method == "initialize": request_id = body.get("id") response = { "jsonrpc": "2.0", "id": request_id, "result": { "protocolVersion": "2024-11-05", "capabilities": { "tools": { "listChanged": False } }, "serverInfo": { "name": "spec3-mcp-server", "version": "0.1.0" } } } elif method == "notifications/initialized": # No response needed for notifications return JSONResponse(content={}) elif method == "tools/list": request_id = body.get("id") # Return the available tools tools = [ { "name": "echo_message", "description": "Echo back a message with additional formatting.", "inputSchema": { "type": "object", "properties": { "message": { "type": "string", "description": "The message to echo back" } }, "required": ["message"] } }, { "name": "get_server_info", "description": "Get comprehensive information about the MCP server.", "inputSchema": { "type": "object", "properties": {}, "required": [] } }, { "name": "list_available_tools", "description": "List all available tools with their descriptions.", "inputSchema": { "type": "object", "properties": {}, "required": [] } } ] response = { "jsonrpc": "2.0", "id": request_id, "result": { "tools": tools } } elif method == "tools/call": request_id = body.get("id") params = body.get("params", {}) tool_name = params.get("name", "") arguments = params.get("arguments", {}) try: # Execute the tool using FastMCP if tool_name == "echo_message": message = arguments.get("message", "") result = f"Echo: {message} (processed by Spec3 MCP Server)" elif tool_name == "get_server_info": result = { "name": "Spec3 MCP Server", "version": "0.1.0", "description": "A Model Context Protocol server for Spec3 tools and services", "capabilities": [ "Basic messaging", "Utility functions", "Server information" ], "status": "running", "tools_count": 3 } elif tool_name == "list_available_tools": result = [ {"name": "echo_message", "description": "Echoes back a message with formatting"}, {"name": "get_server_info", "description": "Returns comprehensive server information"}, {"name": "list_available_tools", "description": "Lists all available tools and their descriptions"} ] else: raise ValueError(f"Unknown tool: {tool_name}") response = { "jsonrpc": "2.0", "id": request_id, "result": { "content": [ { "type": "text", "text": str(result) if not isinstance(result, str) else result } ] } } except Exception as e: logger.error(f"Error calling tool {tool_name}: {e}") response = { "jsonrpc": "2.0", "id": request_id, "error": { "code": -32603, "message": f"Internal error: {str(e)}" } } else: response = { "jsonrpc": "2.0", "id": body.get("id"), "error": { "code": -32601, "message": f"Method not found: {method}" } } logger.info(f"Sending MCP response: {response}") return JSONResponse(content=response) except Exception as e: logger.error(f"Error processing request: {e}") return JSONResponse( status_code=500, content={ "jsonrpc": "2.0", "id": None, "error": { "code": -32603, "message": f"Internal error: {str(e)}" } } ) def main(): """Run the HTTP server.""" logger.info("Starting Spec3 MCP Server HTTP interface...") uvicorn.run( "http_server:app", host="0.0.0.0", # Listen on all interfaces for Windows host access port=8000, log_level="info", access_log=True ) if __name__ == "__main__": main()