1. 先弄明白MCP到底在解决什么问题

很多朋友第一次看到 MCP（Model 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，不依赖官方 SDK

2.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, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, bufsize=1, text=True, ) 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_id=1) def list_tools(self): return self._send_request("tools/list", {}, msg_id=2) def call_tool(self, name: str, arguments: dict): return self._send_request("tools/call", { "name": name, "arguments": arguments }, msg_id=3) 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'], indent=2)}") 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_id=1, 2, 3这样递增的整数。实际生产中，id 通常用 uuid 或者原子递增器，因为 Client 可能需要同时发出多个请求，然后根据返回消息里的 id 去匹配对应的请求。这就像你同时寄了好几封信，每封信都编了号，收信人回信的时候带上编号，你才知道哪封回的是哪封。

communication loop 的出口条件：stdout.readline()会一直阻塞直到收到完整一行。如果 Server 进程崩溃了，这个方法会返回空字符串，所以完整的实现里必须加一个空响应判断，避免死循环。这个问题在后面的“常见问题”里会再提到。

4. 实战：让 AI 模型真正“用起来”MCP 服务

到这里，我们已经实现了一个完整的“手动挡”MCP 链路：Client 能发现工具、能调用工具、能拿到结果。但要让 AI 模型自动决定“该调用哪个工具”，还需要再加一层“意图映射”。

4.1 从模型意图到工具调用的映射机制

真正的 Agent 应用里，大致流程是这样的：

1. 客户端从 MCP Server 获取工具列表（tools/list）。
2. 把每个工具的name、description、inputSchema转成模型能理解的功能调用定义。
3. 模型在对话过程中，根据上下文决定是否调用某个函数，并返回一个结构化的调用指令。
4. 客户端解析这个调用指令，提取tool_name和tool_arguments，然后向 MCP Server 发送tools/call。
5. 拿到执行结果后，把结果作为新的上下文回传给模型，模型基于结果继续生成回答。

为了方便演示，我写了一个不依赖真实模型 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 排查问题速查表

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 最有效的一条路。