本周，A2A协议的官方Python SDK发布，下面这篇文章，通过实战案例讲解如何使用A2A Python SDK构建多智能体系统： 一、事件检测智能体与A2A客户端之间的通信 系统中有三个角色：用户、事件智能体（负责检测事件）和警报智能体（负责向用户发出事件警报）。它们都使用消息（Messages）进行通信。每条消息代表A2A协议中的一次通信轮次。 智能体被封装于暴露HTTP端点的A2A服务器中。每个A2A服务器包含事件队列（Event queues），用于缓冲智能体的异步执行和服务器响应处理。 A2A客户端（A2A Client）负责发起通信，A2A服务器亦可扮演客户端角色。【图1】展示了客户端和服务器在协议内如何进行通信。 EventQueue存储了消息（Messages）、任务（Tasks）、任务状态更新事件（TaskStatusUpdateEvent）、任务工件更新事件（TaskArtifactUpdateEvent）、A2AError 和JSONRPCError对象。 任务（Task）是构建多智能体系统的核心概念： 客户端发送消息时，智能体可能创建有状态任务（如“生成报告”）。 每个任务有唯一ID和生命周期（submitted, working, input-required, completed, failed） 任务是有状态的，可能涉及客户端与服务器间的多次消息交换。 本系统中的任务包括：检测事件和提醒用户。 二、为事件智能体构建A2A服务器 首先，需为事件检测智能体定义智能体卡片（Agent Card）。【图2】 这是一个JSON文档，描述了服务器身份、能力（Capabilities）、技能（Skills）、服务端点（Service endpoint）URL，以及客户端如何进行身份验证和交互。 智能体本身是一个Uvicorn服务器，main() 方法负责服务器启动。 所有请求由a2a-python SDK的DefaultRequestHandler处理。该处理程序需配置TaskStore用于任务存储，以及AgentExecutor实例来执行代理的核心逻辑。 main() 方法的最后组件是A2AStarletteApplication，一个实现A2A协议服务器端点的Starlette应用。它需通过智能体卡片和 DefaultRequestHandler进行初始化。最终，使用uvicorn运行应用程序。【图4】 三、创建EventAgentExecutor 现在是时候构建智能体的核心，并最终了解如何使用任务让智能体彼此交互了。 EventAgentExecutor类继承自AgentExecutor接口，要求实现 execute() 和 cancel() 方法。 两者均接收RequestContext和EventQueue对象作为参数。RequestContext封装当前请求信息，而EventQueue充当智能体异步执行与服务器响应处理间的缓冲区。 智能体逻辑简化为检测用户消息中是否包含字符串 "event"。若存在，则通过向警报智能体发送消息来调用它。 这采用静态配置策略，即为事件智能体配置警报智能体的智能体卡片URL。在此场景下，事件智能体扮演A2A客户端角色。 执行器构建步骤如下：首先创建主任务（事件检测任务），实例化 TaskUpdater对象（辅助智能体向任务事件队列发布更新的类），提交任务并通过 start_work() 方法宣布开始处理。【图5】 用户发送给智能体的消息包含一个或多个Part，代表独立内容片段，可以是TextPart、FilePart或DataPart。本例使用TextPart，需在执行器中对其进行解包。【图6】 若消息中不包含字符串 "event"，则无需调用警报智能体，任务即告完成。【图7】 四、为用户创建A2A客户端 构建A2A客户端以测试智能体。客户端通过 A2AClient 类中的 get_client_from_agent_card_url() 方法获取智能体卡片。 随后，将消息封装于SendMessageRequest对象中，并利用客户端的 send_message() 方法将其发送给智能体。【图8】 五、让事件智能体与警报智能体对话 为了让事件智能体与警报智能体对话，事件智能体也将扮演客户端的角色。【图9】 调用警报智能体，就像作为用户调用事件智能体一样，当警报智能体的任务完成时，就完成原始的事件智能体任务。现在再次调用事件智能体，但这次带上一个事件。【图10】 更多内容，欢迎查看原文：-agent-communication-with-the-a2a-python-sdk/
