Upstash Workflow Js

Upstash Workflow SDK

Quick Start

The Upstash Workflow SDK lets you expose serverless workflow endpoints and run them reliably using QStash under the hood.

Install:

npm install @upstash/workflow

Define a simple workflow endpoint:

import { serve } from "@upstash/workflow";

export const { POST } = serve(async (context) => {
  await context.run("step-1", () => console.log("step 1"));
  await context.run("step-2", () => console.log("step 2"));
});

Trigger it from your backend:

import { Client } from "@upstash/workflow";

const client = new Client({ token: process.env.QSTASH_TOKEN! });
await client.trigger({ url: "https://your-app.com/api/workflow" });

Other Skill Files

These files contain the full documentation. Use them for details, patterns, and advanced behavior.

• basics:
• basics/serve – How to expose workflow endpoints.
• basics/context – Full API for workflow context (steps, waits, webhooks, events, invoke, etc.).
• basics/client – Using the Workflow client to trigger, cancel, inspect, and notify runs.
• features:
• features/invoke – Cross‑workflow invocation.
• features/reliability – Retries, failure callbacks, and DLQ.
• features/flow-control – Rate limits, concurrency, and parallelism.
• features/wait-for-event – Notify and wait-for-event patterns.
• features/webhooks – Webhook creation and consumption.
• how to:
• how-to/local-dev – Local QStash dev server and tunneling.
• how-to/realtime – Realtime and human‑in‑the‑loop workflows.
• how-to/migrations – Migrating workflows safely.
• how-to/middleware – Adding middleware to workflows.
• other files:
• rest-api – Low-level REST endpoints for interacting with QStash/Workflow.
• troubleshooting – Common debugging and environment issues.
• agents – Using Workflow with agents, orchestrators, and automation patterns.

Workflow Agents

This Skill provides guidance for defining and orchestrating agents, tools, models, and tasks inside Upstash Workflow using TypeScript. It explains core concepts, common patterns, workflow configuration, and pitfalls.

Quick Overview

Use this Skill when:

• Building agents with tools and backgrounds
• Creating multi-step or multi-agent tasks
• Implementing patterns like evaluator-optimizer or prompt chaining
• Integrating LangChain or AI SDK tools
• Running workflows reliably inside Upstash Workflow

The Workflow Agents API centers around four elements:

• Models — LLM providers (OpenAI, Anthropic, or any AI SDK provider)
• Tools — Functions agents can call
• Agents — LLM instances with background, maxSteps, and tools
• Tasks — Executable prompts assigned to one or more agents

---

Defining Models

A model defines which provider the agent uses and includes optional configuration for retries, timeouts, or rate‑limits.

const agents = agentWorkflow(context);

// Basic OpenAI model
const model = agents.openai("gpt-3.5-turbo");

// OpenAI-compatible provider
const deepseek = agents.openai("deepseek-chat", {
  baseURL: "https://api.deepseek.com",
  apiKey: process.env.DEEPSEEK_API_KEY,
});

// AI SDK provider (ex: Anthropic)
const anthropic = agents.AISDKModel({
  context,
  provider: createAnthropic, // imported from @ai-sdk/anthropic
  providerParams: { apiKey: process.env.ANTHROPIC_KEY },
  agentCallParams: {
    timeout: 1000,
    retries: 0,
  },
});

Key parameters:

• callSettings / agentCallParams — timeout, retries, flow control
• provider / providerParams — when using AI SDK providers

Pitfall: If you use an OpenAI-compatible provider, you must set baseURL.

---

Defining Tools

Tools extend what agents can do. Workflow supports:

• WorkflowTool (native)
• AI SDK tools
• LangChain tools
• Agentic toolkits

import { z } from "zod";
import { tool } from "ai";
import { WorkflowTool } from "@upstash/workflow-agents";
import { WikipediaQueryRun } from "@langchain/community/tools/wikipedia_query_run";

const mathTool = tool({
  description: "Evaluate a math expression",
  parameters: z.object({ expression: z.string() }),
  execute: async ({ expression }) => mathjs.evaluate(expression),
});

const workflowMath = new WorkflowTool({
  description: "Evaluate math (workflow step aware)",
  schema: z.object({ expression: z.string() }),
  invoke: async ({ expression }) => mathjs.evaluate(expression),
  executeAsStep: false, // allows context.call, etc.
});

const wikiTool = new WikipediaQueryRun({
  topKResults: 1,
  maxDocContentLength: 500,
});

Common mistakes:

• Workflow wraps execute/invoke in context.run by default, so you cannot use context.call unless executeAsStep: false is set.
• LangChain tools must return strings, not objects.

---

Defining Agents

Agents wrap a model and add behavior via:

• maxSteps — how many LLM calls the agent is allowed to make
• background — system prompt
• tools — available actions

const generator = agents.agent({
  model,
  name: "generator",
  maxSteps: 1,
  background: "Generate text from prompts.",
  tools: {},
});

const evaluator = agents.agent({
  model,
  name: "evaluator",
  maxSteps: 1,
  background: "Evaluate responses and give corrections.",
  tools: {},
});

Tips:

• Pick maxSteps carefully; too low prevents tool use; too high increases cost.
• Names appear in Upstash Console logs; keep them descriptive.

---

Tasks (Single & Multi Agent)

A task is a single execution of an agent or a group of agents.

// Single agent task
const single = agents.task({
  agent: generator,
  prompt: "Explain quantum mechanics.",
});
const { text } = await single.run();

// Multi-agent with manager agent
const multi = agents.task({
  model, // manager LLM
  agents: [generator, evaluator],
  maxSteps: 3,
  prompt: "Generate text and refine it until quality improves.",
});
const result = await multi.run();

Tip: In multi-agent mode, the model becomes a "manager" system that decides which agent to call.

---

Common Agent Patterns

Below are the patterns supported by the source files in this skill.

Prompt Chaining

Sequential agent calls where each output becomes the next input. Useful for: stepwise research, multi‑stage content generation, breaking down complex tasks.

Pitfall: watch maxSteps for agents that need both tool calls and summarization.

Evaluator‑Optimizer

Loop until evaluator returns a PASS. Simple feedback‑refinement pattern.

Pitfall: Always check evaluator output with .includes("PASS"), not strict equality.

Parallelization

Use multiple agents with Promise.all and then aggregate.

Pitfall: Avoid extremely large aggregated prompts; summarizing before combining is recommended.

Orchestrator‑Workers

Manager delegates sub‑tasks to specialized workers. Useful for structured Q&A, multi‑topic analysis, or complex synthesis.

Pitfall: The manager must have enough maxSteps to orchestrate multiple workers.

---

Best Practices

• Give each agent a _clear_ background; ambiguous roles cause incorrect tool use.
• Define tools with strict schemas so LLMs call them reliably.
• Use multi-agent tasks when the problem requires specialization.
• Inspect console logs to debug tool calls and agent decisions.
• Use local QStash dev server during development to avoid rate limits.

---

Example: Combined Setup (models + tools + agents + tasks)

This shows all core fields together in one concise example.

export const { POST } = serve(async (context) => {
  const agents = agentWorkflow(context);
  const model = agents.openai("gpt-4o");

  const mathTool = tool({
    description: "Compute math",
    parameters: z.object({ expression: z.string() }),
    execute: async ({ expression }) => mathjs.evaluate(expression),
  });

  const researcher = agents.agent({
    model,
    name: "researcher",
    maxSteps: 2,
    background: "Research topics using wiki.",
    tools: { wikiTool },
  });

  const mathematician = agents.agent({
    model,
    name: "math",
    maxSteps: 2,
    background: "Solve numeric problems.",
    tools: { mathTool },
  });

  const task = agents.task({
    model, // manager
    agents: [researcher, mathematician],
    maxSteps: 3,
    prompt: "Tell me about 3 stars and compute the sum of their masses.",
  });

  return (await task.run()).text;
});

