企业官网建设流程全解析

Agent Zero a2a_chat 工具解析：基于 FastA2A 的跨实例 Agent 对话与上下文复用

【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zero

导读

a2a_chat是 Agent Zero 内置的 Agent 间通信工具（位于 tools/a2a_chat.py），它让当前 Agent 实例通过 FastA2A 协议向任意外部 A2A Agent 发送消息，并从远端任务结果中提取最终的 assistant 回复。本文以其维护文档 tools/a2a_chat.py.dox.md 为核心骨架，结合源码、Prompt 指令与测试用例，完整讲解该工具的参数契约、会话缓存机制、底层调用链、响应提取逻辑与错误处理，帮助读者理解并正确使用 Agent Zero 的多 Agent 协作能力。

A2A 与 a2a_chat 在 Agent Zero 中的定位

A2A（Agent-to-Agent）是面向 Agent 间直接通信的协议。Agent Zero 既可作为 A2A服务器暴露自己的推理与对话能力（配置方法见 docs/guides/a2a-setup.md），也可通过a2a_chat工具以客户端身份主动联系外部 A2A Agent。

a2a_chat的职责非常聚焦：让 Agent 给外部 A2A Agent 发一条消息，并取回对方的 assistant 回复。从源码结构看，它的运行依赖三个模块：

工具与 Prompt 指令（prompts/agent.system.tool.a2a_chat.md）、测试用例（tests/test_tool_action_contracts.py）三者共同构成该功能的完整闭环，任何参数或行为变更都需要同步维护这三处。

工具运行时契约：Tool 基类与 Response 返回

Agent Zero 的所有工具模块都必须遵循统一的运行时契约（定义于 helpers/tool.py）：

• 定义helpers.tool.Tool的子类；
• 在execute(...)方法中返回helpers.tool.Response对象。

Response是一个数据类，包含三个字段：

@dataclass class Response: message: str # 返回给 Agent 的文本内容 break_loop: bool # 是否终止 agent 循环 additional: dict | None = None # 附加结构化数据

A2AChatTool严格遵循该契约：它继承Tool，实现async def execute(self, **kwargs) -> Response，并且所有路径（成功、参数缺失、远端失败、异常）都返回Response(message=..., break_loop=False)。break_loop恒为False，意味着一次 a2a 对话结束后 Agent 循环继续运行，模型可以基于返回内容决定下一步行动。

工具输出要求简洁、可被模型直接理解，并且适合持久化进历史记录——Tool.after_execution会把Response.message清洗后写入 agent 历史（见 helpers/tool.py），因此返回给模型的就是一段纯文本 assistant 回复，不夹带冗余协议字段。

参数契约：agent_url、message、attachments 与 reset

execute()接收四个参数，Prompt 指令文件对其做了精确定义：

源码中的校验逻辑（tools/a2a_chat.py）会拒绝空值或非字符串类型的agent_url与message，并返回agent_url argument missing/message argument missing的错误提示。另外有一个明确的使用禁忌：不要自行传入context_id——会话上下文由工具内部自动管理与传递，模型侧无需关心。

agent_url的三种写法最终都会在底层连接层被规范化（见下文“底层调用链”），例如weather.example.com:8000会被自动补上http://前缀。

一次典型的工具调用示例

以下 JSON 来自 prompts/agent.system.tool.a2a_chat.md，展示了模型侧应如何组织调用：

{ "thoughts": ["I need to ask a remote agent and keep the session for follow-up."], "headline": "Contacting remote FastA2A agent", "tool_name": "a2a_chat", "tool_args": { "agent_url": "http://weather.example.com:8000/a2a", "message": "What's the forecast for Berlin today?", "attachments": [], "reset": false } }

会话缓存与上下文复用：_session_key 与 agent 数据持久化

A2A 对话的上下文复用是a2a_chat的核心能力之一，其设计目标正如 DOX 文档所写：让根路径 URL 与显式/a2a路径的 URL 共享同一个会话缓存。

会话键规范化

_session_key(agent_url)的实现（tools/a2a_chat.py）分两步：

