feat: 支持 a2a 1.0 协议

[bochencwx]

• 同时兼容 A2A 0.3 与 1.0 协议。用户升级仍使用 0.3 的可以不调整，需要使用 1.0 版本的才调整
• 修复 openai 3.x httpx2 流式响应关闭时报错的问题

[helloopenworld]

AI Code Review

很好，方法存在，且 0.3 版本也使用了同样的私有 API。没有新问题。

让我检查最后一件事——_a2a_agent_executor.py 中的 execute 从 call_context.state["headers"] 中提取 OpenTelemetry header。如果 state 为 None 或 headers 不存在，代码中有处理。但 headers = state.get("headers") if isinstance(state, dict) else {} —— 如果 state 是非 dict 的对象，headers 会变为 {}，然后 isinstance(headers, dict) and headers 为 False，因此不会执行 attach。很好。

我现在已经掌握了足够的信息。让我再验证一下 _ensure_card_has_url 的顺序问题：_ensure_card_has_url 在 request_handler 构建之后被调用，但在此之前 _ensure_v0_3_interface(card) 已经被调用（它可能会追加一个 0.3 接口）。_ensure_card_has_url 中的检查使用了 any(i.url for i in card.supported_interfaces) —— 在 _ensure_v0_3_interface 追加了重用空 URL 的 0.3 接口后，如果所有 URL 均为空，此检查仍然返回 False 并抛出异常。很好。

现在让我检查一个潜在的实际问题：当 enable_v0_3_compat=True 且卡片具有 URL 时，_ensure_v0_3_interface 会追加一个 0.3 接口。然后 create_agent_card_routes(card, card_url="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/.well-known/agent.json") 会发布这张修改后的卡片（其中包含一个 0.3 接口）。但顶级的 agent-card.json 路由使用了带有 0.3 接口的同一张卡片 —— 这意味着 1.0 的 agent-card.json 也会包含一个 0.3 接口。这是否会导致 1.0 客户端看到 0.3 接口？可能没问题，因为 1.0 客户端会根据 protocol_version 进行过滤。文档记录了这种行为。这不是明显的 bug。

让我验证一下 _jsonrpc_path_from_card 对 HTTP+JSON 的处理，并确认 JSON-RPC 挂载不会出错。该函数返回第一个 JSONRPC/HTTP+JSON URL 的路径。我认为没有问题。

现在让我看看 _event_converter 中的一个具体正确性问题：_get_event_type 引用了 event.is_streaming_tool_call()。以及 _collect_parts 使用了 Optional[callable]（小写 callable）——这是一个类型注解问题，但 Python 3.12 将 callable 作为类型注解是可以正常工作的，因为它在运行时不会被求值（from future import annotations）。实际上，在运行时对默认值进行 Optional[callable] 注解求值——使用 from __future__ import annotations 时，它不会被求值。没问题。

让我检查一下 _convert_a2a_data_part 中 metadata_type 为 None 的情况——它会返回 genai_types.Part(text=json.dumps(data))。这意味着一个没有已知类型的数据部分会变成一个包含 JSON 字符串的文本部分。这是合理的后备方案。

我已经有足够的内容来撰写审查报告了。让我写下最终结论。让我重新确认一下 initialize 中的事件循环（event loop）问题是否值得标记——它是从 0.3 版本复制过来的，但既然这是一个针对 a2a 1.0 支持的新文件，并且代码是从 ADK 复制/修改而来的，如果存在正在运行的循环（loop being run），那么这个 bug 确实存在，但这是已有的模式。我不标记它。

让我最后检查一下 _remote_a2a_agent 的 initialize 返回 False（失败）的情况，但 _run_async_impl 会检查 self._initialized 并发出错误。这处理得很好。但是 initialize 会捕获所有异常并返回 False，从而隐藏了根本原因——对于诊断来说这并不理想，但会记录错误。没问题。

让我根据新代码验证一个真正的顾虑：TrpcRemoteA2aAgent.initialize 创建了 httpx.AsyncClient(timeout=httpx.Timeout(timeout=None)) —— 没有超时意味着挂起的远程端会永远阻塞。对于取消检查路径，存在 _stream_with_cancel_check，所以流式传输是可以取消的。但卡片发现（_discover_card → resolver.get_agent_card()）没有超时限制 —— 可能会挂起。这是现有的模式（0.3 版本也这样做）。跳过。

现在，让我写下审查报告。最值得关注的是：

1. OpenAI 模型 patch — 非流式响应路径：在 chat completions 的非流式 _generate_async 的 finally 块中，现在调用了 _aclose_openai_stream(response)。但 _patch_stream_response_to_drain_http_body 仅在流式分支中被调用。在非流式路径中，response 是一个普通的 ChatCompletion 对象，而不是带有 aclose 方法的流。_aclose_openai_stream 通过 getattr 安全地检查了 aclose/close 方法，所以这是没问题的。

