kestra-flow

Kestra Flow 技能

使用此技能，根据实时架构生成可用于生产环境的 Kestra Flow YAML。

适用场景

在请求包含以下内容时使用此技能：

• 从头生成新的 Kestra Flow
• 修改、扩展或调试现有 Flow
• 将工作流描述转换为有效的 Kestra YAML

必需输入

• 所需 Flow 行为的描述
• 命名空间（如适用，还需租户 ID）
• 如果是修改请求，提供现有 Flow YAML

工作流

步骤 1 — 获取 Flow 架构

使用 curl 获取完整架构并直接读取，不要通过任何解释器处理：

curl -s https://api.kestra.io/v1/plugins/schemas/flow

读取原始 JSON 输出，验证输出中使用的每种类型、属性名称和结构。架构可用之前，不要生成任何内容。

步骤 2 — 收集上下文

从用户消息或对话中确定：

• id — Flow 标识符（如已提供则保留）
• namespace — 目标命名空间（如已提供则保留）
• 现有 Flow YAML（用于修改请求）
• 这是新增／删除／修改，还是完全重写

步骤 3 — 生成 YAML

应用以下所有生成规则，然后仅输出原始 YAML。

生成规则

符合架构

• 只能使用获取的架构中明确定义的任务类型和属性。绝不虚构或猜测类型或属性名称。
• 每个任务或区块中的属性键必须唯一。

保留结构

• 始终保留已提供的根级 id 和 namespace。
• 对于修改请求，只修改相关部分。不要重构或重写无关部分。
• 避免重复已有意图（例如，替换日志消息，而不是再添加一条）。

触发器

• 如果执行应基于事件或计划启动，至少包含一个触发器。
• 除非明确要求定期执行，否则不要添加 Schedule 触发器。
• 通过 {{ trigger.outputName }} 访问触发器输出；只能使用触发器已声明输出中定义的变量。

循环

• 对集合执行重复操作时使用 ForEach。
• 基于条件循环时使用 LoopUntil。

Flow 输出

• 只有用户明确要求从执行中返回值时，才包含 Flow 级别的 outputs。

执行间的状态跟踪

• 如需检测状态变化，使用 KV 任务（io.kestra.plugin.core.kv.Set / io.kestra.plugin.core.kv.Get）存储并比较各次执行之间的状态。

JDBC 插件

• 使用 JDBC 任务时，始终设置 fetchType: STORE。

Pebble 中的日期处理

• 日期运算使用 dateAdd 和 date 过滤器。
• 数值比较前应用 | number。

凭据和密钥

• 绝不嵌入密钥或硬编码凭据。
• 使用类型为 SECRET 的 Flow inputs 或 Pebble 表达式（例如 {{ secret('MY_SECRET') }}）。

API 和连接

• 优先使用公开／无需身份验证的 API，除非用户另有说明。
• 绝不假定本地端口；使用远程 URL。

引号

• 优先使用双引号；如有需要，在双引号字符串中使用单引号。

错误处理

• 如果请求无法仅使用架构中定义的类型和属性完成，请准确输出：

  I cannot generate a valid Kestra Flow YAML for this request based on the available schema.
  

输出格式

• 仅输出原始 YAML — 不要在 YAML 外添加文字、Markdown 代码围栏或解释。
• 对任何注意事项、假设或警告，使用输出开头的 # 注释。
• 输出必须可以直接粘贴到 Kestra UI 中，或通过 kestractl 部署。

示例提示词

• “编写一个 Kestra Flow，每小时获取一次公开 API 并将结果存储到 KV 存储中。”
• “在这个现有 Flow 中添加一个 Slack 通知任务，在任何任务失败时发送通知。”
• “在命名空间 prod.data 中生成一个 Flow，从 Postgres 表读取数据并将结果写入 S3。”
• “调试这个 Flow YAML——其中引用了一个不存在的触发器变量。”