实战教程：用 MCP + A2A 构建跨框架 Agent 协作系统（LangGraph + CrewAI + 纯 A2A）

// mcp-servers/postgres-server/server.ts
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import pg from "pg";

const pool = new pg.Pool({
  host: process.env.PG_HOST || "localhost",
  port: parseInt(process.env.PG_PORT || "5432"),
  database: process.env.PG_DATABASE || "market_data",
  user: process.env.PG_USER || "agent_reader",
  password: process.env.PG_PASSWORD,
  max: 5, // 连接池最大 5 个连接
  idleTimeoutMillis: 30000,
});

const server = new McpServer({
  name: "postgres-market-data",
  version: "1.0.0",
});

// 工具 1: 执行只读 SQL 查询
server.tool(
  "query",
  "执行只读 SQL 查询，返回结果集。仅支持 SELECT 语句。",
  {
    sql: z.string().describe("SELECT SQL 语句"),
    params: z.array(z.any()).optional().describe("SQL 参数（防注入）"),
  },
  async ({ sql, params }) => {
    // 安全检查：只允许 SELECT
    if (!sql.trim().toUpperCase().startsWith("SELECT")) {
      return {
        content: [{ type: "text", text: "错误: 仅允许 SELECT 查询" }],
        isError: true,
      };
    }
    
    const client = await pool.connect();
    try {
      const result = await client.query(sql, params || []);
      return {
        content: [
          {
            type: "text",
            text: JSON.stringify({
              rows: result.rows,
              rowCount: result.rowCount,
              fields: result.fields.map((f) => f.name),
            }, null, 2),
          },
        ],
      };
    } catch (err) {
      return {
        content: [{ type: "text", text: `查询错误: ${err.message}` }],
        isError: true,
      };
    } finally {
      client.release();
    }
  }
);

// 工具 2: 列出所有表
server.tool(
  "list_tables",
  "列出数据库中的所有用户表",
  {},
  async () => {
    const result = await pool.query(
      "SELECT table_name FROM information_schema.tables WHERE table_schema = 'public' ORDER BY table_name"
    );
    return {
      content: [
        {
          type: "text",
          text: result.rows.map((r) => r.table_name).join("\n"),
        },
      ],
    };
  }
);

// 工具 3: 描述表结构
server.tool(
  "describe_table",
  "获取指定表的列定义（名称、类型、是否可空）",
  {
    tableName: z.string().describe("表名"),
  },
  async ({ tableName }) => {
    const result = await pool.query(
      `SELECT column_name, data_type, is_nullable 
       FROM information_schema.columns 
       WHERE table_name = $1 
       ORDER BY ordinal_position`,
      [tableName]
    );
    return {
      content: [
        {
          type: "text",
          text: JSON.stringify(result.rows, null, 2),
        },
      ],
    };
  }
);

// 启动
const transport = new StdioServerTransport();
await server.connect(transport);
console.error("PostgreSQL MCP Server running on stdio");

// agents/data-analyst/index.ts
import { StateGraph, END } from "@langchain/langgraph";
import { ChatOpenAI } from "@langchain/openai";
import { MCPClient } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientParameters } from "@modelcontextprotocol/sdk/client/stdio.js";
import { A2AServer, Task, Message, Part, TaskStatus } from "@a2a-protocol/server";

// ─── 状态定义 ───
interface AnalysisState {
  request: string;           // 用户的分析需求
  tables: string[];          // 需要查询的表
  queries: string[];         // 生成的 SQL 查询
  results: any[];            // 查询结果
  report: string;            // 分析报告
  needsRevision: boolean;    // 是否需要修正
  messages: any[];           // LLM 消息历史
}

// ─── MCP Client 初始化 ───
const mcpClient = new MCPClient();
const serverParams: StdioClientParameters = {
  command: "node",
  args: ["../../mcp-servers/postgres-server/server.ts"],
};
await mcpClient.connect(serverParams);

// 获取可用工具列表
const tools = await mcpClient.listTools();
const pgTools = tools.tools.filter(t => 
  ["query", "list_tables", "describe_table"].includes(t.name)
);

// ─── LLM 初始化 ───
const llm = new ChatOpenAI({ model: "gpt-4o", temperature: 0 });
const llmWithTools = llm.bindTools(pgTools.map(t => ({
  type: "function",
  function: { name: t.name, description: t.description, parameters: t.inputSchema },
})));

