掌握MCP协议：轻松为模型添加工具，提升开发效率（收藏必备）

本文介绍了MCP（Model Context Protocol）协议，它提供了一种标准化的方式来向模型暴露工具、资源和提示模板，避免了重复造轮子和私有协议的混乱。通过langchain-mcp-adapters库，开发者可以轻松地将MCP服务器上的能力转换为标准的LangChain工具，并将其集成到Agent中。文章还详细讲解了如何使用MultiServerMCPClient连接多个MCP服务器，以及如何通过stdio和HTTP传输协议来执行工具。此外，本文还探讨了如何编写自己的MCP服务，以及如何使用拦截器来注入用户身份、短路危险工具和修改HTTP头等。最后，文章提供了一个完整的示例，展示了如何将自建的数学服务与本地工具混合使用。

一、MCP 解决什么问题

没有 MCP 时，接外部能力通常是：读对方 SDK → 包一层@tool→ 处理鉴权与错误 → 再挂到 Agent。换一家又来一遍。

有了 MCP：

1. 服务端按协议暴露 tools（可执行动作）、resources（可读数据）、prompts（可复用提示模板）。

2. 客户端（这里是你的 Agent）用统一方式发现并调用，不绑死某一家 SDK。

3. 传输可以是本地子进程（stdio），也可以是远程 HTTP。

它解决的是「怎么标准地插工具」，不是「替你写业务」。权限、PII、HITL 仍要你在 Agent / 拦截器里做。第三方 MCP 等于在跑别人的代码或连别人的服务，注意安全边界别松。

MCP协议里的三块能力：

Agent 日常最常用的是 Tools。Resources / Prompts 适合在应用层按需加载，不一定每轮都塞进模型。

二、MultiServerMCPClient

角色先分清：

1. MCP Server：外面的能力提供者（数学服务、天气服务、某家 SaaS 的官方 MCP）。

2. MCP Client：你的进程里连这些 server 的那一侧。

3. Agent：只认 LangChain Tool；不直接讲 MCP 协议。

langchain-mcp-adapters做的的就是第 2→3 步：发现 server 上的工具，转成@tool同款对象，再交给create_agent。日常用的最多的类是MultiServerMCPClient，他一个客户端可以挂多个 server。

pip install langchain-mcp-adapters

配置是一个字典：键是你起的逻辑名（"math"/"weather"），值是传输方式 + 地址。Agent 调工具时不关心名字，名字主要方便你排错、开会话（第六节session("math")）。

示例：

