Conductor A2A 端到端测试实战从 MockWebServer 到真实 Agent 的完整验证体系【免费下载链接】conductorConductor is an event driven agentic workflow engine providing durable and highly resilient execution engine for applications and AI Agents项目地址: https://gitcode.com/GitHub_Trending/co/conductorConductor 的 A2AAgent2Agent模块在 ai/src/main/java/org/conductoross/conductor/ai/a2a/ 提供了面向远程 Agent 的 JSON-RPC 2.0 客户端与系统任务实现。本文基于 ai/src/test/resources/a2a/README.md 展开完整梳理其端到端测试体系——从 OkHttpMockWebServer的线协议测试到内置嵌入式 Agent 的真实 HTTP 测试再到针对官方a2a-sdk参考 Agent 的可选集成测试以及真实外部 Agent的 opt-in 验证路径。读完本文你将掌握如何在本仓库中启动一个真实的 A2A Agent、运行跨实现互操作测试、完成全服务器链路工作流 → 引擎 →AGENT系统任务 → 真实 Agent的手动验证并理解AGENT/GET_AGENT_CARD/CANCEL_AGENT三类任务及底层客户端的实现原理。一、A2A 测试体系的整体分层ai/src/test/resources/a2a/README.md将该目录定位为Conductor A2A 客户端针对真实 Agent的测试辅助目录核心实现位于ai/src/main/java/org/conductoross/conductor/ai/a2a/。整个测试体系按下表分层覆盖从纯线协议到真实引擎 真实外部 Agent的完整纵深测试层覆盖内容是否在 CI 中运行A2AServiceTest通过 OkHttpMockWebServer验证 JSON-RPC 客户端线行为是A2AEndToEndTestDiscovery、send、poll-to-completion、流式SSE、cancel——针对一个真实的内嵌 A2A HTTP 服务器EmbeddedA2AAgent通过 loopback 运行是A2ACallbackResourceTest推送通知接收端完成任务的场景引擎 mock针对内嵌 Agent是A2ASdkInteropTestDiscovery send poll streaming message-mode——针对官方a2a-sdk参考 Agent以子进程方式启动找到带a2a-sdk的 Python 时运行设置A2A_PYTHONA2ADurableEngineEndToEndTest位于test-harness模块AGENT任务走真实引擎decider AsyncSystemTaskExecutor Redis验证崩溃/重启恢复是需要 Docker 跑 RedisA2ARealAgentIntegrationTestDiscovery AGENT针对真实外部 Agent仅在设置A2A_AGENT_URL时运行关键设计点在于EmbeddedA2AAgent是一个真正讲 A2A JSON-RPC SSE 的 HTTP 服务器而不是 mock因此 CI 套件已经在验证真实的线协议而最下面的 opt-in 层则把同一个客户端指向第三方 Agent例如基于官方a2a-sdk构建的 Agent专门验证跨实现互操作。对应测试源码集中在 ai/src/test/java/org/conductoross/conductor/ai/a2a/其中A2AEndToEndTest通过 spy 绕过 SSRF 校验后在127.0.0.1上驱动内嵌 Agent 完成发现、发送、轮询、流式与取消的全链路验证见 A2AEndToEndTest.java。二、快速开始启动一个基于官方 a2a-sdk 的真实 Agent仓库在ai/src/test/resources/a2a/下直接提供了一个基于官方 Pythona2a-sdk构建的最小真实 Agent——echo_agent.py。它由A2AStarletteApplication组装出 Agent Card、技能声明与执行器监听在http://localhost:9999。先用uv创建虚拟环境并安装依赖uv venv --python 3.12 uv pip install a2a-sdk0.2,0.3 uvicorn然后以task模式启动默认此时 Agent 会返回一个带 artifact 的TaskAGENT_MODEtask python ai/src/test/resources/a2a/echo_agent.py # serves http://localhost:9999 # AGENT_MODEmessage - 直接返回一条消息message而不是 Task从源码看echo_agent.py两种模式的行为差异清晰AGENT_MODEtask默认通过TaskUpdater先start_work()再添加名为echo的 artifact内容为echo-task: user input最后complete()走标准的任务状态机AGENT_MODEmessage直接通过event_queue入队一条new_agent_text_message(fecho: {user_text})模拟即时回复型 Agent。其 Agent Cardecho_agent.py声明了nameEcho Agent、capabilities.streamingTrue以及一个idecho的技能并注明了defaultInputModes[text]、defaultOutputModes[text]——这些字段正是GET_AGENT_CARD任务向工作流暴露的发现结果。如果你不想用本仓库自带的 echo Agent也可以运行a2aproject/a2a-samples中的任意 Agent例如samples/python/agents/helloworld或 LangGraph 货币兑换 Agent任何符合 A2A 协议的 Agent 都适用。三、可选的集成测试A2ARealAgentIntegrationTest在 Agent 已启动的前提下可以通过环境变量把集成测试指向它A2A_AGENT_URLhttp://localhost:9999 \ A2A_AGENT_PROMPTconvert 100 USD to EUR \ ./gradlew :conductor-ai:test --tests *A2ARealAgentIntegrationTest可选的环境变量A2A_AGENT_TOKEN会作为Authorization: Bearer token请求头发送给 Agent——这正好对应 10-a2a-call-agent.json 工作流中headers.Authorization的用法。该测试在仓库本地已针对echo_agent.pya2a-sdk 0.2.6验证通过discovery 成功解析出 Agent CardAGENT任务返回statecompleted、textecho-task: convert 100 USD。注意该测试是 opt-in 的只有设置了A2A_AGENT_URL才会执行避免 CI 对第三方服务产生外部依赖。四、全服务器链路手动验证workflow → engine → AGENT → 真实 Agent如果只跑上面的集成测试验证的仍是客户端直连 Agent的路径。要打通完整引擎链路工作流 → 引擎 →AGENT系统任务 → 真实 Agent按以下步骤手动验证第 1 步启动 Agent见上文使用echo_agent.py即可。第 2 步启用 AI 集成并启动 Conductor 服务器。需要配置conductor.integrations.ai.enabledtrue——开启 AI 集成A2AService等组件通过AIIntegrationEnabledCondition条件装配见 A2AService.java如果使用 push 模式pushNotification:true还需要conductor.a2a.callback.urlexternally-reachable-base-url让远程 Agent 能把状态更新推回 Conductor 的A2ACallbackResource。第 3 步注册并运行示例工作流。使用 ai/examples/10-a2a-call-agent.json将inputParameters.agentUrl设置为 Agent 的 URL例如http://localhost:9999。该工作流展示了一个典型配置{ name: a2a_agent, version: 1, schemaVersion: 2, tasks: [ { name: call_currency_agent, taskReferenceName: agent, type: AGENT, inputParameters: { agentType: a2a, agentUrl: http://localhost:9999, text: convert 100 USD to EUR, pollIntervalSeconds: 5, headers: { Authorization: Bearer ${workflow.input.agentToken} } } } ] }第 4 步确认结果。工作流完成后AGENT任务的输出应携带远端 Agent 的state、text、artifacts、taskId、contextId等字段。可以使用conductorCLI 或 skill 启动工作流并检查执行详情。五、三类 A2A 任务类型详解README 给出了三类系统任务的语义结合 A2AWorkers.java 的实现可以看得更透任务类型用途底层协议方法AGENT向远端 Agent 发送消息默认轮询也支持流式streaming:true或推送pushNotification:trueconductor.a2a.callback.url以适配长耗时任务message/send、tasks/get、message/streamGET_AGENT_CARD发现 Agent 的技能/能力Agent Card拉取/.well-known/agent-card.jsonv0.3.x或/.well-known/agent.jsonv0.2.5CANCEL_AGENT取消远端正在运行的 A2A 任务tasks/cancel5.1 AGENT三种执行模式A2AWorkers.executeRemote的执行逻辑A2AWorkers.java是首次执行输出中无taskId走startRemote发起message/send并把远端taskId写入任务输出后续执行若已有taskId则走pollRemote轮询tasks/get直到远端状态为终态。这与 Conductor 系统任务的幂等重试模型天然契合。轮询默认通过pollIntervalSeconds控制轮询节奏pollRemote内部有连续失败次数的保护逻辑流式streaming:true走message/stream客户端通过 SSE 逐事件聚合task、status-update、artifact-update直到最终状态见 A2AService.java推送pushNotification:true要求配置conductor.a2a.callback.url由 A2ACallbackResource.java 接收远端推送并完成任务若未配置则回退到轮询代码中会记录 warning见 A2AWorkers.java。5.2 GET_AGENT_CARD能力发现对应示例 ai/examples/11-a2a-get-agent-card.json{ name: a2a_get_agent_card_workflow, version: 1, schemaVersion: 2, tasks: [ { name: discover_agent, taskReferenceName: discover, type: GET_AGENT_CARD, inputParameters: { agentUrl: http://localhost:9999 } } ] }其实现核心在A2AService.getAgentCardA2AService.java如果agentUrl直接指向.json文档则原样使用否则依次尝试标准 well-known 路径/.well-known/agent-card.json→/.well-known/agent.json将首个成功的响应反序列化为AgentCard含name、skills、capabilities、url等。5.3 CANCEL_AGENT任务取消CANCEL_AGENT需要agentUrl与taskId参数缺失时任务会以failtrue结束底层调用tasks/cancelJSON-RPC 方法采用 best-effort 语义。另外当 Conductor 需要传播取消时A2AWorkers.cancelA2AWorkers.java也会读取任务输出中的远端taskId发起取消。六、源码级原理A2AService 客户端与安全边界A2AService是一个手写 JSON-RPC 2.0 客户端基于共享的 OkHttp 客户端 Jackson与 MCP 服务的实现方式一致。它面向 A2A v0.3.x 线模型并通过TaskState容错 v1.0 的枚举拼写差异见 A2AService.java。几个值得注意的实现事实支持的方法message/send、message/stream、tasks/get、tasks/cancel以及 Agent Card 的 HTTP 发现重试语义传输层/5xx/瞬时错误抛出A2AException可重试客户端/协议错误4xx、method-not-found、task-not-found 等含TERMINAL_RPC_CODES列出的 JSON-RPC 错误码抛出NonRetryableException终态与任务重试机制正确配合SSRF 防护validateAgentUrl拒绝非 http(s) 协议、解析到 RFC-1918/loopback/link-local 地址的主机并始终拦截云元数据地址IPv4 的 169.254.0.0/16 与 IPv6 的fd00:ec2::254、fe80::a9fe:a9fe。默认允许私有网络属性conductor.a2a.client.allow-private-networktrue因此本地 loopback 演示可直接运行A2A 客户端还关闭了自动重定向防止 30x 跳转到私有/元数据主机绕过守卫A2AService.java端点解析A2AWorkers.resolveRemoteEndpoint优先使用任务元数据中快照的 Agent Card 所声明的 JSON-RPCurl无快照时若配置为.json卡片 URL 则执行时现场发现直接端点 URL 始终有效A2AWorkers.java。此外Conductor 也可以反过来把工作流暴露为 A2A Agent在 workflow 定义中通过metadata.a2a.enabled: true启用参考 ai/examples/12-a2a-server-workflow.json服务端实现位于ai/src/main/java/org/conductoross/conductor/ai/a2a/server/。七、延伸验证互操作演示与持久化Durable A2Aai/src/test/resources/a2a/下还附带两个可独立运行的演示是 README 测试矩阵之外的实战补充interop-demo把 Conductor 指向一个非 Conductor 的第三方 A2A Agent官方a2a-sdk参考 echo Agent工作流先GET_AGENT_CARD发现、再AGENT调用走真实 JSON-RPC 线协议。运行./run-interop-demo.sh需 Java 21、curl 和uv无需 Docker/Redis/API key预期输出包含discovered agent: Echo Agent、agent state: completed等。其自动化版本正是A2ASdkInteropTest可通过A2A_PYTHONvenv/bin/python ./gradlew :conductor-ai:test --tests *A2ASdkInteropTest运行durable-demo演示杀掉 Conductor 服务器再重启订单照样完成。durable_purchase工作流中的AGENT任务把远端任务 id 持久化在 SQLite重启后系统任务工作器恢复轮询并完成交互。脚本会在订单处于 preparation 期间对服务器执行kill -9再以同一持久化存储重启并等待 COMPLETE。它强调了 A2A 协议本身是无状态请求/响应而持久化是编排器的属性——这正是 Conductor 相比纯内存 A2A 宿主的关键差异。注意该演示通过 loopback 调用 Agent因此服务器需以conductor.a2a.client.allow-private-networktrue运行SSRF 守卫默认拦截私有/loopback 地址仅建议在受信网络中放开。结语从A2AServiceTest的 MockWebServer 线协议验证到A2AEndToEndTest对真实内嵌 Agent 的端到端覆盖再到A2ASdkInteropTest/A2ARealAgentIntegrationTest的跨实现互操作验证Conductor 的 A2A 测试体系证明了其客户端可以直接对接官方a2a-sdk生态下的真实 Agent。配合AGENT/GET_AGENT_CARD/CANCEL_AGENT三类系统任务、echo_agent.py快速启动脚本与全服务器手动验证步骤你可以独立完成从协议层到真实引擎链路的全部 A2A 集成验证并在生产场景中按需选择轮询、SSE 流式或推送三种与远端 Agent 交互的方式。【免费下载链接】conductorConductor is an event driven agentic workflow engine providing durable and highly resilient execution engine for applications and AI Agents项目地址: https://gitcode.com/GitHub_Trending/co/conductor创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考