从概念到实践：手把手实现MCP服务器，解决AI工具集成难题-开发者社区

1. 从面试八股到实战工具：我为什么重新审视MCP

最近在准备面试，或者和同行交流大模型应用开发时，MCP（Model Context Protocol）这个词出现的频率越来越高。它常常和LangChain、LangGraph一起被提及，成为“AI应用架构”面试八股文里的一个标准答案点。大家可能都背过：“MCP是一种标准协议，用于将外部工具和数据源安全、统一地暴露给大模型……” 但说实话，在很长一段时间里，我对它的认知也停留在这个层面——一个听起来很美好，但似乎离具体编码有点远的“概念”。

直到我真正开始构建一个需要集成多种数据源（内部API、数据库、搜索引擎）的智能体（Agent）时，问题来了。我用LangChain的Tool接口写一个，用自定义函数包装另一个，每个工具的身份验证、输入输出格式、错误处理都自成一体。项目很快变成了“胶水代码”的泥潭，维护和扩展新工具的成本高得吓人。这时，我才回头仔细看了MCP的协议文档和社区实现，发现它根本不是又一个“为KPI而生的协议”，而是一套切实解决生产级AI应用“工具集成之痛”的工程方案。

简单说，MCP定义了一套客户端（如Code编辑器、AI助手）与服务器（提供工具和数据）之间的标准化通信协议。它的核心价值在于解耦与标准化。开发者可以编写一次MCP服务器，任何支持该协议的客户端（如Claude Code、Cursor、Windmill）都能直接发现并使用其提供的所有工具（Tools）和资源（Resources），无需为每个客户端单独适配。这就像为你的各种能力（查数据库、调API、读文件）提供了统一的USB-C接口，任何拥有对应端口的设备都能即插即用。

本文将彻底抛开面试概念的浮光掠影，从一个实践者的角度，深度解析MCP的协议设计、核心组件，并手把手带你从零实现一个功能完整的MCP服务器。我们会涵盖协议基础、传输层（SSE与Stdio）对比、核心模型（Tools, Resources）的代码实现、错误处理、以及最终如何将其集成到Claude Code或自定义AI应用中。你会发现，理解MCP，最好的方式就是亲手实现它。

2. 撕开协议面纱：MCP的核心模型与通信机制

在开始写代码之前，我们必须先理解MCP协议到底规定了什么。它不是魔法，而是一套基于JSON-RPC 2.0的轻量级规范。整个协议围绕几个核心模型展开，理解了它们，就理解了MCP的骨架。

2.1 核心模型：Tools、Resources与Prompts

MCP服务器主要向客户端暴露三种类型的“能力”：

工具（Tools）：这是最常用的一类。一个工具就是一个可以被AI模型调用的函数，它有名称、描述、严格的输入参数模式（JSON Schema）。例如，一个“查询天气”的工具，输入是{“city”: “string”}，输出是天气信息文本。AI模型（或用户）通过名称和描述来理解何时调用它。
资源（Resources）：代表可读的数据源，如文件、数据库表片段或API的只读结果。资源有唯一的URI（如file:///path/to/doc.md或internal://project/status）和特定的MIME类型。客户端可以“读取”资源内容，供AI模型作为上下文参考。这为RAG（检索增强生成）等场景提供了标准化的数据注入管道。
提示词（Prompts）：预定义的、参数化的提示词模板。客户端可以获取这些模板，填入具体参数后直接用于与大模型对话。这有助于在团队或产品中标准化高质量的提示词。

对于本次实现，我们将重点放在Tools和Resources上，它们是构建AI智能体能力扩展的基石。

2.2 通信传输层：SSE与Stdio的抉择

MCP协议支持多种传输方式，最常用的是Server-Sent Events (SSE)和标准输入输出（Stdio）。选择哪种，取决于你的部署场景。