1. 去掉 URL 末尾的/；
2. 若以/a2a结尾，则剥掉这 4 个字符并再次去尾斜杠。

因此http://localhost:32080/a2a与http://localhost:32080会映射到同一个会话键http://localhost:32080。这一点由测试 tests/test_tool_action_contracts.py 明确验证：

def test_a2a_session_key_normalizes_explicit_a2a_path(monkeypatch): module = _load_a2a_chat_tool(monkeypatch) assert module._session_key("http://localhost:32080/a2a") == "http://localhost:32080" assert module._session_key("http://localhost:32080") == "http://localhost:32080"

该设计的实际价值是：即使模型这次传了/a2a后缀、下次没传，远端 Agent 仍然能延续同一段对话上下文，而不会因 URL 写法差异导致会话断裂。

会话的存取与 reset 语义

会话映射sessions: dict[str, str]以键_a2a_sessions存放在 Agent 实例的数据区中（self.agent.get_data/self.agent.set_data）：

• 每次调用先读取self.agent.get_data("_a2a_sessions") or {}；
• 以规范化后的cache_key查找已保存的远端context_id；
• 若reset为true且键存在，先sessions.pop(cache_key)清除旧会话，本次以context_id=None开启新对话；
• 若远端在任务结果中返回了新的context_id（字符串类型），则写入缓存并持久化回 agent 数据区。

由此，reset: false时同一agent_url的多次调用自然构成连续的多轮对话；reset: true则可强制开启一段全新会话——这是 Prompt 中“remote context is preserved automatically per agent_url”的底层实现。

底层调用链：从连接握手到任务完成

A2AChatTool.execute的核心流程（tools/a2a_chat.py）是一条清晰的分步链路，全部由 helpers/fasta2a_client.py 支撑：

is_client_available() → connect_to_agent(agent_url) # 建立连接并握手 → AgentConnection.__init__ # 规范化 URL、注入认证头 → get_agent_card() # 拉取 .well-known/agent.json → conn.send_message(...) # 发送用户消息（阻塞模式） → conn.wait_for_completion(task_id) # 轮询任务直到终态 → _extract_latest_assistant_text(final) # 提取 assistant 文本

第一步：客户端可用性检查

is_client_available()返回模块导入时fasta2a.client与httpx是否成功加载（helpers/fasta2a_client.py）。若依赖缺失，工具直接返回"FastA2A client not available on this instance."而不会尝试网络操作。

第二步：连接与握手

AgentConnection在构造时（helpers/fasta2a_client.py）：

• 若 URL 不以http://或https://开头，自动补http://；
• 从环境变量A2A_TOKEN读取令牌（也可显式传入token），存在时同时注入Authorization: Bearer <token>与X-API-KEY: <token>两个请求头；
• 用httpx.AsyncClient承载 A2A 客户端，默认 30 秒超时。

connect_to_agent()会先调用get_agent_card()拉取远端/.well-known/agent.json并打印 Agent 名称与描述，以此验证连通性；若 URL 含/a2a而拉取失败，还会尝试回退到根路径重试一次（helpers/fasta2a_client.py），随后抛出RuntimeError。

第三步：发送消息

send_message()将用户文本封装为 A2A 消息（role=user，parts含一个kind=text部分），若有attachments则逐个追加kind=file、file.uri部分；若调用方未提供context_id，会自动复用连接内已保存的self._context_id。发送使用message/send方法（而非send_task），并携带accepted_output_modes: ['application/json', 'text/plain']与blocking: true配置（helpers/fasta2a_client.py）。响应中的context_id会被捕获并保存在连接实例上，供后续调用延续上下文。

第四步：等待任务完成

wait_for_completion()以 2 秒为间隔轮询任务状态（helpers/fasta2a_client.py）：

• 任务状态为completed、failed或canceled时立即返回最终结果；
• 其他状态继续轮询，最长等待 300 秒，超时抛出TimeoutError。

a2a_chat拿到final后，从final["result"]["context_id"]中取新会话 ID 写回缓存。

