自定义算子与模板¶

自定义算子用于把自己的 Python 处理逻辑注册为 WorkItem；工作流模板用于保存可复用的 DSL 和运行表单。两者都能帮助复用，但生命周期不同：算子是可运行对象，模板只是创建工作流时的设计材料。

选择要创建的资源¶

需求	使用
在工作流中运行自定义 Python 逻辑	自定义算子
保存一份可复制、可修改的工作流 DSL	工作流模板
让自定义算子出现在 WorkItem Catalog	创建并启用自定义算子
直接运行整个工作流	从模板取得 DSL 后部署工作流

创建自定义算子¶

curl -X POST "$PRODUCT_API_BASE_URL/workflow/v2/custom-operators" \
  -H "X-API-Key: $PRODUCT_API_KEY" \
  -H "X-Workspace-ID: $WORKSPACE_ID" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "Text Counter",
    "identifier": "text_counter",
    "description": "Count text length for downstream workflow decisions.",
    "language": "python",
    "handler": "main.handle",
    "version": "v1",
    "enabled": true,
    "input_schema": {
      "type": "object",
      "properties": {
        "text": {
          "type": "string",
          "description": "Text to count"
        }
      },
      "required": ["text"]
    },
    "output_schema": {
      "type": "object",
      "properties": {
        "length": {
          "type": "integer",
          "description": "Character count"
        }
      },
      "required": ["length"]
    },
    "code": "def handle(workspace_id, sdk, input):\n    return {\"length\": len(input[\"text\"])}"
  }'

字段	类型	必需	说明
`name`	string	是	面向工作流设计者的算子名称。
`identifier`	string	是	算子稳定标识。同一标识可以有不同版本。
`description`	string	是	说明算子用途、适用条件和结果，帮助用户和规划器正确选用。
`language`	string	是	当前代码算子使用 `python`。
`handler`	string	是	模块级处理函数，例如 `main.handle`。
`version`	string	是	算子版本。发布不兼容变更时创建新版本，不覆盖旧契约。
`input_schema`	object	是	输入 JSON Schema。每个字段应说明含义。
`output_schema`	object	是	输出 JSON Schema。应与处理函数实际返回值一致。
`code`	string	与 `source_file_id` 二选一	Python 源码。不要同时传递内联源码和源码文件 ID。

创建成功后保存 data.id、data.node_id 和 data.version。工作流 DSL 使用 node_id 选择该 WorkItem，算子管理接口使用数值 id。

测试并启用算子¶

在业务工作流中使用前，调用测试接口并让输入符合 input_schema：

curl -X POST \
  "$PRODUCT_API_BASE_URL/workflow/v2/custom-operators/$OPERATOR_ID/test-run" \
  -H "X-API-Key: $PRODUCT_API_KEY" \
  -H "X-Workspace-ID: $WORKSPACE_ID" \
  -H "Content-Type: application/json" \
  -d '{
    "input": {
      "text": "hello"
    },
    "wait_timeout_seconds": 30
  }'

检查 data.test_run.status、runtime_input、runtime_output 和错误字段。测试超时不表示服务端任务一定已经停止；保留返回的任务或 Case ID 进行排查。

算子的新版本通过再次创建产生。更新接口不能修改既有算子的 version 或 kind；启用和停用使用独立的 /enable、/disable 动作。删除一个版本前，先检查工作流是否仍引用该版本。

创建工作流模板¶

模板保存 DSL，不会自动部署或运行：

curl -X POST "$PRODUCT_API_BASE_URL/workflow-templates" \
  -H "X-API-Key: $PRODUCT_API_KEY" \
  -H "X-Workspace-ID: $WORKSPACE_ID" \
  -H "Content-Type: application/json" \
  -d '{
    "template_key": "catalog-list-template",
    "name": "Catalog list template",
    "description": "Start a workflow that lists catalog resources.",
    "dsl_yaml": "workflow:\n  name: catalog-list-template\n  root: root\nroot:\n  chain:\n    - work_item:\n        name: list_catalog\n        id: catalog:catalog.list\n",
    "runtime_fields": "{\"fields\":[]}",
    "is_builtin": false
  }'

保存返回的模板 id。使用模板时读取最新详情，将 dsl_yaml 和运行字段作为新工作流的起点，完成必要修改后再调用部署接口。模板 ID 不能用于运行工作流。

内置模板由平台管理，不能假定可以更新或删除。自建模板变更也不会自动更新已经从该模板创建的工作流。

常见问题¶

现象	先检查	下一步
算子创建失败	Handler、Python 语法和输入输出 Schema	在本地校验代码与 Schema，再提交同一版本。
WorkItem Catalog 中看不到算子	算子状态、工作区和版本	启用算子，并按 `node_id` 或版本重新查询 WorkItem Catalog。
测试输出与 Schema 不一致	`runtime_output` 和 `output_schema`	修正代码或发布新版本，不让工作流依赖错误契约。
模板可以读取但不能运行	是否已部署工作流	从模板取得 DSL，调用工作流部署接口并保存工作流 ID。

下一步¶

最后更新于 2026 年 8 月 14 日