A2aAgentExecutor 完全指南为 ADK Agent 定制 A2A 服务端的请求拦截与事件翻译【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-pythonA2aAgentExecutor是 ADKAgent Development Kit中位于 A2A 服务器与 ADKRunner之间的适配层它接收来自其他 Agent 的 A2A 请求驱动你的 Agent 运行并把 ADK 事件翻译成调用方能够理解的 A2A 任务更新。本篇指南以 executor/a2a_agent_executor/index.md 为核心骨架结合仓库源码系统讲解它的三种拦截钩子before_agent/after_event/after_agent、完整配置参数与转换器替换机制并给出运行前拒绝请求为终态事件添加元数据两个可直接落地的实战方案。读完你将能够在不改动 Agent 的前提下自主实现鉴权、配额控制、出站内容脱敏、审计追踪等能力。它是什么A2A 服务器与 ADK Runner 之间的翻译层A2aAgentExecutor是 a2a-sdk 中AgentExecutor接口的 ADK 实现见 a2a_agent_executor.py。它做的事情可以用一句话概括接收来自另一个 Agent 的 A2A 请求 → 在你的 Agent 上运行它 → 把产生的 ADK 事件翻译成调用方看得懂的 A2A 任务更新。如果你希望在请求进入、事件出去的任何环节做检查或改动动手点就在这里。A2A 是协议级别的互联调用方可能是队友的 Agent也可能是完全不懂 Python 的其他语言系统因此这一层拦截是跨进程边界上唯一的控制点。何时需要自己构造它正常情况下to_a2a会为你自动构建一个默认配置的A2aAgentExecutor普通部署到此为止。只有当默认翻译不满足需求时你才需要亲手构造它常见场景包括在付出一次模型调用之前就把请求拒掉如校验调用方 Header、元数据字段或剩余配额把请求头中的某个值写入会话在文本离开你的进程之前进行脱敏scrub/redact随任务的来去建立自己的审计追踪。上述所有能力都集中在A2aAgentExecutorConfig.execute_interceptors中构造好 executor 后通过to_a2a(agent, agent_executor_factory...)把它安装回去。导入路径executor 相关的类位于嵌套模块中且不会被再导出必须使用完整导入路径from google.adk.a2a.executor.a2a_agent_executor import A2aAgentExecutor from google.adk.a2a.executor.config import A2aAgentExecutorConfig from google.adk.a2a.executor.config import ExecuteInterceptor从源码看ExecuteInterceptor与A2aAgentExecutorConfig定义在 config.py 中A2aAgentExecutor本身在 a2a_agent_executor.py。除此之外executor 目录 下还包含executor_context.pyExecutorContext、utils.py拦截器链执行逻辑、task_result_aggregator.py终态聚合与a2a_agent_executor_impl.py新版实现。快速开始一个出站脱敏拦截器下面这个拦截器会把 Agent 发回的每一条状态消息文本都替换掉——这正是做脱敏redaction时最典型的起点形态。你交给to_a2a的agent_executor_factory会收到它解析好的 runner并必须返回 executorfrom google.adk.a2a.executor.a2a_agent_executor import A2aAgentExecutor from google.adk.a2a.executor.config import A2aAgentExecutorConfig from google.adk.a2a.executor.config import ExecuteInterceptor from google.adk.a2a.utils.agent_to_a2a import to_a2a async def redact_outgoing(executor_context, a2a_event, adk_event): Replaces the text of every outgoing status message. message getattr(getattr(a2a_event, status, None), message, None) if message is not None: for part in message.parts: if getattr(part, text, None): part.text [redacted] return a2a_event def build_executor(runner): return A2aAgentExecutor( runnerrunner, configA2aAgentExecutorConfig( execute_interceptors[ ExecuteInterceptor(after_eventredact_outgoing) ] ), ) a2a_app to_a2a(root_agent, port8001, agent_executor_factorybuild_executor)然后照常启动 ASGI 服务uvicorn my_module:a2a_app --host localhost --port 8001几点说明ExecuteInterceptor是一个包含三个可选钩子的 dataclass只用一个钩子时其余两个留空即可每个钩子都是 async 的即使实现里没有任何await也必须是async defagent_executor_factory由to_a2a在服务启动阶段调用传入它解析好的 runner详见 agent_to_a2a 指南 的agent_executor_factory参数说明。工作原理一次 A2A 请求的完整旅程A2A 是协议因此你 Agent 应答的每个请求都走同一条固定序列。这个序列在 executor 介入之前就已经开始了——如果你要写拦截器能触碰到的步骤从第 3 步开始前两步发生在客户端进程内部客户端读取你的 Agent 卡agent card。卡片告诉它往哪里 POST、你的 Agent 能做什么。客户端向 RPC 端点发送一条消息并指明该消息所属的会话。从这一刻起客户端在监听单个 A2A 任务task它收到的每一个答复——包括最后一个——都是以发布在该任务上的更新形式到达的而不是某个调用的返回值。这正是一个完全不产生更新的请求会让客户端一直等待而不是报错的原因。executor 解析 runner并取用请求中指定的会话若会话不存在则创建它。executor 打开任务。第一条更新宣布一个全新任务处于submitted状态下一条把它推进到携带app_name、user_id、session_id等 ADK 元数据的working状态。对照源码 a2a_agent_executor.pyworking状态事件正是通过_compat.make_task_status_update_event发布并把三个 ADK 元数据键写入metadata。executor 在传入消息上运行你的 Agent并把运行产生的每个 ADK 事件转换成任务上的一条更新。客户端是实时看到这些更新还是通过轮询收集取决于 Agent 卡to_a2a默认构建的卡片不声明 streaming而 A2A 服务器会拒绝针对这种卡片的流式请求因此默认部署走的是轮询。任务以三种方式之一收尾收尾方式就是客户端得知发生了什么的手段运行以working状态正常结束且带内容 → 发布携带该内容的artifact 更新随后跟一条completed状态运行以其他状态结束 → 把该状态作为最终状态发布运行抛异常 → 发布携带异常文本的failed状态。这样Agent 内部的崩溃对调用方呈现为任务失败而不是连接被切断。第 6 点的实现细节可以在 a2a_agent_executor.py 中看到TaskResultAggregator见 task_result_aggregator.py负责收集运行期间的状态信号优先级为failedauth_requiredinput_requiredworking最终要么发布TaskArtifactUpdateEventcompleted要么发布聚合出的最终状态。三个钩子The hooks你的拦截器位于这条序列中的三个点Hook运行时机接收参数返回值执行顺序before_agent一次Agent 开始之前RequestContextRequestContext列表顺序after_event每个出站 A2A 事件入队之前ExecutorContext、A2A 事件、产生它的 ADK 事件事件 / 事件列表 /None列表顺序after_agent一次在终态状态事件上ExecutorContext、终态TaskStatusUpdateEvent该事件逆序列表反向对照源码 utils.py 的三个执行函数可以确认execute_before_agent_interceptors按列表顺序逐个调用before_agentexecute_after_event_interceptors按列表顺序调用after_event且支持一个事件换成多个事件的列表返回见下文execute_after_agent_interceptors用reversed(execute_interceptors)反转列表后调用after_agent。每个before_agent钩子接收的是上一个钩子返回的RequestContext所以一串钩子可以像流水线一样组合。after_agent的逆序是有意为之也是最容易踩坑的地方当拦截器列表是[a, b]时before_agent先跑a再跑b而after_agent先跑b再跑a——也就是说每个拦截器自己的两个钩子会嵌套包裹住列表中排在它后面的拦截器。而after_event不反转。ExecutorContext钩子里的只读上下文ExecutorContext是一个小型只读对象携带app_name、user_id、session_id和runner四个字段见 executor_context.py。当钩子需要知道自己正在处理哪个会话或者需要经由runner触达会话服务如runner.session_service、runner.artifact_service时就使用它。三个必须牢记的行为陷阱1.after_event返回None会丢弃事件并且会把任务余下的部分一起带走。终态的completed状态和 artifact 更新与其他事件一样会经过钩子因此一个无条件返回None的拦截器会让客户端盯着一辈子都不结束的任务干等。务必只针对你确实想丢弃的具体事件做过滤其余原样返回。2.after_event可以返回列表用一个事件替换出多个事件。每个事件按顺序入队并且链中的下一个拦截器会逐个处理它们。源码 utils.py 中可见若返回None则跳过相当于丢弃若返回列表则展开合并进待处理队列当展开后队列为空整条链提前返回空列表。3. 就地修改事件的影响范围比你想的大。最终 artifact 与生成它的状态消息共享同一份内容所以改写状态更新的文本也会改写那个 artifact。做脱敏时这通常正是你想要的但如果只是想加注解这就是个意外。配置选项A2aAgentExecutor的四个构造参数A2aAgentExecutor本身接受四个参数全部是 keyword-only见 a2a_agent_executor.pyOptionTypeDefaultDescriptionrunnerRunner \| Callable[..., Runner \| Awaitable[Runner]]requiredRunner或可返回 Runner 的工厂。configA2aAgentExecutorConfig \| NoneNone拦截器与转换器。省略时构建默认配置。use_legacyboolFalse无论请求如何强制使用旧版实现。force_new_versionboolFalse无论请求如何强制使用新版实现。关于runner的延迟解析。传入可调用对象callable会把 Runner 的构建推迟到第一个请求到来时这对于Runner 持有你不希望在 import 时就打开的连接池的场景非常重要。同步和异步 callable 都可以源码 a2a_agent_executor.py 显示_resolve_runner会先检查是否已是Runner实例否则调用 callable 并用inspect.isawaitable判断是否需要await解析结果会被缓存供后续请求复用。既不是Runner也不是 callable 的值会在第一个请求时抛出TypeError而非构造时报错。A2aAgentExecutorConfigPydantic 配置模型A2aAgentExecutorConfig是一个 Pydantic 模型持有拦截器以及沿途使用的转换器定义见 config.pyOptionTypeDefaultDescriptionexecute_interceptorslist[ExecuteInterceptor] \| NoneNone上文介绍的三个钩子。a2a_part_convertercallableconvert_a2a_part_to_genai_part把入站 A2A part 转为 GenAI part。gen_ai_part_convertercallableconvert_genai_part_to_a2a_part把出站 GenAI part 转为 A2A part。request_convertercallableconvert_a2a_request_to_agent_run_request把 A2A 请求转为 runner 参数。event_convertercallableconvert_event_to_a2a_events把一个 ADK 事件转为 A2A 事件。旧版实现使用。adk_event_convertercallableconvert_event_to_a2a_events同样的工作供新版实现使用。从源码可以补充几个有价值的细节入站方向convert_a2a_request_to_agent_run_request见 request_converter.py把 A2A 请求转成一个AgentRunRequest模型——它最终以**vars(run_request)的形式展开为runner.run_async(...)的关键字参数。session_id默认取context_iduser_id在 A2A 服务器启用了认证时取call_context.user.user_name否则退化为A2A_USER_{context_id}请求的 A2A 元数据会被包进RunConfig(custom_metadata{a2a_metadata: ...})在 Agent 侧可通过run_config读取。part 转换convert_a2a_part_to_genai_part/convert_genai_part_to_a2a_part见 part_converter.py覆盖文本、文件URI 与内联字节、数据 part 的相互转换其中函数调用、函数响应、代码执行结果等通过带type标签的 A2A data part 携带——这也是 human-in-the-loop 与鉴权请求响应的传输基础。内置 artifact 拦截器executor 目录还内置了一个include_artifacts_in_a2a_event_interceptor见 interceptors/include_artifacts_in_a2a_event.py它通过runner.artifact_service把 ADK 事件中的artifact_delta转为 A2ATaskArtifactUpdateEvent可作为自定义拦截器的参考实现。何时该替换转换器而非加拦截器替换转换器是比重加拦截器重得多的改动因为这意味着你接管了该阶段的整个翻译过程而不是调整它产出的结果。正确策略是先尝试拦截器只有当翻译本身的形态就不适合你时才考虑转换器。高级应用下面两个例子从一次运行的两端使用同样的钩子第一个在 Agent 什么都没做之前行动此时拒绝几乎零成本第二个作用于任务的最后一个事件此时一切结果都已知。在 Agent 运行前拒绝请求解决的问题在付出一次模型调用之前先检查调用方的某些信息——比如 Header、元数据字段或剩余配额。实现方式从before_agent中 raise。async def enforce_quota(context): if _over_quota(context): raise PermissionError(quota exhausted for this caller) return context关键约束before_agent必须返回一个RequestContext这里没有返回None即中止的约定。返回你收到的 context可以修改过是唯一非抛异常的结果路径。在这里 raise 与在后面失败不一样调用方能够分辨出区别从before_agent抛出的异常会到达 A2A 服务器被转换为JSON-RPC 错误{code: -32603, message: quota exhausted for this caller}并且根本不会创建任务在after_event或after_agent中 raise 的钩子给客户端的是一个以failed状态结束的任务携带同样的文本并且是在已经发布的那些事件之后到达的。两条路径都能把你的消息送到调用方但尽早拒绝留下的碎片更少。注解终态事件解决的问题让每个完成的任务都携带客户端可读的 trace id 或成本数字。实现方式在after_agent中向最终事件的元数据添加内容。该事件已经带有adk_前缀的 app name、user id、session id、invocation id、author 和 event id。async def stamp_trace(executor_context, final_event): final_event.metadata[trace_id] _trace_id_for(executor_context.session_id) return final_event对照源码 a2a_agent_executor.py终态事件的元数据正是由 executor 在运行结束后统一组装除了三个会话维度的键之外还会从最后一个 ADK 事件上取invocation_id、author、event_id分别写入adk_invocation_id、adk_author、adk_event_id。你在after_agent里追加的自定义键会与这些键并存客户端可以在终态事件上一次性读取全部上下文。已知局限两套实现由客户端选择。哪个实现处理请求取决于调用方是否请求了https://google.github.io/adk-docs/a2a/a2a-extension/扩展源码中的判定逻辑见 a2a_agent_executor.py 与_NEW_A2A_ADK_INTEGRATION_EXTENSION因此同一个服务器面对两个客户端可能表现出不同行为。需要固定行为时用use_legacy或force_new_version钉死。拦截器不按名字或优先级安装。它们按列表顺序运行after_agent反转且一个拦截器没有办法跳过其余拦截器——除非在after_event中把事件列表清空。after_event永远看不到被转换器丢弃的 ADK 事件。没有内容的 ADK 事件例如工作流节点只发出Event(output...)不会产生任何 A2A 事件也就不会触发任何钩子。当一个运行产生的所有事件都被这样丢弃时就无内容可发布了任务停留在working且没有 artifact客户端会一直等待一个永远不会来的完成事件。被服务的Workflow默认就会踩中这个坑——因为从节点返回值是写工作流的常规方式详见 to_a2a 指南 的对应局限说明至少要让一个节点额外yield Event(message...)携带调用方应接收的文本。钩子内的失败会拖垮整个请求。没有按拦截器隔离的错误机制。after_event或after_agent中的异常会让任务以failed结束而已经发布的事件仍然保持已发布所以客户端可能看到已产出 artifact 的任务随后失败。实验特性。executor 与其 config 都被a2a_experimental装饰见 experimental.py构造时会发出UserWarning。设置环境变量ADK_SUPPRESS_A2A_EXPERIMENTAL_FEATURE_WARNINGS可以静默该警告。注意A2A 协议本身不是实验性的实验性的是 ADK 对它的实现。相关示例与指南A2A root agent 通过to_a2a提供 Agent 服务即用默认配置构建本 executor示例中的客户端形态可参考 agent.py它使用RemoteA2aAgent指向服务端卡片A2A human in the loop 是一个在任务中途暂停的被服务 Agent见 agent.py它会派生出你的钩子会看到的非终态状态更新如input-required类信号to_a2a 是构建本 executor 的函数其中详细记录了把 executor 换成你自定义版本的agent_executor_factory参数AgentCardBuilder 描述客户端在向你发送请求之前就会读取的那张卡片A2A remote agent configuration 介绍同一段对话客户端一侧的镜像拦截器——RemoteA2aAgent的A2aRemoteAgentConfig提供了request_interceptors、card_request_interceptors与入站事件转换器与服务端本 executor 的钩子一一对应。单元测试方面test_a2a_agent_executor.py 覆盖了 executor 的拦截器调用顺序、None丢弃语义、终态事件组装等行为可以作为理解钩子语义的补充参考。【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 A2aAgentExecutor 完全指南：为 ADK Agent 定制 A2A 服务端的请求拦截与事件翻译
A2aAgentExecutor 完全指南为 ADK Agent 定制 A2A 服务端的请求拦截与事件翻译【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-pythonA2aAgentExecutor是 ADKAgent Development Kit中位于 A2A 服务器与 ADKRunner之间的适配层它接收来自其他 Agent 的 A2A 请求驱动你的 Agent 运行并把 ADK 事件翻译成调用方能够理解的 A2A 任务更新。本篇指南以 executor/a2a_agent_executor/index.md 为核心骨架结合仓库源码系统讲解它的三种拦截钩子before_agent/after_event/after_agent、完整配置参数与转换器替换机制并给出运行前拒绝请求为终态事件添加元数据两个可直接落地的实战方案。读完你将能够在不改动 Agent 的前提下自主实现鉴权、配额控制、出站内容脱敏、审计追踪等能力。它是什么A2A 服务器与 ADK Runner 之间的翻译层A2aAgentExecutor是 a2a-sdk 中AgentExecutor接口的 ADK 实现见 a2a_agent_executor.py。它做的事情可以用一句话概括接收来自另一个 Agent 的 A2A 请求 → 在你的 Agent 上运行它 → 把产生的 ADK 事件翻译成调用方看得懂的 A2A 任务更新。如果你希望在请求进入、事件出去的任何环节做检查或改动动手点就在这里。A2A 是协议级别的互联调用方可能是队友的 Agent也可能是完全不懂 Python 的其他语言系统因此这一层拦截是跨进程边界上唯一的控制点。何时需要自己构造它正常情况下to_a2a会为你自动构建一个默认配置的A2aAgentExecutor普通部署到此为止。只有当默认翻译不满足需求时你才需要亲手构造它常见场景包括在付出一次模型调用之前就把请求拒掉如校验调用方 Header、元数据字段或剩余配额把请求头中的某个值写入会话在文本离开你的进程之前进行脱敏scrub/redact随任务的来去建立自己的审计追踪。上述所有能力都集中在A2aAgentExecutorConfig.execute_interceptors中构造好 executor 后通过to_a2a(agent, agent_executor_factory...)把它安装回去。导入路径executor 相关的类位于嵌套模块中且不会被再导出必须使用完整导入路径from google.adk.a2a.executor.a2a_agent_executor import A2aAgentExecutor from google.adk.a2a.executor.config import A2aAgentExecutorConfig from google.adk.a2a.executor.config import ExecuteInterceptor从源码看ExecuteInterceptor与A2aAgentExecutorConfig定义在 config.py 中A2aAgentExecutor本身在 a2a_agent_executor.py。除此之外executor 目录 下还包含executor_context.pyExecutorContext、utils.py拦截器链执行逻辑、task_result_aggregator.py终态聚合与a2a_agent_executor_impl.py新版实现。快速开始一个出站脱敏拦截器下面这个拦截器会把 Agent 发回的每一条状态消息文本都替换掉——这正是做脱敏redaction时最典型的起点形态。你交给to_a2a的agent_executor_factory会收到它解析好的 runner并必须返回 executorfrom google.adk.a2a.executor.a2a_agent_executor import A2aAgentExecutor from google.adk.a2a.executor.config import A2aAgentExecutorConfig from google.adk.a2a.executor.config import ExecuteInterceptor from google.adk.a2a.utils.agent_to_a2a import to_a2a async def redact_outgoing(executor_context, a2a_event, adk_event): Replaces the text of every outgoing status message. message getattr(getattr(a2a_event, status, None), message, None) if message is not None: for part in message.parts: if getattr(part, text, None): part.text [redacted] return a2a_event def build_executor(runner): return A2aAgentExecutor( runnerrunner, configA2aAgentExecutorConfig( execute_interceptors[ ExecuteInterceptor(after_eventredact_outgoing) ] ), ) a2a_app to_a2a(root_agent, port8001, agent_executor_factorybuild_executor)然后照常启动 ASGI 服务uvicorn my_module:a2a_app --host localhost --port 8001几点说明ExecuteInterceptor是一个包含三个可选钩子的 dataclass只用一个钩子时其余两个留空即可每个钩子都是 async 的即使实现里没有任何await也必须是async defagent_executor_factory由to_a2a在服务启动阶段调用传入它解析好的 runner详见 agent_to_a2a 指南 的agent_executor_factory参数说明。工作原理一次 A2A 请求的完整旅程A2A 是协议因此你 Agent 应答的每个请求都走同一条固定序列。这个序列在 executor 介入之前就已经开始了——如果你要写拦截器能触碰到的步骤从第 3 步开始前两步发生在客户端进程内部客户端读取你的 Agent 卡agent card。卡片告诉它往哪里 POST、你的 Agent 能做什么。客户端向 RPC 端点发送一条消息并指明该消息所属的会话。从这一刻起客户端在监听单个 A2A 任务task它收到的每一个答复——包括最后一个——都是以发布在该任务上的更新形式到达的而不是某个调用的返回值。这正是一个完全不产生更新的请求会让客户端一直等待而不是报错的原因。executor 解析 runner并取用请求中指定的会话若会话不存在则创建它。executor 打开任务。第一条更新宣布一个全新任务处于submitted状态下一条把它推进到携带app_name、user_id、session_id等 ADK 元数据的working状态。对照源码 a2a_agent_executor.pyworking状态事件正是通过_compat.make_task_status_update_event发布并把三个 ADK 元数据键写入metadata。executor 在传入消息上运行你的 Agent并把运行产生的每个 ADK 事件转换成任务上的一条更新。客户端是实时看到这些更新还是通过轮询收集取决于 Agent 卡to_a2a默认构建的卡片不声明 streaming而 A2A 服务器会拒绝针对这种卡片的流式请求因此默认部署走的是轮询。任务以三种方式之一收尾收尾方式就是客户端得知发生了什么的手段运行以working状态正常结束且带内容 → 发布携带该内容的artifact 更新随后跟一条completed状态运行以其他状态结束 → 把该状态作为最终状态发布运行抛异常 → 发布携带异常文本的failed状态。这样Agent 内部的崩溃对调用方呈现为任务失败而不是连接被切断。第 6 点的实现细节可以在 a2a_agent_executor.py 中看到TaskResultAggregator见 task_result_aggregator.py负责收集运行期间的状态信号优先级为failedauth_requiredinput_requiredworking最终要么发布TaskArtifactUpdateEventcompleted要么发布聚合出的最终状态。三个钩子The hooks你的拦截器位于这条序列中的三个点Hook运行时机接收参数返回值执行顺序before_agent一次Agent 开始之前RequestContextRequestContext列表顺序after_event每个出站 A2A 事件入队之前ExecutorContext、A2A 事件、产生它的 ADK 事件事件 / 事件列表 /None列表顺序after_agent一次在终态状态事件上ExecutorContext、终态TaskStatusUpdateEvent该事件逆序列表反向对照源码 utils.py 的三个执行函数可以确认execute_before_agent_interceptors按列表顺序逐个调用before_agentexecute_after_event_interceptors按列表顺序调用after_event且支持一个事件换成多个事件的列表返回见下文execute_after_agent_interceptors用reversed(execute_interceptors)反转列表后调用after_agent。每个before_agent钩子接收的是上一个钩子返回的RequestContext所以一串钩子可以像流水线一样组合。after_agent的逆序是有意为之也是最容易踩坑的地方当拦截器列表是[a, b]时before_agent先跑a再跑b而after_agent先跑b再跑a——也就是说每个拦截器自己的两个钩子会嵌套包裹住列表中排在它后面的拦截器。而after_event不反转。ExecutorContext钩子里的只读上下文ExecutorContext是一个小型只读对象携带app_name、user_id、session_id和runner四个字段见 executor_context.py。当钩子需要知道自己正在处理哪个会话或者需要经由runner触达会话服务如runner.session_service、runner.artifact_service时就使用它。三个必须牢记的行为陷阱1.after_event返回None会丢弃事件并且会把任务余下的部分一起带走。终态的completed状态和 artifact 更新与其他事件一样会经过钩子因此一个无条件返回None的拦截器会让客户端盯着一辈子都不结束的任务干等。务必只针对你确实想丢弃的具体事件做过滤其余原样返回。2.after_event可以返回列表用一个事件替换出多个事件。每个事件按顺序入队并且链中的下一个拦截器会逐个处理它们。源码 utils.py 中可见若返回None则跳过相当于丢弃若返回列表则展开合并进待处理队列当展开后队列为空整条链提前返回空列表。3. 就地修改事件的影响范围比你想的大。最终 artifact 与生成它的状态消息共享同一份内容所以改写状态更新的文本也会改写那个 artifact。做脱敏时这通常正是你想要的但如果只是想加注解这就是个意外。配置选项A2aAgentExecutor的四个构造参数A2aAgentExecutor本身接受四个参数全部是 keyword-only见 a2a_agent_executor.pyOptionTypeDefaultDescriptionrunnerRunner \| Callable[..., Runner \| Awaitable[Runner]]requiredRunner或可返回 Runner 的工厂。configA2aAgentExecutorConfig \| NoneNone拦截器与转换器。省略时构建默认配置。use_legacyboolFalse无论请求如何强制使用旧版实现。force_new_versionboolFalse无论请求如何强制使用新版实现。关于runner的延迟解析。传入可调用对象callable会把 Runner 的构建推迟到第一个请求到来时这对于Runner 持有你不希望在 import 时就打开的连接池的场景非常重要。同步和异步 callable 都可以源码 a2a_agent_executor.py 显示_resolve_runner会先检查是否已是Runner实例否则调用 callable 并用inspect.isawaitable判断是否需要await解析结果会被缓存供后续请求复用。既不是Runner也不是 callable 的值会在第一个请求时抛出TypeError而非构造时报错。A2aAgentExecutorConfigPydantic 配置模型A2aAgentExecutorConfig是一个 Pydantic 模型持有拦截器以及沿途使用的转换器定义见 config.pyOptionTypeDefaultDescriptionexecute_interceptorslist[ExecuteInterceptor] \| NoneNone上文介绍的三个钩子。a2a_part_convertercallableconvert_a2a_part_to_genai_part把入站 A2A part 转为 GenAI part。gen_ai_part_convertercallableconvert_genai_part_to_a2a_part把出站 GenAI part 转为 A2A part。request_convertercallableconvert_a2a_request_to_agent_run_request把 A2A 请求转为 runner 参数。event_convertercallableconvert_event_to_a2a_events把一个 ADK 事件转为 A2A 事件。旧版实现使用。adk_event_convertercallableconvert_event_to_a2a_events同样的工作供新版实现使用。从源码可以补充几个有价值的细节入站方向convert_a2a_request_to_agent_run_request见 request_converter.py把 A2A 请求转成一个AgentRunRequest模型——它最终以**vars(run_request)的形式展开为runner.run_async(...)的关键字参数。session_id默认取context_iduser_id在 A2A 服务器启用了认证时取call_context.user.user_name否则退化为A2A_USER_{context_id}请求的 A2A 元数据会被包进RunConfig(custom_metadata{a2a_metadata: ...})在 Agent 侧可通过run_config读取。part 转换convert_a2a_part_to_genai_part/convert_genai_part_to_a2a_part见 part_converter.py覆盖文本、文件URI 与内联字节、数据 part 的相互转换其中函数调用、函数响应、代码执行结果等通过带type标签的 A2A data part 携带——这也是 human-in-the-loop 与鉴权请求响应的传输基础。内置 artifact 拦截器executor 目录还内置了一个include_artifacts_in_a2a_event_interceptor见 interceptors/include_artifacts_in_a2a_event.py它通过runner.artifact_service把 ADK 事件中的artifact_delta转为 A2ATaskArtifactUpdateEvent可作为自定义拦截器的参考实现。何时该替换转换器而非加拦截器替换转换器是比重加拦截器重得多的改动因为这意味着你接管了该阶段的整个翻译过程而不是调整它产出的结果。正确策略是先尝试拦截器只有当翻译本身的形态就不适合你时才考虑转换器。高级应用下面两个例子从一次运行的两端使用同样的钩子第一个在 Agent 什么都没做之前行动此时拒绝几乎零成本第二个作用于任务的最后一个事件此时一切结果都已知。在 Agent 运行前拒绝请求解决的问题在付出一次模型调用之前先检查调用方的某些信息——比如 Header、元数据字段或剩余配额。实现方式从before_agent中 raise。async def enforce_quota(context): if _over_quota(context): raise PermissionError(quota exhausted for this caller) return context关键约束before_agent必须返回一个RequestContext这里没有返回None即中止的约定。返回你收到的 context可以修改过是唯一非抛异常的结果路径。在这里 raise 与在后面失败不一样调用方能够分辨出区别从before_agent抛出的异常会到达 A2A 服务器被转换为JSON-RPC 错误{code: -32603, message: quota exhausted for this caller}并且根本不会创建任务在after_event或after_agent中 raise 的钩子给客户端的是一个以failed状态结束的任务携带同样的文本并且是在已经发布的那些事件之后到达的。两条路径都能把你的消息送到调用方但尽早拒绝留下的碎片更少。注解终态事件解决的问题让每个完成的任务都携带客户端可读的 trace id 或成本数字。实现方式在after_agent中向最终事件的元数据添加内容。该事件已经带有adk_前缀的 app name、user id、session id、invocation id、author 和 event id。async def stamp_trace(executor_context, final_event): final_event.metadata[trace_id] _trace_id_for(executor_context.session_id) return final_event对照源码 a2a_agent_executor.py终态事件的元数据正是由 executor 在运行结束后统一组装除了三个会话维度的键之外还会从最后一个 ADK 事件上取invocation_id、author、event_id分别写入adk_invocation_id、adk_author、adk_event_id。你在after_agent里追加的自定义键会与这些键并存客户端可以在终态事件上一次性读取全部上下文。已知局限两套实现由客户端选择。哪个实现处理请求取决于调用方是否请求了https://google.github.io/adk-docs/a2a/a2a-extension/扩展源码中的判定逻辑见 a2a_agent_executor.py 与_NEW_A2A_ADK_INTEGRATION_EXTENSION因此同一个服务器面对两个客户端可能表现出不同行为。需要固定行为时用use_legacy或force_new_version钉死。拦截器不按名字或优先级安装。它们按列表顺序运行after_agent反转且一个拦截器没有办法跳过其余拦截器——除非在after_event中把事件列表清空。after_event永远看不到被转换器丢弃的 ADK 事件。没有内容的 ADK 事件例如工作流节点只发出Event(output...)不会产生任何 A2A 事件也就不会触发任何钩子。当一个运行产生的所有事件都被这样丢弃时就无内容可发布了任务停留在working且没有 artifact客户端会一直等待一个永远不会来的完成事件。被服务的Workflow默认就会踩中这个坑——因为从节点返回值是写工作流的常规方式详见 to_a2a 指南 的对应局限说明至少要让一个节点额外yield Event(message...)携带调用方应接收的文本。钩子内的失败会拖垮整个请求。没有按拦截器隔离的错误机制。after_event或after_agent中的异常会让任务以failed结束而已经发布的事件仍然保持已发布所以客户端可能看到已产出 artifact 的任务随后失败。实验特性。executor 与其 config 都被a2a_experimental装饰见 experimental.py构造时会发出UserWarning。设置环境变量ADK_SUPPRESS_A2A_EXPERIMENTAL_FEATURE_WARNINGS可以静默该警告。注意A2A 协议本身不是实验性的实验性的是 ADK 对它的实现。相关示例与指南A2A root agent 通过to_a2a提供 Agent 服务即用默认配置构建本 executor示例中的客户端形态可参考 agent.py它使用RemoteA2aAgent指向服务端卡片A2A human in the loop 是一个在任务中途暂停的被服务 Agent见 agent.py它会派生出你的钩子会看到的非终态状态更新如input-required类信号to_a2a 是构建本 executor 的函数其中详细记录了把 executor 换成你自定义版本的agent_executor_factory参数AgentCardBuilder 描述客户端在向你发送请求之前就会读取的那张卡片A2A remote agent configuration 介绍同一段对话客户端一侧的镜像拦截器——RemoteA2aAgent的A2aRemoteAgentConfig提供了request_interceptors、card_request_interceptors与入站事件转换器与服务端本 executor 的钩子一一对应。单元测试方面test_a2a_agent_executor.py 覆盖了 executor 的拦截器调用顺序、None丢弃语义、终态事件组装等行为可以作为理解钩子语义的补充参考。【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考