circuitpython_mcp_server.py

# SPDX-FileCopyrightText: 2026
# SPDX-License-Identifier: MIT
"""Minimal MCP server helpers for CircuitPython.

This module implements JSON-RPC MCP routing on top of adafruit_httpserver.
Board-specific code should provide network setup and tool callback functions.
"""

import json

from adafruit_httpserver import DELETE, GET, POST, JSONResponse, Response

try:
    from adafruit_httpserver import SSEResponse
    _SSE_AVAILABLE = True
except ImportError:
    SSEResponse = None
    _SSE_AVAILABLE = False


__version__ = "0.1.0"

DEFAULT_PROTOCOL_VERSIONS = ["2025-06-18", "2025-03-26", "2024-11-05"]

DEFAULT_CORS_HEADERS = {
    "Access-Control-Allow-Origin": "*",
    "Access-Control-Allow-Methods": "GET, POST, DELETE, OPTIONS",
    "Access-Control-Allow-Headers": "Content-Type, Accept, MCP-Protocol-Version",
    "Access-Control-Max-Age": "86400",
}


class MCPError(Exception):
    """JSON-RPC error raised by MCP handlers."""

    def __init__(self, code, message, data=None):
        Exception.__init__(self, message)
        self.code = code
        self.message = message
        self.data = data


class ToolError(MCPError):
    """Invalid tool name or invalid tool arguments."""

    def __init__(self, message, data=None):
        MCPError.__init__(self, -32602, "Invalid params", data or message)


def schema_int(minimum, maximum):
    return {"type": "integer", "minimum": minimum, "maximum": maximum}


def schema_number(minimum, maximum):
    return {"type": "number", "minimum": minimum, "maximum": maximum}


def schema_object(properties=None, required=None, additional_properties=False):
    schema = {
        "type": "object",
        "properties": properties or {},
        "additionalProperties": additional_properties,
    }
    if required:
        schema["required"] = required
    return schema


def tool_result(text, structured_content=None, is_error=False):
    if structured_content is None:
        structured_content = {}
    return {
        "content": [{"type": "text", "text": text}],
        "structuredContent": structured_content,
        "isError": is_error,
    }


def jsonrpc_result(rpc_id, result):
    return {"jsonrpc": "2.0", "result": result, "id": rpc_id}


def jsonrpc_error(rpc_id, code, message, data=None):
    error = {"code": code, "message": message}
    if data is not None:
        error["data"] = data
    return {"jsonrpc": "2.0", "error": error, "id": rpc_id}


def _copy_headers(headers):
    copied = {}
    for key in headers:
        copied[key] = headers[key]
    return copied


