从零手写MCP协议服务：实现AI工具即插即用

我大概是从去年底开始把AI应用从“纯聊天的Demo”往“能干活的工作流”上改的。改到一半发现真正的麻烦都在接口上对话模型想查一下订单状态、读取日历安排、调一次内部API要么把函数签名写死在系统提示词里要么靠一连串流程编排代码硬拼。直到在某开源社区看到10.2MCP协议实战的讨论帖我才意识到这类问题本可以抽成一套标准协议来做。MCP是Model Context Protocol的缩写核心就一句话让模型、客户端、服务端三方用同一套JSON-RPC信封通信外部工具像USB外设一样随插随用。下文不是概念搬运而是把一个最小可用的MCP服务从零跑通然后把那些文档不会写的坑逐个点名。对于正在搭私有知识库、内部运维机器人或者想把大模型接到自己业务系统的人这篇笔记的参考价值应该很直接。接下来我会先用一个没有MCP的旧场景解释它到底解决了什么再按“环境准备—协议拆解—完整实战—踩坑复盘”的顺序展开。动手部分以最小依赖优先尽量让每个人都能在半小时内复现。1. 在没有MCP的时代工具接入为什么这么费劲1.1 从“把答案喂给模型”到“让模型自己去取”我先讲一个没有MCP时很典型的场景。假设业务上需要让对话机器人回复“订单A10086目前是已发货还是已签收”。很多团队一开始的做法是写一个“查询订单”的Python函数挂在某个HTTP服务上然后从模型侧分两步走先用提示词让模型把用户问题里的订单号提取出来再让外部编排代码去调那个HTTP接口把返回的文本拼接进下一轮提问的上下文里。这个流程前几次看着挺好用但用上两周就会开始难受。订单信息字段稍微多一点比如加个“预计到达时间”所有相关的提示词模板都要跟着改用户在多轮对话里提到了好几个订单号拼接上下文时很快就会被塞满更麻烦的是每次底层数据接口返回格式有变化模型回复的“话术”就得重新迭代因为模型看到的永远是“二次加工后的文本”而不是结构化的数据。有人会说可以用函数调用。确实成熟模型基本都支持函数调用但这会引入一个新的耦合点每个模型对工具参数格式、返回字段的约定都略有不同而且工具列表通常也要写死在客户端的配置里。换一个模型厂商可能连工具调用请求的长相都要改一遍。MCP要解决的正是这个“最后一公里”的标准化问题。它把这个过程重构成三件事模型不直接看到工具实现而是看到一份统一的工具清单客户端知道怎么按协议把模型意图转换成工具请求服务端只需要实现固定的几个方法。任何一端的变化都不会要求另外两端整体重写。1.2 MCP到底是哪一层可以用一个生活里的类比来理解MCP的位置。过去我们接各种外设鼠标、键盘、U盘、显示器每个设备厂商都用自己的接头结果桌面下面一堆转换线。后来通用串行总线接口把“连接规范”统一了设备里的芯片、驱动、线材才彻底解耦。MCP在AI工具接入里扮演的正是这条“总线协议”的角色。它定义了宿主、客户端、服务端三个角色。宿主是运行模型和UI的应用程序比如桌面AI助手或IDE插件客户端在宿主导体内负责按照MCP协议和服务端对话服务端是具体工具能力的提供方可能封装了数据库、文件系统、内部API或搜索服务。模型本身并不直接调用服务端它只是给出意图真正的网络请求由客户端代发服务端返回结果后再由客户端交回模型做下一轮理解。这个边界很重要也是实战中最容易忽略的MCP不是让模型绕过所有约束直接执行命令它只是让工具暴露的过程标准化。这三个角色的解耦带来了三个很实际的好处。第一是动态发现宿主连接一个MCP服务端后先通过协议获取工具清单不需要在配置里预写全部工具。第二是统一传输无论服务端后端是Python、Java还是Node只要遵守同一套信封客户端都不用改逻辑。第三是安全边界工具能力可以被服务端按权限裁剪宿主可以决定模型在什么条件下调用工具这比把凭证都塞进提示词里可靠得多。2. 实战前的准备先选一条从开发到验证都不卡的路2.1 传输方式取舍MCP协议目前最常见的传输有两种stdio和Streamable HTTP。stdio方式适合本地插拔宿主进程直接启动服务端子进程通过标准输入输出传消息延迟低也不需要开端口但它不够直观一个问题抛出来你很难单独用命令行去测。Streamable HTTP方式是服务端提供一个HTTP端点客户端通过POST发JSON-RPC消息开发时可以拿curl一条条手动请求排障体验好很多。我把“直接能用curl验证”放在选型的首位倒不是因为它比stdio更高级而是对于第一次上手的人来说能看到“请求—响应”的完整往返比什么都重要。所以下面的实战演示用HTTP传输依赖也控制在最少范围。等把协议流程跑通了再回切stdio就是一两行配置的事不会有什么心智负担。2.2 目录与依赖推荐创建一个干净的项目目录把服务端和客户端分开省得后面调试时互相污染mcp-demo/ ├── server/ │ └── app.py └── client/ └── probe.py服务端我用Python的FastAPI搭主要原因不是性能而是代码量最小路由和异步都处理好了。客户端只需要一个requests库。如果机器上没有装过直接执行安装pip install fastapi uvicorn requests这里不推荐一上来就用封装很厚的SDK。倒不是说SDK不好而是它会把协议细节包起来遇到问题你往往不知道是协议没写对还是SDK版本不匹配。先用裸HTTP把协议跑明白后面再用SDK时心里是有一张底图的。2.3 骨架服务先跑起来先在server目录下建一个最小入口只做一件事把收到的POST请求原样打出来再回一个空的JSON-RPC响应。这个骨架不是为了实现协议而是为了确认本地网络和请求格式都通。from fastapi import FastAPI, Request, Response import json app FastAPI() app.post(/mcp) async def mcp_endpoint(request: Request): payload await request.json() print(RECV:, json.dumps(payload, ensure_asciiFalse, indent2)) return Response( contentjson.dumps({jsonrpc: 2.0, id: payload.get(id), result: {}}), media_typeapplication/json, )然后启动它cd mcp-demo/server uvicorn app:app --host 127.0.0.1 --port 8000另开一个终端用curl随便发一条消息curl -s http://127.0.0.1:8000/mcp \ -H Content-Type: application/json \ -d {jsonrpc:2.0,id:1,method:ping,params:{}}能看到服务端控制台打印出完整JSON并发回带相同id的响应说明网络路径就位了。这里有个小细节值得留意MCP协议里绝大多数请求都要带同一个id往返响应里的id和请求的id必须严格一致。客户端如果同时发多个请求就靠这个id把结果对回目标。3. 协议核心一条消息怎么流转3.1 建立会话initialize 到底谈了些什么MCP客户端的第一次请求几乎总是initialize它做的事情不是调工具而是握手协商。客户端会带上自己的协议版本号、客户端信息、希望启用的能力服务端则要返回它支持的协议版本号、服务端信息、服务端能力。两边都各退一步选出一个双方都支持的版本组合。一次最小握手请求大概长这样{ jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 2025-03-26, capabilities: {}, clientInfo: { name: mcp-probe, version: 0.1.0 } } }对应响应{ jsonrpc: 2.0, id: 1, result: { protocolVersion: 2025-03-26, capabilities: { tools: { listChanged: false } }, serverInfo: { name: mcp-demo-server, version: 0.1.0 } } }这里有个特别容易翻车的点initialize完成后客户端必须再发一条notifications/initialized通知服务端才会认为会话真正就绪。这个通知是“通知”不是“请求”所以它不带id也不需要响应。很多新手手搓客户端时漏掉这一步结果后面发tools/list总被服务端拒绝日志里又不给明确原因。协议顺序很严格initialize → notifications/initialized → tools/list → tools/call。3.2 工具发现服务端把自己的能力摆出来握手结束后的第一个实质动作是动态发现工具对应method叫tools/list。这个设计背后的意图是客户端事先不需要知道服务端能干什么只需要在连接后问一次服务端返回的JSON里会包含一个工具数组。每个工具必须给出名称、描述和输入参数schema。输入schema遵循JSON Schema语法用来告诉模型“这个参数叫什么、是不是必填、大概长什么样”。一个最小又完整的tools/list响应是这个样子{ jsonrpc: 2.0, id: 2, result: { tools: [ { name: get_order_status, description: 根据订单号查询订单当前状态, inputSchema: { type: object, properties: { order_id: { type: string } }, required: [order_id] } } ] } }注意inputSchema里的required数组。如果服务端把这个字段写错比如漏了它模型在调用时可能完全不传order_id服务端拿到的参数就是空的。如果required写得太严比如除了order_id还要求一个只在特定场景才有的token字段那模型大概率会因为凑不齐参数而频繁放弃调用。3.3 工具调用模型表达意图客户端代发请求当模型根据用户话术判断出需要查订单时宿主并不会自己直接去调订单API而是由客户端构造一个tools/call请求发给服务端。服务端收到后执行具体逻辑把结果包装成MCP定义的Tool Result返回。这里有一个新手很迷惑的点返回结果不是普通的字符串而是结构化content数组里面每一项有type字段最常见的是text类型。一次典型的tools/call请求{ jsonrpc: 2.0, id: 3, method: tools/call, params: { name: get_order_status, arguments: { order_id: A10086 } } }服务端返回{ jsonrpc: 2.0, id: 3, result: { content: [ { type: text, text: {order_id: A10086, status: shipped, estimate: 2025-06-10} } ], isError: false } }content数组里可以有多项也可以用image类型返回图片资源。isError字段用来标记工具是否失败但即使业务上失败HTTP层通常也返回200只是把isError置为true。这一点和普通REST习惯很不一样后面排障时一定要记得先看isError别光盯着HTTP状态码。4. 一个能直接跑起来的实战用例4.1 用例目标让对话系统能查订单状态下面这个用例不接任何真实第三方数据都写在服务端内存里方便完整复现。需求可以描述为对话模型在收到“帮我查A10086这个订单到哪了”时能通过MCP工具返回真实的订单字段而不是模型自己编一个状态出来。这个目标分开来看关键是两件事一是工具清单能被客户端发现二是服务端收到调用后能根据参数返回对应结果。先把第一个目标用tools/list验证再把第二个目标用客户端脚本验证。如果这两步都通过接入任何支持MCP的宿主App都只是配置的问题。4.2 服务端实现手写一个最小MCP端点把server/app.py扩展成完整版本。这里只实现initialize、tools/list、tools/call三个方法加上一个模拟订单数据的查询函数。from fastapi import FastAPI, Request, Response import json app FastAPI() ORDER_DB { A10086: {status: shipped, estimate: 2025-06-10}, B20240: {status: delivered, estimate: 2025-06-01}, } TOOLS [ { name: get_order_status, description: 根据订单号查询订单当前状态, inputSchema: { type: object, properties: { order_id: {type: string} }, required: [order_id] } } ] def make_response(payload_id, result): return Response( contentjson.dumps({jsonrpc: 2.0, id: payload_id, result: result}), media_typeapplication/json, ) app.post(/mcp) async def mcp_endpoint(request: Request): payload await request.json() method payload.get(method) req_id payload.get(id) params payload.get(params) or {} if method initialize: return make_response(req_id, { protocolVersion: 2025-03-26, capabilities: {tools: {listChanged: False}}, serverInfo: {name: mcp-demo-server, version: 0.1.0} }) if method notifications/initialized: return Response(status_code202) if method tools/list: return make_response(req_id, {tools: TOOLS}) if method tools/call: name params.get(name) args params.get(arguments) or {} if name get_order_status: order_id args.get(order_id, ) info ORDER_DB.get(order_id) if info is None: return make_response(req_id, { content: [{type: text, text: 订单号不存在}], isError: True }) return make_response(req_id, { content: [{type: text, text: json.dumps(info, ensure_asciiFalse)}], isError: False }) return make_response(req_id, { content: [{type: text, text: funknown method: {method}}], isError: True })实现里故意把notifications/initialized返回202而不是200是为了顺手演示“通知不需要响应”这个约定。实际接入时有的客户端不检查响应内容但也有的客户端会解析这个值只要别让它等出超时就行。4.3 客户端实现模拟完整会话流程client/probe.py负责按正确顺序发起协议交互。它要做的事依次是发起initialize、发notifications/initialized通知、调用tools/list、拿到工具后调用tools/call把最终结果打印出来。import requests import json BASE http://127.0.0.1:8000/mcp HEADERS {Content-Type: application/json} def send(req): resp requests.post(BASE, headersHEADERS, jsonreq, timeout10) return resp.json() # step 1: initialize init_resp send({ jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 2025-03-26, capabilities: {}, clientInfo: {name: mcp-probe, version: 0.1.0} } }) print([initialize], json.dumps(init_resp, ensure_asciiFalse, indent2)) # step 2: initialized notification send({ jsonrpc: 2.0, method: notifications/initialized, params: {} }) # step 3: tools/list tool_list send({ jsonrpc: 2.0, id: 2, method: tools/list, params: {} }) print([tools/list], json.dumps(tool_list, ensure_asciiFalse, indent2)) # step 4: tools/call call_resp send({ jsonrpc: 2.0, id: 3, method: tools/call, params: { name: get_order_status, arguments: {order_id: A10086} } }) print([tools/call], json.dumps(call_resp, ensure_asciiFalse, indent2))启动服务端后运行cd mcp-demo/client python probe.py正常情况会看到initialize返回协议版本tools/list返回工具数组tools/call返回包含订单状态的结果。这一段跑通后你已经亲手实现了MCP协议的最小闭环而不是只在SDK示例里按了个按钮。4.4 接入真实宿主的观测方法如果你的目标是把服务端接到真正带界面的AI客户端里其实只剩一个配置问题把服务端地址填到客户端的MCP服务器配置里。不过这里我强烈建议先做一步“日志观测”。服务端里已经加了print每次收到请求都会打印完整JSON。当你从客户端界面发出“查一下A10086订单”时观察控制台你就能看到一条清晰的链路宿主把用户问题送给模型模型输出工具调用意图客户端发出tools/call服务端打印具体参数。很多问题正是在这个观测过程中暴露的。比如模型把order_id理解成了数字类型而你的inputSchema只接受string那么工具可能压根不会被调用又比如客户端连着发多个tools/call但服务端处理太快响应id错位宿主那边就可能显示“工具没有可用结果”。这些现象用日志一对照原因就非常清楚了。5. 实战踩坑记录排障思路与解决方案5.1 content返回格式不对模型拿不到关键信息我在第一次把MCP服务端接入业务系统时犯过一个非常基础的错tools/call返回结果时没有按content数组规范而是直接返回了一段自定义JSON比如{result: xxx}。客户端本身是宽容的也能把原始JSON交给模型但模型在理解“结果”时容易把字段名解释错。后来规范写法是任何工具结果都放在content里再用text承载序列化文本这样模型拿到的信息边界最清晰。这里有一个判断原则工具返回的是“人要看的文本”还是“模型要解析的数据”。如果不是给人直接阅读的建议在text里输出紧凑JSON同时通过description告诉模型这个字符串是结构化数据、可以直接提取字段。我在实测中发现这种写法的工具调用成功率远高于随意返回字符串。5.2 inputSchema写太严或太松都会出问题inputSchema是协议里最需要抠细节的部分。写太严比如必填字段过多模型凑不齐参数就会拒绝调用写太松比如没有任何必填约束模型可能自己脑补一个不存在的order_id传给服务端。我自己的经验是先按最小可用参数写能用一个字段解决的问题绝不放两个字段把description写完整比如“订单号格式为字母开头的字符串”模型对格式的理解会明显改善。还有一点规范上的提醒不要把大型枚举值直接塞进schema里。有的工具状态字段有几十种可能每招首token一次会占用大量上下文而且模型不一定能正确处理。简洁起见先让模型传原始key服务端做映射校验反馈也更可控。5.3 不能把MCP端点当成普通REST接口调日常开发时大家习惯性关心HTTP状态码但MCP的很多“错误”都发生在200响应内部。isError字段为true才是真正的业务级失败。另外一个常见误区是超时设置。MCP服务端有时候可能要执行数据库查询、调用上游API响应时间不是毫秒级的客户端超时设太短会频繁报错。建议初始值至少放到30秒再根据业务情况逐步缩短。如果是Streamable HTTP传输还要注意服务端可能需要支持流式返回。协议规范里客户端可以声明自己是否接受流式服务端如果不支持流式响应却收到相应客户端偏好它应当明确回一个错误。这个小点不细看文档很难注意到实战中你的第一个客户端往往是不支持流式的但只要它能和服务端正常完成工具调用就不影响最小验证。5.4 权限边界要在服务端锁死别指望协议本身MCP协议并不负责鉴权它只定义消息格式和流程。所以服务端暴露的每个工具都必须自己做权限判断。我曾经见过一个demo把内部员工接口直接封装成MCP工具结果任何能连到该端点的客户端都能调用这显然是把协议和信任模型混为一谈了。最小实践是工具内部校验用户身份宿主能控制的只是工具列表服务端要控制的才是能不能调。6. 常见问题速查表10个高频坑6.1 现象和原因对照现象可能原因优先检查顺序客户端连接后看不到工具未完成initialize握手先看服务端日志里是否收到initialize再看notifications/initialized是否发送tools/list返回空数组inputSchema格式非法用独立JSON解析工具检查schema再检查工具数组是否拼错字段名模型始终不调用工具工具description不明确或参数必填过多把description改成含具体格式精简inputSchema必填字段工具调用报参数缺失arguments结构不对查看tools/call请求里的arguments是否被客户端正确序列化响应id对不上多请求并发时未做关联检查服务端是否在异步处理时丢失request_id返回内容被模型误解content格式不规范确保结果放在content[].text内isError按实际置位服务端启动后curl无响应路由路径或CORS配置问题先用最简单POST curl排除路径错误连接进程反复崩溃传输方式不一致确认客户端和服务端都使用Streamable HTTP或都使用stdio超时频繁工具内部逻辑阻塞给工具执行加超时用并发隔离慢接口鉴权被绕过没有服务端校验在工具内部获取上下文并做身份权限判断6.2 一份顺手可用的排障清单我建议把排查顺序固定下来效率会高很多。第一步看服务端是否收到请求没收到就往路由和网络层查第二步看请求method是否按规范写漏字母或大小写都会导致客户端感知为“服务端没有该能力”第三步看返回JSON结构特别是content和isError字段第四步看协议顺序是否在initialize之前就发了tools/list。只要把以上四步走完大部分问题都能定位。剩下极少数的意外通常都出在客户端宿主对协议版本的选择上。MCP版本更新很快如果服务端和客户端版本相差过大handshake返回的protocolVersion可能不会被接受。这时候不用强行改代码直接对齐两边支持的版本号就行。最后再分享一点我的实际操作体会协议这个东西看十遍文档不如把一次完整往返打印出来看一遍。我做完这个最小闭环之后再去翻官方文档很多原本晦涩的段落突然变得很好理解因为你已经知道每一步丢掉的字段会在什么时候被对方捡起来。如果你是刚开始接触MCP建议先别急着在业务代码里大范围铺开而是用这种最小服务跑通一次链路再谈架构和性能。这套经验我用过不止一次换到流式传输、换到长连接场景底层思路仍然成立。