<!-- title: MCP Server长任务编排：进度到异步全实战 -->

# MCP Server长任务编排：进度追踪到异步Task全实战

## 概述

当MCP Tool需要执行超过30秒的复杂操作——批量代码重构、大规模数据分析、远程部署流水线——你会遇到三大问题：客户端超时断开、用户无法感知进度、意外中断无法恢复。本教程从MCP通知机制入手，逐步构建支持进度推送、优雅取消和异步Task恢复的生产级长任务MCP Server，最终交付一个可运行的代码审计引擎。

**适用人群**：已有MCP Server基础开发经验，需要处理耗时操作的TypeScript开发者。

## 前置条件

| 条件 | 说明 |
|------|------|
| Node.js >= 18 | 运行时环境 |
| `@modelcontextprotocol/sdk` >= 1.20 | 支持通知与Task的SDK版本 |
| Claude Code CLI | 用于集成测试 |
| 基础MCP知识 | 了解Tool/Resource/Prompt三原语 |

```bash
# 快速环境验证
node -v  # v18+
npm ls @modelcontextprotocol/sdk  # 确认版本
claude --version  # Claude Code CLI
```

---

## 第一章：MCP通知体系——从静默到实时

### 1.1 为什么通知是长任务的基石

MCP协议基于JSON-RPC 2.0，默认是请求-响应模式。一次Tool调用如果耗时5分钟，客户端只能干等。通知（Notification）机制打破了这种沉默——Server可以主动向Client推送消息，且不期望回复。

```mermaid
sequenceDiagram
    participant Client as Claude Code
    participant Server as MCP Server

    Client->>Server: tools/call {name: "analyze_codebase", progressToken: "p1"}
    
    Note over Server: 开始分析...
    Server-->>Client: notifications/progress {progressToken: "p1", progress: 20, total: 100}
    Note over Client: 显示进度 20%
    
    Note over Server: 解析AST...
    Server-->>Client: notifications/progress {progress: 50, total: 100}
    Note over Client: 显示进度 50%
    
    Note over Server: 生成报告...
    Server-->>Client: notifications/progress {progress: 90, total: 100}
    
    Server->>Client: Tool Result {content: "分析完成"}
```

### 1.2 五种通知类型速查

MCP规范定义了三类通知方向，每类包含具体类型：

| 方向 | 通知类型 | 用途 |
|------|---------|------|
| Server→Client | `notifications/progress` | 工具执行进度 |
| Server→Client | `notifications/resource/updated` | 资源内容变更 |
| Server→Client | `notifications/tools/list_changed` | 工具列表变化 |
| Client→Server | `notifications/cancelled` | 取消正在执行的请求 |
| 双向 | `notifications/message`（日志） | 调试日志输出 |

### 1.3 第一个带进度的Tool

创建项目结构：

```bash
mkdir mcp-longtask && cd mcp-longtask
npm init -y
npm install @modelcontextprotocol/sdk zod
```

**核心文件 `server.ts`**：

```typescript
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

const server = new McpServer({
  name: "longtask-demo",
  version: "1.0.0",
});

// 工具一：带进度通知的批量处理
server.tool(
  "batch_process",
  "批量处理数据文件，实时返回进度",
  {
    fileCount: z.number().min(1).max(100).describe("待处理文件数量"),
    operation: z.enum(["analyze", "transform", "validate"]).describe("操作类型"),
  },
  {
    // 声明此工具支持长时间运行，可被取消
    destructiveHint: false,
    openWorldHint: false,
  },
  async ({ fileCount, operation }, { sendNotification, signal }) => {
    const results: string[] = [];

    for (let i = 1; i <= fileCount; i++) {
      // 检查取消信号
      if (signal?.aborted) {
        return {
          content: [{
            type: "text" as const,
            text: `已取消：处理了 ${i - 1}/${fileCount} 个文件。已完成的结果：\n${results.join("\n")}`,
          }],
          isError: true,
        };
      }

      // 模拟处理单个文件
      await new Promise((resolve) => setTimeout(resolve, 800));
      results.push(`文件 ${i}: ${operation} 完成`);

      // 发送进度通知
      if (sendNotification) {
        await sendNotification({
          method: "notifications/progress",
          params: {
            progressToken: "batch-process-1",  // 与请求中的 progressToken 对应
            progress: i,
            total: fileCount,
          },
        });
      }
    }

    return {
      content: [{
        type: "text" as const,
        text: `全部完成！共处理 ${fileCount} 个文件。\n${results.join("\n")}`,
      }],
    };
  }
);

async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
}

main().catch(console.error);
```

**package.json 添加启动脚本**：

```json
{
  "name": "mcp-longtask",
  "type": "module",
  "scripts": {
    "start": "npx tsx server.ts",
    "build": "npx tsx server.ts"
  }
}
```

### 1.4 在Claude Code中注册

在项目根目录创建 `.mcp.json`：

```json
{
  "mcpServers": {
    "longtask": {
      "command": "npx",
      "args": ["tsx", "server.ts"],
      "cwd": "."
    }
  }
}
```

启动Claude Code后输入：

```
请使用 batch_process 工具处理 10 个文件，执行 analyze 操作
```

**预期结果**：Claude Code在等待过程中会显示进度百分比，从10%逐步递增到100%，最终返回完整处理报告。

### 常见问题

