精读 LangChain 官方文档（十）MCP 篇：把外部工具生态接入 Agent 运行结构

Agent 一旦进入真实工程环境，就会遇到一个很硬的边界：工具能力不会只存在于当前代码仓库里。

订单查询可能在内部 CRM，文件搜索可能在企业网盘，运维动作可能在一套独立脚本里，浏览器、数据库、知识库、审批系统也各有自己的接入方式。如果每接一个系统都重新写一套 LangChain 工具，Agent 的工具层很快会变成一堆临时适配代码。

MCP 要解决的正是这个生态连接问题。它把“外部系统怎样向大模型应用暴露工具、资源和上下文”标准化，让 Agent 不必直接理解每个系统的私有协议。

在 LangChain 里，这个连接主要通过 langchain-mcp-adapters 完成。它把 MCP Server 暴露出来的 tools、resources、prompts 转成 LangChain 能使用的工具、数据对象和消息结构，再交给 create_agent 或其他 workflow 使用。

这篇文档的核心主线可以概括成一句话：

MCP = external capability protocol，LangChain MCP Adapter = protocol-to-agent bridge。

换成工程视角就是：MCP 不负责替 Agent 思考，也不是把所有工具写进 LangChain；它负责把外部能力变成标准协议，LangChain adapter 再把这些协议对象接入 Agent runtime。

下面这张图先把 MCP 篇的主线压成一层结构：外部系统在 MCP Server 中暴露能力，LangChain 通过 MCP Client 加载能力，再把它们放进 Agent 循环。

读这篇文档时，最好始终抓住这条链路：

External System -> MCP Server -> MultiServerMCPClient -> LangChain Tools -> Agent Runtime

理解这条链路后，MultiServerMCPClient、transport、ClientSession、structuredContent、resources、prompts、tool_interceptors、Callbacks 和 elicitation 就不再是分散 API，而是同一套“外部能力接入 Agent”的工程结构。

1. MCP（模型上下文协议）到底解决什么问题

它解决的问题：
MCP 解决的是“外部系统如何用统一协议向 Agent 暴露工具和上下文”。

在前一篇 Tools 里，重点是把 Python 函数变成 Agent 可以调用的工具。MCP 再往外走一步：工具不一定写在当前 Python 进程里，它可能来自一个独立服务、一个本地子进程、一个企业插件系统，甚至是另一个团队维护的能力中心。

如果没有协议层，每个外部系统都要单独写适配器：搜索系统一套、数据库一套、文件系统一套、浏览器自动化一套。这样的问题不是“代码多一点”，而是工具发现、参数约束、鉴权、会话生命周期、错误处理、进度通知都无法统一。

MCP 把这些外部能力抽象成几类标准对象：

• tools：可执行动作，例如查询数据库、调用 API、创建文件。
• resources：可读取数据，例如文件、数据库记录、接口响应。
• prompts：可复用提示词模板，适合把服务端沉淀的任务模板交给客户端使用。
• content：工具返回的文本、结构化数据或多模态内容。

示例：

import os

from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI


# 创建一个用于连接多个 MCP Server 的客户端
client = MultiServerMCPClient(
    {
        "order": {
            "transport": "http",
            "url": "http://localhost:8000/mcp",
        }
    }
)

tools = await client.get_tools()

model = ChatOpenAI(
    model="qwen3.7-max",
    api_key=os.environ["QWEN_API_KEY"],
    base_url=os.environ["QWEN_BASE_URL"],
)

agent = create_agent(
    model=model,
    tools=tools,
    system_prompt="你是一个中文订单助手，需要在必要时调用外部订单工具。",
)

result = await agent.ainvoke(
    {"messages": [{"role": "user", "content": "帮我查一下订单 A10086 的物流进度"}]}
)

这里：

• MultiServerMCPClient：LangChain MCP Adapter 提供的客户端，用来连接一个或多个 MCP Server。
• transport：传输方式，常见值包括 http 和 stdio。
• url：HTTP MCP Server 的访问地址。
• client.get_tools()：从 MCP Server 加载工具，并转成 LangChain Agent 可使用的工具对象。
• create_agent：把模型和 MCP 工具一起放进 Agent 运行结构。

