MCP 进度通知

接下来我们继续学习 MCP 中一项非常关键的能力：MCP 进度通知。

我们先拆分两端：一端是 MCP 服务端 Server，内部定义了业务工具；另一端是 MCP 客户端 Client。我们设定一个典型业务场景：服务端提供一个处理大文件的工具，这个操作属于耗时任务，整个流程会持续好几秒，比如执行 1 秒完成 10% 进度、再经过 1 秒推进到 50%、最后执行到 100% 全部完成。

大家都清楚客户端调用服务端工具的完整流程，但有一个痛点：如果单次工具调用整体耗时 5 秒，在这 5 秒的等待过程里，客户端没有任何反馈，完全不知道服务端工具处理到哪一步了。那有没有办法让客户端实时获取工具执行进度？ 答案是有的，MCP 协议专门提供了Context上下文对象来解决这个实时通信问题。

所以，在 MCP 中：

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

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

构建 MCP 服务器

Context 概述

当客户端和服务端建立连接后，如果双方需要在工具执行的过程中实时双向通信，就依靠Context上下文对象实现。

Context 是 FastMCP 为工具函数提供的上下文对象，封装了进度报告、日志记录、用户交互等 MCP 协议功能。 获取方式：这个对象不需要客户端手动传递，只需要在服务端中的工具函数签名中添加类型为 Context 的参数，FastMCP 会在工具函数被调用时自动注入到工具函数的入参中。

我们写工具函数时，在普通业务参数末尾追加ctx: Context参数即可拿到这个对象。

from fastmcp import Context

mcp = FastMCP("name")

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

这里我们先谈一下整体的操作框架，然后再一一深入学习：

服务端发送进度：ctx.report_progress ()

客户端想要实时获取工具处理进度，服务端在工具执行的对应节点，调用ctx.report_progress()方法，就能主动向客户端推送进度通知。 调用时可以传递三类信息：

  • 当前进度数值；

  • 总进度总量；

  • 一段文本描述，说明当前正在执行的操作。

举个例子：总量设置为 1000，当前执行到 125，就代表整体完成 12.5%，同时附带文字说明 “正在读取 CSV 头部”，这条进度消息会实时推送给客户端。

客户端接收进度：注册进度回调函数

客户端想要接收服务端推送的进度消息，在初始化MultiServerMCPClient创建客户端实例时，除了填写服务端连接配置，还需要额外配置Callbacks回调集合。

我们需要在回调集合中指定on_progress参数，绑定自定义的进度接收回调函数，所有服务端推送的进度通知，都会统一进入这个回调函数处理。

完整通信流程梳理：

  • 服务端工具执行到对应节点，调用ctx.report_progress()推送进度；

  • 进度消息通过 MCP 长连接传输到客户端；

  • 客户端预先注册的on_progress回调函数捕获这条消息；

  • 我们在回调函数内自定义逻辑，打印进度、展示进度条、更新前端界面等。

这里给大家拓展一个知识点：Context不只是用来推送进度通知，后续我们还要学习的MCP 日志记录、引导式用户交互输入这两大能力，全部都是依靠Context对象内部封装的不同方法实现。

这三类能力底层逻辑完全一致：服务端通过Context主动推送消息，客户端通过注册对应类型的回调函数接收消息，只是回调绑定的参数名称、Context 调用的方法不同。

进度报告方法：report_progress

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

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

我们先拆解服务端推送进度的核心方法ctx.report_progress()，它一共接收三个参数：

  • progress：浮点型float，代表当前已完成的进度数值；

  • total：浮点型float或空值None，代表任务总进度总量；通过progress / total * 100可以计算出完成百分比；

  • message：字符串或空值，用来填写当前进度对应的文字描述，可选参数。

如果只需要推送进度，只需要掌握这个方法即可，Context里其他日志、交互相关方法我们后面讲到对应功能再细说。

