Strands Agents A2A 协议实战：让多个 AI Agent 互相对话
最近在搞一个业务场景——订单处理 Agent 需要调用库存查询 Agent 的能力，库存不够时还要找采购 Agent 下补货单。
三个 Agent 各有各的职责，问题是怎么让它们互相"交流"。
翻了一下 Strands Agents SDK 的最新版本，发现它刚加了 A2A（Agent-to-Agent）协议支持，专门解决多 Agent 通信的问题。Agent 之间不用共享内存或数据库，而是通过标准 HTTP 协议互相发消息。
A2A 协议是什么
A2A 是一个开放协议（由 Google 发起，AWS 跟进支持），定义了 Agent 之间的通信标准：
- 每个 Agent 暴露一个
AgentCard（类似 API 文档） - Agent 之间通过 HTTP JSON 消息通信
- 支持同步请求-响应和异步任务
- 有标准的能力发现机制
类比一下：MCP 是"Agent 调工具"，A2A 是"Agent 调 Agent"。
搭建第一个 A2A Agent
安装 Strands SDK（需要 0.3.0+）：
pip install strands-agents strands-agents-tools
先写一个库存查询 Agent，让它暴露 A2A 接口：
from strands import Agent
from strands.tools import tool
from strands.a2a import A2AServer@tool
def check_inventory(product_id: str) -> dict:"""查询产品库存数量"""# 实际场景连 DynamoDBinventory_db = {"SKU-001": {"name": "机械键盘", "stock": 42, "warehouse": "华东"},"SKU-002": {"name": "无线鼠标", "stock": 0, "warehouse": "华东"},"SKU-003": {"name": "显示器支架", "stock": 15, "warehouse": "华南"},}item = inventory_db.get(product_id)if not item:return {"error": f"产品 {product_id} 不存在"}return item@tool
def batch_check(product_ids: list) -> list:"""批量查询多个产品库存"""results = []for pid in product_ids:results.append({"product_id": pid, **check_inventory(pid)})return results# 创建库存 Agent
inventory_agent = Agent(model="us.anthropic.claude-sonnet-4-20250514-v1:0",system_prompt="""你是库存查询助手。- 收到产品ID时查询库存- 返回库存数量、仓库位置- 库存为0时标记"缺货"""",tools=[check_inventory, batch_check]
)# 启动 A2A Server
server = A2AServer(agent=inventory_agent,name="inventory-agent",description="查询产品库存数量和仓库位置",port=8001
)
server.run()
订单 Agent 调用库存 Agent
另一个服务里，订单 Agent 通过 A2A 协议调用库存 Agent：
from strands import Agent
from strands.a2a import A2AClient
from strands.tools import tool# 连接库存 Agent
inventory_client = A2AClient(agent_card_url="http://inventory-agent:8001/.well-known/agent.json"
)@tool
def query_inventory(product_id: str) -> dict:"""通过 A2A 协议调用库存 Agent 查询库存"""response = inventory_client.send_message(message=f"查询产品 {product_id} 的库存")return response.result@tool
def create_order(product_id: str, quantity: int, customer_id: str) -> dict:"""创建订单"""# 先检查库存stock_info = query_inventory(product_id)if stock_info.get("stock", 0) < quantity:return {"status": "failed","reason": f"库存不足，当前库存: {stock_info.get('stock', 0)}"}return {"status": "success","order_id": f"ORD-{customer_id}-{product_id}","product_id": product_id,"quantity": quantity,"warehouse": stock_info.get("warehouse")}order_agent = Agent(model="us.anthropic.claude-sonnet-4-20250514-v1:0",system_prompt="你是订单处理助手。收到下单请求时先查库存再创建订单。",tools=[query_inventory, create_order]
)# 这个 Agent 也可以暴露 A2A 接口
order_server = A2AServer(agent=order_agent,name="order-agent",description="处理订单创建、查询库存、生成物流单",port=8002
)
order_server.run()
部署到 ECS
三个 Agent 部署为 ECS 服务，互相通过 Service Discovery 发现：
# docker-compose.yml（本地开发用）
version: "3.8"
services:inventory-agent:build: ./inventory-agentports:- "8001:8001"environment:- AWS_REGION=us-east-1- BEDROCK_MODEL_ID=us.anthropic.claude-sonnet-4-20250514-v1:0order-agent:build: ./order-agentports:- "8002:8002"environment:- AWS_REGION=us-east-1- INVENTORY_AGENT_URL=http://inventory-agent:8001- BEDROCK_MODEL_ID=us.anthropic.claude-sonnet-4-20250514-v1:0procurement-agent:build: ./procurement-agentports:- "8003:8003"environment:- AWS_REGION=us-east-1- INVENTORY_AGENT_URL=http://inventory-agent:8001- ORDER_AGENT_URL=http://order-agent:8002- BEDROCK_MODEL_ID=us.anthropic.claude-sonnet-4-20250514-v1:0
生产环境用 ECS Service Connect 或 Cloud Map 做服务发现：
# 创建 namespace
aws servicediscovery create-private-dns-namespace \--name agents.local \--vpc vpc-0123456789abcdef0# 注册服务
aws servicediscovery create-service \--name inventory-agent \--namespace-id ns-xxx \--dns-config "NamespaceId=ns-xxx,DnsRecords=[{Type=A,TTL=30}]"
AgentCard 自动发现
A2A 协议的核心是 AgentCard——每个 Agent 在 /.well-known/agent.json 暴露自己的能力描述：
{"name": "inventory-agent","description": "查询产品库存数量和仓库位置","url": "http://inventory-agent:8001","version": "1.0.0","capabilities": {"streaming": true,"pushNotifications": false},"skills": [{"id": "check-inventory","name": "查询单个产品库存","description": "输入产品ID，返回库存数量和仓库位置"},{"id": "batch-check","name": "批量库存查询","description": "同时查询多个产品的库存状态"}]
}
其他 Agent 只需要知道 AgentCard URL，就能自动发现对方的能力并调用。
踩坑记录
-
超时设置：Agent 调 Agent 比调普通 API 慢（因为中间有 LLM 推理），默认 30s 超时可能不够。建议设到 120s。
-
重试策略：LLM 推理偶尔会失败（throttling、模型过载），A2A 调用要加指数退避重试。
-
消息大小：如果 Agent 返回的结果很大（比如批量查询 1000 个 SKU），可能超出下游 Agent 的 context window。建议分页。
-
认证：A2A 默认不带认证。生产环境要加 IAM Auth 或 mutual TLS：
server = A2AServer(agent=inventory_agent,name="inventory-agent",port=8001,auth_handler=iam_auth_handler # 自定义 IAM 认证
)
- 可观测性：A2A 调用链不会自动上报 X-Ray。需要手动在 A2AClient 里注入 trace header。
参考链接
- Strands Agents SDK：https://github.com/strands-agents/sdk-python
- A2A 协议规范：https://github.com/google/a2a-protocol
- Strands 文档：https://docs.strands.aws/
- Amazon Bedrock 模型列表：https://docs.aws.amazon.com/bedrock/latest/userguide/models-supported.html