业务场景：
企业内部已经有订单、知识库、合同、审批、运维脚本等能力时，MCP 可以把它们以标准方式开放给 Agent。Agent 不再和每个系统做私有耦合，而是通过 MCP Server 获取能力。

最简记法：
MCP 是外部能力接入 Agent 的标准插座。

2. MultiServerMCPClient：一个 Agent 同时接多个 MCP Server

它解决的问题：
MultiServerMCPClient 解决的是“Agent 怎样从多个 MCP Server 汇总工具能力”。

官方 Quickstart 里最核心的对象就是 MultiServerMCPClient。它接受一个字典配置，每个 key 表示一个 server 名称，每个 value 描述这个 server 如何连接。比如 math 走本地 stdio 子进程，weather 走远程 HTTP 服务。

这一步建立的是“工具发现层”。Agent 不需要知道每个工具的底层代码在哪里，只需要拿到 client.get_tools() 返回的工具列表。

下面这张图对应 MultiServerMCPClient 的位置：它像一个能力聚合器，把多个 MCP Server 的 tools 汇总成 LangChain Agent 可以消费的工具集合。

示例：

import os

from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI


# 配置多个 MCP Server 的连接方式
client = MultiServerMCPClient(
    {
        "math": {
            "transport": "stdio",
            "command": "python",
            "args": ["C:/services/math_server.py"],
        },
        "weather": {
            "transport": "http",
            "url": "http://localhost:8000/mcp",
        },
    }
)

tools = await client.get_tools()

model = ChatOpenAI(
    model="qwen3.7-max",
    api_key=os.environ["QWEN_API_KEY"],
    base_url=os.environ["QWEN_BASE_URL"],
)

agent = create_agent(
    model=model,
    tools=tools,
    system_prompt="你是一个中文业务助手，需要根据问题选择合适的外部工具。",
)

result = await agent.ainvoke(
    {"messages": [{"role": "user", "content": "北京明天天气如何？顺便计算 8 乘以 12"}]}
)

这里：

• "math"：本地数学工具服务的 server 名称。
• "weather"：天气工具服务的 server 名称。
• command：启动 stdio MCP Server 的命令。
• args：传给本地子进程的参数，通常是 server 脚本路径。
• tools：从多个 server 聚合后的工具列表。

业务场景：
一个企业助手可以同时连接知识库 MCP Server、订单 MCP Server、财务 MCP Server 和运维 MCP Server。不同系统由不同团队维护，但 Agent 看到的是统一工具菜单。

最简记法：
MultiServerMCPClient 是 MCP 工具的聚合入口。

3. Custom servers（自定义 MCP Server）：把业务能力封装成协议服务

它解决的问题：
Custom servers 解决的是“自己的业务函数怎样变成 MCP Server 暴露出去”。

LangChain 文档推荐用 FastMCP 创建自定义 MCP Server。这个方向和 @tool 很像，但目标不同：@tool 是把函数直接放进当前 LangChain 程序；FastMCP 是把函数封装成一个独立 MCP 服务，让任意支持 MCP 的客户端都能连接。

这个边界很关键。只服务于当前项目的小函数，可以直接写成 LangChain tool；要复用给多个 Agent、多个 IDE、多个自动化客户端的能力，更适合封装成 MCP Server。

示例：

from fastmcp import FastMCP

mcp = FastMCP("Order")


# 定义一个通过 MCP 暴露订单状态的工具
@mcp.tool()
def get_order_status(order_id: str) -> str:
    """根据订单号查询订单状态。"""
    return f"订单 {order_id} 当前状态：已支付，等待发货。"


# 定义一个通过 MCP 暴露退款规则的工具
@mcp.tool()
def get_refund_policy(product_type: str) -> str:
    """根据商品类型查询退款规则。"""
    return f"{product_type} 类商品支持 7 天内申请退款，特殊定制商品除外。"


if __name__ == "__main__":
    mcp.run(transport="stdio")

这里：

• FastMCP("Order")：创建名为 Order 的 MCP Server。
• @mcp.tool()：把函数注册成 MCP 工具。
• get_order_status：订单查询能力。
• get_refund_policy：退款规则查询能力。
• mcp.run(transport="stdio")：以标准输入输出方式启动本地 MCP Server。

