A2A协议原理

1 Agent2Agent Protocol

2025年4月9日，Google 正式发布了 Agent2Agent Protocol （以下简称 “A2A”）。该协议为不同类型的智能体之间搭建了一座高效沟通与协作的桥梁，无论是独立Agent与独立Agent、独立Agent与企业Agent，亦或是企业Agent与企业Agent，都能借助该协议实现通信交互和协作。

（1）Secure collaboration (安全协作): 在整个任务处理过程中，A2A 协议确保所有智能体间的通信和数据交换都是安全的。这包括身份验证、数据加密和访问控制，以防止未经授权的访问或数据泄露。

场景示例:

• “旅行规划师”与“TicketAgentServer”之间的所有通信都通过安全协议进行。
• 即使在处理涉及个人信息（如你的姓名、联系方式）的火车票预订任务时，这些数据也得到了协议层面的保护，确保只有授权的智能体才能访问和处理。

（2）Task and state mgmt (任务与状态管理): 这表示协议能够有效地管理任务的整个生命周期及其状态变化。例如，一个任务从提交到完成、失败或被取消，协议都能进行跟踪和管理 。

场景示例:

• “旅行规划师”发送一个 Task 给 “TicketAgentServer”，请求预订从北京到上海的火车票。
• “TicketAgentServer”接收到 Task，并将其状态标记SUBMITTED。
• 当它成功模拟预订后，会将 Task 状态更新为 COMPLETED，并返回包含预订结果的 artifacts。

（3）UX negotiation (用户体验协商): 这项功能允许智能体之间就如何更好地为用户提供服务进行协商和调整，以优化最终的用户体验 。

工作方式: 这项能力允许智能体在协作过程中调整和优化用户体验。例如，一个智能体可能会在发现无法完成任务时，主动请求用户提供更多信息，而不是直接失败。

场景示例:

• 当你发送“帮我预订去上海的火车票”的请求时，“TicketAgentServer”发现缺少出发地信息。
• 它不会直接返回错误，而是通过 A2A 协议向“旅行规划师”发送一个包含 INPUT_REQUIRED 状态的任务更新 ，并在消息中说明“需要您提供出发地城市”，从而让“旅行规划师”可以反过来向你提问，形成一个流畅的交互，而不是僵硬的错误提示。

（4）Capability discovery (能力发现): 这是指 A2A 协议使得智能体能够自动发现其他智能体的能力或功能。智能体通过类似 AgentCard 的“名片”来广播其服务，其他智能体则可以根据这些信息来选择最合适的合作伙伴 。

场景示例:

• “旅行规划师”找到了两个智能体：一个叫 “TicketAgentServer” 的智能体，其 AgentCard 描述它拥有“预订火车票”的能力。
• 另一个叫 “WeatherAgentServer” 的智能体，其 AgentCard 描述它拥有“查询天气”的能力。

Agent2Agent 架构剖析

A2A核心角色：

• User：用户是协议中的关键主体，主要负责进行认证和授权操作，确保交互的安全性和合法性。
• Client Agent：客户端 Agent 是任务的发起者，它代表用户提出需求或请求。
• Server Agent：服务端 Agent 是任务的执行者，它接收来自客户端 Agent 的请求，并执行相应的操作。

客户端 Client 与服务端 Client 之间的通信，本质上是基于任务的请求与响应机制。每个请求都对应一个具体任务，服务端 Agent 处理任务后返回结果。值得注意的是，一个 Agent 既可以作为客户端 Agent 发起任务，也可以作为服务端 Agent 执行任务，具有双重角色的灵活性。

Agent2Agent 核心概念

Client Agent 和 Server Agent 交互的过程中，会涉及到一些 Entity：AgentCard、Task 、Artifact等。

AgentSkill

AgentSkill描述代理的具体能力或功能模块，例如处理特定任务的技能。它包括技能名称、描述、示例、输入/输出模式等。在A2A协议中，技能是代理卡片（AgentCard）的组成部分，用于细粒度服务发现。支持扩展标签和示例，便于代理间匹配调用。

代码示例：

