Webhooks

Webhooks let you receive results pushed directly to a URL you control when an asynchronous request completes — instead of polling the Status Service. This is the recommended approach for production pipelines, batch jobs, and event-driven architectures.

When to Use Webhooks

Polling works well for short, interactive tasks. Webhooks are better when:

Jobs can take more than a few seconds (e.g., Tailored Generation fine-tuning, video processing, batch image editing).
You're building event-driven or serverless backends (AWS Lambda, queues, etc.).
You want to avoid maintaining open connections or polling loops at scale.

Setting Up a Webhook

Via REST API

Pass webhook_url in the request body of any async endpoint:

curl -X POST https://engine.prod.bria-api.com/v2/image/generate \
  -H "api_token: YOUR_API_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "a serene mountain landscape at dawn",
    "webhook_url": "https://your-app.com/api/bria/webhook"
  }'

The API immediately returns a request_id:

{
  "request_id": "a1b2c3d4-...",
  "status_url": "https://engine.prod.bria-api.com/v2/status/a1b2c3d4-..."
}

When processing completes, Bria POSTs the result to your webhook_url.

Via Python SDK

webhook_url is a separate parameter on .submit() — pass it as a keyword argument, not inside your payload dict. The SDK merges it (along with sync: false) into the request body sent to the API; your original payload dict is not modified. Works on both BriaSyncClient and BriaAsyncClient.

from bria_client import BriaAsyncClient

async with BriaAsyncClient() as client:
    response = await client.submit(
        endpoint="image/generate",
        payload={"prompt": "a serene mountain landscape at dawn"},
        webhook_url="https://your-app.com/api/bria/webhook",
    )

print(f"Submitted: {response.request_id}")
# Your webhook URL will receive the result — no polling needed

Webhook Delivery

When the job reaches a terminal state, Bria sends a POST to your URL with three signed headers:

Header	Description
`Bria-Webhook-Id`	The job's `request_id`.
`Bria-Webhook-Timestamp`	Unix epoch (seconds) when the signature was generated.
`Bria-Webhook-Signature`	`v1=<base64>`.

The delivered body is identical to the response you'd get by polling the Status Service — same schema, same fields. If you already handle status-poll responses, the same parsing code works for webhooks.

Successful result

{
  "request_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "status": "COMPLETED",
  "result": {
    "image_url": "https://...",
    "seed": 42,
    "prompt": "a serene mountain landscape at dawn"
  }
}

Error result

{
  "request_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "status": "ERROR",
  "error": {
    "message": "Invalid parameter: prompt is required",
    "code": "VALIDATION_ERROR"
  }
}

Handling Webhook Deliveries

Your endpoint should:

Respond 2xx within 10 seconds — Bria expects an acknowledgment inside this window. Defer heavy work to a background job or queue.
Verify the signature first — reject any request that doesn't validate (see below).

Retries & Idempotency

If your endpoint is unreachable or returns a non-2xx response, Bria retries delivery with exponential backoff — up to 5 attempts over 45 minutes. A 2xx response stops retries. Because retries (and rare duplicate deliveries on success) are possible, your handler must be idempotent: use the Bria-Webhook-Id header (the job's request_id) as a deduplication key.

Security: Verifying Webhook Signatures

Bria signs every webhook with HMAC-SHA256. The convention is Svix-inspired but Bria-specific — it borrows the signed-message format, three-header layout, and versioned signature prefix from Svix, but the signing-key derivation is Bria's own.

Signed message format follows the Svix convention: {webhook_id}.{timestamp}.{raw_body}.
Signing key is derived from your API token via HMAC-SHA256 (Bria-specific): HMAC-SHA256(api_token, "bria-webhook-signing-v1"). Your raw API token is never used directly to sign payloads.
Signature header carries a single v1=<base64> token.

Python (SDK helper)

The SDK ships a verifier — no extra dependencies:

from bria_client.toolkit import verify_webhook_signature

ok = verify_webhook_signature(
    payload=raw_body_bytes,
    webhook_id=request.headers["Bria-Webhook-Id"],
    timestamp=request.headers["Bria-Webhook-Timestamp"],
    signature_header=request.headers["Bria-Webhook-Signature"],
    api_token=os.environ["BRIA_API_TOKEN"],
)
if not ok:
    raise HTTPException(401, "Invalid webhook signature")

Reference implementation (any language)

If you're not using the Python SDK, reproduce the same scheme:

import base64, hashlib, hmac

def verify(payload: bytes, webhook_id: str, timestamp: str,
           signature_header: str, api_token: str) -> bool:
    if not signature_header.startswith("v1="):
        return False
    signing_key = hmac.new(api_token.encode(), b"bria-webhook-signing-v1", hashlib.sha256).digest()
    message = f"{webhook_id}.{timestamp}.{payload.decode()}".encode()
    expected = base64.b64encode(hmac.new(signing_key, message, hashlib.sha256).digest()).decode()
    return hmac.compare_digest(signature_header[3:], expected)

Always use a constant-time comparison (hmac.compare_digest or equivalent) when checking the signature.

Full Example: FastAPI Webhook Receiver

This is the pattern shipped in examples/webhook_handler.py — verify, acknowledge in under 10s, process in the background.

import json, os
from fastapi import BackgroundTasks, FastAPI, Header, HTTPException, Request
from bria_client.toolkit import verify_webhook_signature

app = FastAPI()

@app.post("/webhook")
async def receive_webhook(
    request: Request,
    background_tasks: BackgroundTasks,
    bria_webhook_id: str = Header(...),
    bria_webhook_timestamp: str = Header(...),
    bria_webhook_signature: str = Header(...),
):
    body = await request.body()

    if not verify_webhook_signature(
        payload=body,
        webhook_id=bria_webhook_id,
        timestamp=bria_webhook_timestamp,
        signature_header=bria_webhook_signature,
        api_token=os.environ["BRIA_API_TOKEN"],
    ):
        raise HTTPException(status_code=401, detail="Invalid webhook signature")

    data = json.loads(body)
    background_tasks.add_task(process_result, data)
    return {"ok": True}

async def process_result(data: dict):
    if data["status"] == "COMPLETED":
        ...  # store result, notify user, trigger next step
    elif data["status"] == "ERROR":
        ...  # log / alert

Express.js receiver

import express from 'express';
import crypto from 'crypto';

const app = express();

function verify(rawBody, id, ts, sigHeader, apiToken) {
  if (!sigHeader.startsWith('v1=')) return false;
  const signingKey = crypto.createHmac('sha256', apiToken).update('bria-webhook-signing-v1').digest();
  const expected = crypto.createHmac('sha256', signingKey).update(`${id}.${ts}.${rawBody}`).digest('base64');
  const got = Buffer.from(sigHeader.slice(3));
  const exp = Buffer.from(expected);
  return got.length === exp.length && crypto.timingSafeEqual(got, exp);
}

app.post('/api/bria/webhook', express.raw({ type: 'application/json' }), (req, res) => {
  const ok = verify(
    req.body.toString(),
    req.header('Bria-Webhook-Id'),
    req.header('Bria-Webhook-Timestamp'),
    req.header('Bria-Webhook-Signature'),
    process.env.BRIA_API_TOKEN,
  );
  if (!ok) return res.sendStatus(401);

  res.sendStatus(200); // acknowledge first

  const { request_id, status, result, error } = JSON.parse(req.body);
  if (status === 'COMPLETED') console.log(`Job ${request_id} done:`, result.image_url);
  else if (status === 'ERROR') console.error(`Job ${request_id} failed:`, error.message);
});

app.listen(3000);