参数 类型 说明
progress float 当前进度值
total float | None 总量值，可选
message str | None 进度描述信息，可选
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 获取当前上下文标识符
构建 MCP 服务器：处理大文件时周期性报告进度

场景描述：假设有一个 MCP 服务器提供 process_large_file 工具，处理一个大文件时会周期性报告进度。使用 FastMCP 创建服务器，在工具函数内通过 ctx.report_progress() 发送进度通知。

具体步骤：

  • 导入FastMCP、Context、asyncio；

  • 实例化 FastMCP 服务，命名为file_server；

  • 定义工具process_large_file，入参包含业务参数file_path和自动注入的ctx: Context；

  • 在工具内部模拟分阶段处理大文件，设置总进度total=1000，每个阶段调用一次await ctx.report_progress()推送进度，使用asyncio.sleep(0.1)模拟耗时操作；

  • 最后配置streamable-http传输协议，默认监听 8000 端口启动服务。

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")

代码逻辑说明：真实生产环境中，report_progress()要放在真实的循环、耗时读写逻辑内，我们这里只是分阶段模拟，每一段进度推送都要加await，因为该方法是异步函数。

构建 MCP 客户端

注册进度回调

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

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

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

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

客户端接收进度的回调函数固定有 4 个入参，前三个和服务端report_progress传递的参数一一对应，第四个是元数据上下文：

  • progress：对应服务端传入的当前进度数值；

  • total：对应服务端传入的总进度总量，允许为None；

  • message：对应服务端传入的进度描述文本；

  • context: CallbackContext：回调上下文元数据，记录这条进度通知来自哪一台 MCP 服务、具体是哪一个工具推送的消息，用来区分多服务、多工具场景下的进度来源。

回调函数 on_progress 参数：

参数名 类型 说明
progress float | None 当前进度值（具体含义由服务器定义，通常为已处理单元数量或比例）
total float | None 总量值。若服务器未提供总量，则为 None，此时只能展示绝对进度
message str | None 服务器发送的进度描述文本，如 "正在处理第 3 项..."
context CallbackContext 包含当前调用上下文的元数据（服务名称、工具名称等） 
class CallbackContext:
    server_name: str   # 发送该进度通知的 MCP 服务器名称
    tool_name: str     # 当前正在执行的工具名称（仅在工具调用期间有效）
    # 其他场景下可能为空。
构建 MCP 客户端：接受服务端进度报告

导入依赖：MultiServerMCPClient、Callbacks、CallbackContext、create_agent、asyncio；

自定义异步回调函数on_progress，处理进度打印逻辑：

  • 判断total是否存在，存在则计算百分比格式化打印；

  • total为空时，直接打印原始进度数值；

  • 通过context.tool_name获取推送进度的工具名称，拼接打印日志；

创建客户端实例，填写服务端 HTTP 连接地址，通过callbacks=Callbacks(on_progress=on_progress)绑定进度回调；

拉取服务端所有工具，创建 Agent，发起用户提问 “请处理大文件 report.csv”；

执行agent.ainvoke()发起调用，最终打印完整执行结果。

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 | None,
    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": [("user", "请处理大文件 report.csv")]}
    )
    print(result)

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

实操踩坑说明

代理环境问题：本地 HTTP 服务端口 8000，如果开启了全局代理，会出现 502 无法连接，需要配置对应环境变量关闭代理；

大模型不调用工具：如果运行后大模型直接给出文字回答、没有触发process_large_file工具，需要在用户提问里增加提示，强制大模型必须调用文件处理工具；

进度打印过快：可以把模拟耗时sleep(0.1)调整为sleep(0.5)，方便逐条观察进度输出。

输出：

启动服务端后再运行客户端，Agent 发起工具调用，服务端会分阶段推送进度消息，客户端回调函数逐条打印日志：

📦 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="文件 report.csv 已成功处理。", ...),
        AIMessage(content="我已处理了文件 report.csv。", ...)
    ]
}

