# CheapAIAPI image Job webhooks

CheapAIAPI can POST a signed terminal callback when an image Job reaches `done`
or `failed`. Every image account can use webhooks. Read `webhook_secret` from
`GET /v1/balance` and keep it on your backend.

Recommended production flow for image Jobs:

1. Call `GET /v1/balance` and store `webhook_secret`.
2. Call `POST /v1/images/generations?async=true` with `webhook_url`.
3. Store the returned Job `id`.
4. Handle the signed callback as the primary completion path.
5. Recover with `GET /v1/jobs/{id}` if no callback arrives.

Chat Completions and video do not accept `webhook_url`. Polling remains the
recovery path for image Jobs and is still required so you can store the `id`.
Webhooks remove polling traffic and polling-interval delay; they do not reduce
generation time.

## Low-latency integration

These rules are the fast path. Each one removes avoidable delay that is not
image generation time; apply them by default.

1. **Create Jobs with `?async=true` plus `webhook_url` and treat the signed
   callback as the completion path.** The create call returns immediately and
   the callback is sent as soon as the Job reaches `done` or `failed`, so
   nothing waits on an idle HTTP connection or a polling interval.
2. **Pass reference images as inline `data:image/<mime>;base64,...` bytes
   instead of `https://` URLs.** A URL input may have to be fetched before
   generation can start; inline bytes never need that round trip.
3. **Send each reference image at the resolution you actually need and do not
   compress the request body.** One reference image may be at most 25 MiB and
   the whole `image` array at most 50 MiB, and gzipping already-compressed
   image bytes only costs CPU on both ends.
4. **Send a stable `Idempotency-Key` per client-side Job and reuse it when you
   retry.** For 24 hours the same key replays the original Job instead of
   starting a second paid generation, so a timed-out retry costs you nothing.
5. **Poll `GET /v1/jobs/{id}` only as recovery, every 5 seconds and never
   faster than every 2 seconds.** Polling is the slow path: it adds interval
   delay on top of generation time and never makes a Job finish sooner.
6. **Download `data[].url` as soon as the Job is `done`.** Result links are not
   durable storage: treat the `X-Result-Retention-Days` response header
   (currently 7) as an upper bound, expect at most 25 MiB per image, and copy
   anything you need to keep into your own storage.
7. **Reuse one HTTP/1.1 keep-alive or HTTP/2 connection pool for every call.**
   Opening a new TLS connection per request pays a full handshake before the
   request is even sent.
8. **Tell us before you sustain more than 8 image requests per second.** 8 RPS
   is the published operating baseline for standard image workloads, not a
   universal hard limit; higher sustained or bursty rates are provisioned on
   request through https://cheapaiapi.org/contact.

Fast-path request:

```bash
curl --request POST \
  'https://cheapaiapi.org/v1/images/generations?async=true' \
  --header 'Authorization: Bearer sk_cheap_...' \
  --header 'Content-Type: application/json' \
  --header 'Idempotency-Key: customer-job-123' \
  --data '{
    "model": "nano-banana-pro-2k",
    "prompt": "Place the product on a clean studio background.",
    "image": ["data:image/jpeg;base64,/9j/4AAQSkZJRgABAQ..."],
    "response_format": "url",
    "webhook_url": "https://api.example.com/webhooks/cheapaiapi"
  }'
```

## Create an image Job with a callback

Pass `webhook_url` and optional `custom_data` on the normal asynchronous
generation request:

```json
{
  "model": "nano-banana-pro-2k",
  "prompt": "Place the product on a clean studio background.",
  "response_format": "url",
  "webhook_url": "https://api.example.com/webhooks/cheapaiapi",
  "custom_data": {"customer_request_id": "customer-job-123"}
}
```

Submit this body to `POST /v1/images/generations?async=true`. Webhook URLs must
be publicly reachable HTTPS endpoints. Callback Jobs require
`response_format: "url"` so callback bodies contain durable result URLs rather
than large inline image bytes. `custom_data` is an optional JSON object of at
most 4096 serialized bytes; it is stored and echoed unchanged. Do not put
secrets in `custom_data`.

## Callback body

The callback is the frozen terminal projection of the public Job response, with
an additional stable `event_id`. A successful callback looks like this:

```json
{
  "event_id": "0bd05147-b6a0-45fa-b176-9cfe008bb350",
  "id": "3e7487ca-a942-4ef0-bc47-4cb8171a2ca8",
  "status": "done",
  "model": "nano-banana-pro-2k",
  "created_at": "2026-07-14T12:20:00+00:00",
  "finished_at": "2026-07-14T12:20:14+00:00",
  "custom_data": {"customer_request_id": "customer-job-123"},
  "data": [{"url": "https://cheapaiapi.org/v1/jobs/3e7487ca-a942-4ef0-bc47-4cb8171a2ca8/results/0?sig=..."}],
  "error": null
}
```

For a failed Job, `status` is `failed`, `data` is empty, and `error` contains
the same sanitized public error shape returned by polling. The submitted
`webhook_url` and prompt are not echoed.

## Signing secret and verification

Every callback has one signing header:

```text
X-CheapAIAPI-Signature: t=<unix_timestamp>,v1=<hex_hmac_sha256>
```

The signed bytes are `<unix_timestamp>.<exact raw request body>`. Read the raw
body before JSON parsing, reject timestamps more than five minutes from your
clock, and compare digests in constant time.

```python
import hashlib
import hmac
import json
import time

from fastapi import FastAPI, HTTPException, Request, Response

app = FastAPI()
WEBHOOK_SECRET = "<from GET /v1/balance webhook_secret>"


def verify_signature(header: str, raw_body: bytes) -> bool:
    try:
        parts = dict(part.split("=", 1) for part in header.split(","))
        timestamp = int(parts["t"])
        supplied = parts["v1"]
    except (KeyError, TypeError, ValueError):
        return False
    if abs(int(time.time()) - timestamp) > 300:
        return False
    signed = f"{timestamp}.".encode() + raw_body
    expected = hmac.new(
        WEBHOOK_SECRET.encode(), signed, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(expected, supplied)


@app.post("/webhooks/cheapaiapi")
async def receive_webhook(request: Request) -> Response:
    raw_body = await request.body()
    signature = request.headers.get("X-CheapAIAPI-Signature", "")
    if not verify_signature(signature, raw_body):
        raise HTTPException(status_code=400, detail="invalid signature")
    event = json.loads(raw_body)
    # Insert event["event_id"] into an inbox table with a UNIQUE constraint,
    # and apply your business effect in the same database transaction.
    return Response(status_code=204)
```

Return any `2xx` quickly to acknowledge delivery. Transient delivery failures
are retried automatically, so the same signed `event_id` may arrive more than
once. Deduplicate by `event_id`; on a duplicate, skip the business effect and
return `2xx`. If no callback arrives, recover with `GET /v1/jobs/{id}`.