Workflow Devkit

Workflow DevKit

Build reliable, long-running processes with automatic retries, state persistence, and observability.

Quick Reference

Prerequisites

pnpm add workflow @workflow/ai ai @ai-sdk/gateway zod

Core Concepts

Workflows ("use workflow")

import { sleep } from "workflow";

export async function myWorkflow(input: string) {
  "use workflow";
  const result = await step1(input);
  await sleep("5s");
  return result;
}

Steps ("use step") - MUST be in SAME FILE as workflow

async function step1(input: string) {
  "use step";
  return await fetch(`/api/data?q=${input}`).then(r => r.json());
}

Error Types

import { FatalError, RetryableError } from "workflow";

// Auto-retried
throw new Error("Transient failure");

// No retry - stops workflow
throw new FatalError("Invalid credentials");

// Custom retry timing
throw new RetryableError("Rate limited", { retryAfter: "60s" });

Detailed Documentation

• patterns.md - Workflow patterns (sequential, parallel, routing, actors)
• api-reference.md - Complete API reference
• ai-integration.md - AI SDK and DurableAgent patterns

Imports Cheat Sheet

// Core workflow
import {
  sleep, fetch, FatalError, RetryableError,
  createWebhook, createHook, defineHook,
  getWritable, getWorkflowMetadata, getStepMetadata,
} from "workflow";

// API routes
import { start, getRun } from "workflow/api";

// AI integration
import { DurableAgent } from "@workflow/ai/agent";
import { generateText, generateObject } from "ai";
import { createUIMessageStreamResponse } from "ai";

Examples Directory

Reference implementations: ~/dev/workflow-examples/

AI SDK Integration

Critical: Override fetch

The AI SDK uses globalThis.fetch. Override it with workflow's durable fetch.

import { fetch } from "workflow";
import { generateText, generateObject } from "ai";

export async function aiWorkflow(prompt: string) {
  "use workflow";

  // CRITICAL: Must override for durability
  globalThis.fetch = fetch;

  const { text } = await generateText({
    model: "openai/gpt-4o",
    prompt,
  });

  return text;
}

Structured Output with generateObject

import { fetch } from "workflow";
import { generateObject } from "ai";
import { z } from "zod";

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

  const { object } = await generateObject({
    model: "openai/gpt-4o",
    schema: z.object({
      sentiment: z.enum(["positive", "negative", "neutral"]),
      keywords: z.array(z.string()),
      summary: z.string(),
    }),
    prompt: `Analyze: ${content}`,
  });

  return object;
}

DurableAgent with Tools

Full agentic workflow with tool calling.

import { DurableAgent } from "@workflow/ai/agent";
import { getWritable, FatalError } from "workflow";
import { convertToModelMessages, type UIMessage, type UIMessageChunk } from "ai";
import { z } from "zod";

