10分钟入门A2A协议

在本教程中，你将使用 Python SDK 探索一个简单的“回显”（echo：就是直接返回一个固定的字符串）A2A 服务器。这将帮助你了解 A2A 服务器的基本概念和核心组件。

本教程分为以下步骤：

环境设置（Setup）：准备你的 Python 环境及 A2A SDK。
智能体技能与智能体卡片（Agent Skills & Agent Card）：定义你的智能体能做什么，以及它如何描述自身。
智能体执行器（The Agent Executor）：理解智能体逻辑是如何实现的。
启动服务器（Starting the Server）：运行 Helloworld A2A 服务器。
与服务器交互（Interacting with the Server）：向你的智能体发送请求。

环境设置（Setup）

要求：

Python 3.10 或更高版本

这里默认你已经会配置Python相关环境了

pipinstalla2a-sdk

我测试的版本是0.3.16

智能体技能（Agent Skills）与智能体卡片（Agent Card）

在 A2A 智能体能够执行任何操作之前，它必须先明确两件事：

它能做什么（即它的技能，Skills）；
其他智能体或客户端如何了解这些能力（即通过它的智能体卡片，Agent Card）。

上面这两句话已经把Agent Skills和Agent Card概况的非常准确了。
核心是Agent Card，这个需要给外部调用，而Agent Skills仅仅是Agent Card的一个组成部分，把它单独出来是为了更加明了的定义该智能体能做什么。

智能体技能（Agent Skills）

智能体技能（Agent Skill）描述了该智能体所能执行的某项具体能力或功能。它是客户端理解“这个智能体能处理哪些任务”的基本构建单元。
在 a2a.types 中定义的 AgentSkill 包含以下关键属性：

id：该技能的唯一标识符。
name：人类可读的技能名称。
description：对该技能功能的更详细说明。
tags：用于分类和发现的关键词标签。
examples：示例提示（prompts）或使用场景。
inputModes / outputModes：支持的输入和输出媒体类型（Media Types），例如 “text/plain” 或 “application/json”。

所以一个Helloworld 智能体的技能很简单：

skill=AgentSkill(id='hello_world',name='Returns hello world',description='just returns hello world',tags=['hello world'],examples=['hi','hello world'],)

这个技能非常简单：它的名称是 “Returns hello world”，主要处理纯文本输入，并原样返回固定响应。

智能体卡片（Agent Card）

智能体卡片（Agent Card）是一个 JSON 文档，由 A2A 服务器对外提供，通常可通过 .well-known/agent-card.json 端点访问。你可以把它看作是该智能体的“数字名片”。
在 a2a.types 中定义的 AgentCard 包含以下关键属性：

name, description, version：智能体的基本身份信息。
url：A2A 服务的访问端点地址。
capabilities：声明该智能体支持的 A2A 高级特性，例如流式传输（streaming）或推送通知（pushNotifications）。
defaultInputModes / defaultOutputModes：该智能体默认支持的输入和输出的类型，比如text。
skills：该智能体提供的 AgentSkill 对象列表。

一个HelloWorld智能体卡片定义如下：

# 这将是对外公开的智能体卡片public_agent_card=AgentCard(name='Hello World Agent',description='Just a hello world agent',url='http://localhost:9999/',version='1.0.0',default_input_modes=['text'],default_output_modes=['text'],capabilities=AgentCapabilities(streaming=True),skills=[skill],# 公开卡片中仅包含基础技能supports_authenticated_extended_card=True,)

这张卡片告诉我们：

智能体名为 “Hello World Agent”；
运行在 http://localhost:9999/；
支持纯文本（text）的输入与输出；
提供 hello_world 技能；
同时声明支持流式响应（streaming=True）；
并且支持通过认证获取扩展版智能体卡片（supports_authenticated_extended_card=True），但公开版本无需凭证即可访问。

理解智能体卡片至关重要，因为它是客户端发现智能体并学习如何与其交互的主要方式。当一个客户端想要调用某个 A2A 智能体时，它首先会获取该智能体的 Agent Card，从中解析出可用技能、通信格式、端点地址等信息，从而发起正确的请求。