| 问题 | 原因 | 解决 |
|------|------|------|
| 看不到进度通知 | `sendNotification` 未传入或为空 | 检查SDK版本 >= 1.20 |
| 进度Token不匹配 | Token未与客户端请求关联 | 使用请求中传入的 `progressToken` |
| 工具执行超时被杀 | 超过Claude Code默认超时 | 拆分为更小的子任务 |

---

## 第二章：优雅取消与中断恢复

### 2.1 取消信号的传递链路

MCP的取消机制依赖 `AbortSignal`，从Claude Code客户端一直穿透到你的Tool实现：

```mermaid
flowchart LR
    A[用户点击取消] --> B[Claude Code发送<br>notifications/cancelled]
    B --> C[SDK拦截并触发<br>AbortSignal.abort]
    C --> D[Tool中检查<br>signal.aborted]
    D --> E[清理资源<br>返回部分结果]
```

### 2.2 可中断的深度分析工具

在 `server.ts` 中追加：

```typescript
// 工具二：可中断的代码扫描器
server.tool(
  "code_scan",
  "深度扫描代码库，支持中断并返回已扫描结果",
  {
    directory: z.string().describe("扫描目录路径"),
    rules: z.array(z.string()).describe("检查规则列表，如 security, performance, style"),
  },
  async ({ directory, rules }, { sendNotification, signal }) => {
    const findings: Array<{ file: string; rule: string; severity: string; message: string }> = [];
    
    // 模拟文件列表
    const files = Array.from({ length: 20 }, (_, i) => `src/module-${i + 1}.ts`);
    let scannedCount = 0;

    for (const file of files) {
      // 优雅退出检查点
      if (signal?.aborted) {
        // 发送日志通知告知中断
        if (sendNotification) {
          await sendNotification({
            method: "notifications/message",
            params: {
              level: "warning",
              data: `扫描中断于第 ${scannedCount} 个文件，返回已完成结果`,
            },
          });
        }
        
        return {
          content: [
            {
              type: "text" as const,
              text: `⚠️ 扫描被中断\n\n已完成: ${scannedCount}/${files.length} 个文件\n发现: ${findings.length} 个问题\n\n部分结果:\n${JSON.stringify(findings, null, 2)}`,
            },
          ],
          isError: false,  // 中断不算错误，返回可用结果
        };
      }

      // 模拟对每个文件应用每条规则
      for (const rule of rules) {
        if (Math.random() > 0.7) {
          findings.push({
            file,
            rule,
            severity: Math.random() > 0.5 ? "high" : "medium",
            message: `在 ${file} 中发现 ${rule} 相关问题`,
          });
        }
      }

      scannedCount++;

      // 发送进度
      if (sendNotification) {
        await sendNotification({
          method: "notifications/progress",
          params: {
            progressToken: "code-scan-1",
            progress: scannedCount,
            total: files.length,
          },
        });
      }

      await new Promise((resolve) => setTimeout(resolve, 300));
    }

    // 生成汇总报告
    const summary = {
      totalFiles: files.length,
      totalFindings: findings.length,
      bySeverity: {
        high: findings.filter(f => f.severity === "high").length,
        medium: findings.filter(f => f.severity === "medium").length,
      },
      byRule: rules.reduce((acc, rule) => {
        acc[rule] = findings.filter(f => f.rule === rule).length;
        return acc;
      }, {} as Record<string, number>),
    };

    return {
      content: [{
        type: "text" as const,
        text: `✅ 扫描完成\n\n${JSON.stringify(summary, null, 2)}\n\n详细发现:\n${JSON.stringify(findings, null, 2)}`,
      }],
    };
  }
);
```

### 2.3 中断恢复的三种策略

| 策略 | 实现方式 | 适用场景 |
|------|---------|---------|
| **快照恢复** | Tool返回中间状态，下次调用传入继续 | 文件处理、数据转换 |
| **检查点持久化** | 写入临时文件/数据库，自动从断点续传 | 大规模分析、编译任务 |
| **分页分批** | 将大任务拆为固定大小的批次，按页执行 | API遍历、日志扫描 |

**检查点持久化示例**：

