Pipecat Cloud Webhooks

import express from "express";
import { Webhook, WebhookVerificationError } from "standardwebhooks";

const app = express();
const wh = new Webhook(process.env.PIPECAT_WEBHOOK_SECRET!);

// Deliveries name the signature headers with a provider prefix; Standard
// Webhooks libraries look for them unprefixed. Empty strings rather than
// `undefined` so a request with no signature fails verification (a 400)
// instead of throwing (a 500).
function signatureHeaders(headers: Record<string, unknown>) {
  return {
    "webhook-id": (headers["svix-id"] as string) ?? "",
    "webhook-timestamp": (headers["svix-timestamp"] as string) ?? "",
    "webhook-signature": (headers["svix-signature"] as string) ?? "",
  };
}

// `express.raw` is essential: the JSON body parser would hand you an object,
// and the signature is over the bytes that arrived on the wire.
app.post(
  "/webhooks/pipecat",
  express.raw({ type: "application/json" }),
  (req, res) => {
    let event: any;
    try {
      event = wh.verify(req.body, signatureHeaders(req.headers));
    } catch (err) {
      if (err instanceof WebhookVerificationError) {
        // Wrong secret, tampered body, or a replayed old delivery.
        return res.status(400).send("Invalid signature");
      }
      throw err;
    }

    // Acknowledge immediately, then do the real work out of band. The
    // response has already gone out, so nothing downstream will surface a
    // failure in here and the delivery will not be retried: catch it
    // yourself, and hand off to a real queue in production.
    res.status(204).end();
    handleEvent(event).catch((err) => {
      console.error("Webhook handler failed", err);
    });
  },
);

async function handleEvent(event: any) {
  switch (event.event_type) {
    case "build.succeeded":
      console.log(
        `Build ${event.data.build_id} produced ${event.data.image_uri}`,
      );
      break;
    case "deployment.updated":
      console.log(
        `${event.data.service_name} redeployed in ${event.data.region}`,
      );
      break;
    default:
      // Subscribing to a new event type in the dashboard should not break
      // a running handler, so ignore anything you do not handle yet.
      break;
  }
}

app.listen(3000);

import os

from fastapi import BackgroundTasks, FastAPI, Request, Response
from standardwebhooks import Webhook, WebhookVerificationError

app = FastAPI()
wh = Webhook(os.environ["PIPECAT_WEBHOOK_SECRET"])


def signature_headers(headers) -> dict[str, str]:
    """Deliveries name the signature headers with a provider prefix; Standard
    Webhooks libraries look for them unprefixed. Defaulting to "" rather than
    raising means a request with no signature fails verification (a 400)
    instead of erroring (a 500)."""
    return {
        "webhook-id": headers.get("svix-id", ""),
        "webhook-timestamp": headers.get("svix-timestamp", ""),
        "webhook-signature": headers.get("svix-signature", ""),
    }


@app.post("/webhooks/pipecat")
async def pipecat_webhook(request: Request, background_tasks: BackgroundTasks):
    # The raw bytes, not a parsed model: the signature is over what arrived.
    payload = await request.body()

    try:
        event = wh.verify(payload, signature_headers(request.headers))
    except WebhookVerificationError:
        # Wrong secret, tampered body, or a replayed old delivery.
        return Response(status_code=400)

    # Acknowledge immediately, then do the real work out of band.
    background_tasks.add_task(handle_event, event)
    return Response(status_code=204)


async def handle_event(event: dict):
    event_type = event["event_type"]
    data = event["data"]

    if event_type == "build.succeeded":
        print(f"Build {data['build_id']} produced {data.get('image_uri')}")
    elif event_type == "deployment.updated":
        print(f"{data['service_name']} redeployed in {data['region']}")
    # Subscribing to a new event type in the dashboard should not break a
    # running handler, so ignore anything you do not handle yet.

const messageId = req.headers["svix-id"] as string;

if (await seen.has(messageId)) {
  return res.status(204).end(); // Already handled; acknowledge and stop.
}
await seen.add(messageId, { ttlSeconds: 60 * 60 * 24 });

message_id = request.headers["svix-id"]

if await seen.has(message_id):
    return Response(status_code=204)  # Already handled; acknowledge and stop.
await seen.add(message_id, ttl_seconds=60 * 60 * 24)