A2A Python SDK 源码架构解读：一个请求是如何被处理的

本文是 A2A 协议学习系列的第三篇。前两篇分别介绍了协议全景和规范细节，本文将走进a2a-pythonSDK 的源码，看看这个 Python 库是如何把协议规范变成可运行的代码的。

为了让内容更好理解，本文会大量使用类比和流程图，尽量避免堆砌抽象概念。

一、SDK 是什么？一句话说清楚

a2a-python（pip 包名a2a-sdk）是 A2A 协议的官方 Python 实现。

打个比方：如果 A2A 协议是"快递行业标准"（规定了包裹格式、运单格式、签收流程等），那么a2a-sdk就是一套"快递收发工具包"——它帮你打包、发送、接收、拆包，你只需要关心"包裹里装什么"。

具体来说，SDK 提供了两大能力：

• 客户端：帮你向远程 Agent 发送请求、接收响应（相当于"寄快递"）
• 服务端：帮你搭建一个 A2A 服务，接收请求、执行任务、返回结果（相当于"开快递站"）

1.1 技术栈一览

核心依赖只有 5 个包，其余都是可选的。最小安装只需pip install a2a-sdk，需要什么能力再按需加装，比如pip install "a2a-sdk[http-server]"加装 FastAPI 服务端支持。

1.2 代码目录结构

先看一下 SDK 的代码是怎么组织的，后面会逐个讲解：

a2a/ ├── types.py # 所有数据结构的定义（Task、Message、Part 等） ├── _base.py # 数据结构的基类配置 │ ├── client/ # 客户端（发送请求的一方） │ ├── client.py # 客户端接口定义 │ ├── client_factory.py # 客户端工厂（自动选择通信方式） │ ├── base_client.py # 客户端核心实现 │ ├── transports/ # 通信方式的具体实现 │ │ ├── jsonrpc.py # JSON-RPC 方式 │ │ ├── rest.py # REST 方式 │ │ └── grpc.py # gRPC 方式 │ └── middleware.py # 中间件（如自动加认证头） │ ├── server/ # 服务端（接收请求的一方） │ ├── agent_execution/ # Agent 执行引擎 │ │ ├── agent_executor.py # 开发者实现的 Agent 逻辑 │ │ └── context.py # 请求上下文信息 │ ├── request_handlers/ # 请求路由和处理 │ │ ├── default_request_handler.py # 核心调度器 │ │ ├── jsonrpc_handler.py # JSON-RPC 格式适配 │ │ └── rest_handler.py # REST 格式适配 │ ├── apps/ # Web 应用（FastAPI/Starlette） │ ├── tasks/ # 任务管理 │ │ ├── task_store.py # 任务存储接口 │ │ ├── task_manager.py # 任务状态管理 │ │ └── task_updater.py # Agent 更新任务的工具 │ └── events/ # 事件队列 │ └── event_queue.py # Agent 和服务端之间的消息通道 │ ├── grpc/ # gRPC 相关的生成代码 └── utils/ # 工具函数

二、数据模型：SDK 怎么表示协议中的数据结构

2.1 为什么用 Pydantic 而不是 protobuf？

A2A 协议用 Protocol Buffers（proto 文件）定义数据结构。按理说，SDK 可以直接用 protobuf 自动生成的 Python 类。但 SDK 选择了 Pydantic v2 来重新定义这些数据结构。

为什么？因为 protobuf 生成的 Python 类用起来不太"Pythonic"：

• 没有类型提示，IDE 补全不友好
• 没有运行时校验（传错类型不会报错）
• 序列化控制不灵活

而 Pydantic 是 Python 生态中最流行的数据校验库，写出来的代码更自然：

# protobuf 生成的类（不太好用）task=a2a_pb2.Task()task.id="xxx"task.context_id="yyy"# Pydantic 类（更 Pythonic）task=Task(id="xxx",context_id="yyy")# 自动校验：如果 id 不是字符串，立刻报错# IDE 自动补全：输入 task. 就能看到所有字段

2.2 命名转换：Python 风格 ↔ JSON 风格

A2A 协议要求 JSON 中使用 camelCase（如contextId），但 Python 习惯用 snake_case（如context_id）。SDK 通过一个基类自动处理这个转换：

