ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

MCP协议实战:从Host/Client/Server到Agent工具接入完整指南

MCP协议实战:从Host/Client/Server到Agent工具接入完整指南 如果你最近在折腾大模型Agent八成听过MCP协议这个名字。我上个月被同事安利后用周末半天时间把手里的内部小助手重构成了MCP架构随后两周又把工作用的知识库查询Agent也迁了过去。回头看我之前写过的各种工具接入代码感觉以前真是白费了不少力气。这篇文章我会用完整可运行的代码把MCP协议讲清楚它到底解决什么问题、Host/Client/Server三个角色之间怎么协作、工具如何暴露给大模型以及我踩过的那些坑。文章里的代码我都跑通过基于Python的MCP SDK和任意OpenAI兼容的大模型API你只要能pip install基本就能照着把一套Agent工具框架搭起来。代码可以直接抄走适合正在做Agent开发、准备把Agent接上真实工具的同学。1. 为什么Agent开发需要MCP从连接器爆炸说起1.1 传统工具接入方式到底难在哪先说个实际场景。我之前给一个内部运维助手接工具时遇到的第一个问题不是大模型不行而是工具太多了。要查数据库写一段psycopg2连接逻辑要发企业微信通知写一个requests调用要搜日志再封装一个查询接口。每个工具都有自己的鉴权方式、参数格式、返回结构Agent想正确使用这些工具首先得在系统提示词里塞进一堆说明文档其次每接入一个新工具主程序就要跟着改一轮。这套做法在工具只有两三个的时候还能顶住。工具超过五个麻烦就来了每个工具的调用协议不统一大模型经常把参数传错。工具返回的数据结构五花八门Agent解析逻辑越写越臃肿。换一套大模型之后原来的提示词工具描述要重新调优。团队里别的项目想复用这套工具基本只能复制粘贴。我当时最头疼的是提示词膨胀。为了告诉模型怎么调工具系统提示词里塞满了JSON Schema、示例调用、错误码说明。模型能力稍弱一点就开始胡编参数或者把两个工具的用法混在一起。这本质上是因为工具接入这件事没有一个标准协议大家都在各自造轮子。1.2 MCP解决的是接入协议而非接入动作MCP协议做的事简单说就是把工具如何被描述、如何被调用、结果如何返回这三件事统一成一套标准化格式。它不是替你做具体的业务逻辑而是定义了一套通用的握手语言让大模型宿主Host、协议客户端Client和工具服务器Server之间可以互相理解。我拿它跟传统的HTTP API做对比理解起来更快维度传统函数/API接入MCP接入工具描述写在系统提示词里格式靠人约定通过tools/list动态暴露标准JSON Schema调用方式由Agent代码硬编码转发通过tools/call统一分发参数JSON格式返回结构各写各的解析靠if-else统一的ContentBlock结构新工具上线改Agent代码在Server端加一个装饰器函数即可跨模型移植提示词重写一遍工具描述随Schema自动适配上面这张表是我迁完几个工具之后实实在在的感受。最爽的是最后一行工具描述不再依赖某一个大模型看得懂的特殊写法而是变成标准的JSON Schema格式。只要模型支持OpenAI风格的Function Calling或类似的工具调用协议MCP工具列表就能直接透传过去。1.3 为什么这项协议被看作是Agent开发的新纪元我理解新纪元这个词不是说MCP发明了什么惊天动地的算法而是它把Agent工具生态从手工作坊推向标准化流水线。MCP没有发明JSON-RPC也没有发明Schema它做的是把原本散落在各家SDK里的工具接入规范统一收口了。现在的生态支持情况也确实撑得起这个定位。Claude全家桶原生支持MCP Server配置很多开源IDE插件通过MCP暴露代码上下文LangChain、LlamaIndex也都有MCP适配器。我试过同一个MCP Server从Claude Desktop切到自己的Python Agent配置只改了连接方式Server代码一行没动。这个体验放在一年前是不敢想的。还有一个容易忽略的价值是安全边界。传统方式下Agent常常直接获得一个全量的API Key或数据库账号密码MCP Server则可以在中间层做参数白名单、权限校验、敏感输出过滤模型拿到的更多是工具给你的结果视图。后面我会在踩坑部分专门讲这块怎么设计。2. MCP协议核心概念一次完整调用背后的三张牌2.1 Host、Client、Server三者的分工第一次看MCP文档的人很容易被Host、Client、Server这几个词绕晕。我用自己的话梳理一下它们的关系。Host是大模型所在的主应用比如Claude Desktop、IDE插件、你自己写的Agent主程序。它负责持有对话上下文、决定什么时候调用工具。Client是Host内部的一个协议组件负责替Host跟外部Server沟通。注意这里的Client不是一个独立进程它往往只是SDK里的一个对象。Server则是暴露工具、资源、提示词模板的独立服务进程可以跑在本地也可以部署在远程。我最近在做一个团队共享工具网关就有种明显的感觉MCP这套角色划分很接近前后端分离的思路。Host只关心当前要不要调用工具、用返回结果做什么Client负责按协议把调用请求发给ServerServer只关心工具逻辑本身。各层职责单一出了问题也容易定位——工具执行报错去查Server工具没被发现大概率是Client初始化有问题模型来回调用异常则集中在Host侧。2.2 传输层与JSON-RPC消息流转MCP协议的消息格式基于JSON-RPC 2.0传输层则是可插拔的。官方SDK里最常见的两种是stdio和HTTP/SSE。stdio就是通过标准输入输出传输JSON-RPC消息Server作为子进程被拉起Client通过管道读写消息。HTTP/SSE则是Server作为网络服务Client通过HTTP请求发起调用、通过SSE接收服务器推送。不管底层的传输方式怎么变消息流动的骨架是固定的。一次工具调用的完整过程大致是这样Host启动ClientClient拉起Server进程或连接Server地址。Client发送initialize请求Server返回协议版本和服务能力。双方交换initialized通知后Client调用tools/list获取工具列表。Host把工具列表含参数Schema交给大模型模型决定调用哪个工具。Client发送tools/call请求携带工具名和参数对象。Server执行工具逻辑返回CallToolResult内容以文本块或图片块包裹。Host把结果回传给大模型模型生成最终回答。我最初自己写Client早了卡在第六步拿到返回结果不知道怎么解析。后来打开SDK源码才发现工具返回的内容是content数组里面每个元素是TextContent或ImageContent判断isError字段才能知道这次调用是不是炸了。这几个字段不算难但没接触过的人很容易直接取某个字符串属性然后踩一脚空指针。2.3 Tools、Resources、Prompts的三类能力边界MCP协议里的Server不止能暴露工具Tools还有另外两类能力Resources和Prompts。我建议刚开始做Agent开发的同学不要忽略它们因为三类能力对应三种完全不同的用法。Tools以函数形态存在需要参数、执行动作、返回结果。适合查天气、发消息、查数据库这些操作型场景。Resources以资源URI形式暴露数据可以理解为只读上下文。适合给模型预加载知识库文档、配置文件、代码片段。Prompts预定义的提示词模板。比如问题复现报告模板、代码评审模板适合沉淀固定工作流的起始提示词。我在自己的项目里主要用Tools但后来发现一个很实用的套路把团队内部规范这类高频引用文档挂成Resources模型在回答相关问题时能直接通过resources/read把规范原文拉进上下文比塞进系统提示词省了很多token而且内容可以随时在Server端更新不需要重新发版。3. 动手写一个MCP Server完整代码与运行过程3.1 环境准备与项目结构接下来这部分是整篇文章的核心建议直接在你的项目里建一个新目录跟着做。我的示例做了一个本地运维助手风格的Server提供两个工具一个获取服务器时间一个在指定目录的Markdown笔记里按关键字搜索。这两个工具逻辑足够简单方便你验证MCP协议链路同时又覆盖了无参数调用和带参数调用两种典型形态。环境上只需要Python 3.10以上版本。先安装官方SDKpip install mcp[cli] openai这里我把openai一并装了因为后面要把工具接到大模型Agent做Tool Calling演示。mcp[cli]会额外安装一个mcp命令行工具等下我们用它的调试器检查Server。项目结构保持简单mcp_demo/ ├── server.py # MCP Server暴露两个工具 ├── client.py # MCP Client直连Server调用工具 ├── agent_demo.py # 大模型Agent循环自动调用工具 └── notes/ # 被搜索的示例笔记目录3.2 FastMCP快速实现ServerSDK目前最友好的写法是FastMCP。它把协议细节包了一层我们只需要注册函数并加装饰器SDK会自动帮你生成JSON Schema和请求分发逻辑。下面是完整的server.py# server.py import datetime import os from zoneinfo import ZoneInfo from mcp.server.fastmcp import FastMCP mcp FastMCP(ops-assistant) mcp.tool() def get_server_time(timezone: str Asia/Shanghai) - str: 获取服务器当前时间时区默认使用 Asia/Shanghai也可传入其他合法时区。 try: tz ZoneInfo(timezone) now datetime.datetime.now(tz).strftime(%Y-%m-%d %H:%M:%S %Z) except Exception: now datetime.datetime.utcnow().strftime(%Y-%m-%d %H:%M:%S UTC) return now mcp.tool() def search_notes(keyword: str, base_dir: str ./notes) - str: 在指定目录的Markdown笔记中搜索关键字返回文件路径、行号和匹配行内容。 if not os.path.exists(base_dir): return f目录 {base_dir} 不存在请检查路径。 results [] for root, _, files in os.walk(base_dir): for filename in files: if not filename.endswith(.md): continue path os.path.join(root, filename) with open(path, encodingutf-8) as fh: for line_no, line in enumerate(fh, 1): if keyword in line: results.append(f{path}:{line_no}: {line.strip()}) if not results: return 未找到匹配内容。 return \n.join(results[:20]) if __name__ __main__: mcp.run(transportstdio)这里有一个细节值得注意FastMCP会根据函数的类型注解和默认值自动生成参数Schema。比如search_notes的keyword是必填参数base_dir因为给了默认值就成了可选参数模型端拿到的Schema会正确体现这一点。所以我给函数写类型注解时要准确别图省事全写成str或Any否则模型会对参数类型产生误判。另外函数返回的字符串会由SDK自动包装进标准的结果结构里我不需要自己去构造ContentBlock。这极大简化了Server端代码也降低了我这种习惯性先写老式Server类的人的犯错概率。在notes目录里随便放两个Markdown文件比如写一段MCP协议笔记Host、Client、Server职责的内容方便后面测试搜索工具。3.3 通过MCP Inspector在本地调试工具写完Server别急着写客户端先用SDK自带的调试器验证。在项目目录执行mcp dev server.py这个命令会启动一个本地调试面板默认地址通常是http://127.0.0.1:6277。在里面可以方便地查看Server暴露了哪些工具单个工具对应的输入Schema长什么样然后直接填参数调用看到真实返回结果。我第一次用mcp dev其实是有点土的调了半天client.py怎么都显示连接失败后来才知道是Windows终端编码问题把进程给拉崩了。换成调试面板后一眼就看到Server进程的启动日志问题迎刃而解。所以我建议新手一定先用mcp dev跑通再去看Client代码。4. 用MCP Client连接Server跑通端到端调用4.1 Stdio传输的客户端代码Server本身不会说话需要一个Client去连接它。下面这段client.py演示的是stdio传输的标准姿势Client拉起一个子进程运行server.py通过管道读写JSON-RPC消息。# client.py import asyncio import sys from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): server_params StdioServerParameters( commandsys.executable, # 使用当前Python解释器避免环境变量混乱 args[server.py], ) async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() print(可用工具列表) for tool in tools: print(f- {tool.name}: {tool.description}) print(\n调用 get_server_time) time_result await session.call_tool(get_server_time, {timezone: Asia/Shanghai}) print(time_result.content[0].text) print(\n调用 search_notes) search_result await session.call_tool(search_notes, {keyword: MCP}) print(search_result.content[0].text) if __name__ __main__: asyncio.run(main())运行方式python client.py正常情况下你会看到工具列表、服务器当前时间和笔记搜索结果依次打印。这里我想提醒几点commandsys.executable是我后来改的。原来写死python在Linux服务器上经常启动不了因为服务器里python指向的是Python 2或者压根没进虚拟环境。用sys.executable可以保证Client和Server跑在同一个解释器里减少环境不一致导致的低级事故。session.call_tool的返回结果里可能有多个ContentBlock示例里直接取content[0]是因为我们自己定义的工具只会返回一段文本。如果接入别人的Server最好遍历content逐块处理别做单元素假设。ClientSession还支持session.read_resource(...)和session.list_prompts()如果你接的Server暴露了Resources和Prompts能力用同样模式调用就行。4.2 把工具接入大模型AgentTool Calling实战现在我们已经有一套能通过MCP协议查询的工具了接下来就是把它们真正交给大模型使用。这一步我选择用OpenAI兼容接口来演示因为现在主流大模型服务基本都兼容这个协议包括各种本地部署的方案你只需要把base_url和api_key换成自己的就好。核心思路不复杂把MCP Client拉到的工具列表转换成OpenAI的tools参数然后在对话循环里监听模型的tool_calls请求命中哪个工具就通过MCP的session.call_tool去执行再把结果以roletool的消息回传。# agent_demo.py import asyncio import json import os import sys from openai import AsyncOpenAI from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client SYSTEM_PROMPT ( 你是运维助手。需要查时间或搜索笔记时请调用对应工具 回答要简洁基于工具返回的真实结果不要编造。 ) async def run_agent(llm_client, model_name, server_params): messages [{role: system, content: SYSTEM_PROMPT}] async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() mcp_tools await session.list_tools() tools [ { type: function, function: { name: t.name, description: t.description, parameters: t.inputSchema or {type: object, properties: {}}, }, } for t in mcp_tools ] while True: response await llm_client.chat.completions.create( modelmodel_name, messagesmessages, toolstools, ) msg response.choices[0].message if not msg.tool_calls: print(最终回答, msg.content) break messages.append(msg) for call in msg.tool_calls: print(f模型请求调用工具{call.function.name}) args json.loads(call.function.arguments or {}) result await session.call_tool(call.function.name, args) content result.content[0].text if result.content else 空结果 messages.append({ role: tool, tool_call_id: call.id, content: content, }) async def main(): base_url os.getenv(LLM_BASE_URL, https://api.openai.com/v1) api_key os.getenv(LLM_API_KEY, sk-xxx) model_name os.getenv(LLM_MODEL, gpt-4o-mini) llm_client AsyncOpenAI(base_urlbase_url, api_keyapi_key) server_params StdioServerParameters(commandsys.executable, args[server.py]) await run_agent(llm_client, model_name, server_params) if __name__ __main__: asyncio.run(main())这段代码里最关键的几个点t.inputSchema是MCP工具自带的参数Schema直接透传给OpenAI的parameters字段。模型看到的工具定义和你用原始工具时完全一致不会因为中间过了一层MCP而失真。循环中messages.append(msg)不能漏。模型第一次返回的tool_calls消息如果不回传后续请求它就不知道之前发生了什么可能陷入重复调用同一个工具的死循环。roletool消息中的tool_call_id必须和模型给的call.id一一对应这个ID是它们之间的对账凭证。对不上部分API会直接报错。4.3 用OpenAI兼容接口测试多轮工具调用运行方式很简单export LLM_BASE_URL你的大模型服务地址 export LLM_API_KEY你的API密钥 export LLM_MODEL你使用的模型名 python agent_demo.py如果模型支持Function Calling且工具定义正确你输入类似帮我查一下现在北京时间几点然后搜一下笔记里关于MCP的内容这样的任务应该能看到模型先调用get_server_time再调用search_notes最后汇总回答。这里我踩过一个比较隐蔽的坑有些模型服务对工具名称有要求比如不允许包含下划线或必须小写。MCP工具名如果叫get_server_time没问题但如果你的工具名比较有个性可能要在转换为OpenAI工具时做一次重命名映射不然API直接拒绝请求。格式校验这种东西每家服务都有自己的小脾气接入前先查文档。另一个建议是准备一个简单测试集。我平时会把查时间搜关键词这种组合任务录成脚本每次改完Server代码都跑一遍回归确认没有破坏已有调用链。Agent开发越是往后回归测试越重要因为模型行为本身有随机性没有固定用例容易改坏而不自知。5. 实战心得与踩坑记录5.1 stdio和HTTP/SSE怎么选官方SDK支持两种传输方式但很多人一开始不知道差别包括我。简单说stdioServer是子进程适合本地开发、单租户场景。配置简单天然隔离进程环境调试方便。缺点是不支持远程访问也不适合多客户端并发连接。HTTP/SSEServer是独立网络服务适合部署在服务器上供多个Host共享。可以做鉴权、负载均衡、监控但复杂度也高很多。就我的经验如果Agent跑在用户本机比如Claude Desktop场景stdio是绝对首选。如果要给团队内部做共享工具网关直接上HTTP/SSE省得后面再迁移一次。我当初图省事先用stdio部署到一台共享服务器结果同一时间只能维持一条会话被人吐槽后老老实实改成HTTP模式。技术债欠得越早还起来越痛。5.2 工具参数校验与错误处理的隐藏坑MCP协议本身不做参数类型强校验。FastMCP虽然能根据类型注解做简单转换但遇到空值、缺参数、非法枚举值等情况返回给模型的内容如果不够明确模型就会瞎猜然后一直调同一个工具失败。我现在的做法是每个工具函数内部先做一轮防御式校验mcp.tool() def search_notes(keyword: str, base_dir: str ./notes) - str: if not keyword or not keyword.strip(): return 错误keyword 不能为空请重新输入关键词。 if not os.path.isdir(base_dir): return f错误目录 {base_dir} 不存在。 ...这类面向模型的错误信息很重要。模型不像人看到OSError: [Errno 2] No such file or directory不一定能推断出下一步该怎么做。但你告诉它目录不存在请检查路径格式它往往会修正路径后重试。工具的错误返回本质上也是给模型的一种反馈信号写得越清晰Agent的自动纠错能力越强。5.3 性能、鉴权与安全设计建议MCP Server暴露的是能执行代码的能力比单纯调个REST API风险高一个量级。如果工具里涉及读写文件、执行命令、修改数据库务必要做最小权限原则。我在团队网关里加了这么几层工具级白名单外部Host只能调用分配给它的工具没有注册过的工具不暴露。参数白名单像base_dir这类路径参数我会在Server内部做规范化校验禁止越出允许的根目录防止模型被提示词注入后去读/etc/passwd。输出过滤工具返回前对敏感字段做脱敏比如数据库连接串、访问密钥直接替换成占位符从源头避免模型把密钥当上下文输出。调用审计每个tools/call都记录工具名、参数、耗时、结果状态。后面排查模型为什么抽样工具有据可循。性能上最容易忽略的是超时。大模型服务本身响应就慢如果某个工具执行还要几十秒整个Agent用户体验会非常糟糕。我给耗时长的工具都加了异步执行进度提示的设计工具返回的文本里带上任务已提交请稍后查询之类的说明让模型知道任务不是马上有结果避免它误以为调用失败。5.4 SDK版本差异与调试细节写这篇文章的时候我用的mcp版本是1.x新版session.list_tools()返回的是工具列表可以直接遍历老版本返回的是ListToolsResult包了一层需要访问.tools属性。如果你打开某个教程发现代码对不上先检查SDK版本pip show mcp版本差异还体现在FastMCP的装饰器名称上。早期版本可能叫server.tool()现在推荐的是mcp.tool()。我一开始照着一篇老文章写直接报错后来看官方仓库的examples才对齐。建议以官方仓库的examples为准别轻信搜索引擎排名靠前的教程。调试方面除了前面说的mcp dev还可以给server.py加环境变量MCP_DEBUG1SDK会输出更详细的协议日志。我曾经靠这个抓到一个Zombie进程问题stdio模式下Client异常退出Server子进程没跟着关闭占着端口不释放。后来在Client的finally里显式关闭会话才解决。6. 我想继续做的那些扩展MCP的生态还在快速膨胀我接下来打算做三件事也给你一个扩展方向参考。第一把现有的stdio Server改造成HTTP/SSE模式部署成团队内部工具网关让不同项目的Agent共享同一套运维工具能力省得每个项目都重复接一遍。第二把知识库文档挂载成Resources让模型在回答规范类问题的时候能主动读取文件内容而不是靠每次在提示词里塞文档片段——这样上下文占用会更稳定文档更新也不用重新发版。第三尝试用MCP Server封装一个多Agent协作总线让不同职责的Agent通过标准MCP调用共享工作状态。这条路走下来我对Agent开发最大的体会是与其花心思把一个工具在提示词里描述得天花乱坠不如把工具接入做成标准化、可复用、可治理的工程。MCP不一定是最終形态但它至少让我摆脱了每个工具都得重新发明连接方式的原始阶段。如果你也正在做Agent开发建议拿出半天时间跑一遍上面的代码亲手感受一下从写工具到模型自动调用完整个链路——这个感觉跟只看文档完全不一样。
返回列表