// ─── 状态图节点 ───
async function understandNode(state: AnalysisState): Promise<Partial<AnalysisState>> {
  const response = await llmWithTools.invoke([
    { role: "system", content: "你是数据分析专家。根据需求确定需要查询的数据库表。先用 list_tables 了解可用表。" },
    { role: "user", content: state.request },
  ]);
  
  // 执行工具调用
  const tableList = await executeToolCalls(response.tool_calls);
  
  return {
    tables: extractTableNames(tableList),
    messages: [...state.messages, response],
  };
}

async function analyzeNode(state: AnalysisState): Promise<Partial<AnalysisState>> {
  const response = await llmWithTools.invoke([
    { role: "system", content: `基于以下表结构生成并执行分析查询:\n${JSON.stringify(state.tables)}` },
    { role: "user", content: state.request },
    ...state.messages.slice(-3),
  ]);
  
  const queryResults = await executeToolCalls(response.tool_calls);
  
  return {
    results: queryResults,
    queries: extractQueries(response.tool_calls),
    messages: [...state.messages, response],
  };
}

async function formatNode(state: AnalysisState): Promise<Partial<AnalysisState>> {
  const response = await llm.invoke([
    { role: "system", content: "将分析结果格式化为结构化报告，包含：摘要、关键发现、数据支撑、建议。" },
    { role: "user", content: `需求: ${state.request}\n结果: ${JSON.stringify(state.results)}` },
  ]);
  
  return { report: response.content as string };
}

async function reviewNode(state: AnalysisState): Promise<Partial<AnalysisState>> {
  const response = await llm.invoke([
    { role: "system", content: "检查分析结果是否合理。如果数据量异常（0行或>10000行），标记需要修正。" },
    { role: "user", content: `查询结果行数: ${state.results.length}\n报告: ${state.report}` },
  ]);
  
  return { needsRevision: response.content.includes("NEEDS_REVISION") };
}

// ─── 构建状态图 ───
const graph = new StateGraph<AnalysisState>({
  channels: {
    request: { value: (prev, next) => next, default: () => "" },
    tables: { value: (prev, next) => next, default: () => [] },
    queries: { value: (prev, next) => next, default: () => [] },
    results: { value: (prev, next) => next, default: () => [] },
    report: { value: (prev, next) => next, default: () => "" },
    needsRevision: { value: (prev, next) => next, default: () => false },
    messages: { value: (prev, next) => next, default: () => [] },
  },
});

graph.addNode("understand", understandNode);
graph.addNode("analyze", analyzeNode);
graph.addNode("format", formatNode);
graph.addNode("review", reviewNode);

graph.addEdge("understand", "analyze");
graph.addEdge("analyze", "format");
graph.addEdge("format", "review");
graph.addConditionalEdges("review", (state) => 
  state.needsRevision ? "analyze" : END
);

graph.setEntryPoint("understand");
const compiledGraph = graph.compile();

// ─── A2A Server 包装 ───
const a2aServer = new A2AServer({
  name: "数据分析 Agent",
  description: "专业的数据库分析 Agent，擅长 SQL 查询和市场数据分析",
  url: "http://localhost:3001/a2a",
  skills: [{
    id: "data-analysis",
    name: "数据库分析",
    tags: ["sql", "data", "analytics"],
  }],
});

a2aServer.onTask(async (task: Task): Promise<Task> => {
  const userRequest = task.messages[0].parts[0].text;
  
  const result = await compiledGraph.invoke({
    request: userRequest,
    tables: [],
    queries: [],
    results: [],
    report: "",
    needsRevision: false,
    messages: [],
  });
  
  task.status = TaskStatus.COMPLETED;
  task.artifacts = [{
    type: "text",
    name: "analysis_report",
    content: result.report,
  }];
  
  return task;
});

a2aServer.listen(3001);
console.log("数据分析 Agent 启动在 :3001 (A2A + LangGraph + MCP)");

# agents/writer/main.py
from crewai import Agent, Task, Crew, Process
from crewai_tools import MCPTool
from a2a_server import A2AServer
import asyncio

# ─── MCP 工具集成 ───
# 连接搜索引擎 MCP Server
search_mcp = MCPTool.from_server(
    server_command="node",
    server_args=["../../mcp-servers/search-server/server.js"],
    tools=["web_search", "fetch_page"],
)

# 连接文档工具 MCP Server
doc_mcp = MCPTool.from_server(
    server_command="node",
    server_args=["../../mcp-servers/doc-server/server.js"],
    tools=["create_doc", "export_pdf"],
)

# ─── 角色定义 ───
researcher = Agent(
    role="研究员",
    goal="收集与主题相关的高质量素材和数据",
    backstory="你是一位资深研究员，擅长从海量信息中筛选出最有价值的内容。"
              "你追求数据的准确性和来源的权威性。",
    tools=[search_mcp.web_search, search_mcp.fetch_page],
    allow_delegation=False,
    verbose=True,
)