2. 让我检查 test_patch_closes_connection_before_retrying_hung_drain —— 它修补了 _HTTP_BODY_DRAIN_TIMEOUT_S，但补丁在函数内部将超时时间读取为模块级常量 _HTTP_BODY_DRAIN_TIMEOUT_S。由于它是在 aclose 内部引用的，所以测试期间打补丁的值是生效的。很好。

现在开始写审查报告。

发现的问题

🚨 Critical

• trpc_agent_sdk/server/a2a_v1/executor/_a2a_agent_executor.py:319-335：_handle_request 直接调用 runner._new_invocation_context(...)（私有方法）构建 invocation_context，但随后把该 context 传给 convert_event_to_a2a_events(trpc_event, invocation_context, ...)。convert_event_to_a2a_events 仅用 invocation_context 来读取 app_name/user_id/session_id 等信息构建事件 metadata，而 runner.run_async(**run_args) 内部会自己再创建一个独立的 invocation context 运行 agent。也就是说此处 _new_invocation_context 创建的 context 并未被 agent 实际使用，仅为转换 metadata 而存在。
  • 这本身能工作，但它绕过了 Runner 的正常入口（run_async），手动重建 context，容易在 Runner 内部逻辑（如 cancellation event、session state 绑定）变更时失同步。建议确认 runner.run_async 返回的 event 已携带可用的 invocation_id/branch，并优先复用 event 自身字段而非手动重建 context；若必须重建，应与 0.3 实现保持一致的私有 API 契约并加注释。

实际上这点更像可维护性。让我重新评估等级——不是 Critical，因为没有功能性错误。我把它降为 Warning/Suggestion。

🚨 Critical

• trpc_agent_sdk/server/a2a_v1/_application.py:48-70：create_a2a_application 中 _jsonrpc_path_from_card(card) 用卡片第一个带 url 的 JSONRPC/HTTP+JSON 接口的 path 作为 JSON-RPC 挂载路径。当 enable_v0_3_compat=True 时，_ensure_v0_3_interface 已先追加一个 0.3 JSONRPC 接口（复用已有 url）。若原始 1.0 接口的 url 为空但 0.3 接口继承了非空 url……实际上 0.3 接口 url 来自已有 url，顺序上 1.0 接口在前。
  • 风险点：_ensure_card_has_url(required=True) 在 url 全空时抛错，能兜住。但若 1.0 接口 url 为空、0.3 接口 url 也为空（都空），会先抛错——OK。此处逻辑自洽，不构成 bug。撤回。

让我重新确认一个真正的问题点——在 _jsonrpc_path_from_card 里 advertised = next((i.url for i in card.supported_interfaces if i.protocol_binding in ("JSONRPC","HTTP+JSON") and i.url), None)。当 enable_v0_3_compat 时，card 是含 0.3 接口的副本。第一个匹配是 1.0 接口（在前）。OK，无 bug。

让我聚焦真正可验证的问题。重新审视 _handle_request 末尾的终态处理：

if aggregator.task_state == TaskState.TASK_STATE_WORKING:
    if (aggregator.task_status_message is not None and aggregator.task_status_message.parts):
        ... enqueue final artifact chunk ...
    await event_queue.enqueue_event(create_completed_status_event(...))
    return
await event_queue.enqueue_event(create_final_status_event(state=aggregator.task_state, message=aggregator.task_status_message))

当 agent 正常完成、aggregator 仍为 WORKING（没有收到 failed/completed 状态），且有 message parts 时，会发一个 last_chunk=True 的 artifact + completed status。这里 final_meta: dict[str, Any] = {"partial": False} 只设置了 partial，没有通过 _build_event_metadata 设置 object_type 等。但这是直接构造 TaskArtifactUpdateEvent，metadata 只是 {"partial": False}。这会被远端 _resolve_partial(result.metadata) 读到 → partial=False。OK。

但有一个真实问题：aggregator.task_status_message 来自最后一个 WORKING 状态的 event.status.message。而 convert_event_to_a2a_events 只在 _create_status_update_event 时 _notify(status_event)，status_event 的 message 是转换后的 message。aggregator 记录的 task_status_message 就是这个。最终用它的 parts 作为 final artifact。这看起来合理。

我已经审查得足够深入。让我把真正值得报告的、可从 diff 验证的问题整理出来。重新聚焦：

真正可定位的新问题：

1. _remote_a2a_agent.py 取消时未关闭流生成器 — 但 0.3 同样，不算新引入。不报或作为 Warning 通用项。

