MCP 驱动的 AI Agent 工作流：从任务分解到自动执行

AI Agent 的真正价值不在于回答问题，而在于自主完成任务。MCP 为 Agent 提供了丰富的工具集，使得复杂的自动化工作流成为可能。这篇文章将展示如何用 MCP 构建能够自主规划和执行任务的 Agent 工作流。

Agent 工作流的基本模式

ReAct 模式

ReAct（Reasoning + Acting）是最基础的 Agent 模式：LLM 先推理（Reasoning），再行动（Acting），观察结果后继续推理。

用户任务 → 推理 → 选择工具 → 执行 → 观察结果 → 推理 → ... → 最终回答

Plan-and-Execute 模式

先制定完整计划，再逐步执行：

用户任务 → 制定计划 [步骤1, 步骤2, ...] → 执行步骤1 → 执行步骤2 → ... → 汇总结果

反思模式

在执行后进行自我评估，必要时调整策略：

用户任务 → 执行 → 反思 → 调整 → 重新执行 → ... → 满意 → 输出

实现：自动化研究工作流

让我们实现一个完整的研究工作流——给定一个主题，Agent 自动搜索信息、分析数据、生成报告。

工作流架构

┌─────────────────────────────────────────┐
│              Orchestrator                │
│                                         │
│  ┌──────┐  ┌──────┐  ┌──────┐  ┌─────┐ │
│  │规划器│→│研究员│→│分析师│→│写作者│ │
│  └──────┘  └──────┘  └──────┘  └─────┘ │
│                                         │
└─────────────────────────────────────────┘
        ↓           ↓           ↓
   ┌────────┐  ┌────────┐  ┌────────┐
   │搜索 MCP│  │数据 MCP│  │文档 MCP│
   │ Server │  │ Server │  │ Server │
   └────────┘  └────────┘  └────────┘

实现代码

import Anthropic from '@anthropic-ai/sdk';
import { MCPClientManager } from './mcp-client.js';

const claude = new Anthropic();
const mcp = new MCPClientManager();

// 初始化 MCP Server
await mcp.addServer({
  name: 'search',
  command: 'npx',
  args: ['-y', '@modelcontextprotocol/server-brave-search'],
  env: { BRAVE_API_KEY: process.env.BRAVE_API_KEY! },
});

await mcp.addServer({
  name: 'filesystem',
  command: 'npx',
  args: ['-y', '@modelcontextprotocol/server-filesystem', '/tmp/research'],
});

// 阶段 1：制定研究计划
async function planResearch(topic: string): Promise<string[]> {
  const response = await claude.messages.create({
    model: 'claude-sonnet-4-20250514',
    max_tokens: 1000,
    messages: [{
      role: 'user',
      content: `为以下主题制定研究计划，输出 3-5 个具体的搜索查询：\n\n${topic}\n\n直接输出 JSON 数组格式的查询列表。`,
    }],
  });

  const text = response.content[0].text;
  return JSON.parse(text.match(/\[.*\]/s)?.[0] || '[]');
}

// 阶段 2：执行搜索
async function executeSearches(queries: string[]): Promise<string[]> {
  const results: string[] = [];

  for (const query of queries) {
    try {
      const result = await mcp.callTool('brave_web_search', {
        query,
        count: 5,
      });
      results.push(result.content[0].text);
    } catch (error) {
      console.error(`搜索失败: ${query}`, error);
    }
  }

  return results;
}

// 阶段 3：分析和综合
async function analyzeResults(topic: string, searchResults: string[]): Promise<string> {
  const response = await claude.messages.create({
    model: 'claude-sonnet-4-20250514',
    max_tokens: 4000,
    messages: [{
      role: 'user',
      content: `基于以下搜索结果，为主题 "${topic}" 生成一份深度分析报告。\n\n搜索结果：\n${searchResults.join('\n\n---\n\n')}\n\n要求：\n1. 提取关键事实和数据\n2. 识别趋势和模式\n3. 给出专业见解\n4. 输出 Markdown 格式`,
    }],
  });

  return response.content[0].text;
}

