1. MCP 概述

MCP（模型上下文协议）是一套让大模型标准化连接外部工具、数据和服务的协议。它类似 AI 世界里的 USB 接口，让模型不需要为每个工具单独开发调用方式，而是通过统一协议发现和使用外部能力。

传统 Agent每接一个工具，都需要自己写接口和适配。

MCP：

             Agent
               |
          MCP Client
               |
          MCP协议
               |
   -------------------------
   |          |            |
 文件系统   GitHub      数据库
 MCP Server MCP Server MCP Server

MCP主要包含三个部分：

组件	作用
MCP Host	运行AI应用，例如Claude、Cursor、Agent程序
MCP Client	负责与MCP Server通信
MCP Server	提供外部能力

MCP Server提供三类能力：

类型	作用
Tools	提供可调用的函数，例如查询天气、操作数据库
Resources	提供数据资源，例如文件、数据库内容
Prompts	提供预定义提示模板

MCP和普通Tool Calling区别：

	Tool Calling	MCP
工具来源	开发者自己编写	外部MCP Server提供
接口标准	各自定义	统一协议
工具发现	手动配置	自动发现
复用性	较低	较高

MCP就是让AI能够通过统一标准访问各种外部工具和数据的协议，相当于给Agent提供了一套通用的“外接设备接口”。

2. MCP 传输方式

MCP 支持两种主要的 客户端-服务器 通信传输机制。

2.1 HTTP（也称 streamable-http）

• 通过 HTTP 请求通信。有关详细信息，请参见 MCP HTTP 运输规范。
• 适合远程服务器、云部署。
• 支持传递自定义请求头（如认证token）和实现 httpx.Auth 接口的认证机制。示例自定义身份验证实现

配置示例:

client = MultiServerMCPClient({
    "weather": {
        "transport": "http",
        "url": "http://localhost:8000/mcp",
        "headers": {                # 可选
            "Authorization": "Bearer YOUR_TOKEN"
        },
        # "auth": custom_auth_object     # 可选，实现 httpx.Auth
    }
})

2.2 stdio

• 客户端将服务器作为子进程启动，通过标准输入/输出通信。
• 适合本地工具、简单配置。
• 有状态特性：子进程在客户端连接期间持续存在，但 MultiServerMCPClient 默认仍为每次工具调用创建新会话。

配置示例:

client = MultiServerMCPClient({
    "math": {
        "transport": "stdio",
        "command": "python",
        "args": ["/path/to/your_server.py"],
    }
})

3. MCP 快速上手

3.1 自定义 MCP 服务器

使用 FastMCP 库创建自己的 MCP 服务器。

安装:

pip install fastmcp

示例 1: 数学服务器 (stdio 传输)

from fastmcp import FastMCP

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__":
    mcp.run(transport="stdio")。# 适用于本地

启动:

示例 2: 天气服务器 (streamable HTTP 传输)

from fastmcp import FastMCP

mcp = FastMCP("Weather")

@mcp.tool()
async def get_weather(city: str) -> str:
    """获取天气"""
    return f"{city} 天气晴朗！"


if __name__ == "__main__":
    mcp.run(transport="streamable-http", port=8000)

启动:

3.2 定义 MCP 客户端

使用 langchain-mcp-adapters 库让 LangChain Agent 调用 MCP 服务器上定义的工具。

安装:

pip install langchain-mcp-adapters

核心用法:

• 创建 MultiServerMCPClient，配置一个或多个 MCP 服务器（支持 stdio 和 http 传输）。
• 调用 client.get_tools() 获取所有工具。
• 将工具传入 create_agent，构建 Agent。

示例: 同时连接数学 (本地) 和天气 (远程) 服务器

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent

# FastMCP 客户端是异步的，因此我们需要使用 asyncio.run 来运行客户端
async def main():
    client = MultiServerMCPClient(
        {
            "Math": {
                "transport": "stdio",  # 本地子进程通信
                "command": "python",
                # "math_server.py"文件的绝对路径
                "args": ["C:/Users/26892/Desktop/new-langchain/test19.py"],
            },
            "Weather": {
                "transport": "streamable-http",  # 基于 HTTP 的远程服务器
                # 务必在 8000 端口上启动天气服务器。
                "url": "http://localhost:8000/mcp",
            }
        }
    )

    tools = await client.get_tools()

    agent = create_agent(
        "gpt-5-mini",
        tools
    )

    math_response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "(3 + 5) × 12 等于多少?"}]}
    )

    weather_response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "上海的天气怎么样？"}]}
    )

    print(math_response)
    print(weather_response)


if __name__ == "__main__":
    asyncio.run(main())

注意事项:

• FastMCP 客户端是异步的，因此需要使用 asyncio.run 来运行客户端。
• 默认情况下，MultiServerMCPClient 是无状态的：每次工具调用都会创建一个全新的 MCP ClientSession，执行工具后立即清理。

3.3 有状态会话

如果需要控制 MCP 会话的生命周期，如当处理维护跨工具调用上下文的有状态服务器时，可以使用 client.session() 创建持久的 ClientSession。使用 client.session("服务器名称") 上下文管理器，配合 load_mcp_tools 加载工具。

示例:

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
from langchain_mcp_adapters.tools import load_mcp_tools


async def main():
    client = MultiServerMCPClient(
        {
            "Math": {
                "transport": "stdio",  # 本地子进程通信
                "command": "python",
                # "math_server.py"文件的绝对路径
                "args": ["C:/Users/26892/Desktop/new-langchain/test19.py"],
            },
            "Weather": {
                "transport": "streamable-http",  # 基于 HTTP 的远程服务器
                # 务必在 8000 端口上启动天气服务器。
                "url": "http://localhost:8000/mcp",
            }
        }
    )

    async with client.session("Weather") as session:  # 持久会话
        tools = await load_mcp_tools(session)  # 从该会话加载工具
        agent = create_agent("gpt-5-mini", tools)

        # 该 agent 的所有工具调用将复用同一个会话
        weather_response_1 = await agent.ainvoke(
            {
                "messages": [
                    {
                        "role": "user",
                        "content": "上海的天气怎么样？"
                    }
                ]
            }
        )

        weather_response_2 = await agent.ainvoke(
            {
                "messages": [
                    {
                        "role": "user",
                        "content": "北京的天气怎么样？"
                    }
                ]
            }
        )

        print(weather_response_1)
        print(weather_response_2)


if __name__ == "__main__":
    asyncio.run(main())

此时会话的生命周期由 async with 块管理，块结束后会话自动关闭。

4. MCP 三大件

4.1 MCP 服务器

4.1.1 MCP 服务器概述

MCP (模型上下文协议) 服务器通过标准化接口向 AI 应用暴露特定能力。其核心功能由三个构件组成：

1. Tools (工具)

• 定义: AI 模型可主动调用的函数，由模型决定使用时机。支持写操作（如修改文件、调用 API、写数据库）。
• 控制方: 模型

2. Resources (资源)

• 定义: 只读的被动数据源，为模型提供上下文信息（如文件内容、数据库模式、API 文档）。
• 控制方: 应用
• 资源类型:
  • 直接资源: 固定 URI，如 calendar://events/2024
  • 资源模板: 动态 URI，支持参数，如 weather://forecast/{city}/{date}

