Conductor Metadata API 完全指南：用 REST 管理 Workflow 与 Task 定义

Conductor Metadata API 完全指南用 REST 管理 Workflow 与 Task 定义【免费下载链接】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 的 Metadata API基础路径/api/metadata该 API 负责编排蓝图workflow definition与任务定义task definition的注册、更新、校验与删除是所有 Conductor 工作流落地前的第一道工序。读完本文你将掌握全部 15 个元数据端点的请求/响应格式、参数语义与默认值并能结合源码理解其校验规则、版本管理与批量更新的底层行为直接用于 CI/CD 流水线或日常运维。Metadata endpoints manage definition objects. 与运行时执行数据不同Metadata API 管理的是 Conductor 用来编排执行的定义对象blueprints。本文对应的规范文档是 docs/documentation/api/metadata.md字段级契约以仓库根目录下的 schemas/WorkflowDef.json 与 schemas/TaskDef.json 为准也可参考 Schemas 文档。运行时对象则由 Workflow API 和 Task API 返回。Metadata API 概览所有端点统一挂在基础路径/api/metadata下。REST 层入口是 MetadataResource.java 中的MetadataResource控制器其类级注解RequestMapping(value METADATA)引用了常量METADATA API_PREFIX metadata定义于 RequestMappingConstants.java即/api/metadata。控制器本身只做参数绑定与 HTTP 语义映射真正的业务逻辑全部委托给MetadataService接口MetadataService.java实现MetadataServiceImpl.javaMetadataServiceImpl通过MetadataDAO访问持久化层并在每次定义变更注册、更新、删除后触发MetadataChangeListener回调供缓存失效、事件通知等扩展场景使用。所有接口行为均有对应的单元测试覆盖见 MetadataResourceTest.java。Workflow Definitions工作流定义端点总览EndpointMethodDescription/metadata/workflowGETGet all workflow definitions/metadata/workflowPOSTCreate a new workflow definition/metadata/workflowPUTCreate or update workflow definitions (batch)/metadata/workflow/{name}GETGet a workflow definition by name/metadata/workflow/{name}/{version}DELETEDelete a workflow definition by name and version/metadata/workflow/validatePOSTValidate a workflow definition without saving/metadata/workflow/names-and-versionsGETGet all workflow names and versions (no definition bodies)/metadata/workflow/namesGETGet distinct workflow names only/metadata/workflow/{name}/versionsGETGet lightweight version summaries for one workflow/metadata/workflow/latest-versionsGETGet only the latest version of each workflow definition获取全部工作流定义GET /api/metadata/workflow返回所有已注册的工作流定义列表curl http://localhost:8080/api/metadata/workflowResponse200 OK[ { name: order_processing, version: 1, tasks: [...], inputParameters: [], outputParameters: {}, schemaVersion: 2 } ]从源码看该端点还支持一个可选的classifier查询参数GET /metadata/workflow?classifierxxx当提供该参数时控制器会按WorkflowClassifier.classifierOf(wd)对每个定义打标并过滤如workflow匹配未打标的普通定义其他值按标签字面量匹配。这主要用于 Agent 定义与普通工作流定义的视图区分见 MetadataResource.java 及对应测试 testGetAllWorkflowDefFilteredByClassifier。创建工作流定义POST /api/metadata/workflow注册一个新的工作流定义。请求体是一个 Workflow Definition其 JSON Schema 契约见 schemas/WorkflowDef.jsonname与tasks为必填schemaVersion固定为 2。curl -X POST http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json \ -d { name: my_workflow, version: 1, tasks: [ { name: my_task, taskReferenceName: my_task_ref, type: SIMPLE } ], schemaVersion: 2, ownerEmail: devexample.com }Response200 OK— no response body.关于覆盖行为的源码细节POST /metadata/workflow实际接收一个可选的overwrite查询参数默认false。控制器逻辑为见 MetadataResource.java若同名同版本定义不存在则直接注册若已存在且overwritetrue则升级为更新操作若已存在且overwritefalse抛出ConflictExceptionHTTP 409。批量创建或更新工作流定义PUT /api/metadata/workflow以批量方式创建或更新工作流定义。请求体是 Workflow Definitions 的列表返回一个BulkResponse逐条标识每个定义的成败。curl -X PUT http://localhost:8080/api/metadata/workflow \ -H Content-Type: application/json \ -d [ {name: workflow_a, version: 1, tasks: [...], schemaVersion: 2}, {name: workflow_b, version: 1, tasks: [...], schemaVersion: 2} ]Response200 OK{ bulkSuccessfulResults: [workflow_a, workflow_b], bulkErrorResults: {} }底层实现在MetadataServiceImpl.updateWorkflowDef(ListWorkflowDef)见 MetadataServiceImpl.java对列表逐条执行更新成功的 name 追加进bulkSuccessfulResults失败的在bulkErrorResults中记录异常消息单条失败不会中断整批处理。按名称获取工作流定义GET /api/metadata/workflow/{name}?version{version}ParameterDescriptionRequirednameWorkflow nameYes (path)versionWorkflow versionNo (defaults to latest)curl http://localhost:8080/api/metadata/workflow/my_workflow?version1Response200 OK— returns the full workflow definition JSON.服务层实现MetadataServiceImpl.javaversion缺省时调用metadataDAO.getLatestWorkflowDef(name)取最新版本定义不存在时抛出NotFoundExceptionHTTP 404。删除工作流定义DELETE /api/metadata/workflow/{name}/{version}按名称与版本移除一个工作流定义。不会删除与该定义关联的工作流执行实例执行历史与审计数据得以保留。ParameterDescriptionRequirednameWorkflow nameYes (path)versionWorkflow versionYes (path)curl -X DELETE http://localhost:8080/api/metadata/workflow/my_workflow/1Response200 OK— no response body.校验工作流定义不落库POST /api/metadata/workflow/validate在不注册的前提下校验工作流定义是否合法非常适合 CI/CD 流水线或上线前的预检。curl -X POST http://localhost:8080/api/metadata/workflow/validate \ -H Content-Type: application/json \ -d { name: my_workflow, version: 1, tasks: [ { name: my_task, taskReferenceName: my_task_ref, type: SIMPLE } ], schemaVersion: 2 }Response200 OKif valid.400 Bad Requestwith error details if invalid.校验机制并非手写业务逻辑WorkflowDef类上带有Valid注解Bean Validationjakarta.validation会在请求体反序列化时自动执行字段约束校验因此控制器中的validate方法体本身为空见 MetadataServiceImpl.java。任何违反约束如name、tasks缺失都会返回 400 及具体错误明细。获取工作流名称与版本不含定义体GET /api/metadata/workflow/names-and-versions返回工作流名称到可用版本的轻量映射不含定义体适合构建 UI 列表或做可用工作流盘点。curl http://localhost:8080/api/metadata/workflow/names-and-versionsResponse200 OK{ order_processing: [ {name: order_processing, version: 1}, {name: order_processing, version: 2} ], user_onboarding: [ {name: user_onboarding, version: 1} ] }仅获取最新版本GET /api/metadata/workflow/latest-versions返回每个工作流定义的最新版本每个 name 只返回一条version 取最大值。curl http://localhost:8080/api/metadata/workflow/latest-versionsResponse200 OK— returns a list of workflow definitions (one per workflow name, latest version only).该端点对应服务层getWorkflowDefsLatestVersions()调用metadataDAO.getAllWorkflowDefsLatestVersions()见 MetadataServiceImpl.java。无定义体的轻量发现接口GET /api/metadata/workflow/names GET /api/metadata/workflow/{name}/versionsGET /metadata/workflow/names返回去重后的工作流名称 JSON 数组对应metadataDAO.getWorkflowNames()。GET /metadata/workflow/{name}/versions返回该工作流的轻量WorkflowDefSummary列表。WorkflowDefSummary仅含name、version、createTime、updateTime四个字段组装逻辑见 MetadataServiceImpl.java并按版本排序去重。当调用方只需要发现数据、不需要下载完整定义时优先使用这两条路由以降低传输与解析开销。Task Definitions任务定义任务定义TaskDef是SIMPLE任务类型的注册配置描述 worker 如何被调度执行——包括重试策略、超时、限流与并发约束。字段级契约见 schemas/TaskDef.json 与 TaskDef.java。端点总览EndpointMethodDescription/metadata/taskdefsGETGet all task definitions/metadata/taskdefsPOSTCreate new task definitions/metadata/taskdefsPUTUpdate a task definition/metadata/taskdefs/{taskType}GETGet a task definition by name/metadata/taskdefs/{taskType}DELETEDelete a task definitionTaskDef 关键字段与默认值结合 TaskDef.java 与 TaskDef.json常用字段及默认行为如下字段说明默认值name任务唯一名称必填不能为空—retryCount失败重试次数3retryLogic重试策略FIXED/EXPONENTIAL_BACKOFF/LINEAR_BACKOFFFIXEDretryDelaySeconds重试前等待秒数60timeoutSeconds任务超时秒数超时后按timeoutPolicy处理必须显式设置timeoutPolicy超时策略RETRY/TIME_OUT_WF/ALERT_ONLYTIME_OUT_WFresponseTimeoutSeconds等待 worker 响应秒数超时则重新入队36001 小时rateLimitPerFrequency/rateLimitFrequencyInSeconds限流每窗口最大执行数 / 窗口秒数0 表示不限流窗口缺省 1 秒不限流concurrentExecLimit同一时刻IN_PROGRESS任务上限0 表示不限制不限制ownerEmail负责人邮箱—此外还支持pollTimeoutSeconds、backoffScaleFactorLINEAR_BACKOFF乘数、maxRetryDelaySeconds指数/线性退避的延迟上限、backoffJitterMs重试延迟抖动避免惊群、totalTimeoutSeconds含重试的总超时、isolationGroupId/executionNameSpace隔离与命名空间以及inputSchema/outputSchema/enforceSchema输入输出 Schema 校验。所有字段均通过Min、NotEmpty、OwnerEmailMandatoryConstraint等注解在服务端强制校验。获取全部任务定义GET /api/metadata/taskdefscurl http://localhost:8080/api/metadata/taskdefsResponse200 OK[ { name: my_task, retryCount: 3, retryLogic: FIXED, retryDelaySeconds: 10, timeoutSeconds: 300, timeoutPolicy: TIME_OUT_WF, responseTimeoutSeconds: 180 } ]创建任务定义POST /api/metadata/taskdefs注册新的任务定义。注意请求体是 Task Definitions 的列表ListTaskDef见 MetadataResource.java即使只创建一个也要用数组包裹。curl -X POST http://localhost:8080/api/metadata/taskdefs \ -H Content-Type: application/json \ -d [ { name: my_task, retryCount: 3, retryLogic: FIXED, retryDelaySeconds: 10, timeoutSeconds: 300, timeoutPolicy: TIME_OUT_WF, responseTimeoutSeconds: 180, ownerEmail: devexample.com } ]Response200 OK— no response body.服务层在注册时会自动填充审计字段createdBy取自当前调用方WorkflowContext的clientAppcreateTime取当前毫秒时间戳见 MetadataServiceImpl.java。更新任务定义PUT /api/metadata/taskdefs更新一个已存在的任务定义。与 POST 不同请求体是单个Task Definition。curl -X PUT http://localhost:8080/api/metadata/taskdefs \ -H Content-Type: application/json \ -d { name: my_task, retryCount: 5, retryLogic: EXPONENTIAL_BACKOFF, retryDelaySeconds: 5, timeoutSeconds: 600, timeoutPolicy: TIME_OUT_WF, responseTimeoutSeconds: 300 }Response200 OK— no response body.更新时会校验目标定义存在若metadataDAO.getTaskDef(name)返回 null抛出NotFoundExceptionHTTP 404。同时保留原createTime/createdBy仅刷新updateTime/updatedBy见 MetadataServiceImpl.java。按名称获取任务定义GET /api/metadata/taskdefs/{taskType}curl http://localhost:8080/api/metadata/taskdefs/my_taskResponse200 OK— returns the task definition JSON.定义不存在时同样返回 404No such taskType found by name。删除任务定义DELETE /api/metadata/taskdefs/{taskType}curl -X DELETE http://localhost:8080/api/metadata/taskdefs/my_taskResponse200 OK— no response body.实战建议上线前先校验在 CI/CD 中对每个WorkflowDef.json先调用POST /api/metadata/workflow/validate做预检通过后再走PUT /api/metadata/workflow批量落地校验失败返回的 400 明细可直接映射为流水线报错。注意 POST 与 PUT 的覆盖语义差异POST /metadata/workflow默认幂等拒绝需?overwritetrue才允许覆盖而PUT /metadata/workflow天然支持创建或更新批量发布推荐使用 PUT。任务定义走数组、工作流定义批量走列表POST /metadata/taskdefs请求体必须是数组PUT /metadata/workflow请求体是定义列表并返回BulkResponse便于逐条追踪成败。用轻量发现接口做列表页names-and-versions、names、{name}/versions、latest-versions四条路由均不返回定义体适合 UI 工作流选择器、版本清单等高频只读场景。删除不删执行历史删除WorkflowDef只会移除蓝图本身已产生的工作流执行Workflow与任务Task实例不受影响可安全用于清理废弃版本。审计与权限定义上建议始终携带ownerEmail——当服务端开启ownerEmail强制校验ConductorProperties.isOwnerEmailMandatory()见 MetadataServiceImpl.java时缺失将直接报错createdBy/updatedBy/createTime/updateTime由服务端自动维护。【免费下载链接】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），仅供参考