全部进度推送完成后，工具返回处理成功的文本，完整消息列表会打印在控制台，包含用户提问、AI 工具调用决策、工具返回结果、最终 AI 回复。

注意：

  • 进度通知是单向推送，客户端无法通过回调干预工具执行。

  • CallbackContext 提供了 server_name 和 tool_name，便于区分多个服务或工具。

  • 服务端 Context 除进度报告外，还支持日志、用户交互、资源访问等 MCP 高级功能。

总结

进度通知的完整实现逻辑可以总结为两点：

  • 服务端：依靠 FastMCP 自动注入的Context对象，调用ctx.report_progress()主动推送实时进度；

  • 客户端：初始化客户端时注册on_progress回调函数，捕获并处理所有进度推送消息。

依靠这套机制，我们就能在耗时工具执行期间，给用户提供实时进度反馈，适配文件处理、大数据查询、长文本推理等各类慢任务场景。

到这里，MCP 进度通知的全部概念、参数、代码实操就全部讲解完毕，接下来我们继续学习 MCP 日志记录相关能力。

MCP 日志记录

接下来我们学习 MCP 的日志记录功能，这项功能依旧是围绕 MCP 服务端工具展开的。

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

部署在服务端的业务工具运行时，可以主动输出各类日志内容，这些日志不会只留在服务端本地，而是依托工具自动注入的Context上下文对象，实时推送给客户端；客户端只需要提前注册对应的回调函数，就能捕获、展示服务端工具打印的全部日志，完整还原工具完整执行流程。

简单来说：服务端工具输出的日志通过Context推送，客户端靠回调接收，就能直观看到工具每一步的执行细节。

下面我们分服务端、客户端两段完整编写代码实操演示。

构建 MCP 服务器

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

我们新建服务端文件，自定义一个数据查询工具，命名为fetch_data。

  • 工具入参：第一个业务参数query，用来接收用户传入的查询关键词；末尾追加自动注入的ctx: Context上下文对象；函数返回值为字符串str。

  • 工具返回逻辑：模拟查询无匹配数据，最终返回文本结果: 您查询的 '{query}' 没有匹配的数据。

  • 工具注释标注功能：数据获取工具，执行过程中会推送不同级别的日志给客户端，日志区分不同等级，适配调试、告警、正常信息等不同场景。

服务端推送日志核心逻辑：Context 分级日志方法

依靠工具内置的ctx上下文对象，调用对应分级方法即可异步推送日志，所有日志推送方法都是异步函数，调用必须加await：

  • info 普通信息日志：await ctx.info(f"开始处理查询: {query}")，用于打印正常流程节点，比如任务启动、任务完成；

  • debug 调试日志：await ctx.debug("正在连接数据源...")，用于打印底层执行细节，开发调试阶段使用；

  • warning 告警日志：await ctx.warning("数据源响应较慢，请稍候...")，用于打印风险、缓慢、异常预警类信息；

结尾补充一条 info 日志标记任务收尾：await ctx.info(f"查询 '{query}' 处理完成")。

我们可以用asyncio.sleep(1)模拟接口、数据源耗时等待，完整还原真实业务的执行节奏。写完工具后，配置streamable-http传输协议，默认 8000 端口启动 MCP 服务端。

from fastmcp import FastMCP, Context

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

@mcp.tool()
async def fetch_data(query: str, ctx: Context) -> str:
    """数据获取工具，执行过程中会发送不同级别的日志。"""
    # 发送 info 级别日志
    await ctx.info(f"开始处理查询: {query}")

    # 模拟一些处理步骤
    await ctx.debug("正在连接数据源...")

    await ctx.warning("数据源响应较慢，请稍候...")

    # 模拟返回结果
    await ctx.info(f"查询 '{query}' 处理完成")
    return f"结果: 您查询的 '{query}' 没有匹配数据。"

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

说明：

  • 工具函数 fetch_data 接收一个 Context 对象 ctx，通过 ctx.log 发送不同级别的日志。

  • 日志级别支持 debug、info、warning、error 等，符合 MCP 日志规范。

