Webhooks - Upstash Documentation

This guide explains how to handle webhooks effectively in your Upstash Workflow applications. We'll walk through:

setting up webhook endpoints
verifying webhook requests
and processing webhook events

You can find the project source code on GitHub.

Overview

Webhooks allow external services to notify your application when events occur. For example, you can use webhooks to receive notifications when a new order is placed in your e-commerce store, a new user signs up, or a new message is sent in your chat application.

Upstash Workflow provides a simple way to receive these events and trigger workflows based on the incoming data autonomously.

Setting Up Webhook Endpoints

Basic Setup

To create a webhook endpoint, use the serve function from @upstash/workflow:

TypeScript

import { serve } from "@upstash/workflow/nextjs";export const { POST } = serve(  async (context) => {    // Your webhook handling logic here  },  {    initialPayloadParser: (payload) => {      return payload;    },  });

Python

from fastapi import FastAPIfrom upstash_workflow.fastapi import Servefrom upstash_workflow import AsyncWorkflowContextapp = 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

Request Validation

Always validate incoming webhook requests to ensure they're legitimate. This way, no one other than the original webhook source can trigger your workflow. Here's an example using Clerk webhooks with Svix:

Validate and Parse in Workflow - TypeScript

export const { POST } = serve<string>(async (context) => {	const payloadString = context.requestPayload;	const headerPayload = context.headers;    let event: WebhookEvent;    try {    	event = await validateRequest(payloadString, headerPayload);    } catch {    	return    }    // Next steps based on the event})

Validate and Parse in Workflow - Python

async def validate_request(payload_string: str, header_payload: dict):    # Validate the request    pass@serve.post("/api/example")async def example(context: AsyncWorkflowContext[str]) -> None:    payload_string = context.request_payload    header_payload = context.headers    try:        event = await validate_request(payload_string, header_payload)    except:        return    # Next steps based on the event

Validation Function - TypeScript

import { Webhook } from "svix";import { WebhookEvent } from "@clerk/nextjs/server";const webhookSecret = "YOUR_WEBHOOK_SECRET";async function validateRequest(payloadString: string, headerPayload: Headers) {  const svixHeaders = {    "svix-id": headerPayload.get("svix-id") as string,    "svix-timestamp": headerPayload.get("svix-timestamp") as string,    "svix-signature": headerPayload.get("svix-signature") as string,  };  const wh = new Webhook(webhookSecret);  return wh.verify(payloadString, svixHeaders) as WebhookEvent;}

Handling Webhook events

Use the context.run method to process webhook events in discrete, trackable steps:

TypeScript

export const { POST } = serve(async (context) => {  // ... Parse and validate the incoming request  const user = await context.run<false | UserPayload>(    "handle-webhook-event",    async () => {      if (event.type === "user.created") {        const { id: clerkUserId, email_addresses, first_name } = event.data;        const primaryEmail = email_addresses.find(          (email) => (email.id = event.data.primary_email_address_id)        );        if (!primaryEmail) {          return false;        }        return {          event: event.type,          userId: clerkUserId,          email: primaryEmail.email_address,          firstName: first_name,        } as UserPayload;      }      return false;    }  );});

Python

@serve.post("/api/example")async def example(context: AsyncWorkflowContext[str]) -> None:    # ... Parse and validate the incoming request    async def _handle_webhook_event():        if event.type == "user.created":            clerk_user_id = event.data["id"]            email_addresses = event.data["email_addresses"]            first_name = event.data["first_name"]            primary_email = next(                (                    email                    for email in email_addresses                    if email.id == event.data["primary_email_address_id"]                ),                None,            )            if not primary_email:                return False            return {                "event": event.type,                "user_id": clerk_user_id,                "email": primary_email["email_address"],                "first_name": first_name,            }        return False    user = await context.run("handle-webhook-event", _handle_webhook_event)

After validating the webhook and extracting the initial user data, you'll often need to perform additional operations like creating customer records, sending welcome emails etc.

TypeScript

export const { POST } = serve(async (context) => {  // ... Previous validation and user data extraction  if (!user) {    return;  }  const customer = await context.run("create-stripe-customer", async () => {    return await stripe.customers.create({      email: user.email,      name: `${user.firstName} ${user.lastName}`,      metadata: {        userId: user.userId,      },    });  });  /// ... Additional steps});

Python

@serve.post("/api/example")async def example(context: AsyncWorkflowContext[str]) -> None:    # ... Previous validation and user data extraction    if not user:        return    async def _create_stripe_customer():        return await stripe.customers.create(            email=user["email"],            name=f"{user['first_name']} {user['last_name']}",            metadata={"user_id": user["user_id"]},        )    customer = await context.run("create-stripe-customer", _create_stripe_customer)    # ... Additional steps

You're now ready to perform any operation in the following steps.