Upstash Workflow

https://upstash.com/docs/workflow
Admin
Upstash Workflow enables developers to write durable, reliable, and performant serverless functions...
Tokens:129,560
Snippets:1,112
Trust Score:10
Update:2 months ago
Show doc for...
Tokens:
Raw
### Create a TanStack Start project

Source: https://upstash.com/docs/workflow/quickstarts/tanstack-start

Initialize a new TanStack Start project using the official CLI.

```bash
pnpm create @tanstack/start@latest
```

--------------------------------

### QStash CLI Output Example

Source: https://upstash.com/docs/workflow/howto/local-development/development-server

Example output from the QStash CLI showing connection details and signing keys.

```plaintext
Upstash QStash development server is runnning at

A default user has been created for you to authorize your requests.
QSTASH_TOKEN=eyJVc2VySUQiOiJkZWZhdWx0VXNlciIsIlBhc3N3b3JkIjoiZGVmYXVsdFBhc3N3b3JkIn0=
QSTASH_CURRENT_SIGNING_KEY=sig_7RvLjqfZBvP5KEUimQCE1pvpLuou
QSTASH_NEXT_SIGNING_KEY=sig_7W3ZNbfKWk5NWwEs3U4ixuQ7fxwE

Sample cURL request:
curl -X POST http://127.0.0.1:8080/v2/publish/https://example.com -H "Authorization: Bearer eyJVc2VySUQiOiJkZWZhdWx0VXNlciIsIlBhc3N3b3JkIjoiZGVmYXVsdFBhc3N3b3JkIn0="

Check out documentation for more details:
https://upstash.com/docs/qstash/howto/local-development
```

--------------------------------

### Install Dependencies

Source: https://upstash.com/docs/workflow/quickstarts/flask

Install the necessary Workflow SDK and Flask packages.

```bash
pip install fastapi uvicorn upstash-workflow
```

--------------------------------

### Start TanStack Start Development Server

Source: https://upstash.com/docs/workflow/quickstarts/tanstack-start

Commands to launch the development server using different package managers.

```bash
pnpm dev
```

```bash
npm run dev
```

--------------------------------

### FastAPI Workflow Setup

Source: https://upstash.com/docs/workflow/examples/authWebhook

Set up an asynchronous workflow in FastAPI using Python. This example demonstrates the basic structure for serving workflows, defining payload types, and initiating asynchronous tasks. Ensure FastAPI and Upstash Workflow libraries are installed.

```python
from fastapi import FastAPI
from typing import Dict, TypedDict
from upstash_workflow.fastapi import Serve
from upstash_workflow import AsyncWorkflowContext

app = FastAPI()
serve = Serve(app)


class UserCreatedPayload(TypedDict):
    name: str
    email: str


class UserStats(TypedDict):
    total_problems_solved: int
    most_interested_topic: str


async def create_user_in_database(name: str, email: str) -> Dict[str, str]:
    print("Creating a user in the database:", name, email)
    return {"userid": "12345"}


async def create_new_user_in_stripe(email: str) -> None:
    # Implement logic to create a new user in Stripe
    print("Creating a user in Stripe for", email)


async def start_trial_in_stripe(email: str) -> None:
    # Implement logic to start a trial in Stripe
    print("Starting a trial of 14 days in Stripe for", email)


```

--------------------------------

### Clone the example repository

Source: https://upstash.com/docs/workflow/quickstarts/nextjs-fastapi

Download the project source code from the official repository.

```bash
git clone https://github.com/upstash/workflow-py.git
cd workflow-py/examples/nextjs-fastapi
```

--------------------------------

### Configure .dev.vars for local server

Source: https://upstash.com/docs/workflow/quickstarts/cloudflare-workers

Example configuration for using the local QStash server.

```txt
QSTASH_URL="http://127.0.0.1:8080"
QSTASH_TOKEN="<QSTASH_TOKEN>"
```

--------------------------------

### Install dependencies

Source: https://upstash.com/docs/workflow/quickstarts/nextjs-fastapi

Install the required Node.js packages.

```bash
npm install
```

--------------------------------

### Install Python Environment and Dependencies

Source: https://upstash.com/docs/workflow/sdk/workflow-py

Commands to initialize a virtual environment and install the necessary packages for Upstash Workflow.

```bash
python -m venv venv
source venv/bin/activate
```

```bash
pip install fastapi uvicorn upstash-workflow
```

--------------------------------

### Install Dependencies

Source: https://upstash.com/docs/workflow/agents/getting-started

