1. 先弄明白MCP到底在解决什么问题很多朋友第一次看到 MCPModel Context Protocol模型上下文协议这个名字第一反应是“又出了一个新框架”。我一开始也这么想直到有一天需要把一个内部的数据查询服务接到 AI 模型上才意识到 MCP 并不是又一个“框架”而是整个 AI 工程化链条里非常基础的一块拼图。1.1 没有 MCP 之前每次接一个新工具都要“重写轮子”先回到最原始的痛苦场景。假设你已经有一个跑得很好的服务比如一个天气查询接口、一个数据库查询入口或者一个内部的知识库搜索工具现在想让 AI 模型能用上它。传统做法是给模型写一个 function calling 的接口定义把入参说明、返回值结构告诉模型模型在对话过程中判断要不要调用这个函数然后你把模型返回的调用参数接住再去调真实的接口把结果塞回给模型。听上去不复杂但问题出在“每一个服务都得来一遍”。天气接口要定义一套入参格式数据库查询要定义一套权限模型知识库搜索要定义一套相关性排序参数。这些接口的定义方式、传输协议、认证方式、错误处理逻辑全都不一样。你每接入一个新工具就要在业务代码里多写几百行胶水代码而这些代码九成以上是在做同一件事把模型的意图翻译成某个服务的请求格式。MCP 想解决的就是这个“重复造轮子”的问题。它把工具接模型这件事标准化了工具提供方只要按 MCP 的规范把能力暴露出来模型这一侧就能用统一的协议去发现、调用、收结果。你可以把它类比成 AI 世界的 USB 接口——以前每种设备都要配专属线现在只要设备都支持 USB一根线解决所有问题。1.2 MCP 的核心架构Host、Client、Server 三段各管一段MCP 的架构拆开看其实只有三个角色Host、Client、Server。Server 是能力提供方负责把真实的服务封装成标准接口。比如一个 Server 可以把数据库查询能力暴露成query_users这个工具把文件操作能力暴露成read_file工具。它只关心“我能做什么”不关心谁来调用。Client 是通信层负责建立和维护模型与 Server 之间的连接。它处理握手、消息收发、错误重试这些底层细节。你可以把它理解成一个“协议翻译官”让两端的消息都变成 MCP 标准格式。Host 是决策方通常是 AI 模型所在的应用层。它负责决定“什么时候调用哪个工具”“传什么参数”“拿结果之后怎么组织回复”。打个比方Server 是后厨配菜师傅把菜切好装盘Client 是传菜员把菜从后厨端到前厅Host 是顾客负责决定点什么菜、怎么吃。没有传菜员配菜师傅和顾客之间就得喊话传输效率低还容易出错。1.3 为什么现在值得静下心来学 MCP这个协议从发布到现在的推广速度非常快主流 AI 应用和三方服务基本都在朝这个标准靠拢。一个很直接的原因是MCP 官方已经提供了成熟的 Python 和 TypeScript SDK你不需要从零实现协议层只需要写自己业务的那部分逻辑。但这里要提醒一句直接用 SDK 写代码和真正理解协议是两回事。如果你只是照着文档调用 FastMCP 几句代码遇到连接异常、参数协商失败、会话失效这些问题的时候很容易被底层错误信息绕晕。这也是我坚持“从零手搓代码”的原因——把协议最核心的部分敲一遍后面再用 SDK 就顺手很多。2. 动手实现一个可运行的 MCP 服务端接下来直接上代码。我会先用官方 SDK 快速搭一个 MCP Server因为这个过程能让你先看到“最终成品”长什么样然后下一节再脱离 SDK从协议层面手搓一个 Client彻底把链路打通。2.1 环境准备与项目结构建议使用 Python 3.10 及以上版本。MCP 的官方 SDK 对异步支持比较依赖新版本语法3.10 以下的版本跑起来容易遇到兼容性问题。mkdir mcp-handson cd mcp-handson python3 -m venv venv source venv/bin/activate pip install mcp项目结构先保持精简方便后面展开mcp-handson/ ├── server.py # MCP Server暴露具体工具能力 └── client_raw.py # 手写的 MCP Client不依赖官方 SDK2.2 代码实现一个带两个真实工具的 MCP Server下面这个服务端暴露两个工具一个回显工具echo_text一个返回当前时间的工具get_current_time。功能很简单但足够说明 MCP 的工具定义机制。# server.py import json from datetime import datetime from mcp.server.fastmcp import FastMCP mcp FastMCP(my-demo-server) mcp.tool() def echo_text(content: str) - dict: 把传入内容原样返回用来验证链路连通性。 return { content: content, length: len(content) } mcp.tool() def get_current_time(format: str iso) - dict: 返回当前 UTC 时间format 支持 iso 或 readable。 now datetime.utcnow() if format readable: return {datetime: now.strftime(%Y-%m-%d %H:%M:%S)} return {datetime: now.isoformat()} if __name__ __main__: mcp.run()有几个细节值得展开函数的 docstring 就是工具的说明书。FastMCP 会把函数的docstring自动映射为 MCP 工具的描述信息这个描述字段非常重要因为模型在决定“什么时候调用这个工具”时主要就是靠这段描述。如果你写得太含糊模型很可能在迭代过程中反复调用错误的工具。建议描述里写清楚“功能是什么”“什么时候应该调用”“什么时候不该调用”。参数类型注解决定了模型收到什么格式。echo_text(content: str)会被自动转换成 JSON Schema 里的string类型get_current_time(format: str iso)则会被转换成带默认值的可选参数。这里有个建议参数默认值尽量给因为模型在不确定该传什么的时候有默认值会比直接报错好得多。2.3 用 MCP Inspector 做可视化调试服务端写完之后建议不要急着进下一步先用官方提供的可视化调试工具确认服务是通的。Inspector 是一个基于网页的调试面板能直接连上你的本地 Server。npx modelcontextprotocol/inspector python server.py启动之后浏览器打开 Inspector 给出的本地地址你会看到一系列测试按钮可以查看服务端的工具列表可以直接调用某个工具并查看原始返回结果还能看到完整的 JSON-RPC 通信日志。我第一次用 Inspector 的时候触动最大的就是能看到每个工具的inputSchema长什么样。有了这个可视化的 Schema 视图你就能直观理解为什么函数签名不能乱写——任何不规范的注解都会直接反映到 Schema 里进而影响模型端的调用效果。3. 写一个不带 SDK 的 MCP Client从零手搓核心链路这一节是全文的核心。官方 SDK 帮你把协议层藏得严严实实但如果你不看底层出了问题就只能瞎猜。我建议你花一两个小时用原生 io 读写来实现一个最小可用的 MCP Client。3.1 为什么要脱离 SDK 手写MCP 底层走的是标准的 JSON-RPC 2.0 协议传输方式默认是 stdio——也就是说Client 和 Server 之间通过标准输入输出流通信。Client 往 Server 的 stdin 写 JSON 消息Server 往自己的 stdout 返回 JSON 响应。这里有一个极其关键、也极其容易被忽视的坑任何额外的 print 调试信息都会污染协议流。SDK 里所有打印日志的功能都是走 stderr 的因为 stdout 被协议占用了。手写 Client 的时候如果你在 Server 代码里加一句print(hello)客户端解析就会直接爆炸——这句话会被当成一条 JSON-RPC 消息来解析。理解了这个传输模型后面所有代码就顺理成章了。3.2 一步步实现 Client先看完整代码再逐段拆解# client_raw.py import json import subprocess import sys import uuid class RawMCPClient: def __init__(self, command: list[str]): # 启动 Server 子进程stdin/stdout 作为通信管道 self.proc subprocess.Popen( command, stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, bufsize1, textTrue, ) def _send_request(self, method: str, params: dict, msg_id: int): payload { jsonrpc: 2.0, id: msg_id, method: method, params: params, } # 每条 JSON 之后必须加换行符作为消息边界 line json.dumps(payload) \n self.proc.stdin.write(line) self.proc.stdin.flush() response_line self.proc.stdout.readline() response json.loads(response_line) if error in response: raise RuntimeError(response[error]) return response def initialize(self): return self._send_request(initialize, { protocolVersion: 2024-11-05, capabilities: {}, clientInfo: { name: raw-client, version: 0.1.0 } }, msg_id1) def list_tools(self): return self._send_request(tools/list, {}, msg_id2) def call_tool(self, name: str, arguments: dict): return self._send_request(tools/call, { name: name, arguments: arguments }, msg_id3) if __name__ __main__: client RawMCPClient([python, server.py]) init_response client.initialize() print(Server info:, init_response[result][serverInfo]) print(Capabilities:, init_response[result][capabilities]) tools_response client.list_tools() tools tools_response[result][tools] print(Available tools:) for tool in tools: print(f- {tool[name]}: {tool[description]}) print(f Input schema: {json.dumps(tool[inputSchema], indent2)}) echo_result client.call_tool(echo_text, {content: hello mcp}) print(Echo result:, echo_result[result][content]) time_result client.call_tool(get_current_time, {format: readable}) print(Time result:, time_result[result][content])这段代码的通信逻辑很简单启动 Server 子进程后按行写入 JSON 请求按行读取 JSON 响应。initialize 握手的消息内容有讲究。protocolVersion必须填一个服务端支持的版本号这里用2024-11-05是当前较新的正式版本。capabilities字段标明了客户端自己支持的能力——即使当前什么都不支持这个字段也要存在格式必须合法。tools/list 是纯发现操作。它不触发任何工具执行只是拿回服务端所有工具的定义列表。返回结果里的inputSchema就是后续让模型判断“应该传什么参数”的依据。tools/call 是真正的执行请求。注意这里只传工具名和参数字典不需要传协议版本这些信息。返回结果在result.content里。3.3 协议细节里的几个关键点消息边界JSON-RPC 消息不是通过“括号配对”来辨认边界的而是通过换行符。这也解释了为什么 Server 端的 stdout 不能有任何额外输出——每条\n结尾的字符串都会被读成一条独立的 JSON。请求 id 的作用你可以看到手写代码里我用的是msg_id1, 2, 3这样递增的整数。实际生产中id 通常用 uuid 或者原子递增器因为 Client 可能需要同时发出多个请求然后根据返回消息里的 id 去匹配对应的请求。这就像你同时寄了好几封信每封信都编了号收信人回信的时候带上编号你才知道哪封回的是哪封。communication loop 的出口条件stdout.readline()会一直阻塞直到收到完整一行。如果 Server 进程崩溃了这个方法会返回空字符串所以完整的实现里必须加一个空响应判断避免死循环。这个问题在后面的“常见问题”里会再提到。4. 实战让 AI 模型真正“用起来”MCP 服务到这里我们已经实现了一个完整的“手动挡”MCP 链路Client 能发现工具、能调用工具、能拿到结果。但要让 AI 模型自动决定“该调用哪个工具”还需要再加一层“意图映射”。4.1 从模型意图到工具调用的映射机制真正的 Agent 应用里大致流程是这样的客户端从 MCP Server 获取工具列表tools/list。把每个工具的name、description、inputSchema转成模型能理解的功能调用定义。模型在对话过程中根据上下文决定是否调用某个函数并返回一个结构化的调用指令。客户端解析这个调用指令提取tool_name和tool_arguments然后向 MCP Server 发送tools/call。拿到执行结果后把结果作为新的上下文回传给模型模型基于结果继续生成回答。为了方便演示我写了一个不依赖真实模型 API 的模拟版本核心是把“模型返回工具调用”这个行为简化成一个规则函数你可以在本地把整条链路跑通理解位置关系后再去接真实的模型服务。# agent_loop_demo.py import json import client_raw def simulate_model_decision(user_input: str) - dict: 模拟模型判断根据输入内容决定调用哪个工具。 if 几点 in user_input or 时间 in user_input: return { tool_name: get_current_time, tool_arguments: {format: readable} } if 回显 in user_input or echo in user_input.lower(): return { tool_name: echo_text, tool_arguments: {content: user_input} } return None def run_agent_cycle(user_input: str): tool_name None tool_args None decision simulate_model_decision(user_input) if decision: tool_name decision[tool_name] tool_args decision[tool_arguments] if tool_name is None: print(模型认为无需调用工具直接回复。) return raw_client client_raw.RawMCPClient([python, server.py]) raw_client.initialize() print(f[Agent] 决定调用工具: {tool_name}, 参数: {tool_args}) result raw_client.call_tool(tool_name, tool_args) print(f[Agent] 工具返回: {result[result][content]}) final_text f根据工具返回{当前时间为 result[result][content][datetime] if tool_name get_current_time else 回显结果为 result[result][content][content]} print(f[Agent] 最终回复: {final_text}) raw_client.proc.terminate() if __name__ __main__: run_agent_cycle(现在几点) run_agent_cycle(帮我回显一下这段话)这个 demo 虽然简单但完整复刻了真实 Agent 的三个关键环节判断工具、执行工具、把结果组织成回复。真实场景里模型返回的不一定是“工具调用决定”而是一段结构化 JSON解析这个 JSON 的逻辑是相通的。4.2 完整 Agent 循环的雏形从返回结果到下一次对话如果继续扩展真实 Agent 会进入循环结构第一次调用工具拿到结果后结果重新作为上下文进入模型模型可能发现信息不足继续要求调用第二个工具。比如用户问“北京现在几点”模型先调用地理编码工具拿到时区再调用时间工具才能给出正确答案。这个循环有很多工程细节要处理结果是结构化字典还是纯文本。很多模型服务要求回传上下文必须是纯文本或消息数组而 MCP 返回的content字段是一个结构化结构数组。你在回传给模型之前需要决定是原样保留结构还是把它压成字符串。我的经验是如果模型服务支持多模态上下文保留结构化数据更好如果只支持纯文本就自己拼一个稳定的格式比如tool_result: { key: value }。多工具并行。模型在一次响应里可能同时要求调用两个互不依赖的工具。这种情况下可以用并发发送多个tools/call然后根据消息 id 把结果归类。但注意 MCP 默认的 stdio 传输是单通道如果服务端不支持并发处理就要做串行化处理否则可能出现返回顺序错乱。工具调用的历史上下文量。每轮对话把工具返回结果都塞进上下文token 会涨得飞快。比较常见的做法是只保留最近两轮的工具调用记录更早的结果可以压缩成摘要比如只保留最终影响回复的关键数值。4.3 工具调用失败的兜底策略超时、空结果、参数错误MCP 链路里最让新手头疼的就是“模型调用工具时报错”。我在实际调试中发现错误绝大多数发生在三个位置对应的处理策略差别很大。超时工具执行时间过长模型服务那边可能先等不住了。建议给 MCP Client 加一层超时控制最简单的方式是用select模块监听 stdout 的可读状态超过 20 秒仍没有数据就主动终止请求。真实场景里耗时长的工具不要直接在 MCP 内部同步执行而是先返回一个“任务已提交task_id 为 xxx”的占位结果客户端后续用单独的查询接口轮询进度。空结果不是错误但比错误更麻烦。工具正常返回了但 content 是空数组模型拿不到任何信息只能自己瞎编。我在代码里会强制校验result.content非空如果为空就拼一条明确的提示“工具执行成功但无返回结果”避免模型的编造倾向。参数错误模型传了 Schema 里不存在的字段。MCP Server 的 SDK 一般会自动过滤掉未定义字段但当你手写 Server 的时候就完全没有这层保护。我的建议是在 Server 内部对所有参数做显式白名单检查不符合 Schema 的字符串裁掉类型不对的尝试强转实在不对就返回结构化错误信息——这比抛异常更友好因为错误信息会直接出现在模型上下文里模型可以据此自我修正。5. 运行过程中踩过的坑和排查记录手写 MCP 链路的过程中我踩了不少坑。有些错误信息非常隐晦明明端口通了、进程也起来了但 Client 就是拿不到响应。按“从高频到低频”的顺序整理一个速查表。5.1 排查问题速查表症状可能原因解决方案Client 启动后长时间无输出Server 进程可能启动失败单独在终端跑一遍python server.py看 stderr 有没有报错收到Parse errorstdout 被调试输出污染检查 Server 代码里是否有print全部改成logger输出到 stderr工具列表为空mcp.tool()装饰器未注册成功确认函数定义在mcp.run()之前且没有被条件判断跳过initialize后调用tools/list报错协议版本不匹配检查protocolVersion是否在 Server 支持的版本列表里调用工具后返回Method not found工具名拼写不一致用tools/list返回结果里的准确name字段不要凭记忆写应响应但一直阻塞stdout 空行检查bufsize1是否设置文本模式下按行读写需要行缓冲stdout.readline()读到空字符串Server 进程已退出打印 stderr 内容大概率是代码执行到一半抛异常两次请求返回结果错乱没有按 id 匹配响应检查是否并发发送了多个请求确认读取循环里对id做了校验5.2 三个最常见的坑重点展开坑一闷头用 SDK看不懂底层错误信息直接用 FastMCP 写 Server 时它内部帮你处理了所有底层通信。遇到错误时报错堆栈往往指向 SDK 内部的异步事件循环新手根本无从下手。这也是我建议“先手写一遍 Client”的原因。手写之后每个字节都是明确的发送时你知道发的是什么读取时你期待的是什么。再回去用 SDK很多错误信息一眼就能定位到具体环节。坑二stdout 污染协议层直接爆炸前面反复提到再敲一遍重点stdout 是协议通道不是日志通道。很多人写 Server 的时候顺手print一个调试信息结果 Client 端直接抛JSONDecodeError而且错误信息完全看不出是哪里丢了一个print。排查思路把所有打印日志都切到stderr模块或者干脆先用注释/日志框架管理输出。坑三异步与事件循环的生命周期官方 SDK 的服务端内部是异步事件循环如果你在一个已被事件循环控制的进程里再启动一个同步的 Server大概率会卡住或者冲突。比较常见的场景是在 Flask 或 FastAPI 的请求处理函数里调用mcp.run()这会导致两个框架抢事件循环。正确的做法是让 MCP Server 作为一个独立的子进程运行而不是嵌入到其他 Web 进程里。如果非要嵌入需要把 MCP 的异步函数放进同一个事件循环里协调调度。6. 给初学者的进阶建议写完这遍最小可运行的 MCP 链路之后我建议你做三件事来加深理解并且这三件事我在不同项目里都亲手验证过收益非常直接。第一把 Server 换成 TypeScript 版本再写一遍。MCP 的生态里 Node.js 是一等公民很多现成服务都是 JS 实现的。Python 能读懂协议规则TypeScript 能帮你确认这套协议跟编程语言无关——它是纯粹的 JSON 消息交换。第二给 Server 加一个真正的“有状态工具”。比如一个计数器工具第一次调用返回 1第二次返回 2。这个练习会让你注意 MCP Server 进程的存活时间、状态在子进程中如何保持、会话结束后状态是否重置。这些细节在无状态工具里完全暴露不出来。第三把 stdout 传输换成 SSE 模式跑一遍。MCP 支持的不只是 stdin/stdout还有更接近真实生产的 SSE 传输。实现一个最小 SSE 版本能让你理解为什么说 stdio 适合本地工具集成而 SSE 适合远程服务。最后再分享一个个人经验学习 MCP 不要绕过基础协议直接硬啃 SDK。凡是能让你少写代码的封装都会同时让你少看到一些真相。花一个下午把协议层的请求响应看透后面写任何工具的集成代码都不会再有“黑盒恐惧”。这个投入的回报率极高至少对我来说它是彻底搞懂 MCP 最有效的一条路。
1. 先弄明白MCP到底在解决什么问题很多朋友第一次看到 MCPModel Context Protocol模型上下文协议这个名字第一反应是“又出了一个新框架”。我一开始也这么想直到有一天需要把一个内部的数据查询服务接到 AI 模型上才意识到 MCP 并不是又一个“框架”而是整个 AI 工程化链条里非常基础的一块拼图。1.1 没有 MCP 之前每次接一个新工具都要“重写轮子”先回到最原始的痛苦场景。假设你已经有一个跑得很好的服务比如一个天气查询接口、一个数据库查询入口或者一个内部的知识库搜索工具现在想让 AI 模型能用上它。传统做法是给模型写一个 function calling 的接口定义把入参说明、返回值结构告诉模型模型在对话过程中判断要不要调用这个函数然后你把模型返回的调用参数接住再去调真实的接口把结果塞回给模型。听上去不复杂但问题出在“每一个服务都得来一遍”。天气接口要定义一套入参格式数据库查询要定义一套权限模型知识库搜索要定义一套相关性排序参数。这些接口的定义方式、传输协议、认证方式、错误处理逻辑全都不一样。你每接入一个新工具就要在业务代码里多写几百行胶水代码而这些代码九成以上是在做同一件事把模型的意图翻译成某个服务的请求格式。MCP 想解决的就是这个“重复造轮子”的问题。它把工具接模型这件事标准化了工具提供方只要按 MCP 的规范把能力暴露出来模型这一侧就能用统一的协议去发现、调用、收结果。你可以把它类比成 AI 世界的 USB 接口——以前每种设备都要配专属线现在只要设备都支持 USB一根线解决所有问题。1.2 MCP 的核心架构Host、Client、Server 三段各管一段MCP 的架构拆开看其实只有三个角色Host、Client、Server。Server 是能力提供方负责把真实的服务封装成标准接口。比如一个 Server 可以把数据库查询能力暴露成query_users这个工具把文件操作能力暴露成read_file工具。它只关心“我能做什么”不关心谁来调用。Client 是通信层负责建立和维护模型与 Server 之间的连接。它处理握手、消息收发、错误重试这些底层细节。你可以把它理解成一个“协议翻译官”让两端的消息都变成 MCP 标准格式。Host 是决策方通常是 AI 模型所在的应用层。它负责决定“什么时候调用哪个工具”“传什么参数”“拿结果之后怎么组织回复”。打个比方Server 是后厨配菜师傅把菜切好装盘Client 是传菜员把菜从后厨端到前厅Host 是顾客负责决定点什么菜、怎么吃。没有传菜员配菜师傅和顾客之间就得喊话传输效率低还容易出错。1.3 为什么现在值得静下心来学 MCP这个协议从发布到现在的推广速度非常快主流 AI 应用和三方服务基本都在朝这个标准靠拢。一个很直接的原因是MCP 官方已经提供了成熟的 Python 和 TypeScript SDK你不需要从零实现协议层只需要写自己业务的那部分逻辑。但这里要提醒一句直接用 SDK 写代码和真正理解协议是两回事。如果你只是照着文档调用 FastMCP 几句代码遇到连接异常、参数协商失败、会话失效这些问题的时候很容易被底层错误信息绕晕。这也是我坚持“从零手搓代码”的原因——把协议最核心的部分敲一遍后面再用 SDK 就顺手很多。2. 动手实现一个可运行的 MCP 服务端接下来直接上代码。我会先用官方 SDK 快速搭一个 MCP Server因为这个过程能让你先看到“最终成品”长什么样然后下一节再脱离 SDK从协议层面手搓一个 Client彻底把链路打通。2.1 环境准备与项目结构建议使用 Python 3.10 及以上版本。MCP 的官方 SDK 对异步支持比较依赖新版本语法3.10 以下的版本跑起来容易遇到兼容性问题。mkdir mcp-handson cd mcp-handson python3 -m venv venv source venv/bin/activate pip install mcp项目结构先保持精简方便后面展开mcp-handson/ ├── server.py # MCP Server暴露具体工具能力 └── client_raw.py # 手写的 MCP Client不依赖官方 SDK2.2 代码实现一个带两个真实工具的 MCP Server下面这个服务端暴露两个工具一个回显工具echo_text一个返回当前时间的工具get_current_time。功能很简单但足够说明 MCP 的工具定义机制。# server.py import json from datetime import datetime from mcp.server.fastmcp import FastMCP mcp FastMCP(my-demo-server) mcp.tool() def echo_text(content: str) - dict: 把传入内容原样返回用来验证链路连通性。 return { content: content, length: len(content) } mcp.tool() def get_current_time(format: str iso) - dict: 返回当前 UTC 时间format 支持 iso 或 readable。 now datetime.utcnow() if format readable: return {datetime: now.strftime(%Y-%m-%d %H:%M:%S)} return {datetime: now.isoformat()} if __name__ __main__: mcp.run()有几个细节值得展开函数的 docstring 就是工具的说明书。FastMCP 会把函数的docstring自动映射为 MCP 工具的描述信息这个描述字段非常重要因为模型在决定“什么时候调用这个工具”时主要就是靠这段描述。如果你写得太含糊模型很可能在迭代过程中反复调用错误的工具。建议描述里写清楚“功能是什么”“什么时候应该调用”“什么时候不该调用”。参数类型注解决定了模型收到什么格式。echo_text(content: str)会被自动转换成 JSON Schema 里的string类型get_current_time(format: str iso)则会被转换成带默认值的可选参数。这里有个建议参数默认值尽量给因为模型在不确定该传什么的时候有默认值会比直接报错好得多。2.3 用 MCP Inspector 做可视化调试服务端写完之后建议不要急着进下一步先用官方提供的可视化调试工具确认服务是通的。Inspector 是一个基于网页的调试面板能直接连上你的本地 Server。npx modelcontextprotocol/inspector python server.py启动之后浏览器打开 Inspector 给出的本地地址你会看到一系列测试按钮可以查看服务端的工具列表可以直接调用某个工具并查看原始返回结果还能看到完整的 JSON-RPC 通信日志。我第一次用 Inspector 的时候触动最大的就是能看到每个工具的inputSchema长什么样。有了这个可视化的 Schema 视图你就能直观理解为什么函数签名不能乱写——任何不规范的注解都会直接反映到 Schema 里进而影响模型端的调用效果。3. 写一个不带 SDK 的 MCP Client从零手搓核心链路这一节是全文的核心。官方 SDK 帮你把协议层藏得严严实实但如果你不看底层出了问题就只能瞎猜。我建议你花一两个小时用原生 io 读写来实现一个最小可用的 MCP Client。3.1 为什么要脱离 SDK 手写MCP 底层走的是标准的 JSON-RPC 2.0 协议传输方式默认是 stdio——也就是说Client 和 Server 之间通过标准输入输出流通信。Client 往 Server 的 stdin 写 JSON 消息Server 往自己的 stdout 返回 JSON 响应。这里有一个极其关键、也极其容易被忽视的坑任何额外的 print 调试信息都会污染协议流。SDK 里所有打印日志的功能都是走 stderr 的因为 stdout 被协议占用了。手写 Client 的时候如果你在 Server 代码里加一句print(hello)客户端解析就会直接爆炸——这句话会被当成一条 JSON-RPC 消息来解析。理解了这个传输模型后面所有代码就顺理成章了。3.2 一步步实现 Client先看完整代码再逐段拆解# client_raw.py import json import subprocess import sys import uuid class RawMCPClient: def __init__(self, command: list[str]): # 启动 Server 子进程stdin/stdout 作为通信管道 self.proc subprocess.Popen( command, stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, bufsize1, textTrue, ) def _send_request(self, method: str, params: dict, msg_id: int): payload { jsonrpc: 2.0, id: msg_id, method: method, params: params, } # 每条 JSON 之后必须加换行符作为消息边界 line json.dumps(payload) \n self.proc.stdin.write(line) self.proc.stdin.flush() response_line self.proc.stdout.readline() response json.loads(response_line) if error in response: raise RuntimeError(response[error]) return response def initialize(self): return self._send_request(initialize, { protocolVersion: 2024-11-05, capabilities: {}, clientInfo: { name: raw-client, version: 0.1.0 } }, msg_id1) def list_tools(self): return self._send_request(tools/list, {}, msg_id2) def call_tool(self, name: str, arguments: dict): return self._send_request(tools/call, { name: name, arguments: arguments }, msg_id3) if __name__ __main__: client RawMCPClient([python, server.py]) init_response client.initialize() print(Server info:, init_response[result][serverInfo]) print(Capabilities:, init_response[result][capabilities]) tools_response client.list_tools() tools tools_response[result][tools] print(Available tools:) for tool in tools: print(f- {tool[name]}: {tool[description]}) print(f Input schema: {json.dumps(tool[inputSchema], indent2)}) echo_result client.call_tool(echo_text, {content: hello mcp}) print(Echo result:, echo_result[result][content]) time_result client.call_tool(get_current_time, {format: readable}) print(Time result:, time_result[result][content])这段代码的通信逻辑很简单启动 Server 子进程后按行写入 JSON 请求按行读取 JSON 响应。initialize 握手的消息内容有讲究。protocolVersion必须填一个服务端支持的版本号这里用2024-11-05是当前较新的正式版本。capabilities字段标明了客户端自己支持的能力——即使当前什么都不支持这个字段也要存在格式必须合法。tools/list 是纯发现操作。它不触发任何工具执行只是拿回服务端所有工具的定义列表。返回结果里的inputSchema就是后续让模型判断“应该传什么参数”的依据。tools/call 是真正的执行请求。注意这里只传工具名和参数字典不需要传协议版本这些信息。返回结果在result.content里。3.3 协议细节里的几个关键点消息边界JSON-RPC 消息不是通过“括号配对”来辨认边界的而是通过换行符。这也解释了为什么 Server 端的 stdout 不能有任何额外输出——每条\n结尾的字符串都会被读成一条独立的 JSON。请求 id 的作用你可以看到手写代码里我用的是msg_id1, 2, 3这样递增的整数。实际生产中id 通常用 uuid 或者原子递增器因为 Client 可能需要同时发出多个请求然后根据返回消息里的 id 去匹配对应的请求。这就像你同时寄了好几封信每封信都编了号收信人回信的时候带上编号你才知道哪封回的是哪封。communication loop 的出口条件stdout.readline()会一直阻塞直到收到完整一行。如果 Server 进程崩溃了这个方法会返回空字符串所以完整的实现里必须加一个空响应判断避免死循环。这个问题在后面的“常见问题”里会再提到。4. 实战让 AI 模型真正“用起来”MCP 服务到这里我们已经实现了一个完整的“手动挡”MCP 链路Client 能发现工具、能调用工具、能拿到结果。但要让 AI 模型自动决定“该调用哪个工具”还需要再加一层“意图映射”。4.1 从模型意图到工具调用的映射机制真正的 Agent 应用里大致流程是这样的客户端从 MCP Server 获取工具列表tools/list。把每个工具的name、description、inputSchema转成模型能理解的功能调用定义。模型在对话过程中根据上下文决定是否调用某个函数并返回一个结构化的调用指令。客户端解析这个调用指令提取tool_name和tool_arguments然后向 MCP Server 发送tools/call。拿到执行结果后把结果作为新的上下文回传给模型模型基于结果继续生成回答。为了方便演示我写了一个不依赖真实模型 API 的模拟版本核心是把“模型返回工具调用”这个行为简化成一个规则函数你可以在本地把整条链路跑通理解位置关系后再去接真实的模型服务。# agent_loop_demo.py import json import client_raw def simulate_model_decision(user_input: str) - dict: 模拟模型判断根据输入内容决定调用哪个工具。 if 几点 in user_input or 时间 in user_input: return { tool_name: get_current_time, tool_arguments: {format: readable} } if 回显 in user_input or echo in user_input.lower(): return { tool_name: echo_text, tool_arguments: {content: user_input} } return None def run_agent_cycle(user_input: str): tool_name None tool_args None decision simulate_model_decision(user_input) if decision: tool_name decision[tool_name] tool_args decision[tool_arguments] if tool_name is None: print(模型认为无需调用工具直接回复。) return raw_client client_raw.RawMCPClient([python, server.py]) raw_client.initialize() print(f[Agent] 决定调用工具: {tool_name}, 参数: {tool_args}) result raw_client.call_tool(tool_name, tool_args) print(f[Agent] 工具返回: {result[result][content]}) final_text f根据工具返回{当前时间为 result[result][content][datetime] if tool_name get_current_time else 回显结果为 result[result][content][content]} print(f[Agent] 最终回复: {final_text}) raw_client.proc.terminate() if __name__ __main__: run_agent_cycle(现在几点) run_agent_cycle(帮我回显一下这段话)这个 demo 虽然简单但完整复刻了真实 Agent 的三个关键环节判断工具、执行工具、把结果组织成回复。真实场景里模型返回的不一定是“工具调用决定”而是一段结构化 JSON解析这个 JSON 的逻辑是相通的。4.2 完整 Agent 循环的雏形从返回结果到下一次对话如果继续扩展真实 Agent 会进入循环结构第一次调用工具拿到结果后结果重新作为上下文进入模型模型可能发现信息不足继续要求调用第二个工具。比如用户问“北京现在几点”模型先调用地理编码工具拿到时区再调用时间工具才能给出正确答案。这个循环有很多工程细节要处理结果是结构化字典还是纯文本。很多模型服务要求回传上下文必须是纯文本或消息数组而 MCP 返回的content字段是一个结构化结构数组。你在回传给模型之前需要决定是原样保留结构还是把它压成字符串。我的经验是如果模型服务支持多模态上下文保留结构化数据更好如果只支持纯文本就自己拼一个稳定的格式比如tool_result: { key: value }。多工具并行。模型在一次响应里可能同时要求调用两个互不依赖的工具。这种情况下可以用并发发送多个tools/call然后根据消息 id 把结果归类。但注意 MCP 默认的 stdio 传输是单通道如果服务端不支持并发处理就要做串行化处理否则可能出现返回顺序错乱。工具调用的历史上下文量。每轮对话把工具返回结果都塞进上下文token 会涨得飞快。比较常见的做法是只保留最近两轮的工具调用记录更早的结果可以压缩成摘要比如只保留最终影响回复的关键数值。4.3 工具调用失败的兜底策略超时、空结果、参数错误MCP 链路里最让新手头疼的就是“模型调用工具时报错”。我在实际调试中发现错误绝大多数发生在三个位置对应的处理策略差别很大。超时工具执行时间过长模型服务那边可能先等不住了。建议给 MCP Client 加一层超时控制最简单的方式是用select模块监听 stdout 的可读状态超过 20 秒仍没有数据就主动终止请求。真实场景里耗时长的工具不要直接在 MCP 内部同步执行而是先返回一个“任务已提交task_id 为 xxx”的占位结果客户端后续用单独的查询接口轮询进度。空结果不是错误但比错误更麻烦。工具正常返回了但 content 是空数组模型拿不到任何信息只能自己瞎编。我在代码里会强制校验result.content非空如果为空就拼一条明确的提示“工具执行成功但无返回结果”避免模型的编造倾向。参数错误模型传了 Schema 里不存在的字段。MCP Server 的 SDK 一般会自动过滤掉未定义字段但当你手写 Server 的时候就完全没有这层保护。我的建议是在 Server 内部对所有参数做显式白名单检查不符合 Schema 的字符串裁掉类型不对的尝试强转实在不对就返回结构化错误信息——这比抛异常更友好因为错误信息会直接出现在模型上下文里模型可以据此自我修正。5. 运行过程中踩过的坑和排查记录手写 MCP 链路的过程中我踩了不少坑。有些错误信息非常隐晦明明端口通了、进程也起来了但 Client 就是拿不到响应。按“从高频到低频”的顺序整理一个速查表。5.1 排查问题速查表症状可能原因解决方案Client 启动后长时间无输出Server 进程可能启动失败单独在终端跑一遍python server.py看 stderr 有没有报错收到Parse errorstdout 被调试输出污染检查 Server 代码里是否有print全部改成logger输出到 stderr工具列表为空mcp.tool()装饰器未注册成功确认函数定义在mcp.run()之前且没有被条件判断跳过initialize后调用tools/list报错协议版本不匹配检查protocolVersion是否在 Server 支持的版本列表里调用工具后返回Method not found工具名拼写不一致用tools/list返回结果里的准确name字段不要凭记忆写应响应但一直阻塞stdout 空行检查bufsize1是否设置文本模式下按行读写需要行缓冲stdout.readline()读到空字符串Server 进程已退出打印 stderr 内容大概率是代码执行到一半抛异常两次请求返回结果错乱没有按 id 匹配响应检查是否并发发送了多个请求确认读取循环里对id做了校验5.2 三个最常见的坑重点展开坑一闷头用 SDK看不懂底层错误信息直接用 FastMCP 写 Server 时它内部帮你处理了所有底层通信。遇到错误时报错堆栈往往指向 SDK 内部的异步事件循环新手根本无从下手。这也是我建议“先手写一遍 Client”的原因。手写之后每个字节都是明确的发送时你知道发的是什么读取时你期待的是什么。再回去用 SDK很多错误信息一眼就能定位到具体环节。坑二stdout 污染协议层直接爆炸前面反复提到再敲一遍重点stdout 是协议通道不是日志通道。很多人写 Server 的时候顺手print一个调试信息结果 Client 端直接抛JSONDecodeError而且错误信息完全看不出是哪里丢了一个print。排查思路把所有打印日志都切到stderr模块或者干脆先用注释/日志框架管理输出。坑三异步与事件循环的生命周期官方 SDK 的服务端内部是异步事件循环如果你在一个已被事件循环控制的进程里再启动一个同步的 Server大概率会卡住或者冲突。比较常见的场景是在 Flask 或 FastAPI 的请求处理函数里调用mcp.run()这会导致两个框架抢事件循环。正确的做法是让 MCP Server 作为一个独立的子进程运行而不是嵌入到其他 Web 进程里。如果非要嵌入需要把 MCP 的异步函数放进同一个事件循环里协调调度。6. 给初学者的进阶建议写完这遍最小可运行的 MCP 链路之后我建议你做三件事来加深理解并且这三件事我在不同项目里都亲手验证过收益非常直接。第一把 Server 换成 TypeScript 版本再写一遍。MCP 的生态里 Node.js 是一等公民很多现成服务都是 JS 实现的。Python 能读懂协议规则TypeScript 能帮你确认这套协议跟编程语言无关——它是纯粹的 JSON 消息交换。第二给 Server 加一个真正的“有状态工具”。比如一个计数器工具第一次调用返回 1第二次返回 2。这个练习会让你注意 MCP Server 进程的存活时间、状态在子进程中如何保持、会话结束后状态是否重置。这些细节在无状态工具里完全暴露不出来。第三把 stdout 传输换成 SSE 模式跑一遍。MCP 支持的不只是 stdin/stdout还有更接近真实生产的 SSE 传输。实现一个最小 SSE 版本能让你理解为什么说 stdio 适合本地工具集成而 SSE 适合远程服务。最后再分享一个个人经验学习 MCP 不要绕过基础协议直接硬啃 SDK。凡是能让你少写代码的封装都会同时让你少看到一些真相。花一个下午把协议层的请求响应看透后面写任何工具的集成代码都不会再有“黑盒恐惧”。这个投入的回报率极高至少对我来说它是彻底搞懂 MCP 最有效的一条路。
平方复杂度无法被常规优化消除｜TaoToken 统一 Key 通道实测)
