EMQX A2A Registry 的 Agent Card 存活状态属性覆写机制解析fix-17010【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址: https://gitcode.com/gh_mirrors/em/emqx导读本篇文章围绕 EMQX 开源仓库中的变更记录 changes/ee/fix-17010.en.md 展开深入讲解 A2AAgent-to-AgentAgent Card 中a2a-status与a2a-status-source两个用户属性User Property的覆写机制当 Agent Card 自身携带这两个属性时EMQX 会用自己维护的 Agent 存活状态liveness信息将其覆盖以避免同一 Card 中出现重复或冲突的状态属性。读完本文你将理解该机制的设计动机、源码级实现链路包括 hook 注册、状态查询、属性删除与重写以及如何通过测试用例、CLI 和 REST API 验证其行为。变更背景为什么需要覆写 Agent Card 中的状态属性在 A2A 生态中Agent Card 是描述一个 Agent 能力、接口、技能等信息的标准化文档其 JSON Schema 见 apps/emqx_a2a_registry/priv/agent_card_schema.json。EMQX 通过 A2A Registry 模块apps/emqx_a2a_registry将这些 Card 以**保留消息retained message**的形式发布到 discovery 主题供其他 Agent 或客户端订阅发现。Agent Card 的发布者Agent 本体在发布时可能自行携带一组 MQTT 用户属性例如a2a-status声明 Agent 当前状态a2a-status-source声明该状态的来源例如agent表示由 Agent 自己上报。而 EMQX 的 A2A Registry 本身也需要在每条 Card 消息上附加存活状态信息用于向订阅者实时通告该 Agent 是否在线。这就会产生一个冲突如果 Agent 自己上报的属性与 EMQX 计算出的存活状态同时存在订阅方会看到两份语义可能不一致的a2a-status造成歧义。正是为了解决这一问题fix-17010 引入了覆写逻辑。变更记录原文如下Now,a2a-statusanda2a-status-sourceuser properties present in A2A Agent Cards are overridden with EMQXs liveness information to avoid duplicate properties.该变更同样被收录进版本更新说明 changes/6.2.1.en.md第 226 行对应 PR #17010属于 6.2.1 版本的行为变更。覆写规则删除旧值、写入 EMQX 权威状态覆写逻辑的核心是先删除、再重写而不是简单拼接。从源码实现来看整个处理发生在message.delivered钩子回调中位于 apps/emqx_a2a_registry/src/emqx_a2a_registry_hookcb.erl 的augment_message_metadata/1函数第 135–148 行augment_message_metadata(Msg0) - ClientId emqx_message:from(Msg0), Props0 emqx_message:get_header(properties, Msg0, #{}), UserProperties0 maps:get(User-Property, Props0, []), UserProperties1 proplists:delete(?A2A_PROP_STATUS_KEY, UserProperties0), UserProperties2 proplists:delete(?A2A_PROP_STATUS_SOURCE_KEY, UserProperties1), Status emqx_a2a_registry:lookup_agent_status(ClientId), UserProperties [ {?A2A_PROP_STATUS_KEY, Status}, {?A2A_PROP_STATUS_SOURCE_KEY, emqx} | UserProperties2 ], Props maps:put(User-Property, UserProperties, Props0), emqx_message:set_header(properties, Props, Msg0).处理流程可拆解为四步读取原始用户属性从消息头的User-Property中取出属性列表删除冲突键分别用proplists:delete移除原消息中携带的a2a-status与a2a-status-source确保不会出现重复属性查询权威状态调用emqx_a2a_registry:lookup_agent_status(ClientId)获取该 Agent 的实时存活状态前置写入新值将{a2a-status, Status}与{a2a-status-source, emqx}放在属性列表头部作为第一个出现值MQTT 协议中同名 User Property 可重复订阅方一般取首个完成覆写。属性名与取值的常量定义上述代码中的?A2A_PROP_STATUS_KEY等宏定义在 apps/emqx_a2a_registry/src/emqx_a2a_registry_internal.hrl宏值语义?A2A_PROP_STATUS_KEYa2a-status状态属性键名?A2A_PROP_STATUS_SOURCE_KEYa2a-status-source状态来源属性键名?A2A_PROP_ONLINE_VALonline在线状态值?A2A_PROP_OFFLINE_VALoffline离线状态值可以看到覆写后a2a-status-source被固定为emqx明确告知订阅方该状态由 EMQX Broker 权威计算得出而非 Agent 自报。存活状态如何计算lookup_agent_status/1定义在 apps/emqx_a2a_registry/src/emqx_a2a_registry.erl第 109–116 行lookup_agent_status(ClientId) - FormatFn undefined, case emqx_cm:lookup_client_and_format({clientid, ClientId}, FormatFn) of [{_Chan, #{conn_state : connected} _Info, _Stats}] - ?A2A_PROP_ONLINE_VAL; _ - ?A2A_PROP_OFFLINE_VAL end.它通过emqx_cm:lookup_client_and_format查询连接管理表中的客户端会话只要该 ClientId 存在且conn_state为connected即判定为online否则一律返回offline。也就是说状态是 Broker 在消息投递时刻实时计算的而非依赖 Agent 在 Card 中自报的值——这正是避免重复属性、以 EMQX 为准的意义所在。触发时机message.delivered 钩子与作用范围覆写逻辑并非对任意消息生效而是有严格的作用范围与触发条件由 apps/emqx_a2a_registry/src/emqx_a2a_registry_hookcb.erl 的钩子机制控制register_hooks() - ok emqx_hooks:add(message.publish, ?MESSAGE_PUBLISH_HOOK, ?HP_RETAINER 1), ok emqx_hooks:add(message.delivered, ?MESSAGE_DELIVERED_HOOK, ?HP_RETAINER - 1), ok.message.publish钩子负责在消息发布入站时校验Card 消息——只有保留消息retain标志为真且主题匹配 A2A discovery 主题才会放行非保留消息会被拒绝设置allow_publish false非 discovery 主题的消息则原样放行。同时还会校验发布者 ClientId 必须与主题中的org_id/unit_id/agent_id三段一致并校验 Card JSON 符合 Schemavalidate_card_schema/1。message.delivered钩子在消息投递给订阅者时触发负责给 discovery 主题的消息附加/覆写存活状态属性即本次变更的核心逻辑其他主题消息不做任何处理。两个钩子的执行优先级分别锚定在 Retainer 前后?HP_RETAINER ± 1确保与保留消息的存储、读取时序正确配合。也就是说覆写发生在从保留消息存储中取出、投递给订阅者的路径上因此订阅者在任何时刻收到的最新 Card 消息都携带 Broker 计算的实时状态。触发前提功能开关与保留消息覆写逻辑受开关控制两个钩子的回调入口都会先检查emqx_a2a_registry_config:is_enabled()未启用时直接放行原消息。开关定义在 apps/emqx_a2a_registry/src/emqx_a2a_registry_schema.erla2a_registry { enable false # 总开关默认关闭 validate_schema true # 是否校验 Agent Card 符合 JSON Schema默认开启 }另外由于 Agent Card 依赖保留消息存储emqx_retainer该功能还要求 Retainer 已启用retainer.enable true否则 CLI 与 API 会返回retainer_disabled错误提示。作用范围仅限 A2A discovery 主题do_on_message_delivered会先调用maybe_augment_message_metadata解析消息主题只有能成功解析为 A2A discovery 主题格式为$a2a/v1/discovery/{org_id}/{unit_id}/{agent_id}支持命名空间前缀的消息才会被覆写属性其他主题的消息原样透传不受影响。discovery 主题的拼接逻辑见 apps/emqx_a2a_registry/src/emqx_a2a_registry.erl 的discovery_topic/4第 45–63 行主题命名空间常量定义于 apps/emqx_a2a_registry/src/emqx_a2a_registry_internal.hrl$a2a/v1/discovery。测试验证属性被覆写的完整行为样例仓库的集成测试对本次覆写行为有直接且完整的覆盖见 apps/emqx_a2a_registry/test/emqx_a2a_registry_SUITE.erl第 268–289 行。测试构造了一个故意携带冲突属性的发布%% If a2a-status and/or a2a-source keys is present in user properties, they are %% overridden. {ok, _} emqtt:publish( Agent2, discovery_topic(?ORG_ID, ?UNIT_ID, ?AGENT_ID), #{ User-Property [ {?A2A_PROP_STATUS_KEY, whatever}, {?A2A_PROP_STATUS_SOURCE_KEY, agent} ] }, sample_card_bin(), [{qos, 1}, {retain, true}] ), {publish, #{properties : #{User-Property : Props3}}} ?assertReceive({publish, _}), ?assertMatch( #{ ?A2A_PROP_STATUS_KEY : [?A2A_PROP_ONLINE_VAL], ?A2A_PROP_STATUS_SOURCE_KEY : [emqx] }, get_a2a_props(Props3) ),测试要点Agent 携带a2a-status whatever、a2a-status-source agent发布 CardQoS 1 retain订阅方收到的消息中这两个属性已被覆写为a2a-status online该测试中 Agent 处于在线状态与a2a-status-source emqx同一测试用例第 233–247 行还验证了 Agent 断开后重新订阅会收到a2a-status offline证明状态随 Broker 侧连接状态实时变化。此外测试还覆盖了非保留 Card 消息被拒绝、非 discovery 主题消息正常透传、功能关闭后客户端可任意发布等边界行为可作为理解整套 A2A Registry 消息处理语义的参考。如何观察与使用覆写后的状态覆写发生在message.delivered路径因此订阅 discovery 主题的任何客户端都能直接观察到最终属性。此外A2A Registry 提供的查询入口也统一使用同一套状态语义CLIapps/emqx_a2a_registry/src/emqx_a2a_registry_cli.erla2a_registry list [--status online|offline]、a2a_registry get ORG_ID UNIT_ID AGENT_ID、a2a_registry delete ...、a2a_registry register ORG_ID UNIT_ID AGENT_ID agent-card.json、a2a_registry stats。其中list的--status参数只接受online/offline两个取值第 331–338 行校验与属性值常量保持一致REST APIapps/emqx_a2a_registry/src/emqx_a2a_registry_api.erlCard 输出结构card_out中的status字段同样是online/offline枚举第 157 行该值由 apps/emqx_a2a_registry/src/emqx_a2a_registry.erl 的format_card/4第 179–207 行在读取保留消息时通过lookup_agent_status实时计算与投递路径上的覆写逻辑保持一致。也就是说无论你是通过 MQTT 订阅 discovery 主题、通过emqx ctl a2a_registry命令还是通过 REST API 查询看到的a2a-status/status都是同一份EMQX 视角的存活状态不会再出现 Agent 自报状态与 Broker 实际状态不一致导致的重复或矛盾。总结fix-17010 通过message.delivered钩子对 A2A discovery 主题消息执行删除旧值 前置写入新值的覆写策略将a2a-status与a2a-status-source统一为 EMQX 基于连接管理表实时计算的存活状态online/offline来源固定为emqx。这一机制既保证了订阅者始终拿到权威、一致的 Agent 状态又避免了属性重复带来的歧义是整个 A2A Registry 发现链路上关键的一环。核心实现位于 apps/emqx_a2a_registry/src/emqx_a2a_registry_hookcb.erl 与 apps/emqx_a2a_registry/src/emqx_a2a_registry.erl行为契约由 apps/emqx_a2a_registry/test/emqx_a2a_registry_SUITE.erl 中的集成测试锁定可作为后续二次开发或排障时的直接参考。【免费下载链接】emqxThe most scalable and reliable MQTT broker for AI, IoT, IIoT and connected vehicles项目地址: https://gitcode.com/gh_mirrors/em/emqx创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考