```typescript
import { writeFileSync, readFileSync, existsSync } from "fs";

const CHECKPOINT_FILE = "/tmp/mcp-scan-checkpoint.json";

interface ScanCheckpoint {
  lastFileIndex: number;
  timestamp: number;
  findings: Array<{ file: string; rule: string; severity: string; message: string }>;
}

function saveCheckpoint(data: ScanCheckpoint) {
  writeFileSync(CHECKPOINT_FILE, JSON.stringify(data));
}

function loadCheckpoint(): ScanCheckpoint | null {
  if (!existsSync(CHECKPOINT_FILE)) return null;
  return JSON.parse(readFileSync(CHECKPOINT_FILE, "utf-8"));
}

// 工具三：支持断点续传的扫描器
server.tool(
  "resumable_scan",
  "支持断点续传的代码扫描器",
  {
    directory: z.string().describe("目标目录"),
    resume: z.boolean().default(false).describe("是否从上次断点继续"),
  },
  async ({ directory, resume }, { sendNotification, signal }) => {
    const files = Array.from({ length: 50 }, (_, i) => `src/deep/module-${String(i).padStart(3, "0")}.ts`);
    
    let checkpoint = resume ? loadCheckpoint() : null;
    let startIndex = checkpoint?.lastFileIndex ?? 0;
    let findings = checkpoint?.findings ?? [];

    if (checkpoint) {
      if (sendNotification) {
        await sendNotification({
          method: "notifications/message",
          params: {
            level: "info",
            data: `从第 ${startIndex + 1} 个文件恢复，已有 ${findings.length} 个发现`,
          },
        });
      }
    }

    for (let i = startIndex; i < files.length; i++) {
      if (signal?.aborted) {
        // 保存检查点
        saveCheckpoint({ lastFileIndex: i, timestamp: Date.now(), findings });
        return {
          content: [{
            type: "text" as const,
            text: `⏸ 已暂停于第 ${i + 1}/${files.length} 个文件。检查点已保存。\n再次调用 resumable_scan 并设置 resume=true 可继续。`,
          }],
        };
      }

      // 模拟扫描
      if (Math.random() > 0.8) {
        findings.push({
          file: files[i],
          rule: "security",
          severity: "high",
          message: `潜在安全问题: ${files[i]}`,
        });
      }

      // 每10个文件保存一次检查点
      if ((i + 1) % 10 === 0) {
        saveCheckpoint({ lastFileIndex: i + 1, timestamp: Date.now(), findings });
      }

      if (sendNotification) {
        await sendNotification({
          method: "notifications/progress",
          params: {
            progressToken: "resumable-scan",
            progress: i + 1,
            total: files.length,
          },
        });
      }

      await new Promise((resolve) => setTimeout(resolve, 200));
    }

    // 清理检查点文件
    try { require("fs").unlinkSync(CHECKPOINT_FILE); } catch {}

    return {
      content: [{
        type: "text" as const,
        text: `✅ 全量扫描完成\n文件: ${files.length}\n发现: ${findings.length} 个问题`,
      }],
    };
  }
);
```

---

## 第三章：资源订阅——实时数据推送

### 3.1 Resource Subscription机制

MCP的Resource默认是静态数据。但通过订阅机制，Server可以在数据变化时主动推送更新：

```mermaid
sequenceDiagram
    participant CC as Claude Code
    participant S as MCP Server

    CC->>S: resources/subscribe {uri: "build://status"}
    S-->>CC: 订阅确认
    
    Note over S: 构建状态变更...
    S-->>CC: notifications/resources/updated {uri: "build://status"}
    CC->>S: resources/read {uri: "build://status"}
    S->>CC: 返回最新构建状态
    
    Note over S: 构建完成...
    S-->>CC: notifications/resources/updated {uri: "build://status"}
    CC->>S: resources/read {uri: "build://status"}
    S->>CC: 构建成功 ✓
    
    CC->>S: resources/unsubscribe {uri: "build://status"}
```

### 3.2 构建状态监控资源

在 `server.ts` 中添加：

```typescript
// 模拟构建状态存储
interface BuildStatus {
  id: string;
  project: string;
  status: "queued" | "running" | "success" | "failed";
  progress: number;
  startedAt: string;
  logs: string[];
}

const buildStore = new Map<string, BuildStatus>();

// 添加定期更新的构建状态资源
server.resource(
  "build-status",
  "build://status",
  async (uri) => {
    const statuses = Array.from(buildStore.values());
    return {
      contents: [{
        uri: uri.href,
        mimeType: "application/json",
        text: JSON.stringify(statuses, null, 2),
      }],
    };
  }
);

// 单个构建资源
server.resource(
  "build-detail",
  new z.ZodType<string, z.ZodStringDef>().optional().describe("build://status/{buildId}"),
  // 使用模板URI
  async (uri) => {
    const buildId = uri.href.replace("build://status/", "");
    const status = buildStore.get(buildId);
    if (!status) {
      return {
        contents: [{
          uri: uri.href,
          text: `构建 ${buildId} 不存在`,
        }],
      };
    }
    return {
      contents: [{
        uri: uri.href,
        mimeType: "application/json",
        text: JSON.stringify(status, null, 2),
      }],
    };
  }
);

// 工具四：触发构建并监控
server.tool(
  "trigger_build",
  "触发项目构建，实时推送状态更新",
  {
    project: z.string().describe("项目名称"),
    command: z.string().describe("构建命令"),
  },
  async ({ project, command }, { sendNotification }) => {
    const buildId = `build-${Date.now()}`;
    const status: BuildStatus = {
      id: buildId,
      project,
      status: "running",
      progress: 0,
      startedAt: new Date().toISOString(),
      logs: [`开始构建: ${command}`],
    };
    buildStore.set(buildId, status);

    // 模拟构建步骤
    const steps = ["依赖安装", "编译", "测试", "打包", "部署"];
    for (let i = 0; i < steps.length; i++) {
      await new Promise((resolve) => setTimeout(resolve, 1000));
      
      status.progress = Math.round(((i + 1) / steps.length) * 100);
      status.logs.push(`[${steps[i]}] 完成 (${status.progress}%)`);

      // 推送资源更新通知
      if (sendNotification) {
        await sendNotification({
          method: "notifications/resources/updated",
          params: {
            uri: `build://status/${buildId}`,
          },
        });
      }
    }

    status.status = "success";
    status.logs.push("构建完成 ✓");

    return {
      content: [{
        type: "text" as const,
        text: `构建 ${buildId} 已完成\n项目: ${project}\n状态: 成功\n日志:\n${status.logs.map(l => `  - ${l}`).join("\n")}`,
      }],
    };
  }
);
```

---

## 第四章：MCP Tasks——异步任务原语

### 4.1 Tasks vs 普通Tool调用

MCP Tasks是协议规范2025-11-25版本引入的异步任务原语，专为长时间运行的操作设计：

```mermaid
flowchart TB
    subgraph 传统Tool调用
        T1[Client调用Tool] --> T2[同步等待]
        T2 --> T3[超时风险]
    end

    subgraph MCP Tasks模式
        M1[Client发起Task] --> M2[Server返回taskToken]
        M2 --> M3[Client轮询状态]
        M3 --> M4{Task完成?}
        M4 -->|否| M3
        M4 -->|是| M5[获取结果]
    end

    style T3 fill:#ff6b6b,color:#fff
    style M5 fill:#51cf66,color:#fff