Workflow Client Basics

This skill provides guidance for using the Workflow Client to trigger, cancel, inspect, resume, and manage workflow runs. It focuses on TypeScript usage patterns, argument structures, and common pitfalls.

---

Core Operations

Triggering Workflow Runs (client.trigger)

Starts one or multiple workflow runs with optional payload, metadata, retry logic, and flow control.

// Single and multiple triggers
const single = await client.trigger({
  url: "https://your-endpoint/workflow", // required
  body: { msg: "hello" }, // optional payload
  headers: { "x-user": "123" }, // optional headers
  workflowRunId: "custom-id", // custom unique identifier
  retries: 3, // retry count for steps
  retryDelay: "1000 * (1 + retried)", // expression-based retry delay
  delay: "10s", // start later
  notBefore: Math.floor(Date.now() / 1000) + 60, // absolute schedule
  label: "my-run", // dashboard filtering
  disableTelemetry: true, // opt‑out telemetry
  flowControl: {
    // concurrency & rate limiting
    key: "user-key",
    rate: 10,
    parallelism: 5,
    period: "10m",
  },
});

const multiple = await client.trigger([
  { url: "https://api/runA", body: { a: 1 } },
  { url: "https://api/runB", retries: 5 },
]);

Pitfalls:

• workflowRunId must be unique; collisions cause errors.
• delay is ignored if notBefore is provided.
• Using url incorrectly (missing protocol or dynamic segments) leads to invalid workflow endpoints.

---

Canceling Workflow Runs (client.cancel)

Cancel runs by ID, by URL prefix, or all active/pending runs.

await client.cancel({ ids: ["wfr_123", "wfr_456"] });
await client.cancel({ urlStartingWith: "https://your-endpoint.com" });
await client.cancel({ all: true });

• ids accepts both string and array, but mixing with other filters is not supported.
• urlStartingWith cancels _all_ descendant paths—use carefully.

---

Retrieving Logs (client.logs)

Fetch workflow execution logs with filtering and cursor-based pagination.

const { runs, cursor } = await client.logs({
  workflowRunId: "wfr_123", // filter a specific run
  workflowUrl: "https://endpoint", // fetch logs for an endpoint
  state: "RUN_FAILED", // filter by execution state
  count: 50, // limit return size
  workflowCreatedAt: 1700000000, // Unix timestamp
  cursor: undefined, // pagination
});

---

Notifying Events (client.notify)

Notify workflows paused at context.waitForEvent.

await client.notify({
  eventId: "order-paid",
  eventData: { orderId: 1 }, // delivered to waiting workflow
});

Pitfall: If no workflow is waiting, no error is thrown.

---

Fetching Waiting Workflows (client.getWaiters)

Retrieve active waiters waiting for a specific event.

const waiters = await client.getWaiters({ eventId: "order-paid" });

This is useful for debugging or coordinating external signals.

---

Dead Letter Queue (DLQ) Operations

Listing DLQ Messages (client.dlq.list)

const { messages, cursor } = await client.dlq.list({
  cursor,
  count: 20,
  filter: {
    fromDate: Date.now() - 86400000,
    toDate: Date.now(),
    url: "https://your-endpoint.com",
    responseStatus: 500,
  },
});

• Date fields use Unix ms, not seconds.

---

Retry Failure Callback (client.dlq.retryFailureFunction)

If a workflow's failureFunction/failureUrl call failed:

await client.dlq.retryFailureFunction({ dlqId: "dlq-123" });

---

Restarting DLQ Runs (client.dlq.restart)

Restart from the beginning of the workflow.

await client.dlq.restart({
  dlqId: ["dlq-1", "dlq-2"],
  retries: 5,
  flowControl: {
    key: "restart-group",
    parallelism: 10,
  },
});

---

Resuming DLQ Runs (client.dlq.resume)

Continue from the failed step.

await client.dlq.resume({
  dlqId: "dlq-123",
  retries: 3,
  flowControl: {
    key: "resume-group",
    rate: 5,
  },
});

Common confusion:

• restart = new run from step 0.
• resume = same run continues at the failed step.

---

General Tips for TS Consumers

• Use flow control when bulk‑triggering or restarting large batches to avoid rate limits.

Workflow Context Basics

This skill documents all workflow context functions available inside a TypeScript workflow. It focuses on how these APIs behave, how their parameters interact, and common pitfalls when composing them into reliable long‑running workflows.

---

context.api

context.api provides typed integrations for OpenAI, Anthropic, and Resend. These calls behave like context.call but enforce the correct operation names and body types.

Key points

• Each provider exposes a .call(stepName, config) function.
• token is always required.
• The operation field selects the API method.
• body must match the provider's type definitions.
• Returns { status, body }.

Example (combined)

// OpenAI + Anthropic + Resend usage
const openai = await context.api.openai.call("openai chat", {
  token: process.env.OPENAI_API_KEY!,
  operation: "chat.completions.create",
  body: {
    model: "gpt-4o",
    messages: [
      { role: "system", content: "Hello" },
      { role: "user", content: "Hi" },
    ],
  },
});

const anthropic = await context.api.anthropic.call("anthropic message", {
  token: process.env.ANTHROPIC_API_KEY!,
  operation: "messages.create",
  body: {
    model: "claude-3-5-sonnet-20241022",
    max_tokens: 1024,
    messages: [{ role: "user", content: "Hello" }],
  },
});

// Resend email
const email = await context.api.resend.call("send email", {
  token: process.env.RESEND_API_KEY!,
  headers: { "content-type": "application/json" },
  body: {
    from: "Acme <onboarding@resend.dev>",
    to: ["someone@example.com"],
    subject: "Welcome",
    html: "<p>Hi!</p>",
  },
});

---

context.call

Performs an HTTP request with long execution windows (up to 12 hours). Always returns a response, even for non‑2xx status codes.

Important parameters

• url: required.
• method: defaults to GET.
• headers: object.
• body: body as string.
• retries: number of retry attempts.
• retryDelay: retry delay expression as string or number (in milliseconds).
• flowControl: throttling config.
• timeout: seconds for each attempt.
• workflow: used only when invoking a workflow inside serveMany.

Example (showing most fields)

const response = await context.call("sync user", {
  url: "https://api.example.com/user",
  method: "POST",
  headers: { authorization: `Bearer ${process.env.TOKEN}` },
  body: JSON.stringify({ id: userId, name }),
  retries: 3,
  retryDelay: "pow(2, retried)", // exponential
  timeout: 20,
  flowControl: {
    key: "user-sync",
    rate: 5,
    parallelism: 2,
  },
});

Pitfalls

• Localhost requests fail unless tunneled.

---

context.cancel

Cancels the current workflow run.

• Workflow is marked canceled, not failed.
• failureFunction does not run.
• No DLQ entry.

Example

const result = await context.run("check something", () => ...);
if (result.cancel) await context.cancel();

---

context.createWebhook

Creates a reusable webhook URL for external systems.

Example

const { webhookUrl, eventId } = await context.createWebhook("create webhook");
// Use webhookUrl with an external service

---

context.invoke

Invokes another workflow (must be served under the same serveMany) and waits for its completion.

Key parameters

• workflow: the workflow definition.
• body: becomes requestPayload of invoked workflow.
• headers: forwarded headers.
• workflowRunId: optional override.
• retries, retryDelay, flowControl: control invocation retry strategy.

Example

const { body, isFailed, isCanceled } = await context.invoke("invoke child", {
  workflow: childWorkflow,
  body: { id: 123 },
  headers: { foo: "bar" },
  retries: 2,
});

---

context.notify

Notifies waiting workflows created via context.waitForEvent.

Example

const { notifyResponse } = await context.notify("notify step", "order-updated", {
  status: "shipped",
});