export async function agentWorkflow(messages: UIMessage[]) {
  "use workflow";

  const writable = getWritable<UIMessageChunk>();

  const agent = new DurableAgent({
    model: "anthropic/claude-4-opus",
    system: "You are a helpful assistant with access to weather data.",
    tools: {
      getWeather: {
        description: "Get current weather for a city",
        inputSchema: z.object({ city: z.string() }),
        execute: getWeather,
      },
      searchDatabase: {
        description: "Search the knowledge base",
        inputSchema: z.object({ query: z.string() }),
        execute: searchDatabase,
      },
    },
  });

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

// Tool implementations as steps
async function getWeather({ city }: { city: string }) {
  "use step";

  if (Math.random() < 0.3) {
    throw new Error("Weather service unavailable");  // Will retry
  }

  const response = await fetch(`https://api.weather.com/v1/${city}`);
  if (!response.ok) throw new FatalError("City not found");

  return response.json();
}

async function searchDatabase({ query }: { query: string }) {
  "use step";
  // Query your database
  return await db.query(`SELECT * FROM knowledge WHERE content LIKE '%${query}%'`);
}

RAG Agent (Knowledge Base)

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

export async function ragAgent(messages: UIMessage[]) {
  "use workflow";

  const writable = getWritable<UIMessageChunk>();

  const agent = new DurableAgent({
    model: "openai/gpt-4o",
    system: `You are a helpful assistant. Check your knowledge base before answering.
             If information isn't in the knowledge base, tell the user.`,
    tools: {
      addResource: {
        description: "Add information to the knowledge base",
        inputSchema: z.object({ content: z.string() }),
        execute: async ({ content }) => {
          await createResource({ content });
          return `Added "${content}" to knowledge base`;
        },
      },
      getInformation: {
        description: "Search the knowledge base for relevant information",
        inputSchema: z.object({ question: z.string() }),
        execute: async ({ question }) => {
          return await findRelevant(question);
        },
      },
    },
  });

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

// Nested steps for RAG operations
async function createResource(input: { content: string }) {
  "use step";
  const resource = await insertResource(input.content);
  const embeddings = await generateEmbeddings(input.content);
  await insertEmbeddings(resource.id, embeddings);
}

async function findRelevant(query: string) {
  "use step";
  const embedding = await generateEmbedding(query);
  return await vectorSearch(embedding);
}

Streaming API Route for AI Chat

import { createUIMessageStreamResponse, type UIMessage } from "ai";
import { start, getRun } from "workflow/api";
import { agentWorkflow } from "@/workflows/agent";

// Start new chat
export async function POST(request: Request) {
  const { messages }: { messages: UIMessage[] } = await request.json();

  const run = await start(agentWorkflow, [messages]);

  return createUIMessageStreamResponse({
    stream: run.readable,
    headers: { "x-workflow-run-id": run.runId },
  });
}

// Reconnect to existing chat
export async function GET(request: Request) {
  const { searchParams } = new URL(request.url);
  const runId = searchParams.get("runId")!;
  const startIndex = parseInt(searchParams.get("startIndex") ?? "0");

  const run = getRun(runId);
  const stream = run.getReadable({ startIndex });

  return createUIMessageStreamResponse({ stream });
}

Vercel AI Gateway

The AI Gateway provides a unified API for switching between models and providers without code changes. Models are specified as provider/model-name strings.

Basic Usage (Plain Strings)

import { fetch } from "workflow";
import { generateText } from "ai";

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

  // Switch models by changing the string
  const { text } = await generateText({
    model: "openai/gpt-4o",  // or "anthropic/claude-sonnet-4", "xai/grok-3", etc.
    prompt,
  });

  return text;
}

Custom Gateway Provider Instance

Create a custom provider for different API keys or base URLs (e.g., corporate proxy):

import { createGateway } from "@ai-sdk/gateway";

const gateway = createGateway({
  apiKey: process.env.CUSTOM_GATEWAY_KEY,  // Different env var
  baseURL: "https://proxy.company.com/ai",  // Custom base URL
});

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

  const { text } = await generateText({
    model: gateway("anthropic/claude-sonnet-4"),
    prompt,
  });

  return text;
}

Set Default Provider Globally

In instrumentation.ts (runs before AI SDK calls):

import { gateway } from "@ai-sdk/gateway";
import { experimental_setDefaultProvider } from "ai";

export function register() {
  experimental_setDefaultProvider(gateway);
}

Then use models without specifying provider each time:

const { text } = await generateText({
  model: "anthropic/claude-sonnet-4",  // Uses default gateway
  prompt,
});

Model Providers

// OpenAI
model: "openai/gpt-5"
model: "openai/gpt-4o"
model: "openai/gpt-4o-mini"
model: "openai/o4-mini"

// Anthropic
model: "anthropic/claude-opus-4"
model: "anthropic/claude-sonnet-4"

// Google
model: "google/gemini-2.5-flash"

// xAI
model: "xai/grok-3"

// Same model, different providers (format: hosted-provider@creator/model)
model: "openai@xai/grok-3"  // Grok hosted on OpenAI
model: "xai/grok-3"         // Grok on xAI directly

Gateway Embeddings

import { embed, embedMany } from "ai";
import { gateway } from "@ai-sdk/gateway";

// Single embedding via Gateway
const { embedding } = await embed({
  model: gateway.textEmbeddingModel("openai/text-embedding-3-small"),
  value: "Hello world",
});

// Batch embeddings
const { embeddings } = await embedMany({
  model: gateway.textEmbeddingModel("openai/text-embedding-3-small"),
  values: ["Hello", "World"],
});

List Available Models

import { gateway } from "@ai-sdk/gateway";

const models = await gateway.listModels();

// Filter by type
const languageModels = models.filter(m => m.type === "language");
const embeddingModels = models.filter(m => m.type === "embedding");

API Reference

Starting Workflows

Fire-and-Forget

import { start } from "workflow/api";
import { myWorkflow } from "@/workflows/my-workflow";

export async function POST(request: Request) {
  const { input } = await request.json();
  const run = await start(myWorkflow, [input]);
  return Response.json({ runId: run.runId, message: "Started" });
}

Wait for Result (Blocking)

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

