Strands中A2A协议实践：远程Agent路由与动态发现

作为 Strands Agents 多智能体编排系列文章的收尾，我将讲解如何基于 A2A 协议实现远程智能体，使用确定路由和动态发现机制。我通过两个用例来演示：privacy-aware-routing（隐私感知路由）和 multi-agent-trip-planner（多智能体旅行规划器）。测试环境使用了 Gemma4 + Ollama（本地）以及 Gemini API（免费层）。

核心思路是让你在自己的 notebook 中用简单的例子实验 A2A 通信协议的各种场景，无需 AWS 账号或 Bedrock 等服务。Strands 作为协议的包装层，使得与远程 AI 智能体的通信变得流畅，不论它们使用什么框架或语言实现。

🗒 我选定的用例

Privacy aware routing（隐私感知路由）：一个面向客户的金融/银行业务助手（编排器，通过 API key 调用 Gemini）需要处理各种查询，有时会涉及敏感数据（如账号、DNI、账户余额）。编排器不直接让通用智能体处理，而是检测到查询涉及敏感数据时，将这部分委托给一个本地 A2A 智能体（独立服务器，通过 Ollama 运行 Gemma，监听 localhost）。

Multi agent trip planner（多智能体旅行规划器）：一个面向客户的旅行助手，编排器（通过 API key 调用 Gemini）事先不知道该将查询的哪部分委托给哪个智能体。它在运行时动态发现三个专业智能体：航班、酒店和行程。每个都作为本地 A2A 服务器暴露（通过 Ollama 运行 Gemma），并根据需要解决的问题动态决定调用哪一个。

接下来的章节我将介绍 A2A 协议在 Strands 中的关键概念，以及这两个用例的设计与实现。

➡ Strands 中的 A2A 协议

A2A（Agent-to-Agent）是一个开放协议——并非 Strands 或 AWS 的独创——它定义了运行在不同进程、不同机器甚至不同组织中的智能体如何相互发现和通信，底层协议为 HTTP/JSON-RPC。

Strands 没有重新发明协议：它用一层熟悉的接口将其包装起来，就像我们调用普通 Strands 智能体一样。开发者看不到协议，只看到一个智能体。简单示例：

from strands.agent.a2a_agent import A2AAgent

# Create an A2AAgent pointing to a remote A2A server
privacy_agent = A2AAgent(endpoint="http://localhost:9000")

# Invoked exactly like a local Agent — Strands' wrapper keeps
# the A2A protocol completely hidden from the caller
result = privacy_agent("I need the balance for account 1234-5678-9")

print(result.message)
# {'role': 'assistant', 'content': [{'text': 'The balance for account 1234-5678-9 is $85,400'}]}

A2AAgent 的构造函数接受多个参数：

两个组件共同实现了 Strands 中的包装：

A2AServer：将任意 Strands 智能体暴露为 A2A 服务，自动在 /.well-known/agent-card.json 发布一张 agent card（智能体卡片）→ 这是智能体的自我介绍（名称、描述、提供的能力）。

A2AAgent：客户端侧。包装一个远程智能体的端点，用与本地智能体相同的接口调用它。

与本地 Agent 一样，A2AAgent 支持异步调用（invoke_async）和响应流式传输（stream_async）。为简化起见，本文示例中使用同步调用，但如果你的用例需要不阻塞主线程或逐 token 显示响应，这些变体都是可用的。

远程智能体的元数据就是 agent card，可以通过 get_agent_card() 显式查询：

card = await privacy_agent.get_agent_card()

print(f"Agent: {card.name}")
print(f"Description: {card.description}")
print(f"Skills: {card.skills}")

agent card 中的每个 skill 描述了智能体的一项具体能力：标识符、自然语言描述，以及可选的使用示例。这使得编排器能够推理出"这个智能体会查航班"，而无需开发者显式编码。委托给谁的决策基于用户查询与所发现 skill 之间的匹配。

上面这个方法在 multi-agent-trip-planner 示例中将真正大放异彩：在那里，编排器不是调用一个事先已知的端点，而是查询多个远程智能体的 agent card，动态决定将每个任务委托给谁。