3. Prompts (提示词)

• 定义: 预置的指令模板，指导模型配合特定工具和资源完成任务。
• 控制方: 用户
• 特点: 支持参数化、参数补全、显式调用（如斜杠命令）

4.1.2 MCP 服务器核心功能

4.1.2.1 自定义服务器

对于 MCP 服务器的创建，使用 FastMCP 库。

import json

from fastmcp import FastMCP

mcp = FastMCP("MyServer")

# 1. 工具 是客户端调用以执行操作或访问外部系统的方法。
@mcp.tool
def multiply(a: float, b: float) -> float:
    """将两个数字相乘。"""
    return a * b

# 2. 资源 接受客户端读取的数据—被动数据源，而不是可调用的函数。
# resource 函数必须返回三种类型之一:
# - str: 作为 TextResourceContents (默认情况下为 mime_type="text/plain") 发送。
# - bytes: Base64 编码并作为 BlobResourceContents 发送。
#       您应该指定一个合适的 mime_type (例如，"image/png"、"application/octet-stream")。
# - ResourceResult: 对内容、MIME 类型和元数据的完全控制。参考:
#   https://gofastmcp.com/servers/resources#resourceresult
# 注意: 要返回字典或列表等结构化数据，使用 json.dumps() 将它们序列化为 JSON 字符串。
# 这种显式的方法确保您的类型检查器在开发过程中而不是在客户端读取资源时捕获错误。
@mcp.resource(uri="data://config")
def get_config() -> str:
    return json.dumps({"topic": "dark", "version": "1.0"})

# 3. 提示 是可以重复使用的消息模板，用于指导 LLM 交互。
@mcp.prompt
def analyze_data(data_points: list[float]) -> str:
    formatted_data = ", ".join(str(point) for point in data_points)
    return f"请对这些数据点进行分析：{formatted_data}"

if __name__ == "__main__":
    mcp.run(transport="streamable-http")

每个组件类型都有详细的文档: Tools，Resources（包括 Resource Templates）和 Prompts。

4.1.2.2 MCP 社区服务器

在 MCP 爱好者社区中，活跃着众多平台与技术人员，致力于为 MCP 用户提供交流互动、资源分享等服务。

为了帮助大家更好地发现优质服务器，推荐几个广受认可的 MCP 社区：

• 魔搭社区: https://modelscope.cn/mcp
• 阿里百炼: https://bailian.console.aliyun.com/cn-beijing?spm=a2c4g.11186623.0.0.33f5494ctCRYJv&tab=mcp#/mcp-market
• MCP官方参考服务器: https://github.com/modelcontextprotocol/servers?tab=readme-ov-file#reference-servers

4.2 MCP 客户端

4.2.1 MCP 客户端核心功能

与 MCP 服务器一样，FastMCP 本身支持创建客户端，参考这里。但在 LangChain 中，其自身封装了一个 MCP Client，可使用 MultiServerMCPClient 连接一个或多个 MCP 服务器。

4.2.1.1 获取 Tools

MCP Tools 允许 MCP 服务器向 LLM 暴露可执行的函数，用于查询数据库、调用 API 或与外部系统交互。

在 LangChain 中，要通过 langchain-mcp-adapters 将这些工具转换为原生 LangChain Tool 对象，可直接集成到 Agent 或工作流中。

4.2.1.1.1 加载 Tools

使用 MultiServerMCPClient 连接一个或多个 MCP 服务器，然后调用 get_tools() 获取所有可用工具。

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

# 配置多个 MCP 服务器 (stdio / HTTP)
client = MultiServerMCPClient(
    {
        "Math": {
            "transport": "stdio",  # 本地子进程通信
            "command": "python",
            # "math_server.py"文件的绝对路径
            "args": ["C:/Users/26892/Desktop/new-langchain/test19.py"],
        },
        "Weather": {
            "transport": "streamable-http",  # 基于 HTTP 的远程服务器
            # 务必在 8000 端口上启动天气服务器。
            "url": "http://localhost:8000/mcp",
        }
    }
)

# 加载所有工具
tools = await client.get_tools()

# 创建 Agent 并使用
agent = create_agent(
    "gpt-5-mini",
    tools
)

result = await agent.ainvoke({
    "messages": [{"role": "user", "content": "(3+5)*12 等于多少？"}]
})

在持久化对话中可以这样获取工具：

# 使用 async with 创建并管理一个持久会话
    async with client.session("server") as session:  # 持久会话
        tools = await load_mcp_tools(session)  # 从该会话加载工具
        agent = create_agent(model, tools)
        response = await agent.ainvoke(
            {"messages": [{"role": "user", "content": "xxx"}]}
        )
        print(response)

4.2.1.1.2 结构化内容

MCP 工具可以返回结构化数据（如 JSON）与人类可读文本。
LangChain 将其包装为 MCPToolArtifact，通过 ToolMessage.artifact 访问。

4.2.1.1.3 提取结构化内容

from langchain.messages import ToolMessage

weather_response = await agent.ainvoke(
    {"messages": [{"role": "user", "content": "上海的天气怎么样？"}]}
)

# 执行 Agent 后
for message in weather_response["messages"]:
    if isinstance(message, ToolMessage) and message.artifact:
        structured = message.artifact["structured_content"]
        print(structured)  # 机器可解析的数据

打印:

{'result': '上海 天气晴朗！'}

4.2.1.1.4 通过拦截器自动追加到对话历史

若希望模型也能看到结构化内容，可使用 Tool Interceptor 将其附加到工具返回的文本中。

import json

from langchain_mcp_adapters.interceptors import MCPToolCallRequest
from mcp.types import TextContent

async def append_structured_content(request: MCPToolCallRequest, handler):
    result = await handler(request) # 异步获取模型的调用结果

    if result.structuredContent: # 判断工具有没有结构化返回
        result.content += [ # 把 JSON 再复制一份，塞进文本
            TextContent(
                type="text",
                text=json.dumps(result.structuredContent) 
            )
        ]

    return result



async def main():
    client = MultiServerMCPClient(
        {...},
        tool_interceptors=[append_structured_content]  # 拦截器，用于在工具调用后处理结果
    )

    tools = await client.get_tools()

    agent = create_agent(
        "gpt-5-mini",
        tools
    )

    weather_response = await agent.ainvoke(
        {
            "messages": [
                {
                    "role": "user",
                    "content": "上海的天气怎么样？"
                }
            ]
        }
    )

    print(weather_response)

打印结果:

{
    "messages": [

        HumanMessage(
            content="上海的天气怎么样？"
        ),


        AIMessage(
            content="",
            tool_calls=[
                {
                    "name": "get_weather",
                    "args": {
                        "city": "上海"
                    }
                }
            ]
        ),


        ToolMessage(
            content=[
                {
                    "type": "text",
                    "text": "上海 天气晴朗！"   # 原本的结果
                },
                {
                    "type": "text",
                    "text": "{\"result\": \"上海 天气晴朗！\"}"  # 这里是拦截器中加上的json
                }
            ],

            name="get_weather",

            artifact={
                "structured_content": {
                    "result": "上海 天气晴朗！"
                }
            }
        ),


        AIMessage(
            content=
            "上海现在天气晴朗！如果您需要更详细的信息，可以告诉我..."
        )
    ]
}