```

**关键差异**：

| 特性 | 普通Tool | MCP Task |
|------|---------|----------|
| 执行模式 | 同步阻塞 | 异步轮询 |
| 超时风险 | 高（客户端等待） | 无（立即返回token） |
| 状态追踪 | 仅进度通知 | 完整生命周期 |
| 中断恢复 | 需自行实现 | 协议原生支持 |
| 适用时长 | < 30秒 | 分钟到小时级 |

### 4.2 Task的生命周期

```mermaid
stateDiagram-v2
    [*] --> pending: 创建Task
    pending --> running: 开始执行
    running --> completed: 正常完成
    running --> failed: 执行出错
    running --> cancelled: 用户取消
    running --> input_required: 需要用户输入
    input_required --> running: 用户提供输入
    completed --> [*]
    failed --> [*]
    cancelled --> [*]
```

`input_required` 状态是MCP Tasks的精髓——Server可以在执行过程中暂停，向用户请求额外信息后继续。这实现了人机协作的异步工作流。

### 4.3 实现异步Task引擎

在 `server.ts` 中添加完整的Task管理系统：

```typescript
// Task管理系统
interface McpTask {
  id: string;
  status: "pending" | "running" | "completed" | "failed" | "cancelled" | "input_required";
  progress: number;
  total: number | null;
  result: unknown;
  error: string | null;
  createdAt: number;
  updatedAt: number;
  metadata: Record<string, unknown>;
}

const taskStore = new Map<string, McpTask>();
let taskIdCounter = 0;

function createTask(metadata: Record<string, unknown> = {}): McpTask {
  const task: McpTask = {
    id: `task-${++taskIdCounter}`,
    status: "pending",
    progress: 0,
    total: null,
    result: null,
    error: null,
    createdAt: Date.now(),
    updatedAt: Date.now(),
    metadata,
  };
  taskStore.set(task.id, task);
  return task;
}

function updateTask(id: string, updates: Partial<McpTask>) {
  const task = taskStore.get(id);
  if (!task) throw new Error(`Task ${id} 不存在`);
  Object.assign(task, updates, { updatedAt: Date.now() });
  taskStore.set(id, task);
  return task;
}

// 工具五：启动异步代码审计Task
server.tool(
  "async_audit",
  "启动异步代码审计任务，立即返回Task ID，可后续查询进度",
  {
    target: z.string().describe("审计目标（目录或文件）"),
    depth: z.enum(["quick", "standard", "deep"]).default("standard").describe("审计深度"),
    categories: z.array(z.enum(["security", "performance", "maintainability", "style"]))
      .default(["security", "performance"])
      .describe("审计类别"),
  },
  async ({ target, depth, categories }) => {
    const task = createTask({ target, depth, categories });
    
    // 异步启动后台任务
    runAuditInBackground(task.id, target, depth, categories);

    return {
      content: [{
        type: "text" as const,
        text: `✅ 审计任务已创建\n\nTask ID: ${task.id}\n目标: ${target}\n深度: ${depth}\n类别: ${categories.join(", ")}\n\n使用 task_status 工具查询进度，使用 task_result 获取结果。`,
      }],
    };
  }
);

async function runAuditInBackground(
  taskId: string,
  target: string,
  depth: string,
  categories: string[]
) {
  const task = taskStore.get(taskId)!;
  
  // 根据深度确定扫描步骤数
  const stepsPerCategory = { quick: 5, standard: 12, deep: 25 } as const;
  const totalSteps = categories.length * stepsPerCategory[depth as keyof typeof stepsPerCategory];
  
  updateTask(taskId, { status: "running", total: totalSteps, progress: 0 });

  const findings: Array<{ category: string; severity: string; file: string; message: string }> = [];

  try {
    let step = 0;
    for (const category of categories) {
      const steps = stepsPerCategory[depth as keyof typeof stepsPerCategory];
      
      for (let i = 0; i < steps; i++) {
        step++;
        
        // 检查取消状态
        const currentTask = taskStore.get(taskId)!;
        if (currentTask.status === "cancelled") {
          return; // 已被外部取消
        }
        
        // 模拟扫描逻辑
        await new Promise((resolve) => setTimeout(resolve, 200));
        
        if (Math.random() > 0.75) {
          findings.push({
            category,
            severity: ["critical", "high", "medium", "low"][Math.floor(Math.random() * 4)],
            file: `${target}/module-${Math.floor(Math.random() * 50)}.ts`,
            message: `${category} 相关: 发现潜在${["SQL注入", "内存泄漏", "循环依赖", "硬编码密钥", "未处理异常"][Math.floor(Math.random() * 5)]}问题`,
          });
        }
        
        updateTask(taskId, { progress: step });
      }
    }

    updateTask(taskId, {
      status: "completed",
      progress: totalSteps,
      result: {
        target,
        depth,
        findings,
        summary: {
          total: findings.length,
          critical: findings.filter(f => f.severity === "critical").length,
          high: findings.filter(f => f.severity === "high").length,
          medium: findings.filter(f => f.severity === "medium").length,
          low: findings.filter(f => f.severity === "low").length,
        },
        completedAt: new Date().toISOString(),
      },
    });
  } catch (error) {
    updateTask(taskId, {
      status: "failed",
      error: error instanceof Error ? error.message : String(error),
    });
  }
}