➡ A2AAgent 与 Strands 其他编排模式的组合

A2AAgent 可以与 Strands 中其他多智能体编排模式结合使用。

作为编排器智能体的 tool：这正是我们在 privacy-aware-routing POC 中采用的方式——A2AAgent 被包装成 @tool，父智能体像调用普通函数一样调用它，就像调用一个子智能体。

作为 Graph 中的节点：A2AAgent 可以作为远程节点集成到 Graph 流程中，在同一个依赖图中混合本地和远程智能体。

在 Swarm 中：目前不支持，因为该模式依赖于基于 tool 的移交机制，而 A2A 协议尚未支持这一特性。

看过客户端视角后，现在来看看如何暴露一个智能体供其他智能体通过 A2A 消费。

➡ 创建 A2A Server

在服务端，将 Strands 智能体暴露为 A2A 服务非常简单，只需用 A2AServer 包装它：

from strands import Agent
from strands.multiagent.a2a import A2AServer

def create_agent(context_id: str) -> Agent:
    return Agent(
        name="Privacy Agent",
        description="Resolves queries containing sensitive customer data",
    )

a2a_server = A2AServer(agent_factory=create_agent)
a2a_server.serve()

服务器自动在 /.well-known/agent-card.json 发布 agent card，并在配置的端口（默认 9000）上监听 A2A 请求。

一个重要的设计细节：建议传递一个 agent_factory（创建新智能体的函数）而不是单一的 Agent 实例。原因是按会话隔离：A2A 用 context_id 标识每个会话，服务器为每个 context_id 创建并复用专属的智能体，使不同会话永远不会相互混淆历史记录。传递共享的单一 Agent 实例已废弃！

对于本文的用例，这种隔离不是核心关注点：每个服务器处理一种特定任务，不维护大量会话上下文。但如果你的远程智能体需要在多轮对话中记住会话线索，这个概念是关键所在。

➡ Strands A2A Tool：动态发现

当有多个远程智能体可用，又不想硬编码"什么任务对应什么智能体"的映射时，Strands 提供了 A2AClientToolProvider（属于 strands-agents-tools 包）。它为编排器智能体提供了一套 tools，用于发现 A2A 智能体并用自然语言与它们通信：

from strands import Agent
from strands_tools.a2a_client import A2AClientToolProvider

# known_agent_urls 是可选的：可以传入已知端点，
# 也可以让 provider 在运行时发现它们
provider = A2AClientToolProvider(
    known_agent_urls=[
        "http://localhost:9001",  # Flights Agent
        "http://localhost:9002",  # Hotels Agent
        "http://localhost:9003",  # Itinerary Agent
    ]
)

orchestrator = Agent(tools=provider.tools)
response = orchestrator("Quiero ir a Barcelona en marzo, 5 días")

该 provider 暴露三种能力：

Agent Discovery：自动发现可用的 A2A 智能体及其能力（通过读取它们的 agent card）。

Protocol Communication：使用标准化的 A2A 协议发送消息。

Natural Language Interface：编排器用自然语言与远程智能体交互，开发者无需显式编码"如果查询是关于航班的，就调用这个端点"。

这是本文 multi-agent-trip-planner POC 的核心：在这里 agent card 不再只是一个实现细节，而成为让编排器能够推理将每个子任务委托给谁的机制——且这个映射关系没有写在代码里。

➡ 评估（Evals）

Strands 拥有一套官方评估框架，从简单的结果评估到复杂的多智能体交互分析都能支持。

为了评估这些工作流，我使用了 strands-agents-evals，配置与 .env 中相同的 provider（Gemini）作为评判模型，通过 LiteLLM 调用，以保持项目使用单一凭证且无需 AWS 账户。在生产环境中，标准做法是使用一个更强大、更独立的模型作为 judge。

在两个项目中，我结合了两种类型的评估，以展示 SDK 提供的两个家族：

确定性评估（纯代码，无需 LLM）：快速且廉价，非常适合 CI（持续集成）。它通过检查编排器调用了哪些工具（其轨迹）来验证决策。这是每个 A2A 模式的核心。

