1.进展跟踪概述 #

Model Context Protocol (MCP) 提供了强大的进展跟踪功能，允许对长时间运行的操作进行实时进度监控。这个功能通过通知消息机制实现，让任何一方都可以发送进度更新，为用户提供更好的操作体验。

2.核心概念 #

2.1 什么是进展跟踪？ #

进展跟踪是MCP协议中的一个可选功能，用于：

• 监控长时间运行的操作状态
• 提供实时的进度反馈
• 改善用户体验，让用户了解操作进展

2.2 工作原理 #

1. 请求方在发送请求时包含一个progressToken
2. 接收方可以选择性地发送进度通知
3. 进度通知包含当前进度值、总进度等信息
4. 操作完成后停止发送进度通知

3.进度流程详解 #

3.1 第一步：发送带进度Access Token的请求 #

当一方希望接收请求的进度更新时，需要在请求的元数据中包含progressToken：

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "some_method",
  "params": {
    "_meta": {
      "progressToken": "abc123"
    }
  }
}

重要规则：

• 进度Access Token必须是一个字符串或整数值
• 发送方可以通过任何方式选择进度Access Token，但必须在所有活跃请求中保持唯一性

3.2 第二步：接收进度通知 #

接收器可能会发送包含以下内容的进度通知：

• 原始进度Access Token
• 当前的进度值
• 一个可选的"total"值
• 一个可选的"message"值

示例进度通知：

{
  "jsonrpc": "2.0",
  "method": "notifications/progress",
  "params": {
    "progressToken": "abc123",
    "progress": 50,
    "total": 100,
    "message": "正在处理数据..."
  }
}

进度通知规则：

• progress值必须随着每次通知而增加，即便总数未知
• progress和total值可能是浮点数
• message字段应该提供相关的人类可读进度信息

4.行为要求 #

4.1 进度Access Token管理 #

进度通知必须仅引用满足以下条件的Access Token：

• 在一个活跃请求中提供
• 与进行中的操作相关联

4.2 接收方行为 #

进度请求的接收者可能：

• 选择不发送任何进度通知
• 按照他们认为合适的频率发送通知
• 若总价值未知则省略total字段

5.实现说明 #

5.1 最佳实践 #

1. Access Token跟踪

   • 发送方与接收方应该跟踪活跃进度Access Token
   • 确保Access Token的唯一性和有效性

2. 速率限制

   • 双方应该实施速率限制以防止流量泛滥
   • 避免过于频繁的进度通知

3. 通知管理

   • 进度通知必须在操作完成后停止
   • 及时清理已完成的Access Token

5.2 典型流程示例 #

发送者 → 接收者：带progressToken的方法请求
接收者 → 发送者：进度通知 (0.2/1.0)
接收者 → 发送者：进度通知 (0.6/1.0)  
接收者 → 发送者：进度通知 (1.0/1.0)
接收者 → 发送者：方法响应

6.使用场景 #

6.1 适用场景 #

• 文件上传/下载：显示传输进度
• 数据处理：显示处理步骤和进度
• 模型训练：显示训练轮次和准确率
• 批量操作：显示已完成项目数量

6.2 注意事项 #

1. 性能考虑

   • 不要过于频繁地发送进度通知
   • 考虑网络延迟和带宽限制

2. 用户体验

   • 提供有意义的进度消息
   • 合理设置进度更新频率

3. 错误处理

   • 处理进度Access Token失效的情况
   • 优雅处理网络中断

7.代码实现示例 #

7.1 Python 实现示例 #

# 导入json模块，用于序列化和反序列化JSON数据
import json
# 导入uuid模块，用于生成唯一标识符
import uuid
# 导入time模块，用于获取当前时间戳
import time
# 从typing模块导入Dict和Optional类型，用于类型注解
from typing import Dict, Optional

# 定义MCPProgressTracker类，用于管理进度跟踪
class MCPProgressTracker:
    # 构造函数，初始化活跃进度Access Token字典
    def __init__(self):
        # active_tokens用于存储所有活跃的进度Access Token及其相关信息
        self.active_tokens: Dict[str, dict] = {}

    # 创建唯一的进度Access Token
    def create_progress_token(self) -> str:
        """创建唯一的进度Access Token"""
        # 使用uuid4生成唯一标识符，并转换为字符串
        return str(uuid.uuid4())

    # 开始跟踪某个进度
    def start_progress_tracking(self, token: str, total: Optional[float] = None):
        """开始跟踪进度"""
        # 在active_tokens中添加新的进度Access Token，初始化当前进度为0，总进度为total，记录开始时间
        self.active_tokens[token] = {
            'current': 0.0,
            'total': total,
            'start_time': time.time()
        }

    # 更新进度
    def update_progress(self, token: str, progress: float, message: str = ""):
        """更新进度"""
        # 检查进度Access Token是否有效
        if token not in self.active_tokens:
            raise ValueError(f"无效的进度Access Token: {token}")

        # 检查进度值是否递增
        if progress < self.active_tokens[token]['current']:
            raise ValueError("进度值不能减少")

        # 更新当前进度值
        self.active_tokens[token]['current'] = progress

        # 构造进度通知消息
        notification = {
            "jsonrpc": "2.0",
            "method": "notifications/progress",
            "params": {
                "progressToken": token,
                "progress": progress,
                "message": message
            }
        }

        # 如果有总进度，则加入total字段
        if self.active_tokens[token]['total'] is not None:
            notification["params"]["total"] = self.active_tokens[token]['total']

        # 返回进度通知
        return notification

    # 完成进度跟踪，清理Access Token
    def complete_progress(self, token: str):
        """完成进度跟踪"""
        # 如果Access Token存在于active_tokens中，则删除
        if token in self.active_tokens:
            del self.active_tokens[token]

