Workflow

Workflow DevKit

TypeScript framework for building durable, resumable functions. Workflows survive crashes, deployments, and can pause for days/months.

Website: https://useworkflow.dev | GitHub: https://github.com/vercel/workflow

Installation

npm i workflow
# or: pnpm add workflow | yarn add workflow | bun add workflow

Core Concepts

Directives

"use workflow";  // First line - makes async function durable
"use step";      // First line - makes function a cached, retryable unit

Workflow Function

Orchestrates steps, survives restarts, replays deterministically:

import { sleep } from "workflow";

export async function onboardUser(email: string) {
  "use workflow";

  const user = await createUser(email);      // Step 1
  await sendWelcomeEmail(user);              // Step 2
  await sleep("3 days");                     // Pause (no resources consumed)
  await sendFollowupEmail(user);             // Step 3
  return { userId: user.id };
}

Step Function

Has full Node.js access, auto-retries on failure:

async function createUser(email: string) {
  "use step";
  return await db.users.create({ email });
}

Essential APIs

sleep - Pause Workflow

import { sleep } from "workflow";

await sleep("5s");      // 5 seconds
await sleep("10m");     // 10 minutes
await sleep("1h");      // 1 hour
await sleep("7 days");  // 7 days
await sleep(5000);      // milliseconds
await sleep(new Date("2025-01-01")); // specific date

createHook - Wait for External Input

import { createHook } from "workflow";

const hook = createHook<{ approved: boolean }>();
console.log("Token:", hook.token);  // Send to external system
const result = await hook;          // Pauses until resumed

Resume externally:

import { resumeHook } from "workflow/api";
await resumeHook(token, { approved: true });

createWebhook - HTTP Callback Endpoint

import { createWebhook } from "workflow";

const webhook = createWebhook();
await sendToExternalService(webhook.url);  // Auto-generated URL
const request = await webhook;              // Pauses until HTTP POST
const data = await request.json();

start - Trigger Workflow

import { start } from "workflow/api";

const run = await start(onboardUser, ["user@example.com"]);
console.log(run.runId);  // Returns immediately

Error Handling

import { FatalError, RetryableError } from "workflow";

// Stop retries permanently
throw new FatalError("User not found");

// Retry with delay
throw new RetryableError("Rate limited", { retryAfter: "5m" });

Fetch in Workflows

Workflows run sandboxed. Import fetch from workflow:

import { fetch } from "workflow";

export async function myWorkflow() {
  "use workflow";
  globalThis.fetch = fetch;  // Enable for libraries (AI SDK)

  const response = await fetch("https://api.example.com/data");
}

Streaming (Steps Only)

import { getWritable } from "workflow";

async function streamData(items: string[]) {
  "use step";
  const writer = getWritable();
  for (const item of items) {
    await writer.write({ item });
  }
  writer.releaseLock();
}

Idempotency

Use stepId for external API calls:

import { getStepMetadata } from "workflow";

async function chargeCard(amount: number) {
  "use step";
  const { stepId } = getStepMetadata();
  await stripe.charges.create(
    { amount },
    { idempotencyKey: `charge:${stepId}` }
  );
}

What Cannot Be Serialized

• Functions/callbacks - define logic in steps
• Class instances with methods - use plain objects
• Symbols, WeakMap, WeakSet
• Node.js modules (fs, http, crypto) - use in steps only

Run Status

pending | running | completed | failed | cancelled

References

• Framework Setup: See frameworks.md for Next.js, Express, Hono, Vite, Astro, Fastify, Nitro, Nuxt, SvelteKit
• AI Agents: See ai-agents.md for DurableAgent and streaming patterns
• Advanced Patterns: See patterns.md for hooks, webhooks, control flow
• Common Errors: See errors.md for troubleshooting
• API Reference: See api-reference.md for complete API

Quick Checklist

• [ ] Add "use workflow" as first line of workflow function
• [ ] Add "use step" as first line of step functions
• [ ] Import fetch from workflow and assign to globalThis.fetch
• [ ] Use sleep() instead of setTimeout
• [ ] Move Node.js module usage to step functions
• [ ] Use FatalError for permanent failures, RetryableError for retries
• [ ] Configure framework (withWorkflow, nitro module, etc.)