LLM-as-a-judge：用一个模型配合评分标准，对最终响应的更主观的维度打分：Correctness（数据是否与参考答案一致？）和 Faithfulness（是否基于工具的数据，没有编造或回避？）。

区别在于确定性层在每个模式中测量的是什么，因为每个模式做出的决策不同：

解决方案描述

📦 一个 GitHub 仓库包含两个项目：github.com/reinalau/strands-a2a-patterns

该仓库包含两个独立项目，每个项目都是一个 A2A 用例：

privacy-aware-routing —— 向单个远程 AI 智能体的确定性路由：编排器根据查询内容决定是自己回答，还是将敏感部分委托给一个 100% 本地运行的 AI 智能体。

multi-agent-trip-planner —— 动态发现 AI 智能体：编排器并不预先了解各专家 AI 智能体，而是在运行时通过读取它们的 agent cards 来发现它们，并将每个子任务委托给相应的 AI 智能体。此外，每个 AI 智能体都有可用的工具。

两者共享相同的技术基础（Strands，以 Gemini 作为编排器，本地 Gemma 作为远程 AI 智能体），以相同的方式运行。以下设置对两者同样适用。使用两个模型：通过 Ollama 本地运行的和通过 API key 调用 Gemini 的：

a. Ollama + 小型模型 gemma4:e2b-it-qat（大小仅略超过 4gb），在 Docker 中运行。为此，你需要安装并运行 Docker Desktop；启动后，下面的命令创建 Ollama 容器并下载模型。在 multi-agent-trip-planner 的情况下，有三个远程 AI 智能体可以几乎同时接收请求；如果你希望这些调用并行执行，需要告诉 Ollama，因为它的默认行为是一次处理一个推理：

# Start the Ollama server with a persistent volume
docker run -d --name ollama -p 11434:11434 -v ollama_data:/root/.ollama ollama/ollama

# Start the Ollama server with real parallelism (3 simultaneous requests)
# docker run -d --name ollama -p 11434:11434 -v ollama_data:/root/.ollama -e OLLAMA_NUM_PARALLEL=3  ollama/ollama

# Download the model
docker exec -it ollama ollama pull gemma4:e2b-it-qat

# Test that the model responds
# If the container has already been created and 
# is currently stopped, simply use: docker start ollama
docker exec -it ollama ollama run gemma4:e2b-it-qat

# Verify that the model is running
docker exec -it ollama ollama ps

b. Gemini API。可以从这里免费生成 API key，至少可以使用以下模型（尝试你的账户允许的模型）：

gemini-3.6-flash
gemini-3.5-flash-lite

两个项目的代码都是 Python，结构设计得更具解释性；所有细节都在每个项目的 README.md 中：

strands-a2a-patterns/
├── README.md                          # Overview of both use cases
│
├── privacy-aware-routing/             # Case 1: deterministic routing, single remote agent
│   ├── README.md
│   ├── requirements.txt
│   ├── .env.example
│   ├── common/                        # shared: prompts, config, logging
│   ├── remote_agent/
│   │   ├── server.py                  # A2AServer — Gemma via Ollama
│   │   └── bank_tools.py              # mock bank tools with fixed data
│   ├── orchestrator/
│   │   └── main.py                    # Gemini — A2AAgent as tool
│   ├── tests/
│   │   ├── test_routing_logic.py      # does the orchestrator delegate when it should?
│   │   └── test_a2a_integration.py    # real integration test against the running local server
│   ├── evals/
│   │   ├── eval_routing.py            # deterministic: routing decision
│   │   └── eval_response_quality.py   # LLM-judge: correctness + faithfulness
│   └── logs/
│       └── .gitkeep                    # run logs land here (gitignored)
│
└── multi-agent-trip-planner/          # Case 2: dynamic discovery, 3 remote agents
    ├── README.md
    ├── requirements.txt
    ├── .env.example
    ├── common/                        # shared: prompts, config, logging
    ├── remote_agents/
    │   ├── travel_tools.py            # mock travel tools with fixed data
    │   ├── flights_server.py          # A2AServer — Gemma via Ollama (port 9001)
    │   ├── hotels_server.py           # A2AServer — Gemma via Ollama (port 9002)
    │   └── itinerary_server.py        # A2AServer — Gemma via Ollama (port 9003)
    ├── orchestrator/
    │   └── main.py                    # Gemini — A2AClientToolProvider
    ├── tests/
    │   ├── test_discovery.py          # does the orchestrator discover before delegating?
    │   └── test_a2a_integration.py    # real integration test against the 3 running servers
    ├── evals/
    │   ├── eval_discovery.py          # deterministic: discovery + delegation decision
    │   └── eval_response_quality.py   # LLM-judge: correctness + faithfulness
    └── logs/
        └── .gitkeep                    # run logs land here (gitignored)

