摘要：MCP通知机制实战，实现长任务进度上报（progress通知）和实时状态推送（resource更新通知），附异步任务管理代码和客户端订阅实现。

通知机制实战 长任务进度上报与实时状态推送

写 MCP 专栏到最后一篇了。前面 23 篇讲的都是请求-响应模式，客户端发请求，服务器返回结果。但实际项目里有一类场景这个模式搞不定，就是长任务。比如批量处理 1000 个文件，可能要跑几分钟，用户不想干等着，想看到进度条。MCP 的通知机制就是解决这个问题的。这篇我来讲讲进度上报、日志通知和资源变更通知这三种通知机制的实现。

MCP 通知机制原理

MCP 的通知机制基于 JSON-RPC 2.0 的 notification 消息。跟普通请求不同的是，通知不需要响应，是单向的。服务器可以随时发通知给客户端，客户端也可以发通知给服务器。

通知机制有三种主要类型。进度通知通过 notifications/progress 发送，携带进度 token、当前进度值和可选的总量。日志通知通过 notifications/message 发送，携带日志级别和日志数据。资源变更通知通过 notifications/resources/list_changed 和 notifications/resources/updated 发送，分别表示资源列表变化和单个资源内容变化。

进度通知的核心是 progress token。客户端在发起请求时，在请求的 _meta 字段里带上 progressToken，这个 token 可以是字符串或整数。服务器在处理这个请求的过程中，用这个 token 发送进度通知。请求完成后，进度通知必须停止。

// 客户端请求时携带 progress token
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "batch-process",
    "arguments": { "files": ["a.txt", "b.txt"] },
    "_meta": {
      "progressToken": "task-abc-123"
    }
  }
}

// 服务器发送进度通知
{
  "jsonrpc": "2.0",
  "method": "notifications/progress",
  "params": {
    "progressToken": "task-abc-123",
    "progress": 50,
    "total": 100,
    "message": "已处理 50/100 个文件"
  }
}

进度上报实现

在 Python SDK 里实现进度上报需要直接操作底层的消息发送。FastMCP 提供了 Context 对象来简化这个过程。下面是一个完整的进度上报工具实现

import asyncio
import json
from fastmcp import FastMCP, Context

# 创建服务器实例
mcp = FastMCP("notification-demo-server")


@mcp.tool
async def batch_process_files(
    file_paths: list[str],
    ctx: Context,
) -> str:
    """
    批量处理文件，实时上报进度
    
    Args:
        file_paths: 要处理的文件路径列表
        ctx: FastMCP 自动注入的上下文对象
    """
    # 参数校验
    if not file_paths:
        return "错误: 文件列表为空"
    
    total = len(file_paths)
    results = []
    
    # 逐个处理文件，每处理一个就上报进度
    for i, file_path in enumerate(file_paths):
        # 模拟文件处理
        # 实际项目中这里可能是读取、解析、转换等操作
        await asyncio.sleep(0.5)
        
        # 模拟处理结果
        # 80% 成功，20% 失败
        success = (i % 5 != 3)
        results.append({
            "file": file_path,
            "status": "success" if success else "failed",
            "index": i,
        })
        
        # 发送进度通知
        # progress 必须递增，total 可选
        # message 是人类可读的进度描述
        await ctx.report_progress(
            progress=i + 1,
            total=total,
            message=f"已处理 {i + 1}/{total}: {file_path}",
        )
        
        # 同时发送日志通知
        # 日志级别遵循 syslog 标准
        if success:
            await ctx.log.info(
                f"文件处理成功: {file_path}",
                logger="batch-processor",
            )
        else:
            await ctx.log.warning(
                f"文件处理失败: {file_path}",
                logger="batch-processor",
            )
    
    # 汇总结果
    success_count = sum(1 for r in results if r["status"] == "success")
    failed_count = total - success_count
    
    return json.dumps({
        "total": total,
        "success": success_count,
        "failed": failed_count,
        "results": results,
    }, ensure_ascii=False)