AI Agents with Workflow

Build durable AI agents that survive restarts and stream responses.

Installation

npm i @workflow/ai ai

Supports AI SDK v5 and v6.

DurableAgent

Creates AI agents that maintain state across workflow steps:

import { DurableAgent } from "@workflow/ai/agent";
import { fetch, getWritable } from "workflow";
import { z } from "zod";

const agent = new DurableAgent({
  model: "anthropic/claude-haiku-4.5",
  system: "You are a helpful assistant.",
  temperature: 0.7,
  maxSteps: 10,
  tools: {
    searchWeb: {
      description: "Search the web for information",
      parameters: z.object({ query: z.string() }),
      execute: async ({ query }) => {
        "use step";
        return await webSearch(query);
      },
    },
    getWeather: {
      description: "Get current weather",
      parameters: z.object({ city: z.string() }),
      execute: async ({ city }) => {
        "use step";
        return await weatherAPI.get(city);
      },
    },
  },
});

export async function chatWorkflow(messages: Message[]) {
  "use workflow";
  globalThis.fetch = fetch;  // Required for AI SDK

  const writable = getWritable();
  const result = await agent.stream({
    messages,
    writable,
  });

  return result.messages;
}

DurableAgent Options

Callbacks

const agent = new DurableAgent({
  model: "anthropic/claude-haiku-4.5",
  onFinish: (result) => console.log("Done:", result),
  onError: (error) => console.error("Error:", error),
  onAbort: () => console.log("Aborted"),
  onStepFinish: (step) => console.log("Step:", step),
});

Streaming AI Responses

In Workflow

export async function streamingWorkflow(prompt: string) {
  "use workflow";
  globalThis.fetch = fetch;

  const writable = getWritable();
  await agent.stream({ messages: [{ role: "user", content: prompt }], writable });
}

Reading Stream (Client)

import { start } from "workflow/api";

const run = await start(streamingWorkflow, ["Hello"]);

for await (const chunk of run.readable) {
  console.log(chunk);
}

Resumable Streams

// Resume from specific position
const readable = run.getReadable({ startIndex: lastIndex });

for await (const chunk of readable) {
  process.stdout.write(chunk);
}

WorkflowChatTransport

Transport layer for AI SDK with automatic reconnection:

import { WorkflowChatTransport } from "@workflow/ai";

const transport = new WorkflowChatTransport({
  maxConsecutiveErrors: 3,
  onChatSendMessage: (response) => {
    // Extract workflow run ID from header
    const runId = response.headers.get("x-workflow-run-id");
  },
  onChatEnd: ({ chatId, chunkIndex }) => {
    console.log("Chat ended at chunk:", chunkIndex);
  },
});

Piping AI SDK Responses

async function generateWithAI(prompt: string) {
  "use step";

  const writable = getWritable();
  const response = await generateText({ prompt, stream: true });

  await response.pipeThrough(writable);
}

Human-in-the-Loop AI

Combine AI with human approval:

export async function contentApproval(topic: string) {
  "use workflow";
  globalThis.fetch = fetch;

  // AI generates draft
  const draft = await generateDraft(topic);

  // Wait for human review
  const hook = createHook<{ approved: boolean; feedback?: string }>();
  await notifyEditor(draft.id, hook.token);

  const review = await hook;

  if (review.approved) {
    await publish(draft);
    return { status: "published", draft };
  } else {
    // AI revises based on feedback
    const revised = await reviseDraft(draft, review.feedback);
    return { status: "revised", draft: revised };
  }
}

async function generateDraft(topic: string) {
  "use step";
  const writable = getWritable();
  return await agent.stream({
    messages: [{ role: "user", content: `Write about: ${topic}` }],
    writable,
  });
}

Multi-Day AI Agents

Agents that operate over extended periods:

export async function longRunningAgent(task: string) {
  "use workflow";
  globalThis.fetch = fetch;

  // Initial processing
  const plan = await createPlan(task);

  for (const step of plan.steps) {
    await executeStep(step);

    // Wait for external systems
    await sleep("1h");

    // Check status
    const status = await checkProgress(step);
    if (status.needsReview) {
      const hook = createHook<{ proceed: boolean }>();
      await notifyTeam(step, hook.token);
      const decision = await hook;
      if (!decision.proceed) break;
    }
  }

  return await generateReport(plan);
}

Important Notes

1. Always set globalThis.fetch - AI SDK requires it 2. Tools must use "use step" - For durability and retry 3. Use getWritable() in steps only - Cannot stream from workflow function 4. Release writer locks - Call writer.releaseLock() after writing

Complete API Reference

workflow package

sleep(duration)

Pause workflow without consuming resources.

import { sleep } from "workflow";

await sleep("5s");              // String duration
await sleep("10m");             // 10 minutes
await sleep("1h");              // 1 hour
await sleep("7 days");          // 7 days
await sleep(5000);              // Milliseconds
await sleep(new Date("2025-12-31"));  // Until specific date

Duration formats: ${number}${'ms'|'s'|'m'|'h'|'d'}

fetch(url, options)

HTTP request with automatic retry. Use instead of global fetch.

import { fetch } from "workflow";

export async function myWorkflow() {
  "use workflow";
  globalThis.fetch = fetch;  // Enable for libraries

  const response = await fetch("https://api.example.com/data");
  const data = await response.json();
}

createHook<T>(options?)

Create suspension point for external input.

import { createHook } from "workflow";

const hook = createHook<{ approved: boolean }>();
// hook.token - string identifier for resumption
const result = await hook;  // Pauses until resumed

Options:

• token?: string - Custom deterministic token
• metadata?: any - Attached metadata

Returns: Hook<T> - PromiseLike & AsyncIterable

defineHook<I, O>(config)

Type-safe hook with schema validation.

import { defineHook } from "workflow";
import { z } from "zod";

const myHook = defineHook({
  schema: z.object({
    id: z.string(),
    value: z.number(),
  }),
});

// Create hook
const hook = myHook.create({ token: `custom:${id}` });

// Resume with validation
await myHook.resume(`custom:${id}`, { id: "123", value: 42 });

createWebhook(options?)

Create HTTP webhook endpoint.

import { createWebhook } from "workflow";

const webhook = createWebhook();
console.log(webhook.url);  // Auto-generated URL
const request = await webhook;  // Pauses until HTTP POST

Options:

• token?: string - Custom token
• metadata?: any - Attached metadata
• respondWith?: Response | "manual" - Response mode

Returns: Webhook<Request> - PromiseLike & AsyncIterable with .url

getWritable<T>(options?)

Access stream writer (steps only).

import { getWritable } from "workflow";

async function streamStep() {
  "use step";

  const writer = getWritable();
  await writer.write({ data: "chunk" });
  writer.releaseLock();
}

Options:

• namespace?: string - Stream namespace

Returns: WritableStream<T>

getWorkflowMetadata()

Get workflow run context.

import { getWorkflowMetadata } from "workflow";

export async function myWorkflow() {
  "use workflow";

  const { runId, workflowName } = getWorkflowMetadata();
}

getStepMetadata()

Get step execution context.

import { getStepMetadata } from "workflow";

async function myStep() {
  "use step";

  const { stepId, attemptNumber } = getStepMetadata();
  // Use stepId for idempotency keys
}

FatalError

Stop retries permanently.

import { FatalError } from "workflow";

throw new FatalError("User not found");
throw new FatalError("Invalid input", { cause: originalError });

RetryableError

Trigger retry with delay.

import { RetryableError } from "workflow";

throw new RetryableError("Rate limited", { retryAfter: "5m" });
throw new RetryableError("Retry", { retryAfter: 5000 });
throw new RetryableError("Retry", { retryAfter: new Date("2025-01-01") });

---

workflow/api package

start(workflow, args, options?)

Start workflow run.

import { start } from "workflow/api";