一旦仓库克隆完成且带有 gemma 4 模型的 Ollama Docker 启动（或生成了 Gemini API key），我们就着手构建环境。两个项目的步骤相同；每个项目有自己的 requirements.txt 和 .env，所以需要进入你想运行的项目文件夹。

requirements.txt 中包含（带有注释解释每个依赖的用途）：

strands-agents[a2a,ollama,litellm]>=1.0.0
strands-agents-tools>=0.1.0
python-dotenv>=1.0.0
pytest>=8.0.0
pytest-asyncio>=0.23.0
strands-agents-evals>=1.2.0

建议在虚拟环境中安装依赖，以与系统其他部分隔离：

python -m venv .venv
# Activar: source .venv/bin/activate (Linux/macOS)
pip install -r requirements.txt

在 .env 中配置编排器的 API key 和模型（Gemini）、本地模型（Gemma via Ollama）以及日志级别。只需复制示例并填写 key：

cp .env.example .env

在 common/config.py 中加载和验证这些变量（如果缺少 API key 会提前失败并给出提示），在 common/prompts.py 中是编排器和每个远程 AI 智能体的 system prompt。

📝 关于切换模型：编排器使用 LiteLLMModel，所以不依赖 Gemini：LiteLLM 支持许多 provider（Anthropic、OpenAI、Bedrock 等）。只需在 .env 中将 ORCHESTRATOR_MODEL_ID 改为你想要的 provider 标识符（带前缀，例如 anthropic/…、openai/…）并设置其 API key。完整列表在 Strands 的 model providers 文档中。同样，远程 AI 智能体也不必在 Ollama 上运行：你可以在服务器代码（server.py 或 *_server.py）中用另一个 provider 替换 OllamaModel。

测试位于 tests/，可以逐个运行（推荐！由于本地可能较慢）或一次性全部运行：快速单元测试（判断编排器是否在应该时委托/发现）以及针对 A2A 服务器的真实集成测试（自行作为子进程启动；需要 Ollama 正在运行）。

python -m pytest

接下来执行实际用例。以下是两个项目之间的实际区别：每个项目作为多个独立进程运行，每个终端一个。首先启动远程 AI 智能体（每个 A2AServer 在各自端口上监听），所有 AI 智能体启动后，再启动编排器，它会发现它们并通过网络与它们通信。服务器的启动顺序无关紧要，但必须在运行编排器之前全部启动。

privacy-aware-routing —— 2 个进程（2 个终端）：

multi-agent-trip-planner —— 4 个进程（4 个终端）：

编排器将查询作为参数接受，或者如果不带参数调用，则打开一个交互式聊天。退出远程服务器：Ctrl+C；退出交互式聊天：输入 exit。

调用日志存放在 logs/ 目录下，按进程分隔。里面包含原始数据，可用于追溯 A2A 通信的全过程：编排器发现了什么并委托了什么，每个远程 Agent 执行了什么工具（TOOL CALL: ... 行）。可以此验证最终答案来源于真实数据，而非模型的幻觉。

最终结果以流式方式（逐 token）在屏幕上展示。