业务场景：
公司内部可以把订单、财务、库存、知识库分别封装成 MCP Server。业务团队维护自己的服务，Agent 团队只负责连接和治理，不需要把所有业务逻辑搬进 Agent 项目。

最简记法：
MCP Server 是把业务能力产品化给 Agent 使用的一层服务。

4. Transports（传输方式）：HTTP 和 stdio 的工程边界

它解决的问题：
Transports 解决的是“MCP Client 和 MCP Server 怎样通信，以及连接生命周期如何管理”。

文档重点讲了两类传输方式：

• http 或 streamable-http：适合远程服务、团队共享服务、需要鉴权和网关治理的场景。
• stdio：客户端启动本地子进程，通过标准输入输出通信，适合本地工具、开发调试、单机脚本能力。

两者不只是网络形态不同，工程边界也不同。HTTP 更像“远程能力服务”，需要考虑请求头、鉴权、网络失败、部署地址；stdio 更像“本地工具子进程”，需要考虑进程生命周期、路径、环境变量和本地权限。

下面这张图把 transport 和 session 生命周期放在一起看：连接方式决定了 MCP Server 在哪里运行，也影响状态是否容易保留。

示例：

from langchain_mcp_adapters.client import MultiServerMCPClient


# 配置一个 HTTP MCP Server，适合远程服务
http_client = MultiServerMCPClient(
    {
        "weather": {
            "transport": "http",
            "url": "http://localhost:8000/mcp",
            "headers": {
                "Authorization": "Bearer YOUR_TOKEN",
                "X-Trace-Id": "trace-001",
            },
        }
    }
)


# 配置一个 stdio MCP Server，适合本地子进程
stdio_client = MultiServerMCPClient(
    {
        "math": {
            "transport": "stdio",
            "command": "python",
            "args": ["C:/services/math_server.py"],
        }
    }
)

这里：

• http：通过 HTTP 请求连接 MCP Server。
• headers：HTTP 连接时附带的请求头，可用于鉴权、租户标识、链路追踪。
• Authorization：常见鉴权请求头，示例中使用占位 token。
• stdio：通过本地子进程标准输入输出通信。
• command 和 args：决定本地 server 如何启动。

业务场景：
企业共享知识库通常适合 HTTP MCP Server，因为需要统一部署、鉴权和日志。个人开发环境里的文件搜索、脚本执行、临时计算工具更适合 stdio，因为启动成本低，也不用单独部署服务。

最简记法：
HTTP 像远程能力服务，stdio 像本地工具进程。

5. Stateful sessions（有状态会话）：什么时候需要 ClientSession

它解决的问题：
Stateful sessions 解决的是“MCP Server 需要在多次工具调用之间保留上下文时，客户端如何控制会话生命周期”。

官方文档有一个容易误读的点：MultiServerMCPClient 默认是 stateless。每次工具调用会创建新的 MCP ClientSession，执行工具，然后清理。即使 stdio 子进程本身是长时间存在的，如果不显式管理 session，工具调用层面仍然默认按一次调用一轮会话处理。

这种默认值很适合无状态工具。例如查询天气、计算数学表达式、读取某个文件。但如果 server 需要保留登录态、临时上下文、游标、事务、对话状态，就应该用 client.session() 创建持久 ClientSession。

示例：

import os

from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.tools import load_mcp_tools
from langchain_openai import ChatOpenAI

client = MultiServerMCPClient(
    {
        "crm": {
            "transport": "http",
            "url": "http://localhost:8000/mcp",
        }
    }
)

model = ChatOpenAI(
    model="qwen3.7-max",
    api_key=os.environ["QWEN_API_KEY"],
    base_url=os.environ["QWEN_BASE_URL"],
)


# 显式创建持久会话，用于需要跨调用保留上下文的 MCP Server
async with client.session("crm") as session:
    tools = await load_mcp_tools(session)
    agent = create_agent(
        model=model,
        tools=tools,
        system_prompt="你是一个中文客户运营助手，需要基于 CRM 工具协助分析客户。",
    )

    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "打开客户 A10086 的最近三次沟通记录"}]}
    )

这里：

• client.session("crm")：为名为 crm 的 MCP Server 创建显式会话。
• session：持久 MCP ClientSession，可用于加载工具、资源或提示词。
• load_mcp_tools(session)：从指定 session 加载工具。
• async with：控制会话进入和退出，避免资源泄漏。