const run = await start(myWorkflow, ["arg1", "arg2"]);
console.log(run.runId);

// With options
const run = await start(myWorkflow, ["arg1"], {
  runId: "custom-id",
});

Returns: Run<T> object

Run<T> object

interface Run<T> {
  runId: string;

  // Get current status
  get status(): Promise<RunStatus>;

  // Get return value (waits for completion)
  get returnValue(): Promise<T>;

  // Cancel the run
  cancel(): Promise<void>;

  // Get readable stream
  getReadable<R>(options?: { namespace?: string }): ReadableStream<R>;
}

getRun(runId)

Get existing run by ID.

import { getRun } from "workflow/api";

const run = getRun<ReturnType>("run-id");
const status = await run.status;
const result = await run.returnValue;

resumeHook(token, data)

Resume workflow via hook.

import { resumeHook } from "workflow/api";

const result = await resumeHook(token, { approved: true });
// result.runId, result.hookId

resumeWebhook(token, request)

Resume workflow via webhook.

import { resumeWebhook } from "workflow/api";

await resumeWebhook(token, incomingRequest);

getHookByToken(token)

Get hook metadata.

import { getHookByToken } from "workflow/api";

const hook = await getHookByToken(token);
// hook.status: 'pending' | 'received' | 'disposed'

getWorld()

Access low-level infrastructure.

import { getWorld } from "workflow/api";

const world = getWorld();
const runs = await world.runs.list({ pagination: { cursor } });

---

Types

RunStatus

type RunStatus = 'pending' | 'running' | 'completed' | 'failed' | 'cancelled';

EventType

type EventType =
  | 'step_completed' | 'step_failed' | 'step_retrying' | 'step_started'
  | 'hook_created' | 'hook_received' | 'hook_disposed'
  | 'wait_created' | 'wait_completed'
  | 'workflow_completed' | 'workflow_failed' | 'workflow_started';

Serializable Types

type Serializable =
  | string | number | boolean | null | undefined | bigint
  | Serializable[]
  | { [key: string]: Serializable }
  | Date | URL | RegExp | URLSearchParams
  | Map<Serializable, Serializable> | Set<Serializable>
  | Response | Request | Headers
  | ArrayBuffer | Uint8Array | Int8Array | Float64Array  // all typed arrays
  | ReadableStream | WritableStream;

---

Framework Exports

workflow/next

import { withWorkflow } from "workflow/next";

export default withWorkflow(nextConfig);

workflow/vite

import { workflow } from "workflow/vite";

export default defineConfig({
  plugins: [workflow()],
});

workflow/astro

import { workflow } from "workflow/astro";

export default defineConfig({
  integrations: [workflow()],
});

workflow/nitro

// nitro.config.ts
export default defineNitroConfig({
  modules: ["workflow/nitro"],
});

workflow/nuxt

// nuxt.config.ts
export default defineNuxtConfig({
  modules: ["workflow/nuxt"],
});

workflow/sveltekit

import { workflowPlugin } from "workflow/sveltekit";

export default {
  plugins: [sveltekit(), workflowPlugin()],
};

---

@workflow/ai

DurableAgent

import { DurableAgent } from "@workflow/ai/agent";

const agent = new DurableAgent({
  model: string,
  system?: string,
  temperature?: number,
  tools?: Record<string, ToolDefinition>,
  toolChoice?: "auto" | "required" | "none" | string,
  maxSteps?: number,
  onFinish?: (result) => void,
  onError?: (error) => void,
  onAbort?: () => void,
  onStepFinish?: (step) => void,
});

await agent.stream({ messages, writable });

WorkflowChatTransport

import { WorkflowChatTransport } from "@workflow/ai";

const transport = new WorkflowChatTransport({
  maxConsecutiveErrors?: number,
  onChatSendMessage?: (response) => void,
  onChatEnd?: ({ chatId, chunkIndex }) => void,
});

---

Worlds

@workflow/world-local

Local development with filesystem storage.

// Auto-used in development
// Data in .workflow-data/ or .next/workflow-data/