2. _openai_model.py _patch_stream_response_to_drain_http_body 中 aclose 与 aiter_raw 并发竞态：held dict 在两个闭包间共享，aclose 设置 held["closing"]=True，而 aiter_raw 的 GeneratorExit 分支也会调 _await_drain_task。若 aclose 与消费者 async for 的 GeneratorExit 同时发生，drain_task 可能被创建两次（aclose 路径和 GeneratorExit 路径都调用 _await_drain_task，后者会复用已存在的 task，OK）。但 _await_drain_task 在 task.done() 时会新建。存在轻微竞态但 _await_drain_task 用 held.get 读取，非原子。这是 httpx2 特定补救，影响有限。降级 Suggestion。

3. _application.py _ensure_v0_3_interface 修改的 card 副本同时用于 1.0 agent-card.json 路由：create_agent_card_routes(card) 和 compat 的 agent.json 都用同一张含 0.3 接口的 card。1.0 客户端拉到的 agent-card.json 会多出一个 0.3 接口。文档已说明，且 1.0 客户端按 protocol_version 过滤，影响低。Suggestion。

4. 真正值得 Warning：TrpcA2aAgentService.initialize 用 loop.run_until_complete 在已有运行 loop 时会抛错——但 0.3 既有，不报。

5. 真正值得 Warning：TrpcRemoteA2aAgent 的 httpx client timeout=None，card discovery 与请求无超时，可能永久挂起——0.3 既有模式。

由于绝大多数问题都是沿用 0.3 既有模式、且本 PR 主要是新增 1.0 适配层（copy & modified from ADK），且测试覆盖较好。真正新引入且可验证的问题较少。让我给一个诚实的结论。

让我最后确认一个我能确定的新问题：_openai_model.py 的 _patch_stream_response_to_drain_http_body 通过 http_response.aiter_raw = aiter_raw 直接替换实例方法。但 http_response 可能是一个 httpx2 的 Response，其 aiter_raw 可能是绑定的方法或属性。赋值 http_response.aiter_raw = aiter_raw（一个 async 函数）会替换为实例属性。openai 调用 response.aiter_raw() 时会拿到这个函数。但 openai 3.x 的 AsyncStream 是否调用 http_response.aiter_raw？注释说它调用 response.aclose()。aiter_raw 的包装是为了在 close 时能 drain。逻辑自洽。

我认为没有 Critical。让我给出最终审查。

发现的问题

⚠️ Warning

• trpc_agent_sdk/server/a2a_v1/_remote_a2a_agent.py:131-162：TrpcRemoteA2aAgent.initialize/_run_async_impl 创建的 httpx.AsyncClient(timeout=httpx.Timeout(timeout=None)) 没有 connect/read 超时，且 _discover_card() 与 send_message 均无超时保护。若远端 A2A 服务无响应，card 发现或流式首帧会永久阻塞；虽然流式部分有 _stream_with_cancel_check 兜底取消，但卡片发现阶段无取消保护。建议显式设置 connect/read 超时（与 0.3 实现共用同一隐患，但本 PR 新增 1.x 路径同样受影响）。

• trpc_agent_sdk/server/a2a_v1/_remote_a2a_agent.py:217-258：_stream_with_cancel_check 在取消分支中直接 raise RunCancelledException，但未对 stream_iter（远端 SSE 生成器）调用 aclose()。取消后远端 HTTP 连接不会被显式关闭，可能在高频取消场景下遗留连接。建议在取消时 await stream_iter.aclose()（0.3 实现同样存在，但本 PR 新增 1.x 路径可一并修复）。

• trpc_agent_sdk/models/_openai_model.py:73-138（_patch_stream_response_to_drain_http_body 中的 aiter_raw/aclose 闭包）：held 字典在 aiter_raw 的 GeneratorExit 路径与 aclose 之间共享，且 aclose 可与消费者 async for 并发触发。_await_drain_task 通过 held.get(...) 读取/写入 drain_task，非原子；两条路径都可能创建/等待 drain task，存在轻微竞态（重复创建 drain task 或 held["iterator"] 被一方清空后另一方仍尝试 drain）。该补救逻辑仅作用于 httpx2，影响面有限，但建议在 aclose 与 GeneratorExit 路径间用 held["closing"]/状态机收敛，避免双重 drain。

💡 Suggestion

• trpc_agent_sdk/server/a2a_v1/_application.py:74-80：enable_v0_3_compat=True 时 _ensure_v0_3_interface 修改的是 card 副本，但该副本同时用于 1.0 的 /.well-known/agent-card.json 路由与 0.3 的 /.well-known/agent.json。1.0 客户端拉到的卡片会多出一个 protocol_version="0.3" 接口。文档已说明行为，且 1.0 客户端按 protocol_version 过滤，影响低；若希望 1.0 卡片保持纯净，可分别为两个路由构造不同副本。