class MCPServer:
    """MCP JSON-RPC route adapter for an adafruit_httpserver Server."""

    def __init__(
        self,
        http_server,
        name,
        version,
        title=None,
        instructions="",
        protocol_versions=None,
        cors_headers=None,
        enable_sse=True,
    ):
        self.http_server = http_server
        self.name = name
        self.title = title or name
        self.version = version
        self.instructions = instructions
        self.protocol_versions = protocol_versions or DEFAULT_PROTOCOL_VERSIONS
        self.cors_headers = cors_headers or DEFAULT_CORS_HEADERS
        self.enable_sse = enable_sse
        self.sse_response_class = None
        if enable_sse and _SSE_AVAILABLE:
            self.sse_response_class = SSEResponse
        self.sse_client = None
        self.sse_welcome_pending = False
        self._tools = []
        self._handlers = {}
        self._register_routes()

    def add_tool(self, name, description, input_schema, handler, title=None):
        tool = {
            "name": name,
            "title": title or name,
            "description": description,
            "inputSchema": input_schema or schema_object(),
        }
        self._tools.append(tool)
        self._handlers[name] = handler
        return self

    def start(self, host, port=5000):
        self.http_server.start(host, port=port)

    def poll(self):
        self.http_server.poll()
        self._send_pending_sse_endpoint()

    def mcp_url(self, host, port=5000):
        return "http://" + str(host) + ":" + str(port) + "/mcp"

    def sse_url(self, host, port=5000):
        return "http://" + str(host) + ":" + str(port) + "/sse"

    def _register_routes(self):
        self.http_server.route("/", GET)(self._index)
        self.http_server.route("/mcp", POST)(self._mcp_post)
        self.http_server.route("/mcp", GET)(self._mcp_get)
        self.http_server.route("/mcp", DELETE)(self._mcp_delete)
        self.http_server.route("/mcp", "OPTIONS")(self._mcp_options)
        self.http_server.route("/sse", GET)(self._sse_get)
        self.http_server.route("/messages", POST)(self._messages_post)
        self.http_server.route("/messages", "OPTIONS")(self._messages_options)

    def _tool_names(self):
        names = []
        for tool in self._tools:
            names.append(tool["name"])
        return names

    def _initialize_result(self, params):
        requested = None
        if isinstance(params, dict):
            requested = params.get("protocolVersion")

        protocol_version = self.protocol_versions[0]
        if requested in self.protocol_versions:
            protocol_version = requested

        return {
            "protocolVersion": protocol_version,
            "capabilities": {"tools": {"listChanged": False}},
            "serverInfo": {
                "name": self.name,
                "title": self.title,
                "version": self.version,
            },
            "instructions": self.instructions,
        }

    def handle_rpc(self, message):
        if not isinstance(message, dict):
            return jsonrpc_error(None, -32600, "Invalid Request")

        rpc_id = message.get("id")
        method = message.get("method")
        params = message.get("params")

        if not isinstance(method, str):
            return jsonrpc_error(rpc_id, -32600, "Invalid Request")

        if method == "initialize":
            return jsonrpc_result(rpc_id, self._initialize_result(params))

        if method == "notifications/initialized":
            if "id" in message:
                return jsonrpc_result(rpc_id, {})
            return None

        if method == "ping":
            return jsonrpc_result(rpc_id, {})

        if method == "tools/list":
            return jsonrpc_result(rpc_id, {"tools": self._tools})

        if method == "tools/call":
            return self._handle_tool_call(rpc_id, params)

        if method == "resources/list":
            return jsonrpc_result(rpc_id, {"resources": []})

        if method == "prompts/list":
            return jsonrpc_result(rpc_id, {"prompts": []})

        if method == "logging/setLevel":
            return jsonrpc_result(rpc_id, {})

        return jsonrpc_error(rpc_id, -32601, "Method not found", method)

    def _handle_tool_call(self, rpc_id, params):
        if not isinstance(params, dict):
            return jsonrpc_error(rpc_id, -32602, "Invalid params", "params must be an object")

        name = params.get("name")
        arguments = params.get("arguments", {})
        if not isinstance(name, str):
            return jsonrpc_error(rpc_id, -32602, "Invalid params", "tool name is required")

        handler = self._handlers.get(name)
        if handler is None:
            return jsonrpc_error(rpc_id, -32602, "Invalid params", "unknown tool: " + name)

        try:
            result = handler(arguments)
        except MCPError as error:
            return jsonrpc_error(rpc_id, error.code, error.message, error.data)
        except ValueError as error:
            return jsonrpc_error(rpc_id, -32602, "Invalid params", str(error))
        except Exception as error:
            return jsonrpc_error(rpc_id, -32603, "Internal error", str(error))

        if isinstance(result, str):
            result = tool_result(result)
        elif not isinstance(result, dict):
            result = tool_result(str(result))
        return jsonrpc_result(rpc_id, result)

    def _read_json_rpc(self, request):
        try:
            message = request.json()
        except Exception as error:
            return None, str(error)
        if message is None:
            return None, "empty or invalid JSON body"
        return message, None

    def _headers(self):
        return _copy_headers(self.cors_headers)

    def _json_response(self, request, payload, status=None):
        if status is None:
            return JSONResponse(request, payload, headers=self._headers())
        return JSONResponse(request, payload, status=status, headers=self._headers())

    def _text_response(self, request, body, status=None):
        if status is None:
            return Response(
                request, body=body, content_type="text/plain", headers=self._headers()
            )
        return Response(
            request,
            body=body,
            status=status,
            content_type="text/plain",
            headers=self._headers(),
        )

    def _empty_response(self, request, status):
        return Response(request, body="", status=status, headers=self._headers())

    def _index(self, request):
        return self._json_response(
            request,
            {
                "name": self.name,
                "title": self.title,
                "version": self.version,
                "mcp": "/mcp",
                "sse": "/sse",
                "tools": self._tool_names(),
            },
        )

    def _mcp_post(self, request):
        message, error = self._read_json_rpc(request)
        if error is not None:
            return self._json_response(
                request,
                jsonrpc_error(None, -32700, "Parse error", error),
            )

        response = self.handle_rpc(message)
        if response is None:
            return self._empty_response(request, (202, "Accepted"))
        return self._json_response(request, response)

    def _mcp_get(self, request):
        return self._text_response(
            request,
            "POST JSON-RPC messages to /mcp. GET is not used for this MCP HTTP endpoint.",
            (405, "Method Not Allowed"),
        )

    def _mcp_delete(self, request):
        return self._text_response(
            request, "DELETE /mcp is not supported.", (405, "Method Not Allowed")
        )

    def _mcp_options(self, request):
        return self._empty_response(request, (204, "No Content"))

    def _sse_get(self, request):
        if self.sse_response_class is None:
            return self._text_response(
                request,
                "This build supports MCP streamable HTTP POST at /mcp. "
                "SSEResponse is not available in the installed adafruit_httpserver library.",
                (501, "Not Implemented"),
            )

        if self.sse_client is not None:
            try:
                self.sse_client.close()
            except Exception:
                pass
        self.sse_client = self.sse_response_class(request, headers=self._headers())
        self.sse_welcome_pending = True
        return self.sse_client

    def _messages_post(self, request):
        if self.sse_response_class is None:
            return self._text_response(
                request,
                "SSE is not available; POST JSON-RPC directly to /mcp instead.",
                (404, "Not Found"),
            )

        if self.sse_client is None:
            return self._text_response(
                request,
                "No SSE client is connected. Open /sse first or use /mcp.",
                (409, "Conflict"),
            )

        message, error = self._read_json_rpc(request)
        if error is not None:
            response = jsonrpc_error(None, -32700, "Parse error", error)
        else:
            response = self.handle_rpc(message)

        if response is not None:
            try:
                self.sse_client.send_event(json.dumps(response), event="message")
            except Exception:
                try:
                    self.sse_client.close()
                except Exception:
                    pass
                self.sse_client = None
                return self._text_response(request, "SSE client disconnected.", (409, "Conflict"))

        return self._empty_response(request, (202, "Accepted"))

    def _messages_options(self, request):
        return self._empty_response(request, (204, "No Content"))

    def _send_pending_sse_endpoint(self):
        if self.sse_client is None or not self.sse_welcome_pending:
            return
        try:
            self.sse_client.send_event("/messages", event="endpoint")
            self.sse_welcome_pending = False
        except Exception:
            try:
                self.sse_client.close()
            except Exception:
                pass
            self.sse_client = None
            self.sse_welcome_pending = False