// 工具六：查询Task状态
server.tool(
  "task_status",
  "查询异步任务的当前状态和进度",
  {
    taskId: z.string().describe("任务ID"),
  },
  async ({ taskId }) => {
    const task = taskStore.get(taskId);
    if (!task) {
      return {
        content: [{ type: "text" as const, text: `Task ${taskId} 不存在` }],
        isError: true,
      };
    }

    const elapsed = Math.round((Date.now() - task.createdAt) / 1000);
    const progressPercent = task.total ? Math.round((task.progress / task.total) * 100) : 0;
    const progressBar = "█".repeat(Math.floor(progressPercent / 5)) + "░".repeat(20 - Math.floor(progressPercent / 5));

    return {
      content: [{
        type: "text" as const,
        text: `📋 Task: ${task.id}\n状态: ${task.status}\n进度: ${progressBar} ${progressPercent}%\n耗时: ${elapsed}秒\n${task.error ? `错误: ${task.error}` : ""}`,
      }],
    };
  }
);

// 工具七：获取Task结果
server.tool(
  "task_result",
  "获取已完成异步任务的完整结果",
  {
    taskId: z.string().describe("任务ID"),
  },
  async ({ taskId }) => {
    const task = taskStore.get(taskId);
    if (!task) {
      return {
        content: [{ type: "text" as const, text: `Task ${taskId} 不存在` }],
        isError: true,
      };
    }

    if (task.status !== "completed") {
      return {
        content: [{
          type: "text" as const,
          text: `Task ${taskId} 当前状态为 ${task.status}，尚未完成。请使用 task_status 查询进度。`,
        }],
        isError: true,
      };
    }

    return {
      content: [{
        type: "text" as const,
        text: `📊 审计报告 — ${task.id}\n\n${JSON.stringify(task.result, null, 2)}`,
      }],
    };
  }
);

// 工具八：取消Task
server.tool(
  "task_cancel",
  "取消正在运行的异步任务",
  {
    taskId: z.string().describe("要取消的任务ID"),
  },
  async ({ taskId }) => {
    const task = taskStore.get(taskId);
    if (!task) {
      return {
        content: [{ type: "text" as const, text: `Task ${taskId} 不存在` }],
        isError: true,
      };
    }

    if (task.status !== "running" && task.status !== "pending") {
      return {
        content: [{
          type: "text" as const,
          text: `Task ${taskId} 状态为 ${task.status}，无法取消。只能取消 pending 或 running 状态的任务。`,
        }],
      };
    }

    updateTask(taskId, { status: "cancelled" });

    return {
      content: [{
        type: "text" as const,
        text: `✅ Task ${taskId} 已取消`,
      }],
    };
  }
);
```

---

## 第五章：实战案例——生产级代码审计引擎

### 5.1 完整架构

将前面所有章节的内容整合为一个完整的MCP Server，支持以下工作流：

```mermaid
flowchart TB
    subgraph "Claude Code客户端"
        U[用户发起审计请求] --> R{选择模式?}
        R -->|快速| S1[batch_process<br>同步+进度]
        R -->|标准| S2[code_scan<br>可中断+日志]
        R -->|深度| S3[async_audit<br>异步Task]
    end

    subgraph "MCP Server"
        S1 --> P1[进度通知<br>notifications/progress]
        S2 --> P2[取消处理<br>AbortSignal + 检查点]
        S3 --> P3[后台引擎<br>Task生命周期管理]
        P3 --> TS[task_status<br>task_result<br>task_cancel]
    end

    subgraph "数据层"
        P3 --> FS[文件系统检查点]
        P3 --> MEM[内存Task存储]
    end

    style S1 fill:#74c0fc
    style S2 fill:#ffa94d
    style S3 fill:#69db7c
```

### 5.2 完整server.ts汇总

以下是完整的、可直接运行的服务器代码。将前面章节的所有代码片段合并为一个文件：

```typescript
// server.ts - 生产级代码审计MCP Server
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import { writeFileSync, readFileSync, existsSync, unlinkSync } from "fs";

// ==================== Task管理系统 ====================
interface McpTask {
  id: string;
  status: "pending" | "running" | "completed" | "failed" | "cancelled";
  progress: number;
  total: number | null;
  result: unknown;
  error: string | null;
  createdAt: number;
  updatedAt: number;
}

const taskStore = new Map<string, McpTask>();
let taskIdCounter = 0;

function createTask(): McpTask {
  const task: McpTask = {
    id: `audit-${++taskIdCounter}-${Date.now().toString(36)}`,
    status: "pending",
    progress: 0,
    total: null,
    result: null,
    error: null,
    createdAt: Date.now(),
    updatedAt: Date.now(),
  };
  taskStore.set(task.id, task);
  return task;
}

