多 Agent 编排
我们将使用1 个 Host Agent 通过 A2A 同时指挥 2 个 Remote Agent。
搭一个最小但完整的多 Agent 编排系统：Host Agent 发现两个远程 Agent（计算器 + 骰子），把用户请求拆成子任务并并行委派，最后汇总结果。
这是 A2A 最核心的应用形态，也是企业"Agent 团队"的雏形。
架构
用户请求
"算 42*17，然后再掷一次 2d6"
│
▼
┌───────────────────┐
│ Host Agent │ ① 读取两张 Agent Card
│ （编排者/Client） │ ② 按技能标签路由子任务
└─────┬───────┬─────┘ ③ asyncio 并行委派
A2A │ │ A2A
┌───────────┘ └───────────┐
▼ ▼
┌──────────────────┐ ┌──────────────────┐
│ Calculator Agent │ │ Dice Agent │
│ localhost:10001 │ │ localhost:10002 │
│ 技能标签: math │ │ 技能标签: dice │
└──────────────────┘ └──────────────────┘
（两个独立进程/服务，Host 不知道也不关心其内部实现）
Remote Agents 代码
创建一个remote_agents.py：
一个进程里用 uvicorn 同时起两个独立的 A2A Server（生产中它们通常是 不同团队、不同机器上的服务）：
"""实战三：两个远程 Agent（计算器 + 骰子），供 Host Agent 委派任务。
运行: ./venv/bin/python remote_agents.py
- Calculator Agent -> http://localhost:10001
- Dice Agent -> http://localhost:10002
"""
import asyncio
import random
import re
import uvicorn
from starlette.applications import Starlette
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
from a2a.types import (
AgentCapabilities,
AgentCard,
AgentInterface,
AgentSkill,
Part,
Task,
TaskState,
TaskStatus,
)
class BaseExecutor(AgentExecutor):
"""公共骨架：创建 Task -> working -> 业务逻辑 -> artifact -> completed"""
def handle(self, user_text: str) -> str: # 子类实现
raise NotImplementedError
async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
if context.current_task is None:
await event_queue.enqueue_event(
Task(
id=context.task_id,
context_id=context.context_id,
status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED),
history=[context.message] if context.message else [],
)
)
updater = TaskUpdater(event_queue, context.task_id, context.context_id)
await updater.start_work()
result = self.handle(context.get_user_input())
await updater.add_artifact([Part(text=result)], name="result")
await updater.complete()
async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
await TaskUpdater(event_queue, context.task_id, context.context_id).cancel()
class CalcExecutor(BaseExecutor):
def handle(self, user_text: str) -> str:
m = re.search(r"(-?\d+(?:\.\d+)?)\s*([+\-*/xX×])\s*(-?\d+(?:\.\d+)?)", user_text)
if not m:
return "没有找到算式，请输入如 '42 * 17'"
a, op, b = float(m.group(1)), m.group(2), float(m.group(3))
if op in "xX×":
op = "*"
val = {"+": a + b, "-": a - b, "*": a * b, "/": a / b if b else float("nan")}[op]
return f"{m.group(1)} {op} {m.group(3)} = {val:g}"
class DiceExecutor(BaseExecutor):
def handle(self, user_text: str) -> str:
m = re.search(r"(\d+)\s*[dD]\s*(\d+)", user_text)
n, faces = (int(m.group(1)), int(m.group(2))) if m else (1, 6)
rolls = [random.randint(1, faces) for _ in range(min(n, 100))]
return f"掷 {n}d{faces}: {rolls}，总和 = {sum(rolls)}"
def make_app(name: str, desc: str, port: int, skill: AgentSkill,
executor: AgentExecutor) -> tuple[AgentCard, Starlette]:
card = AgentCard(
name=name,
description=desc,
version="1.0.0",
supported_interfaces=[
AgentInterface(url=f"http://localhost:{port}/",
protocol_binding="JSONRPC", protocol_version="1.0")
],
capabilities=AgentCapabilities(streaming=True),
default_input_modes=["text/plain"],
default_output_modes=["text/plain"],
skills=[skill],
)
handler = DefaultRequestHandler(
agent_executor=executor, task_store=InMemoryTaskStore(), agent_card=card
)
app = Starlette(routes=[
*create_agent_card_routes(card),
*create_jsonrpc_routes(handler, rpc_url="/"),
])
return card, app
async def main() -> None:
_, calc_app = make_app(
"Calculator Agent", "会做四则运算的 Agent", 10001,
AgentSkill(id="calc", name="四则运算", description="计算 a op b",
tags=["math"], examples=["42 * 17"]),
CalcExecutor(),
)
_, dice_app = make_app(
"Dice Agent", "会掷骰子的 Agent", 10002,
AgentSkill(id="roll", name="掷骰子", description="按 NdM 掷骰子",
tags=["dice"], examples=["2d6"]),
DiceExecutor(),
)
servers = [
uvicorn.Server(uvicorn.Config(calc_app, host="0.0.0.0", port=10001,
log_level="warning")),
uvicorn.Server(uvicorn.Config(dice_app, host="0.0.0.0", port=10002,
log_level="warning")),
]
await asyncio.gather(*(s.serve() for s in servers))
if __name__ == "__main__":
asyncio.run(main())
Host Agent 代码
创建一个host_agent.py文件：
"""实战三：Host Agent（编排者）——发现两个远程 Agent 并按需委派子任务。
运行: 先启动 remote_agents.py，再 ./venv/bin/python host_agent.py
"""
import asyncio
import uuid
import httpx
from a2a.client import A2ACardResolver, create_client
from a2a.types import AgentCard, Message, Part, Role, SendMessageRequest
REMOTE_URLS = ["http://localhost:10001", "http://localhost:10002"]
USER_REQUEST = "帮我算一下 42 * 17，然后再掷一次 2d6 骰子"
async def discover(url: str) -> AgentCard:
"""能力发现：读取远程 Agent 的 Agent Card。"""
async with httpx.AsyncClient() as http:
return await A2ACardResolver(http, url).get_agent_card()
async def delegate(card: AgentCard, text: str) -> str:
"""通过 A2A 把子任务委派给远程 Agent，返回其产物文本。"""
client = await create_client(card)
request = SendMessageRequest(
message=Message(
message_id=str(uuid.uuid4()),
role=Role.ROLE_USER,
parts=[Part(text=text)],
)
)
outputs: list[str] = []
async for event in client.send_message(request):
if event.HasField("artifact_update"):
outputs.append("".join(
p.text for p in event.artifact_update.artifact.parts
if p.HasField("text")
))
await client.close()
return "\n".join(outputs)
def pick_agent(cards: list[AgentCard], keywords: list[str]) -> AgentCard:
"""极简路由：根据 Agent Card 的技能标签选择合适的 Agent。
（真实系统中这一步通常交给 LLM 阅读各 Agent Card 后决策）"""
for card in cards:
for skill in card.skills:
if any(k in skill.tags for k in keywords):
return card
raise RuntimeError(f"没有 Agent 能处理: {keywords}")
async def main() -> None:
# 1) 发现所有远程 Agent
cards = list(await asyncio.gather(*(discover(u) for u in REMOTE_URLS)))
print("[Host] 已发现的远程 Agent:")
for c in cards:
tags = sorted({t for s in c.skills for t in s.tags})
print(f" - {c.name}: {c.description} (技能标签: {tags})")
print(f"\n[Host] 用户请求: {USER_REQUEST}")
# 2) 拆分任务并路由（演示用规则路由；生产中可用 LLM 决策）
subtasks = [
(pick_agent(cards, ["math"]), "42 * 17"),
(pick_agent(cards, ["dice"]), "2d6"),
]
# 3) 并行委派两个子任务
results = await asyncio.gather(
*(delegate(card, text) for card, text in subtasks)
)
# 4) 汇总结果
print("\n[Host] 汇总结果:")
for (card, text), result in zip(subtasks, results):
print(f" - 委派给 {card.name}（输入: {text}）-> {result}")
if __name__ == "__main__":
asyncio.run(main())
编排的步骤：
- 发现：并行拉取所有远程 Agent 的 Card；
- 路由：
pick_agent()根据 Card 中技能的tags挑选执行者。演示用关键词规则；生产中通常把所有 Agent Card 喂给 LLM，让它阅读技能描述后决定"谁干什么”（Google ADK、LangGraph 都内置这类 A2A 路由能力）； - 委派与汇总：
asyncio.gather并行发出 A2A 任务，收集 Artifact。
运行
./venv/bin/python remote_agents.py # 终端 1：起两个远程 Agent
./venv/bin/python host_agent.py # 终端 2：编排
A2A + MCP 组合架构
真实企业系统中两个协议各司其职：
用户
│
▼
┌──────────────────┐
│ Host Agent │（LLM 编排：读 Card 做路由）
└───┬─────────┬────┘
A2A（横向）│ │ A2A（横向）
┌───────┘ └────────┐
▼ ▼
┌─────────────────┐ ┌─────────────────┐
│ 财务分析 Agent │ │ 行程规划 Agent │
│ (LangGraph) │ │ (Google ADK) │
└───┬─────────┬───┘ └───┬─────────┬───┘
MCP │ │ MCP MCP │ │ MCP
（纵向） ▼ ▼ ▼ ▼
┌────────┐ ┌─────────┐ ┌─────────┐ ┌──────────┐
│ 数据库 │ │ 报表API │ │ 航班API │ │ 日历服务 │
└────────┘ └─────────┘ └─────────┘ └──────────┘
-
A2A 连接 Agent（横向）：发现、委派、多轮协作、状态跟踪；
-
MCP 连接工具（纵向）：每个 Agent 内部用 MCP 调自己的工具；
总结
A2A什么现在值得关注：
- 行业标准，不站队：Google 发起、Linux Foundation 中立治理，微软（Azure AI Foundry）、AWS、SAP、Salesforce 等均已表态支持，规范已到 v1.0；
- 保护投资：各团队可继续用自己的框架（LangGraph/ADK/CrewAI…），只要对外讲 A2A 就能互通，避免 N×N 定制集成；
- 企业级设计：复用 OAuth2 等现有安全体系；“不透明执行"保证跨部门/跨公司协作时不泄露内部提示词、数据和工具；
- 与 MCP 互补：MCP 管 Agent 用工具，A2A 管 Agent 找同事，两个标准拼出完整的企业 Agent 架构；
- 落地成本低：百行 Python 即可让现有服务具备 A2A 能力。