from python_a2a import  AgentSkill
# 定义一个代理技能
ticket_skill = AgentSkill(
    name="book_ticket",
    description="预订火车票的技能",
    examples=["预订从上海到北京的火车票"],
    input_modes=["text/plain"],
    output_modes=["text/plain"]
)

print(ticket_skill)
print(ticket_skill.to_dict())

运行结果：

AgentSkill(name='book_ticket', description='预订火车票的技能', id='321ec43d-043f-4934-9046-56f730b33356', tags=[], examples=['预订从上海到北京的火车票'], input_modes=['text/plain'], output_modes=['text/plain'])

{'id': '321ec43d-043f-4934-9046-56f730b33356', 'name': 'book_ticket', 'description': '预订火车票的技能', 'tags': [], 'examples': ['预订从上海到北京的火车票'], 'inputModes': ['text/plain'], 'outputModes': ['text/plain']}

AgentCard

AgentCard是A2A协议中Agent代理的元数据描述卡片，用于代理发现和服务注册。它包含代理的名称、描述、URL、版本、技能列表、能力（如流式传输支持）和输入/输出模式等信息。AgentCard允许其他代理或系统查询和调用该代理的服务，是A2A生态系统的入口点。在源码中，它支持序列化为JSON格式，便于网络传输。

AgentCard 是 Server Agent 的名片，它主要描述了 Server Agent 的能力、认证机制等信息。Client Agent通过获取不同 Server Agent 的 AgentCard，了解不同 Server Agent 的能力，来决断具体的任务执行应该调用哪个 Server Agent 。

代码示例:

from python_a2a import AgentCard, AgentSkill
# 创建一个代理技能
ticket_skill = AgentSkill(
    name="book_ticket",
    description="预订火车票的技能",
    examples=["预订从上海到北京的火车票"],
    input_modes=["text/plain"],
    output_modes=["text/plain"]
)
# 创建代理卡片
agent_card = AgentCard(
    name="TicketAgent",
    description="一个可以预订票务的代理",
    url="http://127.0.0.1:5009",
    version="1.0.0",
    skills=[ticket_skill],
    capabilities={"streaming": True}
)
# 打印代理卡片的字典表示（用于序列化）
print(agent_card)
print(agent_card.to_dict())

输出示例：

AgentCard(name='TicketAgent', description='一个可以预订票务的代理',url='http://127.0.0.1:5009', version='1.0.0', authentication=None, capabilities={'streaming': True}, default_input_modes=['text/plain'], default_output_modes=['text/plain'], skills=[AgentSkill(name='book_ticket', description='预订火车票的技能', id='487e3ee8-1cbd-4567-b083-a9627a486d1c', tags=[], examples=['预订从上海到北京的火车票'], input_modes=['text/plain'], output_modes=['text/plain'])], provider=None, documentation_url=None)

{'name': 'TicketAgent', 'description': '一个可以预订票务的代理', 'url': 'http://127.0.0.1:5009', 'version': '1.0.0', 'capabilities': {'streaming': True}, 'defaultInputModes': ['text/plain'], 'defaultOutputModes': ['text/plain'], 'skills': [{'id': '487e3ee8-1cbd-4567-b083-a9627a486d1c', 'name': 'book_ticket', 'description': '预订火车票的技能', 'tags': [], 'examples': ['预订从上海到北京的火车票'], 'inputModes': ['text/plain'], 'outputModes': ['text/plain']}]}

Task

Task 是一个具有明确状态的实体，由 Client Agent 创建并发起，其状态由 Server Agent 负责维护和更新。每个 Task 都旨在实现一个特定的目标或结果。在 Task 的执行过程中，Client Agent 和 Server Agent 通过交换 Message 进行通信，而 Server Agent 执行任务后生成的输出结果被称为 Artifact（人工制品）。

此外，每个 Task 都拥有一个唯一的 session Id。多个 Task 可以共享同一个 session Id，这表明这些 Task 属于同一个会话（Session）的一部分，便于管理和跟踪相关任务的执行流程。

代码示例:

from python_a2a import Task, Message, MessageRole, TextContent

