Cloudflare Workflows 实战指南：Step API、实例管理与多场景触发编排（autoskills cloudflare-deploy 技能参考）

Cloudflare Workflows 实战指南：Step API、实例管理与多场景触发编排（autoskills cloudflare-deploy 技能参考）

发布时间：2026/10/10 1:29:48
Cloudflare Workflows 实战指南：Step API、实例管理与多场景触发编排（autoskills cloudflare-deploy 技能参考） 【免费下载链接】autoskillsOne command. Your entire AI skill stack. Installed.项目地址https://gitcode.com/gh_mirrors/au/autoskills点击查看免费下载本篇以 autoskills 仓库cloudflare-deploy技能SKILL.md中 Workflows 参考文档为主体系统讲解 Cloudflare Workflows 的完整 API 面step.do/sleep/waitForEvent等 Step API、实例的创建/查询/控制/事件投递、Worker/Queue/Cron/Workflow 间等多种触发方式、错误处理与幂等设计、Wrangler CLI 与 REST API 操作。读完你将能独立编写、配置、部署、触发与调试一个具备自动重试、状态持久化与长时运行能力的多步骤工作流并了解其配额限制与常见陷阱。Cloudflare Workflows 是一种持久化的多步骤应用运行时步骤可自动重试、步骤间状态持久化分钟级到数周级、失败不丢失进度、可等待外部事件/人工审批、可在不消耗资源的情况下休眠。该参考位于 references/workflows/ 目录由 README.md核心概念与入门、configuration.md配置、api.mdAPI 速查本文主体、patterns.md实战模式与 gotchas.md陷阱与限额五份文档组成。Workflows 在 Free 与 Paid Workers 套餐上均可用。核心概念Workflow继承WorkflowEntrypoint并实现run方法的类定义工作流逻辑Instance实例一次独立的执行拥有唯一 ID 与独立状态Step步骤通过step.do()定义的可独立重试的最小单元——API 调用、数据库查询、AI 调用等State状态由步骤返回值持久化步骤名即状态缓存键。最小可运行示例来自 README.mdimport { WorkflowEntrypoint, WorkflowStep, WorkflowEvent } from cloudflare:workers; type Env { MY_WORKFLOW: Workflow; DB: D1Database }; type Params { userId: string }; export class MyWorkflow extends WorkflowEntrypointEnv, Params { async run(event: WorkflowEventParams, step: WorkflowStep) { const user await step.do(fetch user, async () { return await this.env.DB.prepare(SELECT * FROM users WHERE id ?) .bind(event.params.userId).first(); }); await step.sleep(wait 7 days, 7 days); await step.do(send reminder, async () { await sendEmail(user.email, Reminder!); }); } }Step API 详解Step API 是工作流的执行原语全部来自 api.md// step.do() const result await step.do(step name, async () { /* logic */ }); const result await step.do(step name, { retries, timeout }, async () {}); // step.sleep() await step.sleep(description, 1 hour); await step.sleep(description, 5000); // ms // step.sleepUntil() await step.sleepUntil(description, Date.parse(2024-12-31)); // step.waitForEvent() const data await step.waitForEventPayloadType(wait, {event: webhook-type, timeout: 24h}); // Default 24h, max 365d try { const event await step.waitForEvent(wait, { event: approval, timeout: 1h }); } catch (e) { /* Timeout */ }step.do 的进阶配置step.do的第二参数支持重试与超时配置来自 configuration.mdawait step.do(api call, { retries: { limit: 10, // 默认5或 Infinity delay: 10 seconds, // 默认10000ms backoff: exponential // constant | linear | exponential }, timeout: 30 minutes // 单次尝试超时默认 10 分钟 }, async () { const res await fetch(https://api.example.com/data); if (!res.ok) throw new Error(Failed); return res.json(); });并行、条件与循环步骤并行步骤用Promise.all并发执行多个step.doconst [user, settings] await Promise.all([ step.do(fetch user, async () this.env.KV.get(user:${id})), step.do(fetch settings, async () this.env.KV.get(settings:${id})) ]);条件步骤分支必须基于步骤输出等确定性来源禁止在步骤外使用Date.now()const config await step.do(fetch config, async () this.env.KV.get(flags, { type: json }) ); // ✅ 确定性分支基于步骤输出 if (config.enableEmail) { await step.do(send email, async () sendEmail()); } // ❌ 非确定性分支步骤外使用 Date.now if (Date.now() deadline) { /* BAD */ }动态步骤循环遍历列表逐个执行步骤const files await step.do(list files, async () this.env.BUCKET.list() ); for (const file of files.objects) { await step.do(process ${file.key}, async () { const obj await this.env.BUCKET.get(file.key); return processData(await obj.arrayBuffer()); }); }实例管理每个实例是一次独立执行支持创建、批量创建、查询状态与生命周期控制api.md// 创建单个实例 const instance await env.MY_WORKFLOW.create({id: crypto.randomUUID(), params: { userId: user123 }}); // id 可选省略时自动生成 // 自定义保留期创建默认免费版 3 天付费版 30 天 const instance await env.MY_WORKFLOW.create({ id: crypto.randomUUID(), params: { userId: user123 }, retention: 30 days // 覆盖默认保留期 }); // 批量创建最多 100 个幂等跳过已存在的 ID const instances await env.MY_WORKFLOW.createBatch([{id: user1, params: {name: John}}, {id: user2, params: {name: Jane}}]); // 获取实例与状态 const instance await env.MY_WORKFLOW.get(instance-id); const status await instance.status(); // {status: queued | running | paused | errored | terminated | complete | waiting | waitingForPause | unknown, error?, output?} // 生命周期控制 await instance.pause(); await instance.resume(); await instance.terminate(); await instance.restart(); // 发送事件type 必须与 waitForEvent 中声明的类型匹配 await instance.sendEvent({type: approval, payload: { approved: true }});需要注意status()返回的状态枚举覆盖排队、运行、暂停、出错、终止、完成、等待等全部生命周期阶段且可能携带error与output字段sendEvent投递的事件类型必须与工作流内waitForEvent等待的事件类型一致否则不会被消费。触发 Workflows 的方式Workflows 可从多种入口触发api.md从 WorkerHTTP 入口触发export default { async fetch(req, env) { const instance await env.MY_WORKFLOW.create({id: crypto.randomUUID(), params: { userId: user123 }}); return Response.json({ id: instance.id }); }};从 Queue 消费消息触发export default { async queue(batch, env) { for (const msg of batch.messages) { await env.MY_WORKFLOW.create({id: job-${msg.id}, params: msg.body}); } }};从 Cron定时调度触发export default { async scheduled(event, env) { await env.CLEANUP_WORKFLOW.create({id: cleanup-${Date.now()}, params: { timestamp: event.scheduledTime }}); }};从另一个 Workflow 触发非阻塞export class ParentWorkflow extends WorkflowEntrypointEnv, Params { async run(event, step) { const child await step.do(start child, async () await this.env.CHILD_WORKFLOW.create({id: child-${event.instanceId}, params: {}})); } }从 Pages Functions 触发通过 service bindings配置于wrangler.jsonc的service_bindings下见 configuration.md// functions/_middleware.ts export const onRequest: PagesFunctionEnv async ({ env, request }) { const instance await env.MY_WORKFLOW.create({ params: { url: request.url } }); return new Response(Started ${instance.id}); };错误处理与幂等NonRetryableError对于校验失败、凭据无效等重试无意义的错误抛出NonRetryableError可让该步骤直接失败而不消耗重试次数普通Error则会触发重试import { NonRetryableError } from cloudflare:workers; await step.do(validate, async () { if (!event.params.paymentMethod) throw new NonRetryableError(Payment method required); const res await fetch(https://api.example.com/charge, { method: POST }); if (res.status 401) throw new NonRetryableError(Invalid credentials); // 不重试 if (!res.ok) throw new Error(Retryable failure); // 会重试 return res.json(); });捕获错误并执行兜底waitForEvent超时、非重试错误等都会抛出可用 try-catch 包裹后执行清理步骤try { await step.do(risky op, async () { throw new NonRetryableError(Failed); }); } catch (e) { await step.do(cleanup, async () {}); }幂等性check-then-execute重试可能导致重复执行副作用操作因此在执行前先检查是否已完成await step.do(charge, async () { const sub await fetch(https://api/subscriptions/${id}).then(r r.json()); if (sub.charged) return sub; // 已完成直接返回 return await fetch(https://api/subscriptions/${id}, {method: POST, body: JSON.stringify({ amount: 10.0 })}).then(r r.json()); });类型约束Rpc.SerializableParams 与步骤返回值必须是Rpc.SerializableT类型api.md// ✅ 合法类型 type ValidParams { userId: string; count: number; tags: string[]; metadata: Recordstring, unknown; }; // ❌ 非法类型 type InvalidParams { callback: () void; // 函数不可序列化 symbol: symbol; // Symbol 不可序列化 circular: any; // 不允许循环引用 }; // 步骤返回值遵循同样规则 const result await step.do(fetch, async () { return { userId: 123, data: [1, 2, 3] }; // ✅ 纯对象 });休眠与调度支持相对时间sleep与绝对时间sleepUntil两种调度方式// 相对时间 await step.sleep(wait 1 hour, 1 hour); await step.sleep(wait 30 days, 30 days); await step.sleep(wait 5s, 5000); // 毫秒 // 绝对时间 await step.sleepUntil(launch date, Date.parse(24 Oct 2024 13:00:00 UTC)); await step.sleepUntil(deadline, new Date(2024-12-31T23:59:59Z));时间单位支持second、minute、hour、day、week、month、year单次休眠最长 365 天。处于休眠状态的实例不占用并发额度因此可以支撑数百万个同时沉睡的工作流实例详见 gotchas.md 的并发限制说明。参数传递从 Worker 传入参数const instance await env.MY_WORKFLOW.create({ id: crypto.randomUUID(), params: { userId: user123, email: userexample.com } });在工作流内访问async run(event: WorkflowEventParams, step: WorkflowStep) { const userId event.params.userId; const instanceId event.instanceId; const createdAt event.timestamp; }WorkflowEvent除params外还暴露instanceId与timestamp可用于生成确定性步骤名与审计。CLI 触发并传参npx wrangler workflows trigger my-workflow {userId:user123}Wrangler CLI 命令完整命令链api.mdnpm create cloudflarelatest my-workflow -- --template cloudflare/workflows-starter npx wrangler deploy npx wrangler workflows list npx wrangler workflows trigger my-workflow {userId:user123} npx wrangler workflows instances list my-workflow npx wrangler workflows instances describe my-workflow instance-id npx wrangler workflows instances pause/resume/terminate my-workflow instance-id部署前建议先执行npx wrangler whoami确认已认证交互式使用wrangler loginCI/CD 使用CLOUDFLARE_API_TOKEN环境变量详见 SKILL.md。REST API工作流实例也可通过 Cloudflare API v4 管理api.md# 创建实例 curl -X POST https://api.cloudflare.com/client/v4/accounts/{account_id}/workflows/{workflow_name}/instances -H Authorization: Bearer {token} -d {id:custom-id,params:{userId:user123}} # 查询状态 curl https://api.cloudflare.com/client/v4/accounts/{account_id}/workflows/{workflow_name}/instances/{instance_id}/status -H Authorization: Bearer {token} # 发送事件 curl -X POST https://api.cloudflare.com/client/v4/accounts/{account_id}/workflows/{workflow_name}/instances/{instance_id}/events -H Authorization: Bearer {token} -d {type:approval,payload:{approved:true}}配置详解wrangler.jsonc 与绑定wrangler.jsonc 基础配置在wrangler.jsonc中声明工作流configuration.md{ name: my-worker, main: src/index.ts, compatibility_date: 2025-01-01, // 新项目使用当前日期 observability: { enabled: true // 启用 Workflows 仪表盘与结构化日志 }, workflows: [ { name: my-workflow, // 工作流名称 binding: MY_WORKFLOW, // Env 绑定名 class_name: MyWorkflow // TS 类名 // script_name: other-worker // 跨脚本调用时使用 } ], limits: { cpu_ms: 300000 // CPU 上限 5 分钟默认 30s } }多个工作流与跨脚本绑定一个 Worker 可声明多个工作流每个类继承WorkflowEntrypoint并拥有自己的Params类型{ workflows: [ {name: user-onboarding, binding: USER_ONBOARDING, class_name: UserOnboarding}, {name: data-processing, binding: DATA_PROCESSING, class_name: DataProcessing} ] }跨脚本调用Worker B 通过script_name指向定义工作流的 Worker A// Worker B调用方 { workflows: [{ name: billing-workflow, binding: BILLING, script_name: billing-worker // 指向 Worker A }] }通过 this.env 访问绑定工作流内通过this.env访问 KV、D1、R2、AI、Vectorize 等 Cloudflare 绑定type Env { MY_WORKFLOW: Workflow; KV: KVNamespace; DB: D1Database; BUCKET: R2Bucket; AI: Ai; VECTORIZE: VectorizeIndex; }; await step.do(use bindings, async () { const kv await this.env.KV.get(key); const db await this.env.DB.prepare(SELECT * FROM users).first(); const file await this.env.BUCKET.get(file.txt); const ai await this.env.AI.run(cf/meta/llama-2-7b-chat-int8, { prompt: Hi }); });实战模式四类典型工作流patterns.md 给出了可直接套用的完整示例图片处理流水线获取 → AI 生成描述 → 等待审批 → 发布export class ImageProcessingWorkflow extends WorkflowEntrypointEnv, Params { async run(event, step) { const imageData await step.do(fetch, async () (await this.env.BUCKET.get(event.params.imageKey)).arrayBuffer()); const description await step.do(generate description, async () await this.env.AI.run(cf/llava-hf/llava-1.5-7b-hf, {image: Array.from(new Uint8Array(imageData)), prompt: Describe this image, max_tokens: 50}) ); await step.waitForEvent(await approval, { event: approved, timeout: 24h }); await step.do(publish, async () await this.env.BUCKET.put(public/${event.params.imageKey}, imageData)); } }用户生命周期欢迎邮件 → 试用期 7 天 → 转化检查 → 到期提醒export class UserLifecycleWorkflow extends WorkflowEntrypointEnv, Params { async run(event, step) { await step.do(welcome email, async () await sendEmail(event.params.email, Welcome!)); await step.sleep(trial period, 7 days); const hasConverted await step.do(check conversion, async () { const user await this.env.DB.prepare(SELECT subscription_status FROM users WHERE id ?).bind(event.params.userId).first(); return user.subscription_status active; }); if (!hasConverted) await step.do(trial expiration email, async () await sendEmail(event.params.email, Trial ending)); } }数据管道提取 → 转换 → 存储 → 分批装载export class DataPipelineWorkflow extends WorkflowEntrypointEnv, Params { async run(event, step) { const rawData await step.do(extract, {retries: { limit: 10, delay: 30s, backoff: exponential }}, async () { const res await fetch(event.params.sourceUrl); if (!res.ok) throw new Error(Fetch failed); return res.json(); }); const transformed await step.do(transform, async () rawData.map(item ({ id: item.id, normalized: normalizeData(item) })) ); const dataRef await step.do(store, async () { const key processed/${Date.now()}.json; await this.env.BUCKET.put(key, JSON.stringify(transformed)); return { key }; }); await step.do(load, async () { const data await (await this.env.BUCKET.get(dataRef.key)).json(); for (let i 0; i data.length; i 100) { await this.env.DB.batch(data.slice(i, i 100).map(item this.env.DB.prepare(INSERT INTO records VALUES (?, ?)).bind(item.id, item.normalized) )); } }); } }人工审批Human-in-the-Loop创建审批记录 → 等待审批事件48h→ 通过/拒绝/超时自动拒绝export class ApprovalWorkflow extends WorkflowEntrypointEnv, Params { async run(event, step) { await step.do(create approval, async () await this.env.DB.prepare(INSERT INTO approvals (id, user_id, status) VALUES (?, ?, ?)).bind(event.instanceId, event.params.userId, pending).run()); try { const approval await step.waitForEvent{ approved: boolean }(wait for approval, { event: approval-response, timeout: 48h }); if (approval.approved) { await step.do(process approval, async () {}); } else { await step.do(handle rejection, async () {}); } } catch (e) { await step.do(auto reject, async () await this.env.DB.prepare(UPDATE approvals SET status ? WHERE id ?).bind(auto-rejected, event.instanceId).run()); } } }编排模式Fan-Out扇出并行Promise.all并行处理多个对象Parent-Child父子工作流父工作流在step.do中启动子工作流非阻塞随后继续其他工作Race竞速Promise.race取最先完成的步骤结果调度链Scheduled ChainCron 触发后工作流内用step.sleep串联日任务 → 7 天后的周任务。测试工作流使用cloudflare/vitest-pool-workers编写测试patterns.md// vitest.config.ts import { defineWorkersConfig } from cloudflare/vitest-pool-workers/config; export default defineWorkersConfig({ test: { poolOptions: { workers: { wrangler: { configPath: ./wrangler.jsonc } } } } });配合cloudflare:test的Introspection API可以等待步骤完成甚至 mock 步骤行为import { introspectWorkflowInstance } from cloudflare:test; const instance await env.MY_WORKFLOW.create({ params: { userId: 123 } }); const introspector await introspectWorkflowInstance(env.MY_WORKFLOW, instance.id); // 等待步骤完成 const result await introspector.waitForStepResult({ name: fetch user, index: 0 }); // Mock 步骤行为 await introspector.modify(async (m) { await m.mockStepResult({ name: api call }, { mocked: true }); });最佳实践清单✅ 应该做来自 patterns.md步骤粒度化一次 API 调用一个步骤除非要证明幂等幂等设计先检查再执行使用幂等键确定性步骤名使用静态名或基于步骤输出的名字用返回值持久化状态通过步骤返回值保存状态而不是局部变量始终 awaitawait step.do()避免悬空 Promise确定性条件分支基于event.payload或步骤输出判断大对象外置存储超过 1 MiB 的数据放入 R2/KV仅返回引用批量创建多个实例用createBatch()。❌ 不要做写一个巨型步骤破坏持久化与重试控制在步骤外保存状态休眠时会丢失修改事件事件不可变应返回新状态在步骤外使用非确定性逻辑Math.random()、Date.now()必须在步骤内在步骤外执行副作用重启后可能重复执行使用非确定性步骤名破坏缓存忽略超时waitForEvent会抛出需 try-catch复用实例 ID保留期内必须唯一。常见陷阱与限制gotchas.md 总结了高频错误与对策错误原因解决方案Step Timeout步骤超过默认 10 分钟超时用{timeout: 30 minutes}自定义或在 wrangler.jsonc 提高 CPU 上限最大 5 分钟 CPU 时间waitForEvent Timeout超时周期内未收到事件默认 24h最大 365dtry-catch 包裹超时后走默认行为非确定性步骤名步骤名中使用Date.now()等动态值使用event.instanceId等确定性值状态丢失用模块级/局部变量存状态休眠后丢失从step.do()返回值会自动持久化非确定性条件步骤外用Date.now()判断把非确定性操作移入步骤const isLate await step.do(check, async () Date.now() deadline)步骤返回值超限步骤返回超过 1 MiB 的数据大数据存 R2只返回{ key: r2-object-key }引用CPU 超限但运行 30s混淆 CPU 时间与墙钟时间网络请求、数据库查询、休眠不计入 CPU30s 指活跃计算时间幂等性违规重试导致重复扣款/重复动作执行前先检查是否已完成实例 ID 冲突复用实例 ID用时间戳保证唯一${userId}-${Date.now()}完成后实例数据消失完成/出错实例在保留期后被自动删除免费 3 天/付费 30 天完成前将关键数据导出到 KV/R2/D1漏写 awaitstep.do()未 await 造成 fire-and-forget始终await step.do(task, ...)限额速查表限制项FreePaid说明每步骤 CPU10ms30s默认5min最大通过 wrangler.jsonc 的limits.cpu_ms设置步骤状态1 MiB1 MiB单步返回值实例状态100 MB1 GB单实例全部状态每工作流步骤数1,0241,024step.sleep()不计入每日执行数100k无限每日执行上限并发实例2510k等待状态的实例不计入排队实例100k1M最大排队实例数每步子请求501,000每步骤最大出站请求状态保留期3 天30 天已完成实例保留时长步骤默认超时10 min10 min单次尝试waitForEvent 默认超时24h24h最大 365 天waitForEvent 最大超时365 天365 天最长等待时间关键提示处于waiting状态step.sleep或step.waitForEvent期间的实例不计入并发实例上限因此可以支撑数百万个休眠中的工作流。计费参考指标FreePaid说明请求100k/天10M/月 $0.30/M工作流调用CPU 时间10ms/次30M CPU-ms/月 $0.02/M CPU-ms实际 CPU 用量存储1 GB1 GB/月 $0.20/GB-月所有实例运行/出错/休眠/完成阅读建议入门顺序先读 configuration.md配置→ api.mdAPI→ patterns.md模式排障查阅 gotchas.md相关技术选型有状态协调/实时场景可对比 durable-objects消息驱动场景参考 queues工作流入口本身运行在 workers 之上部署与认证流程见 SKILL.md。赞分享【免费下载链接】autoskillsOne command. Your entire AI skill stack. Installed.项目地址https://gitcode.com/gh_mirrors/au/autoskills点击查看免费下载相关推荐Cloudflare Workflows API 完整实战指南Step 编排、实例管理与多端触发Workers API 深度解析Cloudflare Workflows API 完整实战指南Step 编排、实例管理与多端触发Workers API 深度解析 本篇技术指南以 Skil人工智能AI 技能AI 插件Klipper 输入整形配置指南消除振纹且不牺牲打印速度Klipper 输入整形配置指南消除振纹且不牺牲打印速度 如果你的 Klipper 打印机在急转弯处留下层层重复的回声纹路振纹英文叫 ringingCloudflare Argo Smart Routing 实战指南智能路由优化、API 配置与故障排查autoskills cloudflare-deploy 技能参考Cloudflare Argo Smart Routing 实战指南智能路由优化、API 配置与故障排查autoskills cloudflare depl上一篇GraphiQL合作伙伴生态系统建设与商业合作下一篇5分钟部署游戏串流服务器Sunshine自动化工具链全攻略创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考