构建 MCP 客户端

使用 MultiServerMCPClient 时，可以向其传递一个 callbacks 的 on_logging_message 参数，其中包含我们自定义的日志处理函数。通过 Callbacks 订阅并打印服务端发送的日志消息，这样才能接收服务端推送的日志消息，整体流程和进度通知的回调设计高度相似，仅回调标识、入参存在区别。

  • 客户端创建MultiServerMCPClient，填入服务端 HTTP 地址http://localhost:8000/mcp；

  • 之前配置进度通知用的是on_progress回调，本次日志推送专用回调参数为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())

日志回调函数入参说明

自定义异步回调函数on_logging_message固定接收两个参数：

params: LoggingMessageNotificationParams：日志核心载体对象，内置两个关键属性

class LoggingMessageNotificationParams(NotificationParams):
    """Parameters for logging message notifications."""

    level: LoggingLevel
    """The severity of this log message."""
    logger: str | None = None
    """An optional name of the logger issuing this message."""
    data: Any
    """
    The data to be logged, such as a string message or an object. Any JSON serializable
    type is allowed here.
    """
    model_config = ConfigDict(extra="allow")

  • params.level：日志级别，对应服务端info/debug/warning/error；

  • params.data：日志正文内容，也就是服务端 ctx 打印的文本；

context: CallbackContext：回调元数据上下文，和进度通知回调的第二个参数完全一致，用来标识这条日志来自哪一台 MCP 服务、哪一个工具，多服务多工具场景下用来区分日志来源。

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

我们在回调函数内格式化打印日志，直观展示全部信息：

  • 先打印服务名称context.server_name，区分日志来源服务；

  • 再打印日志等级params.level，区分日志类型；

  • 最后打印日志正文params.data，展示完整日志内容。

这套日志捕获能力不止局限于控制台打印，生产环境中可以拓展业务逻辑：把日志存入数据库持久化、推送到前端页面实时展示，让终端用户也能看到工具后台完整执行流程，根据业务需求自由拓展。

下面我们开始测试：

  1. 先启动 MCP 服务端，保持后台运行；

  2. 运行客户端代码，构造用户提问：请查询123订单的详细数据；

  3. Agent 会自动识别需求，调用服务端fetch_data工具；

  4. 工具分步执行，每调用一次ctx.xxx()推送日志，客户端回调函数就会同步打印一条日志，按顺序输出 debug、info、warning 各级日志；

  5. 工具执行完毕返回无匹配数据，Agent 整合结果回复用户。

注意：运行时如果控制台没有立刻输出日志，大概率是大模型决策逻辑问题：部分情况下大模型识别查询无数据，会重复调用工具重试，等待几秒后就会完整打印全部日志； 完整消息链路顺序：用户HumanMessage提问 → AI 生成工具调用AIMessage → 服务端推送全量日志 → 工具返回ToolMessage无匹配数据 → AI 生成最终回复。

打印结果：

1 [fetch_data] info: {'msg': '开始处理查询: 查询订单 123 详细数据', 'extra': None}
2 [fetch_data] debug: {'msg': '正在连接数据源...', 'extra': None}
3 [fetch_data] warning: {'msg': '数据源响应较慢，请稍候...', 'extra': None}
4 [fetch_data] info: {'msg': "查询 '查询订单 123 详细数据' 处理完成", 'extra': None}

MCP 日志通知的底层实现和进度通知完全统一：

服务端：依靠 FastMCP 自动注入的Context上下文对象，调用分级日志方法（ctx.info/ctx.debug/ctx.warning）异步推送日志；

客户端：初始化MultiServerMCPClient时，在Callbacks中绑定on_logging_message回调函数，捕获、处理所有服务端推送的日志消息； 通过这套机制，我们可以完整监控、调试、记录 MCP 工具全生命周期执行事件，用于线上监控、线下调试排查问题。到这里 MCP 日志通知功能全部讲解完毕。