outliner = Agent(
    role="大纲师",
    goal="根据研究素材构建逻辑清晰的文章大纲",
    backstory="你是一位结构思维专家，擅长将零散的信息组织成有说服力的叙事结构。"
              "你相信好的大纲是好文章的一半。",
    tools=[],
    allow_delegation=False,
    verbose=True,
)

writer = Agent(
    role="写手",
    goal="根据大纲撰写专业、深度、可读性强的报告",
    backstory="你是一位技术写作专家，擅长将复杂的概念用清晰的语言解释。"
              "你的文章以深度、准确、可读性强著称。",
    tools=[doc_mcp.create_doc, doc_mcp.export_pdf],
    allow_delegation=False,
    verbose=True,
)

# ─── A2A Server 包装 ───
a2a_server = A2AServer(
    name="写作 Agent",
    description="专业的报告写作 Agent，擅长将数据分析结果转化为深度报告",
    port=3002,
)

@a2a_server.on_task
async def handle_task(task):
    user_request = task.messages[0].parts[0].text
    
    # 定义任务链
    research_task = Task(
        description=f"研究主题: {user_request}\n要求: 收集最新数据、权威来源、关键案例",
        expected_output="结构化的研究素材，包含数据点、引用来源、关键洞察",
        agent=researcher,
    )
    
    outline_task = Task(
        description="基于研究素材构建文章大纲（8-10 个章节）",
        expected_output="层次分明的大纲，每个章节有 3-5 个要点",
        agent=outliner,
        context=[research_task],
    )
    
    writing_task = Task(
        description=f"根据大纲撰写完整报告\n用户原始需求: {user_request}",
        expected_output="5000+ 字的深度报告，包含数据支撑和专业分析",
        agent=writer,
        context=[outline_task],
        output_file="report.md",
    )
    
    # 执行 Crew
    crew = Crew(
        agents=[researcher, outliner, writer],
        tasks=[research_task, outline_task, writing_task],
        process=Process.sequential,  # 顺序执行
        verbose=True,
    )
    
    result = crew.kickoff()
    
    # 返回 A2A 结果
    task.status = "completed"
    task.artifacts = [{
        "type": "text",
        "name": "report",
        "content": result.raw,
    }]
    
    return task

a2a_server.start()
print("写作 Agent 启动在 :3002 (A2A + CrewAI + MCP)")

// agents/coordinator/index.ts
import { A2AClient, AgentCard, Task, Message, Part } from "@a2a-protocol/core";
import express from "express";

// ─── Agent 注册表（生产环境应使用 A2A Directory） ───
const AGENT_REGISTRY: Record<string, string> = {
  "data-analyst": "http://localhost:3001",
  "writer": "http://localhost:3002",
};

class CoordinatorAgent {
  private clients: Map<string, A2AClient> = new Map();

  async initialize() {
    // 初始化所有 Agent 的 A2A Client
    for (const [name, url] of Object.entries(AGENT_REGISTRY)) {
      const cardResp = await fetch(`${url}/.well-known/agent.json`);
      const card: AgentCard = await cardResp.json();
      this.clients.set(name, new A2AClient(card));
      console.log(`已连接 Agent: ${name} (${card.skills.map(s => s.id).join(", ")})`);
    }
  }

  async handleRequest(userInput: string): Promise<string> {
    console.log(`\n=== 收到用户请求 ===\n${userInput}\n`);

    // Step 1: 任务分解（使用 LLM 判断需要哪些 Agent）
    const plan = await this.decomposeTask(userInput);
    console.log(`任务计划: ${JSON.stringify(plan, null, 2)}`);

    // Step 2: 按依赖顺序执行
    const results: Record<string, string> = {};
    
    for (const step of plan.steps) {
      console.log(`\n--- 执行步骤: ${step.agent} ---`);
      
      const client = this.clients.get(step.agent);
      if (!client) {
        throw new Error(`Agent 不存在: ${step.agent}`);
      }

      // 注入前置步骤的结果
      const enrichedInput = this.injectContext(step.input, results);

      // 委托任务（带超时）
      try {
        const task = await client.sendTask({
          messages: [new Message({ role: "coordinator", parts: [new Part({ type: "text", text: enrichedInput })] })],
        });

        // 等待完成（最长 5 分钟）
        const completedTask = await client.waitForCompletion(task.id, {
          timeoutMs: 300_000,
          pollIntervalMs: 2_000,
        });

        // 提取结果
        const resultText = completedTask.artifacts
          .filter((a) => a.type === "text")
          .map((a) => a.content)
          .join("\n");

        results[step.agent] = resultText;
        console.log(`步骤完成: ${step.agent} → ${resultText.length} 字`);

      } catch (err) {
        console.error(`步骤失败: ${step.agent} → ${err.message}`);
        // 降级处理：使用空结果继续
        results[step.agent] = `[该步骤执行失败: ${err.message}]`;
      }
    }

    // Step 3: 整合最终结果
    return this.synthesizeResults(userInput, results, plan);
  }

