atlas/home-voice-agent/mcp-server/server/mcp_server.py

#!/usr/bin/env python3
"""
MCP Server - Model Context Protocol implementation.

This server exposes tools via JSON-RPC 2.0 protocol.
"""

import json
import logging
import sys
from pathlib import Path
from typing import Any, Dict, List, Optional
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse, Response, HTMLResponse
from pydantic import BaseModel

# Add parent directory to path to import tools
# This allows running from mcp-server/ directory
parent_dir = Path(__file__).parent.parent
if str(parent_dir) not in sys.path:
    sys.path.insert(0, str(parent_dir))
from tools.registry import ToolRegistry

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

app = FastAPI(title="MCP Server", version="0.1.0")

# Initialize tool registry
tool_registry = ToolRegistry()


class JSONRPCRequest(BaseModel):
    """JSON-RPC 2.0 request model."""
    jsonrpc: str = "2.0"
    method: str
    params: Optional[Dict[str, Any]] = None
    id: Optional[Any] = None


class JSONRPCResponse(BaseModel):
    """JSON-RPC 2.0 response model."""
    jsonrpc: str = "2.0"
    result: Optional[Any] = None
    error: Optional[Dict[str, Any]] = None
    id: Optional[Any] = None


def create_error_response(
    code: int,
    message: str,
    data: Optional[Any] = None,
    request_id: Optional[Any] = None
) -> JSONRPCResponse:
    """Create a JSON-RPC error response."""
    error = {"code": code, "message": message}
    if data is not None:
        error["data"] = data
    
    return JSONRPCResponse(
        jsonrpc="2.0",
        error=error,
        id=request_id
    )


def create_success_response(
    result: Any,
    request_id: Optional[Any] = None
) -> JSONRPCResponse:
    """Create a JSON-RPC success response."""
    return JSONRPCResponse(
        jsonrpc="2.0",
        result=result,
        id=request_id
    )


@app.post("/mcp")
async def handle_mcp_request(request: JSONRPCRequest):
    """
    Handle MCP JSON-RPC requests.
    
    Supported methods:
    - tools/list: List all available tools
    - tools/call: Execute a tool
    """
    try:
        method = request.method
        params = request.params or {}
        request_id = request.id
        
        logger.info(f"Received MCP request: method={method}, id={request_id}")
        
        if method == "tools/list":
            # List all available tools
            tools = tool_registry.list_tools()
            return create_success_response({"tools": tools}, request_id)
        
        elif method == "tools/call":
            # Execute a tool
            tool_name = params.get("name")
            arguments = params.get("arguments", {})
            
            if not tool_name:
                return create_error_response(
                    -32602,  # Invalid params
                    "Missing required parameter: name",
                    request_id=request_id
                )
            
            try:
                result = tool_registry.call_tool(tool_name, arguments)
                return create_success_response(result, request_id)
            except ValueError as e:
                # Tool not found or invalid arguments
                return create_error_response(
                    -32602,  # Invalid params
                    str(e),
                    request_id=request_id
                )
            except Exception as e:
                # Tool execution error
                logger.error(f"Tool execution error: {e}", exc_info=True)
                return create_error_response(
                    -32603,  # Internal error
                    "Tool execution failed",
                    data=str(e),
                    request_id=request_id
                )
        
        else:
            # Unknown method
            return create_error_response(
                -32601,  # Method not found
                f"Unknown method: {method}",
                request_id=request_id
            )
    
    except Exception as e:
        logger.error(f"Request handling error: {e}", exc_info=True)
        return create_error_response(
            -32603,  # Internal error
            "Internal server error",
            data=str(e),
            request_id=request.id if hasattr(request, 'id') else None
        )


@app.get("/health")
async def health_check():
    """Health check endpoint."""
    return {
        "status": "healthy",
        "tools_registered": len(tool_registry.list_tools())
    }


@app.get("/")
async def root():
    """Root endpoint with server information."""
    # Get tool count from registry
    try:
        tools = tool_registry.list_tools()
        tool_count = len(tools)
        tool_names = [tool["name"] for tool in tools]
    except Exception as e:
        logger.error(f"Error getting tools: {e}")
        tool_count = 0
        tool_names = []
    
    return {
        "name": "MCP Server",
        "version": "0.1.0",
        "protocol": "JSON-RPC 2.0",
        "status": "running",
        "tools_registered": tool_count,
        "tools": tool_names,
        "endpoints": {
            "mcp": "/mcp",
            "health": "/health",
            "docs": "/docs"
        }
    }


@app.get("/api")
async def api_info():
    """API information endpoint (JSON)."""
    try:
        tools = tool_registry.list_tools()
        tool_count = len(tools)
        tool_names = [tool["name"] for tool in tools]
    except Exception as e:
        logger.error(f"Error getting tools: {e}")
        tool_count = 0
        tool_names = []
    
    return {
        "name": "MCP Server",
        "version": "0.1.0",
        "protocol": "JSON-RPC 2.0",
        "status": "running",
        "tools_registered": tool_count,
        "tools": tool_names,
        "endpoints": {
            "mcp": "/mcp",
            "health": "/health",
            "docs": "/docs"
        }
    }


@app.get("/favicon.ico")
async def favicon():
    """Handle favicon requests - return 204 No Content."""
    return Response(status_code=204)


if __name__ == "__main__":
    import uvicorn
    # Ensure we're running from the mcp-server directory
    import os
    script_dir = Path(__file__).parent.parent
    os.chdir(script_dir)
    uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info")