// ==================== 检查点管理 ====================
const CHECKPOINT_DIR = "/tmp/mcp-audit";

function saveCheckpoint(taskId: string, data: object) {
  try {
    writeFileSync(`${CHECKPOINT_DIR}-${taskId}.json`, JSON.stringify(data));
  } catch { /* 静默失败 */ }
}

function loadCheckpoint(taskId: string): object | null {
  try {
    if (!existsSync(`${CHECKPOINT_DIR}-${taskId}.json`)) return null;
    return JSON.parse(readFileSync(`${CHECKPOINT_DIR}-${taskId}.json`, "utf-8"));
  } catch { return null; }
}

// ==================== MCP Server ====================
const server = new McpServer({
  name: "code-audit-engine",
  version: "2.0.0",
});

// ----- 工具1: 同步快速扫描（带进度） -----
server.tool(
  "quick_scan",
  "快速同步扫描，带实时进度反馈，适合小型项目",
  { path: z.string(), rules: z.array(z.string()).default(["security"]) },
  async ({ path, rules }, { sendNotification }) => {
    const files = Array.from({ length: 15 }, (_, i) => `${path}/file-${i}.ts`);
    const issues: string[] = [];

    for (let i = 0; i < files.length; i++) {
      await new Promise(r => setTimeout(r, 400));
      if (Math.random() > 0.7) issues.push(`[${rules[0]}] ${files[i]}: 发现问题`);

      await sendNotification?.({
        method: "notifications/progress",
        params: { progressToken: "qs", progress: i + 1, total: files.length },
      });
    }

    return {
      content: [{ type: "text" as const, text: `扫描 ${files.length} 文件，发现 ${issues.length} 问题\n${issues.join("\n")}` }],
    };
  }
);

// ----- 工具2: 可中断深度扫描 -----
server.tool(
  "deep_scan",
  "深度扫描，支持中断恢复。中断后可使用 resume_scan 继续",
  {
    path: z.string(),
    fileCount: z.number().default(50),
  },
  async ({ path, fileCount }, { sendNotification, signal }) => {
    const files = Array.from({ length: fileCount }, (_, i) => `${path}/module-${i}.ts`);
    const findings: Array<{ file: string; type: string; msg: string }> = [];

    for (let i = 0; i < files.length; i++) {
      if (signal?.aborted) {
        saveCheckpoint("deep-scan", { lastIndex: i, findings, totalFiles: fileCount });
        return {
          content: [{
            type: "text" as const,
            text: `⏸ 已暂停于 ${i}/${fileCount}。使用 resume_scan 继续扫描。`,
          }],
        };
      }

      await new Promise(r => setTimeout(r, 150));
      if (Math.random() > 0.75) {
        findings.push({
          file: files[i],
          type: ["security", "perf", "style"][Math.floor(Math.random() * 3)],
          msg: "需要人工审查",
        });
      }

      await sendNotification?.({
        method: "notifications/progress",
        params: { progressToken: "ds", progress: i + 1, total: fileCount },
      });
    }

    return {
      content: [{
        type: "text" as const,
        text: `✅ 深度扫描完成\n扫描: ${fileCount} 文件\n发现: ${findings.length} 问题\n${JSON.stringify(findings, null, 2)}`,
      }],
    };
  }
);

// ----- 工具3: 断点续传 -----
server.tool(
  "resume_scan",
  "从上次中断处继续深度扫描",
  {},
  async (_, { sendNotification, signal }) => {
    const checkpoint = loadCheckpoint("deep-scan") as {
      lastIndex: number; findings: Array<{ file: string; type: string; msg: string }>; totalFiles: number;
    } | null;

    if (!checkpoint) {
      return { content: [{ type: "text" as const, text: "没有找到检查点，请先运行 deep_scan" }], isError: true };
    }

    const { lastIndex, findings, totalFiles } = checkpoint;
    const files = Array.from({ length: totalFiles }, (_, i) => `src/module-${i}.ts`);

    for (let i = lastIndex; i < totalFiles; i++) {
      if (signal?.aborted) {
        saveCheckpoint("deep-scan", { lastIndex: i, findings, totalFiles });
        return {
          content: [{ type: "text" as const, text: `⏸ 再次暂停于 ${i}/${totalFiles}` }],
        };
      }

      await new Promise(r => setTimeout(r, 150));
      if (Math.random() > 0.75) {
        findings.push({
          file: files[i],
          type: "security",
          msg: "续扫发现",
        });
      }

      await sendNotification?.({
        method: "notifications/progress",
        params: { progressToken: "rs", progress: i + 1 - lastIndex, total: totalFiles - lastIndex },
      });
    }

    // 清理检查点
    try { unlinkSync(`${CHECKPOINT_DIR}-deep-scan.json`); } catch {}

    return {
      content: [{
        type: "text" as const,
        text: `✅ 续传扫描完成（从第 ${lastIndex} 个文件恢复）\n总计: ${findings.length} 个问题`,
      }],
    };
  }
);