业务场景：
如果 MCP Server 后面连接的是一个需要登录的业务系统，或者工具调用要维持一个临时任务上下文，就不应完全依赖默认 stateless 调用。持久 session 能让多步操作保持在同一条上下文里。

最简记法：
无状态查询用默认 client，需要跨调用上下文时用 client.session()。

6. Loading tools（加载工具）：MCP 工具如何进入 Agent 循环

它解决的问题：
Loading tools 解决的是“外部 MCP 工具怎样变成 LangChain Agent 能调用的工具”。

MCP Server 暴露出来的是协议层工具。LangChain adapter 会把它们转换成 LangChain tools，之后就可以像普通工具一样交给 create_agent。这一层转换让 Agent 无需关心工具来自本地函数、MCP Server 还是其他工具生态。

工具执行失败时也有一个重要默认行为：MCP 工具失败会以 status="error" 的 tool message 回给模型，而不是直接抛异常。这样模型有机会读懂错误、修正参数、重新尝试。只有设置 handle_tool_errors=False 时，工具执行错误才会以异常形式向外抛出。

示例：

import os

from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI

client = MultiServerMCPClient(
    {
        "knowledge": {
            "transport": "http",
            "url": "http://localhost:8000/mcp",
        }
    }
)

tools = await client.get_tools()

model = ChatOpenAI(
    model="qwen3.7-max",
    api_key=os.environ["QWEN_API_KEY"],
    base_url=os.environ["QWEN_BASE_URL"],
)

agent = create_agent(
    model=model,
    tools=tools,
    system_prompt="你是一个中文知识库助手，回答前要优先使用 MCP 知识库工具。",
)

result = await agent.ainvoke(
    {"messages": [{"role": "user", "content": "公司报销流程里，差旅住宿标准是多少？"}]}
)

这里：

• client.get_tools()：加载 MCP Server 暴露的工具。
• tools：转换后的 LangChain 工具列表。
• handle_tool_errors：控制 MCP 工具执行错误是回给模型还是抛异常。
• status="error"：工具失败时回给模型的错误状态，便于 Agent 自行恢复。

业务场景：
客服、知识库、销售助手里，工具调用失败经常是参数缺失、权限不足或数据暂不可用。把错误作为工具消息返回给模型，比让程序直接崩溃更适合交互式 Agent。

最简记法：
MCP 工具进入 Agent 的入口是 client.get_tools()。

7. Structured content 与 multimodal content：工具结果不只有文本

它解决的问题：
Structured content 和 multimodal content 解决的是“MCP 工具返回复杂结果时，LangChain 怎样保留结构和多模态信息”。

传统工具结果常常是一段字符串。但生产系统里，工具返回值往往同时包含两层内容：

• 给模型或用户看的文本摘要。
• 给程序继续处理的结构化数据，例如 JSON、表格字段、对象列表。

MCP 的 structuredContent 可以承载机器可解析的数据。LangChain adapter 会把它包装到 ToolMessage.artifact 里，让程序侧可以读取。多模态内容则会转换成 LangChain 的 standard content blocks，通过 content_blocks 访问文本、图片等块。

下面这张图把工具结果拆成三层：文本给模型读，结构化内容给程序处理，多模态块给统一消息结构承载。

示例：

from langchain.messages import ToolMessage

result = await agent.ainvoke(
    {"messages": [{"role": "user", "content": "查询客户 A10086 的最近订单，并返回结构化明细"}]}
)

for message in result["messages"]:
    if isinstance(message, ToolMessage) and message.artifact:
        structured_content = message.artifact["structured_content"]
        print(structured_content)

    if getattr(message, "type", "") == "tool":
        for block in message.content_blocks:
            if block["type"] == "text":
                print(block["text"])
            elif block["type"] == "image":
                print(block.get("url"))

这里：

• ToolMessage：工具结果在 LangChain 消息体系中的表示。
• artifact：工具消息里给程序侧使用的附加数据。
• structured_content：MCP structuredContent 转换后的结构化结果。
• content_blocks：标准化后的多模态内容块。
• block["type"]：内容块类型，例如 text 或 image。

业务场景：
数据分析助手可以让 MCP 工具返回一段中文摘要，同时把结构化指标放进 artifact 给前端图表使用。浏览器自动化工具可以返回截图图片块，前端再按标准消息块展示。