智能体执行器（The Agent Executor）

A2A 智能体如何处理请求并生成响应或事件的核心逻辑，由智能体执行器（Agent Executor）负责。A2A Python SDK 提供了一个抽象基类 a2a.server.agent_execution.AgentExecutor，你需要继承并实现它。

AgentExecutor 接口（AgentExecutor Interface）

AgentExecutor 类定义了两个主要方法：

async def execute(self, context: RequestContext, event_queue: EventQueue)
处理传入的请求（无论是期望单次响应还是事件流）。它通过 context 获取用户输入，并使用 event_queue 向客户端发送以下类型的事件对象：

Message（消息）
Task（任务）
TaskStatusUpdateEvent（任务状态更新事件）
TaskArtifactUpdateEvent（任务产物更新事件）
async def cancel(self, context: RequestContext, event_queue: EventQueue)

处理取消正在进行的任务的请求。
其中：

RequestContext 提供了关于当前请求的信息，例如用户的输入消息、已有任务的详细信息等。
EventQueue 是执行器用来将结果或中间事件异步发送回客户端的通道。

Helloworld 智能体执行器（Helloworld Agent Executor）

智能体（HelloWorldAgent）

这是一个简单的辅助类，封装了实际的“业务逻辑”：

classHelloWorldAgent:"""Hello World Agent."""asyncdefinvoke(self)->str:return'Hello World'

它只有一个 invoke 方法，异步返回字符串 “Hello World”。

执行器（HelloWorldAgentExecutor）

该类实现了 AgentExecutor 接口。
初始化（init）：

classHelloWorldAgentExecutor(AgentExecutor):"""Test AgentProxy Implementation."""def__init__(self):self.agent=HelloWorldAgent()

在初始化时，它创建了一个 HelloWorldAgent 实例。
执行方法（execute）：

asyncdefexecute(self,context:RequestContext,event_queue:EventQueue,)->None:result=awaitself.agent.invoke()awaitevent_queue.enqueue_event(new_agent_text_message(result))

当收到 message/send 或 message/stream 请求时（在这个简化版执行器中，两者均由 execute 处理）：

调用 self.agent.invoke() 获取 “Hello World” 字符串；
使用工具函数 new_agent_text_message 创建一个符合 A2A 协议的 Message 对象；
将该消息放入 event_queue 中。
底层的 DefaultRequestHandler 会消费这个队列，并将事件发送给客户端：

对于 message/send，这将作为一次性完整响应返回；
对于 message/stream，这将作为一个单独的事件发出，随后流即关闭（因为只有一条消息）。

取消方法（cancel）：

Helloworld 示例中的 cancel 方法直接抛出异常，表明这个基础智能体不支持任务取消：

asyncdefcancel(self,context:RequestContext,event_queue:EventQueue)->None:raiseException('cancel not supported')

AgentExecutor 充当了 A2A 协议层（由请求处理器和服务器应用管理）与你自定义智能体逻辑之间的桥梁。
它接收包含用户请求上下文的 RequestContext，并通过 EventQueue 将执行结果或中间状态以标准 A2A 事件的形式返回给客户端。这种设计使得智能体逻辑与协议通信解耦，便于扩展和维护。

启动服务器（Starting the Server）

现在我们已经有了智能体卡片（Agent Card）和智能体执行器（Agent Executor），就可以配置并启动 A2A 服务器了。
A2A Python SDK 提供了一个名为 A2AStarletteApplication 的类，用于简化符合 A2A 规范的 HTTP 服务器的搭建。该类基于 Starlette Web 框架构建，通常使用 ASGI 服务器（如 Uvicorn）运行。

Helloworld 示例中的服务器配置

下面代码是服务器如何初始化和启动：

