本文介绍了MCPModel Context Protocol协议它提供了一种标准化的方式来向模型暴露工具、资源和提示模板避免了重复造轮子和私有协议的混乱。通过langchain-mcp-adapters库开发者可以轻松地将MCP服务器上的能力转换为标准的LangChain工具并将其集成到Agent中。文章还详细讲解了如何使用MultiServerMCPClient连接多个MCP服务器以及如何通过stdio和HTTP传输协议来执行工具。此外本文还探讨了如何编写自己的MCP服务以及如何使用拦截器来注入用户身份、短路危险工具和修改HTTP头等。最后文章提供了一个完整的示例展示了如何将自建的数学服务与本地工具混合使用。一、MCP 解决什么问题没有 MCP 时接外部能力通常是读对方 SDK → 包一层tool→ 处理鉴权与错误 → 再挂到 Agent。换一家又来一遍。有了 MCP服务端按协议暴露 tools可执行动作、resources可读数据、prompts可复用提示模板。客户端这里是你的 Agent用统一方式发现并调用不绑死某一家 SDK。传输可以是本地子进程stdio也可以是远程 HTTP。它解决的是「怎么标准地插工具」不是「替你写业务」。权限、PII、HITL 仍要你在 Agent / 拦截器里做。第三方 MCP 等于在跑别人的代码或连别人的服务注意安全边界别松。MCP协议里的三块能力能力干什么适配后大致变成Tools可执行函数查库、调 APILangChainBaseTool给模型调用Resources可读数据文件、记录Blob等偏你主动拉取Prompts服务端定义的提示模板LangChain messages 列表Agent 日常最常用的是 Tools。Resources / Prompts 适合在应用层按需加载不一定每轮都塞进模型。二、MultiServerMCPClient角色先分清MCP Server外面的能力提供者数学服务、天气服务、某家 SaaS 的官方 MCP。MCP Client你的进程里连这些 server 的那一侧。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: [rE:/path/to/math_server.py], # 用绝对路径更省事 }, # 远程HTTPstreamable HTTP weather: { transport: http, # 传输方式 url: http://localhost:8000/mcp, # 地址 }, } ) tools await client.get_tools() # 获取工具并聚合成 LangChain tools 列表 agent create_agent( modelget_model(deepseek, temperature0), toolstools, # 也可 toolstools 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(statuserror)还给模型方便重试传输层/会话级错误仍会抛异常。若要让工具错误也抛出来设handle_tool_errorsFalse。三、传输协议MCP支持两套传输协议stdio 与 HTTP。同一套 tools既可以塞进本地子进程也可以挂到 URL 上。选型看部署不看业务函数怎么写。stdioHTTPstreamable HTTP形态客户端拉起子进程stdin/stdout 通信URL 上的远程或本机服务适合本地工具、开发调试、单机脚本共享服务、多客户端、云端托管状态直觉子进程在连接期内活着无状态请求为主有状态靠会话设计鉴权多靠本机环境/权限headers/authBearer、OAuth 等开发期常见组合自己的小工具用 stdio改完即测公司统一提供的能力用 HTTP。第二节示例两种各挂一个就是这个意思。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.AuthOAuth 等配置里传auth: auth。公开演示可先用文档站https://docs.langchain.com/mcp不保证永远可用联调时当例子即可。旧的 SSE 传输在规范里已弃用新项目用 HTTP / streamable-http。2 stdiomath: { transport: stdio, command: python, args: [/abs/path/math_server.py], # 需要时还可加 env: {API_KEY: ...} }路径用绝对路径相对路径依赖你从哪启动进程排错很烦。四、自己写一个 MCP 服务1 服务端和 Agent 不是一回事容易混的一点你在写 MCP Server不是再写一个 LangChain Agent。Server 只负责声明工具、执行工具、按 MCP 协议把结果返回去。它不调用大模型也不知道有没有create_agent。Agent 在另一边通过 Client 发现这些工具由模型决定何时调用。所以下面的代码里看不到get_model()很正常。2 为什么用 FastMCP实现 MCP 服务可以用官方 Python SDK 手写协议细节样板代码偏多。FastMCP 是一层薄封装用装饰器注册工具再run(transport...)选传输我们可以不管协议怎么走可以直接暴露工具。它和langchain-mcp-adapters的分工包站哪一侧干什么fastmcpServer你对外提供能力langchain-mcp-adaptersClientAgent 进程内发现并调用别人或自己的能力生产里也可以不用 FastMCP、直接上官方 SDK 或别的实现也行。本篇用 FastMCP是因为它短、好演示。pip install fastmcp3 最小数学服务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(transportstdio)对应关系可以记成mcp.tool()≈ 第 5 篇的tool但是注册到 MCP 服务不是直接进 Agent。函数名add/multiply就是模型稍后看到的工具名除非你另行配置。docstring 类型注解会进工具 schema模型靠它们决定怎么传参和本地tool一样注意描述要准确。mcp.run(transportstdio)让进程进入「等父进程通话」模式你单独双击运行时终端可能像卡住这是在等 stdin不是死了。先单独确认服务能起来有报错立刻修python math_server.py再开另一个终端跑带MultiServerMCPClient的 Agent。Client 配置里的command/args必须能指到这个文件的绝对路径。4 改成 HTTP 服务若希望别的机器或多个 Agent 共用同一个数学服务换传输即可工具函数不用改if __name__ __main__: # streamable-http按 FastMCP / MCP 当前文档绑 host、port、path # 客户端侧 transport 填 httpurl 指向该服务的 MCP 端点 mcp.run(transportstreamable-http)HTTP 版是你先启动 server再让 Client 连 URLstdio 版是 Client 帮你把python math_server.py拉起来。开发自己玩用 stdio 省事要给团队共用再上 HTTP并加上第三节的headers鉴权。五、Tools / Resources / Prompts1 Toolsget_tools()会向已配置的每个 server 做一次发现把结果合并成列表。拿到来之后和第 5 篇本地工具没有用法差别tools await client.get_tools() agent create_agent(modelget_model(), toolstools)和本地工具混用很常见核心业务自己写通用能力走 MCPfrom langchain.tools import tool tool def ping() - str: 健康检查本地工具。 return pong tools await client.get_tools() agent create_agent(modelget_model(), tools[ping, *tools])结构化内容有的 MCP 工具除了文本还带structuredContent。适配器会放进ToolMessage.artifact如structured_content。模型默认未必看见 artifact若要让模型也读到可用拦截器把 JSON 追加进文本参考文/官方有append_structured_content模式。多模态截图类工具可能返回图片块用message.content_blocks按类型读text/image别假设永远是纯字符串。2 ResourcesResources 是可以读的数据。典型是文件、配置、某条业务记录的快照。适配器把它收成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 全库。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(modelget_model(), toolstools) return await agent.ainvoke( {messages: [{role: user, content: 先 35再把结果乘 12}]} )Resources / Prompts 也有load_mcp_resources/load_mcp_prompt同样可挂在 session 上。注意stdio 子进程生命周期和「每次工具新会话」不是一回事——进程可能还在MCP 会话仍可能每次新建。要跨调用状态用session()。七、拦截器前面学习过context/state/store但是一旦接 MCP 这些上下文和记忆会失效工具拿不到的。以为工具实际跑在别人的进程里协议里也没有类似 LangGraph Runtime 这种字段。这样在create_agent(..., context_schemaContext)里传的user_idMCP 服务默认拿不到服务端若要求user_id参数模型还可能瞎编一个。工具拦截器tool interceptor 卡在「Agent 已决定调某个 MCP 工具」和「请求真正打出去」之间。这里你能读到与本地工具类似的runtime于是可以把user_id/ 权限写进参数或 HTTP 头未登录直接返回错误ToolMessage根本不打到远端做重试、打日志、改Command更新 state。它和中间件很像但作用点更靠外专管 MCP 出站。多个拦截器同样是洋葱序列表里第一个最外层。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], )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_idruntime.tool_call_id, ) return await handler(request)这和第 11 篇防护栏是同一类事只是卡在「出站打 MCP 之前」。3 改 header、重试async def auth_header(request: MCPToolCallRequest, handler): token get_token() # 你的取票逻辑 return await handler( request.override(headers{Authorization: fBearer {token}}) )重试时注意默认工具业务错误不抛异常要靠except抓传输失败或设handle_tool_errorsFalse再统一处理。进度与日志可用Callbacks(on_progress..., on_logging_message...)服务端elicit中途向用户要字段则要配on_elicitation回调。长任务 UI、人机补参可以后做知道有钩子即可。八、安全、混挂与完整示例1 安全清单接第三方前先过一遍最小权限能只读就别给写token 用窄 scope。别把密钥写进 messages放context或拦截器加 header。工具结果当不可信输入可能含注入文案输出侧仍要防护。超时与隔离一个慢 MCP 别拖死整次 invoke按服务拆客户端或加超时。stdio 跑本地代码路径、环境变量、命令注入都要当心。和本地工具混挂时看清名字避免模型分不清search是 MCP 还是你自己的。规划里「接三个真实 server」在环境各异时不好复现。本篇用可自建的 mathstdio 可选远程 HTTP 本地tool把主路径跑通文件系统/数据库类 MCP 换成你环境里真实的 URL/命令即可接法相同。2 完整示例自建 math 本地工具先保存上一节的math_server.py再跑客户端第 14 篇MCP mathstdio 本地工具混挂。 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( modelget_model(deepseek, temperature0), tools[unit_hint, *mcp_tools], system_prompt( 你是计算器助手。算术用 MCP 的 add/multiply 需要单位说明时调用 unit_hint。分步计算不要心算跳步。 ), ) result await agent.ainvoke( { messages: [ { role: user, content: 先算 35再乘 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%免费】