Install the required Upstash Workflow and AI packages.

```bash
npm i @upstash/workflow @upstash/workflow-agents ai zod
```

--------------------------------

### Install required packages

Source: https://upstash.com/docs/workflow/agents/features

Install the necessary dependencies for defining tools and using the AI SDK.

```bash
npm i ai mathjs zod @agentic/ai-sdk @agentic/weather @langchain/core @langchain/community
```

--------------------------------

### Run Development Server

Source: https://upstash.com/docs/workflow/agents/getting-started

Start the local Next.js development server.

```bash
npm run dev
```

--------------------------------

### Start QStash CLI

Source: https://upstash.com/docs/workflow/howto/local-development

Command to launch the QStash development server manually.

```bash
npx @upstash/qstash-cli dev
```

--------------------------------

### Example ngrok HTTP Tunnel Command

Source: https://upstash.com/docs/workflow/howto/local-development/local-tunnel

An example of starting an ngrok tunnel for a server running on port 3000. This command exposes your local server to the internet.

```bash
ngrok http 3000
```

--------------------------------

### Install @upstash/workflow

Source: https://upstash.com/docs/workflow/howto/migrations

Commands to install the new workflow SDK using various package managers.

```bash
npm install @upstash/workflow
```

```bash
pnpm install @upstash/workflow
```

```bash
bun add @upstash/workflow
```

--------------------------------

### Example ngrok Output

Source: https://upstash.com/docs/workflow/howto/local-development/local-tunnel

This is an example of the output you will see after starting an ngrok tunnel. The 'Forwarding' line provides the public URL that can be used with Upstash Workflow.

```plaintext
Session Status                online
Account                       <YOUR-NAME> (Plan: Free)
Version                       3.1.0
Region                        Europe (eu)
Latency                       -
Web Interface                 http://127.0.0.1:4040
Forwarding                    https://e02f-2a02-810d-af40-5284-b139-58cc-89df-b740.eu.ngrok.io -> http://localhost:3000
Connections                   ttl     opn     rt1     rt5     p50     p90
                              0       0       0.00    0.00    0.00    0.00
```

--------------------------------

### QStash CLI Output Example

Source: https://upstash.com/docs/workflow/howto/local-development

Sample output from the QStash CLI showing connection details and a cURL request.

```text
Upstash QStash development server is runnning at

A default user has been created for you to authorize your requests.
QSTASH_TOKEN=eyJVc2VySUQiOiJkZWZhdWx0VXNlciIsIlBhc3N3b3JkIjoiZGVmYXVsdFBhc3N3b3JkIn0=
QSTASH_CURRENT_SIGNING_KEY=sig_7RvLjqfZBvP5KEUimQCE1pvpLuou
QSTASH_NEXT_SIGNING_KEY=sig_7W3ZNbfKWk5NWwEs3U4ixuQ7fxwE

Sample cURL request:
curl -X POST http://127.0.0.1:8080/v2/publish/https://example.com -H "Authorization: Bearer eyJVc2VySUQiOiJkZWZhdWx0VXNlciIsIlBhc3N3b3JkIjoiZGVmYXVsdFBhc3N3b3JkIn0="
```

--------------------------------

### Start Flask Application

Source: https://upstash.com/docs/workflow/quickstarts/flask

Launch the Flask development server on port 8000.

```bash
flask --app main run -p 8000
```

--------------------------------

### Start ngrok HTTP Tunnel

Source: https://upstash.com/docs/workflow/howto/local-development/local-tunnel

Make your local server publicly available by starting an ngrok HTTP tunnel. Replace `<PORT>` with the port your local server is running on (e.g., 3000 for a Next.js server).

```bash
ngrok http <PORT>
```

--------------------------------

### Multiple Middlewares Example

Source: https://upstash.com/docs/workflow/howto/middlewares

Demonstrates how to use multiple middlewares together in a workflow.

```APIDOC
## Multiple Middlewares

You can use multiple middlewares together.

### Request Example
```typescript
import { serve } from "@upstash/workflow/nextjs";
import { loggingMiddleware } from "@upstash/workflow";

export const { POST } = serve(
  async (context) => {
    // Your workflow logic
  },
  {
    middlewares: [
      loggingMiddleware,
      errorTrackingMiddleware,
      performanceMiddleware
    ]
  }
);