classA2ABaseModel(BaseModel):model_config=ConfigDict(alias_generator=to_camel,# 自动生成 camelCase 别名serialize_by_alias=True,# 输出 JSON 时用 camelCasevalidate_by_name=True,# 写代码时用 snake_case)

效果就是：

# 写代码时用 Python 风格task=Task(context_id="abc")# 序列化为 JSON 时自动变成协议要求的格式task.model_dump_json()# → {"contextId": "abc"}

所有协议数据结构（Task、Message、Part、Artifact、AgentCard 等）都继承这个基类，开发者完全不需要操心命名转换。

2.3 核心类型速览

types.py文件约 2000 行，定义了协议中的所有数据结构。几个最重要的：

# 任务状态枚举classTaskState(str,Enum):submitted='TASK_STATE_SUBMITTED'# 已提交working='TASK_STATE_WORKING'# 处理中completed='TASK_STATE_COMPLETED'# 已完成failed='TASK_STATE_FAILED'# 失败canceled='TASK_STATE_CANCELED'# 已取消input_required='TASK_STATE_INPUT_REQUIRED'# 需要用户输入rejected='TASK_STATE_REJECTED'# 被拒绝auth_required='TASK_STATE_AUTH_REQUIRED'# 需要认证# 任务classTask(A2ABaseModel):id:str# 任务 IDcontext_id:str|None=None# 会话 IDstatus:TaskStatus# 当前状态artifacts:list[Artifact]|None=None# 产出物history:list[Message]|None=None# 交互历史# 消息内容（文本、文件、结构化数据三选一）classPart(RootModel):root:TextPart|FilePart|DataPart

三、服务端架构：一个请求是怎么被处理的

服务端是 SDK 中最复杂也最核心的部分。为了让大家理解它的工作方式，这里用一个类比来说明。

3.1 餐厅类比：理解服务端的分层设计

想象一家餐厅：

顾客点餐（HTTP 请求到达） ↓ 前台服务员（FastAPI 应用层）—— 接待顾客，记录订单 ↓ 后厨调度（DefaultRequestHandler）—— 分配任务，协调流程 ↓ 厨师（AgentExecutor）—— 真正做菜的人（开发者实现） ↓ 传菜窗口（EventQueue）—— 厨师做好一道菜就放到窗口 ↓ 服务员上菜（SSE 流式响应）—— 一道一道端给顾客

SDK 的服务端就是这样分层工作的：

HTTP 请求到达 ↓ [Web 应用层] FastAPI/Starlette —— 接收 HTTP 请求 ↓ [协议适配层] JSONRPCHandler —— 拆开 JSON-RPC 信封，识别是什么操作 ↓ [核心调度层] DefaultRequestHandler —— 创建任务、启动 Agent、管理事件 ↓ [Agent 执行层] AgentExecutor —— 开发者写的业务逻辑 ↓ [事件队列] EventQueue —— Agent 产生的事件在这里排队 ↓ [任务管理层] TaskManager + TaskStore —— 把事件转化为任务状态并保存

每一层只做自己的事，互不干扰。下面逐层介绍。

3.2 AgentExecutor：开发者唯一需要写的代码

这是整个 SDK 中最重要的概念——开发者只需要实现一个类，就能创建一个 A2A Agent。

classAgentExecutor(ABC):asyncdefexecute(self,context,event_queue):"""处理请求。从 context 读取用户输入，把结果放到 event_queue 里。"""asyncdefcancel(self,context,event_queue):"""取消任务。"""

就两个方法。SDK 把所有协议细节（HTTP 处理、JSON 序列化、任务状态管理、SSE 流式传输……）都封装好了，开发者只需要关心：

1. 从context里读取用户发了什么
2. 执行自己的业务逻辑
3. 把结果通过event_queue发出去

这就像餐厅里的厨师——你不需要知道前台怎么接单、服务员怎么上菜，你只管做菜就行。

3.3 RequestContext：Agent 能拿到什么信息

当 Agent 的execute方法被调用时，context参数包含了所有需要的信息：

context.message# 用户发来的消息context.task_id# 任务 ID（SDK 自动生成）context.context_id# 会话 ID（SDK 自动生成）context.current_task# 如果是多轮对话，这里有之前的任务信息context.get_user_input()# 便捷方法：直接拿到用户的文本输入

开发者不需要自己生成 ID、不需要自己管理会话，SDK 全部搞定。

3.4 EventQueue 和 TaskUpdater：Agent 怎么返回结果

Agent 通过EventQueue（事件队列）返回结果。但直接操作队列比较底层，所以 SDK 提供了一个更好用的工具——TaskUpdater：

asyncdefexecute(self,context,event_queue):# 创建一个 TaskUpdater，它会帮你往 event_queue 里放事件updater=TaskUpdater(event_queue=event_queue,task_id=context.task_id,context_id=context.context_id,)# 告诉客户端：我开始干活了awaitupdater.start_work()# 执行业务逻辑...result=do_something(context.get_user_input())# 把结果作为产出物返回awaitupdater.add_artifact(parts=[TextPart(text=result)])# 告诉客户端：我干完了awaitupdater.complete()

TaskUpdater提供了一组直观的方法，对应任务的各种状态：

TaskUpdater还有一个安全机制：一旦任务进入终态（完成/失败/取消/拒绝），就不能再更新了。如果你不小心在complete()之后又调用了start_work()，它会直接抛出异常，帮你发现 bug。

3.5 DefaultRequestHandler：幕后的调度中心

DefaultRequestHandler是服务端的"大脑"，负责协调所有组件。当一个请求到达时，它做的事情可以用一张流程图说清楚：

非流式请求（客户端等待最终结果）：

1. 收到客户端消息 2. 创建任务管理器（TaskManager） 3. 创建事件队列（EventQueue） 4. 在后台启动 AgentExecutor.execute() 5. 等待事件队列中的事件： ├── 收到状态更新 → 保存到 TaskStore ├── 收到产出物 → 追加到任务 └── 收到终态事件 → 停止等待 6. 把最终的 Task 返回给客户端

流式请求（客户端实时接收进度）：

1. 收到客户端消息 2. 创建任务管理器 + 事件队列 3. 在后台启动 AgentExecutor.execute() 4. 每收到一个事件，立刻： ├── 保存到 TaskStore └── 通过 SSE 推送给客户端 5. 直到收到终态事件，关闭连接

两种模式的区别就像：

• 非流式 = 你在餐厅点了外卖，等所有菜做好一起打包送来
• 流式 = 你坐在餐厅里，厨师做好一道就上一道

3.6 EventQueue：Agent 和服务端之间的"传菜窗口"

EventQueue是一个异步队列，Agent 往里放事件，服务端从里面取事件：

AgentExecutor EventQueue DefaultRequestHandler │ │ │ │ enqueue(开始工作) │ │ │─────────────────────────────→│ │ │ │ dequeue() → 开始工作 │ │ │───────────────────────────────→│ → 保存状态 │ │ │ │ enqueue(产出物) │ │ │─────────────────────────────→│ │ │ │ dequeue() → 产出物 │ │ │───────────────────────────────→│ → 保存 + 推送给客户端 │ │ │ │ enqueue(完成) │ │ │─────────────────────────────→│ │ │ │ dequeue() → 完成 │ │ │───────────────────────────────→│ → 保存 + 关闭连接

EventQueue有一个巧妙的功能——tap()（分流）。它可以创建一个"子队列"，父队列收到的所有事件会自动复制一份到子队列。这解决了一个实际问题：如果多个客户端同时订阅同一个任务的进度，每个客户端都能收到完整的事件流。

┌─→ 子队列 1 → 客户端 A EventQueue（父）────┤ └─→ 子队列 2 → 客户端 B

3.7 TaskManager 和 TaskStore：任务的保存和管理

TaskManager负责把事件队列中的事件"翻译"成任务状态的变化：

• 收到TaskStatusUpdateEvent→ 更新任务状态，把旧的状态消息存入历史
• 收到TaskArtifactUpdateEvent→ 把产出物追加到任务
• 收到Task对象 → 直接保存

TaskStore是任务的存储后端，SDK 提供了两种实现：

你也可以自己实现一个TaskStore（比如基于 Redis），只需要实现三个方法：

classTaskStore(ABC):asyncdefsave(self,task):...# 保存任务asyncdefget(self,task_id):...# 读取任务asyncdefdelete(self,task_id):...# 删除任务

3.8 协议适配层：同一套逻辑，多种协议格式

A2A 协议支持三种通信格式：JSON-RPC、REST、gRPC。SDK 通过"适配层"让同一套核心逻辑支持所有格式：

JSON-RPC 请求 ──→ JSONRPCHandler ──→ ┐ │ REST 请求 ─────→ RESTHandler ────────→├──→ DefaultRequestHandler（核心逻辑） │ gRPC 请求 ─────→ GRPCHandler ────────→┘

每个 Handler 做的事情很简单：

• JSON-RPC Handler：拆开 JSON-RPC 的"信封"（提取method和params），调用核心逻辑，再把结果包回 JSON-RPC 格式
• REST Handler：从 URL 路径和 HTTP 方法判断操作类型，调用核心逻辑，返回纯 JSON
• gRPC Handler：从 protobuf 消息中提取参数，调用核心逻辑，返回 protobuf 响应

开发者不需要关心这些——选择哪种格式是在搭建服务时决定的，Agent 的业务逻辑完全不受影响。

3.9 Web 应用层：一行代码启动服务

SDK 提供了开箱即用的 FastAPI 和 Starlette 应用，自动注册所有需要的路由：

# JSON-RPC 模式froma2a.server.apps.jsonrpcimportA2AFastAPIApplication app=A2AFastAPIApplication(agent_card=card,http_handler=handler)fastapi_app=app.build()# 自动注册：# GET /.well-known/agent-card.json → 返回 Agent Card# POST / → 处理所有 A2A 请求

# REST 模式froma2a.server.apps.restimportA2ARESTFastAPIApplication app=A2ARESTFastAPIApplication(agent_card=card,http_handler=handler)fastapi_app=app.build()# 自动注册：# GET /.well-known/agent-card.json → 返回 Agent Card# POST /v1/message:send → 发送消息# POST /v1/message:stream → 流式消息# GET /v1/tasks/{id} → 获取任务# POST /v1/tasks/{id}:cancel → 取消任务# ...

四、客户端架构：怎么和远程 Agent 通信

4.1 三层设计：像打电话一样简单

客户端的设计目标是：不管远程 Agent 用的是 JSON-RPC、REST 还是 gRPC，调用方式都一样。

你的代码 ↓ 调用 client.send_message() Client（统一接口） ↓ 自动选择通信方式 ClientTransport（通信层） ↓ 具体的 HTTP/gRPC 调用 JsonRpcTransport / RestTransport / GrpcTransport

这就像打电话——你不需要知道对方用的是移动还是联通，你只管拨号就行，运营商的事情手机帮你处理了。

4.2 ClientFactory：自动选择最佳通信方式

ClientFactory是创建客户端的推荐方式。它会读取 Agent Card 中声明的通信方式，自动选择一个双方都支持的：

# 最简单的用法：给一个 URL，自动搞定一切client=awaitClientFactory.connect('https://agent.example.com')# 也可以更精细地控制client=awaitClientFactory.connect('https://agent.example.com',client_config=ClientConfig(streaming=True,# 我要用流式supported_transports=['JSONRPC'],# 我只支持 JSON-RPC),)

connect方法内部做了这些事：

1. 访问https://agent.example.com/.well-known/agent-card.json获取 Agent Card
2. 看 Agent Card 里声明支持哪些通信方式（JSON-RPC？REST？gRPC？）
3. 和客户端配置对比，找到双方都支持的方式
4. 创建对应的通信层实例
5. 返回一个可以直接使用的Client对象

4.3 使用客户端：发送消息和接收响应

# 发送消息asyncforeventinclient.send_message(Message(role=Role.user,parts=[TextPart(text="今天天气怎么样？")])):ifisinstance(event,Message):# Agent 直接回复了一条消息（没有创建任务）print("Agent 说：",event.parts[0].root.text)else:# Agent 创建了一个任务task,update=eventprint(f"任务状态：{task.status.state}")iftask.artifacts:print(f"产出物：{task.artifacts[0].parts[0].root.text}")

注意send_message返回的是一个异步迭代器（async for）。无论底层是流式还是非流式，上层代码的写法都一样。这是 SDK 的一个巧妙设计——统一了流式和非流式的编程模型。

4.4 中间件：在每个请求上"加料"

客户端支持中间件（Interceptor），可以在每个请求发出前做一些处理，最常见的用途是自动添加认证信息：

classMyAuthInterceptor(ClientCallInterceptor):asyncdefintercept(self,method,payload,http_kwargs,card,context):# 在每个请求的 HTTP 头里加上 Tokenheaders=http_kwargs.setdefault('headers',{})headers['Authorization']=f'Bearer{my_token}'returnpayload,http_kwargs# 创建客户端时传入中间件client=awaitClientFactory.connect('https://agent.example.com',interceptors=[MyAuthInterceptor()],)

五、完整示例：从零搭建一个 A2A Agent

把上面讲的串起来，看看搭建一个完整的 A2A Agent 需要写多少代码：

froma2a.server.agent_executionimportAgentExecutor,RequestContextfroma2a.server.eventsimportEventQueuefroma2a.server.tasksimportTaskUpdater,InMemoryTaskStorefroma2a.server.request_handlersimportDefaultRequestHandler,JSONRPCHandlerfroma2a.server.apps.jsonrpcimportA2AFastAPIApplicationfroma2a.typesimport(AgentCard,AgentCapabilities,AgentInterface,AgentSkill,TextPart,)# ========== 第一步：实现你的 Agent 逻辑 ==========classEchoAgent(AgentExecutor):asyncdefexecute(self,context:RequestContext,event_queue:EventQueue):updater=TaskUpdater(event_queue=event_queue,task_id=context.task_id,context_id=context.context_id,)awaitupdater.start_work()# 你的业务逻辑写在这里user_input=context.get_user_input()result=f"你说的是：{user_input}"awaitupdater.add_artifact(parts=[TextPart(text=result)])awaitupdater.complete()asyncdefcancel(self,context:RequestContext,event_queue:EventQueue):updater=TaskUpdater(event_queue=event_queue,task_id=context.task_id,context_id=context.context_id,)awaitupdater.cancel()# ========== 第二步：定义 Agent Card（你的 Agent 的"名片"）==========card=AgentCard(name="Echo Agent",description="一个简单的回声 Agent，会把你说的话原样返回",version="1.0.0",supported_interfaces=[AgentInterface(url="http://localhost:8000",protocol_binding="JSONRPC",protocol_version="1.0",)],capabilities=AgentCapabilities(streaming=True),default_input_modes=["text/plain"],default_output_modes=["text/plain"],skills=[AgentSkill(id="echo",name="Echo",description="回声服务",tags=["echo"],)],)# ========== 第三步：组装并启动 ==========handler=DefaultRequestHandler(agent_executor=EchoAgent(),task_store=InMemoryTaskStore(),)jsonrpc_handler=JSONRPCHandler(agent_card=card,request_handler=handler)app=A2AFastAPIApplication(agent_card=card,http_handler=jsonrpc_handler)fastapi_app=app.build()# 启动命令：uvicorn main:fastapi_app --host 0.0.0.0 --port 8000

三步就搞定了：

1. 写 Agent 逻辑（EchoAgent）
2. 填 Agent 名片（AgentCard）
3. 组装启动（把各个组件串起来）

其中真正需要动脑子的只有第一步——你的 Agent 要做什么。剩下的都是"填表"和"接线"。

六、SDK 的设计哲学：可插拔的组件

SDK 的一个核心设计原则是：几乎所有组件都可以替换。就像乐高积木，每个部件都有标准接口，你可以用默认的，也可以换成自己的。

这种设计的好处是：

• 入门简单：用默认组件，几行代码就能跑起来
• 扩展灵活：需要定制时，只替换需要的部分，其他不动
• 测试友好：可以用内存实现做单元测试，用数据库实现做集成测试

七、协议概念到 SDK 代码的对照表

如果你读过前两篇关于协议规范的文章，下面这张表可以帮你快速找到"协议里说的那个东西，在 SDK 里对应哪段代码"：

八、架构亮点总结

回顾整个 SDK 的设计，有几个特别值得学习的地方：

1. AgentExecutor 的极简接口

只有execute和cancel两个方法。所有协议复杂性都被封装在 SDK 内部，开发者的学习成本降到最低。这是"好的抽象"的典范——把复杂的东西藏起来，只暴露简单的接口。

2. EventQueue 的"分流"设计

tap()方法让同一个任务的事件可以同时推送给多个客户端，不需要引入 Redis 或 Kafka 这样的外部消息队列。对于大多数场景来说，这个内置方案已经够用了。

3. Pydantic 的命名转换

一个基类配置就解决了 Python snake_case 和 JSON camelCase 之间的转换问题。开发者写代码时用 Python 风格，序列化时自动变成协议要求的格式，完全无感。

4. TaskUpdater 的安全保护

通过锁和状态标记，在运行时防止 Agent 在任务已经结束后继续发送事件。这种"防呆设计"能帮开发者在开发阶段就发现 bug，而不是等到生产环境才出问题。

5. 可选依赖的分层安装

核心包只有 5 个依赖，FastAPI、gRPC、数据库、OpenTelemetry 等都是可选的。这意味着如果你只需要客户端功能，安装包会非常轻量。

九、总结

a2a-pythonSDK 的核心思路可以用一句话概括：把协议的复杂性封装起来，让开发者只关注业务逻辑。

对于想要构建 A2A Agent 的 Python 开发者来说：

• 你只需要实现一个AgentExecutor（写业务逻辑）
• 填一个AgentCard（描述你的 Agent 能做什么）
• 用 SDK 提供的组件把它们串起来

协议处理、任务管理、事件分发、多协议支持、持久化——这些"脏活累活"全部由 SDK 代劳。

参考资料：

• a2a-python SDK 源码
• a2a-python SDK 文档
• A2A 示例代码
• Pydantic v2 文档