  private async decomposeTask(input: string) {
    // 简化版：直接规则匹配（生产环境应使用 LLM 动态规划）
    if (input.includes("分析") && input.includes("报告")) {
      return {
        steps: [
          { agent: "data-analyst", input: input, dependsOn: [] },
          { agent: "writer", input: "基于以下数据分析结果撰写报告: {{data-analyst}}", dependsOn: ["data-analyst"] },
        ],
      };
    }
    // ... 更多规则
    return { steps: [{ agent: "writer", input, dependsOn: [] }] };
  }

  private injectContext(input: string, results: Record<string, string>): string {
    return input.replace(/\{\{(\w+-\w+)\}\}/g, (_, key) => results[key] || "[无前置结果]");
  }

  private synthesizeResults(input: string, results: Record<string, string>, plan: any): string {
    // 返回最后一个步骤的结果（简化版）
    const lastAgent = plan.steps[plan.steps.length - 1].agent;
    return results[lastAgent] || "任务执行完成，但未产生输出。";
  }
}

// ─── HTTP API 启动 ───
const app = express();
app.use(express.json());
const coordinator = new CoordinatorAgent();
await coordinator.initialize();

app.post("/api/chat", async (req, res) => {
  const { message } = req.body;
  const result = await coordinator.handleRequest(message);
  res.json({ response: result });
});

app.listen(3000, () => console.log("协调 Agent API: :3000"));

#!/bin/bash
# 一键启动整个系统的脚本
# start-all.sh

echo "🚀 启动跨框架 Agent 协作系统..."

# 1. 启动数据分析 Agent (LangGraph + MCP + A2A)
echo "📊 启动数据分析 Agent..."
cd agents/data-analyst
node index.ts &
DATA_PID=$!
cd ../..

# 2. 启动写作 Agent (CrewAI + MCP + A2A)
echo "✍️ 启动写作 Agent..."
cd agents/writer
python main.py &
WRITER_PID=$!
cd ../..

# 3. 等待 Agent 就绪
sleep 3

# 4. 启动协调 Agent
echo "🎯 启动协调 Agent..."
cd agents/coordinator
node index.ts &
COORD_PID=$!
cd ../..

echo ""
echo "✅ 系统启动完成！"
echo "   数据分析 Agent: http://localhost:3001 (PID: $DATA_PID)"
echo "   写作 Agent:     http://localhost:3002 (PID: $WRITER_PID)"
echo "   协调 Agent API: http://localhost:3000 (PID: $COORD_PID)"
echo ""
echo "测试命令："
echo '  curl -X POST http://localhost:3000/api/chat \'
echo '    -H "Content-Type: application/json" \'
echo '    -d '"'"'{"message": "分析 Q2 AI 市场趋势并写报告"}'"'"''
echo ""
echo "按 Ctrl+C 停止所有 Agent"

# 等待并处理退出信号
trap "kill $DATA_PID $WRITER_PID $COORD_PID 2>/dev/null; exit" SIGINT SIGTERM
wait

# 启动 MCP Inspector
npx @modelcontextprotocol/inspector node mcp-servers/postgres-server/server.ts

# 在 Inspector UI 中测试：
# 1. tools/list → 应该返回 3 个工具
# 2. tools/call("list_tables") → 应该返回表名列表
# 3. tools/call("query", {sql: "SELECT count(*) FROM market_data"}) → 应该返回行数

# 测试数据分析 Agent
curl -X POST http://localhost:3001/a2a/tasks \\
  -H "Content-Type: application/json" \\
  -d '{
    "messages": [{"role": "user", "parts": [{"type": "text", "text": "分析 market_data 表中 2026 Q2 的销售趋势"}]}]
  }'

# 应该返回 Task 对象，status 为 "completed"，artifacts 包含分析报告

# 通过协调 Agent 发起完整请求
curl -X POST http://localhost:3000/api/chat \\
  -H "Content-Type: application/json" \\
  -d '{"message": "分析 2026 Q2 中国 AI 市场趋势并生成深度报告"}'

# docker-compose.yml
version: "3.9"