importuvicornfroma2a.server.appsimportA2AStarletteApplicationfroma2a.server.request_handlersimportDefaultRequestHandlerfroma2a.server.tasksimportInMemoryTaskStorefroma2a.typesimport(AgentCapabilities,AgentCard,AgentSkill,)fromagent_executorimport(HelloWorldAgentExecutor,# type: ignore[import-untyped])if__name__=='__main__':skill=AgentSkill(id='hello_world',name='Returns hello world',description='just returns hello world',tags=['hello world'],examples=['hi','hello world'],)extended_skill=AgentSkill(id='super_hello_world',name='Returns a SUPER Hello World',description='A more enthusiastic greeting, only for authenticated users.',tags=['hello world','super','extended'],examples=['super hi','give me a super hello'],)# 这是对外公开的智能体卡片public_agent_card=AgentCard(name='Hello World Agent',description='Just a hello world agent',url='http://localhost:9999/',version='1.0.0',default_input_modes=['text'],default_output_modes=['text'],capabilities=AgentCapabilities(streaming=True),skills=[skill],# 公开卡片仅包含基础技能supports_authenticated_extended_card=True,)# 这是经过认证后可访问的扩展版智能体卡片# 包含额外的 'extended_skill'specific_extended_agent_card=public_agent_card.model_copy(update={'name':'Hello World Agent - Extended Edition',# 为清晰起见使用不同名称'description':'The full-featured hello world agent for authenticated users.','version':'1.0.1',# 甚至可以是不同版本# capabilities、url、default_input_modes 等字段若未在此指定，# 则从 public_agent_card 继承'skills':[skill,extended_skill,],# 扩展卡片包含两个技能})request_handler=DefaultRequestHandler(agent_executor=HelloWorldAgentExecutor(),task_store=InMemoryTaskStore(),)server=A2AStarletteApplication(agent_card=public_agent_card,http_handler=request_handler,extended_agent_card=specific_extended_agent_card,)uvicorn.run(server.build(),host='0.0.0.0',port=9999)

代码解析

DefaultRequestHandler
SDK 提供的 DefaultRequestHandler 负责将 A2A 协议的 RPC 调用路由到你的执行器方法（如 execute 或 cancel）。

它接收你实现的 AgentExecutor（此处为 HelloWorldAgentExecutor）；
同时需要一个 TaskStore（此处使用内存型的 InMemoryTaskStore）。

TaskStore 的作用：管理任务的生命周期，尤其在支持状态保持、流式响应或多轮交互时至关重要。即使你的执行器逻辑很简单，请求处理器仍需要一个 TaskStore 实例。

A2AStarletteApplication
该类用于构建完整的 A2A 服务器应用：

agent_card：传入公开的智能体卡片。服务器会自动将其暴露在默认端点 /.well-known/agent-card.json 上。
http_handler（即 request_handler）：负责处理所有传入的 A2A 方法调用，并与你的 AgentExecutor 交互。
extended_agent_card（可选）：如果设置了 supports_authenticated_extended_card=True，则可通过认证获取此扩展卡片（包含更多技能或能力）。

uvicorn.run(…)

A2AStarletteApplication 的 build() 方法会生成一个标准的 Starlette 应用；
使用 uvicorn.run() 启动该应用，使其通过 HTTP 对外提供服务；
host=‘0.0.0.0’ 表示监听所有网络接口（不仅限于本地回环）；
port=9999 指定监听端口，必须与 AgentCard 中的 url 字段一致（此处为 http://localhost:9999/）。

运行 Helloworld 服务器

在终端中进入 a2a-samples 目录（如果尚未进入），并确保已激活虚拟环境。
运行以下命令启动 Helloworld 服务器：

# 从 a2a-samples 目录执行python samples/python/agents/helloworld/__main__.py

你应该会看到类似以下的输出，表明服务器已成功启动：

INFO: Started server process [xxxxx] INFO: Waiting for application startup. INFO: Application startup complete. INFO: Uvicorn running on http://0.0.0.0:9999 (Press CTRL+C to quit)

现在，你的 A2A Helloworld 智能体已上线，并开始监听来自客户端的请求！

与服务器交互（Interacting with the Server）

现在 Helloworld A2A 服务器正在运行，让我们向它发送一些请求。SDK 提供了一个客户端类 A2AClient，可简化这些交互操作。

Helloworld 测试客户端（The Helloworld Test Client）

客户端代码的功能：

从服务器获取智能体卡片（Agent Card）；
创建一个 A2AClient 实例；
发送非流式请求（message/send）和流式请求（message/stream）。

客户端代码解析（Understanding the Client Code）

获取智能体卡片并初始化客户端

base_url='http://localhost:9999'asyncwithhttpx.AsyncClient()ashttpx_client:# 初始化 A2ACardResolverresolver=A2ACardResolver(httpx_client=httpx_client,base_url=base_url,# agent_card_path 使用默认值，extended_agent_card_path 也使用默认值)

A2ACardResolver 是一个便捷工具类；
它会自动从服务器的 /.well-known/agent-card.json 端点（基于提供的 base_url）获取 AgentCard；
然后可用于初始化 A2AClient。

注意：在正常开发中，resolver 通常还会处理认证以获取扩展卡片，但 Helloworld 示例中仅使用公开卡片。

发送非流式消息（send_message）

client=A2AClient(httpx_client=httpx_client,agent_card=final_agent_card_to_use)logger.info('A2AClient initialized.')send_message_payload:dict[str,Any]={'message':{'role':'user','parts':[{'kind':'text','text':'how much is 10 USD in INR?'}],'messageId':uuid4().hex,},}request=SendMessageRequest(id=str(uuid4()),params=MessageSendParams(**send_message_payload))response=awaitclient.send_message(request)print(response.model_dump(mode='json',exclude_none=True))

send_message_payload 构造了 MessageSendParams 所需的数据；
封装为 SendMessageRequest，符合 JSON-RPC 2.0 格式；
消息对象包含：

role: “user”（表示这是用户输入）；
parts: 一个文本片段列表（此处只有一段纯文本）；
messageId: 唯一消息 ID（使用 UUID 生成）。

尽管用户发送的是 “10美元兑多少印度卢比？”，但 Helloworld 智能体的 execute 方法始终返回 “Hello World”。这体现了它作为“回显”示例的本质——不处理实际逻辑，仅验证通信流程。

服务器的 DefaultRequestHandler 会调用 execute，获取 “Hello World” 消息，并将其作为响应返回；
响应类型为 SendMessageResponse，其 result 字段包含一个 SendMessageSuccessResponse（内含智能体返回的 Message），或在出错时返回 JSONRPCErrorResponse。

关于任务 ID 的说明（Handling Task IDs – Helloworld 特例）

Helloworld 示例不会直接调用 get_task 或 cancel_task，原因如下：

当通过 message/send 调用 Helloworld 智能体时，DefaultRequestHandler 会立即返回一个完整的 Message，而不是一个 Task 对象；
只有在更复杂的智能体中，message/send 才可能返回一个 Task，其 id 可用于后续查询（get_task）或取消（cancel_task）。

发送流式消息（send_message_streaming）

streaming_request=SendStreamingMessageRequest(id=str(uuid4()),params=MessageSendParams(**send_message_payload))stream_response=client.send_message_streaming(streaming_request)asyncforchunkinstream_response:print(chunk.model_dump(mode='json',exclude_none=True))

此方法调用智能体的 message/stream 端点；
DefaultRequestHandler 同样会调用 HelloWorldAgentExecutor.execute；
execute 方法将一条 “Hello World” 消息加入事件队列，然后关闭队列；
客户端会收到一个 SendStreamingMessageResponse 事件（即一个 chunk），随后流结束；
stream_response 是一个 AsyncGenerator，因此使用 async for 遍历。

预期输出（Expected Output）

运行客户端代码后，你会看到两段 JSON 输出：

非流式响应（单条 “Hello World” 消息）：

{"jsonrpc":"2.0","id":"xxxxxxxx","result":{"type":"message","role":"agent","parts":[{"type":"text","text":"Hello World"}],"messageId":"yyyyyyyy"}}

流式响应（单个 chunk，包含 “final”: true 表示流结束）：

{"jsonrpc":"2.0","id":"zzzzzzzz","result":{"type":"message","role":"agent","parts":[{"type":"text","text":"Hello World"}],"messageId":"wwwwwwww","final":true}}