这里有个独家踩坑经验。我一开始用 ctx.report_progress 上报进度，但在某些客户端上完全看不到进度条。排查后发现是 progress token 的问题。有些客户端只在 tools/call 请求的 _meta 里带了 progressToken 时才会监听进度通知。如果客户端没带 token，你发的进度通知会被静默丢弃。这跟日志通知类似，客户端需要先发 logging/setLevel 请求才能收到日志。

所以在实现进度上报时，最好做一个降级处理。如果检测到客户端没有携带 progress token，就把进度信息放到日志通知里，作为备选方案。

@mcp.tool
async def safe_batch_process(
    file_paths: list[str],
    ctx: Context,
) -> str:
    """
    批量处理文件，带降级的进度上报
    
    Args:
        file_paths: 文件路径列表
        ctx: 上下文对象
    """
    total = len(file_paths)
    
    for i, path in enumerate(file_paths):
        # 模拟处理
        await asyncio.sleep(0.3)
        
        # 尝试通过 report_progress 上报
        # 如果客户端没带 progress token，这会被忽略
        await ctx.report_progress(
            progress=i + 1,
            total=total,
        )
        
        # 同时通过日志上报作为降级方案
        # 即使 progress token 不存在，日志仍然有效
        # 前提是客户端设置了日志级别
        await ctx.log.info(
            f"进度: {i + 1}/{total} - 正在处理 {path}",
            logger="batch-processor",
        )
    
    return json.dumps({
        "total": total,
        "message": "批量处理完成",
    }, ensure_ascii=False)

日志通知实现

日志通知是 MCP 的另一个重要通知类型。服务器通过 notifications/message 发送结构化日志给客户端。日志级别遵循 RFC 5424 的 syslog 标准，从 debug 到 emergency 共 8 个级别。

@mcp.tool
async def deploy_application(
    app_name: str,
    version: str,
    ctx: Context,
) -> str:
    """
    模拟应用部署流程，展示多级别日志通知
    
    Args:
        app_name: 应用名称
        version: 版本号
        ctx: 上下文对象
    """
    # 部署流程的各个阶段
    stages = [
        ("debug", "开始部署流程", "deploy-pipeline"),
        ("info", f"拉取镜像 {app_name}:{version}", "deploy-pipeline"),
        ("info", "停止旧版本容器", "deploy-pipeline"),
        ("notice", "数据库迁移开始", "database"),
        ("info", "数据库迁移完成", "database"),
        ("info", "启动新版本容器", "deploy-pipeline"),
        ("debug", "健康检查中...", "health-check"),
        ("warning", "健康检查第1次失败，等待重试", "health-check"),
        ("info", "健康检查第2次成功", "health-check"),
        ("info", "部署完成，清理旧容器", "deploy-pipeline"),
    ]
    
    # 逐个阶段执行并发送日志
    for level, message, logger in stages:
        # 模拟每个阶段的耗时
        await asyncio.sleep(0.3)
        
        # 根据级别发送不同级别的日志通知
        if level == "debug":
            await ctx.log.debug(message, logger=logger)
        elif level == "info":
            await ctx.log.info(message, logger=logger)
        elif level == "notice":
            await ctx.log.info(message, logger=logger)
        elif level == "warning":
            await ctx.log.warning(message, logger=logger)
        elif level == "error":
            await ctx.log.error(message, logger=logger)
    
    return json.dumps({
        "app": app_name,
        "version": version,
        "status": "deployed",
        "message": f"{app_name} {version} 部署成功",
    }, ensure_ascii=False)

日志级别对比

级别数值用途示例
debug7详细调试信息函数入口出口、变量值
info6常规信息操作进度更新
notice5正常但重要的事件配置变更
warning4警告条件弃用功能使用
error3错误条件操作失败
critical2严重条件系统组件故障
alert1需立即行动数据损坏
emergency0系统不可用完全系统故障

