MCP Server

实现一个极简的 MCP（Model Context Protocol）服务端：基于 JSON-RPC 2.0 消息格式， 提供 initialize、tools/list、tools/call 三个核心方法。真实 MCP 通过 stdio 或 SSE 传输，本示例聚焦协议语义，便于离线测试。 对应教程：第7卷《MCP》第2章 MCP Server。

文件路径: examples/volume-07-mcp\mcp_server.py

---

源代码

"""MCP Server 概念实现示例。

实现一个极简的 MCP（Model Context Protocol）服务端：基于 JSON-RPC 2.0 消息格式，
提供 ``initialize``、``tools/list``、``tools/call`` 三个核心方法。真实 MCP 通过 stdio
或 SSE 传输，本示例聚焦协议语义，便于离线测试。

对应教程：第7卷《MCP》第2章 MCP Server。
"""

from __future__ import annotations

import json
from collections.abc import Callable
from dataclasses import dataclass, field
from typing import Any

JSONRPC_VERSION = "2.0"


@dataclass
class McpTool:
    """MCP 工具定义。

    Attributes:
        name: 工具名。
        description: 描述。
        input_schema: 参数 JSON Schema。
        func: 执行函数。
    """

    name: str
    description: str
    input_schema: dict[str, Any]
    func: Callable[..., Any]


@dataclass
class McpServer:
    """MCP 服务端（概念实现）。

    Attributes:
        name: 服务名称。
        version: 服务版本。
    """

    name: str = "demo-mcp-server"
    version: str = "0.1.0"
    _tools: dict[str, McpTool] = field(default_factory=dict)
    _initialized: bool = False

    def register_tool(self, tool: McpTool) -> None:
        """注册工具。

        Args:
            tool: 工具定义。
        """
        self._tools[tool.name] = tool

    def list_tools(self) -> list[dict[str, Any]]:
        """返回工具列表（不含执行函数）。"""
        return [
            {
                "name": t.name,
                "description": t.description,
                "inputSchema": t.input_schema,
            }
            for t in self._tools.values()
        ]

    def call_tool(self, name: str, arguments: dict[str, Any]) -> Any:
        """执行工具。

        Args:
            name: 工具名。
            arguments: 参数。

        Returns:
            工具返回值。

        Raises:
            KeyError: 工具未注册。
        """
        tool = self._tools.get(name)
        if tool is None:
            raise KeyError(f"未注册工具: {name}")
        return tool.func(**arguments)

    def handle_request(self, request: dict[str, Any]) -> dict[str, Any]:
        """处理一条 JSON-RPC 请求并返回响应。

        Args:
            request: JSON-RPC 2.0 请求字典，需含 ``method``、可选 ``params`` 与 ``id``。

        Returns:
            JSON-RPC 2.0 响应字典。
        """
        req_id = request.get("id")
        method = request.get("method")
        params = request.get("params", {}) or {}

        try:
            if method == "initialize":
                self._initialized = True
                result: Any = {
                    "protocolVersion": "2024-11-05",
                    "serverInfo": {"name": self.name, "version": self.version},
                    "capabilities": {"tools": {}},
                }
            elif method == "tools/list":
                result = {"tools": self.list_tools()}
            elif method == "tools/call":
                result = {"content": [{"type": "text", "text": str(self.call_tool(**params))}]}
            else:
                return self._error(req_id, -32601, f"方法不存在: {method}")
        except Exception as exc:  # noqa: BLE001
            return self._error(req_id, -32000, str(exc))

        return {"jsonrpc": JSONRPC_VERSION, "result": result, "id": req_id}

    @staticmethod
    def _error(req_id: Any, code: int, message: str) -> dict[str, Any]:
        return {
            "jsonrpc": JSONRPC_VERSION,
            "error": {"code": code, "message": message},
            "id": req_id,
        }


# ===== 示例工具 =====


def get_time(timezone: str = "UTC") -> str:
    """示例工具：返回当前时间字符串。"""
    from datetime import datetime

    return f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ({timezone})"


def add_numbers(a: int, b: int) -> int:
    """示例工具：两数相加。"""
    return a + b


def build_demo_server() -> McpServer:
    """构造注册了示例工具的 MCP 服务端。

    Returns:
        ``McpServer``。
    """
    server = McpServer(name="demo-mcp-server", version="0.1.0")
    server.register_tool(
        McpTool(
            name="get_time",
            description="获取当前时间",
            input_schema={
                "type": "object",
                "properties": {"timezone": {"type": "string"}},
            },
            func=get_time,
        )
    )
    server.register_tool(
        McpTool(
            name="add_numbers",
            description="两数相加",
            input_schema={
                "type": "object",
                "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}},
                "required": ["a", "b"],
            },
            func=add_numbers,
        )
    )
    return server


def main() -> None:
    """演示 MCP 服务端处理请求。"""
    server = build_demo_server()

    requests = [
        {"jsonrpc": JSONRPC_VERSION, "method": "initialize", "id": 1},
        {"jsonrpc": JSONRPC_VERSION, "method": "tools/list", "id": 2},
        {
            "jsonrpc": JSONRPC_VERSION,
            "method": "tools/call",
            "params": {"name": "add_numbers", "arguments": {"a": 3, "b": 5}},
            "id": 3,
        },
        {"jsonrpc": JSONRPC_VERSION, "method": "unknown", "id": 4},
    ]
    for req in requests:
        resp = server.handle_request(req)
        print(json.dumps(resp, ensure_ascii=False))


if __name__ == "__main__":
    main()