Middlewares are executed in the order they're provided in the array.
```

--------------------------------

### Install Upstash Workflow and Realtime packages

Source: https://upstash.com/docs/workflow/howto/realtime/basic

Install the necessary dependencies for workflow management, real-time communication, and schema validation.

```bash
npm install @upstash/workflow @upstash/realtime @upstash/redis zod
```

--------------------------------

### Start FastAPI application

Source: https://upstash.com/docs/workflow/quickstarts/fastapi

Launch the FastAPI server using uvicorn with auto-reload enabled.

```bash
uvicorn main:app --reload
```

--------------------------------

### Install Vercel AI SDK dependencies

Source: https://upstash.com/docs/workflow/integrations/aisdk

Install the necessary packages to use OpenAI with the Vercel AI SDK and Zod for schema validation.

```bash
npm install @ai-sdk/openai ai zod
```

```bash
pnpm install @ai-sdk/openai ai zod
```

```bash
bun install @ai-sdk/openai ai zod
```

--------------------------------

### Run Local Development Server

Source: https://upstash.com/docs/workflow/sdk/workflow-py

Commands to set up a tunnel, configure environment variables, and start the FastAPI server.

```bash
ngrok http localhost:8000
```

```bash
export UPSTASH_WORKFLOW_URL=
```

```bash
source .env
```

```bash
uvicorn main:app --reload
```

--------------------------------

### Trigger Workflow with Manual Setup

Source: https://upstash.com/docs/workflow/howto/local-development/development-server

Trigger a workflow using the client with manual configuration.

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

const client = Client()

const { workflowRunId } = await client.trigger({
    url: `http://localhost:3000/api/workflow`,
    retries: 3
});
```

```javascript
const BASE_URL = process.env.VERCEL_URL
  ? `https://${process.env.VERCEL_URL}`
  : `http://localhost:3000`

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

--------------------------------

### Example output text

Source: https://upstash.com/docs/workflow/agents/patterns/parallelization

The generated summary output from the parallel agent execution.

```text
Quantum physics explores the behavior of very small particles, such as atoms and subatomic particles, in the strange world of quantum mechanics, where particles can exist in multiple states simultaneously. Key principles include superposition and entanglement, leading to technological advancements like quantum computers. Relativity, developed by Albert Einstein, consists of special relativity and general relativity, explaining the behavior of objects moving at high speeds and the warping of spacetime by massive objects. String theory proposes that the universe's fundamental building blocks are tiny vibrating strings, aiming to unify the four fundamental forces of nature and suggest extra dimensions beyond the familiar ones.
```

--------------------------------

### Source Environment Variables

Source: https://upstash.com/docs/workflow/quickstarts/flask

Load environment variables from a .env file before starting the application.

```bash
source .env
```

--------------------------------

### Basic Webhook Endpoint Setup - Python

Source: https://upstash.com/docs/workflow/howto/use-webhooks

Utilize the `Serve` class from `upstash_workflow.fastapi` to set up a webhook endpoint with FastAPI. The `initial_payload_parser` can be customized.

```python
from fastapi import FastAPI
from upstash_workflow.fastapi import Serve
from upstash_workflow import AsyncWorkflowContext

app = FastAPI()
serve = Serve(app)


def initial_payload_parser(payload):
    return payload


@serve.post("/api/example", initial_payload_parser=initial_payload_parser)
async def example(context: AsyncWorkflowContext[str]) -> None:
    # Your webhook handling logic here

```

--------------------------------

### Run and Trigger Workflow

Source: https://upstash.com/docs/workflow/quickstarts/nuxt

Commands to start the development server and trigger the workflow endpoint via a POST request.

```bash
npm run dev
```

```bash
curl -X POST https://localhost:3000/api/workflow

# result: {"workflowRunId":"wfr_xxxxxx"}
```

--------------------------------

### Resume Workflow Runs

Source: https://upstash.com/docs/workflow/basics/client/dlq/resume

Examples demonstrating how to resume workflow runs using specific IDs, filters, or by resuming all entries with pagination.

```ts
const { messages } = await client.dlq.list();

const response = await client.dlq.resume(messages[0].dlqId, {
  flowControl: {
    key: "my-flow-control-key",
    parallelism: 10,
  },
  retries: 3,
});
```

```ts
const response = await client.dlq.resume(["dlq-12345", "dlq-67890"]);
```

```ts
let cursor: string | undefined;
do {
  const result = await client.dlq.resume({
    filter: {
      label: "my-label",
      workflowUrl: "https://example.com/workflow",
    },
    cursor,
  });
  cursor = result.cursor;
} while (cursor);
```