总结

本 PR 新增 a2a-sdk 1.x 适配层（a2a_v1）并修复 OpenAI 流式响应在 httpx2 下的连接关闭问题，整体结构清晰、测试覆盖（含流式关闭、drain、版本检测、取消等）较充分。未发现必须修复的 Critical 问题；Warning 主要集中在远端客户端无超时保护与取消时未关闭生成器（均与 0.3 既有模式一致，但 1.x 新路径同样受影响），以及 httpx2 drain 闭包的轻微竞态。

测试建议

• 建议补充：TrpcRemoteA2aAgent 在远端无响应/卡片发现超时场景下的行为测试（验证不会永久阻塞、initialize 失败可恢复）。
• 建议补充：取消流式请求后远端生成器被关闭/连接释放的覆盖测试。

[codecov]

Codecov Report

❌ Patch coverage is 94.57895% with 103 lines in your changes missing coverage. Please review.
⚠️ Please upload report for BASE (main@06facc9). Learn more about missing BASE report.

Files with missing lines	Patch %	Lines
...t_sdk/server/a2a_v1/converters/_event_converter.py	90.90909%	30 Missing ⚠️
trpc_agent_sdk/server/a2a_v1/_remote_a2a_agent.py	92.01102%	29 Missing ⚠️
trpc_agent_sdk/server/a2a_v1/logs/_log_utils.py	89.04110%	16 Missing ⚠️
..._sdk/server/a2a_v1/executor/_a2a_agent_executor.py	93.10345%	14 Missing ⚠️
...rpc_agent_sdk/server/a2a_v1/_agent_card_builder.py	97.60766%	5 Missing ⚠️
trpc_agent_sdk/server/a2a_v1/_agent_service.py	91.66667%	5 Missing ⚠️
trpc_agent_sdk/server/_a2a_detect.py	88.00000%	3 Missing ⚠️
trpc_agent_sdk/server/a2a_v1/_application.py	97.87234%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main        #307   +/-   ##
==========================================
  Coverage        ?   88.76291%           
==========================================
  Files           ?         496           
  Lines           ?       47539           
  Branches        ?           0           
==========================================
  Hits            ?       42197           
  Misses          ?        5342           
  Partials        ?           0           

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[helloopenworld]

AI Code Review

我已掌握足够信息，可以给出审查结论。

发现的问题

⚠️ Warning

• trpc_agent_sdk/server/a2a_v1/executor/_a2a_agent_executor.py:403：error_code 事件可能产生重复的终态 FAILED 事件

  • 当 runner 产出带 error_code 的 Event（而非抛异常）时，convert_event_to_a2a_events 已入队一个 TASK_STATE_FAILED 状态事件并被 aggregator 记录；run 正常结束后 aggregator.task_state == FAILED，执行器又会再入队一个 create_final_status_event(state=FAILED, ...)，即向同一 task 流发出两次终态 failed。当前测试只覆盖了抛 RuntimeError/RunLimitException 的异常路径（test_execution_error_*、test_run_limit_error_*），未覆盖“事件携带 error_code”这条路径。建议补一条用例驱动该路径，并确认 DefaultRequestHandler 对重复终态的处理；若 SDK 不去重，应在 convert_event_to_a2a_events 已发出 failed 后跳过末尾的 create_final_status_event。

• trpc_agent_sdk/server/a2a_v1/executor/_a2a_agent_executor.py:155：cancel 的 task metadata 提取路径实为死代码

  • 注释声称从“execute() 写入的 task metadata”提取 app_name/user_id/session_id，但 execute() 入队的初始 Task（:230）未设置 metadata，working_meta 只写到 TaskStatusUpdateEvent.metadata（:299 附近），不会落到 context.current_task.metadata。因此 _get_user_session_from_task_metadata 恒返回 (None,None,None)，cancel 总是走 fallback。功能上 fallback 的 user_id 解析与 run 期一致所以仍能取消成功，但该“主路径”永不命中，注释具有误导性。建议要么真正把 working_meta 写到 Task metadata，要么删除该分支与对应测试，避免误维护。

• pipeline_test/run_all_examples.sh:75：transfer_agent 示例在 CI 中完全失考

  • 本 PR 把 examples/transfer_agent/run_agent.py（0.3）加入 SKIPPED_EXAMPLES，而 run_agent_v1.py 不匹配 discover_run_agent_examples 的 */run_agent.py glob，不会被自动发现；run_a2a_example 也只跑 examples/a2a_v1。结果 transfer_agent 的 1.x 链路（含内嵌 A2A 服务拉起）无任何 CI 覆盖。建议为 run_agent_v1.py 显式增加一个运行步骤，或调整发现逻辑纳入 run_agent_v1.py。