// 阶段 4：保存报告
async function saveReport(topic: string, report: string): Promise<string> {
  const filename = `/tmp/research/${topic.replace(/\s+/g, '-').toLowerCase()}.md`;

  await mcp.callTool('write_file', {
    path: filename,
    content: report,
  });

  return filename;
}

// 主工作流
async function researchWorkflow(topic: string) {
  console.error(`\n📋 研究主题: ${topic}\n`);

  // 阶段 1: 规划
  console.error('🔍 阶段 1: 制定研究计划...');
  const queries = await planResearch(topic);
  console.error(`  生成 ${queries.length} 个搜索查询`);

  // 阶段 2: 搜索
  console.error('🌐 阶段 2: 执行搜索...');
  const searchResults = await executeSearches(queries);
  console.error(`  收集 ${searchResults.length} 条搜索结果`);

  // 阶段 3: 分析
  console.error('📊 阶段 3: 分析和综合...');
  const report = await analyzeResults(topic, searchResults);

  // 阶段 4: 保存
  console.error('💾 阶段 4: 保存报告...');
  const path = await saveReport(topic, report);
  console.error(`\n✅ 报告已保存: ${path}`);

  return report;
}

// 执行
const report = await researchWorkflow('2026 年 MCP 协议发展趋势');
console.log(report);

await mcp.cleanup();

错误恢复机制

工作流中的每一步都可能失败，需要实现健壮的错误恢复：

async function resilientStep<T>(
  name: string,
  fn: () => Promise<T>,
  maxRetries: number = 3
): Promise<T> {
  for (let attempt = 1; attempt <= maxRetries; attempt++) {
    try {
      return await fn();
    } catch (error) {
      console.error(`[Step ${name}] 尝试 ${attempt}/${maxRetries} 失败:`, error.message);

      if (attempt === maxRetries) {
        throw new Error(`步骤 ${name} 在 ${maxRetries} 次尝试后仍然失败`);
      }

      // 指数退避
      await new Promise(r => setTimeout(r, 1000 * Math.pow(2, attempt - 1)));
    }
  }

  throw new Error('Unreachable');
}

// 使用
const results = await resilientStep('search', () => executeSearches(queries));

工作流状态管理

对于长时间运行的工作流，需要持久化状态：

interface WorkflowState {
  id: string;
  status: 'planning' | 'searching' | 'analyzing' | 'completed' | 'failed';
  currentStep: number;
  results: Record<string, any>;
  startedAt: string;
  updatedAt: string;
}

class WorkflowManager {
  private states: Map<string, WorkflowState> = new Map();

  create(id: string): WorkflowState {
    const state: WorkflowState = {
      id,
      status: 'planning',
      currentStep: 0,
      results: {},
      startedAt: new Date().toISOString(),
      updatedAt: new Date().toISOString(),
    };
    this.states.set(id, state);
    return state;
  }

  update(id: string, updates: Partial<WorkflowState>): WorkflowState {
    const state = this.states.get(id);
    if (!state) throw new Error(`Workflow ${id} not found`);

    Object.assign(state, updates, { updatedAt: new Date().toISOString() });
    return state;
  }

  get(id: string): WorkflowState | undefined {
    return this.states.get(id);
  }
}

常见问题（FAQ）

工作流中的 LLM 调用次数如何控制？

设置最大迭代次数。在 ReAct 模式中，限制推理-行动循环的次数。在 Plan-and-Execute 模式中，限制计划步骤的数量。

如何处理工作流中的歧义？

在关键决策点让 LLM 生成多个候选方案，根据置信度选择最合适的。如果置信度都不高，可以请求用户确认。

工作流可以并行执行吗？

可以。独立的步骤可以由不同的 Agent 并行执行。使用 Promise.all 或专门的工作流引擎来管理并行任务。

总结

MCP 为 Agent 工作流提供了标准化的工具调用能力。通过合理的模式选择（ReAct、Plan-and-Execute、反思模式）和健壮的错误处理，你可以构建出能够自主完成复杂任务的 AI Agent 系统。