notifyResponse lists all workflows resumed by this event.

---

context.run

Runs custom synchronous or async logic as a step. Values must be JSON‑serializable.

Example

const user = await context.run("load user", () => getUser());

await context.run("process user", () => process(user));

// Running in parallel
const a = context.run("step A", () => foo());
const b = context.run("step B", () => bar());
await Promise.all([a, b]);

Pitfalls

• Returned class instances lose methods due to serialization. Rehydrate via Object.assign(new Class(), data).

---

context.sleep

Pauses workflow execution for a duration. String or numeric seconds.

Example

await context.sleep("wait 1 day", "1d");

---

context.sleepUntil

Pauses execution until a specific Date, timestamp, or parseable string.

Example

const ts = new Date(Date.now() + 1000 * 60 * 60);
await context.sleepUntil("wait until", ts);

---

context.waitForEvent

Waits for an event (triggered by context.notify) or until timeout.

Example

const { eventData, timeout } = await context.waitForEvent("wait for update", "order-updated", {
  timeout: "2h",
});

This documentation explains how to use serve() to expose workflows as HTTP endpoints.

Core Concept

serve() turns an async route function into a workflow endpoint. When a request arrives, a workflow run is created, and the route function defines the execution logic using context.run().

Full Example With Inline Comments

import { serve } from "@upstash/workflow/nextjs";
import { z } from "zod";
import { Client, Receiver } from "@upstash/qstash";
import { loggingMiddleware } from "@upstash/workflow";

type InitialPayload = { foo: string; bar: number };

const payloadSchema = z.object({ foo: z.string(), bar: z.number() });

export const { POST } = serve<InitialPayload>(
  async (context) => {
    const payload = context.requestPayload;
    await context.run("step-1", async () => `hello ${payload.foo}`);
  },
  {
    // Runs when the workflow fails after all retries
    failureFunction: async ({ context, failStatus, failResponse, failHeaders, failStack }) =>
      console.error(failResponse),

    // Hooks for step lifecycle, run lifecycle, and debug events
    middlewares: [loggingMiddleware],

    // Manually parse the initial request payload before workflow execution
    initialPayloadParser: (raw) => JSON.parse(raw),

    // Automatically validate + parse the request payload using Zod
    schema: payloadSchema,

    // Override the full URL used to schedule subsequent workflow steps
    url: "https://myapp.com/api/workflow",

    // Override only the base portion of the inferred URL (keeps the route path)
    baseUrl: "https://proxy-url.com",

    // Provide a custom QStash client
    qstashClient: new Client({ token: process.env.QSTASH_TOKEN! }),

    // Provide a custom QStash Receiver for verifying that requests come from QStash
    receiver: new Receiver({
      currentSigningKey: process.env.QSTASH_CURRENT_SIGNING_KEY!,
      nextSigningKey: process.env.QSTASH_NEXT_SIGNING_KEY!,
    }),

    // Override environment variables available inside workflow execution
    env: {
      QSTASH_URL: "https://qstash.upstash.io",
      QSTASH_TOKEN: process.env.QSTASH_TOKEN!,
    },

    // Output detailed execution logs to stdout
    verbose: true,

    // Disable anonymous telemetry for this workflow endpoint
    disableTelemetry: true,
  }
);

Common Pitfalls

• Incorrect URL inference behind proxies or local tunnels—set url or baseUrl.

Flow Control

Flow Control defines how many workflow steps may run or start within a given time, ensuring predictable load and preventing overload of internal or third‑party systems. It applies per flow control key, allowing steps across different workflow runs to share the same limits.

This skill documents:

• parallelism
• rate and period
• monitoring
• common behaviors and pitfalls

Core Concepts

Flow Control Key

A string identifier grouping related workflow steps under shared limits. All steps triggered with the same key share the same concurrency and rate budget.

const client = new Client({ token: "<QSTASH_TOKEN>" });

await client.trigger({
  url: "https://<YOUR_ENDPOINT>",
  flowControl: {
    key: "user-onboarding", // Groups steps under this key
    parallelism: 5, // max concurrent steps
    rate: 20, // max starts per period
    period: "1m", // time window
  },
});

Parallelism

Controls how many steps may run concurrently. Works like a token bucket: each running step consumes one token; when it finishes, the token is released.

• If tokens are available → step starts immediately.
• If no tokens → step enters the waitlist.
• Token handoff is not FIFO; later steps may start earlier.

Example Behavior

• parallelism = 3 → at most 3 steps run at the same time.
• Step 4 waits until one of the first 3 completes.

Parallelism affects execution concurrency, not rate‑of‑start per time window.

Rate & Period

Controls how many steps may START within a defined time window.

flowControl: {
  key: "billing-jobs",
  rate: 2,           // allow 2 starts
  period: "10s",    // every 10 seconds
  parallelism: 1,    // can be combined
}

Behavior:

• First rate steps in each period start immediately.
• Additional steps wait until the next window.
• Execution duration does not count toward the period; only start time matters.

This is rate limiting, not concurrency control.

Combining Rate + Parallelism

Both limits apply independently. A step must satisfy _both_ to run.

Example:

• rate = 3 per minute
• parallelism = 7

Even if concurrency slots are free, steps may be delayed by the rate window. Even if rate window is open, steps may be delayed by concurrency.

This enables predictable load patterns.

Monitoring

Flow control queues and waitlists are visible from:

• Console → FlowControl tab (waitlist size per key)
• REST API:
• List all keys
• Get details for one key

Useful for diagnosing bottlenecks, long queues, or misconfigured limits.

Important Behaviors & Pitfalls

Limits are stored on each step

If you deploy new limits, previously published steps keep their original configuration.

No per‑step config (mostly)

All steps inherit the workflow‑run’s flowControl block. Exceptions:

• context.call – may define a separate key to isolate API throttling.
• context.invoke – spawns a new workflow run with its own flowControl.

To isolate heavy or sensitive steps, move them into a separate workflow and invoke them.

Waitlist order is not guaranteed

A later step may receive a token before an earlier one. Do not rely on ordering.

Queueing is automatic

Exceeding limits never rejects steps. Everything is eventually executed.

Full Example

Demonstrates defining all parameters together and how they interact.

import { Client } from "@upstash/workflow";

const client = new Client({ token: "<QSTASH_TOKEN>" });

await client.trigger({
  url: "https://<YOUR_WORKFLOW_ENDPOINT>/process",
  flowControl: {
    key: "media-processing", // Shared grouping key
    parallelism: 4, // Max 4 at the same time
    rate: 10, // Max 10 starts
    period: "60s", // in 60 seconds
  },
});

This configuration:

• ensures no more than 4 processing tasks run concurrently
• ensures no more than 10 tasks start per minute
• queues everything automatically beyond those limits

Workflow Invocation

This documentation covers how to invoke one workflow from another using context.invoke, along with how to define and expose multiple workflows using createWorkflow and serveMany.

---

Invoking Another Workflow

A workflow can start another workflow run and wait for it to finish before continuing. This enables orchestration and dependency chains across workflows.

Use context.invoke to trigger another workflow:

const result = await context.invoke("analyze content", {
  workflow: analyzeContent, // workflow object or function
  body: "input payload", // request body passed to invoked workflow
  headers: { "x-user": "123" }, // optional headers
  retries: 5, // optional, default 3
  flowControl: {
    /* ... */
  },
  workflowRunId: "optional-custom-id",
});

const {
  body, // response returned by invoked workflow
  isFailed, // true if invoked workflow failed
  isCanceled, // true if invoked workflow was canceled
} = result;

Returning a Value

A workflow may return any serializable value, which becomes body for the caller.

Depth Limitation

Workflow invocation depth is limited to 100. Any recursive or cyclic chain exceeding this will fail.

---

Defining Workflows Without Exposing Endpoints