@workflow/world-vercel

Production on Vercel with durable queues.

// Auto-used when deployed to Vercel
// No configuration needed

@workflow/world-postgres

Self-hosted PostgreSQL backend.

import { PostgresWorld } from "@workflow/world-postgres";

@workflow/world-testing

Testing utilities.

import { TestWorld } from "@workflow/world-testing";

Common Errors and Solutions

Quick Reference

fetch-in-workflow

Error: Global fetch is unavailable in workflows.

Cause: Workflows run in a sandboxed environment.

Solution:

import { fetch } from "workflow";

export async function myWorkflow() {
  "use workflow";
  globalThis.fetch = fetch;  // Required for libraries (AI SDK)

  const response = await fetch("https://api.example.com/data");
  const data = await response.json();
}

timeout-in-workflow

Error: setTimeout or setInterval unavailable.

Cause: These are non-deterministic and break replay.

Solution:

import { sleep } from "workflow";

// Instead of setTimeout
await sleep("5s");

// Polling pattern instead of setInterval
while (true) {
  const status = await checkStatus(id);
  if (status === "ready") break;
  await sleep("30s");
}

serialization-failed

Error: Data cannot be serialized.

Cause: Functions, class instances, Symbols, WeakMap/WeakSet.

Solution:

// BAD: Function in data
const config = { callback: () => console.log("hi") };

// GOOD: Configuration data, logic in steps
const config = { logLevel: "info" };

async function processWithLogging(data: any, logLevel: string) {
  "use step";
  if (logLevel === "info") console.log(data);
  return process(data);
}

Supported types:

• Primitives: string, number, boolean, null, undefined, bigint
• Objects: plain objects, arrays, Date, RegExp, URL, URLSearchParams
• Collections: Map, Set, Headers
• Binary: ArrayBuffer, typed arrays (Uint8Array, etc.)
• Web: Request, Response, ReadableStream, WritableStream

start-invalid-workflow-function

Error: Invalid workflow function.

Cause: Missing "use workflow" directive or framework config.

Solution:

// 1. Add directive as FIRST statement
export async function myWorkflow(input: string) {
  "use workflow";  // Must be first
  // ... workflow code
}

// 2. Check framework config
// next.config.ts
import { withWorkflow } from "workflow/next";
export default withWorkflow({ /* config */ });

// nitro.config.ts
export default defineNitroConfig({
  modules: ["workflow/nitro"],
});

node-js-module-in-workflow

Error: Node.js module unavailable in workflow.

Cause: Workflows run sandboxed for determinism.

Restricted: fs, path, http, https, net, dns, child_process, cluster, os, crypto, stream

Solution: Move Node.js code to step functions:

// BAD: Node.js in workflow
export async function badWorkflow() {
  "use workflow";
  const fs = require("fs");  // Error!
}

// GOOD: Node.js in step
async function readFile(path: string) {
  "use step";
  const fs = require("fs");
  return fs.readFileSync(path, "utf-8");
}

export async function goodWorkflow(filePath: string) {
  "use workflow";
  const content = await readFile(filePath);  // Works!
}

webhook-invalid-respond-with-value

Error: Invalid respondWith value.

Solution:

// Valid options:
const webhook = createWebhook();  // Default: auto 202
const webhook = createWebhook({ respondWith: "manual" });
const webhook = createWebhook({ respondWith: new Response("OK") });

webhook-response-not-sent

Error: Manual mode but respondWith() not called.

Solution:

const webhook = createWebhook({ respondWith: "manual" });
const request = await webhook;

await processRequest(request);

// MUST call respondWith in manual mode
await request.respondWith(new Response("OK", { status: 200 }));

Step Retry Exhaustion

Symptom: Step fails after max retries.

Default retry: 5-10 attempts with exponential backoff.

Solution: Handle expected failures explicitly:

async function callAPI() {
  "use step";

  const res = await fetch("...");

  // Don't retry on client errors
  if (res.status >= 400 && res.status < 500) {
    throw new FatalError(`Client error: ${res.status}`);
  }

  // Custom retry delay
  if (res.status === 429) {
    throw new RetryableError("Rate limited", { retryAfter: "5m" });
  }

  return res;
}