SSE (Over HTTP)：服务器作为一个HTTP服务运行，客户端通过HTTP连接与之通信。服务器使用SSE向客户端推送通知（如工具列表更新）。这种方式适合云服务或需要跨网络访问的场景，易于与现有Web架构集成。你需要处理HTTP路由、会话管理和可能的身份验证。
Stdio (标准输入输出)：服务器作为一个本地子进程启动，客户端通过标准输入（stdin）发送请求，通过标准输出（stdout）接收响应。这是本地集成最常见、最轻量的方式。Code编辑器插件（如Claude Code）通常以这种方式启动MCP服务器。它避免了网络端口冲突，安全性也更高（进程间通信）。我们的代码实现将主要采用这种方式，因为它最贴近开发者桌面工具的使用场景。

协议通信的基本单元是JSON-RPC 2.0格式的消息。每个请求（Request）都有id,method,params字段；每个响应（Result）或错误（Error）也都包含对应的id。例如，客户端初始化连接后，会发送“initialize”请求，服务器回复自身能力；接着客户端会发送“tools/list”请求来获取所有可用工具。

注意：MCP协议是双向的，但初始化后，主要由客户端驱动（发送请求）。服务器可以在资源内容变化时，主动通过通知（Notification）告知客户端，客户端再决定是否重新读取。

3. 动手实现：构建一个提供“待办”与“天气”工具的MCP服务器

理论说得再多，不如一行代码。我们将使用Python（因其在AI生态中的广泛使用）从零实现一个MCP服务器。这个服务器将提供两个工具：一个管理简易内存待办列表，另一个模拟查询天气。同时，它还将提供一个资源，用于读取服务器的状态信息。

3.1 项目初始化与依赖选择

首先，创建一个新的项目目录。我们不需要重量级的框架，使用纯Python的asyncio和json库即可处理Stdio的异步读写和JSON-RPC协议。为了简化JSON Schema的定义和验证，我们引入pydantic库，它能让我们的代码更清晰、更健壮。

mkdir mcp-todo-weather-server cd mcp-todo-weather-server python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate pip install pydantic

接下来，创建我们的主文件server.py和协议模型定义文件models.py。

3.2 定义协议数据模型（models.py）

在models.py中，我们使用Pydantic严格定义MCP协议中涉及的核心数据结构。这能确保我们收发的消息格式正确。

# models.py from typing import Any, Dict, List, Optional, Union from enum import Enum from pydantic import BaseModel, Field class JSONRPCRequest(BaseModel): """JSON-RPC 2.0 请求基础模型""" jsonrpc: str = Field(default="2.0") id: Union[int, str, None] method: str params: Optional[Union[Dict[str, Any], List[Any]]] = None class JSONRPCResponse(BaseModel): """JSON-RPC 2.0 成功响应模型""" jsonrpc: str = Field(default="2.0") id: Union[int, str, None] result: Any class JSONRPCError(BaseModel): """JSON-RPC 2.0 错误对象模型""" code: int message: str data: Optional[Any] = None class JSONRPCErrorResponse(BaseModel): """JSON-RPC 2.0 错误响应模型""" jsonrpc: str = Field(default="2.0") id: Union[int, str, None] error: JSONRPCError # MCP 特定模型 class ToolSchema(BaseModel): """工具输入参数的JSON Schema表示（简化版）""" type: str = Field(default="object") properties: Dict[str, Any] required: Optional[List[str]] = None class Tool(BaseModel): """工具定义""" name: str description: str inputSchema: ToolSchema class ResourceContents(BaseModel): """资源内容""" uri: str mimeType: str = Field(default="text/plain") contents: List[Dict[str, str]] # 通常是 [{"text": “内容”}, ...] # 初始化请求/响应参数 class InitializeParams(BaseModel): protocolVersion: str = Field(default="2024-11-05") clientInfo: Optional[Dict[str, str]] = None class InitializeResult(BaseModel): protocolVersion: str = Field(default="2024-11-05") capabilities: Dict[str, Any] = Field(default_factory=dict) serverInfo: Dict[str, str]

这些模型定义了通信的“语言”。JSONRPCRequest/Response是信封，Tool和ResourceContents是信里的具体内容。

3.3 实现服务器主循环与请求分发（server.py）

主服务器的核心是一个异步循环，从sys.stdin读取请求，根据method字段分发给对应的处理函数，然后将结果写入sys.stdout。

