Async requests, webhooks and polling

All Bria v2 endpoints process requests asynchronously by default. This keeps long generations from tying up connections and lets you fan out hundreds of jobs at once. Every request follows the same lifecycle:

1. Submit. POST the request. The API validates it and responds with 202 Accepted, a request_id and a status_url.
2. Receive the result. Either pass a webhook_url and Bria POSTs the result to you when the job finishes (recommended for production), or poll the status_url until it does.
3. Collect. When status is COMPLETED, the result object holds the output: image_url for images, video_url for videos, structured_prompt for structured prompt generation, plus endpoint-specific fields such as seed.

Webhook deliveries and status polls carry the same body, so one parser handles both:

{
  "request_id": "9d4a1a2f5c7e4b6f8a0c2d3e4f5a6b7c",
  "status": "COMPLETED",
  "result": {
    "image_url": "https://.../result.png",
    "seed": 314159
  }
}

Pass webhook_url with any asynchronous request. Use webhooks for pipelines, queues and serverless backends: no polling loops, no open connections, and results arrive the moment a job finishes.

from bria_client import BriaSyncClient

client = BriaSyncClient()  # reads BRIA_API_TOKEN

job = client.submit(
    endpoint="image/generate",
    payload={"prompt": "a serene mountain landscape at dawn"},
    webhook_url="https://your-app.com/api/bria/webhook",
)
print("submitted", job.request_id)  # the result will be POSTed to your webhook

When the job reaches a terminal state, Bria sends a POST with the body shown above and three signed headers:

Your endpoint must answer with any 2xx within 10 seconds; do heavy work afterwards, in a queue or background task. If the endpoint is unreachable or returns a non-2xx response, Bria retries with exponential backoff, up to 5 attempts over 45 minutes. Because retries and rare duplicates are possible, treat deliveries as idempotent and key them on Bria-Webhook-Id.

The signing key is derived from your API token (HMAC-SHA256(api_token, "bria-webhook-signing-v1")), and the signed message is {webhook_id}.{timestamp}.{raw_body}. Verify with a constant-time comparison before trusting a delivery.

import os
from bria_client.toolkit import verify_webhook_signature

def is_valid(raw_body: bytes, headers) -> bool:
    return verify_webhook_signature(
        payload=raw_body,
        webhook_id=headers["Bria-Webhook-Id"],
        timestamp=headers["Bria-Webhook-Timestamp"],
        signature_header=headers["Bria-Webhook-Signature"],
        api_token=os.environ["BRIA_API_TOKEN"],
    )

Verify, acknowledge, then process in the background.

import json
import os

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

app = FastAPI()


@app.post("/api/bria/webhook")
async def receive(
    request: Request,
    background: 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")
    background.add_task(process, json.loads(body))
    return {"ok": True}


async def process(data: dict):
    if data["status"] == "COMPLETED":
        ...  # store data["result"]["image_url"], notify, trigger the next step
    elif data["status"] == "ERROR":
        ...  # log data["error"]

If you would rather pull than push, call GET /v2/status/{request_id} (the status_url from the submit response) until the job reaches a terminal state. The status endpoint reference has the full response schema.

job = client.submit(endpoint="image/generate", payload={"prompt": "a serene mountain landscape at dawn"})
result = client.poll(job, interval=2, timeout=120)  # raises TimeoutError after 120 s
print(result.result.image_url)

# client.run(...) does submit and poll in one call

The status endpoint returns HTTP 200 for all of these; the job state is in the body. A 404 means the request_id does not exist or has expired.

Polling guidance:

• Poll every 1 to 2 seconds for image endpoints and every 5 seconds for video, with a timeout that fits the workload (a minute for images, ten minutes or more for long videos).
• Status calls count toward your rate limit, so do not poll in a tight loop.
• Submissions are not idempotent: retrying a POST creates a new job. Retry only when the submission itself failed (network error or 5xx), never while a job is IN_PROGRESS.
• Download results as soon as they are ready and store them yourself. Result URLs are temporary.

Most endpoints accept "sync": true, which holds the connection open and returns the result in the response body with HTTP 200. It is convenient for quick experiments and low-volume interactive use. Prefer webhooks or polling for production: sync calls occupy a connection for the whole generation, and some options are async-only, for example resolution: "4MP" on /v2/image/generate or placement_type: "automatic" on lifestyle shots. Combining an async-only option with sync: true returns 422.

Some product-shot endpoints are synchronous by default and document their own sync behavior; the endpoint pages in the Product Shot Editing reference state it.