最简记法：
文本是给模型读的，artifact 和 content_blocks 是给系统继续用的。

8. Resources（资源）：让 MCP Server 暴露可读取数据

它解决的问题：
Resources 解决的是“外部系统里的数据怎样以标准对象被 Agent 读取”。

Tools 是动作，Resources 是数据。MCP Server 可以暴露文件、数据库记录、API 响应等资源；LangChain adapter 会把这些资源转换成 Blob 对象。Blob 提供统一接口，让文本内容、二进制内容、MIME 类型和元数据可以用一致方式处理。

这个能力适合“先取资料，再交给模型解释”的场景。它和工具调用的区别在于：Resource 更偏读取和上下文供给，Tool 更偏执行动作。

示例：

from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient(
    {
        "docs": {
            "transport": "http",
            "url": "http://localhost:8000/mcp",
        }
    }
)

# 加载指定 MCP Server 暴露的全部资源
blobs = await client.get_resources("docs")

# 也可以按 URI 加载特定资源
selected_blobs = await client.get_resources(
    "docs",
    uris=["file:///manuals/refund_policy.txt"],
)

for blob in selected_blobs:
    print(blob.metadata["uri"])
    print(blob.mimetype)
    print(blob.as_string())

这里：

• client.get_resources("docs")：从名为 docs 的 server 加载资源。
• uris：指定要加载的资源 URI。
• Blob：LangChain 用来表示文本或二进制内容的统一对象。
• metadata["uri"]：资源来源标识。
• mimetype：资源内容类型。
• as_string()：把文本类资源读取成字符串。

业务场景：
企业文档、合约模板、产品说明、报销制度可以作为 MCP resources 暴露。Agent 在回答前读取资源内容，减少模型凭空猜测业务规则的概率。

最简记法：
Resource 是 MCP 暴露给 Agent 的可读取资料。

9. Prompts（提示词）：把服务端任务模板交给客户端复用

它解决的问题：
Prompts 解决的是“沉淀在 MCP Server 里的提示词模板怎样被 LangChain 工作流复用”。

在复杂系统里，提示词不一定只写在 Agent 代码中。比如代码审查、合同摘要、SQL 生成、客户分层分析，都可能由某个能力服务维护标准模板。MCP prompts 可以让服务端暴露这些模板，客户端按名称和参数获取，再放入 LangChain 消息流。

LangChain adapter 会把 MCP prompt 转成 LangChain messages。这样它可以和普通消息、工具消息、系统消息一起进入模型上下文。

示例：

from langchain_mcp_adapters.client import MultiServerMCPClient

client = MultiServerMCPClient(
    {
        "prompt_hub": {
            "transport": "http",
            "url": "http://localhost:8000/mcp",
        }
    }
)

# 按名称加载提示词模板
messages = await client.get_prompt("prompt_hub", "summarize_contract")

# 带参数加载提示词模板
review_messages = await client.get_prompt(
    "prompt_hub",
    "code_review",
    arguments={
        "language": "python",
        "focus": "安全性和可维护性",
    },
)

for message in review_messages:
    print(message.type, message.content)

这里：

• get_prompt：从 MCP Server 加载提示词模板。
• "prompt_hub"：提示词服务名称。
• "summarize_contract"：提示词模板名称。
• arguments：传给模板的参数。
• messages：转换后的 LangChain 消息列表。

业务场景：
企业可以把合规审查、合同摘要、售后回复、SQL 查询等标准提示词放在统一 prompt 服务里。Agent 项目不需要复制模板，只要按名称加载。

最简记法：
MCP prompt 是服务端维护、客户端复用的任务模板。

10. Tool interceptors（工具拦截器）：把 Runtime 上下文带进 MCP 工具

它解决的问题：
Tool interceptors 解决的是“MCP Server 独立运行时，怎样读取 LangChain Agent 的运行期上下文”。

MCP Server 通常是独立进程或远程服务。它天然拿不到 LangGraph runtime 里的 store、context、agent state 或 tool call id。Interceptors 就是在 MCP 工具调用外面包一层控制逻辑，把运行期上下文、鉴权、重试、请求改写和短路返回接进去。