客户端通过 logging/setLevel 请求设置最低日志级别，服务器只发送该级别及以上的日志。比如客户端设置了 info，服务器发的 debug 日志会被过滤掉。

资源变更通知

资源变更通知有两种。notifications/resources/list_changed 表示资源列表变了，比如新增了文件或删除了表。notifications/resources/updated 表示某个已知的资源内容变了，客户端应该重新读取。

import asyncio
import time
from fastmcp import FastMCP, Context

mcp = FastMCP("resource-notification-server")

# 模拟的数据存储
# 实际项目中可能是数据库或文件系统
_data_store = {
    "sensor-001": {"value": 25.0, "timestamp": ""},
    "sensor-002": {"value": 60.0, "timestamp": ""},
}

# 后台任务: 定期更新传感器数据并通知
async def sensor_update_loop():
    """后台任务: 每 3 秒更新传感器数据"""
    while True:
        await asyncio.sleep(3)
        
        # 更新数据
        for sensor_id in _data_store:
            # 模拟数据变化
            old_value = _data_store[sensor_id]["value"]
            new_value = old_value + (old_value * 0.05 * (1 if time.time() % 2 > 1 else -1))
            _data_store[sensor_id]["value"] = round(new_value, 2)
            _data_store[sensor_id]["timestamp"] = time.strftime("%H:%M:%S")
        
        # 在实际实现中，这里需要通过 server 的通知机制
        # 发送 notifications/resources/updated 给已订阅的客户端
        # FastMCP 内部会处理订阅管理


@mcp.tool
async def get_sensor_data(sensor_id: str, ctx: Context) -> str:
    """
    获取传感器当前数据
    
    Args:
        sensor_id: 传感器 ID
        ctx: 上下文对象
    """
    if sensor_id not in _data_store:
        return f"错误: 传感器 {sensor_id} 不存在"
    
    data = _data_store[sensor_id]
    
    # 发送日志通知告知读取操作
    await ctx.log.info(
        f"读取传感器 {sensor_id}: {data['value']}",
        logger="sensor-service",
    )
    
    return json.dumps({
        "sensor_id": sensor_id,
        "value": data["value"],
        "timestamp": data["timestamp"],
    }, ensure_ascii=False)


@mcp.tool
async def list_sensors(ctx: Context) -> str:
    """列出所有传感器"""
    sensors = [
        {"id": sid, "value": data["value"]}
        for sid, data in _data_store.items()
    ]
    return json.dumps(sensors, ensure_ascii=False)

长任务实战 文件批量处理

把前面学到的进度上报、日志通知和错误处理组合起来，实现一个完整的文件批量处理工具。这个工具会实时上报处理进度，处理失败时发警告日志，完成后发完成通知。

import asyncio
import json
import os
import time
from fastmcp import FastMCP, Context

mcp = FastMCP("file-batch-processor")


