1. Context #

本章主要介绍如何使用 Context 在 MCP 服务器端与客户端之间传递上下文信息。你将学会：

• 如何在工具函数中使用 Context 发送日志、报告进度、读取资源，并访问请求信息；
• 客户端如何通过日志回调和消息处理器观察服务器端的日志与进度；
• 如何调用工具，并观察运行中的日志与进度输出。

2. 服务器 #

新建 C:\mcp-project\context_server.py：

# context_server.py
# 逐行中文注释：在工具函数中使用 Context 发送日志与进度，并读取资源与访问请求信息

# 导入 FastMCP 与 Context 类型
from mcp.server.fastmcp import FastMCP, Context

# 导入 ServerSession，用于给 Context 指定会话类型参数
from mcp.server.session import ServerSession

# 创建 FastMCP 服务器实例，指定服务器名称（客户端可见）
mcp = FastMCP(name="Context Server")


# 注册一个静态资源，供工具在执行时通过 Context 读取
@mcp.resource("data://message")
# 定义静态资源函数，返回字符串内容
def static_message() -> str:
    """示例静态资源：用于演示在工具中通过 Context 读取资源。"""
    # 返回资源内容
    return "这是一条来自 data://message 的资源内容。"


# 注册工具函数：演示发送日志、报告进度、读取资源，并访问请求信息
@mcp.tool()
# 定义异步工具函数，支持任务名、步数和上下文参数
async def long_running_task(task_name: str, ctx: Context, steps: int = 5) -> str:
    """执行一个带进度的任务，期间输出日志、读取资源并返回总结。"""
    # 1) 保护性判断：如果未注入 Context，直接返回
    if ctx is None:
        # 未注入 Context 时，返回提示信息
        return "未注入 Context，无法演示日志/进度/资源读取。"

    # 2) 发送不同级别的日志，便于客户端侧观察
    await ctx.debug(f"开始执行任务：{task_name}")
    await ctx.info("初始化任务环境……")

    # 3) 读取静态资源内容（演示 Context 的资源读取能力）
    #    read_resource 返回 GetResourceResult，其中 contents 里通常有 TextContent
    resource_result = await ctx.read_resource("data://message")
    # 初始化资源预览内容为空
    resource_preview = "<empty>"
    # 遍历资源内容块
    for block in resource_result:
        resource_preview = getattr(block, "content", "<no-text>")
        break
    # 发送日志，说明资源读取成功
    await ctx.info(f"读取资源成功：{resource_preview}")

    # 4) 循环执行步骤并报告进度
    for i in range(steps):
        # 计算当前进度（0~1之间的浮点数）
        progress = (i + 1) / steps
        # 发送进度通知（客户端可在自定义处理器中观察到）
        await ctx.report_progress(
            progress=progress, total=1.0, message=f"Step {i + 1}/{steps}"
        )
        # 也记录到日志，便于在客户端侧统一从日志回调中看到
        await ctx.debug(f"进度：{i + 1}/{steps}")

    # 5) 访问请求相关信息（如 request_id），并返回总结
    req_id = ctx.request_id
    # 发送日志，说明任务完成
    await ctx.info(f"任务完成：{task_name}（request_id={req_id}）")

    # 返回任务总结字符串，包含资源摘要
    return f"任务 '{task_name}' 完成，读取到的资源摘要：{resource_preview}"


# 主入口：以 stdio 方式运行服务器
if __name__ == "__main__":
    # 启动服务器，使用 stdio 作为通信方式
    mcp.run(transport="stdio")

要点：

• 在工具函数签名中加入 ctx: Context[ServerSession, None] 参数，即可自动注入 Context。
• 使用 await ctx.debug/info/warning/error(...) 发送日志；使用 await ctx.report_progress(...) 发送进度。
• await ctx.read_resource(uri) 可在工具中读取资源内容。
• 可通过 ctx.request_id 获取请求 ID。

3. 客户端 #

新建 C:\mcp-project\test_client_context.py：

# test_client_context.py
# 逐行中文注释：连接到 context_server.py，设置日志回调与消息处理器，观察日志与进度并调用工具

