Webhooks

Webhooks let Verbitas push job completion and billing events to your application, so you do not need to poll for status.

Registering a webhook endpoint

In the admin console at https://verbitas.io/admin/webhooks, or via API:

curl -X POST https://api.verbitas.io/v1/webhooks \
  -H "Authorization: Bearer $VERBITAS_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://your-app.example.com/webhooks/verbitas",
    "events": ["job.completed", "job.failed", "anchor.confirmed"],
    "description": "My production webhook"
  }'

Response:

{
  "webhook_id": "wh_01j...",
  "url": "https://your-app.example.com/webhooks/verbitas",
  "events": ["job.completed", "job.failed", "anchor.confirmed"],
  "secret": "whsec_01j...",
  "created_at": "2026-05-09T10:00:00Z"
}

Save the secret. It is shown once and used to verify all incoming webhook payloads.

Webhook events

Event	Triggered by
`job.completed`	Sign job finished successfully
`job.failed`	Sign job failed after all retries
`anchor.confirmed`	Anchor batch confirmed on-chain (OTS or Arbitrum)
`billing.quota_warning`	Usage at 80% of plan limit
`billing.quota_exceeded`	Usage hit plan limit
`key.expiring`	API key will expire in 7 days

Webhook payload

{
  "event_id": "evt_01j...",
  "type": "job.completed",
  "created_at": "2026-05-09T10:05:00Z",
  "data": {
    "job_id": "j_01j...",
    "asset_id": "a_01j...",
    "watermark_id": "w_01j...",
    "manifest_uri": "https://m.verbitas.io/manifests/a_01j.../manifest.c2pa",
    "verifier_url": "https://v.verbitas.io/v/a_01j...",
    "recipe_id": "image-genai-v1",
    "anchor": { "batch_id": "b_01j...", "status": "queued" }
  }
}

Signature verification

Every webhook request includes a Verbitas-Signature header. You must verify this before processing the event.

The signature is HMAC-SHA-256 over the raw request body, using the whsec_ secret as the key.

Python (FastAPI)

from fastapi import Request, HTTPException
import hmac
import hashlib

WEBHOOK_SECRET = "whsec_01j..."  # from env

async def webhook_handler(request: Request):
    payload = await request.body()
    signature = request.headers.get("Verbitas-Signature", "")

    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()

    if not hmac.compare_digest(f"sha256={expected}", signature):
        raise HTTPException(status_code=400, detail="Invalid signature")

    event = json.loads(payload)

    if event["type"] == "job.completed":
        asset_id = event["data"]["asset_id"]
        # update your records

Python SDK shortcut

from fastapi import Request
import verbitas

client = verbitas.Client()

async def webhook_handler(request: Request):
    payload = await request.body()
    signature = request.headers.get("Verbitas-Signature")

    event = client.webhooks.construct_event(
        payload=payload,
        signature=signature,
        secret="whsec_01j..."  # or from env
    )
    # raises verbitas.exceptions.WebhookSignatureError if invalid

    if event.type == "job.completed":
        asset_id = event.data["asset_id"]

TypeScript (Express)

import express from "express";
import { VerbitasClient } from "@verbitas/sdk";

const app = express();
const client = new VerbitasClient();

app.post("/webhooks/verbitas", express.raw({ type: "application/json" }), (req, res) => {
  const signature = req.headers["verbitas-signature"] as string;

  let event;
  try {
    event = client.webhooks.constructEvent(
      req.body,
      signature,
      process.env.VERBITAS_WEBHOOK_SECRET!
    );
  } catch (e) {
    console.error("Invalid webhook signature");
    return res.status(400).send("Invalid signature");
  }

  if (event.type === "job.completed") {
    const assetId = event.data.assetId;
    // update your records
  }

  res.sendStatus(200);
});

Delivery and retries

Verbitas requires a 2xx response within 30 seconds. If the response is non-2xx or times out, the event is retried with exponential backoff:

Attempt	Delay
1	Immediate
2	30 seconds
3	5 minutes
4	30 minutes
5	2 hours

After 5 failed attempts, the event is marked as failed and recorded in the audit log as webhook.failed.

Idempotency

Use event_id to deduplicate retried deliveries. Store processed event IDs in Redis or your database:

redis_client.set(f"webhook:{event['event_id']}", "processed", ex=86400)
if redis_client.exists(f"webhook:{event['event_id']}"):
    return  # already processed

Viewing webhook delivery history

In the admin console, or via API:

curl "https://api.verbitas.io/v1/webhooks/wh_01j.../deliveries?limit=20" \
  -H "Authorization: Bearer $VERBITAS_API_KEY" | jq .

Testing webhooks locally

Use a tunnel such as ngrok or cloudflared to expose your local server:

ngrok http 8000
# → https://abc123.ngrok.io

Or trigger a test delivery from the admin console: https://verbitas.io/admin/webhooks/wh_01j.../test