services:
  # ─── 共享 MCP Servers ───
  postgres-mcp:
    build: ./mcp-servers/postgres-server
    environment:
      PG_HOST: postgres
      PG_PORT: 5432
      PG_DATABASE: market_data
      PG_USER: agent_reader
      PG_PASSWORD: ${PG_PASSWORD}
    networks:
      - agent-net
    healthcheck:
      test: ["CMD", "node", "-e", "process.exit(0)"]
      interval: 30s
      timeout: 5s
      retries: 3

  search-mcp:
    build: ./mcp-servers/search-server
    environment:
      SEARCH_API_KEY: ${SEARCH_API_KEY}
    networks:
      - agent-net

  # ─── Agent 服务 ───
  data-analyst:
    build: ./agents/data-analyst
    environment:
      OPENAI_API_KEY: ${OPENAI_API_KEY}
      A2A_PORT: 3001
      MCP_SERVER_URL: "http://postgres-mcp:3100"  # HTTP 模式
    ports:
      - "3001:3001"
    depends_on:
      postgres-mcp:
        condition: service_healthy
    networks:
      - agent-net
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:3001/.well-known/agent.json"]
      interval: 30s
      timeout: 10s
      retries: 3
    deploy:
      resources:
        limits:
          memory: 2G
          cpus: "1.0"

  writer:
    build: ./agents/writer
    environment:
      OPENAI_API_KEY: ${OPENAI_API_KEY}
      A2A_PORT: 3002
      MCP_SERVER_URL: "http://search-mcp:3100"
    ports:
      - "3002:3002"
    depends_on:
      search-mcp:
        condition: service_healthy
    networks:
      - agent-net
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:3002/.well-known/agent.json"]
      interval: 30s
      timeout: 10s
      retries: 3
    deploy:
      resources:
        limits:
          memory: 2G
          cpus: "1.0"

  coordinator:
    build: ./agents/coordinator
    environment:
      AGENT_REGISTRY: |
        {
          "data-analyst": "http://data-analyst:3001",
          "writer": "http://writer:3002"
        }
    ports:
      - "3000:3000"
    depends_on:
      - data-analyst
      - writer
    networks:
      - agent-net
    deploy:
      resources:
        limits:
          memory: 1G
          cpus: "0.5"

networks:
  agent-net:
    driver: bridge

// 协调 Agent 性能优化：并行执行 + 缓存 + 流式返回
import { A2AClient } from "@a2a-protocol/core";
import Redis from "ioredis";
import { createHash } from "crypto";

const cache = new Redis(process.env.REDIS_URL || "redis://localhost:6379");

class OptimizedCoordinator {
  // 优化 1: 并行执行无依赖任务
  async executeParallel(steps: Step[], results: Record<string, string>) {
    // 找出无依赖的步骤
    const parallelSteps = steps.filter(s => 
      s.dependsOn.length === 0 || s.dependsOn.every(d => results[d])
    );
    
    // 并行执行
    const parallelResults = await Promise.allSettled(
      parallelSteps.map(step => this.executeStep(step, results))
    );
    
    // 收集结果
    parallelSteps.forEach((step, i) => {
      const result = parallelResults[i];
      results[step.agent] = result.status === "fulfilled" 
        ? result.value 
        : `[失败: ${result.reason}]`;
    });
    
    return results;
  }

  // 优化 2: 结果缓存
  async executeWithCache(step: Step, input: string): Promise<string> {
    const cacheKey = `a2a:${step.agent}:${createHash("sha256").update(input).digest("hex").slice(0, 16)}`;
    
    // 检查缓存
    const cached = await cache.get(cacheKey);
    if (cached) {
      console.log(`缓存命中: ${step.agent}`);
      return cached;
    }
    
    // 执行并缓存（TTL 1 小时）
    const result = await this.executeStep(step, {});
    await cache.setex(cacheKey, 3600, result);
    
    return result;
  }

  // 优化 3: 流式返回进度
  async *executeStream(userInput: string): AsyncGenerator<string> {
    const plan = await this.decomposeTask(userInput);
    const results: Record<string, string> = {};
    
    for (const step of plan.steps) {
      yield `🔄 开始执行: ${step.agent}...\n`;
      
      try {
        const result = await this.executeWithCache(step, step.input);
        results[step.agent] = result;
        yield `✅ ${step.agent} 完成 (${result.length} 字)\n`;
      } catch (err) {
        yield `❌ ${step.agent} 失败: ${err.message}\n`;
      }
    }
    
    yield `\n📋 最终结果:\n`;
    yield this.synthesizeResults(userInput, results, plan);
  }
}