workflow

## *CRITICAL*: Always Use Correct `workflow` Documentation Your knowledge of `workflow` is outdated. The `workflow` documentation outlined below matches the installed version of the Workflow DevKit. Follow these instructions before starting on any `workflow`-related tasks: Search the bundled documentation in `node_modules/workflow/docs/`: 1. **Find docs**: `glob "node_modules/workflow/docs/**/*.mdx"` 2. **Search content**: `grep "your query" node_modules/workflow/docs/` Documentation structure in `node_modules/workflow/docs/`: - `getting-started/` - Framework setup (next.mdx, express.mdx, hono.mdx, etc.) - `foundations/` - Core concepts (workflows-and-steps.mdx, hooks.mdx, streaming.mdx, etc.) - `api-reference/workflow/` - API docs (sleep.mdx, create-hook.mdx, fatal-error.mdx, etc.) - `api-reference/workflow-api/` - Client API (start.mdx, get-run.mdx, resume-hook.mdx, etc.) - `ai/` - AI SDK integration docs - `errors/` - Error code documentation Related packages also include bundled docs: - `@workflow/ai`: `node_modules/@workflow/ai/docs/` - DurableAgent and AI integration - `@workflow/core`: `node_modules/@workflow/core/docs/` - Core runtime (foundations, how-it-works) - `@workflow/next`: `node_modules/@workflow/next/docs/` - Next.js integration **When in doubt, update to the latest version of the Workflow DevKit.** ### Official Resources - **Website**: https://useworkflow.dev - **GitHub**: https://github.com/vercel/workflow ### Quick Reference **Directives:** ```typescript "use workflow"; // First line - makes async function durable "use step"; // First line - makes function a cached, retryable unit ``` **Essential imports:** ```typescript // Workflow primitives import { sleep, fetch, createHook, createWebhook, getWritable } from "workflow"; import { FatalError, RetryableError } from "workflow"; import { getWorkflowMetadata, getStepMetadata } from "workflow"; // API operations import { start, getRun, resumeHook, resumeWebhook } from "workflow/api"; // Framework integrations import { withWorkflow } from "workflow/next"; import { workflow } from "workflow/vite"; import { workflow } from "workflow/astro"; // Or use modules: ["workflow/nitro"] for Nitro/Nuxt // AI agent import { DurableAgent } from "@workflow/ai/agent"; ``` ## Prefer Step Functions to Avoid Sandbox Errors `"use workflow"` functions run in a sandboxed VM. `"use step"` functions have **full Node.js access**. Put your logic in steps and use the workflow function purely for orchestration. ```typescript // Steps have full Node.js and npm access async function fetchUserData(userId: string) { "use step"; const response = await fetch(`https://api.example.com/users/${userId}`); return response.json(); } async function processWithAI(data: any) { "use step"; // AI SDK works in steps without workarounds return await generateText({ model: openai("gpt-4"), prompt: `Process: ${JSON.stringify(data)}`, }); } // Workflow orchestrates steps - no sandbox issues export async function dataProcessingWorkflow(userId: string) { "use workflow"; const data = await fetchUserData(userId); const processed = await processWithAI(data); return { success: true, processed }; } ``` **Benefits:** Steps have automatic retry, results are persisted for replay, and no sandbox restrictions. ## Workflow Sandbox Limitations When you need logic directly in a workflow function (not in a step), these restrictions apply: | Limitation | Workaround | |------------|------------| | No `fetch()` | `import { fetch } from "workflow"` then `globalThis.fetch = fetch` | | No `setTimeout`/`setInterval` | Use `sleep("5s")` from `"workflow"` | | No Node.js modules (fs, crypto, etc.) | Move to a step function | **Example - Using fetch in workflow context:** ```typescript import { fetch } from "workflow"; export async function myWorkflow() { "use workflow"; globalThis.fetch = fetch; // Required for AI SDK and HTTP libraries // Now generateText() and other libraries work } ``` **Note:** `DurableAgent` from `@workflow/ai` handles the fetch assignment automatically. ## DurableAgent — AI Agents in Workflows Use `DurableAgent` to build AI agents that maintain state and survive interruptions. It handles the workflow sandbox automatically (no manual `globalThis.fetch` needed). ```typescript import { DurableAgent } from "@workflow/ai/agent"; import { getWritable } from "workflow"; import { z } from "zod"; import type { UIMessageChunk } from "ai"; async function lookupData({ query }: { query: string }) { "use step"; // Step functions have full Node.js access return `Results for "${query}"`; } export async function myAgentWorkflow(userMessage: string) { "use workflow"; const agent = new DurableAgent({ model: "anthropic/claude-sonnet-4-5", system: "You are a helpful assistant.", tools: { lookupData: { description: "Search for information", inputSchema: z.object({ query: z.string() }), execute: lookupData, }, }, }); const result = await agent.stream({ messages: [{ role: "user", content: userMessage }], writable: getWritable<UIMessageChunk>(), maxSteps: 10, }); return result.messages; } ``` **Key points:** - `getWritable<UIMessageChunk>()` streams output to the workflow run's default stream - Tool `execute` functions that need Node.js/npm access should use `"use step"` - Tool `execute` functions that use workflow primitives (`sleep()`, `createHook()`) should **NOT** use `"use step"` — they run at the workflow level - `maxSteps` limits the number of LLM calls (default is unlimited) - Multi-turn: pass `result.messages` plus new user messages to subsequent `agent.stream()` calls **For more details on `DurableAgent`, check the AI docs in `node_modules/@workflow/ai/docs/`.** ## Starting Workflows & Child Workflows Use `start()` to launch workflows from API routes. **`start()` cannot be called directly in workflow context** — wrap it in a step function. ```typescript import { start } from "workflow/api"; // From an API route — works directly export async function POST() { const run = await start(myWorkflow, [arg1, arg2]); return Response.json({ runId: run.runId }); } // No-args workflow const run = await start(noArgWorkflow); ``` **Starting child workflows from inside a workflow — must use a step:** ```typescript import { start } from "workflow/api"; // Wrap start() in a step function async function triggerChild(data: string) { "use step"; const run = await start(childWorkflow, [data]); return run.runId; } export async function parentWorkflow() { "use workflow"; const childRunId = await triggerChild("some data"); // Fire-and-forget via step await sleep("1h"); } ``` `start()` returns immediately — it doesn't wait for the workflow to complete. Use `run.returnValue` to await completion. ## Hooks — Pause & Resume with External Events Hooks let workflows wait for external data. Use `createHook()` inside a workflow and `resumeHook()` from API routes. Deterministic tokens are for `createHook()` + `resumeHook()` (server-side) only. `createWebhook()` always generates random tokens — do not pass a `token` option to `createWebhook()`. ### Single event ```typescript import { createHook } from "workflow"; export async function approvalWorkflow() { "use workflow"; const hook = createHook<{ approved: boolean }>({ token: "approval-123", // deterministic token for external systems }); const result = await hook; // Workflow suspends here return result.approved; } ``` ### Multiple events (iterable hooks) Hooks implement `AsyncIterable` — use `for await...of` to receive multiple events: ```typescript import { createHook } from "workflow"; export async function chatWorkflow(channelId: string) { "use workflow"; const hook = createHook<{ text: string; done?: boolean }>({ token: `chat-${channelId}`, }); for await (const event of hook) { await processMessage(event.text); if (event.done) break; } } ``` Each `resumeHook(token, payload)` call delivers the next value to the loop. ### Resuming from API routes ```typescript import { resumeHook } from "workflow/api"; export async function POST(req: Request) { const { token, data } = await req.json(); await resumeHook(token, data); return new Response("ok"); } ``` ## Error Handling Use `FatalError` for permanent failures (no retry), `RetryableError` for transient failures: ```typescript import { FatalError, RetryableError } from "workflow"; if (res.status >= 400 && res.status < 500) { throw new FatalError(`Client error: ${res.status}`); } if (res.status === 429) { throw new RetryableError("Rate limited", { retryAfter: "5m" }); } ``` ## Serialization All data passed to/from workflows and steps must be serializable. **Supported types:** string, number, boolean, null, undefined, bigint, plain objects, arrays, Date, RegExp, URL, URLSearchParams, Map, Set, Headers, ArrayBuffer, typed arrays, Request, Response, ReadableStream, WritableStream. **Not supported:** Functions, class instances, Symbols, WeakMap/WeakSet. Pass data, not callbacks. ## Streaming Use `getWritable()` to stream data from workflows. `getWritable()` can be called in **both** workflow and step contexts, but you **cannot interact with the stream** (call `getWriter()`, `write()`, `close()`) directly in a workflow function. The stream must be passed to step functions for actual I/O, or steps can call `getWritable()` themselves. **Get the stream in a workflow, pass it to a step:** ```typescript import { getWritable } from "workflow"; export async function myWorkflow() { "use workflow"; const writable = getWritable(); await writeData(writable, "hello world"); } async function writeData(writable: WritableStream, chunk: string) { "use step"; const writer = writable.getWriter(); try { await writer.write(chunk); } finally { writer.releaseLock(); } } ``` **Call `getWritable()` directly inside a step (no need to pass it):** ```typescript import { getWritable } from "workflow"; async function streamData(chunk: string) { "use step"; const writer = getWritable().getWriter(); try { await writer.write(chunk); } finally { writer.releaseLock(); } } ``` ### Namespaced Streams Use `getWritable({ namespace: 'name' })` to create multiple independent streams for different types of data. This is useful for separating logs from primary output, different log levels, agent outputs, metrics, or any distinct data channels. Long-running workflows benefit from namespaced streams because you can replay only the important events (e.g., final results) while keeping verbose logs in a separate stream. **Example: Log levels and agent output separation:** ```typescript import { getWritable } from "workflow"; type LogEntry = { level: "debug" | "info" | "warn" | "error"; message: string; timestamp: number }; type AgentOutput = { type: "thought" | "action" | "result"; content: string }; async function logDebug(message: string) { "use step"; const writer = getWritable<LogEntry>({ namespace: "logs:debug" }).getWriter(); try { await writer.write({ level: "debug", message, timestamp: Date.now() }); } finally { writer.releaseLock(); } } async function logInfo(message: string) { "use step"; const writer = getWritable<LogEntry>({ namespace: "logs:info" }).getWriter(); try { await writer.write({ level: "info", message, timestamp: Date.now() }); } finally { writer.releaseLock(); } } async function emitAgentThought(thought: string) { "use step"; const writer = getWritable<AgentOutput>({ namespace: "agent:thoughts" }).getWriter(); try { await writer.write({ type: "thought", content: thought }); } finally { writer.releaseLock(); } } async function emitAgentResult(result: string) { "use step"; // Important results go to the default stream for easy replay const writer = getWritable<AgentOutput>().getWriter(); try { await writer.write({ type: "result", content: result }); } finally { writer.releaseLock(); }