💡 Suggestion

• trpc_agent_sdk/server/a2a_v1/_remote_a2a_agent.py:430：_events_from_response 处理独立 Message 时调用 convert_a2a_message_to_event(result, ...) 未传 partial，默认 False；而 server 侧通过 artifact 流式分片输出的内容到达 client 时若以 bare Message 形式出现会被当作非 partial 最终消息。如该分支确有可能承载流式分片，应按 metadata 中的 partial/streaming_tool_call 标志解析，与 artifact 路径保持一致。

总结

整体为新增 a2a-sdk 1.x 适配层（a2a_v1）并修复 httpx2 SSE 关闭问题，核心逻辑有较完整单测覆盖，无明显安全或阻断性缺陷。主要风险集中在 error_code 事件路径可能产生重复终态事件（缺测试覆盖）、cancel 的 metadata 提取为死代码，以及 transfer_agent 1.x 示例脱离 CI 覆盖，均建议修复或补测但非必须阻塞。

测试建议

• 补一条 executor 用例：runner 产出带 error_code 的 Event 后正常结束，断言流中 failed 终态事件只出现一次（或验证 SDK 对重复终态的去重行为）。
• 补 cancel 用例：模拟生产流程（execute 正常入队 Task/working status，current_task.metadata 为空），验证 cancel 经 fallback 路径能正确定位并取消 run（user_id 与 run 期一致）。

[helloopenworld]

AI Code Review

重要提示：在 0.3 版本中，错误 Event 使用了 error_message=... 但没有设置 error_code。在 v1 版本中也是一样（error_message=error_message）。两者都使用了 error_message 而没有 error_code。一致。该错误 Event 带有 error_message，框架将其视为错误。由于行为一致，这不是一个新 Bug。

我现在完成审查。让我验证最后一个具体的顾虑：在 v1 的 _run_async_impl 中，错误 Event 的 error_message=error_message 被抛出（yield）了 —— 但它没有 content。在父级 agent 中，带有 error_message 的 Event 是错误信号。没问题。

让我再验证一个潜在的实际 Bug：在 _events_from_response 中，针对 TaskStatusUpdateEvent，当 state 不在允许的集合中时（包括 TASK_STATE_CANCELED、TASK_STATE_FAILED 等），它会进行转换。但当状态为 TASK_STATE_COMPLETED 时，它被排除了（不在允许列表中）。因此，完成时的状态更新 Event 会被丢弃，且 msg 不会作为 Event 抛出。这是合理的，因为 artifact 事件携带了内容。与 0.3 版本一致。没问题。

我已经完成了审查。让我来写结论。

发现的问题

🚨 严重

• trpc_agent_sdk/server/a2a_v1/_application.py:113-132：create_a2a_application 在 enable_v0_3_compat=True 时会调用 create_jsonrpc_routes(request_handler, rpc_url=..., enable_v0_3_compat=...)，并调用 create_agent_card_routes(card, card_url="...")。这两个 a2a-sdk 1.x 路由工厂的参数签名（enable_v0_3_compat 关键字参数、card_url 关键字参数）无法从 diff 本身验证；若实际安装的 a2a-sdk 1.x 不接受这些参数，开启 compat 的服务端启动会直接 TypeError。
  • 影响：v0.3 兼容模式（文档与示例主推的迁移路径）无法启动。默认 enable_v0_3_compat=False 路径不触发，故非阻塞但属高风险。
  • 修复方向：确认 a2a-sdk 1.x 的 create_jsonrpc_routes / create_agent_card_routes 确实接受这些关键字参数；tests/server/a2a_v1/test_application.py 中的 test_compat_serves_legacy_agent_json 等用例已通过 TestClient 实际装配路由，CI（a2a-sdk>=1.0.0）跑通即可验证，务必确保该用例在 CI 中执行而非被跳过。

⚠️ 警告

• trpc_agent_sdk/server/a2a_v1/_remote_a2a_agent.py:150,168-173,210：TrpcRemoteA2aAgent 自建了 httpx.AsyncClient（以及由它派生的 a2a_client），但类上没有任何 close/aclose 方法，BaseAgent 也没有统一的关闭生命周期。示例（examples/a2a_v1/test_a2a.py、transfer_agent/run_agent_v1.py）在进程结束前不会关闭该 client，长驻进程里会泄漏连接池。

  • 影响：常驻服务中使用 TrpcRemoteA2aAgent 会逐步累积未关闭的 httpx 连接。
  • 修复方向：增加 async def close()，关闭 _a2a_client（若有 close）与 _httpx_client；或在文档中明确由调用方注入并管理 a2a_client/httpx_client。（注：0.3 适配层同样缺该方法，属历史遗留，但本 PR 新增 v1 时建议一并补齐。）