```ts
let cursor: string | undefined;
do {
  const result = await client.dlq.resume({ all: true, cursor });
  cursor = result.cursor;
} while (cursor);
```

--------------------------------

### Create Environment File

Source: https://upstash.com/docs/workflow/quickstarts/express

Initialize the .env file in the project root.

```bash
touch .env
```

--------------------------------

### Install Agents Package

Source: https://upstash.com/docs/workflow/howto/migrations

Install the separate package required for Agents functionality.

```bash
npm install @upstash/workflow-agents
```

--------------------------------

### Initialize the Workflow Client

Source: https://upstash.com/docs/workflow/basics/client

Create a new client instance using your QStash credentials. The client is stateless and can be reused throughout your application.

```javascript
import { Client } from "@upstash/workflow"

const client = new Client({
  baseUrl: process.env.QSTASH_URL!,
  token: process.env.QSTASH_TOKEN!
})
```

--------------------------------

### Error Tracking Middleware Example

Source: https://upstash.com/docs/workflow/howto/middlewares

An example of a middleware that sends errors to an external monitoring service.

```APIDOC
## Error Tracking Middleware

Send errors to an external monitoring service.

### Request Example
```typescript
import { WorkflowMiddleware } from "@upstash/workflow";

const errorTrackingMiddleware = new WorkflowMiddleware({
  name: "error-tracking",
  callbacks: {
    onError: async ({ workflowRunId, error }) => {
      await fetch("https://your-monitoring-service.com/errors", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({
          workflowRunId,
          error: error.message,
          stack: error.stack,
          timestamp: new Date().toISOString()
        })
      });
    }
  }
});
```
```

--------------------------------

### Agent Response Example

Source: https://upstash.com/docs/workflow/agents/patterns/prompt-chaining

This is an example of the output generated by the chained agents, summarizing the work of famous physicists.

```text
Albert Einstein was a German physicist known for his theory of relativity and the famous equation E=mc^2. He made significant contributions to quantum mechanics and was awarded the Nobel Prize in Physics in 1921. 

Isaac Newton, an English polymath, was a key figure in the Scientific Revolution and the Enlightenment. He is famous for his laws of motion and universal gravitation, as outlined in his book "PhilosophiÃ¦ Naturalis Principia Mathematica."

Marie Curie, a Polish-French physicist and chemist, conducted pioneering research on radioactivity and was the first woman to win a Nobel Prize. She is the only person to win Nobel Prizes in two scientific fields and her work has had a lasting impact on physics and chemistry.

```

--------------------------------

### Initialize Virtual Environment

Source: https://upstash.com/docs/workflow/quickstarts/flask

Create and activate a Python virtual environment for the project.

```bash
python -m venv venv
source venv/bin/activate
```

--------------------------------

### Configure Environment Variables

Source: https://upstash.com/docs/workflow/agents/getting-started

Set up local development environment variables for QStash and OpenAI.

```txt
QSTASH_DEV=true

OPENAI_API_KEY=<OPENAI_API_KEY>
```

--------------------------------

### Create .env.local file

Source: https://upstash.com/docs/workflow/quickstarts/nuxt

Initialize the environment configuration file in the project root.

```bash
touch .env.local
```

--------------------------------

### Initialize Next.js Project

Source: https://upstash.com/docs/workflow/agents/getting-started

Create a new Next.js project to begin development.

```bash
npx create-next-app@latest [project-name] [options]
```

--------------------------------

### Configure Production Environment Variables

Source: https://upstash.com/docs/workflow/quickstarts/express

Set the QSTASH_TOKEN and UPSTASH_WORKFLOW_URL for production connectivity.

```txt
QSTASH_TOKEN="***"
UPSTASH_WORKFLOW_URL="<UPSTASH_WORKFLOW_URL>"
```

--------------------------------

### OpenAPI Specification for Get Signing Keys

Source: https://upstash.com/docs/workflow/api-reference/signing-keys/get-signing-keys

This OpenAPI 3.1.0 specification defines the GET /v2/keys endpoint for retrieving signing keys. It includes request parameters, response schemas, and authentication methods.