引导式输入（Elicitation）

接下来我们学习 MCP 最后一项核心能力：引导式输入。

先对比前面学过的进度通知、日志通知，梳理核心区别：进度、日志通知全都是 MCP 服务端 Server 主动单向给客户端 Client 推送消息；但引导式输入是双向交互，能让客户端主动向服务端回传自定义数据。

这里有一个关键限制大家一定要记牢：客户端不能主动自发给服务端发消息，所有客户端向服务端的数据回传，必须由服务端主动发起请求引导，客户端收到引导请求后，才能回复对应数据，这也是 “引导式” 这个名字的由来。

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

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

我们用「创建用户个人资料」工具完整演示交互逻辑：

  • 用户向 Agent 发送提问：帮我创建小明的个人资料；

  • 大模型判断需要调用create_profile创建资料工具，把入参name="小明"传给 MCP 服务端工具；

  • 工具执行时发现仅拿到姓名，缺少必填的邮箱、年龄两个字段，无法完成资料创建；

  • 服务端主动发起引导：依靠工具自动注入的Context上下文对象，向客户端推送引导请求，告知客户端当前缺失邮箱、年龄，需要补充；

  • 客户端通过预先注册的专属回调函数捕获这条引导消息；

  • 客户端在回调内选择三种响应策略之一，把对应结果回传给服务端；

  • 服务端工具收到客户端的响应后，根据不同策略执行对应分支逻辑，完成工具流程。

客户端三种标准响应策略（MCP 协议固定三选一，无第四种）

MCP 协议规定客户端收到引导请求后，只能返回三类操作，对应三种业务逻辑：

accept 同意提供数据：客户端主动补齐缺失参数（邮箱、年龄），把结构化数据一并回传给服务端，服务端拿到完整字段后正常创建完整用户资料；

decline 拒绝提供参数:客户端拒绝补充邮箱、年龄，仅允许使用已有的姓名创建极简版资料，不中断整个工具流程；

cancel 完全取消操作:直接终止本次创建用户资料的全部流程，不生成任何用户信息，直接结束工具执行。

三者核心区分：decline 只是拒绝补充额外字段，任务仍可简易完成；cancel 是直接废弃本次整个任务，完全终止工具。

构建 MCP 服务器

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

案例场景：用户资料创建服务

我们新建用户资料服务，定义create_profile工具，实现主动引导客户端补全信息。

依赖导入：FastMCP、Context用于 MCP 服务与上下文；pydantic.BaseModel定义需要用户补充的结构化数据规范UserDetails，包含email: str邮箱、age: int年龄两个字段；

工具入参：业务参数name，末尾自动注入ctx: Context上下文对象；

工具注释：创建用户资料，缺失信息通过 elicitation 引导式输入获取；

核心引导方法：await ctx.elicit()，这是 Context 内置专门用于发起引导请求的异步方法，两个核心传参：

  • message：给客户端展示的提示文本，告知用户需要补充什么信息；

  • response_type：传入我们定义好的UserDetails模型，告诉客户端需要返回的数据结构、字段类型；

接收返回结果：result = await ctx.elicit(...)，返回值result包含两大核心属性：

  • result.action：客户端选择的操作（accept/decline/cancel）；

  • result.data：仅当 action 为accept时存在，是客户端回传的结构化数据，可直接读取data.email、data.age；

分支逻辑判断：

  • 如果action == "accept"且result.data存在：拼接姓名、邮箱、年龄，返回完整资料创建成功文本；

  • 如果action == "decline"：返回提示，仅用姓名创建最简资料；

  • 剩余cancel场景：返回提示，本次资料创建操作已取消；

