• 人工智能
• 大模型
• AI Agent
• Agent 框架
• RAG

本篇技术指南以langchain_v1/examples/mcp目录下的可运行示例为骨架，系统讲解 LangChain v1 的langchain.mcp模块如何通过MCPAdapter把 MCP（Model 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 用一张表概括了全部示例的侧重点，这里结合源码补充每个示例的"模型依赖"与"网络依赖"两列的实际含义（✅ 表示该示例需要发起模型调用或访问外网）：

其中两个示例有额外运行前提（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. FastMCP's 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 与 DeepWiki

remote_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_prompt="Answer 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 dev

graph_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, auth=auth, httpx_client_factory=_client_factory), cache=CacheConfig(store=_CACHE, target_id=user, partition=user), ) for name, url in SERVERS.items() }) tools = await MCPAdapter(group).list_tools(cache_mode="use") return create_agent("anthropic:claude-sonnet-5", tools, system_prompt=SYSTEM_PROMPT)

几个值得展开的工程细节：

• 每用户身份：make_graph(runtime)从runtime.user.identity读取当前调用者的身份，为其铸造带身份的 Bearer 令牌（token_for(user)，见 _servers.py 中的 JWT 铸造函数），于是每次运行都以"那个用户"的身份访问舰队。
• 共享连接池与按用户分区的缓存：_POOL是一个被全体共享的httpx2.AsyncHTTPTransport，_SharedPool把它借出而不允许借用者关闭它（aclose为空实现）；InMemoryResponseCacheStore按partition=user分区缓存，配合cache_mode="use"让重复运行直接读各自用户的缓存而非刷新。
• 依赖 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.py

protocol_eras.py 回答一个很现实的问题：MCP 协议版本演进后，新旧服务器如何共存于同一个 Agent？

背景知识（源码注释明确）：MCP 改变了客户端与服务器协商能力的方式——2025-11-25 时代用initialize握手，2026-07-28 时代改用server/discover。FastMCP 按连接逐一协商，所以两个时代的工具可以同时进入同一个 Agent，调用方无需知道哪个是哪个。

关键代码：

legacy: Client[Any] = Client(weather_server(), mode="legacy") modern: Client[Any] = Client(calculator_server(), mode="auto") async with MCPAdapter(legacy) as legacy_adapter, MCPAdapter(modern) as modern_adapter: tools = await legacy_adapter.list_tools() + await modern_adapter.list_tools() print(f"legacy server -> {legacy.protocol_version} (handshake ran: {legacy.initialize_result is not None})") print(f"modern server -> {modern.protocol_version} (handshake ran: {modern.initialize_result is not None})")

• mode="legacy"固定走握手时代；mode="auto"自动协商服务器能理解的最新时代；每个客户端独立协商。
• 只有握手时代会填充initialize_result，所以它顺带充当"实际运行了哪种协商"的证明。
• 示例特意强调一个架构选择：这里是一个服务器配一个适配器，而不是一个MCPConfig舰队同时命名两者。原因是舰队被组合在单个客户端后面，该复合客户端会为其中所有内容协商一个时代——若舰队里混入只讲握手时代的老后端，整个舰队都会降级到那个时代。独立的适配器才能让每条连接保留其服务器支持的最佳时代。

失败的工具结果回到模型：tool_errors.py

tool_errors.py 演示工具失败不应终结整轮运行。当服务器报告isError=True时，适配器把它转成一个带status="error"的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(f"tool call -> status={message.status}: {message.text.strip()[:70]}")

对应的服务器端实现见 _servers.py 的calculator_server()：divide工具在分母为零时raise ValueError(...)，FastMCP 会把工具内的抛错变成isError=True的 MCP 错误结果（而不是传输层故障），这正是模型能看见并重试的原因。传输层故障仍然会抛异常——因为模型无法对这类错误采取行动。示例还在 system prompt 里强制模型"永远用divide工具做算术、不要自己心算"，否则模型会直接用已有知识作答，错误路径根本不会触发——这是构造可复现演示的实用技巧。

调用中途询问人类：elicitation.py 与 interrupt()

elicitation.py 处理一类特殊工具：没有人类的回答就无法完成。MCPAdapter把服务器的问题呈现为 LangGraph 的interrupt()，于是正在审查 Agent 工作的人直接回答它，运行随即恢复：

agent = create_agent("anthropic:claude-sonnet-5", tools, checkpointer=InMemorySaver()) 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(f"server 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，所以示例传入checkpointer=InMemorySaver()。回答按服务器自己的请求键（question["key"]）组织，跨暂停无需维护额外状态；decline或cancel可表示拒绝。

服务器端模式见 _servers.py 的booking_server()：采用守卫模式——工具先检查所需答案是否已到达，若没有就返回描述问题的InputRequiredResult而不是执行任何操作；提前返回使调用在恢复重放时保持安全。

依据元数据门禁破坏性工具：destructive_interrupt.py

destructive_interrupt.py 展示如何基于元数据而非硬编码工具名给危险工具加审批门。MCP 服务器可以用destructiveHint=True标记工具（见 _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_on=interrupt_on)], checkpointer=InMemorySaver(), )

读取提示后构建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", auth=StaticTokenVerifier(tokens={TOKEN: {"client_id": "demo", "scopes": ["read"]}})) ... async with MCPAdapter(Client(f"{url}/mcp", auth=TOKEN)) as adapter: [forecast] = await adapter.list_tools()

适配器侧的auth参数接受一个 Bearer 令牌字符串、字面量"oauth"、或任意httpx2.Auth对象；使用MCPConfig字典时每个服务器取同样的键。示例同时演示了正反两面：带令牌调用成功，不带令牌时list_tools()抛异常。注释还提醒：StaticTokenVerifier以明文内存保存令牌，仅限演示——真实部署应使用JWTVerifier对接 IdP 的 JWKS，或走 OAuth（见下一个示例）。

auth_oauth.py：OAuth 2.1 + 动态客户端注册

auth_oauth.py 演示完整流程。单个进程同时扮演两个通常分离的角色：资源服务器（提供/mcp，对未认证调用以 401 拒绝并指向其受保护资源元数据）和授权服务器（提供 discovery、/register、/authorize、/token）。生产环境两者会拆分——资源服务器是你的，授权服务器是 Auth0 / WorkOS / Okta；FastMCP 为这些 IdP 提供了 provider，而客户端运行的流程完全一致——这正是规范的意义所在。

auth = InMemoryOAuthProvider( base_url=f"http://127.0.0.1:{port}", client_registration_options=ClientRegistrationOptions( enabled=True, 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", auth="oauth")) 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(status="error")回到模型供其重试（tool_errors.py），elicitation 经interrupt()让人类在调用中途介入（elicitation.py），破坏性工具依据元数据门禁在审批之后（destructive_interrupt.py）；
• 认证分级：从静态 Bearer（auth_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 Agent
• Agent 框架
• RAG