关键要点总结:

功能	说明	核心方法/属性
加载工具	从 MCP 服务器获取工具列表	client.get_tools()
结构化内容	工具返回的机器可读数据，存放在ToolMessage.artifact	message.artifact["structured_content"]
拦截器扩展	修改请求/响应、注入上下文、处理结构化内容	tool_interceptors 参数

注意: MultiServerMCPClient 默认无状态，每次工具调用会创建新会话。如需保持状态（如对话上下文），请使用 client.session() 手动管理会话生命周期，在持久会话场景下无法验证拦截器！！！

4.2.1.2 获取 Resources

Resources 是 MCP 服务器向客户端暴露数据的机制，例如：

• 文件内容（本地或远程）
• 数据库记录
• API 响应结果
• 任何可读的数据源

核心特点: Resources 是只读的，用于向 LLM 提供上下文信息，而非执行动作（那是 Tools 的职责）。

Resources vs Tools:

• Resources: 提供静态或动态数据（只读）
• Tools: 执行操作（可写、可触发副作用）

4.2.1.2.1 Blob 对象

客户端获取 Resources 时，LangChain 将 MCP Resource 统一转换为 Blob 对象，提供一致的接口处理文本和二进制内容。

Blob 是 LangChain 统一的数据容器，具有以下属性：

属性/方法	说明
blob.metadata	包含 uri（资源标识符）等元数据
blob.mimetype	MIME 类型，如 text/plain、image/png
blob.as_string()	将内容解码为字符串（文本文件）
blob.as_bytes()	获取原始字节数据（二进制文件）

4.2.1.2.2 加载 Resources

• 加载服务器上的所有 Resources

import asyncio

from langchain_mcp_adapters.client import MultiServerMCPClient


async def main():
    client = MultiServerMCPClient(
        {
            "MyServer": {
                "transport": "streamable-http",  # 基于 HTTP 的远程服务器
                "url": "http://localhost:8000/mcp",
            }
        },
    )

    # 加载服务器上的所有资源
    blobs = await client.get_resources("MyServer")
    for blob in blobs:
        print(f"URI: {blob.metadata['uri']}")   # URI: data://config
        print(f"MIME 类型: {blob.mimetype}")    # MIME 类型: text/plain
        print(blob.as_string())  # 文本内容      # {"topic": "dark", "version": "1.0"}


if __name__ == "__main__":
    asyncio.run(main())

• 按 URI 加载特定 Resources

# 只加载指定的资源 (支持批量)
blobs = await client.get_resources("MyServer", uris=["data://config"])

• 使用 Session 手动控制，适用于会话场景

import asyncio

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.resources import load_mcp_resources


async def main():
    client = MultiServerMCPClient(

        {
            "MyServer": {
                "transport": "streamable-http",  # 基于 HTTP 的远程服务器
                "url": "http://localhost:8000/mcp",
            }
        },
    )

    async with client.session("MyServer") as session:
        # 加载所有资源
        all_blobs = await load_mcp_resources(session)
        # 或按 URI 加载
        specific_blobs = await load_mcp_resources(
            session,
            uris=["data://config"]
        )

    for blob in all_blobs:
        print(f"URI: {blob.metadata['uri']}")
        print(f"MIME 类型: {blob.mimetype}")
        print(blob.as_string())  # 文本内容

    for blob in specific_blobs:
        print(f"URI: {blob.metadata['uri']}")
        print(f"MIME 类型: {blob.mimetype}")
        print(blob.as_string())  # 文本内容


if __name__ == "__main__":
    asyncio.run(main())

4.2.1.3 获取 Prompts

Prompts 是 MCP 协议的核心功能之一，它允许 MCP 服务器暴露可复用的提示模板，供客户端（如我们的 LangChain 应用）检索和使用。

• 定位：与 Tools（工具，用于执行操作）和 Resources（资源，用于读取数据）并列，是 MCP 服务器向 LLM 应用提供上下文和指导的三种主要方式之一。
• 价值：将提示工程（Prompt Engineering）的最佳实践或特定领域的复杂提示逻辑封装在服务器端，实现集中管理和复用。

4.2.1.3.1 加载 Prompts

LangChain 将 MCP 提示转换为可直接在聊天模型中使用的消息列表（messages）。有两种加载方式：

方式	描述	使用场景
通过客户端直接加载	使用 MultiServerMCPClient 实例的 get_prompt() 方法。	简单、直接的调用，适用于大多数情况。
通过会话精确控制	先创建持久会话（client.session()），再使用 load_mcp_prompt() 函数。	需要更精细地管理 MCP 会话生命周期时（例如有状态服务器）。

示例 1：加载一个简单的提示

此例展示如何从名为 MyServer 的服务器加载一个名为 analyze_data 的提示模板。

import asyncio
import json
from langchain_mcp_adapters.client import MultiServerMCPClient

async def main():
    client = MultiServerMCPClient(
        {
            "MyServer": {
                "transport": "streamable-http",  # 基于 HTTP 的远程服务器
                "url": "http://localhost:8000/mcp",
            }
        },
    )

    # 从服务器加载名为 "analyze_data" 的提示
    messages = await client.get_prompt(
        "MyServer",
        "analyze_data",
        arguments={
            "data_points": json.dumps([1.1, 2.2])
        }
    )

    # 获取到的 messages 是一个消息列表，可直接用于聊天模型
    for message in messages:
        print(f"{message.type}: {message.content}")
        # human: 请对这些数据点进行分析：1.1，2.2

if __name__ == "__main__":
    asyncio.run(main())

示例 2：使用会话更精细的控制，适用于会话场景

此例展示如何显式管理 MCP 会话，并使用 load_mcp_prompt 函数加载提示。

import asyncio
import json

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.prompts import load_mcp_prompt

async def main():
    client = MultiServerMCPClient(
        {
            "MyServer": {
                "transport": "streamable-http",  # 基于 HTTP 的远程服务器
                "url": "http://localhost:8000/mcp",
            }
        },
    )

    # 使用 async with 创建并管理一个持久会话
    async with client.session("MyServer") as session:
        # 在该会话中加载提示
        messages = await load_mcp_prompt(
            session,
            "analyze_data",
            arguments={
                "data_points": json.dumps([1.1, 2.2])
            }
        )

        # ... 在此上下文中使用 messages 进行后续操作 ...

        # 获取到的 messages 是一个消息列表，可直接用于聊天模型
        for message in messages:
            print(f"{message.type}: {message.content}")
            # human: 请对这些数据点进行分析：1.1，2.2

if __name__ == "__main__":
    asyncio.run(main())

加载到的 messages 可以直接用作与 LLM 交互的上下文起点。

4.2.1.4 完整示例：构建一个文档查询 Agent

自定义 MCP 服务器（暴露 Resources）

# docs_server.py
import json

from fastmcp import FastMCP

mcp = FastMCP("DocumentStore")

# 模拟一个文件系统资源

@mcp.resource("file:///help/guide.txt")
def get_guide() -> str:
    return """# 用户指南
1. 首先登录系统
2. 点击“新建项目”
3. 输入项目名称
"""

@mcp.resource("file:///help/faq.json")
def get_faq() -> str:
    return json.dumps({
        "q1": "如何重置密码？",
        "a1": "请点击“忘记密码”链接。"
    })

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