• trpc_agent_sdk/server/a2a_v1/_agent_service.py:86-91：TrpcA2aAgentService.initialize() 用 asyncio.get_running_loop() + loop.run_until_complete(...)；若调用方处于已有运行中的事件循环（例如在 FastAPI/Starlette 的 async startup 里调用），run_until_complete 会抛 RuntimeError: This event loop is already running。

  • 影响：在异步服务启动钩子中初始化会失败；同步 __main__ 调用不受影响（示例均为此路径）。
  • 修复方向：文档中明确"须在进入 async 事件循环前调用 initialize()"，或提供 await svc.initialize_async() 的异步入口供 async 集成使用。

• trpc_agent_sdk/server/a2a_v1/executor/_a2a_agent_executor.py:155-167：cancel() 依赖 a2a-sdk TaskManager 把 working 状态事件的 metadata（app_name/user_id/session_id）合并到持久化 Task 上，再在取消时读回。该合并行为（task.metadata.MergeFrom(event.metadata)）属于 a2a-sdk 内部实现，无法从 diff 验证。

  • 影响：若 1.x 的 DefaultRequestHandler 未做此合并，_get_user_session_from_task_metadata 会读到 None，退回到 get_user_session_id 兜底（用 context_id 当 session_id、A2A_USER_{context_id} 当 user_id），与执行时的真实 user_id/session_id 不一致，导致 cancel_run_async 找不到正在运行的 run，取消静默失效。
  • 修复方向：补一个集成测试覆盖"execute 发起后 cancel 真正中断在跑的 run"（示例 test_a2a_cancel.py 是手动脚本，未进 CI 单测）；若 SDK 不合并，则改由执行器自身维护 task_id -> (app_name,user_id,session_id) 映射。

💡 建议

• trpc_agent_sdk/server/a2a_v1/_agent_card_builder.py:185-187：_build_tool_skills 调用 convert_toolunion_to_tool_list(agent.tools, None) 传入 invocation_context=None，随后对每个 BaseToolSet 调用 toolset.close()。若 agent 注册了自定义 ToolPredicate（会收到 None 并访问其属性），构建卡片时会抛异常并被上层 except 吞掉，导致工具 skill 静默缺失。建议在文档/注释中说明"卡片构建期间 ToolPredicate 会被以 None context 调用"，或对 None context 做防御性短路。

总结

本 PR 主体是为 a2a-sdk 1.x 新增并列的 a2a_v1 适配层（服务端/客户端/转换器/示例/测试），整体结构清晰、测试覆盖较充分，核心非兼容路径风险可控。主要风险集中在两处对 a2a-sdk 1.x 内部行为的隐式依赖：create_jsonrpc_routes(enable_v0_3_compat=...) 的参数签名（严重，但可被 CI 用例验证），以及取消流程依赖 TaskManager 合并 metadata（警告，建议补集成测试）。无明确的必须修复逻辑错误。

测试建议

• 补一个"execute 发起运行 → 通过 A2A tasks/cancel 取消 → 断言远端 run 真正被中断"的集成测试（当前 test_a2a_cancel.py 是手动脚本，未进 pytest），覆盖 _a2a_agent_executor.cancel 读取 task metadata 的路径。
• 确保 tests/server/a2a_v1/test_application.py 中 test_compat_* 与 test_jsonrpc_mount_* 这类实际装配 Starlette 路由的用例在 CI 默认环境（a2a-sdk 1.x）下确实执行，以验证 enable_v0_3_compat 与 card_url 等关键字参数被 a2a-sdk 接受。

[helloopenworld]

v0.3 兼容模式启动可能 TypeError

create_a2a_application 在 enable_v0_3_compat=True 时向 a2a-sdk 1.x 的 create_jsonrpc_routes 传入 enable_v0_3_compat、向 create_agent_card_routes 传入 card_url 关键字参数。若实际安装的 a2a-sdk 1.x 不接受这些参数，开启 compat 的服务端启动会直接 TypeError。建议确认签名并确保 CI 中 test_compat_* 用例实际执行。

[helloopenworld]

AI Code Review

现在我已经获取了行号。让我来编写最终的审查结论。

发现的问题

⚠️ Warning

• trpc_agent_sdk/server/a2a_v1/_remote_a2a_agent.py:180-182：initialize() 失败时 httpx 客户端资源泄漏

  • _httpx_client 在第 170 行创建后，若 _discover_card() / _build_a2a_client() 抛异常，except 分支仅记录日志并 return False，未调用 self._httpx_client.aclose()。对不稳定远端重试 initialize 时会累积未关闭的 AsyncClient/连接池。建议在 except 分支关闭已创建的 httpx client（或用 try/finally 保证清理）。

  except Exception as ex:  # pylint: disable=broad-except
      logger.error("Failed to initialize remote A2A agent %s: %s", self.name, ex)
      return False

  • test_failure_returns_false 仅在断言后由测试侧手动 aclose，未覆盖生产代码路径的清理，泄漏路径无回归保护。