// ----- 工具4: 异步审计Task -----
server.tool(
  "start_audit",
  "启动异步审计任务，立即返回Task ID。适用于大型项目",
  {
    project: z.string(),
    depth: z.enum(["quick", "standard", "deep"]).default("standard"),
  },
  async ({ project, depth }) => {
    const task = createTask();
    runAuditAsync(task.id, project, depth);
    return {
      content: [{
        type: "text" as const,
        text: `🚀 审计已启动\n\nTask ID: ${task.id}\n项目: ${project}\n深度: ${depth}\n\n→ 使用 check_task "${task.id}" 查看进度\n→ 使用 cancel_task "${task.id}" 取消`,
      }],
    };
  }
);

async function runAuditAsync(taskId: string, project: string, depth: string) {
  const stepMap = { quick: 10, standard: 30, deep: 60 };
  const total = stepMap[depth as keyof typeof stepMap] ?? 30;
  const task = taskStore.get(taskId)!;
  Object.assign(task, { status: "running", total });

  const findings: Array<{ severity: string; file: string; message: string }> = [];

  try {
    for (let i = 1; i <= total; i++) {
      if (taskStore.get(taskId)!.status === "cancelled") return;

      await new Promise(r => setTimeout(r, 300));

      if (Math.random() > 0.7) {
        findings.push({
          severity: ["critical", "high", "medium", "low"][Math.floor(Math.random() * 4)],
          file: `${project}/src/${["api", "db", "auth", "utils"][Math.floor(Math.random() * 4)]}/${["handler", "service", "model", "helper"][Math.floor(Math.random() * 4)]}.ts`,
          message: ["硬编码凭证", "未参数化SQL查询", "缺少输入验证", "循环依赖", "未捕获Promise异常"][Math.floor(Math.random() * 5)],
        });
      }

      Object.assign(task, { progress: i, updatedAt: Date.now() });
    }

    Object.assign(task, {
      status: "completed",
      result: {
        project, depth,
        totalFindings: findings.length,
        critical: findings.filter(f => f.severity === "critical").length,
        high: findings.filter(f => f.severity === "high").length,
        findings: findings.slice(0, 20),
        completedAt: new Date().toISOString(),
      },
    });
  } catch (e) {
    Object.assign(task, { status: "failed", error: String(e) });
  }
}

// ----- 工具5: 查询Task -----
server.tool(
  "check_task",
  "查询异步任务状态",
  { taskId: z.string() },
  async ({ taskId }) => {
    const task = taskStore.get(taskId);
    if (!task) return { content: [{ type: "text" as const, text: `Task ${taskId} 不存在` }], isError: true };

    const pct = task.total ? Math.round((task.progress / task.total) * 100) : 0;
    const bar = "█".repeat(Math.floor(pct / 5)) + "░".repeat(20 - Math.floor(pct / 5));

    return {
      content: [{
        type: "text" as const,
        text: `${bar} ${pct}%\n状态: ${task.status}\n${task.status === "completed" ? JSON.stringify(task.result, null, 2) : task.error ?? ""}`,
      }],
    };
  }
);

// ----- 工具6: 取消Task -----
server.tool(
  "cancel_task",
  "取消异步任务",
  { taskId: z.string() },
  async ({ taskId }) => {
    const task = taskStore.get(taskId);
    if (!task) return { content: [{ type: "text" as const, text: "不存在" }], isError: true };
    if (task.status !== "running") return { content: [{ type: "text" as const, text: `状态 ${task.status} 不可取消` }] };
    Object.assign(task, { status: "cancelled" });
    return { content: [{ type: "text" as const, text: `✅ ${taskId} 已取消` }] };
  }
);

// ==================== 启动 ====================
async function main() {
  const transport = new StdioServerTransport();
  await server.connect(transport);
  console.error("[code-audit-engine] MCP Server 已启动");
}

main().catch(console.error);
```

### 5.3 集成测试对话

启动Claude Code后，按以下对话测试全部功能：

```
# 测试1：同步快速扫描
用 quick_scan 扫描 ./src 目录，检查 security 规则

# 测试2：可中断深度扫描
用 deep_scan 扫描 ./project 目录，50个文件
（等待进度到一半时按 Ctrl+C 中断）

# 测试3：断点续传
用 resume_scan 继续刚才中断的扫描

# 测试4：异步审计
用 start_audit 启动对 ./large-project 的 deep 深度审计
（记下返回的 Task ID，如 audit-1-m5x7k）

# 测试5：查询进度
用 check_task 查询 audit-1-m5x7k 的进度

# 测试6：取消任务
用 cancel_task 取消 audit-1-m5x7k
```

---

## 第六章：Streamable HTTP传输——远程部署

### 6.1 为什么需要Streamable HTTP

stdio传输仅限本地进程间通信。当MCP Server需要部署到远程服务器、被多个客户端共享时，需要HTTP传输。MCP规范从2025-03-26版本起，推荐Streamable HTTP替代已废弃的SSE传输。

### 6.2 双传输Server

```typescript
// transport.ts - 双传输支持
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamablehttp.js";