最后，我们使用"Evals"概念对真实的多智能体系统进行评估。每个项目自带两个脚本：一个确定性脚本（验证编排器的决策），以及一个 LLM-as-judge 脚本（对最终回复质量打分）。它们对配置好的模型运行真实流程，在控制台输出结果，并将完整报告保存到 evals/outputs/<name>_<timestamp>.json。（更多细节见上文关键概念部分）

# privacy-aware-routing
python -m evals.eval_routing
python -m evals.eval_response_quality

# multi-agent-trip-planner
python -m evals.eval_discovery
python -m evals.eval_response_quality

两个项目实现的是不同的用例，均基于 A2AServer 暴露智能体的同一套基础设施，但在编排器如何消费这些远程智能体上有所不同：一个是确定性路由，指向已知的单个智能体（A2AAgent，固定 endpoint）；另一个是通过动态发现多个专业智能体（A2AClientToolProvider，在运行时读取 agent cards）。

⭐ 在两个用例中，Strands 的抽象都隐藏了协议层面的 mechanics（agent card 的解析、HTTP、JSON-RPC）。在编排器一侧，向远程进程委托任务的过程几乎和调用本地函数一模一样。

几次运行下来，总结出以下几点经验：

Agent card 就是契约。 在发现场景中，发布的 skills 质量（名称和描述）决定了编排器能否在不借助硬编码映射的情况下做出正确路由。Agent card 不完整会导致决策质量下降。

小参数本地模型使用工具的行为不稳定。 使用 gemma4:e2b-it-qat 时，智能体有时会直接回复而不调用工具，此时可能凭空编造数据。因此（a）在 prompt 中指示其始终调用工具，以及（b）验证执行结果都很重要：TOOL CALL: ... 日志能区分真实数据和幻觉。

评估智能体需要阅读 reasons 字段，不能只看分数。 得分为 0.0 可能源于基础设施问题（judge 返回 503、模型已下线），而非回答本身质量差。

延迟和并发是架构层面的决策。 本地推理速度慢；当三个专业智能体共享一个顺序执行的 Ollama 时，任务会排队等待。启用并行（OLLAMA_NUM_PARALLEL）并调整 A2A 客户端的超时时间，能显著改善体验。

这里举例的两个用例只是 A2A 协议的几种变体。Strands 还支持以下能力，记录下来留待后续测试：

在多智能体模式中使用远程智能体： A2AAgent 可以作为 Graph workflow 中的一个节点，在同一 pipeline 中混合本地和远程智能体。

中断机制（input_required）： 类似于"human-in-the-loop"，远程智能体可以暂停任务并等待批准后再继续（例如确认预订），然后从中断处精确恢复。

对话持久化： 通过为每个 context_id 配置 SessionManager，每段对话可以在进程重启后继续存活，而非仅存在于内存中。

推送通知与分布式部署： 适用于长耗时任务，也适用于将每个智能体真正暴露到生产环境（例如位于负载均衡器后、通过 AWS Fargate 或经由 AgentCore 暴露），使系统可扩展到笔记本之外。

互补使用 MCP： 每个专业智能体可以用 MCP 调用自己的工具，用 A2A 与其他智能体协作。

⭐ 最后，这里用 Strands 以 Python 构建智能体，但 A2A 是一个开放标准（目前隶属 Linux Foundation），运行在 HTTP/JSON-RPC 之上，因此互操作性发生在协议层面，而非框架层面。一个 Strands 智能体可以与用其他框架和语言构建、同样实现了 A2A 的智能体通信。该协议拥有官方 SDK，支持 Python、JavaScript、Java、C#/.NET、Go 和 Rust，各 SDK 之间互不了解彼此的内部逻辑。

Curso Fundamentos Strands Building AI Agent Harnesses with Strands Agents Building AI Agent Harnesses – Video Course

Curso Fundamentos Strands Building AI Agent Harnesses with Strands Agents Building AI Agent Harnesses – Video Course

Documentación Strands Agents Evals Agent-to-Agent (A2A) Protocol

Documentación Strands Agents Evals Agent-to-Agent (A2A) Protocol

Protocolo A2A (Agent2Agent) Sitio oficial del protocolo A2A

Protocolo A2A (Agent2Agent) Sitio oficial del protocolo A2A