Use createWorkflow to define workflows as objects. These are not automatically exposed as HTTP endpoints.

const analyzeContent = createWorkflow(async (context) => {
  await context.sleep("delay", 1);
  return { message: "processed" };
});

const processUser = createWorkflow(async (context) => {
  const { body } = await context.invoke("invoke analyze", {
    workflow: analyzeContent, // type‑safe invocation
    body: "user-1",
  });

  return { result: body };
});

---

Exposing Multiple Workflows With serveMany

To allow workflows to invoke each other without requiring explicit URLs, define all related workflows under a shared parent path.

// app/workflows/[...any]/route.ts
export const { POST } = serveMany({
  analyze: analyzeContent,
  process: processUser,
});

• Keys become route names.
• All workflows that need to invoke each other must appear in the same serveMany call.
• Requests are routed based on the key names.

Triggering a Workflow

import { Client } from "@upstash/workflow";

const client = new Client({ token: "<QSTASH_TOKEN>" });

await client.trigger({
  url: "https://your-app/workflows/analyze",
});

---

Common Pitfalls

• Missing workflow in `serveMany`: If a workflow invokes another that was not included in the same serveMany definition, the call will fail.
• Infinite invocation loops: Depth limit of 100 prevents runaway recursive chains.

Workflow Reliability Features

This documentation provides a consolidated reference for reliability mechanisms in Upstash Workflow, focusing on retries, failure handling, and the Dead Letter Queue (DLQ). It is optimized for AI‑driven execution but remains clear for developers.

---

Core Reliability Concepts

Upstash Workflow ensures that workflow runs are durable, recoverable, and traceable. Reliability is achieved through:

• Automatic retries for transient failures
• Failure handling via failureFunction or failureUrl
• DLQ for storing permanently failed runs
• Manual recovery actions: restart, resume, rerun failure function
• Mechanisms to prevent retries when failure is intentional

These mechanisms work together to guarantee that no failed execution is lost and that failures can be observed and handled.

---

Automatic Retries

Workflow steps automatically retry when they fail, using configurable retry count and delay.

Retry Parameters

• retries: How many times the step will retry
• retryDelay: A math expression (string) returning delay in milliseconds; can use retried variable

Common Pitfalls

• The delay expression must be a _string_; raw numbers disable dynamic calculation.
• A workflow only moves to DLQ after all retries fail.

Example: Configure Retry Count + Delay

import { Client, WorkflowNonRetryableError, serve } from "@upstash/workflow";

// Trigger with custom retry settings
const client = new Client({ token: process.env.WORKFLOW_TOKEN! });
await client.trigger({
  url: "https://example.com/workflow",
  retries: 5,
  retryDelay: "(1 + retried) * 2000", // 2s, 4s, 6s...
});

// Workflow with steps that may intentionally skip retries
export const { POST } = serve(async (context) => {
  const shouldFail = await context.run("check", () => true);

  if (shouldFail) {
    // Skips retry cycle and sends directly to failure handling
    throw new WorkflowNonRetryableError("Bad state");
  }

  await context.run("process", () => doWork());
});

---

Preventing Retries

When to Stop Retries

Use these when failure is expected or should halt execution immediately:

• WorkflowNonRetryableError → Fail fast, trigger failure function, enter DLQ
• context.cancel() → Stop workflow intentionally, no DLQ, no failure function
• Conditional returns → Graceful early exit without error

Pitfalls

• context.cancel() marks the run as canceled, not failed — no failure handling executes.
• Throwing any error _other than_ WorkflowNonRetryableError will still trigger retries.

---

Failure Handling

You can handle workflow failures using either:

• failureFunction (runs in your workflow environment)
• failureUrl (external callback endpoint)

Only one can be configured — they are mutually exclusive.

Failure Function Behavior

• Runs after all retries fail
• Receives failure metadata and workflow context
• Cannot create new workflow steps; context is read‑only
• Is retried if it fails
• Can be manually retried from the DLQ

Failure Function Parameters

• context.workflowRunId
• context.url
• context.requestPayload
• context.headers
• context.env
• failStatus
• failResponse
• failHeaders

Example: Workflow + Failure Function + Prevent‑Retry Logic

export const { POST } = serve(
  async (context) => {
    const ok = await context.run("verify", () => verifySomething());

    if (!ok) {
      throw new WorkflowNonRetryableError("Verification failed");
    }

    const shouldAbort = await context.run("check", () => false);
    if (shouldAbort) {
      await context.cancel();
      return;
    }

    await context.run("execute", () => doWork());
  },
  {
    failureFunction: async ({ context, failStatus, failResponse }) => {
      // Logging / cleanup / alerts
      await logFailure(context.workflowRunId, failStatus, failResponse);
    },
  }
);

---

Dead Letter Queue (DLQ)

A workflow enters the DLQ when:

• A step fails _after all retries_
• A WorkflowNonRetryableError is thrown
• failureFunction also fails (visible as failed in DLQ)

Retention

• Free: 3 days
• Pay‑as‑you‑go: 1 week
• Fixed: Up to 3 months

After retention expires, items cannot be recovered.

---

DLQ Recovery Actions

You may perform these actions from the dashboard or programmatically.

1. Resume

Continue from the failed step while keeping previous step results.

2. Restart

Start the entire workflow from scratch, discarding previous results.

3. Rerun Failure Function

Retry the failure handler only if it previously failed.

Example: All DLQ Actions

import { Client } from "@upstash/workflow";
const client = new Client({ token: process.env.WORKFLOW_TOKEN! });

await client.dlq.resume({ dlqId: "dlq-1", retries: 3 }); // Continue run
await client.dlq.restart({ dlqId: "dlq-2", retries: 3 }); // Restart
await client.dlq.retryFailureFunction({ dlqId: "dlq-3" }); // Retry failure handler

Pitfalls

• retryFailureFunction is only allowed when the failure function failed.
• resume cannot be used if workflow code changed before the failure point.
• restart re-executes everything, including expensive steps.

---

Failure URL (Advanced Option)

Use when failure must be handled by a separate service.

Notes

• Use either failureFunction or failureUrl.
• Recommended only when your primary app might be offline.
• Failure URL receives a structured JSON payload about the failed request.

Example

export const { POST } = serve(
  async (context) => {
    await context.run("work", () => doWork());
  },
  {
    failureUrl: "https://example.com/failure-handler",
  }
);

---

Debugging Failed Runs

DLQ entries store:

• Request + response headers
• Payloads
• Failure metadata
• Failure function state

Use the dashboard filters (workflow run ID, URL, failure function state) to identify and debug root causes.

---

Summary

This skill unifies all Workflow reliability controls:

• Retries (automatic, configurable)
• Failure handling (failureFunction, failureUrl)
• DLQ retention + recovery
• Fast‑fail mechanisms (WorkflowNonRetryableError, cancel, conditional exits)
• Manual recovery operations (resume, restart, retry failure function)

Together these features ensure workflows are resilient, debuggable, and recoverable under all failure scenarios.

Wait for Event

Use this feature to pause workflow execution until an external event arrives. It allows building asynchronous, event‑driven workflows without holding compute resources.

---

How It Works

context.waitForEvent() suspends the workflow and registers a waiter under an event ID. When a matching notify call is sent, the workflow resumes with the provided event data.

• Each waiter has a timeout. If the event does not arrive in time, the workflow continues with { timeout: true }. • Maximum timeout depends on your pricing tier. • Multiple workflow runs may wait on the same event ID; a notify call resumes all of them.

---

Key Concepts & Pitfalls

Event IDs

Use unique event IDs to avoid heavy fan‑out notifications.

Example patterns: • order-123-processing-complete • user-42-email-verified

Race Conditions

A notify call sent _before_ the workflow begins waiting is lost.