这层能力很像 Tools 篇里的 middleware，但作用点更靠近 MCP tool execution。它让 MCP 工具调用不只是“转发外部工具”，而是能进入 LangChain 的上下文治理体系。

下面这张图强调 Interceptor 的桥接作用：MCP Server 本身在外部，但每次工具调用都可以经过运行时上下文处理。

示例：

from dataclasses import dataclass

from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.interceptors import MCPToolCallRequest


@dataclass
class UserContext:
    user_id: str
    tenant_id: str


# 定义一个把用户上下文注入 MCP 工具参数的拦截器
async def inject_user_context(
    request: MCPToolCallRequest,
    handler,
):
    """把当前用户和租户信息注入 MCP 工具调用。"""
    runtime = request.runtime
    user_id = runtime.context.user_id
    tenant_id = runtime.context.tenant_id

    modified_request = request.override(
        args={
            **request.args,
            "user_id": user_id,
            "tenant_id": tenant_id,
        }
    )
    return await handler(modified_request)


client = MultiServerMCPClient(
    {
        "order": {
            "transport": "http",
            "url": "http://localhost:8000/mcp",
        }
    },
    tool_interceptors=[inject_user_context],
)

这里：

• MCPToolCallRequest：MCP 工具调用请求对象。
• request.runtime：本次工具调用对应的 LangChain runtime 信息。
• runtime.context：调用时注入的上下文，例如用户 ID、租户 ID、权限等级。
• request.override(args=...)：创建一个改写后的请求，原请求保持不变。
• tool_interceptors：注册 MCP 工具拦截器列表。

业务场景：
多租户 SaaS 里，订单工具不能让模型自己传 tenant_id。正确做法是在拦截器里从登录态或 runtime context 读取租户信息，再注入 MCP 工具调用。

最简记法：
Interceptor 是 MCP 工具调用进入 Runtime 治理的桥。

11. State、Store 与 Tool call ID：外部工具也要读对上下文

它解决的问题：
这一节解决的是“MCP 工具调用需要会话状态、长期记忆或调用标识时，应该从哪里拿”。

官方文档把 interceptor 访问 runtime context 拆成几个典型入口：

• runtime.context：读取本次调用注入的用户、租户、API key、权限。
• runtime.store：读取长期记忆或用户偏好。
• runtime.state：读取当前会话状态，例如是否登录、当前任务阶段。
• runtime.tool_call_id：构造正确的工具消息，或记录工具调用链路。

这延续了前面 Runtime 和 Tools 篇的三分法：不要让模型猜系统参数，不要把长期记忆塞进 prompt，也不要让外部工具绕开会话状态。

示例：

from langchain.messages import ToolMessage
from langchain_mcp_adapters.interceptors import MCPToolCallRequest


# 定义一个在敏感 MCP 工具前检查登录状态的拦截器
async def require_authentication(
    request: MCPToolCallRequest,
    handler,
):
    """未登录时阻止敏感工具调用，并返回可被 Agent 读取的工具消息。"""
    runtime = request.runtime
    state = runtime.state
    is_authenticated = state.get("authenticated", False)

    sensitive_tools = {"delete_file", "export_data", "update_settings"}

    if request.name in sensitive_tools and not is_authenticated:
        return ToolMessage(
            content="当前操作需要登录后才能继续。",
            tool_call_id=runtime.tool_call_id,
        )

    return await handler(request)

这里：

• runtime.state：当前 Agent 会话状态。
• authenticated：示例中的登录状态字段。
• request.name：当前 MCP 工具名称。
• ToolMessage：返回给 Agent 循环的工具消息。
• runtime.tool_call_id：当前工具调用 ID，保证消息能和调用对应上。

业务场景：
外部 MCP 工具可能包含导出客户数据、修改配置、删除文件等动作。权限判断不能完全放在模型提示词里，应该在 interceptor 中基于 state、context 和真实权限系统做硬控制。

最简记法：
外部工具越接近真实动作，越要从 Runtime 读系统上下文。

12. Command：MCP 工具也能更新 Agent 状态和控制流程

它解决的问题：
Command 解决的是“MCP 工具执行后，怎样更新 Agent 状态或改变图执行流”。

