Advanced Options

Defines a function that executes if the workflow fails after all retries are exhausted.

For details, see failureFunction.

export const { POST } = serve<string>(  async (context) => { ... },  {    failureFunction: async ({      context,      // context during failure      failStatus,   // failure status      failResponse, // failure message      failHeaders,  // failure headers      failStack     // failure stack trace (if available)    }) => {      // handle the failure    }  });

async def failure_function(  context,       # context during failure  fail_status,   # failure status  fail_response, # failure message  fail_headers   # failure headers):  # handle the failure  pass@serve.post("/api/example", failure_function=failure_function)async def example(context: AsyncWorkflowContext[str]) -> None: ...

The failureUrl option defines an external endpoint that will be called if the workflow fails after all retries are exhausted.

This option is an advanced alternative to failureFunction. For more details, see Advanced failureUrl Option.

@serve.post("/api/example", failureUrl="https://<YOUR-FAILURE-ENDPOINT>/...")async def example(context: AsyncWorkflowContext[str]) -> None: ...

Defines the number of retry attempts if a workflow step fails. The default value is 3.

For details, see retry configuration.

@serve.post("/api/example", retries=3)async def example(context: AsyncWorkflowContext[str]) -> None: ...

An array of middleware instances that intercept workflow lifecycle and debug events.

Middlewares allow you to hook into various stages of workflow execution (before/after steps, run start/completion) and debug events (errors, warnings, info logs).

For details and examples, see Middlewares.

import { serve } from "@upstash/workflow/nextjs";import { loggingMiddleware } from "@upstash/workflow";export const { POST } = serve<string>(  async (context) => { ... },  {    middlewares: [loggingMiddleware]  });

Enables custom parsing of the initial request payload.

Use this option if the incoming payload is not plain JSON or a simple string. The parser function lets you transform the raw request into a strongly typed object before workflow execution begins.

type InitialPayload = {  foo: string;  bar: number;};// 👇 1: provide initial payload typeexport const { POST } = serve<InitialPayload>(  async (context) => {    // 👇 3: parsing result is available as requestPayload    const payload: InitialPayload = context.requestPayload;  },  {    // 👇 2: custom parsing for initial payload    initialPayloadParser: (initialPayload) => {      const payload: InitialPayload = parsePayload(initialPayload);      return payload;    },  });

@dataclassclass InitialPayload:    foo: str    bar: intdef initial_payload_parser(initial_payload: str) -> InitialPayload:    return parse_payload(initial_payload)@serve.post("/api/example", initial_payload_parser=initial_payload_parser)async def example(context: AsyncWorkflowContext[InitialPayload]) -> None:    payload: InitialPayload = context.request_payload

Alternative to initialPayloadParser, you can pass a schema in the TypeScript SDK.

The schema is used to validate and parse the initial request payload automatically using Zod.

import { z } from "zod";const parameters = z.object({ expression: z.string() });export const { POST } = serve(  async (context) => {    // context.requestPayload is typed as `{ expression: string }`    const payload = context.requestPayload;  },  {    schema: parameters,  });

Specifies the full endpoint URL of the workflow, including the route path.

By default, Upstash Workflow infers the URL from request.url when scheduling the next step. However, in some environments, request.url may resolve to an internal or unreachable address.

Use this option when running behind a proxy, reverse proxy, or local tunnel during development where request.url cannot be used directly.

export const { POST } = serve<string>(  async (context) => { ... },  {    url: "https://<YOUR-DEPLOYED-APP>.com/api/workflow"  });

@serve.post("/api/example", url="https://<YOUR-DEPLOYED-APP>.com/api/workflow")async def example(context: AsyncWorkflowContext[str]) -> None: ...

Similar to url, but baseUrl only overrides the base portion of the inferred URL rather than replacing the entire path. This is useful when you want to preserve the route structure while changing only the host or scheme.

export const { POST } = serve<string>(  async (context) => {    ...  },  // options:  {    baseUrl: "<LOCAL-TUNNEL-PUBLIC-URL>"  });

@serve.post("/api/example", base_url="<LOCAL-TUNNEL-PUBLIC-URL>")async def example(context: AsyncWorkflowContext[str]) -> None: ...

Use qstashClient if you want to provide your own QStash client instead of letting Workflow use the default from environment variables.

This is useful if you're working with multiple QStash projects in the same app.

import { Client } from "@upstash/qstash";import { serve } from "@upstash/workflow/nextjs";export const { POST } = serve(  async (context) => { ... },  {    qstashClient: new Client({ token: "<QSTASH_TOKEN>" })  });

from qstash import AsyncQStash@serve.post("/api/example", qstash_client=AsyncQStash(os.environ["QSTASH_TOKEN"]))async def example(context: AsyncWorkflowContext[str]) -> None: ...

The Receiver verifies that every request to your endpoint actually comes from QStash, blocking anyone else from triggering your workflow.

The receiver option allows you to pass a QStash Receiver explicitly.

By default, Workflow initializes the Receiver automatically using the environment variables QSTASH_CURRENT_SIGNING_KEY and QSTASH_NEXT_SIGNING_KEY.

This is useful if you're working with multiple QStash projects in the same app.

import { Receiver } from "@upstash/qstash";import { serve } from "@upstash/workflow/nextjs";export const { POST } = serve<string>(  async (context) => { ... },  {    receiver: new Receiver({      currentSigningKey: "<QSTASH_CURRENT_SIGNING_KEY>",      nextSigningKey: "<QSTASH_NEXT_SIGNING_KEY>",    })  });

from qstash import Receiver@serve.post(    "/api/example",    receiver=Receiver(        current_signing_key=os.environ["QSTASH_CURRENT_SIGNING_KEY"],        next_signing_key=os.environ["QSTASH_NEXT_SIGNING_KEY"],    ),)async def example(context: AsyncWorkflowContext[str]) -> None:    ...

By default, Workflow uses process.env to read credentials and initialize QStash. If you're in an environment where process.env isn't available, or you want to inject values manually, you can pass them with env.

Inside your workflow, these values are also exposed on context.env.

import { Receiver } from "@upstash/qstash";import { serve } from "@upstash/workflow/nextjs";export const { POST } = serve<string>(  async (context) => {    // the env option will be available in the env field of the context:    const env = context.env;  },  {    env: {        QSTASH_URL: "<QSTASH_URL>",        QSTASH_TOKEN: "<QSTASH_TOKEN>",        QSTASH_CURRENT_SIGNING_KEY: "<QSTASH_CURRENT_SIGNING_KEY>",        QSTASH_NEXT_SIGNING_KEY: "<QSTASH_NEXT_SIGNING_KEY>",    }  });

@serve.post(    "/api/example",    env={        "QSTASH_CURRENT_SIGNING_KEY": os.environ["QSTASH_CURRENT_SIGNING_KEY"],        "QSTASH_NEXT_SIGNING_KEY": os.environ["QSTASH_NEXT_SIGNING_KEY"],    },)async def example(context: AsyncWorkflowContext[str]) -> None:    ...

Enables verbose mode to print detailed logs of workflow execution to the application's stdout.

Verbose mode is disabled by default.

export const { POST } = serve<string>(  async (context) => { ... },  {    verbose: true  });

Disables anonymous telemetry data collection for this workflow endpoint. Since we don't collect telemetry in Python SDK, this option is only available in the TypeScript SDK.

By default, the Upstash Workflow SDK collects anonymous telemetry data to help improve the service. The collected data includes:

• SDK version
• Platform (Vercel, AWS, etc.)
• Runtime version (Node.js, Python, etc.)

Set disableTelemetry to true to opt out of telemetry for this specific workflow endpoint.

export const { POST } = serve<string>(  async (context) => { ... },  {    disableTelemetry: true  });

@serve.post("/api/example", disable_telemetry=True)async def example(context: AsyncWorkflowContext[str]) -> None: ...