```yaml
openapi: 3.1.0
info:
  title: Upstash Workflow REST API
  description: >
    Upstash Workflow is a serverless workflow orchestration service built on top
    of Upstash QStash and Upstash Redis.
  version: 2.0.0
  contact:
    name: Upstash
    url: https://upstash.com
servers:
  - url: https://qstash-{region}.upstash.io
    description: Regional
    variables:
      region:
        default: eu-central-1
        enum:
          - us-east-1
          - eu-central-1
security:
  - bearerAuth: []
  - bearerAuthQuery: []
paths:
  /v2/keys:
    get:
      tags:
        - Signing Keys
      summary: Get Signing Keys
      description: Retrieve your current and next signing keys
      responses:
        '200':
          description: Signing keys retrieved successfully
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/SigningKeys'
components:
  schemas:
    SigningKeys:
      type: object
      properties:
        current:
          type: string
          description: The current signing key.
        next:
          type: string
          description: The next signing key.
  securitySchemes:
    bearerAuth:
      type: http
      scheme: bearer
      bearerFormat: JWT
      description: QStash authentication token
    bearerAuthQuery:
      type: apiKey
      in: query
      name: qstash_token
      description: QStash authentication token passed as a query parameter

```

--------------------------------

### OpenAPI Specification for Get Global Parallelism

Source: https://upstash.com/docs/workflow/api-reference/flow-control/get-global-parallelism

This OpenAPI 3.1.0 specification defines the GET /v2/globalParallelism endpoint for retrieving global parallelism information. It outlines request parameters, response schemas, and security schemes.

```yaml
openapi: 3.1.0
info:
  title: Upstash Workflow REST API
  description: >
    Upstash Workflow is a serverless workflow orchestration service built on top
    of Upstash QStash and Upstash Redis.
  version: 2.0.0
  contact:
    name: Upstash
    url: https://upstash.com
servers:
  - url: https://qstash-{region}.upstash.io
    description: Regional
    variables:
      region:
        default: eu-central-1
        enum:
          - us-east-1
          - eu-central-1
security:
  - bearerAuth: []
  - bearerAuthQuery: []
paths:
  /v2/globalParallelism:
    get:
      tags:
        - Flow Control
      summary: Get Global Parallelism
      description: >-
        Returns the current global parallelism usage across all flow control
        keys
      responses:
        '200':
          description: Global parallelism info retrieved successfully
          content:
            application/json:
              schema:
                type: object
                properties:
                  parallelismMax:
                    type: integer
                    description: The configured maximum global parallelism
                  parallelismCount:
                    type: integer
                    description: >-
                      The current number of messages running globally in
                      parallel
components:
  securitySchemes:
    bearerAuth:
      type: http
      scheme: bearer
      bearerFormat: JWT
      description: QStash authentication token
    bearerAuthQuery:
      type: apiKey
      in: query
      name: qstash_token
      description: QStash authentication token passed as a query parameter

```

--------------------------------

### Get Flow Control Key OpenAPI Specification

Source: https://upstash.com/docs/workflow/api-reference/flow-control/get-flow-control-key-1

This OpenAPI specification defines the GET /v2/keys/rotate endpoint for retrieving flow control key details. It outlines parameters, request/response schemas, and error handling.

```yaml
openapi: 3.1.0
info:
  title: Upstash Workflow REST API
  description: >
    Upstash Workflow is a serverless workflow orchestration service built on top
    of Upstash QStash and Upstash Redis.
  version: 2.0.0
  contact:
    name: Upstash
    url: https://upstash.com
servers:
  - url: https://qstash-{region}.upstash.io
    description: Regional
    variables:
      region:
        default: eu-central-1
        enum:
          - us-east-1
          - eu-central-1
security:
  - bearerAuth: []
  - bearerAuthQuery: []
paths:
  /v2/keys/rotate:
    get:
      tags:
        - Flow Control
      summary: Get Flow Control Key
      description: >
        Get details of a specific Flow Control key.


        Flow Control keys are used to manage concurrency and rate limiting for
        workflow steps.

        This endpoint returns the current waitlist size for the specified flow
        control key,

        which indicates how many workflow steps are currently waiting due to
        flow control constraints.
      parameters:
        - in: path
          name: flowControlKey
          required: true
          schema:
            type: string
          description: The Flow Control key to retrieve.
      responses:
        '200':
          description: Flow control key details
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/FlowControlKey'
        '400':
          description: Bad Request
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Error'
        '401':
          description: Unauthorized
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Error'
        '404':
          description: Flow Control key not found
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Error'
        '500':
          description: Internal Server Error
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Error'
components:
  schemas:
    FlowControlKey:
      type: object
      properties:
        flowControlKey:
          type: string
          description: The flow control key name
        waitlistSize:
          type: integer
          description: The number of messages waiting due to flow control configuration.
    Error:
      type: object
      required:
        - error
      properties:
        error:
          type: string
          description: Error message
  securitySchemes:
    bearerAuth:
      type: http
      scheme: bearer
      bearerFormat: JWT
      description: QStash authentication token
    bearerAuthQuery:
      type: apiKey
      in: query
      name: qstash_token
      description: QStash authentication token passed as a query parameter

```

