A2A Python 1.1 架构地图:一条消息如何穿过三种传输与任务状态

这篇文章只看 A2A Python SDK 的消息执行主路径：客户端如何选传输，服务端如何把请求交给 Agent，以及事件怎样被聚合成可查询、可续接的任务。先看完整地图，再逐区拆开。

▲ 完整地图把一次 message/send 分成六区；绿色箭头是主请求路径，灰色回边是事件、状态与续接路径。

01.先把边界钉死：这是 SDK，不是 Agent 大脑

本文冻结 A2A Python SDK v1.1.3，tag 指向 commit 4e71245bf2bf4b31f6429f12d97991f1f3d4b3f4，验证日期为 2026 年 8 月 21 日。项目包名是 a2a-sdk，要求 Python >=3.10，许可证为 Apache-2.0。官方 README 说明它实现 A2A Protocol 1.0，同时保留 0.3 兼容层。

这张图的系统边界是“一条 A2A 消息在 Python SDK 内的执行链”，不包括模型推理框架、业务工具、远端身份系统或生产网关。换句话说，A2A SDK 负责协议、路由、执行协调和任务状态；真正回答问题的逻辑仍由应用提供的 AgentExecutor 负责。

主路径按 01 → 02 → 03 → 04 → 05 → 06 阅读：Agent Card 告诉客户端能走哪条接口；ClientFactory 选择传输；服务端路由把协议请求归一成 RequestHandler 调用；DefaultRequestHandler 构造 RequestContext 和 ActiveTask；AgentExecutor 只向 EventQueue 写事件；消费端校验、聚合并落到 TaskStore，最后再把 Message、Task 或事件流返回客户端。

图中关系分三类。README 与公开接口直接声明的能力记为“官方确认”；从 src/a2a 调用链读出的关系记为“源码推导”；对部署方式和工程取舍的判断记为“编辑判断”。这三类不能互相冒充。

02.区域 01：Agent Card 与 ClientFactory 先决定“怎么说话”

▲ 客户端不是先绑定 HTTP 路由，而是先读取 Agent Card，再在支持的接口中选择 JSON-RPC、REST 或 gRPC。

客户端入口的关键不是 send_message() 本身，而是 Agent Card。它列出 Agent 的能力、skills、输入输出模式和 supported_interfaces。在官方端到端测试中，同一张 Card 同时声明 HTTP_JSON、JSONRPC 和 GRPC 三种接口；ClientFactory 根据 ClientConfig.supported_protocol_bindings 与 Card 的接口交集创建具体客户端。

这带来一个很实用的边界：业务调用可以保持 BaseClient.send_message() 这一层不变，传输差异被压到 ClientTransport。JSON-RPC 要封装方法名和 envelope，REST 直接落到资源路由，gRPC 走生成的 protobuf stub；认证、租户和遥测则可以在 call context、interceptor 或 transport decorator 周围叠加。

设计收益是“协议语义”和“线缆选择”分开：同一个 SendMessageRequest 不需要写三套业务流程。成本也很明确：Card 配错接口、协议版本头不匹配、客户端未安装 gRPC extra，都会在真正执行 Agent 之前失败。A2A 没有替你消灭网络问题，只是把失败位置变得更可见。

03.区域 02：三种传输在服务端统一成 RequestHandler

▲ 三种传输拥有不同的协议外壳，但最终都调用同一组 RequestHandler 方法。

服务端为 JSON-RPC、REST 和 gRPC 分别提供 dispatcher、routes 或 handler。它们负责解码协议、检查必填字段、处理版本头和把异常翻译成协议响应；真正的消息语义则落到 RequestHandler.on_message_send()、on_message_send_stream()、on_get_task()、on_cancel_task() 等统一接口。

这一层是架构图里最容易被误画成“三套服务器”的地方。源码实际表达的是三套适配器共享一个请求处理内核。FastAPI 也不是另一套执行引擎：add_a2a_routes_to_fastapi() 只是把 Starlette 路由重新注册成 FastAPI 的 APIRoute，让 OpenAPI 文档能看见 JSON-RPC 与 REST schema。