LangChain Agent 读取 Resources 辅助回答

import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent
from langchain.tools import tool

async def main():
    # 连接 MCP 服务器
    client = MultiServerMCPClient({
        "help_docs": {
            "transport": "stdio",
            "command": "python",
            "args": ["C:/Users/26892/Desktop/new-langchain/test23.py"],
        }
    })

    # 加载资源
    blobs = await client.get_resources("help_docs", uris=["file:///help/faq.json"])

    # 将资源内容转化为一个 Tool（方便 Agent 按需检索）
    @tool
    def search_faq_docs(query: str) -> str:
        """搜索重置密码相关文档"""
        results = []
        for blob in blobs:
            if blob.mimetype == "text/plain":
                content = blob.as_string()
                results.append(content)
        return "\n\n".join(results) if results else "未找到相关文档"

    agent = create_agent("gpt-5-mini", tools=[search_faq_docs])
    response = await agent.ainvoke({
        "messages": [{"role": "user", "content": "如何重置密码？请查阅文档。"}]
    })
    print(response)
    # {
    #     'messages': [
    #         HumanMessage(content='如何重置密码？请查阅文档。', ...),
    #         AIMessage(content='', tool_calls=[{'name': 'search_faq_docs',
    # 'args': {'query': '如何重置密码 文档'}, ...]),
    #         ToolMessage(content='{"q1":
    # "\\u5982\\u4f55\\u91cd\\u7f6e\\u5bc6\\u7801\\uff1f", "a1":
    # "\\u8bf7\\u70b9\\u51fb\\u201c\\u5fd8\\u8bb0\\u5bc6\\u7801\\u201d\\u94fe\\u63a5\\u3002"}', ...),
    #         AIMessage(content='根据文档，重置密码的步骤如下：\n1. 在登录页面点击“忘记密码”链接。 \n2. 按页面提示输入注册时使用的邮箱（或手机号，视系统而定）。 \n3. 系统会向该邮箱发送重置密码的邮件（或发送验证码到手机号）。 \n4. 打开收到的邮件，点击邮件中的重置链接；如果是验证码页，在验证码页面填写验证码。 \n5. 在重置页面输入并确认新密码后，提交后密码即被重置。 \n- 提示与注意事项：\n- 如果没收到邮件，检查垃圾邮箱或等待几分钟后再试；也可尝试重新发送。 \n- 若多次尝试无效或无法访问注册邮箱/手机号，请联系平台客服或管理员提供身份验证后协助重置。 \n- 设置新密码时建议使用长度足够、包含大小写字母、数字和特殊字符的密码，并避免复用其他网站的密码。', ...)
    #     ]
    # }

if __name__ == "__main__":
    asyncio.run(main())

最终，Agent 可以基于文档内容回答用户。

这里说明：MCP Tool 可以直接转换成 LangChain Agent 的工具；而 MCP Resource 和 Prompt 不一定需要转换成 Tool，它们通常由 Client 处理。如果希望 Agent 能够自主调用它们，才会额外封装成 Tool。

4.2.2 调用三方服务器

前面我们已经知道了 MCP 社区中，活跃着众多平台与技术人员，致力于为 MCP 用户提供交流互动、资源分享等服务。

这些网站汇集了丰富的 MCP 服务器信息，支持按人气排名、功能类型等多种条件筛选，帮助用户快速找到心仪的服务器。同时，它们也为开服者提供了展示成果、推广服务器的重要渠道，积极推动了 MCP 生态的繁荣与发展。

• 魔搭社区：https://modelscope.cn/mcp
• 阿里百炼：https://bailian.console.aliyun.com/cn-beijing?spm=a2c4g.11186623.0.0.33f5494ctCRYJv&tab=mcp#/mcp-market
• MCP 官方参考服务器：https://github.com/modelcontextprotocol/servers?tab=readme-overview-file#reference-servers

定义客户端来调用时间 MCP 服务器。

安装时间包：

pip install mcp-server-time

编码：

import asyncio
import json

from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.prompts import load_mcp_prompt
from langchain_mcp_adapters.tools import load_mcp_tools

async def main():
    client = MultiServerMCPClient(
        {
            "time": {
                "command": "python",
                "args": ["-m", "mcp_server_time"],
                "transport": "stdio",  # 本地子进程通信
            }
        },
    )

    # 使用 async with 创建并管理一个持久会话
    async with client.session("time") as session:  # 持久会话
        tools = await load_mcp_tools(session)  # 从该会话加载工具
        agent = create_agent("gpt-5-mini", tools)
        response = await agent.ainvoke(
            {"messages": [{"role": "user", "content": "北京时间的17：50分，对应的英国时间是？"}]}
        )
        print(response)

if __name__ == "__main__":
    asyncio.run(main())

结果：

{
    'messages': [
        HumanMessage(content='北京时间的17：50分，对应的英国时间是？', ...),
        AIMessage(content='',
            tool_calls=[{'name': 'get_current_time', 'args':
{'timezone': 'Asia/Shanghai'}, ...}], ...),
        ToolMessage(content=[{'type': 'text',
"datetime": "2026-04-09T11:40:29+08:00",\n  "timezone": "Asia/Shanghai",\n
...}], ...),
        AIMessage(content='',
            tool_calls=[{'name': 'convert_time', 'args': {
'target_timezone': 'Europe/London'}, ...}], ...),
        ToolMessage(content=[{'type': 'text', "source": {\n    "timezone":
"Asia/Shanghai",\n    "datetime": "2026-04-09T17:50:00+08:00",\n  ...}, ...}],
...),
        AIMessage(content='北京时间（Asia/Shanghai）17:50 对应的英国时间
（Europe/London）是 10:50（当天）。注意：这里英国处于夏令时（BST，UTC+1），因此与北京时间（UTC+8）相差7小时。', ...)
    ]
}

5. MCP 高级功能

5.1 工具栏截器

5.1.1 为什么需要工具栏截器？

MCP 服务器作为独立进程运行，它们无法直接访问 LangGraph 的运行时信息，例如存储（store）、上下文（context）或 Agent 状态（state）。

拦截器（Interceptors）填补了这一空白，它在 MCP 工具执行期间为你提供了访问这些运行时上下文的途径。同时，拦截器也提供了类似中间件的控制能力：可以修改请求、实现重试逻辑、动态添加请求头，甚至完全中断执行。

5.1.2 核心能力 1：访问运行时上下文

当 MCP 工具在 LangChain Agent（通过 create_agent 创建）内部使用时，拦截器会接收到 ToolRuntime 上下文。这使得你可以访问 tool_call_id 工具调用 ID、state 状态、config 配置和 store 存储，从而实现访问用户数据、持久化信息和控制 Agent 行为等强大模式。

典型场景：向 MCP 工具调用注入用户上下文

新增 user_id 参数

from fastmcp import FastMCP

mcp = FastMCP("Weather")

@mcp.tool()
async def get_weather(city: str, user_id: str) -> str:
    """获取天气"""
    return f"用户{user_id}查询: {city} 天气晴朗！"

if __name__ == "__main__":
    mcp.run(transport="streamable-http", port=8000)

假设你在调用 Agent 时传入了用户特定的配置（如用户 ID、API 密钥），你可以通过拦截器将这些信息动态注入到 MCP 工具的参数中。

import asyncio
import dataclasses
from dataclasses import dataclass

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

# 1. 定义上下文数据结构
@dataclass
class Context:
    user_id: str
    api_key: str

# 2. 编写拦截器，从运行时上下文读取信息并修改请求
async def inject_user_context(
    request: MCPToolCallRequest,
    handler,
):
    """将用户凭证注入到 MCP 工具调用中。"""
    runtime = request.runtime
    user_id = runtime.context.user_id  # 从上下文中读取 user_id
    api_key = runtime.context.api_key  # 从上下文中读取 api_key

    # 使用 override() 方法创建修改后的请求（遵循不可变模式）
    modified_request = request.override(
        args={**request.args, "user_id": user_id}
    )
    return await handler(modified_request)

async def main():
    client = MultiServerMCPClient(
        {
            "Weather": {
                "transport": "streamable-http",  # 基于 HTTP 的远程服务器
                # 务必在 8000 端口上启动天气服务器。
                "url": "http://localhost:8000/mcp",
            }
        },
        tool_interceptors=[inject_user_context]
    )

    tools = await client.get_tools()
    agent = create_agent(
        "gpt-5-mini",
        tools,
        context_schema=Context,
    )

    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "上海的天气怎么样？"}],
        context={"user_id": "user_123", "api_key": "sk-..."}
        }
    )
    print(result)