# 使用示例

# 创建MCPProgressTracker实例
tracker = MCPProgressTracker()

# 创建唯一的进度Access Token
progress_token = tracker.create_progress_token()
# 构造一个带有进度Access Token的请求
request = {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "process_data",
    "params": {
        "data": "some_data",
        "_meta": {
            "progressToken": progress_token
        }
    }
}

# 开始跟踪进度，设置总进度为100.0
tracker.start_progress_tracking(progress_token, total=100.0)

# 模拟进度更新，每隔20递增一次
for i in range(0, 101, 20):
    # 更新进度并生成进度通知
    notification = tracker.update_progress(
        progress_token, 
        float(i), 
        f"处理进度: {i}%"
    )
    # 打印格式化后的进度通知
    print(json.dumps(notification, indent=2))

# 完成进度跟踪，清理Access Token
tracker.complete_progress(progress_token)

7.2 JavaScript 实现示例 #

// 定义MCPProgressTracker类，用于进度跟踪
class MCPProgressTracker {
    // 构造函数，初始化一个Map用于存储活跃的进度Access Token
    constructor() {
        this.activeTokens = new Map();
    }

    // 创建唯一的进度Access Token，使用crypto生成UUID
    createProgressToken() {
        return crypto.randomUUID();
    }

    // 开始跟踪进度，记录初始进度、总进度和开始时间
    startProgressTracking(token, total = null) {
        this.activeTokens.set(token, {
            current: 0,           // 当前进度
            total: total,         // 总进度（可选）
            startTime: Date.now() // 记录开始时间
        });
    }

    // 更新进度，并生成进度通知
    updateProgress(token, progress, message = "") {
        // 获取对应的进度数据
        const tokenData = this.activeTokens.get(token);
        // 如果Access Token无效，抛出异常
        if (!tokenData) {
            throw new Error(`无效的进度Access Token: ${token}`);
        }

        // 不允许进度回退
        if (progress < tokenData.current) {
            throw new Error("进度值不能减少");
        }

        // 更新当前进度
        tokenData.current = progress;

        // 构造进度通知对象
        const notification = {
            jsonrpc: "2.0",
            method: "notifications/progress",
            params: {
                progressToken: token, // 进度Access Token
                progress: progress,   // 当前进度
                message: message      // 进度消息
            }
        };

        // 如果有总进度，则加入通知
        if (tokenData.total !== null) {
            notification.params.total = tokenData.total;
        }

        // 返回进度通知
        return notification;
    }

    // 完成进度跟踪，删除对应的Access Token
    completeProgress(token) {
        this.activeTokens.delete(token);
    }
}

// 使用示例

// 创建MCPProgressTracker实例
const tracker = new MCPProgressTracker();

// 创建唯一的进度Access Token
const progressToken = tracker.createProgressToken();

// 构造一个带有进度Access Token的请求对象
const request = {
    jsonrpc: "2.0",
    id: 1,
    method: "processData",
    params: {
        data: "some_data",
        _meta: {
            progressToken: progressToken
        }
    }
};

// 开始跟踪进度，设置总进度为100
tracker.startProgressTracking(progressToken, 100);

// 模拟进度更新，每隔20递增一次
for (let i = 0; i <= 100; i += 20) {
    // 更新进度并生成进度通知
    const notification = tracker.updateProgress(
        progressToken,
        i,
        `处理进度: ${i}%`
    );
    // 打印格式化后的进度通知
    console.log(JSON.stringify(notification, null, 2));
}

// 完成进度跟踪，清理Access Token
tracker.completeProgress(progressToken);

8.最佳实践总结 #

8.1 Access Token管理 #

• 使用UUID或其他唯一标识符作为进度Access Token
• 及时清理已完成的Access Token
• 验证Access Token的有效性

8.2 进度更新策略 #

• 设置合理的更新频率（建议每秒不超过1次）
• 在关键节点发送进度更新
• 提供有意义的进度消息

8.3 错误处理 #

• 处理无效Access Token的情况
• 处理网络中断和重连
• 实现超时机制

8.4 性能优化 #

• 避免过于频繁的进度通知
• 考虑批量操作时的进度聚合
• 实现进度通知的节流机制

9.总结 #

MCP的进展跟踪功能为长时间运行的操作提供了强大的监控能力。通过合理使用进度Access Token和通知机制，可以显著改善用户体验，让用户清楚地了解操作的进展情况。

记住，进展跟踪是可选的，但正确实现可以大大提升应用的可用性和用户满意度。通过遵循本文档中的最佳实践和实现指南，您可以构建出高效、可靠的进展跟踪系统。