Streaming Errors

Symptom: Stream not working.

Solution:

// 1. Only use in steps, not workflow
async function streamData() {
  "use step";  // Required!

  const writer = getWritable();
  await writer.write(data);
  writer.releaseLock();  // 2. Release lock!
}

// 3. Close when done
const writer = getWritable();
await writer.write(lastData);
writer.releaseLock();
await getWritable().close();  // Signal completion

Hook Token Not Found

Error: Invalid or expired hook token.

Cause: Token doesn't exist, workflow completed, or token already consumed.

Solution:

import { resumeHook, getHookByToken } from "workflow/api";

// Check hook status before resuming
const hookInfo = await getHookByToken(token);
if (!hookInfo || hookInfo.status !== "pending") {
  throw new Error("Hook not available");
}

await resumeHook(token, data);

Debugging Tips

1. Use npx workflow web - Visual dashboard for runs 2. Check run status - getRun(runId).status 3. Review event log - Shows step executions, hooks, sleeps 4. Test locally first - Local World for development 5. Add logging in steps - Steps have full console access

Framework Setup

Table of Contents

• Next.js
• Express
• Hono
• Vite
• Astro
• Fastify
• Nitro
• Nuxt
• SvelteKit

Next.js

npm create next-app@latest my-app && cd my-app && npm i workflow

next.config.ts:

import { withWorkflow } from "workflow/next";
export default withWorkflow({ /* next config */ });

tsconfig.json (optional IDE support):

{ "compilerOptions": { "plugins": [{ "name": "workflow" }] } }

Middleware - Exclude workflow routes:

export const config = { matcher: ["/((?!.well-known/workflow).*)"] };

API Route:

// app/api/signup/route.ts
import { start } from "workflow/api";
import { handleUserSignup } from "@/workflows/user-signup";

export async function POST(req: Request) {
  const { email } = await req.json();
  await start(handleUserSignup, [email]);
  return Response.json({ message: "Workflow started" });
}

Express

mkdir my-app && cd my-app && npm init -y
npm i workflow express nitro rollup && npm i -D @types/express

nitro.config.ts:

import { defineNitroConfig } from "nitro/config";

export default defineNitroConfig({
  modules: ["workflow/nitro"],
  vercel: { entryFormat: "node" },
  routes: { "/**": { handler: "./src/index.ts", format: "node" } },
});

src/index.ts:

import express from "express";
import { start } from "workflow/api";

const app = express();
app.use(express.json());

app.post("/api/signup", async (req, res) => {
  await start(handleUserSignup, [req.body.email]);
  return res.json({ message: "Workflow started" });
});

export default app;

Hono

npm create hono@latest my-app -- --template=nodejs
cd my-app && npm i workflow nitro rollup

nitro.config.ts:

import { defineConfig } from "nitro";

export default defineConfig({
  modules: ["workflow/nitro"],
  routes: { "/**": "./src/index.ts" }
});

Vite

npm create vite@latest my-app -- --template react-ts
cd my-app && npm i workflow nitro rollup

vite.config.ts:

import { defineConfig } from "vite";
import { nitro } from "nitro/plugin";
import { workflow } from "workflow/vite";

export default defineConfig({
  plugins: [nitro({ serverDir: "." }), workflow()],
});

Astro

npm create astro@latest my-app -- --template minimal --install --yes
cd my-app && npm i workflow

astro.config.mjs:

import { defineConfig } from "astro/config";
import { workflow } from "workflow/astro";

export default defineConfig({ integrations: [workflow()] });

src/pages/api/signup.ts:

import type { APIRoute } from "astro";
import { start } from "workflow/api";

export const POST: APIRoute = async ({ request }) => {
  const { email } = await request.json();
  await start(handleUserSignup, [email]);
  return Response.json({ message: "Workflow started" });
};

Fastify

mkdir my-app && cd my-app && npm init -y
npm i workflow fastify nitro rollup && npm i -D @types/node typescript

