企业官网建设流程全解析

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/conductor

Conductor 的 A2A（Agent2Agent）模块在 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"的完整纵深：

关键设计点在于：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-sdk>=0.2,<0.3" uvicorn

然后以task模式启动（默认），此时 Agent 会返回一个带 artifact 的Task：

AGENT_MODE=task python ai/src/test/resources/a2a/echo_agent.py # serves http://localhost:9999 # AGENT_MODE=message -> 直接返回一条消息（message），而不是 Task

从源码看（echo_agent.py），两种模式的行为差异清晰：

• AGENT_MODE=task（默认）：通过TaskUpdater先start_work()，再添加名为echo的 artifact（内容为echo-task: <user input>），最后complete()，走标准的任务状态机；
• AGENT_MODE=message：直接通过event_queue入队一条new_agent_text_message(f"echo: {user_text}")，模拟"即时回复"型 Agent。

其 Agent Card（echo_agent.py）声明了name=Echo Agent、capabilities.streaming=True以及一个id=echo的技能，并注明了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_URL=http://localhost:9999 \ A2A_AGENT_PROMPT="convert 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.py（a2a-sdk 0.2.6）验证通过：discovery 成功解析出 Agent Card，AGENT任务返回state=completed、text=echo-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.enabled=true——开启 AI 集成（A2AService等组件通过AIIntegrationEnabledCondition条件装配，见 A2AService.java）；
• 如果使用 push 模式（pushNotification:true），还需要conductor.a2a.callback.url=<externally-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 的实现可以看得更透：

5.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.getAgentCard（A2AService.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参数（缺失时任务会以fail=true结束），底层调用tasks/cancel（JSON-RPC 方法），采用 best-effort 语义。另外，当 Conductor 需要传播取消时，A2AWorkers.cancel（A2AWorkers.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-network=true），因此本地 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 A2A）

ai/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_PYTHON=<venv>/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-network=true运行（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