# 创建任务
message = Message(content=TextContent(text="查询天气"), role=MessageRole.USER)
task = Task(message=message.to_dict())
print(task)

运行结果：

Task(id='5e63778f-a912-4f07-a757-4b47614d0d1b', session_id='7211f559-d616-4f56-82d2-c74976f86390', status=TaskStatus(state=<TaskState.SUBMITTED: 'submitted'>, message=None, timestamp='2025-10-24T23:20:44.444053'), message={'content': {'text': '查询天气', 'type': <ContentType.TEXT: 'text'>}, 'role': 'user', 'message_id': '8a3c1015-0871-4886-b70d-cf1434ad1caf'}, history=[], artifacts=[], metadata={})

TaskState

TaskState任务状态枚举类，TaskState 是一个枚举类，定义了任务的可能状态，如提交（SUBMITTED）、完成（COMPLETED）、失败（FAILED）等。它是任务生命周期的基础，用于确保状态一致性和可读性。

TaskState 状态表格：

状态名称	值	中文描述
SUBMITTED	submitted	任务已提交，等待处理。
WAITING	waiting	任务正在等待，例如等待外部资源或输入。
INPUT_REQUIRED	input-required	任务需要额外用户输入以继续执行。
COMPLETED	completed	任务已成功完成，结果可用。
CANCELED	canceled	任务被取消，未完成执行。
FAILED	failed	任务执行失败，可能包含错误信息。
UNKNOWN	unknown	未知状态，通常用于处理无效或未识别的状态。

tips：

状态名称：TaskState 枚举的成员名称（例如 TaskState.SUBMITTED），用于代码中的类型安全引用。 值：枚举的字符串值（例如 “submitted”），用于序列化（如 JSON）或与外部系统交互。 中文描述：每个状态的作用和场景，帮助开发者理解其在任务生命周期中的意义。

代码示例：

from python_a2a import TaskState  # 只需相关导入
# 检查任务状态
if TaskState.COMPLETED == "completed":
    print("任务完成")
state = TaskState.SUBMITTED
print("转换后的状态值：", state.value)
print(state)

运行结果：

任务完成
转换后的状态值： submitted
TaskState.SUBMITTED

任务完成
转换后的状态值： submitted
TaskState.SUBMITTED

TaskStatus

TaskStatus 表示 A2A 任务的当前状态对象，包括状态枚举（TaskState）、附加消息和时间戳。它用于跟踪任务进度，支持序列化和格式转换，是任务处理的动态表示。

TaskStatus 依赖 TaskState。每个 TaskStatus 实例必须有一个 TaskState 作为其 state 字段。

以下是TaskState 和 TaskStatus结合使用示例。

代码示例：

from python_a2a import TaskStatus, TaskState

status_completed = TaskStatus(
    state=TaskState.COMPLETED,
    message={"info": "任务成功完成"}
)

status_failed = TaskStatus(
    state=TaskState.FAILED,
    message={"error": "无法处理请求"}
)

# 打印字典表示
print("完成状态：", status_completed.to_dict())
print("失败状态：", status_failed.to_dict())

输出日志：

完成状态： {'state': 'completed', 'timestamp': '2025-10-24T22:55:37.883274', 'message': {'info': '任务成功完成'}}
失败状态： {'state': 'failed', 'timestamp': '2025-10-24T22:55:37.883274', 'message': {'error': '无法处理请求'}}

A2AServer

A2AServer是A2A协议的核心实现类，用于 构建代理服务器 。它继承自BaseA2AServer，支持 处理任务 （handle_task）、 消息 （handle_message）和 路由设置 （setup_routes）。它管理任务存储、流式订阅，并支持Google A2A兼容模式。它提供了Flask路由支持、任务处理逻辑和错误处理，确保代理间通信的可靠性。

task:

在继承 A2AServer 的情况下，会有一个task，通常不需要手动创建 Task 对象，因为 A2AServer 的内置机制会自动处理传入的请求并将其解析为 Task 对象，传递给 handle_task 方法。

handle_task : handle_task用于解析任务输入、处理查询、封装结果并返回task任务对象。

代码示例：