尧图网页设计 
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 中的定位A2AAgent-to-Agent是面向 Agent 间直接通信的协议。Agent Zero 既可作为 A2A服务器暴露自己的推理与对话能力配置方法见 docs/guides/a2a-setup.md也可通过a2a_chat工具以客户端身份主动联系外部 A2A Agent。a2a_chat的职责非常聚焦让 Agent 给外部 A2A Agent 发一条消息并取回对方的 assistant 回复。从源码结构看它的运行依赖三个模块依赖模块作用源码路径helpers.fasta2a_client连接建立、消息发送、任务轮询helpers/fasta2a_client.pyhelpers.toolTool基类与Response返回值契约helpers/tool.pyhelpers.print_style错误信息的美化输出helpers/print_style.py工具与 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_loopFalse)。break_loop恒为False意味着一次 a2a 对话结束后 Agent 循环继续运行模型可以基于返回内容决定下一步行动。工具输出要求简洁、可被模型直接理解并且适合持久化进历史记录——Tool.after_execution会把Response.message清洗后写入 agent 历史见 helpers/tool.py因此返回给模型的就是一段纯文本 assistant 回复不夹带冗余协议字段。参数契约agent_url、message、attachments 与 resetexecute()接收四个参数Prompt 指令文件对其做了精确定义参数类型必填说明agent_urlstring是远端 Agent 地址支持host:port、http://host:port或完整的/a2aURLmessagestring是要发送给远端 Agent 的文本attachmentslist[str]否随消息发送的绝对 URI 或本地路径默认为空resetjson boolean否传true时对同一agent_url开启全新会话默认false源码中的校验逻辑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: Whats 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分两步去掉 URL 末尾的/若以/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_idNone开启新对话若远端在任务结果中返回了新的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 消息roleuserparts含一个kindtext部分若有attachments则逐个追加kindfile、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实现了一套多级回退的文本提取逻辑_text_from_part(part)对单个 part 提取文本依次检查text与content键要求是非空字符串并去空白_text_from_message(message)对一条消息提取文本。优先遍历parts列表并拼接各 part 文本若没有 parts则依次尝试text、content、message、output键字符串输入直接strip()_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_loopFalse让模型可以读取错误信息后自行纠正或换路场景返回消息FastA2A 客户端未安装FastA2A client not available on this instance.agent_url缺失/非字符串agent_url argument missingmessage缺失/非字符串message argument missing远端未创建任务无result.idRemote agent failed to create task.任务完成但提取不到任何 assistant 文本A2A_EMPTY_RESPONSE_ERROR常量内容任意异常网络、超时等A2A chat error: {e}同时以PrintStyle.error输出其中空响应被专门设计为显式失败而非成功常量A2A_EMPTY_RESPONSE_ERRORtools/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_argsdocs/guides/a2a-setup.md如何将 Agent Zero 自身配置为 A2A 服务器含连接 URL 格式http://HOST:PORT/a2a/t-TOKEN与项目级 URL.../p-PROJECT_NAMEdocs/developer/connectivity.mdA2A 协议规范与 API 级集成细节helpers/fasta2a_client.py.dox.md 与 tests/test_fasta2a_client.py客户端底层实现及其测试。使用建议综合源码与文档实际使用a2a_chat时有几点值得注意依赖前提当前实例必须安装 FastA2A 客户端fasta2a.client与httpx否则工具会直接返回不可用提示跨实例协作前可先确认目标实例已按 docs/guides/a2a-setup.md 开启 A2A 服务器。URL 写法自由agent_url三种写法均可会话缓存会自动归一无需担心/a2a后缀差异保持会话多轮协作时不要传reset或传false让context_id自动延续需要全新对话时显式传reset: true空响应即失败远端“完成但无文本”不会被视为成功模型应据此重新尝试或切换策略附件传递attachments仅接受绝对 URI 或路径会以kindfile的 A2A part 随消息发送。【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zero创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Agent Zero a2a_chat 工具解析：基于 FastA2A 的跨实例 Agent 对话与上下文复用
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 中的定位A2AAgent-to-Agent是面向 Agent 间直接通信的协议。Agent Zero 既可作为 A2A服务器暴露自己的推理与对话能力配置方法见 docs/guides/a2a-setup.md也可通过a2a_chat工具以客户端身份主动联系外部 A2A Agent。a2a_chat的职责非常聚焦让 Agent 给外部 A2A Agent 发一条消息并取回对方的 assistant 回复。从源码结构看它的运行依赖三个模块依赖模块作用源码路径helpers.fasta2a_client连接建立、消息发送、任务轮询helpers/fasta2a_client.pyhelpers.toolTool基类与Response返回值契约helpers/tool.pyhelpers.print_style错误信息的美化输出helpers/print_style.py工具与 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_loopFalse)。break_loop恒为False意味着一次 a2a 对话结束后 Agent 循环继续运行模型可以基于返回内容决定下一步行动。工具输出要求简洁、可被模型直接理解并且适合持久化进历史记录——Tool.after_execution会把Response.message清洗后写入 agent 历史见 helpers/tool.py因此返回给模型的就是一段纯文本 assistant 回复不夹带冗余协议字段。参数契约agent_url、message、attachments 与 resetexecute()接收四个参数Prompt 指令文件对其做了精确定义参数类型必填说明agent_urlstring是远端 Agent 地址支持host:port、http://host:port或完整的/a2aURLmessagestring是要发送给远端 Agent 的文本attachmentslist[str]否随消息发送的绝对 URI 或本地路径默认为空resetjson boolean否传true时对同一agent_url开启全新会话默认false源码中的校验逻辑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: Whats 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分两步去掉 URL 末尾的/若以/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_idNone开启新对话若远端在任务结果中返回了新的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 消息roleuserparts含一个kindtext部分若有attachments则逐个追加kindfile、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实现了一套多级回退的文本提取逻辑_text_from_part(part)对单个 part 提取文本依次检查text与content键要求是非空字符串并去空白_text_from_message(message)对一条消息提取文本。优先遍历parts列表并拼接各 part 文本若没有 parts则依次尝试text、content、message、output键字符串输入直接strip()_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_loopFalse让模型可以读取错误信息后自行纠正或换路场景返回消息FastA2A 客户端未安装FastA2A client not available on this instance.agent_url缺失/非字符串agent_url argument missingmessage缺失/非字符串message argument missing远端未创建任务无result.idRemote agent failed to create task.任务完成但提取不到任何 assistant 文本A2A_EMPTY_RESPONSE_ERROR常量内容任意异常网络、超时等A2A chat error: {e}同时以PrintStyle.error输出其中空响应被专门设计为显式失败而非成功常量A2A_EMPTY_RESPONSE_ERRORtools/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_argsdocs/guides/a2a-setup.md如何将 Agent Zero 自身配置为 A2A 服务器含连接 URL 格式http://HOST:PORT/a2a/t-TOKEN与项目级 URL.../p-PROJECT_NAMEdocs/developer/connectivity.mdA2A 协议规范与 API 级集成细节helpers/fasta2a_client.py.dox.md 与 tests/test_fasta2a_client.py客户端底层实现及其测试。使用建议综合源码与文档实际使用a2a_chat时有几点值得注意依赖前提当前实例必须安装 FastA2A 客户端fasta2a.client与httpx否则工具会直接返回不可用提示跨实例协作前可先确认目标实例已按 docs/guides/a2a-setup.md 开启 A2A 服务器。URL 写法自由agent_url三种写法均可会话缓存会自动归一无需担心/a2a后缀差异保持会话多轮协作时不要传reset或传false让context_id自动延续需要全新对话时显式传reset: true空响应即失败远端“完成但无文本”不会被视为成功模型应据此重新尝试或切换策略附件传递attachments仅接受绝对 URI 或路径会以kindfile的 A2A part 随消息发送。【免费下载链接】agent-zeroAgent Zero AI framework项目地址: https://gitcode.com/GitHub_Trending/ag/agent-zero创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考