响应提取：_text_from_part / _text_from_message / _extract_latest_assistant_text

远端任务的最终结果结构并不固定（可能包含 history、status、artifacts 等多种载体），因此a2a_chat实现了一套多级回退的文本提取逻辑：

1. _text_from_part(part)：对单个 part 提取文本，依次检查text与content键，要求是非空字符串并去空白；
2. _text_from_message(message)：对一条消息提取文本。优先遍历parts列表并拼接各 part 文本；若没有 parts，则依次尝试text、content、message、output键；字符串输入直接strip()；
3. _extract_latest_assistant_text(task_response)：按以下优先级提取最终 assistant 文本：
   • 遍历result.history（倒序），跳过role == "user"的消息，取第一条能提取出文本的消息；
   • 若 history 缺失或为空，尝试result.status.message；
   • 再尝试result.artifacts（倒序遍历）；
   • 最后回退到对整个result做_text_from_message。

这套策略被 tests/test_tool_action_contracts.py 的两组测试覆盖：一组验证从 history 中提取“4”（跳过 user 消息），另一组验证在 history 为空时能分别从status.message与artifacts中提取文本。

错误处理与边界情况

a2a_chat对失败路径的界定非常明确，全部返回break_loop=False，让模型可以读取错误信息后自行纠正或换路：

其中空响应被专门设计为显式失败而非成功：常量A2A_EMPTY_RESPONSE_ERROR（tools/a2a_chat.py）明确指出“远端任务虽已完成但未找到 assistant 文本，应视为失败的远端响应而非成功”。测试 tests/test_tool_action_contracts.py 验证了空 history 时提取结果为空字符串，且该错误信息同时包含"failed"与"not success"两个关键词，确保模型不会误解成功语义。

从代码结构看，execute整体被try/except Exception包裹，异常只记录并返回消息，不会中断 Agent 主循环，体现了“工具失败不阻断整体运行”的设计取向。

验证方式与配套文档

测试

与a2a_chat直接相关的测试集中在 tests/test_tool_action_contracts.py，覆盖四类行为：

• 会话键规范化（根路径与/a2a路径等价）；
• history 中最新的 assistant 文本提取（跳过 user 消息）；
• history 为空时从 status / artifacts 提取文本；
• 空响应被标记为显式失败。

这些测试通过_load_a2a_chat_tool将helpers.tool替换为桩实现后直接导入tools.a2a_chat，验证的是纯函数逻辑，不依赖真实网络。DOX 文档的 Verification 部分也建议：变更行为后运行针对性的工具与 prompt 契约测试，若无针对性测试则对 Agent 执行做冒烟测试。

配套文档

• prompts/agent.system.tool.a2a_chat.md：工具 Prompt 指令，模型如何组织tool_args；
• docs/guides/a2a-setup.md：如何将 Agent Zero 自身配置为 A2A 服务器（含连接 URL 格式http://HOST:PORT/a2a/t-TOKEN与项目级 URL.../p-PROJECT_NAME）；
• docs/developer/connectivity.md：A2A 协议规范与 API 级集成细节；
• helpers/fasta2a_client.py.dox.md 与 tests/test_fasta2a_client.py：客户端底层实现及其测试。

使用建议

综合源码与文档，实际使用a2a_chat时有几点值得注意：

1. 依赖前提：当前实例必须安装 FastA2A 客户端（fasta2a.client）与httpx，否则工具会直接返回不可用提示；跨实例协作前可先确认目标实例已按 docs/guides/a2a-setup.md 开启 A2A 服务器。
2. URL 写法自由：agent_url三种写法均可，会话缓存会自动归一，无需担心/a2a后缀差异；
3. 保持会话：多轮协作时不要传reset或传false，让context_id自动延续；需要全新对话时显式传reset: true；
4. 空响应即失败：远端“完成但无文本”不会被视为成功，模型应据此重新尝试或切换策略；
5. 附件传递：attachments仅接受绝对 URI 或路径，会以kind=file的 A2A part 随消息发送。

【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zero