@mcp.tool
async def batch_convert_files(
    input_dir: str,
    output_dir: str,
    operation: str,
    ctx: Context,
) -> str:
    """
    批量转换文件，实时上报进度
    
    Args:
        input_dir: 输入目录路径
        output_dir: 输出目录路径
        operation: 操作类型: uppercase/lowercase/count
        ctx: 上下文对象
    """
    # ===== 阶段 1: 初始化和参数校验 =====
    await ctx.log.info(
        f"开始批量转换: {input_dir} -> {output_dir}, 操作: {operation}",
        logger="file-converter",
    )
    
    # 检查输入目录
    if not os.path.isdir(input_dir):
        return f"错误: 输入目录不存在: {input_dir}"
    
    # 检查操作类型
    valid_ops = {"uppercase", "lowercase", "count"}
    if operation not in valid_ops:
        return f"错误: 不支持的操作 '{operation}'，可选: {valid_ops}"
    
    # 创建输出目录
    os.makedirs(output_dir, exist_ok=True)
    
    # ===== 阶段 2: 收集文件列表 =====
    # 递归扫描所有文本文件
    all_files = []
    for root, dirs, files in os.walk(input_dir):
        for f in files:
            # 只处理文本文件
            if f.endswith(('.txt', '.md', '.json', '.csv', '.log')):
                all_files.append(os.path.join(root, f))
    
    total = len(all_files)
    if total == 0:
        await ctx.log.warning(
            "没有找到可处理的文本文件",
            logger="file-converter",
        )
        return "没有找到可处理的文件"
    
    await ctx.log.info(
        f"找到 {total} 个文件待处理",
        logger="file-converter",
    )
    
    # ===== 阶段 3: 逐个处理文件 =====
    results = []
    success_count = 0
    fail_count = 0
    start_time = time.time()
    
    for i, file_path in enumerate(all_files):
        try:
            # 读取文件内容
            with open(file_path, 'r', encoding='utf-8') as f:
                content = f.read()
            
            # 执行转换操作
            if operation == "uppercase":
                converted = content.upper()
            elif operation == "lowercase":
                converted = content.lower()
            elif operation == "count":
                # count 操作只统计不转换
                converted = json.dumps({
                    "chars": len(content),
                    "lines": content.count('\n') + 1,
                    "words": len(content.split()),
                })
            
            # 构造输出路径，保持相对路径结构
            rel_path = os.path.relpath(file_path, input_dir)
            output_path = os.path.join(output_dir, rel_path)
            
            # 创建输出目录
            os.makedirs(os.path.dirname(output_path), exist_ok=True)
            
            # 写入输出文件
            with open(output_path, 'w', encoding='utf-8') as f:
                f.write(converted)
            
            results.append({
                "file": rel_path,
                "status": "success",
                "size": len(converted),
            })
            success_count += 1
            
        except Exception as e:
            # 处理失败，记录错误
            results.append({
                "file": file_path,
                "status": "failed",
                "error": str(e),
            })
            fail_count += 1
            
            # 发送警告日志
            await ctx.log.warning(
                f"文件处理失败: {file_path} - {e}",
                logger="file-converter",
            )
        
        # 上报进度
        # 每处理完一个文件就上报一次
        # progress 值必须递增
        await ctx.report_progress(
            progress=i + 1,
            total=total,
            message=f"已处理 {i + 1}/{total} ({success_count}成功, {fail_count}失败)",
        )
        
        # 每处理 10 个文件发一条 info 日志
        # 避免日志过于频繁
        if (i + 1) % 10 == 0:
            elapsed = time.time() - start_time
            speed = (i + 1) / elapsed if elapsed > 0 else 0
            await ctx.log.info(
                f"进度 {i + 1}/{total}, 速度 {speed:.1f} 文件/秒",
                logger="file-converter",
            )
    
    # ===== 阶段 4: 汇总结果 =====
    elapsed = time.time() - start_time
    
    await ctx.log.info(
        f"批量转换完成: {success_count} 成功, {fail_count} 失败, 耗时 {elapsed:.1f}秒",
        logger="file-converter",
    )
    
    return json.dumps({
        "operation": operation,
        "input_dir": input_dir,
        "output_dir": output_dir,
        "total_files": total,
        "success": success_count,
        "failed": fail_count,
        "elapsed_seconds": round(elapsed, 2),
        "results": results,
    }, ensure_ascii=False, indent=2)

通知机制的客户端配合

服务器发的通知需要客户端正确配合才能发挥作用。客户端需要做三件事。第一，在发起 tools/call 请求时，如果想要进度通知，要在 _meta 里带上 progressToken。第二，发送 logging/setLevel 请求设置日志级别。第三，通过 resources/subscribe 订阅资源变更。

# 客户端示例: 如何正确接收通知
import asyncio
from fastmcp import Client