# server.py import asyncio import json import sys from typing import Any, Dict, Callable, Awaitable from models import ( JSONRPCRequest, JSONRPCResponse, JSONRPCErrorResponse, InitializeParams, InitializeResult, Tool, ToolSchema, ResourceContents ) class MCPServer: def __init__(self): self._request_handlers = {} self._tools = [] self._todo_list = [] self._register_handlers() self._register_tools_and_resources() def _register_handlers(self): """注册所有支持的JSON-RPC方法处理器""" self._request_handlers = { "initialize": self._handle_initialize, "tools/list": self._handle_tools_list, "tools/call": self._handle_tools_call, "resources/list": self._handle_resources_list, "resources/read": self._handle_resources_read, # “notifications/initialized” 和 “shutdown” 等方法暂未实现 } def _register_tools_and_resources(self): """定义本服务器提供的工具和资源""" # 1. 定义工具 todo_add_tool = Tool( name="add_todo_item", description="向待办列表中添加一个新项目。", inputSchema=ToolSchema( properties={ "task": {"type": "string", "description": "待办事项的描述"} }, required=["task"] ) ) todo_list_tool = Tool( name="list_todo_items", description="列出当前所有的待办事项。", inputSchema=ToolSchema(properties={}) # 此工具无需输入参数 ) weather_tool = Tool( name="get_weather", description="获取指定城市的当前天气信息（模拟）。", inputSchema=ToolSchema( properties={ "city": {"type": "string", "description": "城市名称，例如：北京、上海"} }, required=["city"] ) ) self._tools = [todo_add_tool, todo_list_tool, weather_tool] # 2. 定义资源URI（实际内容在_read_resource中动态生成） self._resource_uris = ["internal://server/status"] async def _handle_initialize(self, params: Dict) -> Dict: """处理初始化请求""" _ = InitializeParams(**params) # 验证参数 return InitializeResult( serverInfo={"name": "Todo-Weather-MCP-Server", "version": "0.1.0"}, capabilities={ "tools": {"listChanged": True}, # 告知客户端工具列表可能变化 "resources": {"listChanged": True} } ).dict() async def _handle_tools_list(self, params: Dict) -> Dict: """返回工具列表""" # params 在此方法中通常为空 return {"tools": [tool.dict() for tool in self._tools]} async def _handle_tools_call(self, params: Dict) -> Dict: """调用具体的工具""" tool_name = params.get("name") arguments = params.get("arguments", {}) if tool_name == "add_todo_item": task = arguments.get("task") if not task: raise ValueError("参数 'task' 是必需的") self._todo_list.append(task) return {"content": [{"type": "text", "text": f"已添加待办：{task}"}]} elif tool_name == "list_todo_items": if not self._todo_list: return {"content": [{"type": "text", "text": "当前待办列表为空。"}]} list_str = "\n".join([f"{i+1}. {item}" for i, item in enumerate(self._todo_list)]) return {"content": [{"type": "text", "text": f"当前待办事项：\n{list_str}"}]} elif tool_name == "get_weather": city = arguments.get("city", "未知城市") # 模拟天气数据 import random weather_conditions = ["晴", "多云", "小雨", "阴天"] temperature = random.randint(15, 30) return {"content": [{"type": "text", "text": f"{city}的当前天气：{random.choice(weather_conditions)}，气温{temperature}摄氏度。"}]} else: raise ValueError(f"未知工具：{tool_name}") async def _handle_resources_list(self, params: Dict) -> Dict: """返回资源列表""" # 这里我们返回一个静态的资源URI列表 resources = [{"uri": uri} for uri in self._resource_uris] return {"resources": resources} async def _handle_resources_read(self, params: Dict) -> Dict: """读取指定资源的内容""" uri = params.get("uri") if uri == "internal://server/status": status_text = f"""服务器状态报告： - 名称：Todo-Weather-MCP-Server - 运行中：是 - 已注册工具数：{len(self._tools)} - 当前待办项数：{len(self._todo_list)} - 最后更新：{asyncio.get_event_loop().time():.2f} """ return ResourceContents(uri=uri, contents=[{"text": status_text}]).dict() else: raise ValueError(f"资源未找到：{uri}") async def _dispatch_request(self, request_data: dict) -> dict: """核心分发器：将请求路由到对应的处理器""" try: request = JSONRPCRequest(**request_data) except Exception as e: # 如果连基本请求格式都无效，返回解析错误 return JSONRPCErrorResponse( id=None, error={"code": -32700, "message": "Parse error", "data": str(e)} ).dict() handler = self._request_handlers.get(request.method) if not handler: return JSONRPCErrorResponse( id=request.id, error={"code": -32601, "message": f"Method not found: {request.method}"} ).dict() try: result = await handler(request.params or {}) return JSONRPCResponse(id=request.id, result=result).dict() except Exception as e: # 处理处理器内部错误 return JSONRPCErrorResponse( id=request.id, error={"code": -32603, "message": "Internal error", "data": str(e)} ).dict() async def run_stdio(self): """通过标准输入输出运行服务器（主循环）""" loop = asyncio.get_event_loop() # 包装标准输入为异步流 reader = asyncio.StreamReader() protocol = asyncio.StreamReaderProtocol(reader) await loop.connect_read_pipe(lambda: protocol, sys.stdin) # 包装标准输出 w_transport, w_protocol = await loop.connect_write_pipe( asyncio.streams.FlowControlMixin, sys.stdout ) writer = asyncio.StreamWriter(w_transport, w_protocol, reader, loop) print("MCP Server (Stdio) 已启动，等待请求...", file=sys.stderr) while True: try: # 读取一行（每个JSON-RPC消息以换行符分隔） line = await reader.readline() if not line: break # EOF line = line.decode('utf-8').strip() if not line: continue request_data = json.loads(line) # 分发并处理请求 response_data = await self._dispatch_request(request_data) # 写入响应 writer.write((json.dumps(response_data) + '\n').encode('utf-8')) await writer.drain() except json.JSONDecodeError as e: # 处理无效JSON error_resp = JSONRPCErrorResponse( id=None, error={"code": -32700, "message": "Parse error", "data": str(e)} ).dict() writer.write((json.dumps(error_resp) + '\n').encode('utf-8')) await writer.drain() except Exception as e: # 捕获其他意外错误，避免服务器崩溃 print(f"服务器内部循环错误：{e}", file=sys.stderr) if __name__ == "__main__": server = MCPServer() asyncio.run(server.run_stdio())