if __name__ == "__main__":
    asyncio.run(main())

打印结果：

{
    'messages': [
        HumanMessage(content='上海的天气怎么样？', additional_kwargs={},
response_metadata={}, id='5ebf0d22-e4bd-4495-958d-a66f0e356ce1'),
        AIMessage(content='', additional_kwargs={'refusal': None},
response_metadata={'token_usage': {'completion_tokens': 47, 'prompt_tokens':
131, 'total_tokens': 178, 'completion_tokens_details':
{'accepted_prediction_tokens': None, 'audio_tokens': 0, 'reasoning_tokens': 0,
'rejected_prediction_tokens': None}, 'prompt_tokens_details': {'audio_tokens':
0, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'gpt-5-mini-
2025-08-07', 'system_fingerprint': None, 'id': '8a1cmpl-
Ngeyv04dl9cLpzsN6X9gnMJ4bptc08a7f6aih-dce1s984b341fa-0', 'logprobs':
None}, id='89lcLpzsN6X9gnMJ4bptc08a7f6aih-dce1s984b341fa-0', tool_calls':
[{'name': 'get_weather', 'args': {'city': '上海'}, 'id': 'call_
8al8U38TgeCXdknP7OlcUavj0', 'type': 'tool_call', 'invalid_tool_calls': []}],
'usage_metadata': {'input_tokens': 131, 'output_tokens': 47, 'total_tokens': 178,
'input_token_details': {}, 'output_token_details': {}}),
        ToolMessage(content=[{'type': 'text', 'text': '用户user_123查询: 上海 天气晴朗！'}],
name='get_weather', id='0731ac8e-8d19-4f21-94d7-8e6d9f876504f9',
tool_call_id='call_8al8U38TgeCXdknP7OlcUavj0', artifact={'structured_content':
{'result': '用户user_123查询:上海 天气晴朗！'}}),
        AIMessage(content='查询到上海现在天气晴朗，如果需要我可以帮你查更详细的信息（例如温度、湿度、未来几天预报或穿衣建议），你想看哪一项？', additional_kwargs={'refusal': None},
response_metadata={'token_usage': {'completion_tokens': 51, 'prompt_tokens':
229, 'total_tokens': 229, 'completion_tokens_details':
{'accepted_prediction_tokens': None, 'audio_tokens': 0, 'reasoning_tokens': 0,
'rejected_prediction_tokens': None}, 'prompt_tokens_details': {'audio_tokens':
0, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'gpt-5-mini-
2025-08-07', 'system_fingerprint': None, 'id': 'chatcmpl-
dgemlc15rgnTwh9B1pT2dEUbV6P50-b8f15n78acea2014', 'logprobs': None},
id='ml15rgnTwh9B1pT2dEUbV6P50-b8f15n78acea2014', 'tool_calls': None,
'invalid_tool_calls': [], 'usage_metadata': {'input_tokens': 178, 'output_tokens': 51,
'total_tokens': 229, 'input_token_details': {}, 'output_token_details': {}})
    ]
}

访问 tool_call_id 工具调用 ID、state 状态和 store 存储见下表：

可以获取的元素	使用场景举例	获取方式
state	若用户未通过身份验证，则屏蔽敏感的 MCP 工具	runtime.state
store	使用 store 的偏好设置来个性化 MCP 工具的调用操作	runtime.store
Tool call ID	限制昂贵的 MCP 工具调用的次数	runtime.tool_call_id

1. state — 身份验证拦截

# 拦截器
async def interceptor(request: MCPToolCallRequest, handler):
    """若用户未通过身份验证，则屏蔽敏感的 MCP 工具"""
    runtime = request.runtime
    state = runtime.state
    is_authenticated = state.get("authenticated", False)

    if not is_authenticated:
        # 返回错误信息而非调用工具
        return ToolMessage(
            content="需要进行身份验证。请先登录。",
            tool_call_id=runtime.tool_call_id,
        )
    return await handler(request)

2. store — 个性化偏好设置

# 拦截器
async def interceptor(request: MCPToolCallRequest, handler):
    """使用 store 的偏好设置来个性化 MCP 工具的调用操作。"""
    runtime = request.runtime
    user_id = runtime.context.user_id
    store = runtime.store

    # 从存储中读取用户偏好设置
    prefs = store.get(("preferences",), user_id)
    if prefs and request.name == "search":
        # 应用用户所选的语言及结果限制
        modified_args = {
            **request.args,
            "language": prefs.value.get("language", "en"),
            "limit": prefs.value.get("result_limit", 10),
        }
        request = request.override(args=modified_args)
    return await handler(request)

3. Tool call ID — 速率限制

# 拦截器
async def interceptor(request: MCPToolCallRequest, handler):
    """限制昂贵的 MCP 工具调用的次数。"""
    runtime = request.runtime
    tool_call_id = runtime.tool_call_id

    # 检查速率限制（简化示例）
    if is_rate_limited(request.name):
        return ToolMessage(
            content="速率限制已超。请稍后再试。",
            tool_call_id=tool_call_id,
        )

    result = await handler(request)
    # 工具调用成功记录
    log_tool_execution(tool_call_id, request.name, success=True)
    return result

工具执行成功后提前结束 Agent 运行

使用 Command 的 goto="__end__" 可以立即终止图的执行。

async def end_on_success(
    request: MCPToolCallRequest,
    handler,
):
    """当 mark_complete 工具被调用时，结束整个 Agent 运行。"""
    result = await handler(request)

    if request.name == "mark_complete":
        return Command(
            update={"messages": [result], "status": "done"},
            goto="__end__",  # 特殊值，表示终止执行
        )

    return result

5.1.4 自定义拦截器

5.1.4.1 基本拦截器模式

一个拦截器接收 request 和 handler 两个参数。你可以在调用 handler 前后执行逻辑，或者完全跳过它。

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.interceptors import MCPToolCallRequest

async def logging_interceptor(
    request: MCPToolCallRequest,
    handler,
):
    """在执行前后记录工具调用日志。"""
    print(f"Calling tool: {request.name} with args: {request.args}")
    result = await handler(request)
    print(f"Tool {request.name} returned: {result}")
    return result

# 将拦截器列表传递给客户端
client = MultiServerMCPClient(
    {"math": {"transport": "stdio", "command": "python", "args":
["/path/to/server.py"]}},
    tool_interceptors=[logging_interceptor],
)

5.1.4.2 修改请求参数

使用 request.override() 方法创建修改后的请求。重要：请遵循不可变模式，不要直接修改原始请求对象。

async def double_args_interceptor(
    request: MCPToolCallRequest,
    handler,
):
    """在工具执行前，将所有数值参数翻倍。"""
    modified_args = {k: v * 2 for k, v in request.args.items()}
    modified_request = request.override(args=modified_args)
    return await handler(modified_request)

# 原始调用：add(a=2, b=3) 实际执行：add(a=4, b=6)

5.1.4.3 运行时动态修改 HTTP 请求头

你可以根据请求的具体内容（例如被调用的工具名称）来动态生成认证信息。

async def auth_header_interceptor(
    request: MCPToolCallRequest,
    handler,
):
    """根据被调用的工具名称，动态添加相应的认证头。"""
    token = get_token_for_tool(request.name)  # 假设这是获取令牌的自定义函数
    modified_request = request.override(
        headers={"Authorization": f"Bearer {token}"}
    )
    return await handler(modified_request)

5.1.4.4 错误处理与重试机制

使用拦截器捕获工具执行时的异常，并实现健壮的重试逻辑。

import asyncio

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 e:
            last_error = e
            if attempt < max_retries - 1:
                wait_time = delay * (2 ** attempt)  # 计算指数退避等待时间
                print(f"Tool {request.name} failed (attempt {attempt + 1}),
retrying in {wait_time}s...")
                await asyncio.sleep(wait_time)
    raise last_error  # 所有重试均失败后，抛出最后的异常

5.1.4.5 错误降级处理

捕获特定异常后，可以不抛出错误，而是返回一个兜底值，保证流程继续。

async def fallback_interceptor(
    request: MCPToolCallRequest,
    handler,
):
    """如果工具执行失败，返回一个降级响应。"""
    try:
        return await handler(request)
    except TimeoutError:
        return f"Tool {request.name} timed out. Please try again later."
    except ConnectionError:
        return f"Could not connect to {request.name} service. Using cached data."

5.1.4.6 组合多个拦截器

拦截器是包装工具执行的异步函数，多个拦截器会按照列表顺序组合执行，形成“洋葱”结构：列表中的第一个拦截器是最外层，最后一个是最内层，这和前面讲解的多个中间件的调用顺序是一样的。

async def outer_interceptor(request, handler):
    print("outer: before")
    result = await handler(request)
    print("outer: after")
    return result

async def inner_interceptor(request, handler):
    print("inner: before")
    result = await handler(request)
    print("inner: after")
    return result

client = MultiServerMCPClient(
    # 服务器配置...
    tool_interceptors=[outer_interceptor, inner_interceptor],
)

# 执行顺序：
# outer: before -> inner: before -> [实际工具执行] -> inner: after -> outer: after

5.1.5 总结

核心功能	关键点与原文描述
访问运行时上下文	通过 request.runtime 访问 state，config，store，context。原文案例展示了注入用户凭证到工具参数中。
状态更新与流程控制	返回 Command 对象。原文案例展示了更新状态并跳转（goto="summary_agent"）和提前结束执行（goto="__end__"）。
编写自定义拦截器	基本结构：async def func(request, handler)。 修改请求：使用 request.override()。 动态头：根据 request.name 修改 headers。 错误处理：包含指数退避重试和特定异常降级的完整代码。 组合执行：遵循洋葱模型顺序。

5.2 MCP 进度通知

在 MCP 中：

• 进度通知 允许客户端订阅 MCP 服务器在执行长时间运行工具时发送的进度更新。
• 客户端能接收进度通知的前提是服务端主动发送。在 FastMCP 中，通过工具函数内的 Context 对象实现。

适用场景：文件处理、大型数据集查询、模型推理等耗时操作，为用户提供实时反馈。

5.2.1 构建 MCP 服务器

5.2.1.1 Context 概述

Context 是 FastMCP 为工具函数提供的上下文对象，封装了进度报告、日志记录、用户交互等 MCP 协议功能。

获取方式：在工具函数签名中添加类型为 Context 的参数，FastMCP 会在调用时自动注入。

from fastmcp import Context

async def tool_func(..., ctx: Context) -> ...:

进度报告方法：report_progress

ctx.report_progress() 用于向客户端发送进度更新。

await ctx.report_progress(progress=50, total=100, message="正在处理第50项...")

参数	类型	说明
progress	float	当前进度值
total	float | None	总量值，可选
message	str | None	进度描述信息，可选

5.2.1.2 Context 其他功能

功能	方法/属性	说明
日志记录	ctx.debug(), ctx.info(), ctx.warning(), ctx.error()	向客户端发送日志消息
用户交互	ctx.elicit(message, schema)	请求用户提供结构化输入
资源访问	await ctx.read_resource(uri)	读取服务器注册的资源
LLM 采样	await ctx.sample(messages)	请求客户端 LLM 生成文本
会话/请求标识	ctx.session_id, ctx.request_id, ctx.client_id	获取当前上下文标识符

5.2.1.3 处理大文件时周期性报告进度

场景描述：假设有一个 MCP 服务器提供 process_large_file 工具，处理一个大文件时会周期性报告进度。

使用 FastMCP 创建服务器，在工具函数内通过 ctx.report_progress() 发送进度通知：

import asyncio
from fastmcp import FastMCP, Context

mcp = FastMCP("file_server")

@mcp.tool()
async def process_large_file(file_path: str, ctx: Context) -> str:
    """模拟处理大文件，分阶段发送进度更新。
    实际应用中，进度通知应在真正的耗时循环中调用。
    """
    # 定义总量（例如文件总行数）
    total = 1000.0

    # 阶段1：读取头部
    await ctx.report_progress(125, total, "正在读取 CSV 头部...")
    await asyncio.sleep(0.1)  # 模拟耗时

    # 阶段2：转换数据
    await ctx.report_progress(372, total, "正在转换第 1000 行...")
    await asyncio.sleep(0.1)

    # 阶段3：写入结果
    await ctx.report_progress(728, total, "正在写入结果文件...")
    await asyncio.sleep(0.1)

    # 阶段4：完成
    await ctx.report_progress(1000, total, "处理完成")
    await asyncio.sleep(0.1)

    return f"文件 {file_path} 已成功处理。"

if __name__ == "__main__":
    # 使用 HTTP 传输，监听 8000 端口
    mcp.run(transport="streamable-http")

5.2.2 构建 MCP 客户端

5.2.2.1 注册进度回调

客户端订阅 MCP 服务器的进度更新需通过 Callbacks 注册进度回调函数：

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.callbacks import Callbacks, CallbackContext

async def on_progress(
    progress: float,
    total: float | None,
    message: str | None,
    context: CallbackContext,
):
    """处理来自 MCP 服务器的进度更新"""
    pass

# 创建客户端并注入进度回调
client = MultiServerMCPClient(
    { # 服务器配置（示例占位）
        "math": {
            "transport": "stdio",
            "command": "python",
            "args": ["/path/to/math_server.py"],
        }
    },
    callbacks=Callbacks(on_progress=on_progress),  # 重点：注册进度回调
)

回调函数 on_progress 参数：

参数名	类型	说明
progress	float	当前进度值（具体含义由服务器定义，通常为已处理单元数量或比例）
total	float | None	总量值。若服务器未提供总量，则为 None，此时只能展示绝对进度
message	str | None	服务器发送的进度描述文本，如 “正在处理第3项…”
context	CallbackContext	包含当前调用上下文的元数据（服务器名称、工具名称等）

CallbackContext 结构：

class CallbackContext:
    server_name: str     # 发送该进度通知的 MCP 服务器名称。
    tool_name: str       # 当前正在执行的工具名称（仅在工具调用期间有效），
                         # 其他场景下可能为空。

5.2.2.2 构建 MCP 客户端：接受服务端进度报告

import asyncio

from langchain_mcp_adapters.callbacks import Callbacks, CallbackContext
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.agents import create_agent

# 定义进度回调
async def on_progress(
    progress: float,
    total: float | None,
    message: str | None,
    context: CallbackContext,
):
    """处理来自 MCP 服务器的进度更新信息。"""
    if total:
        percent = progress / total * 100
        print(f"📊 {context.tool_name} 已完成 {percent:.1f}%: {message}")
    else:
        print(f"📊 {context.tool_name} 进度 {progress}: {message}")

async def main():
    client = MultiServerMCPClient(
        {
            "file_server": {
                "transport": "http",
                "url": "http://localhost:8000/mcp",
            }
        },
        callbacks=Callbacks(on_progress=on_progress)
    )

    tools = await client.get_tools()
    agent = create_agent("gpt-5-mini", tools)
    result = await agent.ainvoke({
        "messages": [{"role": "user", "content": "请处理大文件 report.csv"}]
    })
    print(result)

if __name__ == "__main__":
    asyncio.run(main())

输出：

📊 process_large_file 已完成 12.5%: 正在读取 CSV 头部...
📊 process_large_file 已完成 37.2%: 正在转换第 1000 行...
📊 process_large_file 已完成 72.8%: 正在写入结果文件...
📊 process_large_file 已完成 100.0%: 处理完成
{
    'messages': [
        HumanMessage(content='请处理大文件 report.csv', ...),
        AIMessage(content='', tool_calls=[{'name': 'process_large_file',
'args': {'file_path': 'report.csv'}, ...}),
        ToolMessage(content=[{'type': 'text', 'text': '文件 report.csv 已成功处理。', ...}),
        AIMessage(content='我已处理了文件 report.csv。', ...)
    ]
}

• 进度通知是单向推送，客户端无法通过回调干预工具执行。
• CallbackContext 提供了 server_name 和 tool_name，便于区分多个服务或工具。
• 服务端 Context 除进度报告外，还支持日志、用户交互、资源访问等 MCP 高级功能。

5.3 MCP 日志记录

MCP协议允许服务器在运行过程中发送日志通知给客户端。这些日志可用于监控、调试或记录工具执行过程中的关键事件。

5.3.1 构建 MCP 服务器

该服务器提供一个简单的工具，并在执行过程中发送日志通知。

from fastmcp import FastMCP, Context

# 创建 MCP 服务器实例
mcp = FastMCP("FetchData")

@mcp.tool()
async def fetch_data(query: str, ctx: Context) -> str:
    await ctx.info(f"开始处理查询{query}")
    await asyncio.sleep(0.5)

    await ctx.debug("正在连接数据库...")
    await asyncio.sleep(0.5)

    await ctx.warning("数据库连接异常")
    await asyncio.sleep(0.5)

    await ctx.error("查询失败，返回默认数据")
    await asyncio.sleep(0.5)



if __name__ == "__main__":
    # 使用 HTTP 传输，监听 8000 端口
    mcp.run(transport="streamable-http")

说明：

• 工具函数 fetch_data 接收一个 Context 对象 ctx，通过 ctx.log 发送不同级别的日志。
• 日志级别支持 debug、info、warning、error 等，符合 MCP 日志规范。

5.3.2 构建 MCP 客户端

使用 MultiServerMCPClient 时，可以向其传递一个 callbacks 的 on_logging_message 参数，其中包含我们自定义的日志处理函数。通过 Callbacks 订阅并打印服务端发送的日志消息。

import asyncio
from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.callbacks import Callbacks, CallbackContext
from mcp.types import LoggingMessageNotificationParams

async def on_logging_message(
    params: LoggingMessageNotificationParams,
    context: CallbackContext,
):
    """处理来自 MCP 服务器的日志消息。"""
    print(f"[{context.server_name}] {params.level}: {params.data}")

async def main():
    # 配置客户端，连接本地 stdio 服务端
    client = MultiServerMCPClient(
        {
            "fetch_data": {
                "transport": "http",
                "url": "http://localhost:8000/mcp",
            }
        },
        callbacks=Callbacks(on_logging_message=on_logging_message), # 关键配置
    )

    # 获取工具并调用（触发服务端日志）
    tools = await client.get_tools()
    agent = create_agent("gpt-5-mini", tools)
    result = await agent.ainvoke({
        "messages": [{"role": "user", "content": "查询123订单的详细数据"}]
    })
    print(result)

if __name__ == "__main__":
    asyncio.run(main())

参数详解

参数 / 属性	类型	说明
params	LoggingMessageNotificationParams	包含日志的级别（level）和数据（data）
context	CallbackContext	提供上下文信息，如服务器名称（server_name）和工具名称（tool_name）

打印结果

📝math - info: {'msg': '开始处理查询打印运行日志', 'extra': None}
📝math - debug: {'msg': '正在连接数据库...', 'extra': None}
📝math - warning: {'msg': '数据库连接异常', 'extra': None}
📝math - error: {'msg': '查询失败，返回默认数据', 'extra': None}
{'messages': [HumanMessage(content='请调工具打印运行日志', additional_kwargs={}, response_metadata={}, id='028461da-eff8-41ee-b46d-b4373aae80d4'), AIMessage(content='', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 20, 'prompt_tokens': 4469, 'total_tokens': 4489, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_provider': 'openai', 'model_name': 'gpt-5.4', 'system_fingerprint': None, 'id': 'chatcmpl-6b1708fe-dd2f-4ef0-b5e4-b9ad0', 'finish_reason': 'tool_calls', 'logprobs': None}, id='lc_run--019fb6cc-d609-75a0-8d74-b12d31a0d410-0', tool_calls=[{'name': 'fetch_data', 'args': {'query': '打印运行日志'}, 'id': 'call_DbzXXWEREpfgLCZdPBesl7zi', 'type': 'tool_call'}], invalid_tool_calls=[], usage_metadata={'input_tokens': 4469, 'output_tokens': 20, 'total_tokens': 4489, 'input_token_details': {}, 'output_token_details': {}}), ToolMessage(content=[{'type': 'text', 'text': '查询 打印运行日志 失败，已返回默认数据。', 'id': 'lc_12c7dba4-0582-4ec7-9eaf-3fce33c9b42e'}], name='fetch_data', id='bbf79c6d-5efc-4534-8645-b8e42626ed6b', tool_call_id='call_DbzXXWEREpfgLCZdPBesl7zi', artifact={'structured_content': {'result': '查询 打印运行日志 失败，已返回默认数据。'}}), AIMessage(content='已调用工具尝试打印运行日志，结果是：\n- `fetch_data` 返回：`查询 打印运行日志 失败，已返回默认数据。`\n\n如果你愿意，我也可以继续帮你：\n- 排查为什么日志查询失败\n- 换个查询词再试一次\n- 模拟一段运行日志输出', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 75, 'prompt_tokens': 4513, 'total_tokens': 4588, 'completion_tokens_details': None, 'prompt_tokens_details': None}, 'model_provider': 'openai', 'model_name': 'gpt-5.4', 'system_fingerprint': None, 'id': 'chatcmpl-94ad5b05-b517-45e2-ac82-3ae73', 'finish_reason': 'stop', 'logprobs': None}, id='lc_run--019fb6cc-e699-7383-a522-0a70c11d7b36-0', tool_calls=[], invalid_tool_calls=[], usage_metadata={'input_tokens': 4513, 'output_tokens': 75, 'total_tokens': 4588, 'input_token_details': {}, 'output_token_details': {}})]}

5.4 引导式输入（Elicitation）

根据 MCP 规范，Elicitation 允许 MCP 服务器在工具执行过程中向用户请求额外输入，而不是要求所有输入在调用前一次性提供。服务器可以交互式地按需询问信息。

核心作用：实现动态、分步的信息收集，提升用户体验。

5.4.1 构建 MCP 服务器

在 MCP 服务器中，使用 ctx.elicit() 方法向客户端发起引导请求，并定义所需数据的 Schema。
案例场景：用户资料创建服务

from pydantic import BaseModel
from fastmcp import Context, FastMCP

server = FastMCP("Profile")

class UserDetails(BaseModel):
    email: str
    age: int

@server.tool()
async def create_profile(name: str, ctx: Context) -> str:
    """创建用户资料，elicit 询问"""
    result = await ctx.elicit(
        message=f"请为 {name} 的个人资料提供详细信息：",
        schema=UserDetails
    )
    if result.action == "accept" and result.data:
        return f"为 {name} 创建个人资料：email={result.data.email}, age={result.data.age}"
    if result.action == "decline":
        return f"用户拒绝了为 {name} 创建完整信息。"
    return "个人资料创建已取消。"

if __name__ == "__main__":
    server.run(transport="streamable-http")

说明：

• ctx.elicit(message, schema) 向客户端发起一个模式化的输入请求。
• result.action 表示用户的响应动作（accept / decline / cancel）。
• result.data 包含符合 UserDetails 结构的数据。

5.4.2 构建 MCP 客户端

客户端通过向 MultiServerMCPClient 提供 on_elicitation 回调来处理服务器的引导请求。
案例场景：注册回调并返回模拟用户数据

import asyncio
from langchain.agents import create_agent
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.callbacks import Callbacks, CallbackContext
from mcp.shared.context import RequestContext
from mcp.types import ElicitRequestParams, ElicitResult

async def on_elicitation(
    mcp_context: RequestContext,
    params: ElicitRequestParams,
    context: CallbackContext,
) -> ElicitResult:
    """处理来自 MCP 服务器的引导请求"""
    # 实际应用中，此处应根据 params.message 和 params.requestedSchema 提示真实用户输入
    return ElicitResult(
        action="accept",
        content={"email": "user@example.com", "age": "25"},
    )

async def main():
    # 配置客户端
    client = MultiServerMCPClient(
        {
            "profile": {
                "url": "http://localhost:8000/mcp",
                "transport": "http",
            }
        },
        callbacks=Callbacks(on_elicitation=on_elicitation),
    )

    tools = await client.get_tools()
    agent = create_agent("gpt-5-mini", tools)

    result = await agent.ainvoke({
        "messages": [{"role": "user", "content": "创建小明的个人资料"}]
    })
    print(result)

if __name__ == "__main__":
    asyncio.run(main())

说明：

• 回调接收 ElicitRequestParams，其中包含 message 和 requestedSchema。
• 必须返回一个 ElicitResult 对象，指示用户的操作和提供的数据。

响应动作

ElicitResult 支持三种动作，对应不同用户意图：

动作	描述	使用示例
accept	用户提供了有效输入，需在 content 中附带数据	ElicitResult(action="accept", content={"email": "user@example.com", "age": "25"})
decline	用户拒绝提供所请求的信息	ElicitResult(action="decline")
cancel	用户完全取消当前操作	ElicitResult(action="cancel")

打印结果

{
  'messages': [
    HumanMessage(content='创建小明的个人资料', additional_kwargs={}, response_metadata={}, id='5310741c-0dc8-453d-9a10-8c5ac946e2a7'),
    AIMessage(content='', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 25, 'prompt_tokens': 138, 'total_tokens': 163, 'completion_tokens_details': {'accepted_prediction_tokens': None, 'reasoning_tokens': 0, 'rejected_prediction_tokens': None}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'gpt-5-mini-2025-08-07', 'system_fingerprint': None, 'id': 'chatcmpl-DSxtNzNHPVCEgf0RpgIymmxnActJjzZ7', 'finish_reason': 'tool_calls', 'logprobs': None}, id='lc_run--019d75ad-246e-7c81-ab9f-b8999b149c1a-0', tool_calls=[{'name': 'create_profile', 'args': {'name': '小明'}, 'id': 'call_N7duUwSKmogGwSwObHF6bXPF', 'type': 'tool_call'}], invalid_tool_calls=[], usage_metadata={'input_tokens': 138, 'output_tokens': 25, 'total_tokens': 163, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}),
    ToolMessage(content="{'type': 'text', 'text': '为 小明 创建了个人资料：email=user@example.com, age=25'}", name='create_profile', id='aeece4970-7a2a-4b64-8fe7-842e893d4eb3', tool_call_id='call_N7duUwSKmogGwSwObHF6bXPF', artifact={'structured_content': {'result': '为 小明 创建了个人资料：email=user@example.com, age=25'}}),
    AIMessage(content='已为"小明"创建了个人资料。以下是已记录的信息：\n\n- 姓名：小明\n- 电子邮件：user@example.com\n- 年龄：25\n\n如果你想补充更多信息（例如性别、地址、职业、兴趣爱好、电话等），告诉我需要添加的字段和具体内容，我会帮你更新。', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 87, 'prompt_tokens': 183, 'total_tokens': 270, 'completion_tokens_details': {'accepted_prediction_tokens': None, 'reasoning_tokens': 0, 'rejected_prediction_tokens': None}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'gpt-5-mini-2025-08-07', 'system_fingerprint': None, 'id': 'chatcmpl-DSxtNzNHPVCEgfAB5cjwRUmgY1IN3NQi', 'finish_reason': 'stop', 'logprobs': None}, id='lc_run--019d75ad-3ae3-7922-99ab-c5ba0e7494f5-0', tool_calls=[], invalid_tool_calls=[], usage_metadata={'input_tokens': 183, 'output_tokens': 87, 'total_tokens': 270, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})
  ]
}