--------------------------------

### Deploy to Production

Source: https://upstash.com/docs/workflow/quickstarts/nextjs-fastapi

Deploy the project to the production environment on Vercel.

```bash
vercel --prod
```

--------------------------------

### OpenAPI Specification for Get Flow Control Key

Source: https://upstash.com/docs/workflow/api-reference/flow-control/get-flow-control-key

This OpenAPI 3.1.0 specification defines the GET endpoint `/v2/flowControl/{flowControlKey}` for retrieving details of a specific Flow Control key. It includes request parameters, response schemas for success (200) and not found (404) cases, and security scheme definitions.

```yaml
openapi: 3.1.0
info:
  title: Upstash Workflow REST API
  description: >
    Upstash Workflow is a serverless workflow orchestration service built on top
    of Upstash QStash and Upstash Redis.
  version: 2.0.0
  contact:
    name: Upstash
    url: https://upstash.com
servers:
  - url: https://qstash-{region}.upstash.io
    description: Regional
    variables:
      region:
        default: eu-central-1
        enum:
          - us-east-1
          - eu-central-1
security:
  - bearerAuth: []
  - bearerAuthQuery: []
paths:
  /v2/flowControl/{flowControlKey}:
    get:
      tags:
        - Flow Control
      summary: Get Flow Control Key
      description: Get details of a specific Flow Control key
      parameters:
        - name: flowControlKey
          in: path
          required: true
          schema:
            type: string
          description: The Flow Control key to retrieve
      responses:
        '200':
          description: Flow control key details
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/FlowControlKey'
        '404':
          description: Flow Control key not found
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Error'
components:
  schemas:
    FlowControlKey:
      type: object
      properties:
        flowControlKey:
          type: string
          description: The flow control key name
        waitlistSize:
          type: integer
          description: The number of messages waiting due to flow control configuration.
    Error:
      type: object
      required:
        - error
      properties:
        error:
          type: string
          description: Error message
  securitySchemes:
    bearerAuth:
      type: http
      scheme: bearer
      bearerFormat: JWT
      description: QStash authentication token
    bearerAuthQuery:
      type: apiKey
      in: query
      name: qstash_token
      description: QStash authentication token passed as a query parameter

```

--------------------------------

### Navigate to project directory

Source: https://upstash.com/docs/workflow/quickstarts/tanstack-start

Change the current working directory to the newly created project folder.

```bash
cd your-project-name
```

--------------------------------

### Expected Console Output

Source: https://upstash.com/docs/workflow/agents/getting-started

Example logs generated in the console during workflow execution.

```text
Inner thought: I will discuss the future of space
exploration and the potential advancements in
technology and missions.

Final response: The future of space exploration
holds exciting possibilities with advancements
in technology, potential manned missions to
Mars, increased commercial space travel,
and exploration of distant celestial
bodies.
```

--------------------------------

### Customer Onboarding Workflow

Source: https://upstash.com/docs/workflow/examples/customerOnboarding

This workflow registers a new user, sends a welcome email, waits for 3 days, and then periodically checks the user's activity state to send appropriate emails. It requires initial data containing the user's email.

```typescript
import {
  serve
} from "@upstash/workflow/nextjs"

type InitialData = {
  email: string
}

export const { POST } = serve<InitialData>(async (context) => {
  const { email } = context.requestPayload

  await context.run("new-signup", async () => {
    await sendEmail("Welcome to the platform", email)
  })

  await context.sleep("wait-for-3-days", 60 * 60 * 24 * 3)

  while (true) {
    const state = await context.run("check-user-state", async () => {
      return await getUserState()
    })

    if (state === "non-active") {
      await context.run("send-email-non-active", async () => {
        await sendEmail("Email to non-active users", email)
      })
    } else if (state === "active") {
      await context.run("send-email-active", async () => {
        await sendEmail("Send newsletter to active users", email)
      })
    }

    await context.sleep("wait-for-1-month", 60 * 60 * 24 * 30)
  }
})

async function sendEmail(message: string, email: string) {
  // Implement email sending logic here
  console.log(`Sending ${message} email to ${email}`)
}

type UserState = "non-active" | "active"

const getUserState = async (): Promise<UserState> => {
  // Implement user state logic here
  return "non-active"
}

```

