Conductor 工作流如何调用远程 A2A AgentAGENT 任务【免费下载链接】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如果你的工作流需要调用一个部署在别处的 A2AAgent2Agent远程 Agent——比如官方的a2a-sdk参考 Agent 或其他框架实现的 A2A 服务——Conductor 提供了AGENT这类持久化系统任务来完成这件事工作流向远程 Agent 发送message/send由引擎负责轮询、超时、重试和重启恢复远程 Agent 的返回结果状态、文本、artifacts会作为该任务的 output 供下游任务引用。本文的完整操作路径是启用 AI 集成 → 准备一个可被调用的 A2A Agent → 编写并注册包含AGENT任务的工作流 → 运行 → 核对任务输出。主要依据是 A2A 集成文档 与 A2A 端到端测试说明。前置条件启用 AI 集成并放行 Agent 地址AGENT、GET_AGENT_CARD、CANCEL_AGENT这几个任务要求 Conductor 开启 AI 集成默认关闭conductor.integrations.ai.enabledtrue同时注意文档中的 SSRF 防护agentUrl解析到 loopback、RFC-1918 私有网段、link-local 或 IPv6 unique-localfc00::/7地址时会被拒绝云元数据地址则始终被阻断。开发环境里调用本机 Agent如http://localhost:9999需要额外打开conductor.a2a.client.allow-private-networktrue打开该开关后云元数据地址依然被拒。文档建议生产环境优先用网络层出口防火墙控制访问。如果任务报错信息包含… SSRF blocked就是这一层拦截按上面处理即可。准备一个可被调用的远程 A2A Agent测试目标要真正跑通链路你需要一个真实的 A2A 服务端点。仓库自带一个基于官方 Pythona2a-sdk的最小 echo Agentecho_agent.py。副作用说明下面的命令会创建一个 Python 虚拟环境并联网安装a2a-sdk与uvicorn需要uv工具。它不修改 Conductor 仓库本身只在本机准备一个独立的 Agent 进程。uv venv --python 3.12 uv pip install a2a-sdk0.2,0.3 uvicorn AGENT_MODEtask python ai/src/test/resources/a2a/echo_agent.py # serves http://localhost:9999AGENT_MODEtask表示该 Agent 返回一个 A2A Task可轮询AGENT_MODEmessage则直接返回一条 message。该 Agent 起在http://localhost:9999其 Agent Card 位于/.well-known/agent-card.json。如果你有自己部署的 A2A 服务直接把它的基础 URL 当agentUrl用即可——工作流只需要这一个信息。编写工作流AGENT 任务的定义与关键输入agentType决定AGENT任务调用哪一类运行时agentType: a2a默认——调用远程 Agent2Agent 端点agentUrl即本文场景agentType: conductor——按name运行已部署的 Conductor Agent见 Conductor Agents。它不是用来选框架OpenAI Agents、Google ADK、LangGraph 等的无法识别的值会直接让任务失败。完整输入字段参照A2ACallRequest本文用到的字段如下字段说明agentTypea2a默认调远程端点conductor运行已部署 Conductor Agent其他值被拒绝agentUrl远程 Agent 基础 URL必填text/prompt单文本 part 的简写parts/message完整 A2A 消息多 part / data part替代textcontextId、taskId续接已有会话 / 恢复远程 Agent 任务多轮headers每次调用的 HTTP 头如鉴权凭据建议走工作流 input/secrets 而不是硬编码pollIntervalSecondspoll 模式的轮询间隔默认 5streamingtrue→ 消费message/streamSSE并聚合到完成pushNotificationtrue→ 任务完成时由 Agent 回调 Conductor 的 webhookmaxDurationSeconds绝对截止时间默认 86400maxPollFailures容忍的连续瞬时轮询失败次数默认 30示例工作流仓库内ai/examples/10-a2a-call-agent.json展示了最小形态{ 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} } } } }上面的${workflow.input.agentToken}是模板表达式启动工作流时通过 input 传入如果你的 Agent 不需要鉴权删掉整个headers块即可。文档还给出一个更完整的参考工作流a2a_interop_echo先用GET_AGENT_CARD发现 Agent 的 card再调用它agentUrl和 prompt 作为工作流输入同一份定义可以对接任意 A2A Agent{ name: a2a_interop_echo, version: 1, schemaVersion: 2, description: Discover a remote A2A agent, then call it., ownerEmail: a2aexample.com, tasks: [ { name: discover_agent, taskReferenceName: discover, type: GET_AGENT_CARD, inputParameters: { agentUrl: ${workflow.input.agentUrl} } }, { name: call_agent, taskReferenceName: call, type: AGENT, inputParameters: { agentUrl: ${workflow.input.agentUrl}, text: ${workflow.input.prompt}, pollIntervalSeconds: 2 } } ] }GET_AGENT_CARD从/.well-known/agent-card.json解析找不到时回退到旧路径/.well-known/agent.json返回解析后的 skills/capabilities可喂给 LLM 在运行时选 skill。注册并运行工作流把上面的定义保存为a2a_interop_echo.json然后注册、启动# register curl -X POST localhost:8080/api/metadata/workflow \ -H Content-Type: application/json -d a2a_interop_echo.json # run against a reachable A2A agent curl -X POST localhost:8080/api/workflow/a2a_interop_echo \ -H Content-Type: application/json \ -d {agentUrl:http://localhost:9999,prompt:convert 100 USD to EUR}这里agentUrl指向前面启动的 echo Agentlocalhost:9999替换成你自己的 A2A 服务地址即可。验证调用结果仓库的 interop 演示脚本 用下面的方式查询工作流状态与任务输出轮询到终态curl -sS localhost:8080/api/workflow/workflowId判定成功的依据来自 A2A 测试说明 的 Full-server verification工作流状态为COMPLETEDAGENT任务的 output 携带 Agent 返回的state、text、artifacts、taskId、contextId。AGENT任务的 output下称agent.output字段为stateA2A 任务状态、taskId与contextId用于恢复、artifacts、text提取出的文本、agentMessage以及完整task对象。完成调用时文档给出的示例输出文档示例非固定预期{ state: completed, taskId: task-7f3a, contextId: ctx-7f3a, text: 100 USD 92.40 EUR, artifacts: [ { artifactId: result, parts: [ { kind: text, text: 100 USD 92.40 EUR } ] } ] }下游任务用${agent.output.text}、${agent.output.taskId}等表达式引用这些值。仓库还提供了一个端到端演示脚本 run-interop-demo.sh要求 Java 21、curl、uv或已装a2a-sdk的A2A_VENV不需要 Docker、Redis 或 API key。副作用说明它会删除/tmp下的conductor-a2a-interop*状态文件、在/tmp创建 Python venv 并安装依赖、启动本地 Conductor端口 7002与 echo Agent端口 9998进程退出时自动终止这两个进程。运行它可以直接看到文档给出的预期输出示例workflow status : COMPLETED discovered agent: Echo Agent agent state : completed agent reply : echo-task: convert 100 USD to EUR ✓ Conductor discovered and called a real third-party A2A agent. That is A2A interop.长任务三种执行模式怎么选AGENT任务是非阻塞的快速回复立即完成长任务进入IN_PROGRESS后被轮询期间不占 worker 线程重启也能恢复。三种模式Poll默认——任务为IN_PROGRESS按pollIntervalSeconds通过tasks/get轮询轮询之间不占线程调用在重启后存活。Streamingstreaming: true——消费 Agent 的 SSE 流并聚合事件要求 Agent Card 声明capabilities.streamingtrue否则行为退化为 poll。流期间占线程文档建议交互式/短流场景用 streaming长任务优先 poll 或 push。PushpushNotification: true——任务完成时 Agent 回调 Conductor 的 webhook期间不轮询要求配置对外可达的conductor.a2a.callback.url且仍有pushBackstopPollSeconds默认 300兜底轮询防止 webhook 丢失挂死任务。conductor.integrations.ai.enabledtrue conductor.a2a.callback.urlhttps://conductor.example.com失败语义与排查AGENT把远程结果映射到 Conductor 任务状态引擎的正常重试/超时机制随之生效。重试会复用确定性的messageId由workflowInstanceId referenceTaskName iteration派生对按messageId去重的 Agent 相当于 effectively-once 投递。失败原因记录在task.reasonForIncompletion。条件任务状态会重试HTTP 408/429/5xx、连接/读超时、流中断/空流FAILED是JSON-RPC 瞬时错误如-32603internalFAILED是远程 Agent 任务结束为failed/rejectedFAILED是HTTP 4xx408/429 除外FAILED_WITH_TERMINAL_ERROR否JSON-RPC 终态码-32700/-32600/-32601/-32602/-3200{1..5,7}FAILED_WITH_TERMINAL_ERROR否缺agentUrl/ 空消息 /SSRF 拦截FAILED_WITH_TERMINAL_ERROR否超过maxDurationSeconds或连续maxPollFailures次轮询失败FAILED_WITH_TERMINAL_ERROR否文档给出的常见现象对照现象原因 / 处理… SSRF blockedagentUrl解析到私有/loopback/元数据地址。用公网 URL或为可信/开发环境设conductor.a2a.client.allow-private-networktrue云元数据始终拦截streaming: true却表现为 pollAgent Card 是capabilities.streamingfalse客户端只在 Agent 声明支持时才走流N 次轮询失败后失败Agent 不可达——调大maxPollFailures或检查连通性挂起后在截止时间失败Agent 在maxDurationSeconds内未到达终态多轮交互input-required与取消远程任务到达input-required或auth-required时AGENT任务会完成并在 output 中带出 Agent 的问题与taskId/contextId。工作流按状态分支SWITCH再用相同的taskId和contextId发一个新AGENT任务带回答案——这是恢复同一个远程任务而不是开新会话。完整示例见 29-a2a-client-multi-turn.json。取消运行中的远程 Agent 任务用CANCEL_AGENT引用前一步拿到的 taskId{ name: cancel_agent_task, taskReferenceName: cancel, type: CANCEL_AGENT, inputParameters: { agentUrl: https://currency-agent.example.com, taskId: ${agent.output.taskId} } }下一步AGENT是普通持久化任务可用常规 Conductor 算子编排多个 AgentFORK_JOIN并行调用多个 Agent、JOIN汇合结果每条分支各自 crash-safe。仓库内的相关示例10-a2a-call-agent.json — poll 模式调用远程 Agent11-a2a-get-agent-card.json — 发现 Agent 的 skills27-a2a-multi-agent.json —FORK_JOIN → JOIN并行多 Agent28-a2a-llm-pick-skill.json — 发现 → LLM 选 prompt → 调用反向场景把 Conductor 工作流暴露为 A2A Agent 供外部客户端调用与流式服务端行为见 A2A 集成文档 的 server 部分【免费下载链接】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），仅供参考