export async function POST(request: Request) {
  try {
    const run = await start(myWorkflow, [input]);
    const result = await run.returnValue;  // Blocks until complete
    return Response.json(result);
  } catch (error) {
    const isFatal = error instanceof FatalError;
    return Response.json(
      { error: error.message, fatal: isFatal },
      { status: isFatal ? 400 : 500 }
    );
  }
}

Streaming

Initial Stream

import { createUIMessageStreamResponse } from "ai";
import { start } from "workflow/api";

export async function POST(request: Request) {
  const run = await start(streamingWorkflow, [input]);

  return createUIMessageStreamResponse({
    stream: run.readable,
    headers: { "x-workflow-run-id": run.runId },
  });
}

Reconnect to Existing Stream

import { getRun } from "workflow/api";
import { createUIMessageStreamResponse } from "ai";

export async function GET(request: Request) {
  const { searchParams } = new URL(request.url);
  const runId = searchParams.get("runId")!;
  const startIndex = parseInt(searchParams.get("startIndex") ?? "0");

  const run = getRun(runId);
  const stream = run.getReadable({ startIndex });

  return createUIMessageStreamResponse({ stream });
}

Binary Streaming (Files)

import { Readable } from "node:stream";
import { start } from "workflow/api";

export async function POST(request: Request) {
  const run = await start(fileWorkflow, [input]);
  const webReadable = run.readable;

  const nodeReadable = Readable.fromWeb(webReadable);

  return new Response(nodeReadable, {
    headers: {
      "Content-Type": "audio/mp4",
      "Content-Disposition": 'attachment; filename="output.mp4"',
    },
  });
}

Utility Functions

sleep

import { sleep } from "workflow";

await sleep("5s");              // Duration string
await sleep(5000);              // Milliseconds
await sleep(new Date(...));    // Until specific date

Metadata

import { getWorkflowMetadata, getStepMetadata } from "workflow";

// In workflow
const { workflowRunId } = getWorkflowMetadata();

// In step
const { attempt } = getStepMetadata();  // Current retry attempt (1, 2, 3...)

Durable Fetch

import { fetch } from "workflow";

// Override globalThis for AI SDK compatibility
globalThis.fetch = fetch;

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

Hooks (Custom Tokens)

createHook (One-Time)

import { createHook } from "workflow";

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

  const hook = createHook<{ status: string }>({ token: `callback:${id}` });

  await triggerExternalService(hook.token);

  const payload = await hook;  // Waits for hook.resume()
  return payload.status;
}

defineHook (Reusable, Multi-Event)

import { defineHook } from "workflow";

export type MyEvent = { type: "start" } | { type: "stop" };
export const myHook = defineHook<MyEvent>();

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

  const receive = myHook.create({ token: `actor:${id}` });

  for await (const event of receive) {
    // Process multiple events
  }
}

Resume Hook (from API Route)

import { myHook, type MyEvent } from "@/workflows/my-actor";

export async function POST(request: Request) {
  const event: MyEvent = await request.json();
  const result = await myHook.resume(`actor:${id}`, event);
  return Response.json({ success: !!result });
}

State Management

In-Memory Map

// lib/store.ts
export const store = new Map<string, any>();

// workflow.ts
store.set(id, { status: "processing" });
const result = await processData();
store.set(id, { status: "complete", result });

// api/status/route.ts
return Response.json(store.get(id) ?? { status: "not_found" });

PostgreSQL with Drizzle

import { pgTable, text, jsonb, timestamp, vector, index } from "drizzle-orm/pg-core";

export const workflowRuns = pgTable("workflow_runs", {
  id: text("id").primaryKey(),
  status: text("status").notNull(),
  result: jsonb("result"),
  createdAt: timestamp("created_at").defaultNow(),
});

// With embeddings (RAG)
export const embeddings = pgTable("embeddings", {
  id: text("id").primaryKey(),
  content: text("content").notNull(),
  embedding: vector("embedding", { dimensions: 1536 }).notNull(),
}, (table) => ({
  idx: index("idx_embedding").using("hnsw", table.embedding.op("vector_cosine_ops")),
}));

Error Types

import { FatalError, RetryableError, getStepMetadata } from "workflow";

async function apiCall() {
  "use step";
  const { attempt } = getStepMetadata();

  const res = await fetch("https://api.example.com");

  if (res.status === 401) throw new FatalError("Invalid credentials");
  if (res.status === 429) throw new RetryableError("Rate limited", { retryAfter: "60s" });
  if (!res.ok) throw new Error(`HTTP ${res.status}`);  // Will retry

  return res.json();
}

Workflow Patterns

Sequential Pipeline