```python
from fastapi import FastAPI
from typing import Literal, TypedDict
from upstash_workflow.fastapi import Serve
from upstash_workflow import AsyncWorkflowContext

app = FastAPI()
serve = Serve(app)

UserState = Literal["non-active", "active"]


class InitialData(TypedDict):
    email: str


async def send_email(message: str, email: str) -> None:
    # Implement email sending logic here
    print(f"Sending {message} email to {email}")


async def get_user_state() -> UserState:
    # Implement user state logic here
    return "non-active"


@serve.post("/customer-onboarding")
async def customer_onboarding(context: AsyncWorkflowContext[InitialData]) -> None:
    email = context.request_payload["email"]

    async def _new_signup() -> None:
        await send_email("Welcome to the platform", email)

    await context.run("new-signup", _new_signup)

    await context.sleep("wait-for-3-days", 60 * 60 * 24 * 3)

    while True:

        async def _check_user_state() -> UserState:
            return await get_user_state()

        state: UserState = await context.run("check-user-state", _check_user_state)

        if state == "non-active":

            async def _send_email_non_active() -> None:
                await send_email("Email to non-active users", email)

            await context.run("send-email-non-active", _send_email_non_active)
        else:

            async def _send_email_active() -> None:
                await send_email("Send newsletter to active users", email)

            await context.run("send-email-active", _send_email_active)

        await context.sleep("wait-for-1-month", 60 * 60 * 24 * 30)

```

--------------------------------

### GET /v2/workflows/dlq

Source: https://upstash.com/docs/workflow/api-reference/dlq/list-failed-workflow-runs

List and paginate through all failed workflow runs currently in the DLQ.

```APIDOC
## GET /v2/workflows/dlq

### Description
List and paginate through all failed workflow runs currently in the DLQ. Failed workflows end up in the DLQ after exhausting all retry attempts.

### Method
GET

### Endpoint
/v2/workflows/dlq

### Parameters
#### Query Parameters
- **cursor** (string) - Optional - Pagination cursor. If provided, returns the next page of results.
- **count** (integer) - Optional - The maximum number of failed workflow runs to return per page (default: 100, max: 100).
- **fromDate** (integer) - Optional - Filter by starting date in milliseconds (Unix timestamp). This is inclusive.
- **toDate** (integer) - Optional - Filter by ending date in milliseconds (Unix timestamp). This is inclusive.
- **workflowUrl** (array) - Optional - Filter by workflow URL. Supports multiple values.
- **workflowRunId** (string) - Optional - Filter by workflow run ID.
- **workflowCreatedAt** (integer) - Optional - Filter by workflow creation timestamp in milliseconds (Unix timestamp).
- **label** (array) - Optional - Filter by label assigned to the workflow run. Supports multiple values.
- **failureFunctionState** (array) - Optional - Filter by failure function state (CALLBACK_INPROGRESS, CALLBACK_SUCCESS, CALLBACK_FAIL, CALLBACK_CANCELED).
- **callerIp** (array) - Optional - Filter by IP address of the publisher. Supports multiple values.
```

--------------------------------

### Enable Automatic Dev Server

Source: https://upstash.com/docs/workflow/howto/local-development

Set this environment variable to automatically download and connect to the QStash dev server.

```text
QSTASH_DEV=true
```

--------------------------------

### Configure Local QStash Environment Variables

Source: https://upstash.com/docs/workflow/quickstarts/flask

Add local QStash URL and token to the .env file.

```bash
export QSTASH_URL="http://127.0.0.1:8080"
export QSTASH_TOKEN="<QSTASH_TOKEN>"
```

--------------------------------

### Next.js Workflow for User Onboarding

Source: https://upstash.com/docs/workflow/examples/authWebhook

Implement a user onboarding workflow in Next.js using TypeScript. This example handles user creation, Stripe integration, email notifications, and trial period management. Ensure necessary imports are present.