# 导入异步IO库，用于支持异步编程
import asyncio

# 导入os库，用于处理文件和路径操作
import os

# 从mcp模块导入客户端会话、Stdio服务器参数和类型定义
from mcp import ClientSession, StdioServerParameters, types

# 从mcp.client.stdio导入stdio客户端工厂方法
from mcp.client.stdio import stdio_client


# 定义异步日志回调函数，当服务器发送日志消息时会被调用
async def on_logging(params: types.LoggingMessageNotificationParams) -> None:
    # 获取日志级别，若不存在则默认为"info"
    level = getattr(params, "level", "info")
    # 获取日志数据内容，若不存在则为None
    data = getattr(params, "data", None)
    # 获取日志记录器名称，若不存在则为None
    logger = getattr(params, "logger", None)
    # 打印日志信息，若无数据则显示<no-data>
    print(
        f"[LOG][{level}]",
        str(data) if data is not None else "<no-data>",
        f"(logger={logger})",
    )


# 定义通用消息处理器，用于处理服务器的进度通知等消息
async def on_message(message) -> None:
    # 获取消息的方法名，若不存在则为None
    method = getattr(message, "method", None)
    # 判断是否为进度通知
    if method == "notifications/progress":
        # 获取通知参数
        params = getattr(message, "params", None)
        if params is not None:
            # 获取当前进度
            progress = getattr(params, "progress", None)
            # 获取总进度
            total = getattr(params, "total", None)
            # 获取进度消息内容
            msg = getattr(params, "message", None)
            # 打印进度信息
            print(f"[PROGRESS] {progress}/{total} - {msg}")
    # 其他类型的通知在本示例中忽略


# 定义主异步函数
async def main() -> None:
    # 1) 计算服务器脚本的绝对路径
    base_dir = os.path.dirname(os.path.abspath(__file__))
    server_path = os.path.join(base_dir, "context_server.py")

    # 2) 配置以stdio方式启动服务器的参数
    server_params = StdioServerParameters(
        command="python",
        args=[server_path],
        env={},
    )

    # 3) 建立到stdio服务器的连接，并创建客户端会话（设置日志回调与消息处理器）
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(
            read,
            write,
            logging_callback=on_logging,  # 设置日志回调函数
            message_handler=on_message,  # 设置通用消息处理器
        ) as session:
            # 4) 完成初始化握手
            await session.initialize()

            # 5) 调用长任务工具，传入参数，观察运行中的日志与进度输出
            result = await session.call_tool(
                "long_running_task", {"task_name": "demo", "steps": 3}
            )

            # 6) 打印工具返回的结果（提取可读文本内容）
            texts = []
            for block in result.content:
                if isinstance(block, types.TextContent):
                    texts.append(block.text)
            print(
                "[RESULT]",
                " | ".join(texts) or str(getattr(result, "structuredContent", None)),
            )


# 判断是否为主模块入口
if __name__ == "__main__":
    # 在事件循环中运行主异步函数
    asyncio.run(main())

说明：

• 通过 logging_callback 接收服务器端日志，便于在终端中实时查看。
• 通过 message_handler 观察到 notifications/progress 进度通知（本例做了通用健壮处理）。
• 可调用 await session.set_logging_level(...) 提高日志详细程度（如 debug）。

4. 运行与验证 #

cd C:\mcp-project
call .venv\Scripts\activate
python test_client_context.py

预期输出（示例，顺序可能有差异）：

[LOG][debug] 开始执行任务：demo (logger=None)
[LOG][info] 初始化任务环境…… (logger=None)
[LOG][info] 读取资源成功：这是一条来自 data://message 的资源内容。 (logger=None)
[LOG][debug] 进度：1/3 (logger=None)
[LOG][debug] 进度：2/3 (logger=None)
[LOG][debug] 进度：3/3 (logger=None)
[LOG][info] 任务完成：demo（request_id=1） (logger=None)
[RESULT] 任务 'demo' 完成，读取到的资源摘要：这是一条来自 data://message 的资源内容。

如果能看到类似输出，表示你已掌握 Context 的日志与进度能力，并能在客户端侧实时观察。