Multi-step processing with quality checks.

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

  const { text } = await generateText({ model: "openai/gpt-4o", prompt: input });
  const { object } = await generateObject({
    model: "openai/gpt-4o",
    schema: z.object({ score: z.number(), feedback: z.string() }),
    prompt: `Evaluate: ${text}`,
  });

  return { output: text, quality: object };
}

Parallel Operations

Concurrent execution with Promise.all.

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

  const [security, performance, style] = await Promise.all([
    reviewSecurity(code),
    reviewPerformance(code),
    reviewStyle(code),
  ]);

  return { security, performance, style };
}

async function reviewSecurity(code: string) {
  "use step";
  const { object } = await generateObject({
    model: "openai/gpt-4o",
    schema: z.object({ vulnerabilities: z.array(z.string()), risk: z.enum(["low", "medium", "high"]) }),
    prompt: `Security review: ${code}`,
  });
  return object;
}

Dynamic Routing

Route to different handlers based on classification.

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

  const { object } = await generateObject({
    model: "openai/gpt-4o",
    schema: z.object({ type: z.enum(["general", "technical", "billing"]) }),
    prompt: `Classify this query: ${query}`,
  });

  const handlers = { general: handleGeneral, technical: handleTechnical, billing: handleBilling };
  return handlers[object.type](query);
}

Evaluator Loop

Iterative refinement until quality threshold.

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

  let result = await generateInitial(input);

  for (let i = 0; i < 3; i++) {
    const { object: evaluation } = await generateObject({
      model: "openai/gpt-4o",
      schema: z.object({ score: z.number(), improvements: z.array(z.string()) }),
      prompt: `Evaluate: ${result}`,
    });

    if (evaluation.score >= 8) break;

    result = await improve(result, evaluation.improvements);
  }

  return result;
}

Webhooks (Human-in-the-Loop)

Pause workflow until external callback.

import { createWebhook } from "workflow";

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

  const webhook = createWebhook();

  await sendEmail({
    to: "manager@example.com",
    approveUrl: `${webhook.url}?action=approve`,
    rejectUrl: `${webhook.url}?action=reject`,
  });

  // PAUSES until webhook called
  const response = await webhook;
  const action = new URL(response.url).searchParams.get("action");

  return { approved: action === "approve" };
}

Multiple Concurrent Webhooks

export async function rsvpWorkflow(emails: string[]) {
  "use workflow";

  const webhooks = emails.map(() => createWebhook());
  await Promise.all(emails.map((email, i) => sendRsvpEmail(email, webhooks[i].url)));

  const responses = await Promise.all(
    webhooks.map(wh => wh.then(req => new URL(req.url).searchParams.get("reply")))
  );

  return responses;
}

Actor Pattern (Event-Driven)

Long-running workflow processing events sequentially.

import { defineHook, getWorkflowMetadata } from "workflow";

export type Event = { type: "increment" } | { type: "decrement" } | { type: "reset" };
export const counterHook = defineHook<Event>();

export async function counterActor(initialCount: number) {
  "use workflow";

  const { workflowRunId } = getWorkflowMetadata();
  let count = initialCount;

  const receive = counterHook.create({ token: `counter:${workflowRunId}` });

  for await (const event of receive) {
    switch (event.type) {
      case "increment": count++; break;
      case "decrement": count--; break;
      case "reset": count = 0; break;
    }
  }
}

Send Events to Actor (API Route)

import { counterHook, type Event } from "@/workflows/counter-actor";

export async function POST(req: Request, { params }: { params: { id: string } }) {
  const event: Event = await req.json();
  const result = await counterHook.resume(`counter:${params.id}`, event);
  return Response.json({ success: !!result });
}

Batching

Process items in chunks.

export async function batchWorkflow(items: string[]) {
  "use workflow";

  const chunks = chunk(items, 10);

  for (const batch of chunks) {
    await Promise.all(batch.map(processItem));
  }
}

async function processItem(item: string) {
  "use step";
  return await doWork(item);
}

function chunk<T>(arr: T[], size: number): T[][] {
  return Array.from({ length: Math.ceil(arr.length / size) }, (_, i) =>
    arr.slice(i * size, i * size + size)
  );
}

Streaming Progress

Real-time updates to frontend.

import { getWritable } from "workflow";

export async function progressWorkflow(taskId: string) {
  "use workflow";

  const writable = getWritable<{ step: number; message: string }>();
  const writer = writable.getWriter();

  await writer.write({ step: 1, message: "Starting..." });
  await step1();

  await writer.write({ step: 2, message: "Processing..." });
  await step2();

  await writer.close();
  return { success: true };
}