import asyncio from langchain.agents import create_agent from langchain_mcp_adapters.client import MultiServerMCPClient from common.models import get_model async def main(): client = MultiServerMCPClient( { # 本地：拉起子进程，走标准输入输出 "math": { "transport": "stdio", "command": "python", "args": [r"E:/path/to/math_server.py"], # 用绝对路径更省事 }, # 远程：HTTP（streamable HTTP） "weather": { "transport": "http", # 传输方式 "url": "http://localhost:8000/mcp", # 地址 }, } ) tools = await client.get_tools() # 获取工具并聚合成 LangChain tools 列表 agent = create_agent( model=get_model("deepseek", temperature=0), tools=tools, # 也可 tools=tools + my_local_tools ) r = await agent.ainvoke( {"messages": [{"role": "user", "content": "算一下 (3 + 5) * 12"}]} ) print(r["messages"][-1].content) if __name__ == "__main__": asyncio.run(main())

上面示例里的math_server.py第四节才写；先把客户端骨架看懂，下一节再补服务端。

MultiServerMCPClient无状态，每次工具调用新建 MCP 会话，跑完清理。超时、鉴权、跨调用要保留的服务端状态，后面第六节用session()。

工具执行失败时，默认把错误收成ToolMessage(status="error")还给模型，方便重试；传输层/会话级错误仍会抛异常。若要让工具错误也抛出来，设handle_tool_errors=False。

三、传输协议

MCP支持两套传输协议stdio 与 HTTP。同一套 tools，既可以塞进本地子进程，也可以挂到 URL 上。选型看部署，不看业务函数怎么写。

开发期常见组合：自己的小工具用 stdio（改完即测）；公司统一提供的能力用 HTTP。第二节示例两种各挂一个，就是这个意思。

3. 1 HTTP 与请求头

client = MultiServerMCPClient( { "weather": { "transport": "http", "url": "http://localhost:8000/mcp", "headers": { "Authorization": "Bearer YOUR_TOKEN", "X-Request-Id": "trace-001", }, } } )

也可走官方 MCP SDK 的httpx.Auth（OAuth 等），配置里传"auth": auth。公开演示可先用文档站：https://docs.langchain.com/mcp（不保证永远可用，联调时当例子即可）。

旧的 SSE 传输在规范里已弃用，新项目用 HTTP / streamable-http。

3. 2 stdio

"math": { "transport": "stdio", "command": "python", "args": ["/abs/path/math_server.py"], # 需要时还可加 "env": {"API_KEY": "..."} }

路径用绝对路径；相对路径依赖你从哪启动进程，排错很烦。

四、自己写一个 MCP 服务

4. 1 服务端和 Agent 不是一回事

容易混的一点：你在写 MCP Server，不是再写一个 LangChain Agent。

1. Server 只负责：声明工具、执行工具、按 MCP 协议把结果返回去。

2. 它不调用大模型，也不知道有没有create_agent。

3. Agent 在另一边：通过 Client 发现这些工具，由模型决定何时调用。

所以下面的代码里看不到get_model()，很正常。

4. 2 为什么用 FastMCP

实现 MCP 服务可以用官方 Python SDK 手写协议细节，样板代码偏多。FastMCP 是一层薄封装：用装饰器注册工具，再run(transport=...)选传输，我们可以不管协议怎么走，可以直接暴露工具。

它和langchain-mcp-adapters的分工：

生产里也可以不用 FastMCP、直接上官方 SDK 或别的实现也行。本篇用 FastMCP，是因为它短、好演示。

pip install fastmcp

4. 3 最小数学服务（stdio）

把文件存成math_server.py，路径与第二节 / 第八节客户端配置一致：

# math_server.py —— 独立进程；被 MultiServerMCPClient 用 stdio 拉起 from fastmcp import FastMCP # 服务显示名，出现在日志/部分客户端 UI 里；不等于工具名 mcp = FastMCP("Math") @mcp.tool() def add(a: int, b: int) -> int: """两数相加。""" return a + b @mcp.tool() def multiply(a: int, b: int) -> int: """两数相乘。""" return a * b if __name__ == "__main__": # stdio：不监听端口；父进程通过 stdin/stdout 说话 # 被 Client 拉起时不要在这里 print 调试信息到 stdout，会干扰协议 mcp.run(transport="stdio")

对应关系可以记成：

1. @mcp.tool()≈ 第 5 篇的@tool，但是注册到 MCP 服务，不是直接进 Agent。

2. 函数名add/multiply就是模型稍后看到的工具名（除非你另行配置）。

3. docstring + 类型注解会进工具 schema，模型靠它们决定怎么传参，和本地@tool一样，注意描述要准确。

4. mcp.run(transport="stdio")让进程进入「等父进程通话」模式；你单独双击运行时，终端可能像卡住，这是在等 stdin，不是死了。

先单独确认服务能起来（有报错立刻修）：

python math_server.py

再开另一个终端跑带MultiServerMCPClient的 Agent。Client 配置里的command/args必须能指到这个文件的绝对路径。

4. 4 改成 HTTP 服务

若希望别的机器或多个 Agent 共用同一个数学服务，换传输即可，工具函数不用改：

if __name__ == "__main__": # streamable-http：按 FastMCP / MCP 当前文档绑 host、port、path # 客户端侧 transport 填 "http"，url 指向该服务的 MCP 端点 mcp.run(transport="streamable-http")

HTTP 版是你先启动 server，再让 Client 连 URL；stdio 版是 Client 帮你把python math_server.py拉起来。开发自己玩用 stdio 省事；要给团队共用，再上 HTTP，并加上第三节的headers鉴权。

五、Tools / Resources / Prompts

5. 1 Tools

get_tools()会向已配置的每个 server 做一次发现，把结果合并成列表。拿到来之后，和第 5 篇本地工具没有用法差别：

tools = await client.get_tools() agent = create_agent(model=get_model(), tools=tools)

和本地工具混用（很常见：核心业务自己写，通用能力走 MCP）：

from langchain.tools import tool @tool def ping() -> str: """健康检查，本地工具。""" return "pong" tools = await client.get_tools() agent = create_agent(model=get_model(), tools=[ping, *tools])

结构化内容：有的 MCP 工具除了文本，还带structuredContent。适配器会放进ToolMessage.artifact（如structured_content）。模型默认未必看见 artifact，若要让模型也读到，可用拦截器把 JSON 追加进文本（参考文/官方有append_structured_content模式）。

多模态：截图类工具可能返回图片块；用message.content_blocks按类型读text/image，别假设永远是纯字符串。

5. 2 Resources

Resources 是可以读的数据。典型是文件、配置、某条业务记录的快照。适配器把它收成Blob，你在 Python 里读，而不是默认塞进tools=让模型每轮自己捞。

blobs = await client.get_resources("server_name") # 或指定 URI，避免一次拉全库 blobs = await client.get_resources( "server_name", uris=["file:///path/to/file.txt"], ) for blob in blobs: print(blob.metadata.get("uri"), blob.mimetype) print(blob.as_string()) # 文本；二进制用 as_bytes() 一类 API

流程一般是：你的代码按需get_resources→ 截一段摘要 → 再经动态提示或消息注入给模型。注意别把 resources 当免费用的 RAG 全库。

5. 3 Prompts

有的 MCP 服务会自带提示模板，例如「安全代码审查」「按字段总结工单」。get_prompt取回来的是已经拼好的 LangChain messages，可直接当对话素材：

messages = await client.get_prompt( "server_name", "code_review", arguments={"language": "python", "focus": "security"}, ) # 可拼进 ainvoke 的 messages，或抽其中 system 段使用

和@dynamic_prompt怎么分工：MCP prompt 适合「服务提供方定义的标准话术」；和你产品身份、VIP 文案相关的，仍放在自己的中间件里。两边可以同时使用，别两处各写一套互相打架的系统提示。

六、有状态会话session

默认模型可以记成：Client 无状态 = 每次工具调用都是「新建会话 → 调一次 → 拆掉」。
对add(3,5)这种纯函数完全够用，还能避免会话泄漏。

但有些 MCP 服务会在服务端记住「当前页」「向导走到哪一步」「浏览器已登录」等，这样第二次工具调用必须落在同一个 MCP 会话里，默认短会话会把上下文弄丢。这时不要再用裸的get_tools()，改用client.session(...)拿到持久ClientSession，再load_mcp_tools(session)：

from langchain.agents import create_agent from langchain_mcp_adapters.client import MultiServerMCPClient from langchain_mcp_adapters.tools import load_mcp_tools from common.models import get_model async def run_stateful(): client = MultiServerMCPClient({...}) async with client.session("math") as session: tools = await load_mcp_tools(session) # 绑在这个 session 上 agent = create_agent(model=get_model(), tools=tools) return await agent.ainvoke( {"messages": [{"role": "user", "content": "先 3+5，再把结果乘 12"}]} )

Resources / Prompts 也有load_mcp_resources/load_mcp_prompt，同样可挂在 session 上。
注意：stdio 子进程生命周期和「每次工具新会话」不是一回事——进程可能还在，MCP 会话仍可能每次新建。要跨调用状态，用session()。

七、拦截器

前面学习过context/state/store，但是一旦接 MCP 这些上下文和记忆会失效，工具拿不到的。以为工具实际跑在别人的进程里，协议里也没有类似 LangGraph Runtime 这种字段。

这样在create_agent(..., context_schema=Context)里传的user_id，MCP 服务默认拿不到；服务端若要求user_id参数，模型还可能瞎编一个。

工具拦截器（tool interceptor） 卡在「Agent 已决定调某个 MCP 工具」和「请求真正打出去」之间。这里你能读到与本地工具类似的runtime，于是可以：

1. 把user_id/ 权限写进参数或 HTTP 头；

2. 未登录直接返回错误ToolMessage，根本不打到远端；

3. 做重试、打日志、改Command更新 state。

它和中间件很像，但作用点更靠外，专管 MCP 出站。多个拦截器同样是洋葱序：列表里第一个最外层。

7. 1 注入用户身份

from dataclasses import dataclass from langchain_mcp_adapters.interceptors import MCPToolCallRequest @dataclass class Context: user_id: str api_key: str async def inject_user_context(request: MCPToolCallRequest, handler): """MCP 服务本身不知道 user_id，由客户端拦截器注入。""" runtime = request.runtime if runtime is None: return await handler(request) modified = request.override( args={request.args, "user_id": runtime.context.user_id} ) return await handler(modified)

client = MultiServerMCPClient( {...}, tool_interceptors=[inject_user_context], )

7. 2 按 state 短路危险工具

from langchain.messages import ToolMessage async def require_auth(request: MCPToolCallRequest, handler): runtime = request.runtime sensitive = {"delete_file", "export_data"} if request.name in sensitive and not runtime.state.get("authenticated", False): return ToolMessage( content="需要先登录才能执行该操作。", tool_call_id=runtime.tool_call_id, ) return await handler(request)

这和第 11 篇防护栏是同一类事，只是卡在「出站打 MCP 之前」。

7. 3 改 header、重试

async def auth_header(request: MCPToolCallRequest, handler): token = get_token() # 你的取票逻辑 return await handler( request.override(headers={"Authorization": f"Bearer {token}"}) )

重试时注意：默认工具业务错误不抛异常；要靠except抓传输失败，或设handle_tool_errors=False再统一处理。

进度与日志可用Callbacks(on_progress=..., on_logging_message=...)；服务端elicit（中途向用户要字段）则要配on_elicitation回调。长任务 UI、人机补参可以后做，知道有钩子即可。

八、安全、混挂与完整示例

8. 1 安全清单（接第三方前先过一遍）

1. 最小权限：能只读就别给写；token 用窄 scope。

2. 别把密钥写进 messages，放context或拦截器加 header。

3. 工具结果当不可信输入：可能含注入文案，输出侧仍要防护。

4. 超时与隔离：一个慢 MCP 别拖死整次 invoke；按服务拆客户端或加超时。

5. stdio = 跑本地代码：路径、环境变量、命令注入都要当心。

6. 和本地工具混挂时看清名字：避免模型分不清search是 MCP 还是你自己的。

规划里「接三个真实 server」在环境各异时不好复现。本篇用可自建的 math（stdio）+ 可选远程 HTTP + 本地@tool把主路径跑通；文件系统/数据库类 MCP 换成你环境里真实的 URL/命令即可，接法相同。

8. 2 完整示例：自建 math + 本地工具

先保存上一节的math_server.py，再跑客户端：

"""第 14 篇：MCP math（stdio）+ 本地工具混挂。""" import asyncio from pathlib import Path from langchain.agents import create_agent from langchain.tools import tool from langchain_mcp_adapters.client import MultiServerMCPClient from common.models import get_model MATH_SERVER = str(Path(__file__).resolve().parent / "math_server.py") @tool def unit_hint() -> str: """返回单位说明（本地工具，证明可与 MCP 工具并存）。""" return "计算结果为无量纲整数。" async def main(): client = MultiServerMCPClient( { "math": { "transport": "stdio", "command": "python", "args": [MATH_SERVER], } } ) mcp_tools = await client.get_tools() agent = create_agent( model=get_model("deepseek", temperature=0), tools=[unit_hint, *mcp_tools], system_prompt=( "你是计算器助手。算术用 MCP 的 add/multiply；" "需要单位说明时调用 unit_hint。分步计算，不要心算跳步。" ), ) result = await agent.ainvoke( { "messages": [ { "role": "user", "content": "先算 3+5，再乘 12，并说明单位。", } ] } ) print(result["messages"][-1].content) if __name__ == "__main__": asyncio.run(main())

验收：轨迹里应出现 MCP 的add/multiply与本地unit_hint；最终答案 96。若 stdio 起不来，先单独python math_server.py看报错，再查MATH_SERVER路径。

再挂第二个 HTTP server 时，往MultiServerMCPClient({...})字典多加一项即可，get_tools()会合并。工具爆炸时，回到第 9 篇LLMToolSelectorMiddleware或第 10 篇按权限override(tools=...)。

如何学习大模型 AI ？

由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。

但是具体到个人，只能说是：

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】

为什么要学习大模型？

我国在A大模型领域面临人才短缺,数量与质量均落后于发达国家。2023年，人才缺口已超百万，凸显培养不足。随着AI技术飞速发展，预计到2025年,这一缺口将急剧扩大至400万,严重制约我国AI产业的创新步伐。加强人才培养,优化教育体系,国际合作并进是破解困局、推动AI发展的关键。

大模型入门到实战全套学习大礼包

1、大模型系统化学习路线

作为学习AI大模型技术的新手，方向至关重要。 正确的学习路线可以为你节省时间，少走弯路；方向不对，努力白费。这里我给大家准备了一份最科学最系统的学习成长路线图和学习规划，带你从零基础入门到精通！

2、大模型学习书籍&文档

学习AI大模型离不开书籍文档，我精选了一系列大模型技术的书籍和学习文档（电子版），它们由领域内的顶尖专家撰写，内容全面、深入、详尽，为你学习大模型提供坚实的理论基础。

3、AI大模型最新行业报告

2025最新行业报告，针对不同行业的现状、趋势、问题、机会等进行系统地调研和评估，以了解哪些行业更适合引入大模型的技术和应用，以及在哪些方面可以发挥大模型的优势。

4、大模型项目实战&配套源码

学以致用，在项目实战中检验和巩固你所学到的知识，同时为你找工作就业和职业发展打下坚实的基础。

5、大模型大厂面试真题

面试不仅是技术的较量，更需要充分的准备。在你已经掌握了大模型技术之后，就需要开始准备面试，我精心整理了一份大模型面试题库，涵盖当前面试中可能遇到的各种技术问题，让你在面试中游刃有余。

适用人群

第一阶段（10天）：初阶应用

该阶段让大家对大模型 AI有一个最前沿的认识，对大模型 AI 的理解超过 95% 的人，可以在相关讨论时发表高级、不跟风、又接地气的见解，别人只会和 AI 聊天，而你能调教 AI，并能用代码将大模型和业务衔接。

• 大模型 AI 能干什么？
• 大模型是怎样获得「智能」的？
• 用好 AI 的核心心法
• 大模型应用业务架构
• 大模型应用技术架构
• 代码示例：向 GPT-3.5 灌入新知识
• 提示工程的意义和核心思想
• Prompt 典型构成
• 指令调优方法论
• 思维链和思维树
• Prompt 攻击和防范
• …

第二阶段（30天）：高阶应用

该阶段我们正式进入大模型 AI 进阶实战学习，学会构造私有知识库，扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架，抓住最新的技术进展，适合 Python 和 JavaScript 程序员。

• 为什么要做 RAG
• 搭建一个简单的 ChatPDF
• 检索的基础概念
• 什么是向量表示（Embeddings）
• 向量数据库与向量检索
• 基于向量检索的 RAG
• 搭建 RAG 系统的扩展知识
• 混合检索与 RAG-Fusion 简介
• 向量模型本地部署
• …

第三阶段（30天）：模型训练

恭喜你，如果学到这里，你基本可以找到一份大模型 AI相关的工作，自己也能训练 GPT 了！通过微调，训练自己的垂直大模型，能独立训练开源多模态大模型，掌握更多技术方案。

到此为止，大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗？

• 为什么要做 RAG
• 什么是模型
• 什么是模型训练
• 求解器 & 损失函数简介
• 小实验2：手写一个简单的神经网络并训练它
• 什么是训练/预训练/微调/轻量化微调
• Transformer结构简介
• 轻量化微调
• 实验数据集的构建
• …

第四阶段（20天）：商业闭环

对全球大模型从性能、吞吐量、成本等方面有一定的认知，可以在云端和本地等多种环境下部署大模型，找到适合自己的项目/创业方向，做一名被 AI 武装的产品经理。

• 硬件选型
• 带你了解全球大模型
• 使用国产大模型服务
• 搭建 OpenAI 代理
• 热身：基于阿里云 PAI 部署 Stable Diffusion
• 在本地计算机运行大模型
• 大模型的私有化部署
• 基于 vLLM 部署大模型
• 案例：如何优雅地在阿里云私有部署开源大模型
• 部署一套开源 LLM 项目
• 内容安全
• 互联网信息服务算法备案
• …

学习是一个过程，只要学习就会有挑战。天道酬勤，你越努力，就会成为越优秀的自己。

如果你能在15天内完成所有的任务，那你堪称天才。然而，如果你能完成 60-70% 的内容，你就已经开始具备成为一名大模型 AI 的正确特征了。

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】