这段代码构建了一个完整的、可运行的MCP服务器核心。MCPServer类管理工具和资源的注册，并通过_dispatch_request方法将收到的JSON-RPC请求路由到具体的处理函数。run_stdio方法建立了与标准输入输出的异步连接，构成了服务器的主事件循环。

3.4 关键实现细节与避坑指南

在实现过程中，有几个细节至关重要，直接关系到服务器能否与客户端正常对话：

消息分隔符：MCP over Stdio协议规定，每条JSON-RPC消息必须独占一行，以换行符(\n)分隔。我们的代码使用await reader.readline()来读取，并用\n来写入响应。忘记换行符是导致客户端解析失败的最常见原因。
错误处理标准化：JSON-RPC 2.0有预定义的错误码。例如，-32601表示方法未找到，-32603是内部错误。我们必须捕获所有异常，并将其转化为标准错误响应格式返回，而不是让进程崩溃或输出非格式化的错误信息到stdout（这会被客户端认为是响应内容）。
输入验证：我们使用Pydantic模型在入口处验证请求格式。但在工具的具体实现（如_handle_tools_call）中，仍需手动检查参数是否存在、类型是否正确。永远不要信任客户端的输入。
资源内容格式：resources/read返回的contents字段是一个列表，其中每个元素是一个包含type和text（或image等）的字典。对于纯文本，我们使用{"type": "text", "text": “内容”}。这是客户端（尤其是AI模型）期望的格式。

实操心得：在开发调试阶段，一个非常有效的方法是同时运行服务器和一个简单的测试客户端脚本。测试客户端模拟Code编辑器的行为，发送初始化请求、列出工具、调用工具，并打印服务器的响应。这能帮你快速定位是协议格式问题、逻辑错误还是通信问题。不要一开始就尝试与复杂的IDE集成。

4. 从本地测试到IDE集成：让工具真正“活”起来

