拓冰网站定制 
1. 从面试八股到实战工具我为什么重新审视MCP最近在准备面试或者和同行交流大模型应用开发时MCPModel 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与PromptsMCP服务器主要向客户端暴露三种类型的“能力”工具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(default2.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(default2.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(default2.0) id: Union[int, str, None] error: JSONRPCError # MCP 特定模型 class ToolSchema(BaseModel): 工具输入参数的JSON Schema表示简化版 type: str Field(defaultobject) 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(defaulttext/plain) contents: List[Dict[str, str]] # 通常是 [{text: “内容”}, ...] # 初始化请求/响应参数 class InitializeParams(BaseModel): protocolVersion: str Field(default2024-11-05) clientInfo: Optional[Dict[str, str]] None class InitializeResult(BaseModel): protocolVersion: str Field(default2024-11-05) capabilities: Dict[str, Any] Field(default_factorydict) 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( nameadd_todo_item, description向待办列表中添加一个新项目。, inputSchemaToolSchema( properties{ task: {type: string, description: 待办事项的描述} }, required[task] ) ) todo_list_tool Tool( namelist_todo_items, description列出当前所有的待办事项。, inputSchemaToolSchema(properties{}) # 此工具无需输入参数 ) weather_tool Tool( nameget_weather, description获取指定城市的当前天气信息模拟。, inputSchemaToolSchema( 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{i1}. {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(uriuri, 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( idNone, error{code: -32700, message: Parse error, data: str(e)} ).dict() handler self._request_handlers.get(request.method) if not handler: return JSONRPCErrorResponse( idrequest.id, error{code: -32601, message: fMethod not found: {request.method}} ).dict() try: result await handler(request.params or {}) return JSONRPCResponse(idrequest.id, resultresult).dict() except Exception as e: # 处理处理器内部错误 return JSONRPCErrorResponse( idrequest.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) 已启动等待请求..., filesys.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( idNone, 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}, filesys.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], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.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, indent2, ensure_asciiFalse)) # 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, indent2, ensure_asciiFalse)) # 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, indent2, ensure_asciiFalse)) # 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, indent2, ensure_asciiFalse)) # 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, indent2, ensure_asciiFalse)) 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.jsonLinux/macOS或%APPDATA%\Claude\claude_desktop_config.jsonWindows。你需要编辑这个文件添加你的服务器配置{ 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从“面试概念”落地为“生产力工具”的真正价值。 从概念到实践：手把手实现MCP服务器，解决AI工具集成难题
1. 从面试八股到实战工具我为什么重新审视MCP最近在准备面试或者和同行交流大模型应用开发时MCPModel 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与PromptsMCP服务器主要向客户端暴露三种类型的“能力”工具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(default2.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(default2.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(default2.0) id: Union[int, str, None] error: JSONRPCError # MCP 特定模型 class ToolSchema(BaseModel): 工具输入参数的JSON Schema表示简化版 type: str Field(defaultobject) 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(defaulttext/plain) contents: List[Dict[str, str]] # 通常是 [{text: “内容”}, ...] # 初始化请求/响应参数 class InitializeParams(BaseModel): protocolVersion: str Field(default2024-11-05) clientInfo: Optional[Dict[str, str]] None class InitializeResult(BaseModel): protocolVersion: str Field(default2024-11-05) capabilities: Dict[str, Any] Field(default_factorydict) 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( nameadd_todo_item, description向待办列表中添加一个新项目。, inputSchemaToolSchema( properties{ task: {type: string, description: 待办事项的描述} }, required[task] ) ) todo_list_tool Tool( namelist_todo_items, description列出当前所有的待办事项。, inputSchemaToolSchema(properties{}) # 此工具无需输入参数 ) weather_tool Tool( nameget_weather, description获取指定城市的当前天气信息模拟。, inputSchemaToolSchema( 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{i1}. {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(uriuri, 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( idNone, error{code: -32700, message: Parse error, data: str(e)} ).dict() handler self._request_handlers.get(request.method) if not handler: return JSONRPCErrorResponse( idrequest.id, error{code: -32601, message: fMethod not found: {request.method}} ).dict() try: result await handler(request.params or {}) return JSONRPCResponse(idrequest.id, resultresult).dict() except Exception as e: # 处理处理器内部错误 return JSONRPCErrorResponse( idrequest.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) 已启动等待请求..., filesys.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( idNone, 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}, filesys.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], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.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, indent2, ensure_asciiFalse)) # 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, indent2, ensure_asciiFalse)) # 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, indent2, ensure_asciiFalse)) # 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, indent2, ensure_asciiFalse)) # 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, indent2, ensure_asciiFalse)) 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.jsonLinux/macOS或%APPDATA%\Claude\claude_desktop_config.jsonWindows。你需要编辑这个文件添加你的服务器配置{ 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从“面试概念”落地为“生产力工具”的真正价值。