async def client_demo():
    """客户端接收通知的示例"""
    # 连接到 MCP 服务器
    async with Client("python", ["server.py"]) as client:
        # 1. 设置日志级别为 info
        # 服务器只会发送 info 及以上级别的日志
        await client.set_logging_level("info")
        
        # 2. 调用工具并监听进度
        # progress_token 参数告诉服务器我们想要进度通知
        progress_events = []
        
        # 调用批量处理工具
        # on_progress 回调会在收到进度通知时被调用
        result = await client.call_tool(
            "batch_convert_files",
            {
                "input_dir": "./input",
                "output_dir": "./output",
                "operation": "uppercase",
            },
            # 进度回调函数
            on_progress=lambda p: progress_events.append(p),
        )
        
        print(f"处理结果: {result}")
        print(f"收到 {len(progress_events)} 条进度通知")
        
        # 3. 订阅资源变更
        # 订阅后，资源内容变化时会收到通知
        await client.subscribe_resource("system://stats")
        
        # 持续监听资源变更
        # 实际应用中这部分通常在事件循环里自动处理
        await asyncio.sleep(10)

asyncio.run(client_demo())

完整代码

下面是一个完整的通知机制服务器，包含进度上报、日志通知和资源变更

# file: notification_server.py
# 完整的 MCP 通知机制服务器
# 运行方式: python notification_server.py
# 依赖安装: pip install fastmcp

import asyncio
import json
import time
import os
from fastmcp import FastMCP, Context

# 创建服务器
mcp = FastMCP("notification-demo")

# ===== 数据存储 =====
# 模拟的任务状态
_task_status = {}

# 模拟的监控数据
_monitor_data = {
    "cpu": 0.0,
    "memory": 0.0,
    "requests_per_sec": 0,
    "last_update": "",
}


# ===== 工具 1: 带进度上报的批量处理 =====
@mcp.tool
async def process_with_progress(
    task_count: int,
    ctx: Context,
) -> str:
    """
    模拟批量任务处理，展示进度上报
    
    Args:
        task_count: 要处理的任务数量
        ctx: 上下文对象，自动注入
    """
    # 参数校验
    if task_count < 1 or task_count > 500:
        return f"错误: 任务数量必须在 1-500 之间，当前 {task_count}"
    
    # 记录任务开始
    task_id = f"task-{int(time.time())}"
    _task_status[task_id] = {"status": "running", "progress": 0}
    
    await ctx.log.info(
        f"任务 {task_id} 开始，共 {task_count} 个子任务",
        logger="task-manager",
    )
    
    start_time = time.time()
    results = []
    
    # 逐个处理子任务
    for i in range(task_count):
        # 模拟子任务处理
        await asyncio.sleep(0.1)
        
        # 模拟 5% 失败率
        success = (i % 20 != 0) or (i == 0)
        results.append({
            "index": i,
            "status": "success" if success else "failed",
        })
        
        # 更新任务状态
        _task_status[task_id]["progress"] = (i + 1) / task_count
        
        # 上报进度
        await ctx.report_progress(
            progress=i + 1,
            total=task_count,
            message=f"子任务 {i + 1}/{task_count}",
        )
        
        # 每 10 个子任务发一条 info 日志
        if (i + 1) % 10 == 0:
            elapsed = time.time() - start_time
            speed = (i + 1) / elapsed if elapsed > 0 else 0
            await ctx.log.info(
                f"已完成 {i + 1}/{task_count}, 速度 {speed:.1f}/秒",
                logger="task-manager",
            )
    
    # 任务完成
    elapsed = time.time() - start_time
    _task_status[task_id]["status"] = "completed"
    
    success_count = sum(1 for r in results if r["status"] == "success")
    failed_count = task_count - success_count
    
    await ctx.log.info(
        f"任务 {task_id} 完成: {success_count}成功, {failed_count}失败, 耗时{elapsed:.1f}秒",
        logger="task-manager",
    )
    
    return json.dumps({
        "task_id": task_id,
        "total": task_count,
        "success": success_count,
        "failed": failed_count,
        "elapsed": round(elapsed, 2),
    }, ensure_ascii=False)