To avoid this: • Always inspect the notify response. • Retry if no waiters were found.

---

Combined Example

Below is a single TypeScript example showing waiting for an event, handling timeouts, and safely notifying:

import { serve } from "@upstash/workflow/nextjs";
import { Client } from "@upstash/workflow";

// Workflow that waits for an event
export const { POST } = serve(async (context) => {
  const { orderId } = context.requestPayload;
  const eventId = `order-${orderId}-processed`;

  // Wait for the event with timeout
  const { eventData, timeout } = await context.waitForEvent(
    "wait-for-order-processing", // step name
    eventId, // event id
    {
      timeout: "1d", // optional timeout
    }
  );

  if (timeout) {
    // handle timeout here
    await context.run("timeout-handler", async () => {
      console.log("Order processing timed out");
    });
    return;
  }

  // Continue workflow using eventData
  await context.run("process-completed-order", async () => {
    console.log("Order processed:", eventData);
  });
});

// External notifier (safe retry pattern)
const client = new Client({ token: "<WORKFLOW_TOKEN>" });

async function notifyProcessingComplete(orderId: string, payload: any) {
  const eventId = `order-${orderId}-processed`;

  // First attempt - returns array of NotifyResponse
  const waiters = await client.notify({ eventId, eventData: payload });

  if (waiters > 0) return waiters;

  // Retry if no workflows were waiting
  await new Promise((r) => setTimeout(r, 3000));
  return await client.notify({ eventId, eventData: payload });
}

---

Best Practices

• Generate event IDs that uniquely identify a specific workflow instance. • Always handle the timeout case explicitly. • On notification, check the returned array length to detect if any workflows were waiting. • Prefer one event ID per workflow run to avoid notifying large groups.

Webhooks

This skill teaches agents how to use webhooks inside Upstash Workflow to pause execution, wait for external callbacks, and resume reliably. It covers creation, waiting, validation, and multi-step usage.

---

Key Concepts

• Webhook URLs are unique per context.createWebhook() call. • Workflow execution pauses when calling context.waitForWebhook() until the webhook receives a request or times out. • Lookback protection ensures the workflow still receives early webhook calls. • Workflows remain dormant during the waiting period, avoiding compute cost. • Webhooks are ideal for third‑party APIs that support callback URLs.

---

Core Workflow: Create → Call External API → Wait → Resume

Below is a consolidated example showing: • Creating a webhook • Passing it to an external API • Handling the callback • Waiting for multiple updates • Error/timeout handling

import { serve } from "@upstash/workflow/nextjs";

export const { POST } = serve(async (context) => {
  // Create a unique webhook for this workflow
  const webhook = await context.createWebhook("create webhook");

  // Trigger an external service, sending it the callback URL
  await context.call("trigger-external", {
    url: "https://api.example.com/process",
    method: "POST",
    body: { webhookUrl: webhook.webhookUrl },
  });

  // Loop: wait for progress updates until service signals completion
  let updates = [];
  while (true) {
    const res = await context.waitForWebhook(`wait update`, webhook, "5m");

    if (res.timeout) break; // no progress received

    const req = res.request;
    const payload = await req.json();
    updates.push(payload);

    if (req.headers.get("x-task-finished") === "true") break;
  }

  return updates;
});

---

Webhook Reliability: Lookback Protection

Common pitfall: external systems may call the webhook _before_ the workflow reaches waitForWebhook().

Upstash automatically stores early webhook calls. The next waitForWebhook() will immediately resolve with the stored request.

This avoids race conditions and eliminates the need for manual buffering.

---

When to Use Webhooks vs. Wait for Event

Use webhooks when: • integrating with third‑party APIs • the external system invokes a callback URL • you need race‑condition‑safe behavior

Use Wait for Event when: • you control the notifying system and can use the Workflow Client • you don't need lookback semantics

Local Development for Upstash Workflow

This documentation explains how to run Upstash Workflow locally using the QStash development server and how to expose your local app using a public tunnel such as ngrok. It focuses on practical usage and TypeScript integration.

Overview

Upstash Workflow uses Upstash QStash under the hood. During development you can:

• Run a local QStash development server to simulate workflow behavior entirely offline.
• Optionally expose your local server using ngrok if you want to test with the production QStash.

Both approaches require updating environment variables and ensuring your workflow trigger URLs correctly reference your local or tunnel address.

---

1. Start the QStash Local Development Server

Use the QStash CLI:

npx @upstash/qstash-cli dev

The CLI prints:

• QSTASH_TOKEN
• QSTASH_CURRENT_SIGNING_KEY
• QSTASH_NEXT_SIGNING_KEY
• Local server URL (default: http://127.0.0.1:8080)

Set these values in your .env file so your workflow client uses the local environment.

---

2. Set Local Environment Variables

Use the environment values printed by the CLI:

QSTASH_URL="http://127.0.0.1:8080"
QSTASH_TOKEN="<token-from-cli>"
QSTASH_CURRENT_SIGNING_KEY="<cur-key>"
QSTASH_NEXT_SIGNING_KEY="<next-key>"

These ensure all workflow requests are routed locally.

---

3. Trigger Workflows Using Local URLs

A common pattern is determining the base URL dynamically based on environment variables. Below is an example that demonstrates:

• Local development (localhost)
• Production deployments (auto-detected env)

import { Client } from "@upstash/workflow";

const client = Client();

// In production, Vercel sets VERCEL_URL. Otherwise use localhost.
const BASE_URL = process.env.VERCEL_URL
  ? `https://${process.env.VERCEL_URL}`
  : `http://localhost:3000`;

// Trigger a workflow with retries
const { workflowRunId } = await client.trigger({
  url: `${BASE_URL}/api/workflow`, // Local or production
  retries: 3, // Optional retry logic
});

console.log("Workflow run:", workflowRunId);

Common mistakes:

• Forgetting to include the full URL including http:// or https://.
• Using a production URL while the local QStash server is running.
• Missing environment variables.

---

4. Using ngrok (Optional)

If your workflow must be reachable from the managed Upstash servers (not local), expose your local server publicly.

Install & authenticate

ngrok config add-authtoken <YOUR-AUTH-TOKEN>

Start a tunnel

ngrok http 3000

ngrok outputs a public URL:

Forwarding  https://1234abcd.ngrok.io -> http://localhost:3000

Use this public URL instead of localhost:

const BASE_URL = "https://1234abcd.ngrok.io"; // Public tunnel

await client.trigger({
  url: `${BASE_URL}/api/workflow`,
  retries: 3,
});

Pitfall: Your ngrok port must match your dev server port, otherwise all workflow calls will fail.

---

Summary

• Start QStash locally using qstash-cli dev.
• Copy generated environment variables into .env.
• Use local URLs in TypeScript clients while developing.
• Optionally expose your server with ngrok if you need remote access.

This setup ensures fast workflow iteration without deploying your app.

Middlewares

This guide explains how to intercept workflow lifecycle and debug events using middleware. It covers how to attach middlewares, define event handlers, initialize resources, and combine functionality into efficient implementations.

Overview

Middlewares plug into the workflow engine and listen to:

• Lifecycle events such as run start, run completion, and step execution
• Debug messages such as errors, warnings, and informational logs

They are useful for logging, monitoring, instrumentation, and integrating with external systems.

Using Built‑in Middleware

Add middleware inside the middlewares array when serving a workflow:

import { serve } from "@upstash/workflow/nextjs";
import { loggingMiddleware } from "@upstash/workflow";

export const { POST } = serve(
  async (ctx) => {
    await ctx.run("step-1", () => "Hello World");
  },
  {
    middlewares: [loggingMiddleware], // emits run/step logs
  }
);

Creating Custom Middleware

You define middleware by instantiating WorkflowMiddleware with callbacks for the events you want to handle.

Direct Callback Definitions

All lifecycle and debug events can be defined in one object.

import { WorkflowMiddleware } from "@upstash/workflow";

const customMiddleware = new WorkflowMiddleware({
  name: "custom-logger",
  callbacks: {
    // Lifecycle events
    runStarted: async ({ context }) => {
      console.log(`Run ${context.workflowRunId} started`);
    },
    beforeExecution: async ({ context, stepName }) => {
      console.log(`→ Executing step: ${stepName}`);
    },
    afterExecution: async ({ context, stepName, result }) => {
      console.log(`✓ Step ${stepName} completed`, result);
    },
    runCompleted: async ({ context, result }) => {
      console.log(`Run ${context.workflowRunId} finished`, result);
    },

    // Debug events
    onError: async ({ workflowRunId, error }) => {
      console.error(`Error in ${workflowRunId}`, error);
    },
    onWarning: async ({ workflowRunId, warning }) => {
      console.warn(`Warning from ${workflowRunId}`, warning);
    },
    onInfo: async ({ workflowRunId, info }) => {
      console.info(`Info from ${workflowRunId}`, info);
    },
  },
});

Initialization With init

Use init when you need setup logic (connecting to a DB, creating clients, loading configuration). init returns an object containing the callbacks.

import { WorkflowMiddleware } from "@upstash/workflow";

const databaseMiddleware = new WorkflowMiddleware({
  name: "database-logger",
  init: async () => {
    const db = await connectToDatabase();

    return {
      runStarted: async ({ context }) => {
        await db.insert({ workflowRunId: context.workflowRunId, status: "started" });
      },
      runCompleted: async ({ context, result }) => {
        await db.update({ workflowRunId: context.workflowRunId, status: "completed", result });
      },
      onError: async ({ workflowRunId, error }) => {
        await db.insert({ workflowRunId, level: "error", message: error.message });
      },
    };
  },
});

Event Reference

Below is a concise list of supported event handlers and what they receive.

Lifecycle

• runStarted: { context }
• beforeExecution: { context, stepName }
• afterExecution: { context, stepName, result }
• runCompleted: { context, result }

Debug

• onError: { workflowRunId?, error }
• onWarning: { workflowRunId?, warning }
• onInfo: { workflowRunId?, info }

Combined Example

Single snippet showing typical usage patterns.

import { serve } from "@upstash/workflow/nextjs";
import { WorkflowMiddleware } from "@upstash/workflow";

// Error tracking
const errorTrackingMiddleware = new WorkflowMiddleware({
  name: "errors",
  callbacks: {
    onError: async ({ workflowRunId, error }) => {
      await fetch("https://monitoring/errors", {
        method: "POST",
        body: JSON.stringify({ workflowRunId, error: error.message }),
      });
    },
  },
});

// Performance measurement
const timings = new Map();
const performanceMiddleware = new WorkflowMiddleware({
  name: "perf",
  callbacks: {
    beforeExecution: ({ stepName }) => timings.set(stepName, Date.now()),
    afterExecution: ({ stepName }) => {
      const t = timings.get(stepName);
      if (t) console.log(`Step ${stepName} took ${Date.now() - t}ms`);
      timings.delete(stepName);
    },
  },
});

export const { POST } = serve(
  async (ctx) => {
    await ctx.run("task", () => "done");
  },
  {
    middlewares: [errorTrackingMiddleware, performanceMiddleware],
  }
);

Pitfalls and Tips

• Ensure init returns all callbacks; missing keys simply mean the event is ignored.
• Debug events may be called without a workflowRunId if errors occur before run id is identified.

Migration Guide for Upstash Workflow (TypeScript)

This document provides a clear, task-focused overview of migrations between Workflow versions. It highlights breaking changes, how to update your code, and common pitfalls to avoid. Code samples combine multiple parameters in each example to reduce redundancy.

January 2026 – Major Release 1.0.0

Agents API → moved to a separate package

The Agents API was removed from the core workflow package and placed into @upstash/workflow-agents so the base SDK no longer depends on large AI-related libraries.

Key changes:

• Must install and import agentWorkflow from the new package.
• Agents access now happens through an agents helper created per workflow invocation.

// Before
import { serve } from "@upstash/workflow/nextjs";
export const { POST } = serve(async (context) => {
  const model = context.agents.openai("gpt-3.5-turbo");
  const agent = context.agents.agent({
    /* ... */
  });
  const task = context.agents.task({
    /* ... */
  });
});

// After
import { serve } from "@upstash/workflow/nextjs";
import { agentWorkflow } from "@upstash/workflow-agents";
export const { POST } = serve(async (context) => {
  const agents = agentWorkflow(context);
  const model = agents.openai("gpt-3.5-turbo");
  const agent = agents.agent({
    /* ... */
  });
  const task = agents.task({
    /* ... */
  });
});

---

keepTriggerConfig and useFailureFunction removed

These fields were redundant—both behaviors are now always enabled.

// Before
await client.trigger({
  url: "...",
  retries: 3,
  keepTriggerConfig: true,
  useFailureFunction: true,
});

// After
await client.trigger({ url: "...", retries: 3 });

---

Configuration moved from serve() → client.trigger()

Options such as retries, flowControl, retryDelay, and failureUrl no longer belong in serve().

// Before
export const { POST } = serve(
  async (context) => {
    /* ... */
  },
  {
    retries: 3,
    retryDelay: "1000 * (1 + retried)",
    flowControl: { key: "my-key", rate: 10 },
  }
);
await client.trigger({ url: "..." });

// After
export const { POST } = serve(async (context) => {
  /* ... */
});
await client.trigger({
  url: "...",
  retries: 3,
  retryDelay: "1000 * (1 + retried)",
  flowControl: { key: "my-key", rate: 10 },
});

---

stringifyBody removed from context.call and context.invoke

Bodies must now be strings.

// Before
await context.call("step", {
  url: "https://api.example.com",
  method: "POST",
  body: { key: "value" },
  stringifyBody: true,
});

// After
await context.call("step", {
  url: "https://api.example.com",
  method: "POST",
  body: JSON.stringify({ key: "value" }),
});

// Same change applies to invoke
await context.invoke("other", {
  workflow: otherWorkflow,
  body: JSON.stringify({ key: "value" }),
});

---

Logger removed → middleware system added

Removed WorkflowLogger class and added WorkflowMiddleware. Updated verbose param to only allow true (in this case, loggingMiddleware will be used which prints to console).

import { loggingMiddleware, WorkflowMiddleware } from "@upstash/workflow";

const stepFinish = new WorkflowMiddleware({
  name: "step-finish",
  callbacks: {
    afterExecution: async ({ stepName, result }) =>
      console.log(`Step ${stepName} finished`, result),
  },
});

export const { POST } = serve(
  async (context) => {
    /* ... */
  },
  {
    middlewares: [loggingMiddleware, stepFinish],
  }
);

---

October 2024 – Migration from @upstash/qstash

Install the new package

Replace all workflow usage from @upstash/qstash with @upstash/workflow.

Serve import and return shape changed

Most environments now require destructuring { POST }.

// Before
import { serve } from "@upstash/qstash/nextjs";
export const POST = serve(...);

// After
import { serve } from "@upstash/workflow/nextjs";
export const { POST } = serve(...);

context.call updated

Now returns { status, headers, body } and does not fail the workflow if the HTTP call fails.

const { status, headers, body } = await context.call("call-step", {
  url: "https://example.com/api",
  method: "POST",
});

Pitfall: Old runs may not contain status or headers until fully migrated.

Errors renamed

• QStashWorkflowError → WorkflowError
• QStashWorkflowAbort → WorkflowAbort

---

Summary of Common Mistakes

• Using context.agents instead of the new agentWorkflow() helper.
• Leaving old serve-config options (retries, flowControl, etc.).
• Forgetting to JSON.stringify bodies.
• Assuming context.call failures stop workflow execution.
• Mix-and-matching @upstash/qstash and @upstash/workflow imports.

This guide ensures smooth migration with minimal friction while highlighting where behavioral changes may break existing workflows.

Realtime Workflow Integration (TypeScript)

This Skill provides guidance for building real‑time and human‑in‑the‑loop workflows using Upstash Workflow + Upstash Realtime. It covers event schemas, workflow patterns, emitting/receiving updates, and handling interactive pauses.

Key capabilities:

• Emit strongly typed workflow events.
• Subscribe to live updates in the frontend.
• Implement pause/resume workflows using waitForEvent and notify.

---

Core Concepts

Realtime schema design

Define all workflow event types in one schema. Keep event definitions minimal and stable.

// lib/realtime.ts
const schema = {
  workflow: {
    // Completion events
    runFinish: z.object({}),
    stepFinish: z.object({ stepName: z.string(), result: z.unknown().optional() }),

    // Human‑in‑the‑loop events
    waitingForInput: z.object({ eventId: z.string(), message: z.string() }),
    inputResolved: z.object({ eventId: z.string() }),
  },
};

export const realtime = new Realtime({ schema, redis });
export type RealtimeEvents = InferRealtimeEvents<typeof realtime>;

Common mistakes:

• Forgetting to wrap events under a namespace (e.g. workflow.*).

---

Emitting Events Inside Workflows

Emit inside context.run() whenever possible to avoid duplicate emissions on retries.

// app/api/workflow/route.ts
export const { POST } = serve(async (context) => {
  const channel = realtime.channel(context.workflowRunId);

  // Emit step updates
  await context.run("validate", async () => {
    const result = { ok: true };
    await channel.emit("workflow.stepFinish", { stepName: "validate", result });
    return result;
  });

  // Final event
  await context.run("finish", () => channel.emit("workflow.runFinish", {}));
});

Pitfall:

• Emitting events _outside_ of context.run() risks duplicate delivery if retries happen.

---

Human‑in‑the‑Loop Workflows

Use waitForEvent() with unique eventId values and emit a waitingForInput event immediately.

// Step that pauses for human input
const eventId = `approval-${context.workflowRunId}`;

const [{ eventData, timeout }] = await Promise.all([
  context.waitForEvent("wait-for-approval", eventId, { timeout: "5m" }),
  context.run("notify-wait", () =>
    channel.emit("workflow.waitingForInput", {
      eventId,
      message: "Waiting for approval...",
    })
  ),
]);

if (timeout) return { success: false, reason: "timeout" };

await context.run("resolved", () => channel.emit("workflow.inputResolved", { eventId }));

Important details:

• Always handle timeout: otherwise the workflow may appear stuck.
• Emit inputResolved so the UI can clear any pending state.
• Use deterministic eventId prefixes per action/approval.

---

Notify Endpoint (User Response)

The frontend sends input back to the workflow using client.notify().

// app/api/notify/route.ts
export async function POST(req: NextRequest) {
  const { eventId, eventData } = await req.json();

  await workflowClient.notify({ eventId, eventData }); // resumes workflow
  return NextResponse.json({ success: true });
}

Common mistakes:

• Sending the wrong eventId → workflow never resumes.
• Sending eventData that doesn’t match the awaited type.

---

Realtime Subscription in the Frontend

Use a single hook to consume all workflow‑related events and manage state.

// lib/realtime-client.ts & useWorkflow hooks
useRealtime({
  enabled: !!workflowRunId,
  channels: [workflowRunId],
  events: [
    "workflow.stepFinish",
    "workflow.runFinish",
    "workflow.waitingForInput",
    "workflow.inputResolved",
  ],
  onData({ event, data }) {
    switch (event) {
      case "workflow.stepFinish":
        setSteps((prev) => [...prev, data]);
        break;
      case "workflow.waitingForInput":
        setWaitingState(data);
        break;
      case "workflow.inputResolved":
        setWaitingState((prev) => (prev?.eventId === data.eventId ? null : prev));
        break;
      case "workflow.runFinish":
        setIsRunFinished(true);
    }
  },
});

Pitfalls:

• Forgetting to clear waiting state on inputResolved.
• Forgetting to reset state on re‑trigger.

---

Triggering the Workflow

One simple endpoint triggers both basic and human‑in‑the‑loop flows.

// app/api/trigger/route.ts
export async function POST(req: NextRequest) {
  const workflowUrl = `${req.nextUrl.origin}/api/workflow`; // or /human-in-loop
  const { workflowRunId } = await workflowClient.trigger({
    url: workflowUrl,
    body: { userId: "123", action: "process" },
  });
  return NextResponse.json({ workflowRunId });
}

---

Recommended Patterns

• Use one channel per workflow run (workflowRunId).
• Always place event emissions inside context.run().
• Use stable eventId formats for all interactive steps.
• Subscribe to only the events your UI needs.
• Reset UI state before every new trigger.

Workflow REST API Skill

This Skill provides TypeScript-focused guidance for interacting with the Upstash QStash Workflow Dead Letter Queue (DLQ) REST API. It includes practical usage examples, consolidated patterns, and key considerations for safely restarting, resuming, listing, deleting, and inspecting failed workflow runs.

The goal is to help agents construct correct TypeScript API requests, handle optional fields, and avoid common pitfalls when working with workflow recovery operations.

---

Core Concepts

• DLQ Entries represent failed workflow runs containing metadata such as payload, headers, timestamps, failure callback state, and workflow configuration. • Restart recreates the entire workflow run from the beginning (fresh execution, no preserved state). • Resume creates a new run continuing from the failed step (preserves successful steps). • Bulk operations process up to 50 DLQ entries per request and may return a cursor for pagination. • Flow-Control and Retry Overrides are passed via headers and apply only to the newly created workflow run.

Be careful: changing workflow code before the failed step may cause resume operations to break.

---

HTTP Endpoints Overview

Below is a complete summary of the API endpoints supported by this skill.

• GET /v2/workflows/dlq — List DLQ entries with filtering. • GET /v2/workflows/dlq/{dlqId} — Retrieve a single DLQ entry. • DELETE /v2/workflows/dlq/{dlqId} — Delete a DLQ entry. • POST /v2/workflows/dlq/restart/{dlqId} — Restart a workflow from scratch. • POST /v2/workflows/dlq/resume/{dlqId} — Resume a workflow from point of failure. • POST /v2/workflows/dlq/restart — Bulk restart (up to 50). • POST /v2/workflows/dlq/resume — Bulk resume (up to 50). • POST /v2/workflows/dlq/callback/{dlqId} — Rerun a failed failure-callback. • GET /v2/flowControl and /v2/flowControl/{key} — Flow-control inspection.

---

TypeScript Usage Patterns

Below is a single integrated example illustrating several endpoints, optional query parameters, header overrides, and response field handling.

import fetch from "node-fetch";

const token = process.env.QSTASH_TOKEN;
const base = "https://qstash.upstash.io/v2";

async function example() {
  // 1. List DLQ items with filters
  const list = await fetch(
    `${base}/workflows/dlq?workflowUrl=https://my.app/workflow&fromDate=1700000000000`,
    {
      headers: { Authorization: `Bearer ${token}` },
    }
  ).then((r) => r.json());

  const firstDlqId = list.messages?.[0]?.dlqId;

  // 2. Restart or Resume a specific run (headers may override flow control or retries)
  const restart = await fetch(`${base}/workflows/dlq/restart/${firstDlqId}`, {
    method: "POST",
    headers: {
      Authorization: `Bearer ${token}`,
      "Upstash-Flow-Control-Key": "my-key", // optional
      "Upstash-Flow-Control-Value": "parallelism=1", // optional
      "Upstash-Retries": "3", // optional
    },
  }).then((r) => r.json());

  // 3. Bulk resume using filters and cursor handling
  const bulkResume = await fetch(`${base}/workflows/dlq/resume`, {
    method: "POST",
    headers: { Authorization: `Bearer ${token}` },
    body: JSON.stringify({
      fromDate: 1700000000000,
      workflowUrl: "https://my.app/workflow",
      dlqIds: [firstDlqId], // optional; filters also supported
    }),
  }).then((r) => r.json());

  // 4. Rerun failure callback
  const callback = await fetch(`${base}/workflows/dlq/callback/${firstDlqId}`, {
    method: "POST",
    headers: { Authorization: `Bearer ${token}` },
  }).then((r) => r.json());

  // 5. Delete a DLQ entry
  await fetch(`${base}/workflows/dlq/${firstDlqId}`, {
    method: "DELETE",
    headers: { Authorization: `Bearer ${token}` },
  });

  // 6. Inspect Flow-Control
  const fc = await fetch(`${base}/flowControl/my-key`, {
    headers: { Authorization: `Bearer ${token}` },
  }).then((r) => r.json());

  console.log({ list, restart, bulkResume, callback, fc });
}

---

Key Fields and Behaviors

Request Query Fields

• dlqIds: string[] — Exact DLQ IDs for targeting specific failed runs (bulk restart/resume). • fromDate / toDate: number — Millisecond timestamps; inclusive range filters. • workflowUrl: string — Filters by workflow endpoint URL. • workflowRunId: string — Can match full run ID or prefix. • workflowCreatedAt: number — Timestamp filter for creation time. • responseStatus: number — Filter by HTTP status of failed run. • callerIP: string — Filter by origin IP. • failureCallbackState: string — Filter runs whose failure callback succeeded or failed. • cursor: string — Pagination cursor; returned by list & bulk ops. • count: number — Limit for listing DLQ entries (default/max 100).

Header Overrides

These apply only when creating a new workflow run (restart or resume).

• Upstash-Flow-Control-Key — Override flow-control key. • Upstash-Flow-Control-Value — Override flow-control config (e.g., "parallelism=1"). • Upstash-Retries — Override step retry configuration.

All are optional; original values are reused if omitted.

Response Fields

• cursor?: string — If returned, additional entries exist. • workflowRuns: { workflowRunId: string; workflowCreatedAt: number }[] — Returned by bulk restart/resume. • message objects (DLQ list/get) include:

• workflowRunId, workflowCreatedAt, workflowUrl
• responseStatus, responseHeader, responseBody
• failureCallbackInfo.state — Identify failed callbacks.
• dlqId — Unique DLQ entry identifier.

---

Pitfalls & Best Practices

• Resume only works if workflow code before the failed step stays unchanged. Any change may cause resume to fail. • Bulk operations process max 50 items. Always check cursor to continue processing. • Deleting a DLQ entry is permanent; once removed, it cannot be resumed or restarted. • Failure callback reruns are independent and do not affect workflow execution. They only retry the failure-notification step. • Prefix matching on workflowRunId can unintentionally match more runs than expected—use full ID for precision.

---

Recommended Patterns

• Fetch → Check cursor → Continue looping for bulk operations. • Always log failureCallbackInfo.state to determine whether a callback rerun is required. • Apply header overrides sparingly; flow-control misconfiguration may stall large batches. • Prefer filtering in bulk operations instead of manually passing large dlqIds arrays.

Workflow Troubleshooting (TypeScript)

This guide provides clear, TypeScript‑focused patterns for diagnosing and fixing common issues in Upstash Workflow. It highlights correct usage of core workflow features and shows how to avoid pitfalls that commonly occur in real applications.

---

Steps Inside try/catch

Workflow steps (context.run, context.sleep, context.sleepUntil, context.call) intentionally throw WorkflowAbort after completing. Catching this error prevents the workflow engine from progressing.

Correct pattern: rethrow WorkflowAbort or avoid try/catch around steps.

import { WorkflowAbort } from "@upstash/workflow";

try {
  // Any step will throw WorkflowAbort
  const result = await context.run("step", () => "ok");
} catch (err) {
  if (err instanceof WorkflowAbort) throw err; // required
  // handle real errors
}

If you _must_ handle errors, move the try/catch into the function passed to context.run.

---

requestPayload Becoming Undefined

During context.call, the workflow endpoint is invoked multiple times internally. Before the first step executes, context.requestPayload may appear undefined.

Fix: capture the payload via a step:

export const { POST } = serve(async (context) => {
  const payload = await context.run("get-payload", () => context.requestPayload);

  // payload remains stable even during context.call
  await context.call("my-call", {
    url: "https://example.com",
    body: payload,
  });
});

Header considerations: When triggering workflows, ensure payload‑friendly headers like:

• Content-Type: text/plain
• Content-Type: application/json

---

Signature Verification Failures

If QStash request verification fails:

• Ensure the workflow is triggered through QStash (client.trigger or publish)
• Verify QSTASH_CURRENT_SIGNING_KEY and QSTASH_NEXT_SIGNING_KEY
• Pass appropriate Content-Type headers when triggering

Error looks like:

Failed to verify that the Workflow request comes from QStash: ...

---

Authorization Errors from Early Returns

If your workflow returns before running any step, the SDK interprets this as failed authorization:

HTTP 400 – Failed to authenticate Workflow request.

To safely perform non‑deterministic checks, wrap them in a step:

export const { POST } = serve(async (context) => {
  const shouldExit = await context.run("check-condition", () => Math.random() > 0.5);
  if (shouldExit) return;

  // remaining workflow
});

---

Retry Configuration

Retries come from two locations:

• Workflow start (client.trigger): default 3, applies to workflow steps
• context.call: default 0
• context.invoke: default 0

Use these in combination to tune behavior correctly.

---

Verbose Mode Diagnostics

Enable verbose logging to debug rare edge cases:

serve(handler, { verbose: true });

Useful warnings include:

• Localhost in workflow URL
• Duplicate step execution
• Network response anomalies
• Parallel context.call race warnings

Each of these indicates environmental or routing issues rather than workflow logic problems.

---

HTTPS Protocol Issues (Proxy Environments)

If deployed behind a proxy that downgrades HTTPS → HTTP (e.g., Railway), the SDK may infer the wrong protocol.

Fix: explicitly set:

UPSTASH_WORKFLOW_URL=https://your-deployment-url

---

Workflow Stuck on First Step

If the first step appears stuck:

• Check step logs in the dashboard
• Ensure the endpoint URL is correct and reachable
• Verify SDK versions
• If inference is failing, explicitly set baseUrl:

serve(handler, { baseUrl: "https://your-url" });

---

Non‑workflow Destination Error

Occurs when triggering a non‑workflow endpoint or using mismatched SDK versions.

detected a non-workflow destination for trigger/invoke

Ensure:

• The URL points to a serve() workflow endpoint
• Both caller and workflow endpoint use the latest SDK

---

Vercel Preview Deployment Protection

Preview deployments block external requests unless bypassed.

Steps:

1. Create a bypass secret in Vercel → Deployment Protection 2. Add it to the QStash client + pass it when triggering

import { Client as QStash } from "@upstash/qstash";
import { serve } from "@upstash/workflow/nextjs";

export const { POST } = serve(
  async (context) => {
    // workflow logic
  },
  {
    qstashClient: new QStash({
      token: process.env.QSTASH_TOKEN!,
      headers: {
        "x-vercel-protection-bypass": process.env.VERCEL_AUTOMATION_BYPASS_SECRET!,
      },
    }),
  }
);

// Triggering
await client.trigger({
  url: "https://preview.vercel.app/workflow",
  headers: {
    "x-vercel-protection-bypass": process.env.VERCEL_AUTOMATION_BYPASS_SECRET!,
  },
  body: "hello",
});

Both trigger header and client header are required.