Verify a Webhook

WebhooksAPI.verify() checks the HMAC signature on an inbound webhook that Vendo signed before forwarding it to your deployment. It returns the parsed event or raises ValidationError on a bad / missing / stale signature.

Verification is entirely local — no network call. The secret comes from the VENDO_WEBHOOK_SECRET env var (per-deployment, injected automatically in Vendo mode).

Outbound lifecycle events — connection.connected, billing.balance_changed, etc. — are delivered over SSE, not as outbound HTTP webhooks. Subscribe with vendo.events.subscribe() (Python) or vendo.events.subscribe() (TypeScript async iterable) to consume those. See the SSE events example below. The webhooks.verify helper on this page is for inbound provider webhooks that Vendo's hooks worker re-signs before forwarding (Telegram is the only provider on that path today).

Verify a forwarded webhook

from vendo import Vendo
from vendo.errors import ValidationError

client = Vendo()

def handle_webhook(headers: dict, body: str) -> dict:
    # body MUST be the raw request body as a string (NOT bytes).
    # If your framework gives you bytes, decode it first: body.decode("utf-8").
    try:
        event = client.webhooks.verify(headers, body)
    except ValidationError:
        return {"status": 400, "error": "invalid signature"}

    # event.type is the event kind; event.data is the parsed payload.
    if event.type == "connection.connected":
        slug = event.data.get("slug")
        if isinstance(slug, str):
            client.invalidate(slug)

    return {"status": 200, "ok": True, "event_id": event.id, "at": event.occurred_at}

import { Vendo, ValidationError } from "@vendodev/sdk";

const vendo = new Vendo();

// Next.js App Router route handler.
export async function POST(request: Request) {
  const rawBody = await request.text();
  const headers = Object.fromEntries(request.headers.entries());

  let event;
  try {
    event = vendo.webhooks.verify(headers, rawBody);
  } catch (e) {
    if (e instanceof ValidationError) {
      return Response.json({ error: "invalid signature" }, { status: 400 });
    }
    throw e;
  }

  if (event.type === "connection.connected") {
    const slug = event.data["slug"];
    if (typeof slug === "string") {
      vendo.invalidate(slug);
    }
  }

  return Response.json({ ok: true, eventId: event.id, at: event.occurredAt });
}

How signature verification works

Vendo signs the forwarded body with HMAC-SHA256 over a timestamp-prefixed payload and sends two headers:

verify():

1. Reads both headers. Raises ValidationError if either is missing.
2. Rejects timestamps more than 5 minutes (max_age_sec=300) out of clock alignment to prevent replay.
3. Computes the expected HMAC over "${timestamp}.${body}" and compares it to Vendo-Signature with hmac.compare_digest / timingSafeEqual.
4. Parses the body as JSON and returns a WebhookEvent.

The body argument must be the raw request body as a string (Python str, TypeScript string). The Python SDK rejects bytes — decode the body before passing it. Do not parse the body before calling verify(); the signature is over the exact bytes the sender hashed.

Hand-rolled verification (no SDK)

If you can't use the SDK — verifying in a language Vendo doesn't ship an SDK for, or inside a Cloudflare Worker — the scheme is straightforward:

import hmac
import hashlib
import os

def verify(headers, body: str) -> bool:
    secret = os.environ["VENDO_WEBHOOK_SECRET"].encode()
    sig = headers.get("Vendo-Signature", "").strip()
    ts = headers.get("Vendo-Timestamp", "").strip()
    if not sig or not ts:
        return False
    expected = hmac.new(secret, f"{ts}.{body}".encode(), hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, sig)

Always include the timestamp inside the HMAC input — signing just the body lets an attacker replay any captured request indefinitely.

Consuming lifecycle events over SSE

Connection-lifecycle and billing events flow over the SDK's SSE stream at /api/deployments/me/events, not as outbound HTTP webhooks. The SDK exposes them as an iterator:

from vendo import Vendo

client = Vendo()

for msg in client.events.subscribe():
    # msg.type is the event kind ("connection.connected", "billing.balance_changed", ...)
    # msg.data is the parsed payload dict (matches the wire shape in web/src/lib/sse/types.ts).
    if msg.type == "connection.connected":
        slug = msg.data.get("slug") if isinstance(msg.data, dict) else None
        if isinstance(slug, str):
            client.invalidate(slug)

// docs-examples-ci: skip
import { Vendo } from "@vendodev/sdk";

const vendo = new Vendo();

for await (const msg of vendo.events.subscribe()) {
  if (msg.type === "connection.connected") {
    const slug = (msg.data as { slug?: string } | null)?.slug;
    if (typeof slug === "string") vendo.invalidate(slug);
  }
}

The iterator reconnects automatically on transient errors with exponential backoff. Both flavors require Vendo mode (VENDO_API_KEY set) and raise VendoOnlyFeature in OSS.

Event kinds on the SSE stream

These are the kinds the SDK can currently surface. The wire field is kind; the SDK normalizes it to msg.type on EventStreamMessage.

Source of truth: web/src/lib/sse/types.ts in the monorepo.

Notes

• VENDO_WEBHOOK_SECRET is per-deployment. In Vendo mode it's injected at boot. In OSS mode you set it yourself when running a tool that verifies forwarded provider webhooks.
• The 5-minute replay window is configurable on the API constructor: WebhooksAPI(max_age_sec=600) (Python) / new WebhooksAPI({ maxAgeSec: 600 }) (TypeScript).
• Telegram webhook delivery has its own per-provider verification (X-Telegram-Bot-Api-Secret-Token) that the hooks worker checks before forwarding. The SDK helper covers the Vendo-signed envelope added on top. See the Telegram tutorial for the dual-verification pattern.