三协议归一后：MCP+A2A双栈多Agent协作模式实战拆解

┌─────────────────────────────────────────────────────┐
│                  应用层（业务逻辑）                    │
├─────────────────────────────────────────────────────┤
│  A2A层（水平）: Agent发现 · 任务委派 · 结果交换       │
│  Agent Card · Task生命周期 · 签名验证 · 多租户       │
├─────────────────────────────────────────────────────┤
│  MCP层（垂直）: 工具调用 · 资源读取 · Prompt模板      │
│  JSON-RPC · Streamable HTTP · 10K+ Server生态       │
├─────────────────────────────────────────────────────┤
│          传输层：HTTP/gRPC/JSON-RPC/stdio            │
└─────────────────────────────────────────────────────┘

sequenceDiagram
    participant User
    participant Orchestrator
    participant Researcher
    participant Coder
    participant Reviewer

    User->>Orchestrator: 提交需求
    Orchestrator->>Orchestrator: 解析Agent Card
    Orchestrator->>Researcher: A2A Task: 调研
    Researcher->>Researcher: MCP: 搜索/Web/数据库
    Researcher-->>Orchestrator: 返回调研结果
    Orchestrator->>Coder: A2A Task: 编码
    Coder->>Coder: MCP: GitHub/代码执行
    Coder-->>Orchestrator: 返回代码
    Orchestrator->>Reviewer: A2A Task: 审查
    Reviewer->>Reviewer: MCP: 测试/安全扫描
    Reviewer-->>Orchestrator: 返回审查报告
    Orchestrator-->>User: 汇总输出

from google.adk import Agent
from google.adk.a2a import RemoteA2aAgent

# 编排者发现远程Agent
researcher = RemoteA2aAgent("https://agents.corp.com/researcher")
coder = RemoteA2aAgent("https://agents.corp.com/coder")

orchestrator = Agent(
    name="orchestrator",
    model="gemini-2.5-pro",
    sub_agents=[researcher, coder],
    instruction="分解用户需求，委派给合适的子Agent执行"
)

# LangGraph流水线实现
from langgraph.graph import StateGraph

graph = StateGraph(PipelineState)
graph.add_node("extract", extract_agent)
graph.add_node("analyze", analyze_agent)  
graph.add_node("format", format_agent)
graph.add_node("publish", publish_agent)

graph.add_edge("extract", "analyze")
graph.add_edge("analyze", "format")
graph.add_edge("format", "publish")

# 每个节点通过MCP访问专属工具
# extract → MCP: PDF解析/OCR
# analyze → MCP: 数据库/BI工具
# format → MCP: 模板引擎/图表生成
# publish → MCP: Notion/Slack/Email

Agent A ←── A2A Task ──→ Agent B
  │                         │
  MCP                      MCP
(工具集A)                (工具集B)
  │                         │
  └──── 共享状态 ────────────┘

                    ┌─────────────┐
                    │  CEO Agent  │  战略层
                    └──────┬──────┘
              ┌────────────┼────────────┐
        ┌─────┴─────┐ ┌────┴────┐ ┌────┴─────┐
        │产品经理    │ │工程经理  │ │QA经理    │  战术层
        │Agent      │ │Agent    │ │Agent     │
        └─────┬─────┘ └────┬────┘ └────┬─────┘
           ┌──┴──┐     ┌───┼───┐     ┌─┴─┐
        设计  文案   前端 后端  DevOps  测试  安全  执行层

from crewai import Agent, Crew, Task

manager = Agent(
    role="项目总监",
    goal="协调各子团队高效完成项目",
    backstory="10年项目管理经验",
    allow_delegation=True  # 允许向下委派
)

frontend_lead = Agent(
    role="前端负责人",
    goal="管理前端开发团队",
    allow_delegation=True
)

crew = Crew(
    agents=[manager, frontend_lead, ...],
    tasks=[...],
    process="hierarchical",  # 层级模式
    manager_agent=manager
)

from langgraph.graph import StateGraph

graph = StateGraph(ParallelState)

# 添加并行节点（LangGraph自动并行执行无依赖节点）
graph.add_node("source_a", agent_query_a)
graph.add_node("source_b", agent_query_b)
graph.add_node("source_c", agent_query_c)
graph.add_node("merge", merge_agent)

graph.add_edge("source_a", "merge")
graph.add_edge("source_b", "merge")
graph.add_edge("source_c", "merge")
# source_a/b/c并行执行 → 全部完成后触发merge

class AdaptiveOrchestrator:
    def select_pattern(self, task: Task) -> str:
        complexity = task.estimate_complexity()
        dependencies = task.analyze_dependencies()
        agent_count = task.required_agents()

        if agent_count == 1:
            return "single_agent"          # 单Agent直接处理
        elif dependencies == "sequential":
            return "pipeline"              # 流水线
        elif dependencies == "independent":
            return "parallel"              # 并行
        elif agent_count > 7:
            return "hierarchical"          # 层级控制
        else:
            return "orchestrator_worker"   # 编排者-工作者

graph TB
    subgraph "A2A层 - Agent间通信"
        RG[路由Agent] -->|A2A Task| BA[计费Agent]
        RG -->|A2A Task| TA[技术支持Agent]
        RG -->|A2A Task| SA[销售Agent]
    end

    subgraph "MCP层 - 工具访问"
        BA -->|MCP| SF[(Salesforce)]
        BA -->|MCP| ST[(Stripe)]
        TA -->|MCP| KB[(知识库)]
        TA -->|MCP| JIRA[(Jira)]
        SA -->|MCP| CRM[(CRM)]
        SA -->|MCP| EM[(Email)]
    end

    subgraph "基础设施"
        AR[Agent Card<br/>注册中心]
        GW[MCP网关<br/>权限/审计/限流]
        OB[可观测性<br/>OpenTelemetry]
    end

    RG --> AR
    BA --> AR
    TA --> AR
    SA --> AR
    BA --> GW
    TA --> GW
    SA --> GW
    RG --> OB

{
  "name": "BillingAgent",
  "version": "1.0.0",
  "capabilities": ["invoice_query", "payment_processing"],
  "supportedInterfaces": [{
    "protocolVersion": "1.0",
    "transport": "grpc"
  }],
  "authentication": {
    "type": "oauth2",
    "scopes": ["read:billing", "write:payments"]
  }
}

你的任务需要几个Agent？
├─ 1个 → 单Agent + MCP（不需要A2A）
├─ 2-7个
│  ├─ 任务有依赖关系？
│  │  ├─ 是，且顺序固定 → 流水线模式
│  │  └─ 是，但需动态路由 → 编排者-工作者模式
│  └─ 任务完全独立？
│     ├─ 是 → 并行执行模式
│     └─ 否，需要协商 → 对等协作模式
└─ 7个以上
   └─ 层级控制模式（上层管理下层）

需求多样化、流量波动大？
└─ 混合自适应模式

应用层    业务逻辑
协调层    A2A (Agent间通信)
工具层    MCP (Agent到工具)
商业层    AP2/UCP (支付与SLA)  ← 即将到来
发现层    ANP (全局Agent发现)  ← 探索阶段