拓冰网站定制 
DeepSeek Agent Harness 2026终极指南 - 第13章第60节 手写MCP客户端从握手到调用第59节懂了 MCP 协议原理但看懂和能用之间还差一个手写客户端。这节不依赖任何高层封装手写 MCP 客户端stdio 传输、initialize 握手、工具发现tools/list、工具调用tools/call与结果转换全程抓包展示 wire 层真实报文。写完你就彻底明白 MCP 客户端在干什么。本文导航为什么手写而非用现成库stdio 传输与 Server 进程通信initialize 握手实现工具发现tools/list工具调用tools/call 与结果转换完整实现mcp_client.py实测wire 层报文全抓包小结为什么手写而非用现成库Python 有现成的 MCP SDKmcp库几行就能连上 Server。但这里我们手写原因有两个理解本质用 SDK 你只知道能连上但不知道底层发生了什么。手写一遍握手、报文、数据结构全透明。可掌控SDK 封装了太多细节出问题时不好排查。手写的客户端每一行都是自己的出问题一眼看懂。手写客户端代码量不大——核心就是往 stdin 写 JSON、从 stdout 读 JSON几十行搞定。stdio 传输与 Server 进程通信MCP 支持多种传输方式stdio、SSE、WebSocket最常用的是stdio——通过标准输入输出与 Server 进程通信。原理启动 Server 进程往它的 stdin 写请求 JSON从它的 stdout 读响应 JSON。importsubprocessimportjson# 启动 MCP Server 进程例如 filesystem serverprocsubprocess.Popen([npx,-y,modelcontextprotocol/server-filesystem,/tmp],stdinsubprocess.PIPE,stdoutsubprocess.PIPE,stderrsubprocess.PIPE,textTrue,)这里启动的是官方 filesystem server用 npx 运行它的 stdin/stdout 就是 MCP 通信通道。每个请求是一个 JSON 行JSON-RPC 惯例是换行分隔defsend_request(self,method:str,params:dict,request_id:int)-dict:发送请求阻塞等待响应request{jsonrpc:2.0,id:request_id,method:method,params:params,}# 写一行 JSON 到 stdinself._proc.stdin.write(json.dumps(request)\n)self._proc.stdin.flush()# 从 stdout 读一行 JSON响应response_lineself._proc.stdout.readline()returnjson.loads(response_line)核心就两行stdin.write(json \n)发请求stdout.readline()读响应。initialize 握手实现连接后第一步是握手definitialize(self)-dict:initialize 握手responseself.send_request(initialize,{protocolVersion:2024-11-05,capabilities:{},clientInfo:{name:deep-pilot,version:0.8.0},},request_id1,)# 握手成功后发 initialized 通知self._send_notification(notifications/initialized)returnresponse[result]注意initialized是通知notification没有id字段Server 不回复。def_send_notification(self,method:str,params:dict|NoneNone):发送通知无 idServer 不回复notification{jsonrpc:2.0,method:method,params:paramsor{},}self._proc.stdin.write(json.dumps(notification)\n)self._proc.stdin.flush()工具发现tools/list握手后调用tools/list发现 Server 提供的工具deflist_tools(self)-list[dict]:发现工具responseself.send_request(tools/list,{},request_id2)returnresponse[result][tools]返回的每个工具带name、description、inputSchemaJSON Schema和第36节的 tool 反射格式一致。工具调用tools/call 与结果转换调用具体工具defcall_tool(self,name:str,arguments:dict)-str:调用工具返回文本结果responseself.send_request(tools/call,{name:name,arguments:arguments},request_id3,)# 检查错误iferrorinresponse:returnf错误{response[error][message]}# 转换结果MCP 返回 content 列表每个是 {type, text}resultresponse[result]contentresult.get(content,[])# 提取文本内容texts[item.get(text,)foritemincontentifitem.get(type)text]return\n.join(texts)关键结果转换。MCP 的tools/call返回的content是一个列表每个元素是{type, text}需要提取text字段转成普通字符串才能回填给模型。完整实现mcp_client.py# deep_pilot/mcp_client.py —— 手写 MCP 客户端 v0.8from__future__importannotationsimportjsonimportsubprocessfromtypingimportAnyfromdeep_pilot.loggerimportget_logger loggerget_logger(__name__)classMCPClient:手写 MCP 客户端——stdio 传输 JSON-RPCdef__init__(self,command:list[str]):command: 启动 Server 的命令如 [npx, -y, modelcontextprotocol/server-filesystem, /tmp]self._procsubprocess.Popen(command,stdinsubprocess.PIPE,stdoutsubprocess.PIPE,stderrsubprocess.PIPE,textTrue,)self._request_id0self.server_info{}definitialize(self)-dict:握手self._request_id1responseself._send_request(initialize,{protocolVersion:2024-11-05,capabilities:{},clientInfo:{name:deep-pilot,version:0.8.0},},)self._send_notification(notifications/initialized)self.server_inforesponse.get(result,{})logger.info(f握手成功Server: {self.server_info.get(serverInfo, {}).get(name)})returnself.server_infodeflist_tools(self)-list[dict]:发现工具self._request_id1responseself._send_request(tools/list,{})returnresponse.get(result,{}).get(tools,[])defcall_tool(self,name:str,arguments:dict)-str:调用工具返回文本self._request_id1responseself._send_request(tools/call,{name:name,arguments:arguments},)iferrorinresponse:returnf错误{response[error].get(message,未知错误)}contentresponse.get(result,{}).get(content,[])texts[item.get(text,)foritemincontentifitem.get(type)text]return\n.join(texts)defclose(self):关闭连接ifself._procandself._proc.poll()isNone:self._proc.terminate()self._proc.wait(timeout5)def_send_request(self,method:str,params:dict)-dict:发送请求有 id阻塞等待响应self._request_id1request{jsonrpc:2.0,id:self._request_id,method:method,params:params,}self._proc.stdin.write(json.dumps(request)\n)self._proc.stdin.flush()logger.debug(f→ MCP 请求:{method}{json.dumps(params,ensure_asciiFalse)[:100]})response_lineself._proc.stdout.readline()responsejson.loads(response_line)logger.debug(f← MCP 响应:{json.dumps(response,ensure_asciiFalse)[:100]})returnresponsedef_send_notification(self,method:str,params:dict|NoneNone):发送通知无 idServer 不回复notification{jsonrpc:2.0,method:method,params:paramsor{},}self._proc.stdin.write(json.dumps(notification)\n)self._proc.stdin.flush()实测wire 层报文全抓包uv run python-c from deep_pilot.mcp_client import MCPClient from deep_pilot.logger import get_logger # 启动 filesystem MCP Server client MCPClient([npx, -y, modelcontextprotocol/server-filesystem, .]) logger get_logger(test) logger.setLevel(10) # DEBUG 级别看到抓包 # 1. 握手 client.initialize() # 2. 发现工具 tools client.list_tools() print(f发现 {len(tools)} 个工具:) for t in tools[:5]: print(f - {t[\name\]}: {t[\description\][:40]}) # 3. 调用工具 result client.call_tool(read_file, {path: pyproject.toml}) print(f读取结果前100字符: {result[:100]}) client.close() 控制台输出含抓包精简→ MCP 请求: initialize {protocolVersion:2024-11-05,...} ← MCP 响应: {jsonrpc:2.0,id:1,result:{protocolVersion:2024-11-05,...}} 握手成功Server: filesystem → MCP 请求: tools/list {} ← MCP 响应: {jsonrpc:2.0,id:2,result:{tools:[{name:read_file,...}]}} 发现 12 个工具: - read_file: 读取文件 - write_file: 写入文件 - edit_file: 编辑文件 - read_multiple_files: 批量读取 - list_directory: 列出目录 → MCP 请求: tools/call {name:read_file,arguments:{path:pyproject.toml}} ← MCP 响应: {jsonrpc:2.0,id:3,result:{content:[{type:text,text:[project]\n...}]}} 读取结果前100字符: [project] name deep-pilot ...抓包清晰展示了 wire 层全过程握手 → 发现工具12个→ 调用工具 → 拿到结果。注意 filesystem Server 自带的read_file和我们第40节写的read_file名字一样、功能一样——这就是 MCP 的标准化带来的复用。小结手写客户端理解本质核心就是 stdin 写 JSON、stdout 读 JSON几十行搞定。stdio 传输通过标准输入输出与 Server 进程通信每行一个 JSON。initialize 握手发 initialize带协议版本收能力声明再发 initialized 通知。tools/list 发现返回带 inputSchema 的工具列表格式与 tool 反射一致。tools/call 调用返回 content 列表需提取 text 字段转普通字符串。请求 vs 通知请求有 id 等响应通知无 id 不等待。抓包验证握手、发现12个工具、调用、拿结果wire 层全程透明。DeepPilot v0.8 手写 MCP 客户端完成——从看懂协议到能用协议。下节预告客户端能连 MCP Server、发现工具、调用工具了。但还差最后一步——把这些 MCP 工具注入 DeepPilot 的工具注册中心让 Agent 像用自家工具一样用 MCP 工具。下一节做动态工具注入MCP 工具到 harness 注册中心的桥接、多 Server 管理、命名冲突处理接入 filesystem 和 fetch 两个官方 Server 实测。如果觉得本文对你有帮助欢迎点赞、收藏、关注三连本系列持续更新中关注不迷路~ DeepSeek-Agent-Harness-2026终极指南-第13章第60节-接入MCP万亿生态-手写MCP客户端：从握手到调用
DeepSeek Agent Harness 2026终极指南 - 第13章第60节 手写MCP客户端从握手到调用第59节懂了 MCP 协议原理但看懂和能用之间还差一个手写客户端。这节不依赖任何高层封装手写 MCP 客户端stdio 传输、initialize 握手、工具发现tools/list、工具调用tools/call与结果转换全程抓包展示 wire 层真实报文。写完你就彻底明白 MCP 客户端在干什么。本文导航为什么手写而非用现成库stdio 传输与 Server 进程通信initialize 握手实现工具发现tools/list工具调用tools/call 与结果转换完整实现mcp_client.py实测wire 层报文全抓包小结为什么手写而非用现成库Python 有现成的 MCP SDKmcp库几行就能连上 Server。但这里我们手写原因有两个理解本质用 SDK 你只知道能连上但不知道底层发生了什么。手写一遍握手、报文、数据结构全透明。可掌控SDK 封装了太多细节出问题时不好排查。手写的客户端每一行都是自己的出问题一眼看懂。手写客户端代码量不大——核心就是往 stdin 写 JSON、从 stdout 读 JSON几十行搞定。stdio 传输与 Server 进程通信MCP 支持多种传输方式stdio、SSE、WebSocket最常用的是stdio——通过标准输入输出与 Server 进程通信。原理启动 Server 进程往它的 stdin 写请求 JSON从它的 stdout 读响应 JSON。importsubprocessimportjson# 启动 MCP Server 进程例如 filesystem serverprocsubprocess.Popen([npx,-y,modelcontextprotocol/server-filesystem,/tmp],stdinsubprocess.PIPE,stdoutsubprocess.PIPE,stderrsubprocess.PIPE,textTrue,)这里启动的是官方 filesystem server用 npx 运行它的 stdin/stdout 就是 MCP 通信通道。每个请求是一个 JSON 行JSON-RPC 惯例是换行分隔defsend_request(self,method:str,params:dict,request_id:int)-dict:发送请求阻塞等待响应request{jsonrpc:2.0,id:request_id,method:method,params:params,}# 写一行 JSON 到 stdinself._proc.stdin.write(json.dumps(request)\n)self._proc.stdin.flush()# 从 stdout 读一行 JSON响应response_lineself._proc.stdout.readline()returnjson.loads(response_line)核心就两行stdin.write(json \n)发请求stdout.readline()读响应。initialize 握手实现连接后第一步是握手definitialize(self)-dict:initialize 握手responseself.send_request(initialize,{protocolVersion:2024-11-05,capabilities:{},clientInfo:{name:deep-pilot,version:0.8.0},},request_id1,)# 握手成功后发 initialized 通知self._send_notification(notifications/initialized)returnresponse[result]注意initialized是通知notification没有id字段Server 不回复。def_send_notification(self,method:str,params:dict|NoneNone):发送通知无 idServer 不回复notification{jsonrpc:2.0,method:method,params:paramsor{},}self._proc.stdin.write(json.dumps(notification)\n)self._proc.stdin.flush()工具发现tools/list握手后调用tools/list发现 Server 提供的工具deflist_tools(self)-list[dict]:发现工具responseself.send_request(tools/list,{},request_id2)returnresponse[result][tools]返回的每个工具带name、description、inputSchemaJSON Schema和第36节的 tool 反射格式一致。工具调用tools/call 与结果转换调用具体工具defcall_tool(self,name:str,arguments:dict)-str:调用工具返回文本结果responseself.send_request(tools/call,{name:name,arguments:arguments},request_id3,)# 检查错误iferrorinresponse:returnf错误{response[error][message]}# 转换结果MCP 返回 content 列表每个是 {type, text}resultresponse[result]contentresult.get(content,[])# 提取文本内容texts[item.get(text,)foritemincontentifitem.get(type)text]return\n.join(texts)关键结果转换。MCP 的tools/call返回的content是一个列表每个元素是{type, text}需要提取text字段转成普通字符串才能回填给模型。完整实现mcp_client.py# deep_pilot/mcp_client.py —— 手写 MCP 客户端 v0.8from__future__importannotationsimportjsonimportsubprocessfromtypingimportAnyfromdeep_pilot.loggerimportget_logger loggerget_logger(__name__)classMCPClient:手写 MCP 客户端——stdio 传输 JSON-RPCdef__init__(self,command:list[str]):command: 启动 Server 的命令如 [npx, -y, modelcontextprotocol/server-filesystem, /tmp]self._procsubprocess.Popen(command,stdinsubprocess.PIPE,stdoutsubprocess.PIPE,stderrsubprocess.PIPE,textTrue,)self._request_id0self.server_info{}definitialize(self)-dict:握手self._request_id1responseself._send_request(initialize,{protocolVersion:2024-11-05,capabilities:{},clientInfo:{name:deep-pilot,version:0.8.0},},)self._send_notification(notifications/initialized)self.server_inforesponse.get(result,{})logger.info(f握手成功Server: {self.server_info.get(serverInfo, {}).get(name)})returnself.server_infodeflist_tools(self)-list[dict]:发现工具self._request_id1responseself._send_request(tools/list,{})returnresponse.get(result,{}).get(tools,[])defcall_tool(self,name:str,arguments:dict)-str:调用工具返回文本self._request_id1responseself._send_request(tools/call,{name:name,arguments:arguments},)iferrorinresponse:returnf错误{response[error].get(message,未知错误)}contentresponse.get(result,{}).get(content,[])texts[item.get(text,)foritemincontentifitem.get(type)text]return\n.join(texts)defclose(self):关闭连接ifself._procandself._proc.poll()isNone:self._proc.terminate()self._proc.wait(timeout5)def_send_request(self,method:str,params:dict)-dict:发送请求有 id阻塞等待响应self._request_id1request{jsonrpc:2.0,id:self._request_id,method:method,params:params,}self._proc.stdin.write(json.dumps(request)\n)self._proc.stdin.flush()logger.debug(f→ MCP 请求:{method}{json.dumps(params,ensure_asciiFalse)[:100]})response_lineself._proc.stdout.readline()responsejson.loads(response_line)logger.debug(f← MCP 响应:{json.dumps(response,ensure_asciiFalse)[:100]})returnresponsedef_send_notification(self,method:str,params:dict|NoneNone):发送通知无 idServer 不回复notification{jsonrpc:2.0,method:method,params:paramsor{},}self._proc.stdin.write(json.dumps(notification)\n)self._proc.stdin.flush()实测wire 层报文全抓包uv run python-c from deep_pilot.mcp_client import MCPClient from deep_pilot.logger import get_logger # 启动 filesystem MCP Server client MCPClient([npx, -y, modelcontextprotocol/server-filesystem, .]) logger get_logger(test) logger.setLevel(10) # DEBUG 级别看到抓包 # 1. 握手 client.initialize() # 2. 发现工具 tools client.list_tools() print(f发现 {len(tools)} 个工具:) for t in tools[:5]: print(f - {t[\name\]}: {t[\description\][:40]}) # 3. 调用工具 result client.call_tool(read_file, {path: pyproject.toml}) print(f读取结果前100字符: {result[:100]}) client.close() 控制台输出含抓包精简→ MCP 请求: initialize {protocolVersion:2024-11-05,...} ← MCP 响应: {jsonrpc:2.0,id:1,result:{protocolVersion:2024-11-05,...}} 握手成功Server: filesystem → MCP 请求: tools/list {} ← MCP 响应: {jsonrpc:2.0,id:2,result:{tools:[{name:read_file,...}]}} 发现 12 个工具: - read_file: 读取文件 - write_file: 写入文件 - edit_file: 编辑文件 - read_multiple_files: 批量读取 - list_directory: 列出目录 → MCP 请求: tools/call {name:read_file,arguments:{path:pyproject.toml}} ← MCP 响应: {jsonrpc:2.0,id:3,result:{content:[{type:text,text:[project]\n...}]}} 读取结果前100字符: [project] name deep-pilot ...抓包清晰展示了 wire 层全过程握手 → 发现工具12个→ 调用工具 → 拿到结果。注意 filesystem Server 自带的read_file和我们第40节写的read_file名字一样、功能一样——这就是 MCP 的标准化带来的复用。小结手写客户端理解本质核心就是 stdin 写 JSON、stdout 读 JSON几十行搞定。stdio 传输通过标准输入输出与 Server 进程通信每行一个 JSON。initialize 握手发 initialize带协议版本收能力声明再发 initialized 通知。tools/list 发现返回带 inputSchema 的工具列表格式与 tool 反射一致。tools/call 调用返回 content 列表需提取 text 字段转普通字符串。请求 vs 通知请求有 id 等响应通知无 id 不等待。抓包验证握手、发现12个工具、调用、拿结果wire 层全程透明。DeepPilot v0.8 手写 MCP 客户端完成——从看懂协议到能用协议。下节预告客户端能连 MCP Server、发现工具、调用工具了。但还差最后一步——把这些 MCP 工具注入 DeepPilot 的工具注册中心让 Agent 像用自家工具一样用 MCP 工具。下一节做动态工具注入MCP 工具到 harness 注册中心的桥接、多 Server 管理、命名冲突处理接入 filesystem 和 fetch 两个官方 Server 实测。如果觉得本文对你有帮助欢迎点赞、收藏、关注三连本系列持续更新中关注不迷路~