nitro.config.ts:

import { defineNitroConfig } from "nitro/config";
export default defineNitroConfig({ modules: ["workflow/nitro"] });

Nitro

npx create-nitro-app my-app && cd my-app && npm i workflow

nitro.config.ts:

export default defineConfig({
  serverDir: "./server",
  modules: ["workflow/nitro"],
});

Nuxt

npm create nuxt@latest my-app && cd my-app && npm i workflow

nuxt.config.ts:

export default defineNuxtConfig({
  modules: ["workflow/nuxt"],
  compatibilityDate: "latest",
});

SvelteKit

npx sv create my-app --template=minimal --types=ts --no-add-ones
cd my-app && npm i workflow

vite.config.ts:

import { sveltekit } from "@sveltejs/kit/vite";
import { workflowPlugin } from "workflow/sveltekit";

export default { plugins: [sveltekit(), workflowPlugin()] };

Run and Inspect

npm run dev
npx workflow web  # Dashboard to inspect runs

HTTP Endpoints (Auto-generated)

Advanced Workflow Patterns

Table of Contents

• Hooks
• Webhooks
• Control Flow
• Streaming
• Error Handling
• Real-World Examples

Hooks

Basic Hook

import { createHook } from "workflow";

export async function approvalWorkflow() {
  "use workflow";

  const hook = createHook<{ approved: boolean; comment: string }>();
  console.log("Send approval to:", hook.token);

  const result = await hook;  // Pauses until resumed
  return result.approved ? "Approved" : "Rejected";
}

Custom Deterministic Tokens

Reconstruct tokens from external data:

const hook = createHook<SlackMessage>({
  token: `slack:${channelId}:${threadTs}`
});

// External handler reconstructs token:
await resumeHook(`slack:${channelId}:${threadTs}`, slackEvent);

Multiple Events (AsyncIterable)

export async function dataCollectionWorkflow() {
  "use workflow";

  const hook = createHook<{ value: number; done?: boolean }>();
  const values: number[] = [];

  for await (const payload of hook) {
    values.push(payload.value);
    if (payload.done) break;
  }

  return { total: values.reduce((a, b) => a + b, 0), count: values.length };
}

Type-Safe Hooks with Schema

import { defineHook } from "workflow";
import { z } from "zod";

const approvalHook = defineHook({
  schema: z.object({
    requestId: z.string(),
    approved: z.boolean(),
    approvedBy: z.string(),
    comment: z.string().transform((v) => v.trim()),
  }),
});

export async function documentApproval(documentId: string) {
  "use workflow";

  const hook = approvalHook.create({ token: `approval:${documentId}` });
  const approval = await hook;

  return approval.approved;
}

// Resume with validation (throws on invalid data)
await approvalHook.resume(`approval:${documentId}`, approvalData);

Supported validators: Zod, Valibot, ArkType, Effect Schema

Webhooks

Basic Webhook

import { createWebhook } from "workflow";

export async function paymentWorkflow(orderId: string) {
  "use workflow";

  const webhook = createWebhook();
  await notifyPaymentProvider(orderId, webhook.url);

  const request = await webhook;  // Pauses until HTTP POST
  const data = await request.json();

  if (data.status === "completed") {
    await fulfillOrder(orderId);
  }
}

Manual Response Mode

const webhook = createWebhook({ respondWith: "manual" });
const request = await webhook;

await processRequest(request);

// Must respond when using manual mode
await request.respondWith(new Response("OK", { status: 200 }));

Automatic Custom Response

const webhook = createWebhook({
  respondWith: new Response("Received", { status: 200 })
});

Multiple Webhook Events

for await (const request of webhook) {
  const data = await request.json();
  await processEvent(data);
  if (data.final) break;
}

Hooks vs Webhooks

Control Flow

Parallel Steps

export async function parallelWorkflow(urls: string[]) {
  "use workflow";

  // Execute steps in parallel
  const results = await Promise.all(
    urls.map(url => fetchAndProcess(url))
  );

  return results;
}

Conditional Branching