最后配置streamable-http传输协议，8000 端口启动服务。

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:
    """创建用户资料，缺失信息通过 elicitation 询问"""
    result = await ctx.elicit(
        message=f"请为 {name} 的个人资料提供详细信息: ",
        response_type=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 结构的数据。

构建 MCP 客户端

客户端通过向 MultiServerMCPClient 提供 on_elicitaiton 回调来处理服务器的引导请求。

案例场景：注册回调并返回模拟用户数据

客户端想要接收服务端的引导请求、回传响应，必须注册专属回调函数on_elicitaiton，和进度、日志回调配置方式统一，仅回调标识、入参不同。

1. 引导回调函数固定三个入参

async def on_elicitaiton(
    mcp_context: RequestContext,
    params: ElicitRequestParams,
    context: CallbackContext,
) -> ElicitResult:
    pass

mcp_context: RequestContext：MCP 本次请求全局上下文，存放请求 ID、会话元数据，业务开发中极少用到；

from dataclasses import dataclass, field
from typing import Generic, Any

@dataclass
class RequestContext(Generic[SessionT, LifespanContextT, RequestT]):
    """
    MCP 单次请求上下文对象
    贯穿一次客户端工具调用/消息交互全生命周期，承载会话、请求元数据、流式通道、扩展实验能力等全部上下文信息
    核心用途：给耗时任务提供进度推送、会话读写、请求生命周期控制能力
    """
    # 当前请求唯一标识，用于全链路追踪、日志打点、区分并发请求
    request_id: RequestId
    # 请求携带的元数据（客户端附加参数、调用来源、超时配置等），无元数据时为 None
    meta: RequestParams.Meta | None
    # 当前请求绑定的会话实例，存储客户端长连接会话状态、权限、缓存等会话级数据
    session: SessionT
    # 服务全局生命周期上下文，服务启动时初始化，整个服务运行期间复用，存放全局共享资源（数据库连接池、公共配置等）
    lifespan_context: LifespanContextT

    # 实验性功能扩展对象
    # 注：此处标注为 Any 是为规避循环导入问题：
    # 真实运行类型为 mcp.server.experimental.request_context.Experimental
    # 若在此处直接导入会触发导入环：mcp.server.__init__ → fastmcp → tools → 当前模块
    # 服务启动后会由 Server 内部自动赋值为 Experimental 实例
    experimental: Any = field(default=None)

    # 当前原始请求报文对象，解析后的客户端完整请求体，无则为 None
    request: RequestT | None = None

    # SSE 标准流式通道关闭回调函数
    # 当需要主动断开服务端推送进度/日志的 SSE 长流时调用，用于清理连接、释放资源
    close_sse_stream: CloseSSEStreamCallback | None = None

    # 独立模式 SSE 流关闭回调函数
    # 区分标准会话 SSE，用于单独创建、不绑定主会话的独立流式通道销毁逻辑
    close_standalone_sse_stream: CloseSSEStreamCallback | None = None

params: ElicitRequestParams：核心参数，承载服务端发来的全部引导信息：

  • params.message：服务端推送的提示文字；

  • params.requestedSchema：服务端要求补充的数据结构（我们定义的 UserDetails 规范）；

class ElicitRequestURLParams(RequestParams):
    """Parameters for URL mode elicitation requests.

    URL mode directs users to external URLs for sensitive out-of-band interactions
    like OAuth flows, credential collection, or payment processing.
    """

    mode: Literal["url"] = "url"
    """The elicitation mode (always "url" for this type)."""

    message: str
    """The message to present to the user explaining why the interaction is needed."""

    url: str
    """The URL that the user should navigate to."""

    elicitationId: str
    """
    The ID of the elicitation, which must be unique within the context of the server.
    The client MUST treat this ID as an opaque value.
    """

    model_config = ConfigDict(extra="allow")


# Union type for elicitation request parameters
ElicitRequestParams: TypeAlias = ElicitRequestURLParams | ElicitRequestFormParams
"""Parameters for elicitation requests - either form or URL mode."""

class ElicitRequestFormParams(RequestParams):
    """Parameters for form mode elicitation requests.

    Form mode collects non-sensitive information from the user via an in-band form
    rendered by the client.
    """

    mode: Literal["form"] = "form"
    """The elicitation mode (always "form" for this type)."""

    message: str
    """The message to present to the user describing what information is being requested."""

    requestedSchema: ElicitRequestedSchema
    """
    A restricted subset of JSON Schema defining the structure of expected response.
    Only top-level properties are allowed, without nesting.
    """

    model_config = ConfigDict(extra="allow")

context: CallbackContext：回调元数据上下文，标识这条引导请求来自哪一台 MCP 服务、哪一个工具，多服务多工具场景区分来源。

2. 回调返回规范

回调函数必须返回ElicitResult对象，仅支持两个字段：

class ElicitResult(Result):
    """The client's response to an elicitation request."""

    action: Literal["accept", "decline", "cancel"]
    """
    The user action in response to the elicitation.
    - "accept": User submitted the form/confirmed the action (or consented to URL navigation)
    - "decline": User explicitly declined the action
    - "cancel": User dismissed without making an explicit choice
    """

    content: dict[str, str | int | float | bool | list[str] | None] | None = None
    """
    The submitted form data, only present when action is "accept" in form mode.
    Contains values matching the requested schema. Values can be strings, integers,
    booleans, or arrays of strings.
    For URL mode, this field is omitted.
    """

  • action：三选一，accept/decline/cancel；

  • content：仅当 action 为accept时生效，传入符合 Schema 规范的字典数据，存放用户补充的字段；如果是 decline、cancel，content 传值无任何作用。

3. 客户端完整流程

初始化MultiServerMCPClient，填入服务端 HTTP 地址http://localhost:8000/mcp，在Callbacks中绑定on_elicitaiton=on_elicitaiton引导回调；

拉取服务端全部工具，创建 LangChain Agent；

构造用户提问：创建小明的个人资料，调用agent.ainvoke()发起请求；

Agent 触发create_profile工具，服务端推送引导请求，客户端回调捕获消息；

回调内返回action="accept"，模拟补充邮箱user@example.com、年龄25；

服务端收到数据，拼接完整信息返回工具结果，控制台打印完整消息链路。

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_elicitaiton(
    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():
    # 配置客户端，连接本地 stdio 服务端
    client = MultiServerMCPClient(
        {
            "profile": {
                "url": "http://localhost:8000/mcp",
                "transport": "http",
            }
        },
        callbacks=Callbacks(on_elicitaiton=on_elicitaiton),
    )
    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")

打印结果：

控制台完整输出消息顺序：

  • 用户HumanMessage提问；

  • AI 生成AIMessage，决策调用create_profile工具，传入name="小明"；

  • 服务端工具执行，通过ctx.elicit()发起引导请求，客户端回调捕获并返回补充的邮箱、年龄；

  • 工具拿到完整字段，生成ToolMessage返回创建成功文本；

  • AI 整合全部信息，生成最终回复告知用户资料创建完成，并展示已记录的姓名、邮箱、年龄。

{
    'messages': [
        HumanMessage(content='创建小明的个人资料', additional_kwargs={}, id='5310741c-0dc8-453d-9a10-8c5ac946e2a7'),
        AIMessage(content='', additional_kwargs={'response_metadata': {'token_usage': {'completion_tokens': 25, 'prompt_tokens': 138, 'total_tokens': 163, 'completion_tokens_details': {'accepted_prediction_tokens': None, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'gpt-5-mini-DS25H-08-07', 'system_fingerprint': None, 'finish_reason': 'tool_calls', 'logprobs': None}}, id='llm-run--019d714b-74a-76c1-8c4-1984b34f1a7', 'tool_calls': [{'name': 'create_profile', 'args': {'name': '小明'}, 'id': 'call_N7dduwSKmgwSpuHb6kXPF', '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='为 小明 创建了个人资料: email=user@example.com, age=25', name='create_profile', id='lc-32aac82d-4419-05-4a7-81d4-26c597650479', tool_call_id='call_N7dduwSKmgwSpuHb6kXPF', artifact={'structured_content': [{'result': '为 小明 创建了个人资料: email=user@example.com, age=25'}]}),
        AIMessage(content='已为"小明"创建了个人资料。以下是已记录的信息：\n姓名：小明\n电子邮件：user@example.com\n年龄：25\n如果你想补充更多信息（例如性别、地址、职业、兴趣爱好、电话等），告诉我需要添加的字段和具体内容，我会帮你更新。', additional_kwargs={'response_metadata': {'token_usage': {'completion_tokens': 87, 'prompt_tokens': 183, 'total_tokens': 270, 'completion_tokens_details': {'accepted_prediction_tokens': None, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_provider': 'openai', 'model_name': 'gpt-5-mini-DS25H-08-07', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None}}, id='llm-run--019d714b-74a-76c1-8c4-57cae20e14', 'invalid_tool_calls': [], 'usage_metadata': {'input_tokens': 163, 'output_tokens': 87, 'total_tokens': 270, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}}]
}

MCP 整体总结

到这里 MCP 整套知识点全部讲解完毕，我们完整梳理这阶段的脉络：

MCP 基础定义：MCP 是专门为大模型设计的通信协议，打通模型与外部工具、资源、提示词的交互通道；

服务端开发（FastMCP）：使用 FastMCP 快速搭建 MCP 服务，自定义工具、资源、提示词模板；社区也提供大量现成开箱即用的 MCP 服务；

客户端开发（langchain-mcp-adapters）：LangChain 配套封装MultiServerMCPClient客户端，和 LangChain Agent 无缝适配，可批量拉取服务端工具、资源、提示词；

工具拦截器：拦截 MCP 工具调用链路，实现前置参数注入、权限校验、后置状态更新、流程跳转、重试降级、日志打印等自定义业务逻辑；

服务端与客户端双向通信三大能力（全部依托Context上下文对象实现）

  • 进度通知：服务端主动推送工具执行进度，客户端注册on_progress回调接收；

  • 日志记录：服务端分级推送工具运行日志（debug/info/warning/error），客户端注册on_logging_message回调捕获日志；

  • 引导式输入：服务端主动发起引导请求，客户端通过on_elicitaiton回调回传结构化补充数据，支持同意、拒绝、取消三种交互策略。

整套 MCP 体系依托Context上下文实现服务端、客户端双向实时通信，搭配各类回调函数处理推送消息，拓展性极强，能够适配文件处理、数据查询、多轮信息采集、线上监控调试等各类复杂大模型业务场景。

# python # 数据库 # 网络
Logo
MCP技术社区

欢迎加入 MCP 技术社区！与志同道合者携手前行，一同解锁 MCP 技术的无限可能！

更多推荐

Vibe Coding 真的很难

四个星期做了四个项目——AI新闻日报、汽车资讯日报、听障学习小程序、Skill MCP管理工具。表面看是一个字"累"，但深入体会是"敬畏"。真实上手后发现四大难题：上下文记忆模糊导致有序性难保证、Agent之间改来改去连贯性缺失、本地与外部安全性隐患、跨设备可用性依赖配置。这四座大山，才是Vibe Coding真正的修炼场。正如沙龙上听到的：AI时代的大爆发还没来，至少不是现在。共勉。

avatar MCP技术社区

Python 开发者进阶指南，这套大模型课值不值得深钻

本文深度解析 Python 开发者进阶大模型架构的实战路径。课程涵盖 LangGraph 智能体编排、MCP 协议标准互联，以及 Qwen3 与 DeepSeek 模型的微调私有化部署。通过多模态融合与 RAG 架构演进，助力开发者从脚本编写者蜕变为具备企业级落地能力的大模型应用专家。

avatar MCP技术社区
cover

MCP 实战：从零搭建 Model Context Protocol Server，让 AI 连接你的工具

avatar MCP技术社区