export async function startWithTransport(server: any, mode: "stdio" | "http") {
  if (mode === "stdio") {
    const transport = new StdioServerTransport();
    await server.connect(transport);
    console.error("[Server] stdio 模式启动");
  } else {
    const express = await import("express");
    const app = express.default();
    app.use(express.default.json());

    const transport = new StreamableHTTPServerTransport({
      sessionIdGenerator: undefined,  // 无状态模式
    });

    app.post("/mcp", async (req: any, res: any) => {
      await transport.handleRequest(req, res, req.body);
    });

    app.get("/mcp", async (req: any, res: any) => {
      // SSE 端点用于服务器推送通知
      await transport.handleRequest(req, res);
    });

    await server.connect(transport);

    const PORT = process.env.PORT || 3001;
    app.listen(PORT, () => {
      console.error(`[Server] Streamable HTTP 模式启动: http://localhost:${PORT}/mcp`);
    });
  }
}
```

**远程配置 `.mcp.json`**：

```json
{
  "mcpServers": {
    "remote-audit": {
      "url": "http://your-server:3001/mcp",
      "transport": "streamable-http"
    }
  }
}
```

---

## 第七章：生产级增强

### 7.1 并发控制

长任务Server最怕资源耗尽。添加简单的并发限制：

```typescript
const MAX_CONCURRENT_TASKS = 5;
const activeTasks = new Set<string>();

function canStartTask(): boolean {
  return activeTasks.size < MAX_CONCURRENT_TASKS;
}

function registerTask(taskId: string) {
  activeTasks.add(taskId);
}

function unregisterTask(taskId: string) {
  activeTasks.delete(taskId);
  // 清理超过1小时的已完成任务
  const oneHourAgo = Date.now() - 3600000;
  for (const [id, task] of taskStore) {
    if (task.updatedAt < oneHourAgo && (task.status === "completed" || task.status === "failed" || task.status === "cancelled")) {
      taskStore.delete(id);
    }
  }
}
```

### 7.2 内存监控

```typescript
// 定期报告内存使用
setInterval(() => {
  const mem = process.memoryUsage();
  const mb = (bytes: number) => `${Math.round(bytes / 1024 / 1024)}MB`;
  console.error(`[监控] RSS: ${mb(mem.rss)} | Heap: ${mb(mem.heapUsed)}/${mb(mem.heapTotal)} | Tasks: ${taskStore.size}`);
}, 30000);
```

### 7.3 错误恢复

```typescript
// 进程级异常捕获，防止崩溃
process.on("uncaughtException", (err) => {
  console.error("[致命错误]", err.message);
  // 将所有运行中的任务标记为失败
  for (const [id, task] of taskStore) {
    if (task.status === "running") {
      Object.assign(task, { status: "failed", error: "服务器异常中断" });
    }
  }
});

process.on("unhandledRejection", (reason) => {
  console.error("[未处理Promise拒绝]", reason);
});
```

---

## 完整工作流总览

```mermaid
flowchart LR
    subgraph "用户决策"
        A[发起扫描请求] --> B{任务规模?}
    end

    B -->|小型<15文件| C[quick_scan<br>同步+进度通知]
    B -->|中型15-50文件| D[deep_scan<br>可中断+检查点]
    B -->|大型>50文件| E[start_audit<br>异步Task引擎]

    C --> F[实时进度条]
    D --> G{是否中断?}
    G -->|否| H[完整报告]
    G -->|是| I[保存检查点]
    I --> J[resume_scan<br>断点续传]
    J --> H

    E --> K[立即返回TaskID]
    K --> L[check_task<br>轮询状态]
    L --> M{完成?}
    M -->|否| L
    M -->|是| N[完整审计报告]

    style C fill:#74c0fc
    style D fill:#ffa94d
    style E fill:#69db7c
    style H fill:#b2f2bb
    style N fill:#b2f2bb
```

## 参考来源

1. [MCP Specification 2025-11-25 — Progress Tracking](https://modelcontextprotocol.io/specification/2025-11-25) — MCP协议最新规范，定义Progress、Task等原语
2. [MCP Async Tasks: Building Long-Running Workflows for AI Agents — WorkOS](https://workos.com/blog/mcp-async-tasks-ai-agent-workflows) — 深度解读MCP Tasks异步原语设计理念
3. [MCP For Software Engineers Part 2: Interactive & Long-Running Tools](https://newsletter.victordibia.com/p/mcp-for-software-engineers-part-2) — 长时间运行工具的交互模式与最佳实践
4. [Why MCP Deprecated SSE and Went with Streamable HTTP — fka.dev](https://blog.fka.dev/blog/2025-06-06-why-mcp-deprecated-sse-and-go-with-streamable-http/) — Streamable HTTP传输协议迁移原因分析
5. [MCP TypeScript SDK — Official GitHub Repository](https://github.com/modelcontextprotocol/typescript-sdk) — 官方TypeScript SDK源码与API文档

> [!tip] 51学通信知识星球
> **星主：爱卫生**（微信：gprshome201101）——公众号「51学通信」创办者，19 年通信行业运维与讲师经验，著有《GPRS网络信令实例详解》《EPC原理与实践》《5G核心网原理与实践》三本纸质书（各大电商平台有售）。
> **星球定位**：深耕 5G 核心网、IMS、NFV/AI、云原生等 ICT 领域，主打运维岗位与信令分析。
> **当前规模**：5200 篇主题 / 990 个回答，198 元/年。
> **加入福利**：加入星球后请加爱卫生微信，拉您进知识星球年度陪伴与答疑群——不限量回复您的各种问题，另有群内分享。
> **AI 加持**：51学通信知识星球已作为 skill 接入 WorkBuddy，使用 deepseek-4-flash 在海量星球帖子中精准找贴、找视频、找 FAQ 和案例，做总结、找文档和 PPT 等场景全覆盖。

![51学通信知识星球海报-扫码加入](https://wiki.500nian.com/images/zhixingqiu_qrcode.png)