服务器写好了，但它现在还只是一个孤立的进程。我们需要让它被AI客户端识别和使用。这里我们分两步走：先进行本地手动测试验证基本功能，再集成到真实的AI辅助编程环境中。

4.1 手动测试：模拟客户端验证协议

创建一个test_client.py脚本，通过子进程启动我们的服务器，并模拟发送协议消息。

# test_client.py import subprocess import json import time def send_request(process, request): """向服务器进程发送一个JSON-RPC请求""" message = json.dumps(request) + '\n' process.stdin.write(message.encode('utf-8')) process.stdin.flush() def read_response(process): """从服务器进程读取一行响应""" line = process.stdout.readline().decode('utf-8').strip() return json.loads(line) if line else None # 启动服务器进程 server_proc = subprocess.Popen( ['python', 'server.py'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) try: # 1. 发送初始化请求 init_request = { "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2024-11-05", "clientInfo": {"name": "TestClient"} } } send_request(server_proc, init_request) init_response = read_response(server_proc) print("初始化响应:", json.dumps(init_response, indent=2, ensure_ascii=False)) # 2. 列出所有工具 list_request = {"jsonrpc": "2.0", "id": 2, "method": "tools/list"} send_request(server_proc, list_request) list_response = read_response(server_proc) print("\n工具列表响应:", json.dumps(list_response, indent=2, ensure_ascii=False)) # 3. 调用添加待办工具 call_request = { "jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": { "name": "add_todo_item", "arguments": {"task": "学习MCP协议"} } } send_request(server_proc, call_request) call_response = read_response(server_proc) print("\n调用工具响应:", json.dumps(call_response, indent=2, ensure_ascii=False)) # 4. 再次列出待办，确认添加成功 list_todo_request = { "jsonrpc": "2.0", "id": 4, "method": "tools/call", "params": { "name": "list_todo_items", "arguments": {} } } send_request(server_proc, list_todo_request) list_todo_response = read_response(server_proc) print("\n列出待办响应:", json.dumps(list_todo_response, indent=2, ensure_ascii=False)) # 5. 读取资源 read_resource_request = { "jsonrpc": "2.0", "id": 5, "method": "resources/read", "params": {"uri": "internal://server/status"} } send_request(server_proc, read_resource_request) resource_response = read_response(server_proc) print("\n读取资源响应:", json.dumps(resource_response, indent=2, ensure_ascii=False)) except Exception as e: print(f"测试过程中发生错误: {e}") finally: # 读取可能的错误输出 stderr_output = server_proc.stderr.read().decode('utf-8') if stderr_output: print("\n服务器标准错误输出:", stderr_output) server_proc.terminate() server_proc.wait()

运行这个测试脚本(python test_client.py)，你应该能看到一系列格式规范的JSON响应。这证明了你的服务器协议层工作正常。

4.2 集成到Claude Code：配置与验证

Claude Code（或Cursor等支持MCP的编辑器）通常通过一个配置文件来声明本地的MCP服务器。以Claude Code为例，配置文件通常位于~/.config/claude/claude_desktop_config.json（Linux/macOS）或%APPDATA%\Claude\claude_desktop_config.json（Windows）。

你需要编辑这个文件，添加你的服务器配置：

{ "mcpServers": { "my-todo-weather-server": { "command": "python", "args": ["/绝对路径/to/your/mcp-todo-weather-server/server.py"], "env": { "PYTHONPATH": "/绝对路径/to/your/mcp-todo-weather-server" } } } }

关键配置解析：

command: 启动服务器的命令，这里是python。
args: 命令的参数列表，第一个是脚本的绝对路径。使用绝对路径可以避免很多因工作目录引起的找不到模块的问题。
env: 可选的环境变量。如果你的服务器脚本依赖其他本地模块或环境，可以在这里设置PYTHONPATH。

保存配置后，重启Claude Code。重启后，你可以通过一些方式来验证集成是否成功：

在聊天框中，直接询问AI助手：“你能使用哪些工具？” 它应该会列出add_todo_item,list_todo_items,get_weather。
尝试发出指令：“请帮我添加一个待办事项：写项目周报。” AI应该会调用工具并返回成功信息。
检查Claude Code的日志或开发者工具（如果有），查看是否有服务器启动或通信错误。

