作为一个常年写小工具、偶尔折腾智能体的独立开发者，我一度很烦“把工具交给 AI 调用”这件事。明明接口文档写清楚了，模型还是会拼错参数；明明工具就挂在同一个局域网里，它却找不到服务；好不容易调通了，换一个客户端又得重写一套协议。后来我把工具按 MCP（Model Context Protocol）的标准接入，AI 再通过 Streamable HTTP 方式来调用，整个链路才算真正稳定下来。这篇教程会从原理到实现，完整走一遍 AI 调用 MCP 的过程，重点讲基于 Streamable HTTP 的传输方式。如果你正在做智能体工具接入、想把本地脚本暴露给 AI 使用、或者只是想知道 MCP 和普通 HTTP API 到底有什么区别，这篇文章应该能帮你省不少时间。1. 模型为什么需要 MCP：从“猜 REST 接口”到“标准工具说明书”1.1 模型调用普通接口时到底在“猜”什么先聊一个容易被忽略的问题：大模型本身并不知道你的接口长什么样。你给它一个 REST API 的地址和一段自然语言描述，它每一步都在做概率预测——URL 拼对了没有、字段名写对了没有、参数类型对不对，全靠“猜”。我最早做过一个时间查询工具，写成了GET /api/time?tz=Asia/Shanghai，结果模型经常把tz解释成timezone，或者把参数直接塞进路径里。不是模型笨，是我没有给它一个足够结构化的“工具说明书”。MCP 解决的就是这个“说明书”问题。它把工具、数据资源、提示模板统一成标准化的 JSON-RPC 协议，模型侧只需要理解一套协议，就能发现并调用任意 MCP 服务器提供的功能。你可以把 MCP 理解成 USB-C 接口：以前每个外设都有自己的充电线，现在所有设备用同一个接口标准，插上就能用。1.2 MCP 协议里的四个核心概念MCP 协议不只是“工具调用”，它包含四类能力，理解它们有助于你设计好自己的 MCP 服务器。Tools（工具）：模型主动发起调用的函数。类似于 function calling 里的 function，但有了统一的 schema 描述和标准化返回格式。Resources（资源）：只读数据，类似文件或数据库查询结果。模型端可以通过resources/read读取，适合暴露配置、文档、日志。Prompts（提示模板）：服务端预设的提示词模板，客户端可以拿到后组装进对话上下文。Sampling（采样）：服务器反向请求模型生成内容。这个能力更多用于“工具内部想再用一次大模型”的场景。实际使用中最常打交道的还是 Tools。调用链路的逻辑也很清楚：先获取工具的 schema（元数据），再按 schema 填入参数并调用，最后解析结构化结果。这个流程和 OpenAI 的 function calling 在理念上是一致的，但 MCP 强调的是协议层面的标准化和客户端-服务端解耦。1.3 为什么教程选择“Streamable HTTP”而不是别的MCP 协议支持几种传输方式，本文只讲 Streamable HTTP，因为它在远程调用、浏览器端、无服务器部署这些场景里最省事。后面会专门讲它和 stdio、旧版 HTTP+SSE 的取舍。现在你只需要知道：一旦你部署好了 MCP 服务端，AI 客户端能通过一个普通 HTTPS 地址来调用它，不需要共用进程，也不需要维护长连接，非常接近“给 AI 开一个 API”的直觉体验。2. Streamable HTTP 传输方式：为什么我放弃 stdio 和旧版 SSE2.1 三种传输方式的直观对比我先放一张对比表，然后再展开讲各自的使用场景。特性stdio旧版 HTTP+SSEStreamable HTTP（本教程）通信方式子进程标准输入输出先建立 SSE，再用独立端点发消息普通 POST 请求统一发消息服务端位置必须与客户端同机可在远程，但需可见的 SSE 端点远程/本地均可是否需要常驻连接需要需要不需要浏览器/网页端支持不支持较差支持无服务器（Serverless）友好程度低低高协议复杂度低中中我最早用的是 stdio，本地调试非常顺手，因为它不需要网络配置，起一个子进程就能把工具注册给客户端。但一旦工具需要部署到云服务器、或要让多个客户端并发访问，stdio 就完全不够用。进程是跟着客户端走的，客户端退出服务就没了，根本没有“常驻服务”的概念。2.2 旧版 HTTP+SSE 的两个核心痛点随后我尝试了旧版 HTTP+SSE，也就是“先 GET 建 SSE 流、再 POST 发消息”的方式。它解决了远程部署问题，但引入两个新麻烦。第一，客户端必须先向服务器发起 SSE GET 请求，拿到一个 listener 端点，才能开始通信。这意味着你的客户端必须维护一个长期挂起的连接，而且这个连接在网关、代理、无服务器环境中很容易超时。我测试时连着几次遇到“连接建立成功，但消息发送时 listener 已断开”的问题，排查起来极其恼火。第二，它把“建立连接”和“发送消息”拆成了两个独立步骤，这让很多 HTTP 客户端框架难以处理。你不仅要管理一个永久的 HTTP 响应流，还要额外维护一个发送用的 POST 端点，状态管理非常繁琐。2.3 Streamable HTTP 的答案：一个请求里完成通信Streamable HTTP 从根本上调整了交互模式：所有 JSON-RPC 消息都由客户端通过 POST 发送，服务器在同一个 HTTP 响应里返回结果。服务器如果希望流式推送进度，可以在响应头里把Content-Type设为text/event-stream，通过 SSE 逐段返回；如果只是普通工具调用，直接返回application/json也不是问题。这就把 MCP 客户端从“必须维护长连接”的泥潭里解放了出来。你可以把一次工具调用看成一次普通 API 请求，用任何 HTTP 客户端库都能实现。服务端状态通过Mcp-Session-Id响应头来标识，客户端带着它做后续请求，相当于把会话状态外置到了客户端。这个设计非常适合云函数、Web 应用、甚至运行在浏览器沙箱里的客户端。3. 服务器端搭建：30 分钟内吐出一个能被 AI 调用的端点3.1 先规划路由与协议入口服务端本质上就是一个 HTTP 服务，核心工作就是把收到的 JSON-RPC 请求分发到对应的处理器上。以我常用的目录结构为例，整个 MCP 服务可以拆成三层：协议层：负责解析 JSON-RPC 请求、包装响应、处理会话 ID。工具层：注册具体的工具函数和输入参数 schema。传输层：把 Streamable HTTP 请求映射到协议层。如果你自己从零写协议层，需要处理不少细节：会话状态、响应流式切换、协议版本协商、错误码统一……我试过一次，写了三百多行才勉强能用，后来还是切回了官方 SDK。我的建议是：原理要懂，但工程实现优先用现成库。下面我用 Python 的 MCP SDK 演示一个完整可跑的服务端。3.2 用 FastMCP 搭建一个可用的 MCP 服务器FastMCP 是 MCP 官方 SDK 提供的高层封装，支持流式 HTTP 传输。安装依赖：pip install mcp fastmcp uvicorn然后写一个最简单的服务端：from mcp.server.fastmcp import FastMCP mcp = FastMCP("demo-tools") @mcp.tool() def now_time(timezone: str = "Asia/Shanghai") -> str: """Return current time for the given timezone.""" from datetime import datetime from zoneinfo import ZoneInfo return datetime.now(ZoneInfo(timezone)).isoformat() if __name__ == "__main__": mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)启动后，服务会监听 8000 端口，默认的 MCP 端点路径是/mcp。你可以用curl快速验证：curl -X POST http://127.0.0.1:8000/mcp \ -H "Content-Type: application/json" \ -H "Accept: application/json, text/event-stream" \ -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"curl-test","version":"1.0.0"}}}'如果一切正常，响应里会带有Mcp-Session-Id头，body 为服务器信息和协议版本。这就意味着端点已经就绪，可以被任意 MCP 客户端发现并调用了。3.3 关于响应格式：JSON 还是 SSE，由谁决定你可能会问：服务端返回 JSON 还是 SSE，是我自己控制的吗？答案是：由服务端实现决定，客户端需要同时兼容两种格式。对于简单工具调用，服务端完全可以只返回 JSON。但对于耗时较长的工具，比如一个需要执行 30 秒的批处理任务，客户端希望看到分阶段进度，这时候服务端把响应切换成 SSE 流，就能借助notifications/progress事件持续推送进度。FastMCP 会根据工具执行方式和端点配置自动选择合适策略，但你自己实现时要注意：客户端必须根据Content-Type来分流解析逻辑，而不是写死某一种格式。4. 客户端的完整调用链路：初始化、会话、工具发现、工具调用4.1 第一步：initialize 握手与协议版本协商无论你用哪个 MCP 客户端，调用的第一步永远是initialize。这一步有两个目的：协商双方都支持的协议版本；交换客户端/服务端能力信息。如果服务端认为你请求的协议版本不兼容，会返回Unsupported protocol version错误。所以你在写客户端时，要选择官方当时推荐的协议版本。例如我写这篇教程时常用的版本是2025-06-18。此外，initialize响应头里如果带了Mcp-Session-Id，你就必须把它保存下来，后续所有请求都要带上这个 ID。4.2 第二步：发送 initialized 通知握手完成后，客户端必须发送一次notifications/initialized通知。这个名字里的 "notification" 是 JSON-RPC 术语，表示不需要服务端返回响应的消息。它存在的意义是告诉服务端“我已经准备好接受你后续推送的通知，可以开始正常工作了”。如果漏发这个通知，很多服务端会拒绝后续的tools/list请求，或者表现为“客户端一直等待服务端推送，但服务端没有动作”。这是初学者最容易踩的坑之一，我和同事调试时甚至因此以为是网络问题，查了半天才发现是少发了一个通知。4.3 第三步与第四步：tools/list 发现工具，tools/call 调用工具接下来的流程就非常直观了。客户端先发tools/list，拿到服务端所有工具的名称、描述、输入参数 schema。再根据模型的选择，决定调用哪个工具，发tools/call请求。这里要注意的是，MCP 里的工具调用本身不包含“AI 帮你决定调用哪个”的逻辑。MCP 只是提供了标准的工具发现和调用接口；至于模型是否调用、调用哪个，是由客户端把工具列表喂给模型后由模型决策的。所以你完全可以不引入大模型，纯靠代码逻辑调 MCP 工具——这对接自动化脚本很有用。下面是一个用 Pythonrequests库手写的最小客户端示例，没有依赖 MCP SDK，方便看协议：import requests BASE = "http://127.0.0.1:8000/mcp" HEADERS = {"Content-Type": "application/json", "Accept": "application/json, text/event-stream"} s = requests.Session() init_resp = s.post(BASE, json={ "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "simple-client", "version": "0.1.0"} } }, headers=HEADERS) session_id = init_resp.headers.get("Mcp-Session-Id") HEADERS["Mcp-Session-Id"] = session_id s.post(BASE, json={ "jsonrpc": "2.0", "method": "notifications/initialized" }, headers=HEADERS) tools_resp = s.post(BASE, json={ "jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {} }, headers=HEADERS) print("工具列表:", tools_resp.json()) call_resp = s.post(BASE, json={ "jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": { "name": "now_time", "arguments": {"timezone": "Asia/Tokyo"} } }, headers=HEADERS) print("调用结果:", call_resp.json())响应体里的content数组是工具结果的列表，isError字段表示这次调用是否出错。这些字段都是标准化的，客户端可以在拿到结果后交给模型解析。4.4 流式响应时的解析差异如果服务端在tools/call这一步返回的是text/event-stream，上面的resp.json()就会报错。正确做法是先看content-type，如果是 SSE，就按行读取事件流。SSE 格式本身很简单：每个事件以event:开头定义类型，紧接着是data:字符串，中间空行分隔。if "text/event-stream" in call_resp.headers.get("Content-Type", ""): for line in call_resp.iter_lines(): if line.startswith(b"data:"): data = line[5:].strip() # 解析 data 字段中的 JSON print("收到流式数据:", data) else: print(call_resp.json())如果你使用官方 MCP SDK 的客户端，这些细节都被封装好了，你只需要处理最高层的result对象。但我仍然建议你手写一次客户端，因为只有理解了Content-Type分流、SSE 行结构、session 头这些细节，碰到线上问题才不会一头雾水。5. 鉴权模式与多会话管理：线上部署绕不开的事5.1 无鉴权模式只适合本地调试开发初期我直接裸奔跑服务，没有任何 token 校验，用起来确实爽。但一旦把 MCP 端点暴露到公网，基本等于让你的服务器为全世界开放工具调用能力。所有能访问到这个地址的客户端，都能列举你的工具、触发你的工具逻辑。如果你在工具里写了“读取本机文件”“执行 shell”这类能力，后果非常严重。所以我的经验是：无鉴权模式只留在localhost阶段；一旦绑定非回环地址，立刻加鉴权。5.2 Bearer Token 接入方式MCP 的鉴权体系完整来说支持 OAuth 2.1，但对内部工具、私有部署来说，第一步建议先用最简单的 Bearer Token。客户端侧只需要在请求头里带上：Authorization: Bearer your_token_here服务端侧则需要在校验每个请求前检查这个头。FastMCP 里可以通过auth参数注入校验函数，或者更简单的做法是在你的反向代理层（比如 Nginx）统一校验。我实际用的是后者，原因是不想在自己的工具代码里侵入鉴权逻辑。代理层拦一道，既保护了 MCP 端点，也能顺便做限流、日志、白名单。5.3 Session ID 的作用域与生命周期Mcp-Session-Id是 MCP 会话管理的核心，但它不承担身份认证职责。它的作用是让服务端在一个会话期内维护上下文状态，比如“这个客户端已经初始化过了”“它的通信能力是什么”。携带错误的 session ID 会收到Invalid session或类似错误。关于生命周期，有几个要点：服务端返回Mcp-Session-Id后，后续所有请求都必须带上；不同客户端拿到的 session ID 应该互相隔离，不能共用一个全局变量；服务端可以主动让 session 过期，过期后客户端需要重新initialize；多实例部署时，session 尽量存储在共享存储（如 Redis）里，否则负载均衡后请求落到另一个实例就会找不到会话。我踩过最蠢的坑是客户端用了一个全局变量存 session ID，本地单用户没问题，一压测多个并发线程，全乱套了。正确做法是把 session ID 放在每个客户端实例里，跟具体的客户端请求绑定。5.4 再进一步：OAuth 2.1 与权限粒度如果工具要给外部用户使用，建议走 OAuth 2.1 流程。MCP 协议文档里把授权流程设计得比较完整：客户端先获取授权服务器信息，完成用户授权后拿到访问令牌，再带着令牌访问 MCP 端点。这套流程可以做到“哪个用户在用哪个工具，全部可审计”。但它的实现成本比 Bearer Token 高不少，不适合刚起步的项目。我的建议是：先无鉴权/简单 Token 跑通业务，等有真实的多用户需求再升级到 OAuth。6. 实测最常踩的五个坑与排查思路6.1 一个典型故障现象的排查链路先说一个我印象很深的故障。某个下午，我部署在云函数上的 MCP 服务突然“无法调用”，客户端一直报Expecting value: line 1 column 1 (char 0)。从报错看像是 JSON 解析失败，但我本地明明测得好好的。我的排查链路是这样的：先抓包看响应原文，发现返回的是text/event-stream，而不是application/json。本地之所以正常，是因为本地服务工具执行很快，服务端直接返回了 JSON；线上那个工具因为首次冷启动很慢，服务端强制切到了 SSE 模式分批返回。问题根因是客户端把响应格式写死了，没有按Content-Type分流。修复方式就是前面代码里提到的那个if判断。这个案例也说明一个共性原则：先看原始响应，再猜代码逻辑，不要一开始就怀疑底层协议。6.2 坑一：SSE 响应解析失败现象：resp.json()抛JSONDecodeError。原因：服务端返回了 SSE 格式，客户端按 JSON 解析。常见场景包括工具执行时间较长、服务端配置了强制流式响应。排查方式：打印resp.headers.get("Content-Type")，如果是text/event-stream，尝试逐行读取；如果内容被压缩或转码，先关闭代理层的内容编码再抓包。6.3 坑二：Mcp-Session-Id 丢失或串号现象：能initialize，但后续tools/list被拒，报Invalid session或请求直接挂起。原因：客户端没有保存Mcp-Session-Id，或者多个客户端共用了一个 session ID。排查方式：检查initialize的响应头是否含Mcp-Session-Id；检查后续请求头里是否带了这个值；再做一次并发测试，确认不同实例未共用 session。6.4 坑三：Content-Type 被代理层改写我在 Nginx 后面部署时碰到过：上游返回text/event-stream，但经过 Nginx 后响应头变成了text/plain。客户端按Content-Type判断时走了普通 JSON 分支，结果解析失败。处理方式：在代理配置里显式保留或改写 Content-Type，或者要求客户端同时检查响应体前缀而不是只依赖头部。两种方式我都用过，但更推荐前者，因为网络链路越透明，问题越少。6.5 坑四：协议版本不匹配现象：initialize阶段服务端返回Unsupported protocol version。原因：你的客户端写死了旧协议版本，而服务端升级后只支持新版本；或反过来。排查方式：对比客户端和服务端 SDK 版本；升级到一致的版本后，重新确认protocolVersion字段。这里没有捷径，版本号必须显式对齐。6.6 坑五：工具执行时间太长同步请求里跑一个 60 秒的工具，客户端和服务端都很容易遇到 HTTP 超时。正确做法是让工具在执行时先快速返回一个“任务已开始”的状态，再通过 SSE 或消息通知推送最终结果；或者干脆拆分工具，让批处理任务后台执行，只暴露状态查询工具。Streamable HTTP 对这种情况的关键价值就在这：它允许服务端在同一请求内用 SSE 持续输出进度，客户端不需要额外建立连接。如果你遇到的工具耗时普遍超过 30 秒，强烈建议把这种“开工+进度+结果”的交互模式纳入设计。7. 如果让我从头再搭一次如果现在让我重新设计一个“AI 调用 MCP”的项目，我会这样执行：先用 stdio 模式在本地把工具逻辑调试到完美，确保工具本身没有边界问题；再用 FastMCP 切换成 Streamable HTTP，在局域网里验证远程调用；紧接着加一层反向代理做 Bearer Token 校验和限流；最后才考虑 OAuth 和跨域等企业级需求。协议细节会在运行时慢慢暴露，不要试图在第一天就吃透所有边缘情况。我自己的体会是，MCP 的最大价值不是“协议多高效”，而是“把工具接入 AI 这件事从手工作坊变成了标准流水线”。写完这篇文章时，我又去翻了一遍自己的旧项目：以前那个总是拼错参数的时间查询接口，现在只保留了 MCP 版本，因为它对模型来说足够清晰，也足够容易维护。你的工具大概率也一样。
作为一个常年写小工具、偶尔折腾智能体的独立开发者，我一度很烦“把工具交给 AI 调用”这件事。明明接口文档写清楚了，模型还是会拼错参数；明明工具就挂在同一个局域网里，它却找不到服务；好不容易调通了，换一个客户端又得重写一套协议。后来我把工具按 MCP（Model Context Protocol）的标准接入，AI 再通过 Streamable HTTP 方式来调用，整个链路才算真正稳定下来。这篇教程会从原理到实现，完整走一遍 AI 调用 MCP 的过程，重点讲基于 Streamable HTTP 的传输方式。如果你正在做智能体工具接入、想把本地脚本暴露给 AI 使用、或者只是想知道 MCP 和普通 HTTP API 到底有什么区别，这篇文章应该能帮你省不少时间。1. 模型为什么需要 MCP：从“猜 REST 接口”到“标准工具说明书”1.1 模型调用普通接口时到底在“猜”什么先聊一个容易被忽略的问题：大模型本身并不知道你的接口长什么样。你给它一个 REST API 的地址和一段自然语言描述，它每一步都在做概率预测——URL 拼对了没有、字段名写对了没有、参数类型对不对，全靠“猜”。我最早做过一个时间查询工具，写成了GET /api/time?tz=Asia/Shanghai，结果模型经常把tz解释成timezone，或者把参数直接塞进路径里。不是模型笨，是我没有给它一个足够结构化的“工具说明书”。MCP 解决的就是这个“说明书”问题。它把工具、数据资源、提示模板统一成标准化的 JSON-RPC 协议，模型侧只需要理解一套协议，就能发现并调用任意 MCP 服务器提供的功能。你可以把 MCP 理解成 USB-C 接口：以前每个外设都有自己的充电线，现在所有设备用同一个接口标准，插上就能用。1.2 MCP 协议里的四个核心概念MCP 协议不只是“工具调用”，它包含四类能力，理解它们有助于你设计好自己的 MCP 服务器。Tools（工具）：模型主动发起调用的函数。类似于 function calling 里的 function，但有了统一的 schema 描述和标准化返回格式。Resources（资源）：只读数据，类似文件或数据库查询结果。模型端可以通过resources/read读取，适合暴露配置、文档、日志。Prompts（提示模板）：服务端预设的提示词模板，客户端可以拿到后组装进对话上下文。Sampling（采样）：服务器反向请求模型生成内容。这个能力更多用于“工具内部想再用一次大模型”的场景。实际使用中最常打交道的还是 Tools。调用链路的逻辑也很清楚：先获取工具的 schema（元数据），再按 schema 填入参数并调用，最后解析结构化结果。这个流程和 OpenAI 的 function calling 在理念上是一致的，但 MCP 强调的是协议层面的标准化和客户端-服务端解耦。1.3 为什么教程选择“Streamable HTTP”而不是别的MCP 协议支持几种传输方式，本文只讲 Streamable HTTP，因为它在远程调用、浏览器端、无服务器部署这些场景里最省事。后面会专门讲它和 stdio、旧版 HTTP+SSE 的取舍。现在你只需要知道：一旦你部署好了 MCP 服务端，AI 客户端能通过一个普通 HTTPS 地址来调用它，不需要共用进程，也不需要维护长连接，非常接近“给 AI 开一个 API”的直觉体验。2. Streamable HTTP 传输方式：为什么我放弃 stdio 和旧版 SSE2.1 三种传输方式的直观对比我先放一张对比表，然后再展开讲各自的使用场景。特性stdio旧版 HTTP+SSEStreamable HTTP（本教程）通信方式子进程标准输入输出先建立 SSE，再用独立端点发消息普通 POST 请求统一发消息服务端位置必须与客户端同机可在远程，但需可见的 SSE 端点远程/本地均可是否需要常驻连接需要需要不需要浏览器/网页端支持不支持较差支持无服务器（Serverless）友好程度低低高协议复杂度低中中我最早用的是 stdio，本地调试非常顺手，因为它不需要网络配置，起一个子进程就能把工具注册给客户端。但一旦工具需要部署到云服务器、或要让多个客户端并发访问，stdio 就完全不够用。进程是跟着客户端走的，客户端退出服务就没了，根本没有“常驻服务”的概念。2.2 旧版 HTTP+SSE 的两个核心痛点随后我尝试了旧版 HTTP+SSE，也就是“先 GET 建 SSE 流、再 POST 发消息”的方式。它解决了远程部署问题，但引入两个新麻烦。第一，客户端必须先向服务器发起 SSE GET 请求，拿到一个 listener 端点，才能开始通信。这意味着你的客户端必须维护一个长期挂起的连接，而且这个连接在网关、代理、无服务器环境中很容易超时。我测试时连着几次遇到“连接建立成功，但消息发送时 listener 已断开”的问题，排查起来极其恼火。第二，它把“建立连接”和“发送消息”拆成了两个独立步骤，这让很多 HTTP 客户端框架难以处理。你不仅要管理一个永久的 HTTP 响应流，还要额外维护一个发送用的 POST 端点，状态管理非常繁琐。2.3 Streamable HTTP 的答案：一个请求里完成通信Streamable HTTP 从根本上调整了交互模式：所有 JSON-RPC 消息都由客户端通过 POST 发送，服务器在同一个 HTTP 响应里返回结果。服务器如果希望流式推送进度，可以在响应头里把Content-Type设为text/event-stream，通过 SSE 逐段返回；如果只是普通工具调用，直接返回application/json也不是问题。这就把 MCP 客户端从“必须维护长连接”的泥潭里解放了出来。你可以把一次工具调用看成一次普通 API 请求，用任何 HTTP 客户端库都能实现。服务端状态通过Mcp-Session-Id响应头来标识，客户端带着它做后续请求，相当于把会话状态外置到了客户端。这个设计非常适合云函数、Web 应用、甚至运行在浏览器沙箱里的客户端。3. 服务器端搭建：30 分钟内吐出一个能被 AI 调用的端点3.1 先规划路由与协议入口服务端本质上就是一个 HTTP 服务，核心工作就是把收到的 JSON-RPC 请求分发到对应的处理器上。以我常用的目录结构为例，整个 MCP 服务可以拆成三层：协议层：负责解析 JSON-RPC 请求、包装响应、处理会话 ID。工具层：注册具体的工具函数和输入参数 schema。传输层：把 Streamable HTTP 请求映射到协议层。如果你自己从零写协议层，需要处理不少细节：会话状态、响应流式切换、协议版本协商、错误码统一……我试过一次，写了三百多行才勉强能用，后来还是切回了官方 SDK。我的建议是：原理要懂，但工程实现优先用现成库。下面我用 Python 的 MCP SDK 演示一个完整可跑的服务端。3.2 用 FastMCP 搭建一个可用的 MCP 服务器FastMCP 是 MCP 官方 SDK 提供的高层封装，支持流式 HTTP 传输。安装依赖：pip install mcp fastmcp uvicorn然后写一个最简单的服务端：from mcp.server.fastmcp import FastMCP mcp = FastMCP("demo-tools") @mcp.tool() def now_time(timezone: str = "Asia/Shanghai") -> str: """Return current time for the given timezone.""" from datetime import datetime from zoneinfo import ZoneInfo return datetime.now(ZoneInfo(timezone)).isoformat() if __name__ == "__main__": mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)启动后，服务会监听 8000 端口，默认的 MCP 端点路径是/mcp。你可以用curl快速验证：curl -X POST http://127.0.0.1:8000/mcp \ -H "Content-Type: application/json" \ -H "Accept: application/json, text/event-stream" \ -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"curl-test","version":"1.0.0"}}}'如果一切正常，响应里会带有Mcp-Session-Id头，body 为服务器信息和协议版本。这就意味着端点已经就绪，可以被任意 MCP 客户端发现并调用了。3.3 关于响应格式：JSON 还是 SSE，由谁决定你可能会问：服务端返回 JSON 还是 SSE，是我自己控制的吗？答案是：由服务端实现决定，客户端需要同时兼容两种格式。对于简单工具调用，服务端完全可以只返回 JSON。但对于耗时较长的工具，比如一个需要执行 30 秒的批处理任务，客户端希望看到分阶段进度，这时候服务端把响应切换成 SSE 流，就能借助notifications/progress事件持续推送进度。FastMCP 会根据工具执行方式和端点配置自动选择合适策略，但你自己实现时要注意：客户端必须根据Content-Type来分流解析逻辑，而不是写死某一种格式。4. 客户端的完整调用链路：初始化、会话、工具发现、工具调用4.1 第一步：initialize 握手与协议版本协商无论你用哪个 MCP 客户端，调用的第一步永远是initialize。这一步有两个目的：协商双方都支持的协议版本；交换客户端/服务端能力信息。如果服务端认为你请求的协议版本不兼容，会返回Unsupported protocol version错误。所以你在写客户端时，要选择官方当时推荐的协议版本。例如我写这篇教程时常用的版本是2025-06-18。此外，initialize响应头里如果带了Mcp-Session-Id，你就必须把它保存下来，后续所有请求都要带上这个 ID。4.2 第二步：发送 initialized 通知握手完成后，客户端必须发送一次notifications/initialized通知。这个名字里的 "notification" 是 JSON-RPC 术语，表示不需要服务端返回响应的消息。它存在的意义是告诉服务端“我已经准备好接受你后续推送的通知，可以开始正常工作了”。如果漏发这个通知，很多服务端会拒绝后续的tools/list请求，或者表现为“客户端一直等待服务端推送，但服务端没有动作”。这是初学者最容易踩的坑之一，我和同事调试时甚至因此以为是网络问题，查了半天才发现是少发了一个通知。4.3 第三步与第四步：tools/list 发现工具，tools/call 调用工具接下来的流程就非常直观了。客户端先发tools/list，拿到服务端所有工具的名称、描述、输入参数 schema。再根据模型的选择，决定调用哪个工具，发tools/call请求。这里要注意的是，MCP 里的工具调用本身不包含“AI 帮你决定调用哪个”的逻辑。MCP 只是提供了标准的工具发现和调用接口；至于模型是否调用、调用哪个，是由客户端把工具列表喂给模型后由模型决策的。所以你完全可以不引入大模型，纯靠代码逻辑调 MCP 工具——这对接自动化脚本很有用。下面是一个用 Pythonrequests库手写的最小客户端示例，没有依赖 MCP SDK，方便看协议：import requests BASE = "http://127.0.0.1:8000/mcp" HEADERS = {"Content-Type": "application/json", "Accept": "application/json, text/event-stream"} s = requests.Session() init_resp = s.post(BASE, json={ "jsonrpc": "2.0", "id": 1, "method": "initialize", "params": { "protocolVersion": "2025-06-18", "capabilities": {}, "clientInfo": {"name": "simple-client", "version": "0.1.0"} } }, headers=HEADERS) session_id = init_resp.headers.get("Mcp-Session-Id") HEADERS["Mcp-Session-Id"] = session_id s.post(BASE, json={ "jsonrpc": "2.0", "method": "notifications/initialized" }, headers=HEADERS) tools_resp = s.post(BASE, json={ "jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {} }, headers=HEADERS) print("工具列表:", tools_resp.json()) call_resp = s.post(BASE, json={ "jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": { "name": "now_time", "arguments": {"timezone": "Asia/Tokyo"} } }, headers=HEADERS) print("调用结果:", call_resp.json())响应体里的content数组是工具结果的列表，isError字段表示这次调用是否出错。这些字段都是标准化的，客户端可以在拿到结果后交给模型解析。4.4 流式响应时的解析差异如果服务端在tools/call这一步返回的是text/event-stream，上面的resp.json()就会报错。正确做法是先看content-type，如果是 SSE，就按行读取事件流。SSE 格式本身很简单：每个事件以event:开头定义类型，紧接着是data:字符串，中间空行分隔。if "text/event-stream" in call_resp.headers.get("Content-Type", ""): for line in call_resp.iter_lines(): if line.startswith(b"data:"): data = line[5:].strip() # 解析 data 字段中的 JSON print("收到流式数据:", data) else: print(call_resp.json())如果你使用官方 MCP SDK 的客户端，这些细节都被封装好了，你只需要处理最高层的result对象。但我仍然建议你手写一次客户端，因为只有理解了Content-Type分流、SSE 行结构、session 头这些细节，碰到线上问题才不会一头雾水。5. 鉴权模式与多会话管理：线上部署绕不开的事5.1 无鉴权模式只适合本地调试开发初期我直接裸奔跑服务，没有任何 token 校验，用起来确实爽。但一旦把 MCP 端点暴露到公网，基本等于让你的服务器为全世界开放工具调用能力。所有能访问到这个地址的客户端，都能列举你的工具、触发你的工具逻辑。如果你在工具里写了“读取本机文件”“执行 shell”这类能力，后果非常严重。所以我的经验是：无鉴权模式只留在localhost阶段；一旦绑定非回环地址，立刻加鉴权。5.2 Bearer Token 接入方式MCP 的鉴权体系完整来说支持 OAuth 2.1，但对内部工具、私有部署来说，第一步建议先用最简单的 Bearer Token。客户端侧只需要在请求头里带上：Authorization: Bearer your_token_here服务端侧则需要在校验每个请求前检查这个头。FastMCP 里可以通过auth参数注入校验函数，或者更简单的做法是在你的反向代理层（比如 Nginx）统一校验。我实际用的是后者，原因是不想在自己的工具代码里侵入鉴权逻辑。代理层拦一道，既保护了 MCP 端点，也能顺便做限流、日志、白名单。5.3 Session ID 的作用域与生命周期Mcp-Session-Id是 MCP 会话管理的核心，但它不承担身份认证职责。它的作用是让服务端在一个会话期内维护上下文状态，比如“这个客户端已经初始化过了”“它的通信能力是什么”。携带错误的 session ID 会收到Invalid session或类似错误。关于生命周期，有几个要点：服务端返回Mcp-Session-Id后，后续所有请求都必须带上；不同客户端拿到的 session ID 应该互相隔离，不能共用一个全局变量；服务端可以主动让 session 过期，过期后客户端需要重新initialize；多实例部署时，session 尽量存储在共享存储（如 Redis）里，否则负载均衡后请求落到另一个实例就会找不到会话。我踩过最蠢的坑是客户端用了一个全局变量存 session ID，本地单用户没问题，一压测多个并发线程，全乱套了。正确做法是把 session ID 放在每个客户端实例里，跟具体的客户端请求绑定。5.4 再进一步：OAuth 2.1 与权限粒度如果工具要给外部用户使用，建议走 OAuth 2.1 流程。MCP 协议文档里把授权流程设计得比较完整：客户端先获取授权服务器信息，完成用户授权后拿到访问令牌，再带着令牌访问 MCP 端点。这套流程可以做到“哪个用户在用哪个工具，全部可审计”。但它的实现成本比 Bearer Token 高不少，不适合刚起步的项目。我的建议是：先无鉴权/简单 Token 跑通业务，等有真实的多用户需求再升级到 OAuth。6. 实测最常踩的五个坑与排查思路6.1 一个典型故障现象的排查链路先说一个我印象很深的故障。某个下午，我部署在云函数上的 MCP 服务突然“无法调用”，客户端一直报Expecting value: line 1 column 1 (char 0)。从报错看像是 JSON 解析失败，但我本地明明测得好好的。我的排查链路是这样的：先抓包看响应原文，发现返回的是text/event-stream，而不是application/json。本地之所以正常，是因为本地服务工具执行很快，服务端直接返回了 JSON；线上那个工具因为首次冷启动很慢，服务端强制切到了 SSE 模式分批返回。问题根因是客户端把响应格式写死了，没有按Content-Type分流。修复方式就是前面代码里提到的那个if判断。这个案例也说明一个共性原则：先看原始响应，再猜代码逻辑，不要一开始就怀疑底层协议。6.2 坑一：SSE 响应解析失败现象：resp.json()抛JSONDecodeError。原因：服务端返回了 SSE 格式，客户端按 JSON 解析。常见场景包括工具执行时间较长、服务端配置了强制流式响应。排查方式：打印resp.headers.get("Content-Type")，如果是text/event-stream，尝试逐行读取；如果内容被压缩或转码，先关闭代理层的内容编码再抓包。6.3 坑二：Mcp-Session-Id 丢失或串号现象：能initialize，但后续tools/list被拒，报Invalid session或请求直接挂起。原因：客户端没有保存Mcp-Session-Id，或者多个客户端共用了一个 session ID。排查方式：检查initialize的响应头是否含Mcp-Session-Id；检查后续请求头里是否带了这个值；再做一次并发测试，确认不同实例未共用 session。6.4 坑三：Content-Type 被代理层改写我在 Nginx 后面部署时碰到过：上游返回text/event-stream，但经过 Nginx 后响应头变成了text/plain。客户端按Content-Type判断时走了普通 JSON 分支，结果解析失败。处理方式：在代理配置里显式保留或改写 Content-Type，或者要求客户端同时检查响应体前缀而不是只依赖头部。两种方式我都用过，但更推荐前者，因为网络链路越透明，问题越少。6.5 坑四：协议版本不匹配现象：initialize阶段服务端返回Unsupported protocol version。原因：你的客户端写死了旧协议版本，而服务端升级后只支持新版本；或反过来。排查方式：对比客户端和服务端 SDK 版本；升级到一致的版本后，重新确认protocolVersion字段。这里没有捷径，版本号必须显式对齐。6.6 坑五：工具执行时间太长同步请求里跑一个 60 秒的工具，客户端和服务端都很容易遇到 HTTP 超时。正确做法是让工具在执行时先快速返回一个“任务已开始”的状态，再通过 SSE 或消息通知推送最终结果；或者干脆拆分工具，让批处理任务后台执行，只暴露状态查询工具。Streamable HTTP 对这种情况的关键价值就在这：它允许服务端在同一请求内用 SSE 持续输出进度，客户端不需要额外建立连接。如果你遇到的工具耗时普遍超过 30 秒，强烈建议把这种“开工+进度+结果”的交互模式纳入设计。7. 如果让我从头再搭一次如果现在让我重新设计一个“AI 调用 MCP”的项目，我会这样执行：先用 stdio 模式在本地把工具逻辑调试到完美，确保工具本身没有边界问题；再用 FastMCP 切换成 Streamable HTTP，在局域网里验证远程调用；紧接着加一层反向代理做 Bearer Token 校验和限流；最后才考虑 OAuth 和跨域等企业级需求。协议细节会在运行时慢慢暴露，不要试图在第一天就吃透所有边缘情况。我自己的体会是，MCP 的最大价值不是“协议多高效”，而是“把工具接入 AI 这件事从手工作坊变成了标准流水线”。写完这篇文章时，我又去翻了一遍自己的旧项目：以前那个总是拼错参数的时间查询接口，现在只保留了 MCP 版本，因为它对模型来说足够清晰，也足够容易维护。你的工具大概率也一样。