```typescript
import { serve } from "@upstash/workflow/nextjs";
import { WorkflowContext } from '@upstash/qstash/workflow'

/**
 * This can be the payload of the user created webhook event coming from your
 * auth provider (e.g. Firebase, Auth0, Clerk etc.)
 */
type UserCreatedPayload = {
  name: string;
  email: string;
};

export const { POST } = serve<UserCreatedPayload>(async (context) => {
  const { name, email } = context.requestPayload;

  const { userid } = await context.run("sync user", async () => {
    return await createUserInDatabase({ name, email });
  });

  await context.run("create new user in stripe", async () => {
    await createNewUserInStripe(email);
  });

  await context.run("start trial in Stripe", async () => {
    await startTrialInStripe(email);
  });

  await context.run("send welcome email", async () => {
    await sendEmail(
      email,
      "Welcome to our platform!, You have 14 days of free trial."
    );
  });

  await context.sleep("wait", 7 * 24 * 60 * 60);

  // get user stats and send email with them
  const stats = await context.run("get user stats", async () => {
    return await getUserStats(userid);
  });
  await sendProblemSolvedEmail({context, email, stats});

  // wait until there are two days to the end of trial period
  // and check upgrade status
  await context.sleep("wait for trial warning", 5 * 24 * 60 * 60);
  const isUpgraded = await context.run("check upgraded plan", async () => {
    return await checkUpgradedPlan(email);
  });

  // end the workflow if upgraded
  if (isUpgraded) return;

  await context.run("send trial warning email", async () => {
    await sendEmail(
      email,
      "Your trial is about to end in 2 days. Please upgrade your plan to keep using our platform."
    );
  });

  await context.sleep("wait for trial end", 2 * 24 * 60 * 60);

  await context.run("send trial end email", async () => {
    await sendEmail(
      email,
      "Your trial has ended. Please upgrade your plan to keep using our platform."
    );
  });
});

async function sendProblemSolvedEmail({
  context: WorkflowContext<UserCreatedPayload>
  email: string,
  stats: { totalProblemsSolved: number }
}) {
  if (stats.totalProblemsSolved === 0) {
    await context.run("send no answers email", async () => {
      await sendEmail(
        email,
        "Hey, you haven't solved any questions in the last 7 days..."
      );
    });
  } else {
    await context.run("send stats email", async () => {
      await sendEmail(
        email,
        `You have solved ${stats.totalProblemsSolved} problems in the last 7 days. Keep it up!`
      );
    });
  }
}

async function createUserInDatabase({
  name,
  email,
}: {
  name: string;
  email: string;
}) {
  console.log("Creating a user in the database:", name, email);
  return { userid: "12345" };
}

async function createNewUserInStripe(email: string) {
  // Implement logic to create a new user in Stripe
  console.log("Creating a user in Stripe for", email);
}

async function startTrialInStripe(email: string) {
  // Implement logic to start a trial in Stripe
  console.log("Starting a trial of 14 days in Stripe for", email);
}

async function getUserStats(userid: string) {
  // Implement logic to get user stats
  console.log("Getting user stats for", userid);
  return {
    totalProblemsSolved: 10_000,
    mostInterestedTopic: "JavaScript",
  };
}

async function checkUpgradedPlan(email: string) {
  // Implement logic to check if the user has upgraded the plan
  console.log("Checking if the user has upgraded the plan", email);
  return false;
}

async function sendEmail(email: string, content: string) {
  // Implement logic to send an email
  console.log("Sending email to", email, content);
}

```

--------------------------------

### Get Signing Keys

Source: https://upstash.com/docs/workflow/api-reference/signing-keys/get-signing-keys

Retrieves the current and next signing keys for your Upstash account.

```APIDOC
## GET /v2/keys

### Description
Retrieve your current and next signing keys.

### Method
GET

### Endpoint
/v2/keys

### Response
#### Success Response (200)
- **current** (string) - The current signing key.
- **next** (string) - The next signing key.

#### Response Example
```json
{
  "current": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c",
  "next": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjM0NTY3ODkwIiwibmFtZSI6IkpvaG4gRG9lIiwiaWF0IjoxNTE2MjM5MDIyfQ.SflKxwRJSMeKKF2QT4fwpMeJf36POk6yJV_adQssw5c"
}
```
```

--------------------------------

### Create .dev.vars file

Source: https://upstash.com/docs/workflow/quickstarts/cloudflare-workers

Initializes the environment variable file in the project root.

```bash
touch .dev.vars
```

--------------------------------

### Initialize Vercel Deployment

Source: https://upstash.com/docs/workflow/quickstarts/nextjs-fastapi

Initialize the Vercel project configuration for deployment.

```bash
vercel
```

--------------------------------

### Run Workflow Endpoint Locally

Source: https://upstash.com/docs/workflow/quickstarts/cloudflare-workers

Start the Cloudflare Worker locally using the Wrangler CLI.

```bash
npm run wrangler dev
```
Tokens:
Raw