边界同样重要。HTTP server、FastAPI、gRPC、SQL、telemetry 都是 optional extras。核心包可以只做客户端或类型处理；服务端按部署需要选择依赖。生产环境若三种传输都开放，测试矩阵、认证策略和限流面也会同步扩大，不能把“SDK 支持”直接等同于“部署时应该全开”。

04.区域 03：DefaultRequestHandler 把一次请求变成可管理的 ActiveTask

▲ Handler 先补齐 task/context 标识，再把同一 task 的执行、订阅和取消统一收进 ActiveTaskRegistry。

a2a.server.request_handlers.DefaultRequestHandler 在 v1.1.3 实际是 DefaultRequestHandlerV2 的别名。它收到 SendMessageRequest 后，先检查历史长度等参数；若消息带已有 task_id，就从 TaskStore 取回任务并在不存在时返回 TaskNotFound。随后 RequestContextBuilder 解析或生成 task_id、context_id，并把消息、调用上下文和任务信息组成 RequestContext。

接下来，ActiveTaskRegistry 按 task_id 获取或创建 ActiveTask。这个对象不是数据库任务的另一份简单副本，而是“当前正在运行的协调器”：它持有请求队列、Agent 事件源、订阅者事件流和串行化同一任务请求的锁。相同 task 的后续消息因此能进入同一生命周期，而不是绕过当前状态另起炉灶。

同步 message/send、流式 message/stream 与 tasks/subscribe 在这里分叉。同步调用会消费到 Message、终态 Task、输入中断或“立即返回”条件；流式调用则逐个向上游 yield 事件；订阅接口先给现有 Task，再跟随新的状态。分叉发生在返回策略，不发生在 AgentExecutor 内部。

05.区域 04：AgentExecutor 只负责业务执行，EventQueue 是唯一出口

▲ AgentExecutor 不直接写 HTTP 响应；它把 Message、Task、状态或产物事件送入 EventQueue。

AgentExecutor 是一个只有 execute(context, event_queue) 与 cancel(context, event_queue) 两个关键异步方法的抽象接口。应用可以在这里接模型、工具、工作流或人工审批，但 SDK 不要求特定 Agent 框架。这也是 A2A 与 LangGraph、ADK、CrewAI 之类框架的分工线：前者定义 Agent 之间怎样交换消息和任务，后者通常决定 Agent 内部怎样推理和调用工具。

官方端到端测试里的 MockAgentExecutor 展示了两条合法输出路径。短交互可以只 enqueue 一个 Message；长任务则先创建 Task，再通过 TaskUpdater 写 working、artifact、completed 或 input-required 等事件。两种模式不能在同一次响应流里随意混用：消费端会拒绝“已经进入 task 模式后又突然发 Message”、终态后继续发事件、或 task_id 不一致等行为。

这个约束看似严格，却是跨 Agent 协议必须付出的成本。没有事件合法性检查，客户端无法判断一段流到底是一次性回答、持久任务，还是已经结束后仍在漂移的数据。A2A 把这类歧义收敛成显式错误，而不是让每个调用方猜。

06.区域 05：EventConsumer、TaskManager 与 TaskStore 把事件变成状态

▲ EventConsumer 校验事件序列，TaskManager 更新任务，TaskStore 提供跨请求查询与续接边界。

ActiveTask 内部是清晰的 producer/consumer 架构。producer 从请求队列取 RequestContext，在请求锁保护下调用 AgentExecutor，并把开始、完成或异常等内部事件也放进 Agent 事件源。EventConsumer 持续读取这些事件，验证 Message 模式与 Task 模式，驱动 TaskManager 更新历史、状态和 artifacts，再把可公开事件广播给每个订阅者。

TaskStore 是持久化边界而不是执行器。抽象接口提供 save、get、list、delete；默认的 InMemoryTaskStore 按 owner 与 task_id 保存任务，并用 CopyingTaskStoreAdapter 默认返回深拷贝，减少共享 protobuf 对象被外部修改的风险。SDK 也提供数据库实现与 PostgreSQL、MySQL、SQLite extras，但生产选择仍要考虑事务、迁移、租户隔离和恢复策略。