如果 MCP 工具只是返回文本，Agent 最多基于文本继续回答。但很多业务动作执行后，需要同步改变运行状态。例如订单提交成功后，把 task_status 改成 completed；客服流程完成后，跳到总结节点；某个任务满足终止条件后，直接结束执行。

LangChain MCP interceptors 可以返回 Command。这和 LangGraph 的状态更新机制接上了，让外部工具结果不仅进入消息，还能改变图状态和路由。

下面这张图展示 Command 的位置：MCP 工具执行结果经过 interceptor 后，可以同时写消息、更新 state，并把流程跳到下一个节点。

示例：

from langchain.messages import ToolMessage
from langchain_mcp_adapters.interceptors import MCPToolCallRequest
from langgraph.types import Command


# 定义一个在订单提交后更新状态并跳转到总结节点的拦截器
async def handle_order_completion(
    request: MCPToolCallRequest,
    handler,
):
    """订单提交成功后，把任务状态写回 Agent，并交给总结节点继续处理。"""
    result = await handler(request)

    if request.name == "submit_order":
        return Command(
            update={
                "messages": [result] if isinstance(result, ToolMessage) else [],
                "task_status": "completed",
            },
            goto="summary_agent",
        )

    return result

这里：

• Command：LangGraph 状态更新和流程控制对象。
• update：要写回 Agent state 的字段。
• messages：写回消息列表，保留工具执行结果。
• task_status：示例中的业务状态字段。
• goto：下一步跳转目标，例如 summary_agent 或 __end__。

业务场景：
在售后工单、报销审批、订单提交这类流程里，外部 MCP 工具一旦执行成功，Agent 状态也要同步变化。否则模型虽然知道“操作成功”，系统流程却还停在旧状态。

最简记法：
Command 让 MCP 工具结果不只返回文本，还能改变 Agent 运行状态。

13. Custom interceptors（自定义拦截器）：请求改写、组合顺序和错误恢复

它解决的问题：
Custom interceptors 解决的是“MCP 工具调用前后有哪些横切逻辑需要统一处理”。

官方文档把 interceptor 描述成一个 async wrapper：它接收 request 和 handler。你可以在调用 handler 前改写请求，在 handler 后处理响应，也可以不调用 handler 直接短路返回。

多个 interceptor 会按 onion pattern 组合：第一个 interceptor 是最外层，执行顺序是“外层 before -> 内层 before -> 工具执行 -> 内层 after -> 外层 after”。这对日志、鉴权、重试、请求头注入很重要，因为顺序会影响实际行为。

示例：

import asyncio

from langchain_mcp_adapters.interceptors import MCPToolCallRequest


# 定义一个记录 MCP 工具调用前后信息的拦截器
async def logging_interceptor(
    request: MCPToolCallRequest,
    handler,
):
    """记录 MCP 工具名称、参数和返回结果，便于排查工具调用链路。"""
    print(f"调用工具：{request.name}，参数：{request.args}")
    result = await handler(request)
    print(f"工具返回：{result}")
    return result


# 定义一个带指数退避的重试拦截器
async def retry_interceptor(
    request: MCPToolCallRequest,
    handler,
    max_retries: int = 3,
    delay: float = 1.0,
):
    """对传输异常或运行时异常做有限重试，避免短暂故障直接中断流程。"""
    last_error = None

    for attempt in range(max_retries):
        try:
            return await handler(request)
        except Exception as exc:
            last_error = exc
            if attempt < max_retries - 1:
                wait_time = delay * (2 ** attempt)
                await asyncio.sleep(wait_time)

    raise last_error

这里：

• handler(request)：继续执行下一层 interceptor 或真实工具调用。
• request.override(...)：用于改写参数、请求头等信息。
• max_retries：最大重试次数。
• delay：初始等待时间。
• last_error：最后一次失败异常。

业务场景：
调用外部 MCP Server 时，网络波动、服务限流、鉴权刷新、参数修正都很常见。把这些逻辑散落在每个工具里会很难维护，集中在 interceptor 层更清晰。

最简记法：
Interceptor 是 MCP 工具调用的可插拔控制层。

14. Progress、Logging 与 Elicitation：长任务要有反馈，缺信息要能追问

它解决的问题：
Progress、Logging 和 Elicitation 解决的是“MCP 工具执行过程如何被观察，以及参数不完整时如何向用户补齐信息”。