export async function conditionalWorkflow(userId: string) {
  "use workflow";

  const user = await getUser(userId);

  if (user.tier === "premium") {
    await processPremium(user);
  } else {
    await processStandard(user);
  }
}

Polling Pattern

export async function pollUntilReady(jobId: string) {
  "use workflow";

  while (true) {
    const status = await checkStatus(jobId);
    if (status === "ready") return await getResult(jobId);
    if (status === "failed") throw new FatalError("Job failed");
    await sleep("30s");
  }
}

Retry with Backoff

export async function retryWorkflow(data: any) {
  "use workflow";

  let attempts = 0;
  const maxAttempts = 5;

  while (attempts < maxAttempts) {
    try {
      return await processData(data);
    } catch (error) {
      attempts++;
      if (attempts >= maxAttempts) throw error;
      await sleep(`${Math.pow(2, attempts)}s`);  // Exponential backoff
    }
  }
}

Streaming

Namespaced Streams

async function processWithLogs(data: any) {
  "use step";

  const logsWriter = getWritable({ namespace: "logs" });
  const metricsWriter = getWritable({ namespace: "metrics" });

  await logsWriter.write({ level: "info", msg: "Starting" });
  await metricsWriter.write({ event: "start", timestamp: Date.now() });

  const result = await process(data);

  await logsWriter.write({ level: "info", msg: "Done" });
  logsWriter.releaseLock();
  metricsWriter.releaseLock();

  return result;
}

Binary Data

async function writeBinary(data: string) {
  "use step";

  const writer = getWritable();
  const encoder = new TextEncoder();
  await writer.write(encoder.encode(data));
  writer.releaseLock();
}

Error Handling

Categorizing Errors

async function callExternalAPI(url: string) {
  "use step";

  const res = await fetch(url);

  // Permanent failures
  if (res.status === 401 || res.status === 403) {
    throw new FatalError(`Auth error: ${res.status}`);
  }

  // Retryable with delay
  if (res.status === 429) {
    const retryAfter = res.headers.get("Retry-After");
    throw new RetryableError("Rate limited", {
      retryAfter: retryAfter ? parseInt(retryAfter) * 1000 : "5m"
    });
  }

  // Server errors - auto retry
  if (res.status >= 500) {
    throw new Error(`Server error: ${res.status}`);
  }

  return res;
}

Workflow-Level Error Handling

export async function resilientWorkflow(input: any) {
  "use workflow";

  try {
    const result = await riskyOperation(input);
    return { success: true, result };
  } catch (error) {
    await notifyAdmin(error);
    await cleanup(input);
    return { success: false, error: error.message };
  }
}

Real-World Examples

Onboarding Drip Campaign

export async function userOnboarding(email: string) {
  "use workflow";

  await sendWelcomeEmail(email);
  await sleep("3 days");
  await sendTipsEmail(email);
  await sleep("7 days");
  await sendFeedbackRequest(email);
}

Churn Prevention

export async function churnPrevention(userId: string) {
  "use workflow";

  await sleep("7 days");
  if (!await isActive(userId)) {
    await sendReEngagementEmail(userId);
    await sleep("3 days");
    if (!await isActive(userId)) {
      await offerDiscount(userId);
    }
  }
}

Order Processing

export async function processOrder(orderId: string) {
  "use workflow";

  const payment = await chargeCard(orderId);

  // Wait for warehouse webhook
  const webhook = createWebhook();
  await notifyWarehouse(orderId, webhook.url);
  const shipment = await webhook;

  await notifyCustomer(orderId, await shipment.json());
  return { orderId, status: "shipped" };
}

Multi-Step Approval

export async function multiApproval(requestId: string) {
  "use workflow";

  const managers = await getApprovers(requestId);

  for (const manager of managers) {
    const hook = createHook<{ approved: boolean }>();
    await notifyManager(manager, requestId, hook.token);

    const decision = await hook;
    if (!decision.approved) {
      return { status: "rejected", rejectedBy: manager };
    }
  }

  await executeRequest(requestId);
  return { status: "approved" };
}