> For the complete documentation index, see [llms.txt](https://yeasy.gitbook.io/harness_engineering_guide/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://yeasy.gitbook.io/harness_engineering_guide/di-san-bu-fen-xi-tong-ji-cheng-yu-gong-cheng-shi-jian/09_mcp/9.3_server_dev.md).

# 9.3 MCP 服务端开发

本节详细讲解如何从零开始构建 MCP Server，包括基本开发步骤、完整的 Python 实现示例、关键概念说明以及错误处理方案。通过这些实现细节，你将掌握如何定义工具(Tool)、资源(Resource)和提示词(Prompt)，选择合适的传输方式，并确保 Server 的稳定性。

> **示例的协议版本**：本节代码按当前修订版 **2026-07-28** 的无状态模型编写——没有 `initialize` 握手，协议版本与客户端能力随每个请求的 `params._meta` 传入，服务端实现 `server/discover` 供客户端一次性发现，每个 result 带 `resultType`，列表与读取类结果带 `ttlMs`/`cacheScope`。若要对接仍讲 2025-11-25 的旧客户端，见 [9.1 协议设计](/harness_engineering_guide/di-san-bu-fen-xi-tong-ji-cheng-yu-gong-cheng-shi-jian/09_mcp/9.1_protocol_design.md) 的向后兼容小节与本节末尾的旧流量处理建议。

## 9.3.1 开发 MCP Server 的基本步骤

创建一个完整的 MCP Server 需要：

1. 定义 Tools（可调用的函数）
2. 定义 Resources（可访问的数据）
3. 定义 Prompts（提示词模板）
4. 实现处理器
5. 选择传输方式并启动 Server

## 9.3.2 完整的 Python MCP Server 实现

MCP Server 的完整 Python 实现包括四个主要部分：工具定义、资源处理、不同传输方式的实现，以及主函数。下面是教学用的最小协议骨架，用来解释 JSON-RPC 消息结构；生产项目应优先使用官方或社区维护的 MCP SDK。

### 第一部分：数据模型与工具定义

首先需要定义数据模型来表示工具、资源和提示词：

```python
# mcp_server.py
# 一个完整的MCP Server示例,提供文件系统工具和资源访问

import json
import os
import subprocess
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, asdict
from enum import Enum
import asyncio
import sys
from pathlib import Path

@dataclass
class MCPTool:
    """MCP工具定义"""
    name: str
    description: str
    inputSchema: Dict[str, Any]

@dataclass
class MCPResource:
    """MCP资源定义"""
    uri: str
    name: str
    description: str
    mimeType: str  # MIME 类型

@dataclass
class MCPPrompt:
    """MCP提示词定义"""
    name: str
    description: str
    arguments: List[Dict[str, Any]]

class MCPServerBase:
    """MCP Server基类"""

    def __init__(self):
        self.tools: Dict[str, MCPTool] = {}
        self.resources: Dict[str, MCPResource] = {}
        self.prompts: Dict[str, MCPPrompt] = {}
        self.tool_handlers: Dict[str, callable] = {}
        self.resource_readers: Dict[str, callable] = {}
        self.prompt_generators: Dict[str, callable] = {}

    def register_tool(
        self, name: str, description: str,
        input_schema: Dict[str, Any], handler: callable
    ) -> None:
        """注册工具"""
        self.tools[name] = MCPTool(name, description, input_schema)
        self.tool_handlers[name] = handler

    def register_resource(
        self, uri: str, name: str, description: str,
        mime_type: str, reader: callable
    ) -> None:
        """注册资源"""
        self.resources[uri] = MCPResource(uri, name, description, mime_type)
        self.resource_readers[uri] = reader

    def register_prompt(
        self, name: str, description: str,
        arguments: List[Dict[str, Any]], generator: callable
    ) -> None:
        """注册提示词"""
        self.prompts[name] = MCPPrompt(name, description, arguments)
        self.prompt_generators[name] = generator
```

**设计说明**：基类采用了“注册模式”(Registration Pattern)，允许在运行时动态注册工具、资源和提示词。这提供了灵活性，使不同的 Server 实例可以有不同的功能集合。

### 第二部分：请求处理与 JSON-RPC 协议

MCP 使用 JSON-RPC 2.0 协议进行通信。以下是请求处理的核心逻辑：

```python
    # 2026-07-28 无状态时代：协议版本与客户端能力随每个请求的 params._meta 传入
    PROTOCOL_VERSION_KEY = "io.modelcontextprotocol/protocolVersion"
    CLIENT_CAPABILITIES_KEY = "io.modelcontextprotocol/clientCapabilities"
    SERVER_INFO_KEY = "io.modelcontextprotocol/serverInfo"
    SUPPORTED_VERSIONS = ["2026-07-28"]
    SERVER_INFO = {"name": "example-mcp-server", "version": "1.0.0"}

    # 需要客户端能力的方法，值取 ClientCapabilities 的形状。
    # 本例的处理器都不向客户端索取输入，因此这里是空的；只有当某个方法确实
    # 会返回 input_required 去要 elicitation / sampling / roots 时才在此登记，
    # 例如 {"prompts/get": {"elicitation": {"form": {}}}}。登记了但处理器
    # 用不到，只会让没声明该能力的客户端白白拿到 -32021。
    REQUIRED_CLIENT_CAPABILITIES: Dict[str, Dict[str, Any]] = {}

    # 列表与读取类结果必须带新鲜度提示，method 映射到 ttlMs 与 cacheScope
    CACHE_HINTS = {
        "server/discover": (60000, "public"),
        "tools/list": (60000, "public"),
        "prompts/list": (60000, "public"),
        "resources/list": (60000, "public"),
        "resources/read": (5000, "private"),  # 用户私有内容不能标成 public
    }

    async def handle_request(self, request: Dict[str, Any]) -> Dict[str, Any]:
        """处理JSON-RPC请求，没有握手，每个请求自带完整协议信封"""
        method = request.get("method")
        params = request.get("params", {})
        request_id = request.get("id")
        meta = params.get("_meta") or {}

        def error(code: int, message: str, data: Any = None) -> Dict[str, Any]:
            err: Dict[str, Any] = {"code": code, "message": message}
            if data is not None:
                err["data"] = data
            return {"jsonrpc": "2.0", "id": request_id, "error": err}

        # 信封校验一：两个必需键缺一不可，缺失即非法参数
        missing = [
            key
            for key in (self.PROTOCOL_VERSION_KEY, self.CLIENT_CAPABILITIES_KEY)
            if key not in meta
        ]
        if missing:
            return error(-32602, f"Missing required _meta keys: {', '.join(missing)}")

        # 信封校验二：版本不支持时，把本端支持的版本回给客户端去降级
        requested = meta[self.PROTOCOL_VERSION_KEY]
        if requested not in self.SUPPORTED_VERSIONS:
            return error(
                -32022,
                f"Unsupported protocol version: {requested}",
                {"supported": self.SUPPORTED_VERSIONS, "requested": requested},
            )

        # 信封校验三：本次调用要用到的客户端能力必须已在信封里声明
        declared = meta[self.CLIENT_CAPABILITIES_KEY] or {}
        needed = self.REQUIRED_CLIENT_CAPABILITIES.get(method, {})
        lacking = {k: v for k, v in needed.items() if declared.get(k) is None}
        if lacking:
            return error(
                -32021,
                f"Client did not declare the capabilities required by {method}",
                {"requiredCapabilities": lacking},
            )

        try:
            if method == "server/discover":
                # 取代 initialize，无状态时代服务端必须实现的方法
                result = {
                    "supportedVersions": self.SUPPORTED_VERSIONS,
                    "capabilities": {"tools": {}, "resources": {}, "prompts": {}},
                }
            elif method == "tools/list":
                result = self._handle_tools_list()
            elif method == "tools/call":
                result = await self._handle_tools_call(params)
            elif method == "resources/list":
                result = self._handle_resources_list()
            elif method == "resources/read":
                result = await self._handle_resources_read(params)
            elif method == "prompts/list":
                result = self._handle_prompts_list()
            elif method == "prompts/get":
                result = await self._handle_prompts_get(params)
            else:
                return error(-32601, f"Method not found: {method}")

            # 每个result都要带 resultType，列表与读取类再补缓存提示
            result.setdefault("resultType", "complete")
            hint = self.CACHE_HINTS.get(method)
            if hint is not None:
                result.setdefault("ttlMs", hint[0])
                result.setdefault("cacheScope", hint[1])
            # 服务端身份不再由握手交换，改为每个result的 _meta 回带
            result.setdefault("_meta", {})[self.SERVER_INFO_KEY] = dict(self.SERVER_INFO)
            return {"jsonrpc": "2.0", "id": request_id, "result": result}

        except Exception as e:
            return error(-32000, str(e))
```

**设计说明**：请求处理是 MCP 通信的“中枢”。每个请求必须返回合法的 JSON-RPC 2.0 响应，包括 id、result 或 error。这确保了客户端可以准确关联请求和响应，即使在异步场景中也不会混淆。

### 第三部分：资源与提示词处理

现在实现具体的处理方法，展示工具、资源和提示词的导出：

```python
    def _handle_initialize(self) -> Dict[str, Any]:
        """处理初始化请求"""
        return {
            "protocolVersion": "2025-11-25",
            "capabilities": {
                "tools": {},
                "resources": {},
                "prompts": {},
            },
            "serverInfo": {
                "name": "example-mcp-server",
                "version": "1.0.0",
            },
        }

    def _handle_tools_list(self) -> Dict[str, Any]:
        """列出所有工具"""
        tools_list = [asdict(tool) for tool in self.tools.values()]
        return {"tools": tools_list}

    async def _handle_tools_call(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """调用工具"""
        tool_name = params.get("name")
        arguments = params.get("arguments", {})

        if tool_name not in self.tool_handlers:
            raise ValueError(f"Tool not found: {tool_name}")

        handler = self.tool_handlers[tool_name]
        result = await handler(arguments)

        return {
            "content": [
                {
                    "type": "text",
                    "text": result,
                }
            ]
        }

    def _handle_resources_list(self) -> Dict[str, Any]:
        """列出所有资源"""
        resources_list = [asdict(res) for res in self.resources.values()]
        return {"resources": resources_list}

    async def _handle_resources_read(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """读取资源"""
        uri = params.get("uri")

        if uri not in self.resource_readers:
            raise ValueError(f"Resource not found: {uri}")

        reader = self.resource_readers[uri]
        content = await reader()

        return {
            "contents": [
                {
                    "uri": uri,
                    "mimeType": self.resources[uri].mimeType,
                    "text": content,
                }
            ]
        }

    def _handle_prompts_list(self) -> Dict[str, Any]:
        """列出所有提示词"""
        prompts_list = [asdict(prompt) for prompt in self.prompts.values()]
        return {"prompts": prompts_list}

    async def _handle_prompts_get(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """获取提示词"""
        name = params.get("name")
        arguments = params.get("arguments", {})

        if name not in self.prompt_generators:
            raise ValueError(f"Prompt not found: {name}")

        generator = self.prompt_generators[name]
        messages = await generator(arguments)

        return {"messages": messages}
```

**设计说明**：这些处理方法遵循了一致的模式：列表方法返回所有已注册的对象、调用方法执行对应的处理器。注意 `_handle_prompts_get` 支持参数，使提示词能够根据不同的上下文动态生成。上面的返回结构沿用 2025-11-25 修订版；若要适配 2026-07-28 修订版，有两个细节容易被忽略：其一，每个 result 都必须携带 `resultType`，取值是 `"complete"` 或 `"input_required"`；客户端遇到旧服务端返回的、没有 `resultType` 的结果时按 `"complete"` 处理。其二，`tools/list`、`prompts/list`、`resources/list`、`resources/read`、`resources/templates/list` 的结果还必须带 `ttlMs`（新鲜度提示，毫秒）和 `cacheScope`（`"public"` 或 `"private"`），客户端据此缓存——把用户私有的文件内容标成 `"public"` 会让它被跨用户复用。另外，只有 `prompts/get`、`resources/read`、`tools/call` 可以返回 `resultType` 为 `"input_required"` 的结果，用来向客户端索取输入（如获取根目录列表、请求用户确认或请求模型补全），客户端补齐输入后用新的 JSON-RPC id 重试同一请求。

### 第四部分：文件系统实现与传输

最后，我们实现一个具体的文件系统 Server 和两种传输方式。以下展示文件系统工具的实现：

```python
class FileSystemMCPServer(MCPServerBase):
    """提供文件系统访问的MCP Server"""

    def __init__(self, root_path: str = "."):
        super().__init__()
        self.root_path = Path(root_path).resolve()

        # 注册三个文件系统工具
        self.register_tool(
            name="read_file",
            description="Read contents of a file",
            input_schema={
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "Path to the file"},
                    "encoding": {
                        "type": "string",
                        "enum": ["utf-8", "ascii", "latin-1"],
                        "description": "File encoding (default: utf-8)",
                    },
                },
                "required": ["path"],
            },
            handler=self.handle_read_file,
        )

        self.register_tool(
            name="write_file",
            description="Write contents to a file",
            input_schema={
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "Path to the file"},
                    "contents": {"type": "string", "description": "File contents"},
                },
                "required": ["path", "contents"],
            },
            handler=self.handle_write_file,
        )

        self.register_tool(
            name="list_directory",
            description="List files in a directory",
            input_schema={
                "type": "object",
                "properties": {
                    "path": {"type": "string", "description": "Directory path"},
                },
                "required": ["path"],
            },
            handler=self.handle_list_directory,
        )

    async def handle_read_file(self, args: Dict[str, Any]) -> str:
        """处理读文件请求"""
        file_path = self._resolve_path(args["path"])
        encoding = args.get("encoding", "utf-8")
        with open(file_path, "r", encoding=encoding) as f:
            contents = f.read()
        return f"File contents:\n\n{contents}"

    async def handle_write_file(self, args: Dict[str, Any]) -> str:
        """处理写文件请求"""
        file_path = self._resolve_path(args["path"])
        contents = args["contents"]
        file_path.parent.mkdir(parents=True, exist_ok=True)
        with open(file_path, "w") as f:
            f.write(contents)
        return f"File written: {file_path}"

    async def handle_list_directory(self, args: Dict[str, Any]) -> str:
        """处理列出目录请求"""
        dir_path = self._resolve_path(args["path"])
        if not dir_path.is_dir():
            return f"Not a directory: {dir_path}"
        items = []
        for item in dir_path.iterdir():
            item_type = "dir" if item.is_dir() else "file"
            items.append(f"  [{item_type}] {item.name}")
        return f"Contents of {dir_path}:\n" + "\n".join(items)

    def _resolve_path(self, path: str) -> Path:
        """解析路径并严格校验不逃逸根目录"""
        resolved = (self.root_path / path).resolve()
        try:
            resolved.relative_to(self.root_path.resolve())
        except ValueError as e:
            raise ValueError(f"Path traversal not allowed: {path}") from e
        return resolved
```

**设计说明**：文件系统 Server 展示了三个关键的安全做法：(1) 限制可访问的根路径；(2) 验证路径不会逃出根目录；(3) 在执行操作前解析并验证所有路径。这是处理用户输入的敏感操作（如文件访问）的标准防御方式。上面代码使用 `resolved.relative_to(self.root_path.resolve())` 进行边界检查是最佳实践——相比字符串前缀比较（`resolved.startswith()`），它能正确处理 Windows 路径、符号链接和相对路径。完整的五层递进式路径校验实践（长度检查、URL 编码双解码、Unicode 规范化、平台特定规范化、符号链接解析与边界检查）详见第 12.4 节。

### 第五部分：传输层实现

MCP 支持多种传输方式。以下展示 stdio 和 HTTP 两种传输的简化实现：

```python
class StdioMCPServer:
    """使用stdio传输的MCP Server"""

    def __init__(self, server: MCPServerBase):
        self.server = server

    async def run(self) -> None:
        """运行服务器"""
        loop = asyncio.get_running_loop()

        while True:
            try:
                # 读取一行JSON
                line = await loop.run_in_executor(None, sys.stdin.readline)
                if not line:
                    break

                request = json.loads(line.strip())
                response = await self.server.handle_request(request)

                # 写入响应
                json_line = json.dumps(response)
                sys.stdout.write(json_line + "\n")
                sys.stdout.flush()

            except json.JSONDecodeError as e:
                sys.stderr.write(f"JSON decode error: {e}\n")
            except Exception as e:
                sys.stderr.write(f"Error: {e}\n")
```

**stdio 传输的优势**：(1) 无需网络配置，完全通过管道通信；(2) 天然支持进程隔离；(3) 无 HTTP 分帧与连接管理开销，只需逐行 JSON 处理。这使 stdio 成为本地工具集成的理想选择。这里要提醒一个常见误解：进程一直开着并不等于“会话一直在”。上面的示例遵循 2025-11-25 修订版，仍带有 `initialize` 握手；而在 2026-07-28 修订版中，握手与协议级会话已被移除，服务端不得把前一条请求留下的信息当作后一条请求的上下文，协议版本与客户端能力改为随每个请求放在 `params._meta` 中传递。迁移时，凡是原本在握手期建立、后续请求默认可见的状态，都要改成由客户端每次显式传入的标识。

```python
try:
    from aiohttp import web
    HAS_AIOHTTP = True
except ImportError:
    HAS_AIOHTTP = False

# 2026-07-28 的无状态请求信封：协议版本与客户端能力随每个请求放在 params._meta 里
PROTOCOL_VERSION = "2026-07-28"
PROTOCOL_VERSION_META_KEY = "io.modelcontextprotocol/protocolVersion"
CLIENT_CAPABILITIES_META_KEY = "io.modelcontextprotocol/clientCapabilities"
# 这几个方法还要多校验一个 Mcp-Name 头；值是要与该头比对的 params 键
NAME_BEARING_METHODS = {"tools/call": "name", "prompts/get": "name", "resources/read": "uri"}
# JSON-RPC 错误码到 HTTP 状态码的映射；未列出的错误码仍用 200 承载
ERROR_HTTP_STATUS = {
    -32700: 400, -32600: 400, -32602: 400,
    -32020: 400, -32021: 400, -32022: 400,
    -32601: 404,
}

class HttpMCPServer:
    """使用Streamable HTTP传输的MCP Server(2026-07-28 无状态形态)"""

    def __init__(
        self,
        server: MCPServerBase,
        host: str = "127.0.0.1",
        port: int = 8000,
        bearer_token: Optional[str] = None,
        allowed_origins: Optional[set] = None,
    ):
        self.server = server
        self.host = host
        self.port = port
        self.bearer_token = bearer_token
        self.allowed_origins = allowed_origins or {"http://localhost:8000"}
        # 协议级会话已移除，因此没有会话字典；跨请求的状态只能由客户端每次显式带上
        if not HAS_AIOHTTP:
            raise RuntimeError("aiohttp not installed. Install it with: pip install aiohttp")

    def _check_security(self, request):
        origin = request.headers.get("Origin")
        if origin and origin not in self.allowed_origins:
            return web.Response(status=403, text="Origin not allowed")
        if self.bearer_token:
            expected = f"Bearer {self.bearer_token}"
            if request.headers.get("Authorization") != expected:
                return web.Response(status=401, text="Unauthorized")
        return None

    def _validate_envelope(self, request, data):
        """校验请求信封与路由头；通过返回None，否则返回(错误码, 说明, 附加数据)"""
        params = data.get("params")
        meta = params.get("_meta") if isinstance(params, dict) else None
        if not isinstance(meta, dict):
            return (-32602, "params._meta must carry the required envelope keys", None)
        missing = [
            key for key in (PROTOCOL_VERSION_META_KEY, CLIENT_CAPABILITIES_META_KEY)
            if key not in meta
        ]
        if missing:
            return (-32602, f"params._meta is missing: {', '.join(missing)}", None)

        # 头必须与body自洽。本实现只服务一个修订版，所以不一致直接判 -32020；
        # 若要同时兼容旧版，请求会先按 MCP-Protocol-Version 头分流到对应时代的传输层，
        # 那时不一致可能表现为别的错误码
        method = data.get("method")
        if request.headers.get("MCP-Protocol-Version") != meta[PROTOCOL_VERSION_META_KEY]:
            return (-32020, "MCP-Protocol-Version header does not match params._meta", None)
        if request.headers.get("Mcp-Method") != method:
            return (-32020, "Mcp-Method header does not match the body method", None)
        name_key = NAME_BEARING_METHODS.get(method)
        if name_key is not None and params.get(name_key) is not None:
            if request.headers.get("Mcp-Name") != params[name_key]:
                return (-32020, f"Mcp-Name header does not match params.{name_key}", None)

        version = meta[PROTOCOL_VERSION_META_KEY]
        if version != PROTOCOL_VERSION:
            err_data = {"supported": [PROTOCOL_VERSION], "requested": version}
            return (-32022, f"Unsupported protocol version: {version}", err_data)
        return None

    def _error_response(self, request_id, code, message, data=None):
        """把JSON-RPC错误按错误码映射到对应的HTTP状态码"""
        error = {"code": code, "message": message}
        if data is not None:
            error["data"] = data
        return web.json_response(
            {"jsonrpc": "2.0", "id": request_id, "error": error},
            status=ERROR_HTTP_STATUS.get(code, 200),
        )

    async def handle_mcp_post(self, request):
        """处理POST /mcp(每条JSON-RPC消息一次独立POST，请求之间不共享状态)"""
        security_error = self._check_security(request)
        if security_error:
            return security_error
        accept = request.headers.get("Accept", "")
        if "application/json" not in accept or "text/event-stream" not in accept:
            return web.Response(status=406, text="Accept must include JSON and SSE")
        data = await request.json()
        request_id = data.get("id")

        # 通知没有 id，无法承载错误，接受后直接返回 202 空响应体；
        # notifications/initialized 已随握手一起移除，不再有“已初始化”这种状态可置
        if request_id is None:
            return web.Response(status=202)

        rejection = self._validate_envelope(request, data)
        if rejection:
            return self._error_response(request_id, *rejection)

        response = await self.server.handle_request(data)
        error = response.get("error")
        if error:
            # 未知方法是 HTTP 404 加 -32601，其余协议级错误是 400，处理器内部错误仍走 200
            return web.json_response(response, status=ERROR_HTTP_STATUS.get(error["code"], 200))
        return web.json_response(response)

    async def run(self):
        """启动Streamable HTTP服务器"""
        app = web.Application()
        # 单端点只注册POST。GET流端点与DELETE会话删除都已移除，
        # aiohttp 会为同一路径上的其他方法自动回 405 并带上 Allow: POST
        app.router.add_post("/mcp", self.handle_mcp_post)
        runner = web.AppRunner(app)
        await runner.setup()
        site = web.TCPSite(runner, self.host, self.port)
        await site.start()
        print(f"Streamable HTTP MCP Server running at http://{self.host}:{self.port}")
        await asyncio.Event().wait()
```

**HTTP 传输的优势**：(1) 支持远程访问，跨机器通信；(2) GET 流连接用于 Server 推送事件；(3) 标准的 HTTP 协议，易于在网络中部署和监控。

### 第六部分：启动函数

最后，主函数根据命令行参数选择传输方式：

```python
async def main():
    """主函数"""
    # 创建文件系统MCP Server
    fs_server = FileSystemMCPServer(root_path=".")

    # 选择传输方式
    if len(sys.argv) > 1 and sys.argv[1] == "--http":
        # 使用HTTP传输
        transport = HttpMCPServer(fs_server)
        await transport.run()
    else:
        # 使用stdio传输(默认)
        transport = StdioMCPServer(fs_server)
        await transport.run()

if __name__ == "__main__":
    asyncio.run(main())
```

### Server 开发的关键概念

#### 1. 工具定义的 JSON Schema

工具的`inputSchema`应该是完整的 JSON Schema，包含：

* `type`: 必须是“object”
* `properties`: 参数定义
* `required`: 必需参数列表

```json
{
  "type": "object",
  "properties": {
    "name": {
      "type": "string",
      "description": "User's full name"
    },
    "age": {
      "type": "integer",
      "minimum": 0,
      "maximum": 150
    },
    "tags": {
      "type": "array",
      "items": {"type": "string"}
    }
  },
  "required": ["name"],
  "additionalProperties": false
}
```

#### 2. 资源 URI 规范

Resources 应该有结构化的 URI（统一资源标识符）：

```yaml
file:///path/to/document.md
db://postgres/public/users/123
notion://page_abc123def456
github://owner/repo/issues/42
```

#### 3. 提示词参数化

Prompts 可以接受参数，在生成消息时使用这些参数：

```python
async def generate_code_review_prompt(self, args: Dict[str, Any]) -> List[Dict]:
    """生成代码审查提示词"""
    language = args.get("language", "Python")
    focus_areas = args.get("focus_areas", "correctness,performance")

    prompt = f"""You are an expert code reviewer for {language}.
Review the following code focusing on: {focus_areas}

Provide:
1. Issues found (if any)
2. Suggestions for improvement
3. Overall assessment
"""

    return [
        {
            "role": "user",
            "content": {
                "type": "text",
                "text": prompt,
            }
        }
    ]
```

#### 4. 面向旧版客户端的兼容

真实环境里仍有大量按 2025-11-25 修订版实现的客户端和服务端，兼容问题绕不开。如果你的服务端只实现 2026-07-28，就要让旧流量快速失败，而不是让它挂在半路：

* MCP 端点上的 GET 与 DELETE 一律返回 `405 Method Not Allowed`；
* 收到 `Mcp-Session-Id` 请求头直接忽略，不要据此建立或恢复会话；
* 收到 `Last-Event-ID` 请求头直接忽略，新版不再支持 SSE 断点续传。

如果服务端要同时服务两代客户端，就得在同一端点上保留旧的 `initialize`/`notifications/initialized` 握手分支和旧的结果格式，按请求实际使用的修订版分派。这类兼容代码应当有明确的下线时间，而不是长期并存。反过来，客户端探测服务端时的做法是：先按新版发请求，如果收到 HTTP 400 就检查响应体——能解析出新版定义的 JSON-RPC 错误，说明对方是新版服务端，改正请求重试即可；响应体为空或无法识别，才回退到旧的 `initialize` 握手。stdio 场景下没有 HTTP 状态码可用，`server/discover` 就是那次探测请求：旧服务端不认识它，会按未知方法返回 `-32601`。

注意区分“移除”和“弃用”：Roots、Sampling、Logging 属于后者，它们至少还有 12 个月的弃用期，现有服务端继续可用，但新写的服务端不应再依赖——需要目录或文件范围就用工具参数、资源 URI 传入，需要模型补全就直接调用模型提供方的 API，需要日志就写 stderr 或接入 OpenTelemetry。

### 错误处理

MCP Server 应该正确处理和报告错误。除 JSON-RPC 2.0 标准错误码外，2026-07-28 修订版还增加了几个 MCP 专有错误码：`-32020`（请求头中的协议版本与 `_meta` 中的不一致）、`-32021`（客户端未声明服务端所需的能力）、`-32022`（不支持的协议版本）；资源不存在的错误码也从 `-32002` 改成了 `-32602`，但客户端仍应接受旧服务端返回的 `-32002`：

```python
class MCPServerError(Exception):
    """MCP Server错误"""

    def __init__(self, code: int, message: str, data: Optional[Dict[str, Any]] = None):
        self.code = code
        self.message = message
        self.data = data

    def to_json_rpc_error(self) -> Dict[str, Any]:
        error: Dict[str, Any] = {
            "code": self.code,
            "message": self.message,
        }
        if self.data is not None:
            error["data"] = self.data
        return error

# 标准错误码
class MCPErrorCode(Enum):
    PARSE_ERROR = -32700
    INVALID_REQUEST = -32600
    METHOD_NOT_FOUND = -32601  # 未知方法，HTTP 404
    # 缺少必填的_meta字段，HTTP 400；资源不存在也复用此码
    # （旧版服务端可能仍返回 -32002，客户端应继续接受）
    INVALID_PARAMS = -32602
    INTERNAL_ERROR = -32603
    SERVER_ERROR = -32000  # 到 -32099

# MCP专有错误码
class MCPProtocolErrorCode(Enum):
    # MCP-Protocol-Version 请求头与 _meta 中的版本不一致，HTTP 400
    HEADER_MISMATCH = -32020
    # 服务端需要客户端未声明的能力，data.requiredCapabilities 说明缺哪些，HTTP 400
    MISSING_REQUIRED_CLIENT_CAPABILITY = -32021
    # 客户端请求的修订版不受支持，响应中列出服务端支持的版本，HTTP 400
    UNSUPPORTED_PROTOCOL_VERSION = -32022
```

### 本小节小结

开发 MCP Server 的核心是：

1. 定义 Tool、Resource 和 Prompt 的 Schema
2. 实现对应的处理器
3. 选择合适的传输方式
4. 正确处理错误和异常

关键要点：

* JSON Schema 应该准确且完整
* 资源 URI 应该有明确的结构
* 提示词应该支持参数化
* 错误响应应该遵循 JSON-RPC 2.0 规范

下一节将讨论 Harness 如何在系统级别集成多个 MCP Server。