踩坑实录：集成失败最常见的原因有三个。第一，路径错误：确保command和args中的路径完全正确，特别是在Windows上注意反斜杠转义或使用双引号。第二，权限问题：确保脚本有可执行权限（在Unix系统上可能需要chmod +x server.py，或者通过python解释器执行）。第三，依赖缺失：服务器脚本在独立运行时可能因缺少pydantic等库而崩溃。确保你的Python环境已安装所有依赖，或者在配置中使用虚拟环境的Python解释器绝对路径（如“/path/to/venv/bin/python”）。

4.3 进阶思考：从玩具到生产

我们实现了一个基础但功能完整的服务器。但在生产环境中，还需要考虑更多：

状态管理：我们的待办列表存储在内存中，服务器重启即丢失。生产环境需要连接数据库（如SQLite、PostgreSQL）或外部存储服务。
身份验证与授权：如果工具涉及敏感操作（如数据库写入、调用付费API），服务器必须实现身份验证。MCP协议本身不强制规定，但你可以通过初始化参数传递令牌，或在工具调用时验证上下文。
性能与并发：我们的简单服务器是顺序处理请求的。对于高并发场景，需要考虑使用asyncio的并发特性，或者采用多进程/线程模型，确保一个耗时工具调用不会阻塞其他请求。
更复杂的工具：工具可以返回更结构化的内容，例如混合文本和图片，甚至引导用户进行多步交互（虽然当前MCP协议对复杂交互的支持还在演进中）。
动态工具注册：我们的工具是启动时静态注册的。更高级的服务器可以根据配置或外部事件动态添加或移除工具。

5. MCP与LangChain工具调用的深度对比

在文章开头，我提到了用LangChain直接封装工具的痛点。现在，有了MCP的实践经验，我们可以更具体地对比两者的差异，这也是面试中常被问到的点。

LangChain Tool 调用：

定位：LangChain是一个用于构建LLM应用框架，其Tool抽象是框架内的一个组件。
集成方式：工具与LangChain的Agent、Chain强耦合。你需要将工具实例注册到特定的Agent类型（如create_react_agent）中。
通信：工具调用发生在同一个Python进程内，是函数调用。
优点：开发快捷，与LangChain生态无缝集成，适合快速原型验证和单一应用场景。
缺点：紧耦合。你的工具逻辑、LangChain版本、Agent逻辑绑定在一起。很难将同一套工具复用到另一个不基于LangChain的项目或不同的客户端（如一个独立的CLI工具或另一个AI平台）。

MCP 工具调用：

定位：MCP是一个协议，关注于工具能力的标准化暴露和通信。
集成方式：工具位于独立的MCP服务器进程中。客户端（如Claude Code、你的自定义AI应用）通过标准协议与服务器通信。
通信：进程间通信（IPC）或网络通信（HTTP）。是跨进程/网络的RPC调用。
优点：彻底解耦。工具服务器独立开发、部署、升级。任何支持MCP协议的客户端都能立即使用所有工具，无需修改客户端代码。实现了“一次编写，处处可用”。
缺点：引入额外的通信开销（序列化/反序列化、网络延迟），架构稍复杂，需要处理进程生命周期和通信错误。

核心区别与选择建议：可以把LangChain Tools看作“本地库”，而MCP Tools则是“微服务”。如果你的工具只服务于一个特定的、用LangChain构建的AI应用，且没有跨平台共享的需求，直接用LangChain最简单。但如果你希望构建一套能在不同AI助手（Claude、GPTs、自定义前端）、不同环境（本地IDE、云平台）中共享的通用工具集，那么投入时间构建MCP服务器是极具长期价值的投资。MCP解决的是工具生态的“标准化”和“可移植性”问题。

回到我们实现的服务器，它现在可以被Claude Code使用。未来，你可以用同样的服务器配置，让Windmill、Codeium等任何支持MCP的客户端获得管理待办和查询天气的能力，而无需为每个客户端重写一遍工具逻辑。这就是MCP从“面试概念”落地为“生产力工具”的真正价值。