长时间运行的工具不能只让用户等。例如导入大文件、生成报告、执行批量分析、查询外部系统，都需要进度反馈。MCP 支持 progress notifications，LangChain adapter 通过 Callbacks(on_progress=...) 订阅。

MCP Server 也可以发送 logging notifications，客户端通过 Callbacks(on_logging_message=...) 接收。这样外部服务日志可以进入应用侧观察链路。

Elicitation 则处理另一种情况：工具执行到一半发现还缺用户输入。它允许 MCP Server 请求补充信息，客户端通过 on_elicitation 回调返回 accept、decline 或 cancel。

下面这张图把三类回调放在一起：progress 负责进度，logging 负责日志，elicitation 负责补充输入。

示例：

from langchain_mcp_adapters.callbacks import CallbackContext, Callbacks
from langchain_mcp_adapters.client import MultiServerMCPClient
from mcp.shared.context import RequestContext
from mcp.types import ElicitRequestParams, ElicitResult, LoggingMessageNotificationParams


# 定义一个接收 MCP 工具进度通知的回调
async def on_progress(
    progress: float,
    total: float | None,
    message: str | None,
    context: CallbackContext,
):
    """把外部 MCP 工具的进度转换成应用侧可展示的状态。"""
    percent = (progress / total * 100) if total else progress
    print(f"[{context.server_name}] 进度：{percent:.1f}% - {message}")


# 定义一个接收 MCP Server 日志通知的回调
async def on_logging_message(
    params: LoggingMessageNotificationParams,
    context: CallbackContext,
):
    """把 MCP Server 日志接入应用侧日志流。"""
    print(f"[{context.server_name}] {params.level}: {params.data}")


# 定义一个处理 MCP Server 补充信息请求的回调
async def on_elicitation(
    mcp_context: RequestContext,
    params: ElicitRequestParams,
    context: CallbackContext,
) -> ElicitResult:
    """根据 MCP Server 请求补充用户资料，示例中直接返回占位数据。"""
    return ElicitResult(
        action="accept",
        content={"email": "user@example.com", "age": 25},
    )


client = MultiServerMCPClient(
    {
        "profile": {
            "transport": "http",
            "url": "http://localhost:8000/mcp",
        }
    },
    callbacks=Callbacks(
        on_progress=on_progress,
        on_logging_message=on_logging_message,
        on_elicitation=on_elicitation,
    ),
)

这里：

• Callbacks：LangChain MCP Adapter 的回调配置入口。
• on_progress：处理进度通知。
• CallbackContext.server_name：当前回调来自哪个 MCP Server。
• on_logging_message：处理 server 日志通知。
• on_elicitation：处理 server 请求补充输入。
• ElicitResult(action="accept")：表示客户端接受请求并返回数据。
• decline 和 cancel：分别表示拒绝提供信息和取消操作。

业务场景：
报表生成工具可以通过 progress 告诉前端“正在读取数据、正在生成图表、正在汇总结论”。用户资料创建工具如果缺少邮箱或年龄，可以通过 elicitation 请求补充，而不是让整次工具调用失败。

最简记法：
MCP 回调让外部工具从黑盒执行变成可观察、可交互的过程。

总结：MCP 是 Agent 工具生态的协议层

LangChain 的 MCP 文档真正讲的是一件事：当 Agent 的能力不再局限于当前代码仓库时，外部工具、资源、提示词和运行期治理应该通过标准协议接入。

MultiServerMCPClient 负责连接多个 MCP Server；transport 决定本地子进程还是远程服务；ClientSession 处理需要持久上下文的场景；get_tools、get_resources、get_prompt 分别把工具、数据和提示词拉入 LangChain；structuredContent、artifact 和 content_blocks 保留复杂工具结果；tool_interceptors 则把 Runtime 上下文、鉴权、状态更新、重试和错误恢复接进 MCP 工具调用。

把这些层次合起来，MCP 篇建立的是这个心智模型：

MCP 不是单个工具写法，而是 Agent 连接外部能力生态的协议边界。LangChain adapter 负责把这个边界接回 Agent runtime。

理解这一层后，再读 Human-in-the-loop 和 Guardrails 时，重点就会更清楚：当外部工具真的能执行动作，下一步必须把人工审批、安全策略和权限边界放到同一条 Agent 工程链路里。