• trpc_agent_sdk/server/a2a_v1/_remote_a2a_agent.py:387-401：_stream_with_cancel_check 在 cancel 与下一帧同时就绪时丢弃已到达响应

  • asyncio.wait 返回后先判 cancel_wait_task in done 并直接 raise RunCancelledException，即使 next_response_task 也已完成也不消费其结果。这是预期的优先级，但被取消的 next_response_task 的结果在 finally 中仅 aclose 流迭代器，未读取/取消该 task 本身（pending 里的才被 cancel+await）。当两者都在 done 中时 next_response_task 既未 .result() 也未 .cancel()，可能产生“task was destroyed but it is pending”告警。建议在 cancel 分支也显式取消并吸收 next_response_task。

💡 Suggestion

• trpc_agent_sdk/models/_openai_model.py:178-182：_close_http_body follower 路径会重复调用 orig_aclose
  • 即使 leader 已将 held["tcp_closed"] 置 True，follower 仍无条件再次 await orig_aclose(...)，可能导致底层 httpx2 aclose 被多次调用。建议 follower 在 always_tcp_close 时先检查 held["tcp_closed"]，已关闭则跳过。
• .coveragerc:1-3 与 requirements-test.txt：CI 全量切换到 a2a-sdk 1.x 后，0.3 适配层（trpc_agent_sdk/server/a2a/*）在 tests/server/a2a/conftest.py 中被 pytest_ignore_collect 跳过，且 .coveragerc 将其整体 omit，0.3 适配层实质上不再有任何测试或覆盖率覆盖。本 PR 未改动 0.3 适配层主体，风险可控，但后续若改动 0.3 适配层将缺乏 CI 保护，建议在文档/CI 中明确这一状态。

总结

整体改动（新增 a2a-sdk 1.x 并行适配包 a2a_v1、版本探测守卫、OpenAI 流式 HTTP body drain 修复、CI 切换到 1.x 栈）结构清晰、测试覆盖较充分，未发现阻塞级别的安全或逻辑错误。主要风险集中在远端 agent 初始化失败时的 httpx 资源泄漏与流取消时未消费 task 的边缘处理，建议修复后合入。

测试建议

• 补充 initialize() 失败路径的断言：失败后 agent._httpx_client 应为已关闭/为 None，避免资源泄漏回归。
• 补充 _stream_with_cancel_check 在“cancel 与下一帧同时就绪”场景下的用例，断言无 pending task 泄漏告警。

[helloopenworld]

AI Code Review

I have enough to finalize. Let me write the review.

发现的问题

⚠️ Warning

• trpc_agent_sdk/server/a2a_v1/executor/_a2a_agent_executor.py:317-328：agent_run_config 浅合并会整体覆盖服务端配置的 metadata

  • 当 run_config/run_config_factory 提供了 agent_run_config（例如含服务端约定的 metadata、configuration 等键）时，合并用的是字典级浅合并 {**configured, **request}；请求侧的 agent_run_config 仅含 {"metadata": request_metadata}，会直接整键覆盖服务端的 metadata，导致服务端在该键下注入的字段丢失。docstring 称“请求 metadata 在 agent_run_config 中被保留”，但对 metadata 子键并非深度合并。建议对 metadata 子键做嵌套合并（{**configured_meta, **request_meta}），以保留服务端约定字段。

  • ...
    "agent_run_config": {
        **configured_agent_run_config,
        **request_agent_run_config,  # 请求 metadata 整键覆盖服务端 metadata
    },
    ...

• trpc_agent_sdk/server/a2a_v1/_remote_a2a_agent.py:524-535：_resolve_partial 对非 bool/非 str 值默认返回 True

  • 当远端事件 metadata 的 partial 为非布尔、非字符串值（如 protobuf Struct 中以 number value 表示的 0，经 MessageToDict 回到 Python 后是 int）时，函数既非 bool 也非 str，直接 return True，会把一个明确为“非流式”的事件误判为 partial=True。建议对数字值（0/1）一并处理，或当值确定存在但无法识别为 true 时回退到 False 而非 True。

• trpc_agent_sdk/server/a2a_v1/_remote_a2a_agent.py:353-364：run_config.metadata/configuration 透传未覆盖测试

  • _run_async_impl 在 ctx.run_config 存在时取出 agent_run_config["metadata"]/["configuration"] 并直接传入 SendMessageRequest(..., metadata=..., configuration=...)，但该路径在 test_remote_a2a_agent.py 中所有用例均以 run_config=None 运行，未验证 1.x proto 对 plain dict 的构造是否成立（SendMessageRequest.metadata 在 1.x 是 Struct）。建议补一个带 run_config.agent_run_config={"metadata": {...}, "configuration": {...}} 的用例，确认请求构造不抛异常且字段被正确序列化。

💡 Suggestion

• trpc_agent_sdk/server/a2a_v1/executor/_a2a_agent_executor.py:72-80、trpc_agent_sdk/server/a2a_v1/converters/_event_converter.py:79-85、trpc_agent_sdk/server/a2a_v1/logs/_log_utils.py:62-68：_metadata_to_dict 在三处重复定义
  • 同一 Struct/dict 归一化逻辑重复实现，后续若 1.x metadata 处理方式调整需多处同步。可抽到 _utils.py 单一实现并复用，降低维护成本。

总结

整体风险较低：a2a-sdk 1.x 适配、OpenAI httpx2 流体 drain 补丁及互斥测试隔离设计均较完整、覆盖了主要异常与并发路径。未发现必须修复的安全或核心功能缺陷；主要遗留点是 agent_run_config 浅合并覆盖服务端 metadata、_resolve_partial 对非标量值的默认值，以及远程 agent 的 run_config 透传缺测试，建议合并前处理。

测试建议

• 补一个 _handle_request 在 run_config/run_config_factory 提供 agent_run_config.metadata、且请求侧也带 metadata 的用例，断言服务端 metadata 子键未被整键覆盖。
• 补一个 TrpcRemoteA2aAgent._run_async_impl 在 ctx.run_config.agent_run_config 含 metadata/configuration 时 SendMessageRequest 构造成功的用例。

[helloopenworld]

AI Code Review

我已经了解了相关情况。现在我还需要验证另一个潜在问题：_run_async_impl 中的 RunCancelledException 捕获逻辑。当收到取消指令时，代码会尝试调用 cancel_task。但 _a2a_client 可能是 None（如果未进行初始化）——但该路径上方有一个 if not self._initialized 的提前返回逻辑。很好。另外，metadata/configuration 是从 ctx.run_config.agent_run_config 中提取的。如果 ctx.run_config 为 None，则它们保持为 None。没问题。

让我完成审查。

发现的问题

⚠️ Warning

• trpc_agent_sdk/server/a2a_v1/_remote_a2a_agent.py:187：自有的 httpx 客户端在成功路径下未关闭（资源泄漏）

  • TrpcRemoteA2aAgent 在 initialize() 时创建了自有的 httpx.AsyncClient（第 170 行），但 _aclose_owned_httpx_client 仅在初始化失败分支（第 185 行）调用；类上没有 close() 方法，Runner.close() 只清理 BaseToolSet，远程 agent 不是 toolset，因此成功初始化后该客户端从不被关闭，长时间运行进程会泄漏连接。建议为 TrpcRemoteA2aAgent 增加 async def close()，关闭自有的 httpx 客户端与 a2a client。

• trpc_agent_sdk/models/_openai_model.py:196：SSE drain 失败后无超时等待可能挂起连接关闭

  • _close_http_body 在 drain 超时（_HTTP_BODY_DRAIN_TIMEOUT_S=2.0）后先 orig_aclose 关 TCP，再 await _await_drain_task(held, timeout=None) 无超时地等待 drain task 结束。若对端在 TCP 关闭后仍不触发 EOF（半开连接/异常 peer），该 await 将无限挂起，阻塞 openai 流的 finally 与后续 close_http_client。建议给该兜底等待设一个上限，超时则放弃 drain task 并记录告警。

💡 Suggestion

• trpc_agent_sdk/server/a2a_v1/executor/_a2a_agent_executor.py:261：execute() 中的 RunLimitException 分支不可达
  • _handle_request 内部已捕获 RunLimitException（第 388 行）并 return，因此 execute() 第 261 行的 except RunLimitException 永远收不到该异常，属死代码。为避免误读，建议删除 execute() 中的该分支，或将其与 _handle_request 的处理统一收敛到一处。

总结

整体为新增 a2a-sdk 1.x 适配层，结构清晰、测试覆盖较全，未发现阻断性正确性或安全问题。主要风险为远程 agent 自有 httpx 客户端在成功路径下的资源泄漏，以及 OpenAI SSE drain 兜底路径的无超时等待可能挂起；均建议修复但不强制阻断。

测试建议

• 暂无额外测试建议；现有 test_remote_a2a_agent.py 已覆盖初始化与流式取消路径。如采纳上述修复，建议补充一条“成功初始化后调用 close() 能关闭 httpx 客户端、重复调用幂等”的用例。