# ===== 工具 2: 多级别日志通知 =====
@mcp.tool
async def system_health_check(ctx: Context) -> str:
    """
    执行系统健康检查，发送多级别日志
    
    Args:
        ctx: 上下文对象
    """
    checks = [
        # (级别, 检查项, 状态描述, logger 名称)
        ("debug", "CPU", "CPU 使用率 35%，正常", "health-check"),
        ("debug", "Memory", "内存使用率 62%，正常", "health-check"),
        ("info", "Disk", "磁盘空间充足", "health-check"),
        ("info", "Network", "网络延迟 12ms，正常", "health-check"),
        ("notice", "Config", "检测到配置变更", "config-monitor"),
        ("warning", "Database", "数据库连接池使用率 85%，接近上限", "database"),
        ("info", "API", "API 响应时间正常", "api-monitor"),
        ("info", "Cache", "缓存命中率 94%", "cache-monitor"),
    ]
    
    await ctx.log.info("开始系统健康检查", logger="health-check")
    
    issues = []
    
    for level, component, description, logger in checks:
        # 模拟检查耗时
        await asyncio.sleep(0.2)
        
        # 发送对应级别的日志
        if level == "debug":
            await ctx.log.debug(f"[{component}] {description}", logger=logger)
        elif level == "info":
            await ctx.log.info(f"[{component}] {description}", logger=logger)
        elif level == "notice":
            await ctx.log.info(f"[{component}] {description}", logger=logger)
        elif level == "warning":
            await ctx.log.warning(f"[{component}] {description}", logger=logger)
            issues.append({"component": component, "level": "warning", "detail": description})
    
    await ctx.log.info("系统健康检查完成", logger="health-check")
    
    return json.dumps({
        "total_checks": len(checks),
        "issues": issues,
        "summary": "系统整体健康，有 1 个警告需要关注",
    }, ensure_ascii=False)


# ===== 工具 3: 带资源变更通知的监控数据 =====
@mcp.tool
async def start_monitoring(
    interval: int,
    ctx: Context,
) -> str:
    """
    启动监控数据采集，持续更新监控数据
    
    Args:
        interval: 采集间隔（秒），1-60
        ctx: 上下文对象
    """
    # 参数校验
    if interval < 1 or interval > 60:
        return f"错误: 间隔必须在 1-60 秒之间，当前 {interval}"
    
    await ctx.log.info(
        f"启动监控，间隔 {interval} 秒",
        logger="monitor",
    )
    
    # 模拟监控数据更新
    # 实际项目中这会是一个后台任务
    for i in range(5):
        # 模拟数据变化
        _monitor_data["cpu"] = 30.0 + 20.0 * ((time.time() % 10) / 10)
        _monitor_data["memory"] = 55.0 + 15.0 * ((time.time() % 7) / 7)
        _monitor_data["requests_per_sec"] = int(1000 + 500 * ((time.time() % 5) / 5))
        _monitor_data["last_update"] = time.strftime("%H:%M:%S")
        
        # 发送日志通知
        await ctx.log.info(
            f"监控数据更新: CPU={_monitor_data['cpu']:.1f}%, "
            f"MEM={_monitor_data['memory']:.1f}%, "
            f"RPS={_monitor_data['requests_per_sec']}",
            logger="monitor",
        )
        
        # 上报进度
        await ctx.report_progress(
            progress=i + 1,
            total=5,
            message=f"监控采集 {i + 1}/5",
        )
        
        await asyncio.sleep(interval)
    
    await ctx.log.info("监控完成", logger="monitor")
    
    return json.dumps({
        "monitor": _monitor_data,
        "message": "监控数据采集完成",
    }, ensure_ascii=False)


