src/hermes/mcp/server.py

"""MCP Server implementation for the Hermes platform."""

from __future__ import annotations

import json
import logging
from typing import Any

from hermes.config.settings import get_settings
from hermes.tools.base.registry import tool_registry

logger = logging.getLogger(__name__)


class MCPServer:
    """Model Context Protocol server implementation."""

    def __init__(self) -> None:
        self.settings = get_settings()
        self._initialized = False

    async def initialize(self) -> None:
        """Initialize the MCP server."""
        from hermes.tools.register import register_default_tools

        register_default_tools()
        self._initialized = True
        logger.info("MCP server initialized")

    async def handle_request(self, request: dict[str, Any]) -> dict[str, Any]:
        """Handle an MCP protocol request."""
        method = request.get("method", "")
        params = request.get("params", {})
        request_id = request.get("id")

        try:
            if method == "initialize":
                return await self._handle_initialize(params, request_id)
            elif method == "tools/list":
                return await self._handle_list_tools(request_id)
            elif method == "tools/call":
                return await self._handle_call_tool(params, request_id)
            elif method == "resources/list":
                return await self._handle_list_resources(request_id)
            elif method == "resources/read":
                return await self._handle_read_resource(params, request_id)
            else:
                return self._error_response(request_id, -32601, f"Method not found: {method}")

        except Exception as e:
            logger.error(f"MCP request error: {e}")
            return self._error_response(request_id, -32603, str(e))

    async def _handle_initialize(
        self, params: dict[str, Any], request_id: str | None
    ) -> dict[str, Any]:
        """Handle initialization request."""
        return {
            "jsonrpc": "2.0",
            "id": request_id,
            "result": {
                "protocolVersion": "2024-11-05",
                "capabilities": {
                    "tools": {"listChanged": True},
                    "resources": {"listChanged": True},
                },
                "serverInfo": {
                    "name": "hermes-mcp-server",
                    "version": "1.0.0",
                },
            },
        }

    async def _handle_list_tools(self, request_id: str | None) -> dict[str, Any]:
        """Handle tools/list request."""
        schemas = tool_registry.list_schemas()
        return {
            "jsonrpc": "2.0",
            "id": request_id,
            "result": {"tools": schemas},
        }

    async def _handle_call_tool(
        self, params: dict[str, Any], request_id: str | None
    ) -> dict[str, Any]:
        """Handle tools/call request."""
        tool_name = params.get("name", "")
        arguments = params.get("arguments", {})

        tool = tool_registry.get(tool_name)
        if not tool:
            return self._error_response(request_id, -32602, f"Tool not found: {tool_name}")

        try:
            result = await tool.execute(**arguments)
            content = json.dumps(result, default=str) if isinstance(result, dict) else str(result)

            return {
                "jsonrpc": "2.0",
                "id": request_id,
                "result": {
                    "content": [{"type": "text", "text": content}],
                    "isError": False,
                },
            }
        except Exception as e:
            return {
                "jsonrpc": "2.0",
                "id": request_id,
                "result": {
                    "content": [{"type": "text", "text": str(e)}],
                    "isError": True,
                },
            }

    async def _handle_list_resources(self, request_id: str | None) -> dict[str, Any]:
        """Handle resources/list request."""
        return {
            "jsonrpc": "2.0",
            "id": request_id,
            "result": {
                "resources": [
                    {
                        "uri": "hermes://tools",
                        "name": "Available Tools",
                        "description": "List of all available tools",
                        "mimeType": "application/json",
                    },
                    {
                        "uri": "hermes://status",
                        "name": "Server Status",
                        "description": "Current server status",
                        "mimeType": "application/json",
                    },
                ]
            },
        }

    async def _handle_read_resource(
        self, params: dict[str, Any], request_id: str | None
    ) -> dict[str, Any]:
        """Handle resources/read request."""
        uri = params.get("uri", "")

        if uri == "hermes://tools":
            data = json.dumps(tool_registry.list_schemas(), indent=2)
            return {
                "jsonrpc": "2.0",
                "id": request_id,
                "result": {
                    "contents": [
                        {"uri": uri, "mimeType": "application/json", "text": data}
                    ]
                },
            }
        elif uri == "hermes://status":
            data = json.dumps(
                {"status": "running", "tools": len(tool_registry.list_tools())},
                indent=2,
            )
            return {
                "jsonrpc": "2.0",
                "id": request_id,
                "result": {
                    "contents": [
                        {"uri": uri, "mimeType": "application/json", "text": data}
                    ]
                },
            }

        return self._error_response(request_id, -32602, f"Resource not found: {uri}")

    def _error_response(
        self, request_id: str | None, code: int, message: str
    ) -> dict[str, Any]:
        """Create an error response."""
        return {
            "jsonrpc": "2.0",
            "id": request_id,
            "error": {"code": code, "message": message},
        }


mcp_server = MCPServer()