人工智能大模型AI AgentAgent 框架RAG【免费下载链接】langchainThe agent engineering platform.项目地址https://gitcode.com/GitHub_Trending/la/langchain点击查看免费下载本篇技术指南以langchain_v1/examples/mcp目录下的可运行示例为骨架系统讲解 LangChain v1 的langchain.mcp模块如何通过MCPAdapter把 MCPModel Context Protocol服务器暴露的工具无缝适配为 LangChain 工具交给create_agent编排。文中覆盖三种传输方式in-memory、stdio、streamable HTTP、多服务器前缀隔离、LangGraph 图工厂内的长生命周期适配器、MCP 协议新老时代的协商、工具错误回传、运行中的人类介入elicitation、破坏性工具审批门禁以及静态 Bearer Token 与完整 OAuth 2.1 两种认证流程。读完你将获得一套可直接复制运行的 MCP 接入方案并能从源码层面理解每个环节的底层机制。快速开始环境准备与运行方式示例脚本全部位于 libs/langchain_v1/examples/mcp/它们是可运行、自包含的脚本每个脚本都会自行启动它所需的 MCP 服务器因此你无需预先单独拉起任何服务进程。运行前先同步依赖并注入 API 密钥uv sync --extra mcp --extra anthropic export ANTHROPIC_API_KEY... # needed by the examples that run an agent uv run examples/mcp/transports.py两个--extra分别安装langchain.mcp所需的 FastMCP 依赖pyproject.toml 中定义的mcpextra与 Anthropic 模型提供方。凡是要跑 Agent 的示例都需要ANTHROPIC_API_KEY仅演示适配器本身的示例如transports.py不发起模型调用可以省略。每个示例都可以用同样方式单独运行例如uv run examples/mcp/remote_server.py。示例全景一张表看懂十个场景README 用一张表概括了全部示例的侧重点这里结合源码补充每个示例的模型依赖与网络依赖两列的实际含义✅ 表示该示例需要发起模型调用或访问外网示例展示的核心能力模型网络transports.py一个适配器依次连接 in-memory、stdio、HTTP 三种传输remote_server.py将适配器指向公网 MCP 服务器DeepWiki✅✅multi_server.py多个服务器合并到同一个适配器工具按服务器前缀隔离✅graph_factory.py一个长生命周期适配器被langgraph dev图的每次运行共享protocol_eras.py一个 Agent 同时持有来自两个 MCP 协议时代的工具✅tool_errors.py失败的工具结果能回到模型供其自我纠错重试✅elicitation.py服务器在调用中途向人类提问经interrupt()回答后续跑✅destructive_interrupt.py依据工具元数据把破坏性工具门禁在人工审批之后✅auth_bearer.py访问受静态 Bearer Token 保护的服务器auth_oauth.py完整 OAuth 2.1 流程 动态客户端注册其中两个示例有额外运行前提README 明确说明remote_server.py调用 DeepWiki 这一公网 MCP 服务器需要联网auth_oauth.py会打开一个浏览器标签页完成授权跳转——演示用的授权服务器会自动批准因此会立刻重定向回来无需人工操作。另外三个模块_servers.py、_stdio_server.py、_fleet_servers.py是示例共享的基础设施不属于被演示的 API_servers.py存放共享的小型 MCP 服务器_stdio_server.py是经 stdio 以子进程方式启动的入口_fleet_servers.py则为 graph 工厂示例在固定端口拉起两个受令牌保护的服务器。一种目标三种传输transports.py 深入transports.py 是全目录的入门示例。它的核心观点是MCPAdapter从你交给它的目标对象推断传输方式因此从进程内服务器切换到本地脚本stdio再到远程 URL唯一变化的就是目标本身。async def show(label: str, target: MCPAdapterTarget) - None: Adapt one target and call the tool it exposes. async with MCPAdapter(target) as adapter: [forecast] await adapter.list_tools() # An MCP result arrives as LangChain content blocks, not a bare string. [block] await forecast.ainvoke({city: Oslo}) print(f{label:12} {forecast.name} - {block[text]}) async def main() - None: # In-process: no subprocess, no socket. Ideal for tests. await show(in-memory, weather_server()) # A script path is launched over stdio, one subprocess per adapter. await show(stdio, _STDIO_SERVER) # A URL is reached over streamable HTTP. FastMCPs own test helper runs the # server in a subprocess and hands back its URL. with run_server_in_process(run_weather_http) as url: await show(http, f{url}/mcp)值得注意的实战细节in-memory直接传入weather_server()返回的 FastMCP 实例不起子进程、不开 socket最适合单元测试stdio传入_stdio_server.py的路径Path(__file__).parent / _stdio_server.py适配器按路径启动一个子进程每个适配器对应一个子进程HTTP通过 FastMCP 自带的测试辅助函数run_server_in_process(run_weather_http)在子进程里起 HTTP 服务器并把 URL 交回适配器则以{url}/mcp走 streamable HTTP。另一个关键认知MCP 的返回结果是 LangChain 内容块content block而非裸字符串所以示例里用[block] await forecast.ainvoke(...)解包后读取block[text]。这是从 MCP 工具桥接到 LangChain 工具后最常见的坑之一。源码视角MCPAdapter 的目标类型与字符串语义从 libs/langchain_v1/langchain/mcp/adapter.py 可以看到MCPAdapterTarget是一个联合类型涵盖FastMCPClient[Any] | ClientGroup | ClientTransport | FastMCP | MCPServer | AnyUrl | Path | MCPConfig | dict[str, Any] | str即所有fastmcp.Client能接受的传输目标外加一个预构建的fastmcp.Client。其中str目标有一个值得警惕的语义源码注释明确说明它必须是 http/https URL。原因在于fastmcp.Client在从字符串推断传输方式时会先把字符串当作文件系统路径测试再当作 URL 测试——如果应用从配置、请求体或模型输出拿到一个字符串却让它静默地变成启动本地子进程那将是危险的隐式行为。因此MCPAdapter在构造时就用TypeAdapter(AnyUrl)校验字符串见 adapter.py 中_validate_url_target并限定 scheme 只能是http/https。要跑本地 stdio 服务器请显式使用Path(server.py)、一个 fastmcp 传输对象或MCPConfig——裸字符串一律按 URL 解读。这也解释了为什么示例里 stdio 目标用Path构造而不是直接传字符串。指向公网服务器remote_server.py 与 DeepWikiremote_server.py 演示把适配器指向真实互联网上的 MCP 服务器——DeepWiki一个能回答公开 GitHub 仓库问题的 MCP 服务通过 streamable HTTP 提供且无需认证。示例强调这里没有任何针对 DeepWiki 的特殊处理URL 就是全部配置DEEPWIKI https://mcp.deepwiki.com/mcp async with MCPAdapter(DEEPWIKI) as adapter: tools await adapter.list_tools() agent create_agent( anthropic:claude-sonnet-5, tools, system_promptAnswer only from the deepwiki tools. Never answer from memory., ) result await agent.ainvoke( {messages: [{role: user, content: question}]} )示例还给出了一个验证思路遍历result[messages]中type tool的消息打印message.name与返回文本长度以此证明是远端服务器完成了工作而不是模型凭记忆作答。这种让结果可证伪的写法在接入第三方工具时非常实用。多服务器舰队multi_server.py 的前缀隔离multi_server.py 演示多个 MCP 服务器合并到同一个适配器。一个MCPConfig字典命名每个后端FastMCP 连接全部后端并给每个工具加上配置键前缀因此两台服务器即使暴露同名工具在交给模型的工具列表里依然可区分CONFIG { mcpServers: { weather: {command: sys.executable, args: [_STDIO_SERVER, weather]}, calc: {command: sys.executable, args: [_STDIO_SERVER, calculator]}, } } async with MCPAdapter(CONFIG) as adapter: tools await adapter.list_tools()对应源码里工具名会带上weather、calc前缀。同时注意每个后端被独立寻址所以舰队可以混用传输方式——示例里两个都是 stdio但任何一个都可以换成 URL。这是把多种数据源如天气预报、计算器、数据库、文件系统组织进一个 Agent 的标准模式。图工厂中的长生命周期适配器graph_factory.py 与 langgraph devgraph_factory.py 展示最贴近生产的一个场景一个按用户区分的 MCP 舰队被langgraph dev图工厂的每次运行共享。核心结构节选_POOL httpx2.AsyncHTTPTransport() # one pool, shared by everyone _CACHE InMemoryResponseCacheStore() # one cache, partitioned by user def _client_factory(**kwargs): return httpx2.AsyncClient(transport_SharedPool(), **kwargs) async def make_graph(runtime: ServerRuntime) - CompiledStateGraph: user runtime.user.identity if runtime.user is not None else anonymous auth BearerAuth(token_for(user)) group ClientGroup({ name: Client( StreamableHttpTransport(url, authauth, httpx_client_factory_client_factory), cacheCacheConfig(store_CACHE, target_iduser, partitionuser), ) for name, url in SERVERS.items() }) tools await MCPAdapter(group).list_tools(cache_modeuse) return create_agent(anthropic:claude-sonnet-5, tools, system_promptSYSTEM_PROMPT)几个值得展开的工程细节每用户身份make_graph(runtime)从runtime.user.identity读取当前调用者的身份为其铸造带身份的 Bearer 令牌token_for(user)见 _servers.py 中的 JWT 铸造函数于是每次运行都以那个用户的身份访问舰队。共享连接池与按用户分区的缓存_POOL是一个被全体共享的httpx2.AsyncHTTPTransport_SharedPool把它借出而不允许借用者关闭它aclose为空实现InMemoryResponseCacheStore按partitionuser分区缓存配合cache_modeuse让重复运行直接读各自用户的缓存而非刷新。依赖 langgraph.json 接线langgraph.json 声明fleet: ./graph_factory.py:make_graph作为图工厂并挂载自定义认证模块auth.py。注意graph_factory.py顶部用的是运行时导入而非TYPE_CHECKING保护因为langgraph dev通过get_type_hints(make_graph)分类工厂必须能解析每个注解。配套的 run_graph_factory_demo.py 是这条链路的端到端验证脚本它先生成一个共享密钥对写入临时MCP_DEMO_KEYFILE保证工厂铸造的令牌能被服务器验证再拉起两个受令牌保护的 MCP 服务器见 _fleet_servers.py端口 8001/8002然后启动langgraph dev最后让alice、bob两个用户分别运行fleet图校验各自的 Agent 从whoami工具处报出的是各自的身份。而 auth.py 则演示了如何把x-user-id请求头解析为 LangGraph 运行身份真实部署应校验凭证并向 IdP 解析用户x-api-key被 LangGraph SDK 保留故演示改用自定义头。跨越协议时代protocol_eras.pyprotocol_eras.py 回答一个很现实的问题MCP 协议版本演进后新旧服务器如何共存于同一个 Agent背景知识源码注释明确MCP 改变了客户端与服务器协商能力的方式——2025-11-25 时代用initialize握手2026-07-28 时代改用server/discover。FastMCP 按连接逐一协商所以两个时代的工具可以同时进入同一个 Agent调用方无需知道哪个是哪个。关键代码legacy: Client[Any] Client(weather_server(), modelegacy) modern: Client[Any] Client(calculator_server(), modeauto) async with MCPAdapter(legacy) as legacy_adapter, MCPAdapter(modern) as modern_adapter: tools await legacy_adapter.list_tools() await modern_adapter.list_tools() print(flegacy server - {legacy.protocol_version} (handshake ran: {legacy.initialize_result is not None})) print(fmodern server - {modern.protocol_version} (handshake ran: {modern.initialize_result is not None}))modelegacy固定走握手时代modeauto自动协商服务器能理解的最新时代每个客户端独立协商。只有握手时代会填充initialize_result所以它顺带充当实际运行了哪种协商的证明。示例特意强调一个架构选择这里是一个服务器配一个适配器而不是一个MCPConfig舰队同时命名两者。原因是舰队被组合在单个客户端后面该复合客户端会为其中所有内容协商一个时代——若舰队里混入只讲握手时代的老后端整个舰队都会降级到那个时代。独立的适配器才能让每条连接保留其服务器支持的最佳时代。失败的工具结果回到模型tool_errors.pytool_errors.py 演示工具失败不应终结整轮运行。当服务器报告isErrorTrue时适配器把它转成一个带statuserror的ToolMessage携带服务器自己的报错文本模型读到后可以自我纠正result await agent.ainvoke( {messages: [{role: user, content: What is 10 divided by 0? Then try 10 / 4.}]} ) for message in result[messages]: if message.type tool: print(ftool call - status{message.status}: {message.text.strip()[:70]})对应的服务器端实现见 _servers.py 的calculator_server()divide工具在分母为零时raise ValueError(...)FastMCP 会把工具内的抛错变成isErrorTrue的 MCP 错误结果而不是传输层故障这正是模型能看见并重试的原因。传输层故障仍然会抛异常——因为模型无法对这类错误采取行动。示例还在 system prompt 里强制模型永远用divide工具做算术、不要自己心算否则模型会直接用已有知识作答错误路径根本不会触发——这是构造可复现演示的实用技巧。调用中途询问人类elicitation.py 与 interrupt()elicitation.py 处理一类特殊工具没有人类的回答就无法完成。MCPAdapter把服务器的问题呈现为 LangGraph 的interrupt()于是正在审查 Agent 工作的人直接回答它运行随即恢复agent create_agent(anthropic:claude-sonnet-5, tools, checkpointerInMemorySaver()) config {configurable: {thread_id: booking-1}} paused await agent.ainvoke({messages: [{role: user, content: Book a table for 4.}]}, config) [interrupt] paused[__interrupt__] [question] interrupt.value[requests] print(fserver asks ({question[mode]}): {question[message]}) answer {action: accept, content: {date: 2026-09-14}} resumed await agent.ainvoke(Command(resume{responses: {question[key]: answer}}), config)两个源码层面的要点无需任何显式开启适配器会为它构建的每个客户端在网络上声明 elicitation 能力_arm_for_interrupts见 libs/langchain_v1/langchain/mcp/elicitation.py。服务器只向在链路上作出承诺的客户端提问适配器替你作出了这个承诺。可恢复性依赖持久化恢复被中断的运行需要 checkpoint所以示例传入checkpointerInMemorySaver()。回答按服务器自己的请求键question[key]组织跨暂停无需维护额外状态decline或cancel可表示拒绝。服务器端模式见 _servers.py 的booking_server()采用守卫模式——工具先检查所需答案是否已到达若没有就返回描述问题的InputRequiredResult而不是执行任何操作提前返回使调用在恢复重放时保持安全。依据元数据门禁破坏性工具destructive_interrupt.pydestructive_interrupt.py 展示如何基于元数据而非硬编码工具名给危险工具加审批门。MCP 服务器可以用destructiveHintTrue标记工具见 _servers.py 中files_server()的delete_file适配器把它呈现在 LangChain 工具的metadata[mcp][tool][annotations][destructive_hint]上def _is_destructive(tool: BaseTool) - bool: annotations (tool.metadata or {}).get(mcp, {}).get(tool, {}).get(annotations, {}) return annotations.get(destructive_hint, False) interrupt_on { tool.name: InterruptOnConfig(allowed_decisions[approve, reject], description_describe) for tool in tools if _is_destructive(tool) } agent create_agent( anthropic:claude-sonnet-5, tools, middleware[HumanInTheLoopMiddleware(interrupt_oninterrupt_on)], checkpointerInMemorySaver(), )读取提示后构建HumanInTheLoopMiddleware的interrupt_on映射破坏性工具暂停等待审批其余工具原样执行。审批提示由_describe从待执行的工具调用格式化如Approve destructive call delete_file({path: report.md})?。由于门禁派生自元数据服务器未来暴露任何新的破坏性工具都会被自动覆盖。运行流程与 elicitation 类似paused[__interrupt__]里取出action_requests用Command(resume{decisions: [{type: approve}]})批准reject则跳过并告知模型同样依赖InMemorySaver()支持恢复。认证从静态 Bearer Token 到完整 OAuth 2.1认证是 MCP 接入生产环境绕不开的一环示例给了两个端点。auth_bearer.py静态令牌auth_bearer.py 演示认证的最简形态服务器从不签发凭证只校验到达的Authorization: Bearer token——没有发现流程、没有浏览器、没有刷新令牌是带外预置的mcp FastMCP(weather, authStaticTokenVerifier(tokens{TOKEN: {client_id: demo, scopes: [read]}})) ... async with MCPAdapter(Client(f{url}/mcp, authTOKEN)) as adapter: [forecast] await adapter.list_tools()适配器侧的auth参数接受一个 Bearer 令牌字符串、字面量oauth、或任意httpx2.Auth对象使用MCPConfig字典时每个服务器取同样的键。示例同时演示了正反两面带令牌调用成功不带令牌时list_tools()抛异常。注释还提醒StaticTokenVerifier以明文内存保存令牌仅限演示——真实部署应使用JWTVerifier对接 IdP 的 JWKS或走 OAuth见下一个示例。auth_oauth.pyOAuth 2.1 动态客户端注册auth_oauth.py 演示完整流程。单个进程同时扮演两个通常分离的角色资源服务器提供/mcp对未认证调用以 401 拒绝并指向其受保护资源元数据和授权服务器提供 discovery、/register、/authorize、/token。生产环境两者会拆分——资源服务器是你的授权服务器是 Auth0 / WorkOS / OktaFastMCP 为这些 IdP 提供了 provider而客户端运行的流程完全一致——这正是规范的意义所在。auth InMemoryOAuthProvider( base_urlfhttp://127.0.0.1:{port}, client_registration_optionsClientRegistrationOptions( enabledTrue, valid_scopes[calendar:read], default_scopes[calendar:read], ), required_scopes[calendar:read], ) ... # oauth runs discovery, dynamic registration, the browser redirect, # and the token exchange. async with MCPAdapter(Client(f{url}/mcp, authoauth)) as adapter: [whoami] await adapter.list_tools()动态客户端注册是即插即用体验的来源客户端在运行时自行注册而非预先配好 client ID。InMemoryOAuthProvider自动批准授权所以浏览器标签打开后立刻跳转回来真实服务器会在那里展示登录与同意页。令牌保存在内存中因此每次运行都会重走浏览器步骤——需要持久化时给OAuth(..., token_storage...)传入存储即可。服务器端工具whoami通过get_access_token()读取令牌身份并返回client_id与 scopes是验证整条授权链路是否走通的最直观手段。小结一套适配器覆盖 MCP 接入的全部常见形态回顾整组示例langchain.mcp的设计主线清晰可见统一入口MCPAdapter接受进程内服务器、脚本路径、URL、预构建客户端、ClientGroup、MCPConfig等任意目标adapter.py传输方式由目标自动推断字符串目标强制为 http/https URL 以防意外执行本地脚本组合能力多服务器前缀隔离multi_server.py、跨协议时代共存protocol_eras.py、按用户区分的共享连接池与缓存graph_factory.py健壮与安全失败工具以ToolMessage(statuserror)回到模型供其重试tool_errors.pyelicitation 经interrupt()让人类在调用中途介入elicitation.py破坏性工具依据元数据门禁在审批之后destructive_interrupt.py认证分级从静态 Bearerauth_bearer.py到完整 OAuth 2.1 动态注册auth_oauth.py同一套适配器 API 无缝切换。每个示例都可通过uv run examples/mcp/name.py直接运行是理解 LangChain v1 MCP 集成、以及将其移植到自有业务中最快的上手路径。进一步的实现细节可继续研读 adapter.py、elicitation.py 与 tools.py 三个核心模块。赞分享人工智能大模型AI AgentAgent 框架RAG【免费下载链接】langchainThe agent engineering platform.项目地址https://gitcode.com/GitHub_Trending/la/langchain点击查看免费下载相关推荐LangChain MCP 集成指南用 MCPAdapter 将任意 MCP 服务器接入 create_agent 智能体LangChain MCP 集成指南用 MCPAdapter 将任意 MCP 服务器接入 create_agent 智能体 Model Context Pro人工智能大模型AI AgentAgent 框架RAG5分钟搭好 Excalidraw免费开源手绘白板从克隆到协作一次跑通5分钟搭好 Excalidraw免费开源手绘白板从克隆到协作一次跑通 画系统架构流程图时你是否被卡住过专业绘图软件学习曲线太陡十分钟画不出一个框把截前端UI组件协同办公VoltAgent 集成 Zapier MCP用 HTTP 型 MCP 服务器为 Agent 接入第三方服务VoltAgent 集成 Zapier MCP用 HTTP 型 MCP 服务器为 Agent 接入第三方服务 导读 本文以 VoltAgent 官方示例 ex人工智能AI AgentAgent 框架后端多智能体RAG工具调用Agent 记忆Agent 工作流AI 评测MCP 服务MCP Clients语音上一篇Redwood GraphQL Realtime 实战指南Subscriptions、Live Queries 与 Defer/Stream 全解析下一篇goexpect插件开发Generic Spawner接口与第三方协议集成实战创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
LangChain v1 MCP 集成实战指南：用 MCPAdapter 统一接入内存、stdio、HTTP 与多服务器舰队
人工智能大模型AI AgentAgent 框架RAG【免费下载链接】langchainThe agent engineering platform.项目地址https://gitcode.com/GitHub_Trending/la/langchain点击查看免费下载本篇技术指南以langchain_v1/examples/mcp目录下的可运行示例为骨架系统讲解 LangChain v1 的langchain.mcp模块如何通过MCPAdapter把 MCPModel Context Protocol服务器暴露的工具无缝适配为 LangChain 工具交给create_agent编排。文中覆盖三种传输方式in-memory、stdio、streamable HTTP、多服务器前缀隔离、LangGraph 图工厂内的长生命周期适配器、MCP 协议新老时代的协商、工具错误回传、运行中的人类介入elicitation、破坏性工具审批门禁以及静态 Bearer Token 与完整 OAuth 2.1 两种认证流程。读完你将获得一套可直接复制运行的 MCP 接入方案并能从源码层面理解每个环节的底层机制。快速开始环境准备与运行方式示例脚本全部位于 libs/langchain_v1/examples/mcp/它们是可运行、自包含的脚本每个脚本都会自行启动它所需的 MCP 服务器因此你无需预先单独拉起任何服务进程。运行前先同步依赖并注入 API 密钥uv sync --extra mcp --extra anthropic export ANTHROPIC_API_KEY... # needed by the examples that run an agent uv run examples/mcp/transports.py两个--extra分别安装langchain.mcp所需的 FastMCP 依赖pyproject.toml 中定义的mcpextra与 Anthropic 模型提供方。凡是要跑 Agent 的示例都需要ANTHROPIC_API_KEY仅演示适配器本身的示例如transports.py不发起模型调用可以省略。每个示例都可以用同样方式单独运行例如uv run examples/mcp/remote_server.py。示例全景一张表看懂十个场景README 用一张表概括了全部示例的侧重点这里结合源码补充每个示例的模型依赖与网络依赖两列的实际含义✅ 表示该示例需要发起模型调用或访问外网示例展示的核心能力模型网络transports.py一个适配器依次连接 in-memory、stdio、HTTP 三种传输remote_server.py将适配器指向公网 MCP 服务器DeepWiki✅✅multi_server.py多个服务器合并到同一个适配器工具按服务器前缀隔离✅graph_factory.py一个长生命周期适配器被langgraph dev图的每次运行共享protocol_eras.py一个 Agent 同时持有来自两个 MCP 协议时代的工具✅tool_errors.py失败的工具结果能回到模型供其自我纠错重试✅elicitation.py服务器在调用中途向人类提问经interrupt()回答后续跑✅destructive_interrupt.py依据工具元数据把破坏性工具门禁在人工审批之后✅auth_bearer.py访问受静态 Bearer Token 保护的服务器auth_oauth.py完整 OAuth 2.1 流程 动态客户端注册其中两个示例有额外运行前提README 明确说明remote_server.py调用 DeepWiki 这一公网 MCP 服务器需要联网auth_oauth.py会打开一个浏览器标签页完成授权跳转——演示用的授权服务器会自动批准因此会立刻重定向回来无需人工操作。另外三个模块_servers.py、_stdio_server.py、_fleet_servers.py是示例共享的基础设施不属于被演示的 API_servers.py存放共享的小型 MCP 服务器_stdio_server.py是经 stdio 以子进程方式启动的入口_fleet_servers.py则为 graph 工厂示例在固定端口拉起两个受令牌保护的服务器。一种目标三种传输transports.py 深入transports.py 是全目录的入门示例。它的核心观点是MCPAdapter从你交给它的目标对象推断传输方式因此从进程内服务器切换到本地脚本stdio再到远程 URL唯一变化的就是目标本身。async def show(label: str, target: MCPAdapterTarget) - None: Adapt one target and call the tool it exposes. async with MCPAdapter(target) as adapter: [forecast] await adapter.list_tools() # An MCP result arrives as LangChain content blocks, not a bare string. [block] await forecast.ainvoke({city: Oslo}) print(f{label:12} {forecast.name} - {block[text]}) async def main() - None: # In-process: no subprocess, no socket. Ideal for tests. await show(in-memory, weather_server()) # A script path is launched over stdio, one subprocess per adapter. await show(stdio, _STDIO_SERVER) # A URL is reached over streamable HTTP. FastMCPs own test helper runs the # server in a subprocess and hands back its URL. with run_server_in_process(run_weather_http) as url: await show(http, f{url}/mcp)值得注意的实战细节in-memory直接传入weather_server()返回的 FastMCP 实例不起子进程、不开 socket最适合单元测试stdio传入_stdio_server.py的路径Path(__file__).parent / _stdio_server.py适配器按路径启动一个子进程每个适配器对应一个子进程HTTP通过 FastMCP 自带的测试辅助函数run_server_in_process(run_weather_http)在子进程里起 HTTP 服务器并把 URL 交回适配器则以{url}/mcp走 streamable HTTP。另一个关键认知MCP 的返回结果是 LangChain 内容块content block而非裸字符串所以示例里用[block] await forecast.ainvoke(...)解包后读取block[text]。这是从 MCP 工具桥接到 LangChain 工具后最常见的坑之一。源码视角MCPAdapter 的目标类型与字符串语义从 libs/langchain_v1/langchain/mcp/adapter.py 可以看到MCPAdapterTarget是一个联合类型涵盖FastMCPClient[Any] | ClientGroup | ClientTransport | FastMCP | MCPServer | AnyUrl | Path | MCPConfig | dict[str, Any] | str即所有fastmcp.Client能接受的传输目标外加一个预构建的fastmcp.Client。其中str目标有一个值得警惕的语义源码注释明确说明它必须是 http/https URL。原因在于fastmcp.Client在从字符串推断传输方式时会先把字符串当作文件系统路径测试再当作 URL 测试——如果应用从配置、请求体或模型输出拿到一个字符串却让它静默地变成启动本地子进程那将是危险的隐式行为。因此MCPAdapter在构造时就用TypeAdapter(AnyUrl)校验字符串见 adapter.py 中_validate_url_target并限定 scheme 只能是http/https。要跑本地 stdio 服务器请显式使用Path(server.py)、一个 fastmcp 传输对象或MCPConfig——裸字符串一律按 URL 解读。这也解释了为什么示例里 stdio 目标用Path构造而不是直接传字符串。指向公网服务器remote_server.py 与 DeepWikiremote_server.py 演示把适配器指向真实互联网上的 MCP 服务器——DeepWiki一个能回答公开 GitHub 仓库问题的 MCP 服务通过 streamable HTTP 提供且无需认证。示例强调这里没有任何针对 DeepWiki 的特殊处理URL 就是全部配置DEEPWIKI https://mcp.deepwiki.com/mcp async with MCPAdapter(DEEPWIKI) as adapter: tools await adapter.list_tools() agent create_agent( anthropic:claude-sonnet-5, tools, system_promptAnswer only from the deepwiki tools. Never answer from memory., ) result await agent.ainvoke( {messages: [{role: user, content: question}]} )示例还给出了一个验证思路遍历result[messages]中type tool的消息打印message.name与返回文本长度以此证明是远端服务器完成了工作而不是模型凭记忆作答。这种让结果可证伪的写法在接入第三方工具时非常实用。多服务器舰队multi_server.py 的前缀隔离multi_server.py 演示多个 MCP 服务器合并到同一个适配器。一个MCPConfig字典命名每个后端FastMCP 连接全部后端并给每个工具加上配置键前缀因此两台服务器即使暴露同名工具在交给模型的工具列表里依然可区分CONFIG { mcpServers: { weather: {command: sys.executable, args: [_STDIO_SERVER, weather]}, calc: {command: sys.executable, args: [_STDIO_SERVER, calculator]}, } } async with MCPAdapter(CONFIG) as adapter: tools await adapter.list_tools()对应源码里工具名会带上weather、calc前缀。同时注意每个后端被独立寻址所以舰队可以混用传输方式——示例里两个都是 stdio但任何一个都可以换成 URL。这是把多种数据源如天气预报、计算器、数据库、文件系统组织进一个 Agent 的标准模式。图工厂中的长生命周期适配器graph_factory.py 与 langgraph devgraph_factory.py 展示最贴近生产的一个场景一个按用户区分的 MCP 舰队被langgraph dev图工厂的每次运行共享。核心结构节选_POOL httpx2.AsyncHTTPTransport() # one pool, shared by everyone _CACHE InMemoryResponseCacheStore() # one cache, partitioned by user def _client_factory(**kwargs): return httpx2.AsyncClient(transport_SharedPool(), **kwargs) async def make_graph(runtime: ServerRuntime) - CompiledStateGraph: user runtime.user.identity if runtime.user is not None else anonymous auth BearerAuth(token_for(user)) group ClientGroup({ name: Client( StreamableHttpTransport(url, authauth, httpx_client_factory_client_factory), cacheCacheConfig(store_CACHE, target_iduser, partitionuser), ) for name, url in SERVERS.items() }) tools await MCPAdapter(group).list_tools(cache_modeuse) return create_agent(anthropic:claude-sonnet-5, tools, system_promptSYSTEM_PROMPT)几个值得展开的工程细节每用户身份make_graph(runtime)从runtime.user.identity读取当前调用者的身份为其铸造带身份的 Bearer 令牌token_for(user)见 _servers.py 中的 JWT 铸造函数于是每次运行都以那个用户的身份访问舰队。共享连接池与按用户分区的缓存_POOL是一个被全体共享的httpx2.AsyncHTTPTransport_SharedPool把它借出而不允许借用者关闭它aclose为空实现InMemoryResponseCacheStore按partitionuser分区缓存配合cache_modeuse让重复运行直接读各自用户的缓存而非刷新。依赖 langgraph.json 接线langgraph.json 声明fleet: ./graph_factory.py:make_graph作为图工厂并挂载自定义认证模块auth.py。注意graph_factory.py顶部用的是运行时导入而非TYPE_CHECKING保护因为langgraph dev通过get_type_hints(make_graph)分类工厂必须能解析每个注解。配套的 run_graph_factory_demo.py 是这条链路的端到端验证脚本它先生成一个共享密钥对写入临时MCP_DEMO_KEYFILE保证工厂铸造的令牌能被服务器验证再拉起两个受令牌保护的 MCP 服务器见 _fleet_servers.py端口 8001/8002然后启动langgraph dev最后让alice、bob两个用户分别运行fleet图校验各自的 Agent 从whoami工具处报出的是各自的身份。而 auth.py 则演示了如何把x-user-id请求头解析为 LangGraph 运行身份真实部署应校验凭证并向 IdP 解析用户x-api-key被 LangGraph SDK 保留故演示改用自定义头。跨越协议时代protocol_eras.pyprotocol_eras.py 回答一个很现实的问题MCP 协议版本演进后新旧服务器如何共存于同一个 Agent背景知识源码注释明确MCP 改变了客户端与服务器协商能力的方式——2025-11-25 时代用initialize握手2026-07-28 时代改用server/discover。FastMCP 按连接逐一协商所以两个时代的工具可以同时进入同一个 Agent调用方无需知道哪个是哪个。关键代码legacy: Client[Any] Client(weather_server(), modelegacy) modern: Client[Any] Client(calculator_server(), modeauto) async with MCPAdapter(legacy) as legacy_adapter, MCPAdapter(modern) as modern_adapter: tools await legacy_adapter.list_tools() await modern_adapter.list_tools() print(flegacy server - {legacy.protocol_version} (handshake ran: {legacy.initialize_result is not None})) print(fmodern server - {modern.protocol_version} (handshake ran: {modern.initialize_result is not None}))modelegacy固定走握手时代modeauto自动协商服务器能理解的最新时代每个客户端独立协商。只有握手时代会填充initialize_result所以它顺带充当实际运行了哪种协商的证明。示例特意强调一个架构选择这里是一个服务器配一个适配器而不是一个MCPConfig舰队同时命名两者。原因是舰队被组合在单个客户端后面该复合客户端会为其中所有内容协商一个时代——若舰队里混入只讲握手时代的老后端整个舰队都会降级到那个时代。独立的适配器才能让每条连接保留其服务器支持的最佳时代。失败的工具结果回到模型tool_errors.pytool_errors.py 演示工具失败不应终结整轮运行。当服务器报告isErrorTrue时适配器把它转成一个带statuserror的ToolMessage携带服务器自己的报错文本模型读到后可以自我纠正result await agent.ainvoke( {messages: [{role: user, content: What is 10 divided by 0? Then try 10 / 4.}]} ) for message in result[messages]: if message.type tool: print(ftool call - status{message.status}: {message.text.strip()[:70]})对应的服务器端实现见 _servers.py 的calculator_server()divide工具在分母为零时raise ValueError(...)FastMCP 会把工具内的抛错变成isErrorTrue的 MCP 错误结果而不是传输层故障这正是模型能看见并重试的原因。传输层故障仍然会抛异常——因为模型无法对这类错误采取行动。示例还在 system prompt 里强制模型永远用divide工具做算术、不要自己心算否则模型会直接用已有知识作答错误路径根本不会触发——这是构造可复现演示的实用技巧。调用中途询问人类elicitation.py 与 interrupt()elicitation.py 处理一类特殊工具没有人类的回答就无法完成。MCPAdapter把服务器的问题呈现为 LangGraph 的interrupt()于是正在审查 Agent 工作的人直接回答它运行随即恢复agent create_agent(anthropic:claude-sonnet-5, tools, checkpointerInMemorySaver()) config {configurable: {thread_id: booking-1}} paused await agent.ainvoke({messages: [{role: user, content: Book a table for 4.}]}, config) [interrupt] paused[__interrupt__] [question] interrupt.value[requests] print(fserver asks ({question[mode]}): {question[message]}) answer {action: accept, content: {date: 2026-09-14}} resumed await agent.ainvoke(Command(resume{responses: {question[key]: answer}}), config)两个源码层面的要点无需任何显式开启适配器会为它构建的每个客户端在网络上声明 elicitation 能力_arm_for_interrupts见 libs/langchain_v1/langchain/mcp/elicitation.py。服务器只向在链路上作出承诺的客户端提问适配器替你作出了这个承诺。可恢复性依赖持久化恢复被中断的运行需要 checkpoint所以示例传入checkpointerInMemorySaver()。回答按服务器自己的请求键question[key]组织跨暂停无需维护额外状态decline或cancel可表示拒绝。服务器端模式见 _servers.py 的booking_server()采用守卫模式——工具先检查所需答案是否已到达若没有就返回描述问题的InputRequiredResult而不是执行任何操作提前返回使调用在恢复重放时保持安全。依据元数据门禁破坏性工具destructive_interrupt.pydestructive_interrupt.py 展示如何基于元数据而非硬编码工具名给危险工具加审批门。MCP 服务器可以用destructiveHintTrue标记工具见 _servers.py 中files_server()的delete_file适配器把它呈现在 LangChain 工具的metadata[mcp][tool][annotations][destructive_hint]上def _is_destructive(tool: BaseTool) - bool: annotations (tool.metadata or {}).get(mcp, {}).get(tool, {}).get(annotations, {}) return annotations.get(destructive_hint, False) interrupt_on { tool.name: InterruptOnConfig(allowed_decisions[approve, reject], description_describe) for tool in tools if _is_destructive(tool) } agent create_agent( anthropic:claude-sonnet-5, tools, middleware[HumanInTheLoopMiddleware(interrupt_oninterrupt_on)], checkpointerInMemorySaver(), )读取提示后构建HumanInTheLoopMiddleware的interrupt_on映射破坏性工具暂停等待审批其余工具原样执行。审批提示由_describe从待执行的工具调用格式化如Approve destructive call delete_file({path: report.md})?。由于门禁派生自元数据服务器未来暴露任何新的破坏性工具都会被自动覆盖。运行流程与 elicitation 类似paused[__interrupt__]里取出action_requests用Command(resume{decisions: [{type: approve}]})批准reject则跳过并告知模型同样依赖InMemorySaver()支持恢复。认证从静态 Bearer Token 到完整 OAuth 2.1认证是 MCP 接入生产环境绕不开的一环示例给了两个端点。auth_bearer.py静态令牌auth_bearer.py 演示认证的最简形态服务器从不签发凭证只校验到达的Authorization: Bearer token——没有发现流程、没有浏览器、没有刷新令牌是带外预置的mcp FastMCP(weather, authStaticTokenVerifier(tokens{TOKEN: {client_id: demo, scopes: [read]}})) ... async with MCPAdapter(Client(f{url}/mcp, authTOKEN)) as adapter: [forecast] await adapter.list_tools()适配器侧的auth参数接受一个 Bearer 令牌字符串、字面量oauth、或任意httpx2.Auth对象使用MCPConfig字典时每个服务器取同样的键。示例同时演示了正反两面带令牌调用成功不带令牌时list_tools()抛异常。注释还提醒StaticTokenVerifier以明文内存保存令牌仅限演示——真实部署应使用JWTVerifier对接 IdP 的 JWKS或走 OAuth见下一个示例。auth_oauth.pyOAuth 2.1 动态客户端注册auth_oauth.py 演示完整流程。单个进程同时扮演两个通常分离的角色资源服务器提供/mcp对未认证调用以 401 拒绝并指向其受保护资源元数据和授权服务器提供 discovery、/register、/authorize、/token。生产环境两者会拆分——资源服务器是你的授权服务器是 Auth0 / WorkOS / OktaFastMCP 为这些 IdP 提供了 provider而客户端运行的流程完全一致——这正是规范的意义所在。auth InMemoryOAuthProvider( base_urlfhttp://127.0.0.1:{port}, client_registration_optionsClientRegistrationOptions( enabledTrue, valid_scopes[calendar:read], default_scopes[calendar:read], ), required_scopes[calendar:read], ) ... # oauth runs discovery, dynamic registration, the browser redirect, # and the token exchange. async with MCPAdapter(Client(f{url}/mcp, authoauth)) as adapter: [whoami] await adapter.list_tools()动态客户端注册是即插即用体验的来源客户端在运行时自行注册而非预先配好 client ID。InMemoryOAuthProvider自动批准授权所以浏览器标签打开后立刻跳转回来真实服务器会在那里展示登录与同意页。令牌保存在内存中因此每次运行都会重走浏览器步骤——需要持久化时给OAuth(..., token_storage...)传入存储即可。服务器端工具whoami通过get_access_token()读取令牌身份并返回client_id与 scopes是验证整条授权链路是否走通的最直观手段。小结一套适配器覆盖 MCP 接入的全部常见形态回顾整组示例langchain.mcp的设计主线清晰可见统一入口MCPAdapter接受进程内服务器、脚本路径、URL、预构建客户端、ClientGroup、MCPConfig等任意目标adapter.py传输方式由目标自动推断字符串目标强制为 http/https URL 以防意外执行本地脚本组合能力多服务器前缀隔离multi_server.py、跨协议时代共存protocol_eras.py、按用户区分的共享连接池与缓存graph_factory.py健壮与安全失败工具以ToolMessage(statuserror)回到模型供其重试tool_errors.pyelicitation 经interrupt()让人类在调用中途介入elicitation.py破坏性工具依据元数据门禁在审批之后destructive_interrupt.py认证分级从静态 Bearerauth_bearer.py到完整 OAuth 2.1 动态注册auth_oauth.py同一套适配器 API 无缝切换。每个示例都可通过uv run examples/mcp/name.py直接运行是理解 LangChain v1 MCP 集成、以及将其移植到自有业务中最快的上手路径。进一步的实现细节可继续研读 adapter.py、elicitation.py 与 tools.py 三个核心模块。赞分享人工智能大模型AI AgentAgent 框架RAG【免费下载链接】langchainThe agent engineering platform.项目地址https://gitcode.com/GitHub_Trending/la/langchain点击查看免费下载相关推荐LangChain MCP 集成指南用 MCPAdapter 将任意 MCP 服务器接入 create_agent 智能体LangChain MCP 集成指南用 MCPAdapter 将任意 MCP 服务器接入 create_agent 智能体 Model Context Pro人工智能大模型AI AgentAgent 框架RAG5分钟搭好 Excalidraw免费开源手绘白板从克隆到协作一次跑通5分钟搭好 Excalidraw免费开源手绘白板从克隆到协作一次跑通 画系统架构流程图时你是否被卡住过专业绘图软件学习曲线太陡十分钟画不出一个框把截前端UI组件协同办公VoltAgent 集成 Zapier MCP用 HTTP 型 MCP 服务器为 Agent 接入第三方服务VoltAgent 集成 Zapier MCP用 HTTP 型 MCP 服务器为 Agent 接入第三方服务 导读 本文以 VoltAgent 官方示例 ex人工智能AI AgentAgent 框架后端多智能体RAG工具调用Agent 记忆Agent 工作流AI 评测MCP 服务MCP Clients语音上一篇Redwood GraphQL Realtime 实战指南Subscriptions、Live Queries 与 Defer/Stream 全解析下一篇goexpect插件开发Generic Spawner接口与第三方协议集成实战创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