# ===== 启动服务器 =====
if __name__ == "__main__":
    mcp.run()

效果验证

安装依赖并运行

pip install fastmcp
python notification_server.py

用 MCP Inspector 连接后，先发送 logging/setLevel 请求设置日志级别为 debug。然后调用 process_with_progress 工具，传入 {"task_count": 20}，你会看到进度通知从 1/20 逐步到 20/20，同时收到多条 info 和 debug 级别的日志通知。

调用 system_health_check 工具，你会收到 debug、info、notice、warning 四个级别的日志通知，分别对应不同检查项的状态。

调用 start_monitoring 工具，传入 {"interval": 2}，会看到 5 次监控数据更新，每次都带进度通知和日志通知。

常见问题与避坑

1. 进度通知发了但客户端看不到

这个问题的根因通常是客户端没有携带 progress token。MCP 规范要求客户端在请求的 _meta 里带上 progressToken，服务器才能关联通知和请求。如果客户端没带 token，服务器的 ctx.report_progress 调用不会报错，但通知会被静默丢弃。解决办法是在进度通知不可用时降级为日志通知，这样即使没有 progress token 也能传递进度信息。

2. 日志通知级别设置不当导致信息过载或丢失

如果你在服务器里发了大量 debug 级别的日志，而客户端设置了 info 级别，这些 debug 日志全部被丢弃，你可能以为代码有 bug。反过来，如果客户端设置了 debug 级别，你在循环里发 debug 日志可能产生通知风暴。建议在循环里控制日志频率，每 N 次迭代发一次，而不是每次都发。

3. 进度值没有递增导致通知被忽略

MCP 规范要求 progress 值必须递增。如果你发了 progress=50，然后又发了 progress=30，后面的通知会被客户端忽略。我在一个项目里因为浮点数精度问题导致 progress 值偶尔不递增，排查了很久。解决办法是 progress 用整数，或者对浮点数做舍入处理。

4. 通知频率过高导致客户端处理不过来

批量处理 10000 个文件，每个文件都发进度通知和日志通知，客户端可能来不及处理。MCP 规范也要求双方实现速率限制。建议按批次上报，比如每处理 100 个文件才发一次进度通知和日志通知，而不是每个文件都发。

5. 长任务被取消后通知没有停止

当客户端断开连接或取消请求时，正在执行的长任务应该停止发送通知。如果你在循环里没有检查任务是否被取消，会继续发通知给一个已经不存在的客户端，浪费资源。在 asyncio 里，客户端断开会触发 CancelledError，你需要正确捕获并清理。在 except 链里先处理 CancelledError 并 re-raise，再处理其他异常。

小结

MCP 通知机制让长任务和实时数据场景变得可行。进度通知通过 progress token 关联请求和通知，让客户端能展示进度条。日志通知遵循 syslog 级别标准，客户端可以控制接收的详细程度。资源变更通知支持订阅模式，让客户端能实时获取更新的数据。实现时要注意 progress 值必须递增、通知频率需要控制、客户端可能不携带 progress token 需要做降级处理。整个 MCP 专栏到这里就结束了，从协议基础到 SDK 使用，从工具开发到资源、提示和通知，希望能帮你构建自己的 MCP 服务器。

相关推荐

# windows # python # microsoft # MCP
Logo
MCP技术社区

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

更多推荐

cover

Windows 下部署 New API 中转站全过程实录（含 6 个真实踩坑）

avatar MCP技术社区

Google能索引，AI却抓不到

很多技术团队有个错觉：只要Google能正常收录，内容就没问题。但现实是，你的页面在Google搜索结果里好好的，在ChatGPT、Perplexity、Claude的答案里却像从未存在过一样。

avatar MCP技术社区
cover

英伟达AI刷爆ARC-AGI-3！AVO Harness架构让Agent从30%飙升到100%

avatar MCP技术社区