源码里另一个值得注意的变化是 owner scope。InMemoryTaskStore 不是一个裸 task_id 字典，而是先通过 OwnerResolver 得到 owner，再在该作用域内查询。这不等于完整的认证授权方案，却提醒部署者：任务隔离必须绑定可信调用上下文，不能只相信客户端送来的 task_id。

07.区域 06：返回的是 Message、Task，还是持续事件流

▲ 同一条内部事件链可投影为一次性 Message、当前 Task 快照或持续事件流。

同步调用里，DefaultRequestHandler 会订阅 ActiveTask 并等待结果。如果 Agent 只发 Message，就把 Message 直接返回；如果进入 Task 模式，则在终态、input-required、auth-required 或 return_immediately 条件下返回 Task。流式调用不会等聚合结束，而是把 Task、TaskStatusUpdateEvent、TaskArtifactUpdateEvent 等逐个传回客户端。

这里的工程取舍是延迟与完整性的交换。阻塞等待适合短任务，调用方拿到的对象最完整；return_immediately 适合先拿 task_id 再轮询；streaming 适合进度与产物实时展示；push notification 适合客户端不保持长连接。四种方式共享任务状态，但可靠性模型不同，部署时必须分别验证断线、重连、幂等和重复通知。

取消也沿着同一边界工作。Handler 先通过 ActiveTaskRegistry 找到活跃任务，再调用 AgentExecutor.cancel。若任务不存在、已经不可取消，或实现根本不支持取消，SDK 会返回明确错误。接口存在不代表业务一定可抢占；真正能否安全停止取决于 AgentExecutor 和它调用的外部系统。

08.把主路径重新走一遍：哪些地方最容易出事故

▲ 一次请求的风险集中在能力发现、传输适配、事件合法性、持久化隔离和中断恢复五个边界。

现在把整条路径压回一句话：客户端读取 Agent Card，ClientFactory 选出传输，路由把协议请求翻译给 DefaultRequestHandler，Handler 建立 RequestContext 与 ActiveTask，AgentExecutor 向 EventQueue 产出事件，EventConsumer 与 TaskManager 校验并聚合，TaskStore 保存状态，最后 Handler 按阻塞、立即返回或流式策略把结果交回客户端。

最值得复用的设计有三点。第一，协议适配与业务执行分离，使三种传输可以共享一个 RequestHandler。第二，AgentExecutor 不直接控制网络响应，所有输出先进入可校验的事件通道。第三，执行态 ActiveTask 与持久态 TaskStore 分离，让实时订阅和跨请求查询各有清晰责任。

最危险的误用也有三点。其一，把 Agent Card 当成可信身份声明；它是能力描述，不替代认证。其二，用 InMemoryTaskStore 承担生产恢复；进程退出后数据会丢。其三，只测试 REST 成功就宣称三种传输等价；序列化、版本头、连接生命周期和流控都需要独立验证。

09.源码与复现说明

本轮冻结官方 release v1.1.3、commit 4e71245bf2bf4b31f6429f12d97991f1f3d4b3f4、pyproject.toml、README、ClientFactory、三种 transport、DefaultRequestHandlerV2、ActiveTask、EventQueue、TaskStore 与官方 tests/integration/test_end_to_end.py。在 Python 3.10.20 环境中运行该官方端到端文件，结果为 54 passed、101 warnings；JSON-RPC、REST 与 gRPC 的阻塞、非阻塞、流式、任务查询、参数校验和扩展传递路径均通过。

这次复现使用官方 in-memory fixture 与本机临时 gRPC 端口，没有接入真实远端 Agent、生产数据库、认证服务、push notification 接收端或公网故障注入。因此文章确认的是 SDK 内部边界与官方测试覆盖，不是对生产可靠性、吞吐或安全性的背书。

如果你准备把一个现有 Agent 暴露成 A2A 服务，最小下一步不是同时开启三种传输：先选一种接口，用自定义 AgentExecutor 加持久化 TaskStore 跑通 send → stream → get → cancel 四条路径，再补认证、租户隔离和重启恢复测试。