1. 项目概述从理论到实战亲手解剖MCP协议的数据包最近Model Context Protocol (MCP) 在AI和LLM工具集成领域火得一塌糊涂。各种架构图、白皮书和概念解析文章满天飞仿佛人人都成了协议专家。但作为一个在软件工程一线摸爬滚打多年的老手我总觉得少了点什么。大家都在高谈阔论“双向通信”、“能力协商”、“上下文管理”这些宏大概念却没人愿意掀开引擎盖给你看看里面到底是怎么转的。这就像有人滔滔不绝地给你讲解汽车的空气动力学原理却从不让你看一眼发动机的内部构造。对于工程师来说这种隔靴搔痒的感觉实在难受。所以我决定干点“脏活累水”。这篇文章我们不谈哲学只看数据包。我们将彻底抛开那些精美的示意图和抽象的理论直接深入到比特和字节的层面亲手构建一个最原始的MCP客户端并通过标准输入输出与一个真实的MCP服务器对话。我们的目标只有一个亲眼看看在网络上流动的究竟是怎样的JSON。通过这个“外科手术”式的实践你不仅能透彻理解MCP的核心机制更能获得一种“我能从头构建它”的底气和自信。无论你是想深度集成AI能力到现有系统还是仅仅好奇协议底层如何工作这篇从零开始的实战指南都将为你提供最直接的洞察。2. 核心思路拆解为什么选择“裸”协议分析在开始动手之前我们先明确一下这次探索的独特视角和价值。市面上大多数MCP教程都基于官方SDK或高级框架这当然能快速上手但也像用自动挡学开车——你知道了怎么走但不知道离合器、变速箱和发动机是如何协同工作的。2.1 超越SDK理解协议的“第一性原理”使用官方SDK比如JavaScript或Python的MCP库无疑是最便捷的方式。它们封装了连接管理、消息序列化、错误处理等繁琐细节让你可以专注于业务逻辑。然而这种便利性也带来了一层抽象屏蔽了协议最本质的通信模式。当出现网络问题、版本不兼容或需要深度定制时这层抽象就可能变成黑箱让你无从下手。我们的方法截然不同我们将仅使用Python标准库的subprocess和json模块手动构造每一个JSON-RPC消息并通过管道发送。这相当于我们直接在与协议的“语法”对话跳过了所有“编译器”和“运行时”的中间层。这样做有几个不可替代的好处绝对透明每一个字节的发送和接收都完全在你的控制之下没有任何魔法。你可以清晰地看到请求是如何构建的响应是如何解析的。深度理解通过手动实现最基本的握手和工具调用流程你会对MCP的状态机、错误处理边界和设计哲学有肌肉记忆般的理解。调试利器当你使用高级SDK遇到诡异问题时拥有底层协议知识能让你快速定位问题是在应用层、SDK层还是协议层。你可以自己写一个小脚本模拟客户端行为直接测试服务器是否正常。不受限制你不受任何特定SDK版本、设计模式或依赖项的束缚。你可以用任何语言、在任何环境中实现MCP客户端只要它能启动子进程、读写JSON和管道。2.2 聚焦核心工具调用是MCP的“杀手锏”MCP协议规范涵盖了不少内容初始化、资源Resources、提示词Prompts、工具Tools等。但对于绝大多数实际应用场景——尤其是让LLM能够调用外部功能——工具调用Tool Calling是绝对的核心和起点。我们可以暂时忽略资源管理和提示词模板这些高级特性将注意力完全集中在工具调用的生命周期上。这样一来整个复杂的协议就被简化成了一个极其清晰的对话模型客户端“你好我支持MCP协议版本是X这是我的基本信息。”initialize服务器“你好我是XX服务器版本是Y我支持这些能力。”initialized响应客户端“好的我准备好了开始吧。”notifications/initialized客户端“你都能做什么”tools/list服务器“我能做A、B、C这几件事。”tools/list响应客户端“请帮我做A这是参数。”tools/call服务器“做完了这是结果。”tools/call响应看只需要理解这七种消息的交换你就掌握了MCP 80%以上的实用价值。我们的实验也将严格遵循这个精简流程。2.3 实验环境与目标服务器为了进行这次解剖我们需要一个“标本”——一个简单、稳定、开源的MCP服务器。官方提供的weather.py示例服务器是绝佳的选择。它提供了两个工具get_forecast获取天气预报和get_alerts获取天气警报。功能具体代码清晰且完全符合MCP规范。我们的实验目标非常明确手动启动这个天气服务器进程。不使用任何MCP库仅用标准库与其建立通信。完整走通上述7步对话并捕获每一个环节在“线上”即stdin/stdout管道传输的原始JSON数据。分析这些数据包的结构、字段含义和设计意图。注意选择stdin/stdout作为传输层是MCP的常见实践尤其对于本地集成的场景。它简单、高效且避免了网络端口、认证等复杂性。虽然有人诟病其健壮性但对于进程间通信IPC和快速原型来说它是无可替代的利器。我们的实验正是基于这种模式。3. 实战准备搭建你的数字解剖台理论铺垫完毕现在进入实战环节。请确保你有一个可用的Python环境3.7以上我们将从零开始搭建实验环境。3.1 获取并安装天气服务器首先我们需要获取MCP天气服务器的代码。官方示例存放在GitHub上。# 创建一个专门的工作目录 mkdir mcp-wire-dissection cd mcp-wire-dissection # 下载天气服务器示例代码 curl -O https://raw.githubusercontent.com/modelcontextprotocol/quickstart-resources/main/weather-server-python/weather.py # 查看文件确认下载成功 head -20 weather.py这个weather.py文件就是一个完整的MCP服务器实现。它使用了mcp这个Python SDK。因此我们需要安装其依赖。推荐使用现代Python包管理工具uv它比传统的pip更快、更可靠。# 安装 uv (如果尚未安装) # 在MacOS/Linux上 curl -LsSf https://astral.sh/uv/install.sh | sh # 初始化项目并安装依赖 uv init . uv add mcp实操心得使用uv而不是pip和venv能大幅提升依赖管理效率。uv不仅安装极快还能生成精确的锁文件确保环境一致性。对于这类一次性实验项目它能帮你省去很多配置虚拟环境的麻烦。3.2 理解服务器启动方式查看weather.py的末尾通常你会看到类似这样的代码if __name__ __main__: # 使用 mcp 库的 run 函数启动服务器 mcp.run(server)这意味着我们可以直接用python weather.py来启动它。但更常见的做法是使用uv run它能确保在正确的、隔离的依赖环境中运行脚本。这也是我们后续在客户端代码中启动子进程的方式。你可以先手动测试一下服务器是否能正常运行uv run weather.py如果一切正常服务器会启动并等待标准输入上的连接。此时它不会输出任何内容因为还没有客户端连接你可以按CtrlC终止它。这个“安静”的启动是正常的MCP服务器通常以这种“沉默的守护进程”模式运行。4. 核心环节一建立连接与初始化握手现在让我们开始编写我们的“外科医生”——一个纯手工的MCP客户端。我们将创建一个新的Python脚本bare_mcp_client.py。4.1 启动服务器子进程我们使用subprocess.Popen来启动服务器并捕获其标准输入、输出和错误流。这是实现进程间通信的基础。# bare_mcp_client.py import subprocess import json from pprint import pprint import time def start_mcp_server(server_script_path): 启动MCP服务器作为子进程。 返回一个Popen对象通过其stdin/stdout进行通信。 # 关键参数解析 # - stdinsubprocess.PIPE: 我们可以向进程的stdin写入数据我们的请求 # - stdoutsubprocess.PIPE: 我们可以从进程的stdout读取数据服务器的响应 # - stderrsubprocess.DEVNULL: 将服务器的错误输出重定向到空设备避免干扰我们的主控制台。 # 在实际调试时你可以改为 stderrsubprocess.PIPE 来捕获错误日志。 # - [uv, run, ...]: 使用uv在正确的环境中运行脚本。 process subprocess.Popen( [uv, run, server_script_path], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.DEVNULL, # 或 subprocess.PIPE 用于调试 textFalse, # 我们处理字节而不是字符串以获得更精确的控制 bufsize1, # 行缓冲确保每条消息能及时发送 ) print(f[INFO] 服务器进程已启动PID: {process.pid}) return process注意事项将stderr重定向到DEVNULL在实验阶段是干净的但生产环境或调试时一定要捕获并处理stderr。服务器可能会将重要的启动错误或运行时日志输出到标准错误流。4.2 定义消息收发函数MCP over stdio 的约定是newline-delimited JSON (NDJSON)。即每条完整的JSON消息独占一行以换行符\n分隔。这是许多流式JSON协议如JSON-RPC over HTTP/1.1的某些实现的常见模式。def send_mcp_message(process, message_dict): 向MCP服务器进程发送一条JSON-RPC消息。 消息被序列化为JSON字符串末尾添加换行符然后编码为字节流写入stdin。 # 1. 将字典转换为JSON字符串 json_str json.dumps(message_dict) # 2. 添加NDJSON要求的换行符 json_line json_str \n # 3. 编码为字节并写入进程的stdin process.stdin.write(json_line.encode(utf-8)) # 4. 刷新缓冲区确保数据立即发送 process.stdin.flush() print(f[SENT] {json_str[:200]}...) # 打印前200个字符便于观察 def read_mcp_message(process): 从MCP服务器进程的stdout读取一条JSON-RPC消息。 读取一行解码为字符串然后解析为Python字典。 这是一个阻塞调用会等待服务器输出新的一行。 # 读取一行字节数据 line_bytes process.stdout.readline() if not line_bytes: # 如果读到空字节可能意味着管道已关闭或进程结束 return None # 解码为UTF-8字符串并去除末尾的换行符 line_str line_bytes.decode(utf-8).rstrip(\n) if line_str: print(f[RECV] 原始行长度: {len(line_str)} 字符) # 解析JSON message json.loads(line_str) return message return None4.3 执行初始化握手现在让我们发起第一次对话。根据MCP规范连接建立后的第一条消息必须是initialize请求。def perform_handshake(process, client_nameBareMCPClient, client_version0.1.0): 执行MCP协议初始化握手。 返回服务器返回的初始化结果。 # 步骤1: 客户端发送 initialize 请求 init_request { jsonrpc: 2.0, id: 1, # 请求ID用于匹配响应 method: initialize, params: { protocolVersion: 2025-03-26, # 必须与服务器兼容的协议版本 capabilities: {}, # 客户端声明自己支持的能力这里为空 clientInfo: { name: client_name, version: client_version } } } print(\n 步骤 1: 发送 initialize 请求 ) send_mcp_message(process, init_request) # 步骤2: 读取服务器的 initialize 响应 print(\n 等待服务器 initialize 响应 ) init_response read_mcp_message(process) if not init_response: raise ConnectionError(服务器未响应 initialize 请求) print(服务器初始化响应:) pprint(init_response, depth3) # 限制打印深度避免信息过载 # 步骤3: 客户端发送 initialized 通知 # 这是一个“通知”notification没有id字段服务器不回复。 initialized_notification { jsonrpc: 2.0, method: notifications/initialized, params: {} # 参数为空对象 } print(\n 步骤 2: 发送 initialized 通知 ) send_mcp_message(process, initialized_notification) # 发送通知后没有响应需要读取 return init_response让我们逐行分析这个握手过程initialize请求jsonrpc: 2.0: 声明我们使用JSON-RPC 2.0协议。这是MCP的底层传输协议。id: 1: 每个请求需要一个唯一ID服务器会在对应的响应中带回相同的ID。这对于异步通信至关重要。method: initialize: 指定要调用的方法。params: 包含握手参数。protocolVersion: 指定我们希望使用的MCP协议版本。必须与服务器支持的版本匹配否则握手会失败。capabilities: 客户端向服务器宣告自己支持哪些MCP扩展功能如资源订阅、提示词变更通知等。我们这里留空表示只支持基础功能。clientInfo: 客户端的身份信息方便服务器日志记录和识别。服务器initialize响应服务器会返回一个同样包含id: 1 的JSON-RPC响应。result字段中包含服务器的信息如serverInfo名称、版本和capabilities服务器支持的能力。这是我们第一次了解服务器特性的机会。notifications/initialized通知这是一个通知而不是请求。JSON-RPC中通知的id字段为null或直接省略。它用于告知服务器“客户端已准备就绪可以开始正式工作”。服务器收到后不会发送任何响应。这完成了握手的最后一步。运行这部分代码你会看到类似以下的输出具体内容因服务器版本而异[INFO] 服务器进程已启动PID: 12345 步骤 1: 发送 initialize 请求 [SENT] {jsonrpc: 2.0, id: 1, method: initialize, params: {protocolVersion: 2025-03-26, capabilities: {}, clientInfo: {name: BareMCPClient, version: 0.1.0}}}... 等待服务器 initialize 响应 [RECV] 原始行长度: 266 字符 服务器初始化响应: {id: 1, jsonrpc: 2.0, result: {capabilities: {experimental: {}, prompts: {listChanged: False}, resources: {listChanged: False, subscribe: False}, tools: {listChanged: False}}, protocolVersion: 2025-03-26, serverInfo: {name: weather, version: 1.9.4}}} 步骤 2: 发送 initialized 通知 [SENT] {jsonrpc: 2.0, method: notifications/initialized, params: {}}...恭喜你已经成功完成了MCP连接握手。你看到了服务器返回的原始JSON知道了它的名字、版本以及它声明不支持动态工具列表变更 (listChanged: False)。这就是“线上”流动的第一个关键数据包。5. 核心环节二发现与调用工具握手成功后客户端和服务器就进入了工作状态。接下来客户端需要知道服务器能做什么。5.1 查询可用工具列表我们发送tools/list请求来获取服务器暴露的所有工具。def list_available_tools(process): 向服务器请求可用的工具列表。 list_tools_request { jsonrpc: 2.0, id: 2, # 新的请求ID method: tools/list, params: {} # 此请求通常不需要额外参数 } print(\n 步骤 3: 查询工具列表 (tools/list) ) send_mcp_message(process, list_tools_request) print(\n 等待工具列表响应 ) list_response read_mcp_message(process) if not list_response: raise ConnectionError(服务器未响应 tools/list 请求) # 检查是否有错误 if error in list_response: print(f查询工具列表失败: {list_response[error]}) return None tools list_response.get(result, {}).get(tools, []) print(f发现 {len(tools)} 个工具:) for tool in tools: print(f - 名称: {tool[name]}) print(f 描述: {tool.get(description, N/A)[:100]}...) # 截断长描述 # 打印输入参数模式 schema tool.get(inputSchema, {}) req schema.get(required, []) props schema.get(properties, {}) if req or props: print(f 输入参数: {list(props.keys())} (必填: {req})) print() return tools运行后你会看到服务器返回的详细工具列表。以天气服务器为例输出可能如下 步骤 3: 查询工具列表 (tools/list) [SENT] {jsonrpc: 2.0, id: 2, method: tools/list, params: {}}... 等待工具列表响应 [RECV] 原始行长度: 732 字符 发现 2 个工具: - 名称: get_alerts 描述: Get weather alerts for a US state. Args: state: Two-letter US state code (e.g. CA, NY)... 输入参数: [state] (必填: [state]) - 名称: get_forecast 描述: Get weather forecast for a location. Args: latitude: Latitude of the location longitude:... 输入参数: [latitude, longitude] (必填: [latitude, longitude])关键字段解析name: 工具的标识符在调用时必须精确匹配。description: 给LLM看的自然语言描述。这是工具能否被正确调用的关键。LLM根据这段描述来决定是否以及如何使用这个工具。inputSchema: 一个遵循JSON Schema的对象严格定义了调用此工具所需的参数名称、类型和是否必需。这是实现强类型接口和自动验证的基础。实操心得description字段的撰写是一门艺术。它需要足够清晰让LLM能准确理解工具的用途又不能过于冗长以免占用过多上下文窗口。一个好的描述应包含1) 工具的核心功能2) 每个参数的含义和示例3) 可能返回的内容。这是你作为服务器开发者与LLM“沟通”的主要渠道。5.2 构造并发送工具调用请求现在我们知道了服务器有get_alerts工具它需要一个state参数美国州代码。让我们调用它来获取德克萨斯州TX的天气警报。def call_mcp_tool(process, tool_name, arguments, request_id3): 调用指定的MCP工具。 call_request { jsonrpc: 2.0, id: request_id, method: tools/call, params: { name: tool_name, # 必须与 tools/list 返回的 name 完全一致 arguments: arguments # 参数必须符合 inputSchema 的定义 } } print(f\n 步骤 4: 调用工具 {tool_name} ) print(f 参数: {arguments}) send_mcp_message(process, call_request) print(\n 等待工具调用响应 ) call_response read_mcp_message(process) if not call_response: raise ConnectionError(f服务器未响应 tools/call 请求 (id: {request_id})) # 检查响应中是否包含错误 if error in call_response: error call_response[error] print(f[ERROR] 工具调用失败!) print(f 错误码: {error.get(code)}) print(f 消息: {error.get(message)}) if data in error: print(f 详情: {error.get(data)}) return None # 解析成功的结果 result call_response.get(result, {}) is_error result.get(isError, False) content result.get(content, []) print(f调用结果: isError{is_error}) print(f返回了 {len(content)} 个内容块) for i, block in enumerate(content): block_type block.get(type, unknown) print(f\n--- 内容块 {i1} (类型: {block_type}) ---) if block_type text: text block.get(text, ) # 对于长文本只预览开头和结尾 if len(text) 500: print(text[:250] ...\n...\n text[-250:]) else: print(text) elif block_type image: # 如果是图片可能包含数据URI或引用 print(f图片数据 (已省略): {block.get(data, N/A)[:100]}...) # 可以处理其他类型如 audio, video, resource 等 return result现在在主函数中调用它# 在主流程中握手和获取工具列表之后... tools list_available_tools(process) if tools: # 假设我们调用 get_alerts tool_to_call get_alerts # 确保工具存在 if any(t[name] tool_to_call for t in tools): # 准备参数 tool_args {state: TX} # 获取德克萨斯州的警报 result call_mcp_tool(process, tool_to_call, tool_args, request_id3) if result: print(\n✅ 工具调用成功) else: print(\n❌ 工具调用失败或返回错误。) else: print(f\n工具 {tool_to_call} 不在可用列表中。)运行这段代码你将见证奇迹或者说一次标准的API调用。客户端发送一个紧凑的请求服务器返回可能非常庞大的天气警报数据。 步骤 4: 调用工具 get_alerts 参数: {state: TX} [SENT] {jsonrpc: 2.0, id: 3, method: tools/call, params: {name: get_alerts, arguments: {state: TX}}}... 等待工具调用响应 [RECV] 原始行长度: 51305 字符 调用结果: isErrorFalse 返回了 1 个内容块 --- 内容块 1 (类型: text) --- Event: Flood Advisory Area: Hidalgo, TX; Willacy, TX Severity: Minor Description: * WHAT...Flooding caused by excessive rainfall continues. ... Instructions: Turn around, dont drown when encountering flooded roads... The next statement will be issued Tuesday morning at 830 AM CDT. ✅ 工具调用成功响应结构深度解析 你收到的响应是一个标准的JSON-RPC成功响应。核心在于result字段isError:false表示调用成功。如果是true则content中可能包含错误信息。content: 一个数组包含一个或多个“内容块”。每个块有type和具体数据。type: text: 这是最常见的类型数据在text字段中是一个字符串。注意这个字符串内部可以包含任何结构化的文本比如我们看到的格式化天气警报。LLM擅长从这种半结构化的自然语言文本中提取信息。其他类型MCP规范还支持image、audio、video等类型用于多模态输出。resource类型则用于引用服务器管理的资源如文件。关键洞察MCP工具调用的响应设计非常通用。content数组允许服务器返回多种类型、多个部分的结果。text类型的灵活性极高它可以是纯文本、JSON字符串、CSV、HTML甚至是自定义的标记语言。这种设计将“数据呈现”的复杂性留给了服务器和客户端或LLM协议本身只负责传输。这也是为什么你很少在工具定义中看到outputSchema——输出格式是隐含在content的类型和结构中的。6. 核心环节三完整流程整合与错误处理我们已经拆解了每一个步骤。现在让我们把它们组合成一个完整、健壮的客户端脚本并加入必要的错误处理和资源清理。6.1 完整的客户端脚本# bare_mcp_client.py (完整版) import subprocess import json import sys from pprint import pprint class BareMCPClient: def __init__(self, server_script_path): self.server_path server_script_path self.process None self.next_request_id 1 # 用于生成唯一的请求ID def start(self): 启动MCP服务器进程 print(f[启动] 正在启动MCP服务器: {self.server_path}) try: self.process subprocess.Popen( [uv, run, self.server_path], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, # 改为PIPE以捕获错误 textFalse, bufsize1, ) # 启动后先读一下stderr看是否有立即报错 self._drain_stderr() print(f[启动] 服务器进程已启动 (PID: {self.process.pid})) return True except FileNotFoundError as e: print(f[错误] 无法找到服务器脚本或uv命令: {e}) print(f 请确保: 1) 脚本路径正确; 2) uv已安装; 3) 依赖已安装 (uv add mcp)) return False except Exception as e: print(f[错误] 启动服务器时发生未知错误: {e}) return False def _drain_stderr(self): 非阻塞地读取并打印服务器的错误输出 import select if self.process and self.process.stderr: # 检查stderr是否有内容可读非阻塞 rlist, _, _ select.select([self.process.stderr], [], [], 0.1) if rlist: err_output self.process.stderr.read().decode(utf-8, errorsignore) if err_output: print(f[服务器STDERR] {err_output.strip()}) def send(self, method, paramsNone, is_notificationFalse): 发送JSON-RPC请求或通知 if not self.process or self.process.poll() is not None: raise ConnectionError(服务器进程未运行或已终止) request { jsonrpc: 2.0, method: method, } if not is_notification: request[id] self.next_request_id self.next_request_id 1 if params: request[params] params json_line json.dumps(request) \n try: self.process.stdin.write(json_line.encode(utf-8)) self.process.stdin.flush() print(f[发送] {method} (ID: {request.get(id, N/A-通知)})) # 打印精简的请求内容 if params: preview str(params)[:100] if len(str(params)) 100: preview ... print(f 参数: {preview}) return request.get(id) # 返回请求ID用于匹配响应 except BrokenPipeError: print(f[错误] 向服务器写入失败管道已断开) self.stop() raise ConnectionError(与服务器的连接已断开) def receive(self): 从服务器接收一条JSON-RPC消息 if not self.process: return None # 先检查一下stderr看是否有错误输出 self._drain_stderr() try: line_bytes self.process.stdout.readline() if not line_bytes: print([接收] 读到空字节服务器可能已关闭输出流) return None line_str line_bytes.decode(utf-8, errorsreplace).rstrip(\n) if not line_str: return None print(f[接收] 原始数据长度: {len(line_str)} 字符) return json.loads(line_str) except json.JSONDecodeError as e: print(f[错误] 解析服务器响应JSON失败: {e}) print(f 原始行: {line_str[:200]}...) return None except Exception as e: print(f[错误] 读取服务器响应时发生未知错误: {e}) return None def call_and_wait(self, method, paramsNone): 发送请求并等待响应返回响应字典 req_id self.send(method, params, is_notificationFalse) if req_id is None: return None # 循环读取直到找到对应ID的响应或遇到错误/通知 while True: response self.receive() if response is None: print(f[错误] 在等待响应 {req_id} 时连接断开) return None # 如果是通知无id打印并继续等 if id not in response: print(f[接收] 收到通知: {response.get(method, unknown)}) continue # 找到对应ID的响应 if response.get(id) req_id: return response # ID不匹配可能是之前的响应未处理或服务器乱序MCP通常不会 print(f[警告] 收到不匹配的响应ID: 期望 {req_id}, 收到 {response.get(id)}) def stop(self): 停止服务器进程 if self.process: print([停止] 正在终止服务器进程...) try: self.process.terminate() # 发送SIGTERM self.process.wait(timeout5) # 等待进程结束 print([停止] 服务器进程已终止) except subprocess.TimeoutExpired: print([警告] 进程未正常终止强制结束) self.process.kill() # 发送SIGKILL self.process.wait() finally: self.process None def main(): client BareMCPClient(weather.py) if not client.start(): sys.exit(1) try: # 1. 初始化握手 print(\n *50) print(阶段 1: 初始化握手) print(*50) init_response client.call_and_wait(initialize, { protocolVersion: 2025-03-26, capabilities: {}, clientInfo: {name: 解剖客户端, version: 1.0} }) if not init_response or error in init_response: print(初始化失败退出。) return print(初始化成功。服务器信息:) pprint(init_response.get(result, {}).get(serverInfo)) # 2. 发送 initialized 通知 print(\n *50) print(阶段 2: 发送就绪通知) print(*50) client.send(notifications/initialized, {}, is_notificationTrue) print(已通知服务器客户端准备就绪。) # 3. 列出工具 print(\n *50) print(阶段 3: 查询可用工具) print(*50) list_response client.call_and_wait(tools/list, {}) if not list_response or error in list_response: print(获取工具列表失败退出。) return tools list_response.get(result, {}).get(tools, []) print(f发现 {len(tools)} 个工具:) for t in tools: print(f - {t[name]}: {t.get(description, 无描述)[:80]}...) if not tools: print(没有可用工具退出。) return # 4. 调用一个工具 (例如 get_alerts) print(\n *50) print(阶段 4: 调用工具) print(*50) # 选择第一个工具或按名称选择 selected_tool tools[0] # 例如 get_alerts tool_name selected_tool[name] # 根据工具定义构造参数这里硬编码实际应由用户或逻辑决定 if tool_name get_alerts: arguments {state: CA} # 加州 elif tool_name get_forecast: arguments {latitude: 37.7749, longitude: -122.4194} # 旧金山 else: print(f未知工具 {tool_name}使用空参数测试。) arguments {} print(f调用工具: {tool_name}参数: {arguments}) call_response client.call_and_wait(tools/call, { name: tool_name, arguments: arguments }) if not call_response: print(工具调用无响应。) elif error in call_response: print(f工具调用返回错误: {call_response[error]}) else: result call_response.get(result, {}) print(f调用成功! isError: {result.get(isError)}) content result.get(content, []) for i, block in enumerate(content): print(f\n--- 内容块 {i1}/{len(content)} ---) if block.get(type) text: text block.get(text, ) # 显示前500字符作为预览 preview text[:500] if len(text) 500: preview ... print(preview) else: print(f类型: {block.get(type)}, 数据: {str(block)[:200]}...) print(\n *50) print(所有操作完成。) print(*50) except KeyboardInterrupt: print(\n[用户中断]) except Exception as e: print(f\n[未处理的异常] {e}) import traceback traceback.print_exc() finally: client.stop() if __name__ __main__: main()这个完整的客户端类封装了连接、通信、错误处理和资源管理的所有细节。它更加健壮包含了超时处理、错误流监控和更清晰的日志。6.2 关键错误处理与边界情况在实际操作中你会遇到各种问题。以下是一些常见场景及处理思路服务器启动失败表现Popen抛出异常如FileNotFoundError或进程立即退出。排查检查脚本路径、uv命令是否在PATH中、Python环境是否正确、依赖 (mcp) 是否已安装。我们的_drain_stderr方法会捕获并打印服务器的启动错误。初始化握手失败表现服务器对initialize请求返回一个包含error字段的JSON-RPC响应。常见原因protocolVersion不兼容、capabilities格式错误。处理解析error对象根据code和message调整客户端请求。工具调用参数错误表现tools/call响应中isError: true或直接返回JSON-RPC错误。常见原因参数缺失、参数类型不匹配、参数值不符合约束如州代码不是两个字母。处理严格遵循tools/list返回的inputSchema构建参数。实现参数验证逻辑。服务器无响应或超时表现readline()阻塞或长时间无返回。处理在生产环境中需要为readline()设置超时例如使用select或线程。我们的简单示例是阻塞的对于可靠系统需要改进。协议版本升级注意MCP协议仍在快速发展中。protocolVersion字段是关键。如果未来版本不兼容你需要根据服务器响应的版本或错误信息来适配客户端。避坑技巧在开发调试阶段不要将stderr重定向到DEVNULL。服务器的错误日志是诊断问题的第一手资料。许多奇怪的连接问题比如导入错误、权限问题都会打印到标准错误流。7. 协议深度解析与扩展思考通过亲手实现这个“裸”客户端我们已经穿透了抽象层直接触摸到了MCP协议的筋骨。基于此我们可以进行更深入的思考。7.1 MCP协议设计的精妙之处基于JSON-RPC 2.0这是一个非常明智的选择。JSON-RPC是一个简单、成熟、语言无关的RPC协议。它定义了请求、响应、通知和错误的标准格式MCP直接继承省去了重新设计消息格式的麻烦并获得了广泛的客户端库支持。传输层无关性协议规范只定义了消息的语义initialize,tools/call等而不规定传输方式stdio, WebSocket, HTTP等。这使得MCP能灵活适应从本地脚本插件到云端微服务的各种场景。我们使用的 stdio 只是其中一种实现。能力协商机制initialize握手过程中的capabilities字段为未来扩展留下了空间。服务器和客户端可以声明自己支持哪些可选功能如资源订阅、工具动态更新从而实现优雅的降级和功能发现。工具描述的枢纽作用tools/list返回的description和inputSchema是连接LLM世界与代码世界的桥梁。description让LLM理解工具功能inputSchema确保了调用的类型安全。这种设计分离了“意图理解”和“接口契约”。7.2 从“裸”客户端到生产级实现的差距我们的实验客户端是教学和理解的绝佳工具但距离生产环境还有距离连接管理与心跳生产客户端需要处理连接断开重连、心跳保活、服务器崩溃重启等场景。异步与并发一个客户端可能同时发起多个工具调用或者需要处理服务器主动推送的通知如tools/listChanged。这需要异步IO和更复杂的消息分发机制。安全性通过 stdio 通信相对安全本地进程但如果扩展到网络传输如WebSocket则需要考虑认证、授权和加密TLS。更完整的协议实现我们只实现了工具调用部分。完整的MCP客户端还需要处理资源resources/*和提示词prompts/*相关的消息。健壮的错误处理与重试网络波动、服务器临时过载、参数无效等都需要有系统的重试和降级策略。7.3 下一步探索方向理解了底层协议你可以实现其他语言的客户端用Go、Rust、Java甚至Shell脚本实现一个MCP客户端原理完全相同。调试复杂的MCP集成当使用Claude Desktop、Cursor等集成MCP的工具出现问题时你可以编写一个最小化的测试客户端直接与问题服务器对话精准定位是服务器错误、协议不匹配还是SDK问题。开发自定义MCP服务器现在你清楚地知道一个MCP服务器需要监听什么消息、返回什么格式。你可以为你的内部系统、数据库或独特API编写MCP适配器让LLM直接调用。分析协议流量在 stdio 通信中插入一个中间人代理记录和分析所有往来消息用于性能分析、审计或调试。8. 总结与最终建议这次深入线缆wire之下的探险应该已经驱散了MCP周围的大部分迷雾。它不是一个神秘莫测的“AI黑魔法”而是一个设计精良、基于成熟标准的RPC协议其核心目标是为LLM提供一个标准化、可扩展的“手和脚”。我的核心体会是理解任何协议或框架最有效的方式就是亲手实现一次最简版本。当你自己处理了JSON的序列化、管道的读写、状态机的维护那些抽象的概念瞬间就变得具体而清晰。你不再害怕文档中未提及的细节因为你知道数据包就在那里你可以看到它、修改它、创造它。对于想要在项目中应用MCP的开发者我的建议是从官方SDK开始对于大多数应用直接使用modelcontextprotocol/sdk(JavaScript/TypeScript) 或mcp(Python) 等官方SDK是最快、最稳的选择。但保留底层知识将本文的实践作为你的“逃生舱”知识。当SDK行为异常或你需要实现一些非常规功能时这份底层理解能帮你快速找到方向。关注工具描述的质量花时间打磨你的tools/list返回的description字段。清晰、准确、包含示例的描述能极大提升LLM调用工具的准确率。循序渐进先从简单的工具调用开始成功后再逐步尝试资源、提示词等高级特性。MCP的模块化设计允许你按需采纳功能。最后记住那句老话“一切皆文件描述符”。在Unix哲学里进程间通信不过是对文件描述符的读写。MCP over stdio 正是这一哲学的优雅体现——将复杂的AI能力集成简化成了对标准输入输出流的JSON格式化读写。当你掌握了这一点你就掌握了连接智能与代码世界最基础的钥匙。
从零实现MCP协议客户端：深入JSON-RPC与工具调用底层原理
发布时间：2026/5/30 23:22:02
1. 项目概述从理论到实战亲手解剖MCP协议的数据包最近Model Context Protocol (MCP) 在AI和LLM工具集成领域火得一塌糊涂。各种架构图、白皮书和概念解析文章满天飞仿佛人人都成了协议专家。但作为一个在软件工程一线摸爬滚打多年的老手我总觉得少了点什么。大家都在高谈阔论“双向通信”、“能力协商”、“上下文管理”这些宏大概念却没人愿意掀开引擎盖给你看看里面到底是怎么转的。这就像有人滔滔不绝地给你讲解汽车的空气动力学原理却从不让你看一眼发动机的内部构造。对于工程师来说这种隔靴搔痒的感觉实在难受。所以我决定干点“脏活累水”。这篇文章我们不谈哲学只看数据包。我们将彻底抛开那些精美的示意图和抽象的理论直接深入到比特和字节的层面亲手构建一个最原始的MCP客户端并通过标准输入输出与一个真实的MCP服务器对话。我们的目标只有一个亲眼看看在网络上流动的究竟是怎样的JSON。通过这个“外科手术”式的实践你不仅能透彻理解MCP的核心机制更能获得一种“我能从头构建它”的底气和自信。无论你是想深度集成AI能力到现有系统还是仅仅好奇协议底层如何工作这篇从零开始的实战指南都将为你提供最直接的洞察。2. 核心思路拆解为什么选择“裸”协议分析在开始动手之前我们先明确一下这次探索的独特视角和价值。市面上大多数MCP教程都基于官方SDK或高级框架这当然能快速上手但也像用自动挡学开车——你知道了怎么走但不知道离合器、变速箱和发动机是如何协同工作的。2.1 超越SDK理解协议的“第一性原理”使用官方SDK比如JavaScript或Python的MCP库无疑是最便捷的方式。它们封装了连接管理、消息序列化、错误处理等繁琐细节让你可以专注于业务逻辑。然而这种便利性也带来了一层抽象屏蔽了协议最本质的通信模式。当出现网络问题、版本不兼容或需要深度定制时这层抽象就可能变成黑箱让你无从下手。我们的方法截然不同我们将仅使用Python标准库的subprocess和json模块手动构造每一个JSON-RPC消息并通过管道发送。这相当于我们直接在与协议的“语法”对话跳过了所有“编译器”和“运行时”的中间层。这样做有几个不可替代的好处绝对透明每一个字节的发送和接收都完全在你的控制之下没有任何魔法。你可以清晰地看到请求是如何构建的响应是如何解析的。深度理解通过手动实现最基本的握手和工具调用流程你会对MCP的状态机、错误处理边界和设计哲学有肌肉记忆般的理解。调试利器当你使用高级SDK遇到诡异问题时拥有底层协议知识能让你快速定位问题是在应用层、SDK层还是协议层。你可以自己写一个小脚本模拟客户端行为直接测试服务器是否正常。不受限制你不受任何特定SDK版本、设计模式或依赖项的束缚。你可以用任何语言、在任何环境中实现MCP客户端只要它能启动子进程、读写JSON和管道。2.2 聚焦核心工具调用是MCP的“杀手锏”MCP协议规范涵盖了不少内容初始化、资源Resources、提示词Prompts、工具Tools等。但对于绝大多数实际应用场景——尤其是让LLM能够调用外部功能——工具调用Tool Calling是绝对的核心和起点。我们可以暂时忽略资源管理和提示词模板这些高级特性将注意力完全集中在工具调用的生命周期上。这样一来整个复杂的协议就被简化成了一个极其清晰的对话模型客户端“你好我支持MCP协议版本是X这是我的基本信息。”initialize服务器“你好我是XX服务器版本是Y我支持这些能力。”initialized响应客户端“好的我准备好了开始吧。”notifications/initialized客户端“你都能做什么”tools/list服务器“我能做A、B、C这几件事。”tools/list响应客户端“请帮我做A这是参数。”tools/call服务器“做完了这是结果。”tools/call响应看只需要理解这七种消息的交换你就掌握了MCP 80%以上的实用价值。我们的实验也将严格遵循这个精简流程。2.3 实验环境与目标服务器为了进行这次解剖我们需要一个“标本”——一个简单、稳定、开源的MCP服务器。官方提供的weather.py示例服务器是绝佳的选择。它提供了两个工具get_forecast获取天气预报和get_alerts获取天气警报。功能具体代码清晰且完全符合MCP规范。我们的实验目标非常明确手动启动这个天气服务器进程。不使用任何MCP库仅用标准库与其建立通信。完整走通上述7步对话并捕获每一个环节在“线上”即stdin/stdout管道传输的原始JSON数据。分析这些数据包的结构、字段含义和设计意图。注意选择stdin/stdout作为传输层是MCP的常见实践尤其对于本地集成的场景。它简单、高效且避免了网络端口、认证等复杂性。虽然有人诟病其健壮性但对于进程间通信IPC和快速原型来说它是无可替代的利器。我们的实验正是基于这种模式。3. 实战准备搭建你的数字解剖台理论铺垫完毕现在进入实战环节。请确保你有一个可用的Python环境3.7以上我们将从零开始搭建实验环境。3.1 获取并安装天气服务器首先我们需要获取MCP天气服务器的代码。官方示例存放在GitHub上。# 创建一个专门的工作目录 mkdir mcp-wire-dissection cd mcp-wire-dissection # 下载天气服务器示例代码 curl -O https://raw.githubusercontent.com/modelcontextprotocol/quickstart-resources/main/weather-server-python/weather.py # 查看文件确认下载成功 head -20 weather.py这个weather.py文件就是一个完整的MCP服务器实现。它使用了mcp这个Python SDK。因此我们需要安装其依赖。推荐使用现代Python包管理工具uv它比传统的pip更快、更可靠。# 安装 uv (如果尚未安装) # 在MacOS/Linux上 curl -LsSf https://astral.sh/uv/install.sh | sh # 初始化项目并安装依赖 uv init . uv add mcp实操心得使用uv而不是pip和venv能大幅提升依赖管理效率。uv不仅安装极快还能生成精确的锁文件确保环境一致性。对于这类一次性实验项目它能帮你省去很多配置虚拟环境的麻烦。3.2 理解服务器启动方式查看weather.py的末尾通常你会看到类似这样的代码if __name__ __main__: # 使用 mcp 库的 run 函数启动服务器 mcp.run(server)这意味着我们可以直接用python weather.py来启动它。但更常见的做法是使用uv run它能确保在正确的、隔离的依赖环境中运行脚本。这也是我们后续在客户端代码中启动子进程的方式。你可以先手动测试一下服务器是否能正常运行uv run weather.py如果一切正常服务器会启动并等待标准输入上的连接。此时它不会输出任何内容因为还没有客户端连接你可以按CtrlC终止它。这个“安静”的启动是正常的MCP服务器通常以这种“沉默的守护进程”模式运行。4. 核心环节一建立连接与初始化握手现在让我们开始编写我们的“外科医生”——一个纯手工的MCP客户端。我们将创建一个新的Python脚本bare_mcp_client.py。4.1 启动服务器子进程我们使用subprocess.Popen来启动服务器并捕获其标准输入、输出和错误流。这是实现进程间通信的基础。# bare_mcp_client.py import subprocess import json from pprint import pprint import time def start_mcp_server(server_script_path): 启动MCP服务器作为子进程。 返回一个Popen对象通过其stdin/stdout进行通信。 # 关键参数解析 # - stdinsubprocess.PIPE: 我们可以向进程的stdin写入数据我们的请求 # - stdoutsubprocess.PIPE: 我们可以从进程的stdout读取数据服务器的响应 # - stderrsubprocess.DEVNULL: 将服务器的错误输出重定向到空设备避免干扰我们的主控制台。 # 在实际调试时你可以改为 stderrsubprocess.PIPE 来捕获错误日志。 # - [uv, run, ...]: 使用uv在正确的环境中运行脚本。 process subprocess.Popen( [uv, run, server_script_path], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.DEVNULL, # 或 subprocess.PIPE 用于调试 textFalse, # 我们处理字节而不是字符串以获得更精确的控制 bufsize1, # 行缓冲确保每条消息能及时发送 ) print(f[INFO] 服务器进程已启动PID: {process.pid}) return process注意事项将stderr重定向到DEVNULL在实验阶段是干净的但生产环境或调试时一定要捕获并处理stderr。服务器可能会将重要的启动错误或运行时日志输出到标准错误流。4.2 定义消息收发函数MCP over stdio 的约定是newline-delimited JSON (NDJSON)。即每条完整的JSON消息独占一行以换行符\n分隔。这是许多流式JSON协议如JSON-RPC over HTTP/1.1的某些实现的常见模式。def send_mcp_message(process, message_dict): 向MCP服务器进程发送一条JSON-RPC消息。 消息被序列化为JSON字符串末尾添加换行符然后编码为字节流写入stdin。 # 1. 将字典转换为JSON字符串 json_str json.dumps(message_dict) # 2. 添加NDJSON要求的换行符 json_line json_str \n # 3. 编码为字节并写入进程的stdin process.stdin.write(json_line.encode(utf-8)) # 4. 刷新缓冲区确保数据立即发送 process.stdin.flush() print(f[SENT] {json_str[:200]}...) # 打印前200个字符便于观察 def read_mcp_message(process): 从MCP服务器进程的stdout读取一条JSON-RPC消息。 读取一行解码为字符串然后解析为Python字典。 这是一个阻塞调用会等待服务器输出新的一行。 # 读取一行字节数据 line_bytes process.stdout.readline() if not line_bytes: # 如果读到空字节可能意味着管道已关闭或进程结束 return None # 解码为UTF-8字符串并去除末尾的换行符 line_str line_bytes.decode(utf-8).rstrip(\n) if line_str: print(f[RECV] 原始行长度: {len(line_str)} 字符) # 解析JSON message json.loads(line_str) return message return None4.3 执行初始化握手现在让我们发起第一次对话。根据MCP规范连接建立后的第一条消息必须是initialize请求。def perform_handshake(process, client_nameBareMCPClient, client_version0.1.0): 执行MCP协议初始化握手。 返回服务器返回的初始化结果。 # 步骤1: 客户端发送 initialize 请求 init_request { jsonrpc: 2.0, id: 1, # 请求ID用于匹配响应 method: initialize, params: { protocolVersion: 2025-03-26, # 必须与服务器兼容的协议版本 capabilities: {}, # 客户端声明自己支持的能力这里为空 clientInfo: { name: client_name, version: client_version } } } print(\n 步骤 1: 发送 initialize 请求 ) send_mcp_message(process, init_request) # 步骤2: 读取服务器的 initialize 响应 print(\n 等待服务器 initialize 响应 ) init_response read_mcp_message(process) if not init_response: raise ConnectionError(服务器未响应 initialize 请求) print(服务器初始化响应:) pprint(init_response, depth3) # 限制打印深度避免信息过载 # 步骤3: 客户端发送 initialized 通知 # 这是一个“通知”notification没有id字段服务器不回复。 initialized_notification { jsonrpc: 2.0, method: notifications/initialized, params: {} # 参数为空对象 } print(\n 步骤 2: 发送 initialized 通知 ) send_mcp_message(process, initialized_notification) # 发送通知后没有响应需要读取 return init_response让我们逐行分析这个握手过程initialize请求jsonrpc: 2.0: 声明我们使用JSON-RPC 2.0协议。这是MCP的底层传输协议。id: 1: 每个请求需要一个唯一ID服务器会在对应的响应中带回相同的ID。这对于异步通信至关重要。method: initialize: 指定要调用的方法。params: 包含握手参数。protocolVersion: 指定我们希望使用的MCP协议版本。必须与服务器支持的版本匹配否则握手会失败。capabilities: 客户端向服务器宣告自己支持哪些MCP扩展功能如资源订阅、提示词变更通知等。我们这里留空表示只支持基础功能。clientInfo: 客户端的身份信息方便服务器日志记录和识别。服务器initialize响应服务器会返回一个同样包含id: 1 的JSON-RPC响应。result字段中包含服务器的信息如serverInfo名称、版本和capabilities服务器支持的能力。这是我们第一次了解服务器特性的机会。notifications/initialized通知这是一个通知而不是请求。JSON-RPC中通知的id字段为null或直接省略。它用于告知服务器“客户端已准备就绪可以开始正式工作”。服务器收到后不会发送任何响应。这完成了握手的最后一步。运行这部分代码你会看到类似以下的输出具体内容因服务器版本而异[INFO] 服务器进程已启动PID: 12345 步骤 1: 发送 initialize 请求 [SENT] {jsonrpc: 2.0, id: 1, method: initialize, params: {protocolVersion: 2025-03-26, capabilities: {}, clientInfo: {name: BareMCPClient, version: 0.1.0}}}... 等待服务器 initialize 响应 [RECV] 原始行长度: 266 字符 服务器初始化响应: {id: 1, jsonrpc: 2.0, result: {capabilities: {experimental: {}, prompts: {listChanged: False}, resources: {listChanged: False, subscribe: False}, tools: {listChanged: False}}, protocolVersion: 2025-03-26, serverInfo: {name: weather, version: 1.9.4}}} 步骤 2: 发送 initialized 通知 [SENT] {jsonrpc: 2.0, method: notifications/initialized, params: {}}...恭喜你已经成功完成了MCP连接握手。你看到了服务器返回的原始JSON知道了它的名字、版本以及它声明不支持动态工具列表变更 (listChanged: False)。这就是“线上”流动的第一个关键数据包。5. 核心环节二发现与调用工具握手成功后客户端和服务器就进入了工作状态。接下来客户端需要知道服务器能做什么。5.1 查询可用工具列表我们发送tools/list请求来获取服务器暴露的所有工具。def list_available_tools(process): 向服务器请求可用的工具列表。 list_tools_request { jsonrpc: 2.0, id: 2, # 新的请求ID method: tools/list, params: {} # 此请求通常不需要额外参数 } print(\n 步骤 3: 查询工具列表 (tools/list) ) send_mcp_message(process, list_tools_request) print(\n 等待工具列表响应 ) list_response read_mcp_message(process) if not list_response: raise ConnectionError(服务器未响应 tools/list 请求) # 检查是否有错误 if error in list_response: print(f查询工具列表失败: {list_response[error]}) return None tools list_response.get(result, {}).get(tools, []) print(f发现 {len(tools)} 个工具:) for tool in tools: print(f - 名称: {tool[name]}) print(f 描述: {tool.get(description, N/A)[:100]}...) # 截断长描述 # 打印输入参数模式 schema tool.get(inputSchema, {}) req schema.get(required, []) props schema.get(properties, {}) if req or props: print(f 输入参数: {list(props.keys())} (必填: {req})) print() return tools运行后你会看到服务器返回的详细工具列表。以天气服务器为例输出可能如下 步骤 3: 查询工具列表 (tools/list) [SENT] {jsonrpc: 2.0, id: 2, method: tools/list, params: {}}... 等待工具列表响应 [RECV] 原始行长度: 732 字符 发现 2 个工具: - 名称: get_alerts 描述: Get weather alerts for a US state. Args: state: Two-letter US state code (e.g. CA, NY)... 输入参数: [state] (必填: [state]) - 名称: get_forecast 描述: Get weather forecast for a location. Args: latitude: Latitude of the location longitude:... 输入参数: [latitude, longitude] (必填: [latitude, longitude])关键字段解析name: 工具的标识符在调用时必须精确匹配。description: 给LLM看的自然语言描述。这是工具能否被正确调用的关键。LLM根据这段描述来决定是否以及如何使用这个工具。inputSchema: 一个遵循JSON Schema的对象严格定义了调用此工具所需的参数名称、类型和是否必需。这是实现强类型接口和自动验证的基础。实操心得description字段的撰写是一门艺术。它需要足够清晰让LLM能准确理解工具的用途又不能过于冗长以免占用过多上下文窗口。一个好的描述应包含1) 工具的核心功能2) 每个参数的含义和示例3) 可能返回的内容。这是你作为服务器开发者与LLM“沟通”的主要渠道。5.2 构造并发送工具调用请求现在我们知道了服务器有get_alerts工具它需要一个state参数美国州代码。让我们调用它来获取德克萨斯州TX的天气警报。def call_mcp_tool(process, tool_name, arguments, request_id3): 调用指定的MCP工具。 call_request { jsonrpc: 2.0, id: request_id, method: tools/call, params: { name: tool_name, # 必须与 tools/list 返回的 name 完全一致 arguments: arguments # 参数必须符合 inputSchema 的定义 } } print(f\n 步骤 4: 调用工具 {tool_name} ) print(f 参数: {arguments}) send_mcp_message(process, call_request) print(\n 等待工具调用响应 ) call_response read_mcp_message(process) if not call_response: raise ConnectionError(f服务器未响应 tools/call 请求 (id: {request_id})) # 检查响应中是否包含错误 if error in call_response: error call_response[error] print(f[ERROR] 工具调用失败!) print(f 错误码: {error.get(code)}) print(f 消息: {error.get(message)}) if data in error: print(f 详情: {error.get(data)}) return None # 解析成功的结果 result call_response.get(result, {}) is_error result.get(isError, False) content result.get(content, []) print(f调用结果: isError{is_error}) print(f返回了 {len(content)} 个内容块) for i, block in enumerate(content): block_type block.get(type, unknown) print(f\n--- 内容块 {i1} (类型: {block_type}) ---) if block_type text: text block.get(text, ) # 对于长文本只预览开头和结尾 if len(text) 500: print(text[:250] ...\n...\n text[-250:]) else: print(text) elif block_type image: # 如果是图片可能包含数据URI或引用 print(f图片数据 (已省略): {block.get(data, N/A)[:100]}...) # 可以处理其他类型如 audio, video, resource 等 return result现在在主函数中调用它# 在主流程中握手和获取工具列表之后... tools list_available_tools(process) if tools: # 假设我们调用 get_alerts tool_to_call get_alerts # 确保工具存在 if any(t[name] tool_to_call for t in tools): # 准备参数 tool_args {state: TX} # 获取德克萨斯州的警报 result call_mcp_tool(process, tool_to_call, tool_args, request_id3) if result: print(\n✅ 工具调用成功) else: print(\n❌ 工具调用失败或返回错误。) else: print(f\n工具 {tool_to_call} 不在可用列表中。)运行这段代码你将见证奇迹或者说一次标准的API调用。客户端发送一个紧凑的请求服务器返回可能非常庞大的天气警报数据。 步骤 4: 调用工具 get_alerts 参数: {state: TX} [SENT] {jsonrpc: 2.0, id: 3, method: tools/call, params: {name: get_alerts, arguments: {state: TX}}}... 等待工具调用响应 [RECV] 原始行长度: 51305 字符 调用结果: isErrorFalse 返回了 1 个内容块 --- 内容块 1 (类型: text) --- Event: Flood Advisory Area: Hidalgo, TX; Willacy, TX Severity: Minor Description: * WHAT...Flooding caused by excessive rainfall continues. ... Instructions: Turn around, dont drown when encountering flooded roads... The next statement will be issued Tuesday morning at 830 AM CDT. ✅ 工具调用成功响应结构深度解析 你收到的响应是一个标准的JSON-RPC成功响应。核心在于result字段isError:false表示调用成功。如果是true则content中可能包含错误信息。content: 一个数组包含一个或多个“内容块”。每个块有type和具体数据。type: text: 这是最常见的类型数据在text字段中是一个字符串。注意这个字符串内部可以包含任何结构化的文本比如我们看到的格式化天气警报。LLM擅长从这种半结构化的自然语言文本中提取信息。其他类型MCP规范还支持image、audio、video等类型用于多模态输出。resource类型则用于引用服务器管理的资源如文件。关键洞察MCP工具调用的响应设计非常通用。content数组允许服务器返回多种类型、多个部分的结果。text类型的灵活性极高它可以是纯文本、JSON字符串、CSV、HTML甚至是自定义的标记语言。这种设计将“数据呈现”的复杂性留给了服务器和客户端或LLM协议本身只负责传输。这也是为什么你很少在工具定义中看到outputSchema——输出格式是隐含在content的类型和结构中的。6. 核心环节三完整流程整合与错误处理我们已经拆解了每一个步骤。现在让我们把它们组合成一个完整、健壮的客户端脚本并加入必要的错误处理和资源清理。6.1 完整的客户端脚本# bare_mcp_client.py (完整版) import subprocess import json import sys from pprint import pprint class BareMCPClient: def __init__(self, server_script_path): self.server_path server_script_path self.process None self.next_request_id 1 # 用于生成唯一的请求ID def start(self): 启动MCP服务器进程 print(f[启动] 正在启动MCP服务器: {self.server_path}) try: self.process subprocess.Popen( [uv, run, self.server_path], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, # 改为PIPE以捕获错误 textFalse, bufsize1, ) # 启动后先读一下stderr看是否有立即报错 self._drain_stderr() print(f[启动] 服务器进程已启动 (PID: {self.process.pid})) return True except FileNotFoundError as e: print(f[错误] 无法找到服务器脚本或uv命令: {e}) print(f 请确保: 1) 脚本路径正确; 2) uv已安装; 3) 依赖已安装 (uv add mcp)) return False except Exception as e: print(f[错误] 启动服务器时发生未知错误: {e}) return False def _drain_stderr(self): 非阻塞地读取并打印服务器的错误输出 import select if self.process and self.process.stderr: # 检查stderr是否有内容可读非阻塞 rlist, _, _ select.select([self.process.stderr], [], [], 0.1) if rlist: err_output self.process.stderr.read().decode(utf-8, errorsignore) if err_output: print(f[服务器STDERR] {err_output.strip()}) def send(self, method, paramsNone, is_notificationFalse): 发送JSON-RPC请求或通知 if not self.process or self.process.poll() is not None: raise ConnectionError(服务器进程未运行或已终止) request { jsonrpc: 2.0, method: method, } if not is_notification: request[id] self.next_request_id self.next_request_id 1 if params: request[params] params json_line json.dumps(request) \n try: self.process.stdin.write(json_line.encode(utf-8)) self.process.stdin.flush() print(f[发送] {method} (ID: {request.get(id, N/A-通知)})) # 打印精简的请求内容 if params: preview str(params)[:100] if len(str(params)) 100: preview ... print(f 参数: {preview}) return request.get(id) # 返回请求ID用于匹配响应 except BrokenPipeError: print(f[错误] 向服务器写入失败管道已断开) self.stop() raise ConnectionError(与服务器的连接已断开) def receive(self): 从服务器接收一条JSON-RPC消息 if not self.process: return None # 先检查一下stderr看是否有错误输出 self._drain_stderr() try: line_bytes self.process.stdout.readline() if not line_bytes: print([接收] 读到空字节服务器可能已关闭输出流) return None line_str line_bytes.decode(utf-8, errorsreplace).rstrip(\n) if not line_str: return None print(f[接收] 原始数据长度: {len(line_str)} 字符) return json.loads(line_str) except json.JSONDecodeError as e: print(f[错误] 解析服务器响应JSON失败: {e}) print(f 原始行: {line_str[:200]}...) return None except Exception as e: print(f[错误] 读取服务器响应时发生未知错误: {e}) return None def call_and_wait(self, method, paramsNone): 发送请求并等待响应返回响应字典 req_id self.send(method, params, is_notificationFalse) if req_id is None: return None # 循环读取直到找到对应ID的响应或遇到错误/通知 while True: response self.receive() if response is None: print(f[错误] 在等待响应 {req_id} 时连接断开) return None # 如果是通知无id打印并继续等 if id not in response: print(f[接收] 收到通知: {response.get(method, unknown)}) continue # 找到对应ID的响应 if response.get(id) req_id: return response # ID不匹配可能是之前的响应未处理或服务器乱序MCP通常不会 print(f[警告] 收到不匹配的响应ID: 期望 {req_id}, 收到 {response.get(id)}) def stop(self): 停止服务器进程 if self.process: print([停止] 正在终止服务器进程...) try: self.process.terminate() # 发送SIGTERM self.process.wait(timeout5) # 等待进程结束 print([停止] 服务器进程已终止) except subprocess.TimeoutExpired: print([警告] 进程未正常终止强制结束) self.process.kill() # 发送SIGKILL self.process.wait() finally: self.process None def main(): client BareMCPClient(weather.py) if not client.start(): sys.exit(1) try: # 1. 初始化握手 print(\n *50) print(阶段 1: 初始化握手) print(*50) init_response client.call_and_wait(initialize, { protocolVersion: 2025-03-26, capabilities: {}, clientInfo: {name: 解剖客户端, version: 1.0} }) if not init_response or error in init_response: print(初始化失败退出。) return print(初始化成功。服务器信息:) pprint(init_response.get(result, {}).get(serverInfo)) # 2. 发送 initialized 通知 print(\n *50) print(阶段 2: 发送就绪通知) print(*50) client.send(notifications/initialized, {}, is_notificationTrue) print(已通知服务器客户端准备就绪。) # 3. 列出工具 print(\n *50) print(阶段 3: 查询可用工具) print(*50) list_response client.call_and_wait(tools/list, {}) if not list_response or error in list_response: print(获取工具列表失败退出。) return tools list_response.get(result, {}).get(tools, []) print(f发现 {len(tools)} 个工具:) for t in tools: print(f - {t[name]}: {t.get(description, 无描述)[:80]}...) if not tools: print(没有可用工具退出。) return # 4. 调用一个工具 (例如 get_alerts) print(\n *50) print(阶段 4: 调用工具) print(*50) # 选择第一个工具或按名称选择 selected_tool tools[0] # 例如 get_alerts tool_name selected_tool[name] # 根据工具定义构造参数这里硬编码实际应由用户或逻辑决定 if tool_name get_alerts: arguments {state: CA} # 加州 elif tool_name get_forecast: arguments {latitude: 37.7749, longitude: -122.4194} # 旧金山 else: print(f未知工具 {tool_name}使用空参数测试。) arguments {} print(f调用工具: {tool_name}参数: {arguments}) call_response client.call_and_wait(tools/call, { name: tool_name, arguments: arguments }) if not call_response: print(工具调用无响应。) elif error in call_response: print(f工具调用返回错误: {call_response[error]}) else: result call_response.get(result, {}) print(f调用成功! isError: {result.get(isError)}) content result.get(content, []) for i, block in enumerate(content): print(f\n--- 内容块 {i1}/{len(content)} ---) if block.get(type) text: text block.get(text, ) # 显示前500字符作为预览 preview text[:500] if len(text) 500: preview ... print(preview) else: print(f类型: {block.get(type)}, 数据: {str(block)[:200]}...) print(\n *50) print(所有操作完成。) print(*50) except KeyboardInterrupt: print(\n[用户中断]) except Exception as e: print(f\n[未处理的异常] {e}) import traceback traceback.print_exc() finally: client.stop() if __name__ __main__: main()这个完整的客户端类封装了连接、通信、错误处理和资源管理的所有细节。它更加健壮包含了超时处理、错误流监控和更清晰的日志。6.2 关键错误处理与边界情况在实际操作中你会遇到各种问题。以下是一些常见场景及处理思路服务器启动失败表现Popen抛出异常如FileNotFoundError或进程立即退出。排查检查脚本路径、uv命令是否在PATH中、Python环境是否正确、依赖 (mcp) 是否已安装。我们的_drain_stderr方法会捕获并打印服务器的启动错误。初始化握手失败表现服务器对initialize请求返回一个包含error字段的JSON-RPC响应。常见原因protocolVersion不兼容、capabilities格式错误。处理解析error对象根据code和message调整客户端请求。工具调用参数错误表现tools/call响应中isError: true或直接返回JSON-RPC错误。常见原因参数缺失、参数类型不匹配、参数值不符合约束如州代码不是两个字母。处理严格遵循tools/list返回的inputSchema构建参数。实现参数验证逻辑。服务器无响应或超时表现readline()阻塞或长时间无返回。处理在生产环境中需要为readline()设置超时例如使用select或线程。我们的简单示例是阻塞的对于可靠系统需要改进。协议版本升级注意MCP协议仍在快速发展中。protocolVersion字段是关键。如果未来版本不兼容你需要根据服务器响应的版本或错误信息来适配客户端。避坑技巧在开发调试阶段不要将stderr重定向到DEVNULL。服务器的错误日志是诊断问题的第一手资料。许多奇怪的连接问题比如导入错误、权限问题都会打印到标准错误流。7. 协议深度解析与扩展思考通过亲手实现这个“裸”客户端我们已经穿透了抽象层直接触摸到了MCP协议的筋骨。基于此我们可以进行更深入的思考。7.1 MCP协议设计的精妙之处基于JSON-RPC 2.0这是一个非常明智的选择。JSON-RPC是一个简单、成熟、语言无关的RPC协议。它定义了请求、响应、通知和错误的标准格式MCP直接继承省去了重新设计消息格式的麻烦并获得了广泛的客户端库支持。传输层无关性协议规范只定义了消息的语义initialize,tools/call等而不规定传输方式stdio, WebSocket, HTTP等。这使得MCP能灵活适应从本地脚本插件到云端微服务的各种场景。我们使用的 stdio 只是其中一种实现。能力协商机制initialize握手过程中的capabilities字段为未来扩展留下了空间。服务器和客户端可以声明自己支持哪些可选功能如资源订阅、工具动态更新从而实现优雅的降级和功能发现。工具描述的枢纽作用tools/list返回的description和inputSchema是连接LLM世界与代码世界的桥梁。description让LLM理解工具功能inputSchema确保了调用的类型安全。这种设计分离了“意图理解”和“接口契约”。7.2 从“裸”客户端到生产级实现的差距我们的实验客户端是教学和理解的绝佳工具但距离生产环境还有距离连接管理与心跳生产客户端需要处理连接断开重连、心跳保活、服务器崩溃重启等场景。异步与并发一个客户端可能同时发起多个工具调用或者需要处理服务器主动推送的通知如tools/listChanged。这需要异步IO和更复杂的消息分发机制。安全性通过 stdio 通信相对安全本地进程但如果扩展到网络传输如WebSocket则需要考虑认证、授权和加密TLS。更完整的协议实现我们只实现了工具调用部分。完整的MCP客户端还需要处理资源resources/*和提示词prompts/*相关的消息。健壮的错误处理与重试网络波动、服务器临时过载、参数无效等都需要有系统的重试和降级策略。7.3 下一步探索方向理解了底层协议你可以实现其他语言的客户端用Go、Rust、Java甚至Shell脚本实现一个MCP客户端原理完全相同。调试复杂的MCP集成当使用Claude Desktop、Cursor等集成MCP的工具出现问题时你可以编写一个最小化的测试客户端直接与问题服务器对话精准定位是服务器错误、协议不匹配还是SDK问题。开发自定义MCP服务器现在你清楚地知道一个MCP服务器需要监听什么消息、返回什么格式。你可以为你的内部系统、数据库或独特API编写MCP适配器让LLM直接调用。分析协议流量在 stdio 通信中插入一个中间人代理记录和分析所有往来消息用于性能分析、审计或调试。8. 总结与最终建议这次深入线缆wire之下的探险应该已经驱散了MCP周围的大部分迷雾。它不是一个神秘莫测的“AI黑魔法”而是一个设计精良、基于成熟标准的RPC协议其核心目标是为LLM提供一个标准化、可扩展的“手和脚”。我的核心体会是理解任何协议或框架最有效的方式就是亲手实现一次最简版本。当你自己处理了JSON的序列化、管道的读写、状态机的维护那些抽象的概念瞬间就变得具体而清晰。你不再害怕文档中未提及的细节因为你知道数据包就在那里你可以看到它、修改它、创造它。对于想要在项目中应用MCP的开发者我的建议是从官方SDK开始对于大多数应用直接使用modelcontextprotocol/sdk(JavaScript/TypeScript) 或mcp(Python) 等官方SDK是最快、最稳的选择。但保留底层知识将本文的实践作为你的“逃生舱”知识。当SDK行为异常或你需要实现一些非常规功能时这份底层理解能帮你快速找到方向。关注工具描述的质量花时间打磨你的tools/list返回的description字段。清晰、准确、包含示例的描述能极大提升LLM调用工具的准确率。循序渐进先从简单的工具调用开始成功后再逐步尝试资源、提示词等高级特性。MCP的模块化设计允许你按需采纳功能。最后记住那句老话“一切皆文件描述符”。在Unix哲学里进程间通信不过是对文件描述符的读写。MCP over stdio 正是这一哲学的优雅体现——将复杂的AI能力集成简化成了对标准输入输出流的JSON格式化读写。当你掌握了这一点你就掌握了连接智能与代码世界最基础的钥匙。
