AI Video Pipeline with Async Tasks and Webhooks
Build a reliable AI video pipeline: submit async tasks, verify HMAC-signed webhooks, save files before links expire, catch missed events with a poller.
Piotr Olszewski2026-09-276 min
An AI video pipeline works best as an async job system: you submit a generation task, store its id, and let the API call your webhook when the video is ready, instead of holding a connection open for minutes. On Jantaris that means POST /v1/generate with a webhook_url, a receiver that verifies the x-jantaris-signature header (hex HMAC-SHA256 of the raw body), and a small poller as a safety net. The rest is bookkeeping: a task table, idempotent handlers and copying files to your own storage.
This is the architecture I use for batch video jobs. Code is in Python (FastAPI) with a Node.js version of the webhook receiver.
Why AI video generation has to be async
A video generation can take anywhere from under a minute to several minutes, depending on model, duration and resolution. A synchronous HTTP call that long fails in predictable ways:
- load balancers and proxies cut idle connections,
- serverless functions hit their execution limits,
- a deploy or restart kills every request in flight,
- you cannot run 50 jobs in parallel without 50 open connections.
So the API splits the work in two. Submission returns instantly with a task id and status 202. Completion is reported later, either when you ask (GET /v1/tasks/{id}) or when the API tells you (webhook).
The pipeline at a glance
[your app] --POST /v1/generate (webhook_url)--> [API] -> 202 {id, status: queued, estimated_cost}
|
+-- save task row (id, status=queued)
[API] --POST webhook (signed JSON)--> [your receiver]
|- verify signature
|- dedupe by task id + status
|- enqueue: download output to your storage
'- mark task done
[reconciler, every few minutes] --GET /v1/tasks/{id}--> for tasks stuck in queued/runningFour parts: submitter, receiver, downloader, reconciler. Each one is small.
Step 1: submit tasks and record them
import os
import sqlite3
import requests
API = "https://jantaris.com/v1"
HEADERS = {"Authorization": f"Bearer {os.environ['AI_API_KEY']}"}
WEBHOOK_URL = "https://yourapp.example.com/hooks/video"
db = sqlite3.connect("tasks.db", check_same_thread=False)
db.execute("""
CREATE TABLE IF NOT EXISTS tasks (
id TEXT PRIMARY KEY,
model TEXT,
prompt TEXT,
status TEXT,
estimated_cost TEXT,
cost TEXT,
file_path TEXT,
error TEXT,
updated_at TEXT DEFAULT CURRENT_TIMESTAMP
)""")
def submit_video(prompt, model="seedance-2.5", duration=5, resolution="720p", aspect_ratio="9:16"):
body = {
"model": model,
"input": {
"prompt": prompt,
"duration": duration,
"resolution": resolution,
"aspect_ratio": aspect_ratio,
},
"webhook_url": WEBHOOK_URL,
}
r = requests.post(f"{API}/generate", headers=HEADERS, json=body, timeout=30)
data = r.json()
if r.status_code != 202:
raise RuntimeError(f"{r.status_code} {data['error']['type']}: {data['error']['message']}")
db.execute(
"INSERT INTO tasks (id, model, prompt, status, estimated_cost) VALUES (?, ?, ?, ?, ?)",
(data["id"], model, prompt, data["status"], data["estimated_cost"]["amount"]),
)
db.commit()
return data["id"]Two details worth copying:
- Record the task before anything else happens. If the webhook arrives before your insert commits (it can, with fast models), your receiver should still cope. Inserting right after the 202 keeps that window tiny, and the handler below treats unknown ids gracefully.
- Store
estimated_cost. It is calculated before the job runs, so you can enforce a budget per user or per batch before costs pile up.
Handling submission errors
| Code | Meaning | Pipeline action |
|---|---|---|
| 402 | Balance below estimated_cost | Pause the queue, alert, resume after top-up |
| 422 | Invalid input | Mark the job invalid, do not retry |
| 429 | Rate limited | Retry with exponential backoff |
| 502 | Provider error | Retry after a delay |
Step 2: receive and verify the webhook
When a task finishes, Jantaris sends a POST to your webhook_url with the same JSON you would get from GET /v1/tasks/{id}:
{
"id": "task_...",
"status": "succeeded",
"model": "seedance-2.5",
"output": [{ "type": "video", "url": "https://...", "duration": 5, "width": 720, "height": 1280 }],
"cost": { "amount": "X.XX", "currency": "USD" },
"error": null,
"created_at": "...",
"completed_at": "..."
}The x-jantaris-signature header holds a hex-encoded HMAC-SHA256 of the raw request body, keyed with the webhook secret from your dashboard. Verify it against the exact bytes you received. Parsing the JSON and re-serializing it changes whitespace and key order, and the signature will not match.
FastAPI receiver
import hashlib
import hmac
import json
from fastapi import FastAPI, Request, HTTPException, BackgroundTasks
app = FastAPI()
SECRET = os.environ["WEBHOOK_SECRET"].encode()
def valid_signature(raw: bytes, header: str | None) -> bool:
if not header:
return False
expected = hmac.new(SECRET, raw, hashlib.sha256).hexdigest()
return hmac.compare_digest(expected, header)
@app.post("/hooks/video")
async def video_hook(request: Request, background: BackgroundTasks):
raw = await request.body()
if not valid_signature(raw, request.headers.get("x-jantaris-signature")):
raise HTTPException(status_code=401, detail="bad signature")
task = json.loads(raw)
row = db.execute("SELECT status FROM tasks WHERE id = ?", (task["id"],)).fetchone()
if row and row[0] in ("failed", "stored"):
return {"ok": True, "duplicate": True} # already handled
if task["status"] == "succeeded":
background.add_task(store_output, task)
elif task["status"] == "failed":
db.execute(
"UPDATE tasks SET status = 'failed', cost = '0', error = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?",
(json.dumps(task["error"]), task["id"]),
)
db.commit()
return {"ok": True}Rules I follow for every webhook receiver:
- Verify first, parse second. Reject anything with a missing or wrong signature.
- Use
hmac.compare_digest, not==, to avoid timing leaks. - Answer fast. Return 2xx right away and do slow work (downloads, transcoding) in the background.
- Be idempotent. Design for the same event arriving twice. Checking the stored status before acting makes duplicates harmless.
Node.js / Express receiver
The classic Express mistake is letting express.json() parse the body before you verify it. Use express.raw() on the webhook route:
import crypto from "node:crypto";
import express from "express";
const app = express();
const SECRET = process.env.WEBHOOK_SECRET;
app.post("/hooks/video", express.raw({ type: "application/json" }), (req, res) => {
const signature = req.get("x-jantaris-signature") || "";
const expected = crypto.createHmac("sha256", SECRET).update(req.body).digest("hex");
const a = Buffer.from(expected, "utf8");
const b = Buffer.from(signature, "utf8");
if (a.length !== b.length || !crypto.timingSafeEqual(a, b)) {
return res.status(401).send("bad signature");
}
const task = JSON.parse(req.body.toString("utf8"));
res.status(200).json({ ok: true });
if (task.status === "succeeded") {
queueDownload(task); // your job queue
} else if (task.status === "failed") {
markFailed(task.id, task.error);
}
});
app.listen(3000);Step 3: copy outputs to your own storage
Output URLs are valid for a limited time. Treat them as a pickup point. The moment a task succeeds, download the file and put it somewhere you control (S3, R2, GCS, a disk).
import pathlib
STORAGE = pathlib.Path("videos")
STORAGE.mkdir(exist_ok=True)
def store_output(task):
item = task["output"][0]
path = STORAGE / f"{task['id']}.mp4"
with requests.get(item["url"], stream=True, timeout=300) as r:
r.raise_for_status()
with open(path, "wb") as f:
for chunk in r.iter_content(1 << 20):
f.write(chunk)
db.execute(
"UPDATE tasks SET status = 'stored', cost = ?, file_path = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?",
(task["cost"]["amount"], str(path), task["id"]),
)
db.commit()If a download fails, the reconciler below will fetch the task again and retry while the link is still valid.
Step 4: a reconciler for missed webhooks
Webhooks are the fast path, not the only path. Your server might be down during a deploy, a firewall might block a request, a background download might crash. A reconciler closes the gap by polling tasks that have been pending too long:
import time
def reconcile(max_age_minutes=10):
rows = db.execute(
"""SELECT id FROM tasks
WHERE status IN ('queued', 'running')
AND updated_at < datetime('now', ?)""",
(f"-{max_age_minutes} minutes",),
).fetchall()
for (task_id,) in rows:
t = requests.get(f"{API}/tasks/{task_id}", headers=HEADERS, timeout=30).json()
if t["status"] == "succeeded":
store_output(t)
elif t["status"] == "failed":
db.execute("UPDATE tasks SET status = 'failed', cost = '0', error = ? WHERE id = ?",
(json.dumps(t["error"]), task_id))
db.commit()
else:
db.execute("UPDATE tasks SET status = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?",
(t["status"], task_id))
db.commit()
if __name__ == "__main__":
while True:
reconcile()
time.sleep(300)Run it as a cron job or a background worker. In a healthy system it finds nothing. When something breaks, it quietly repairs the state.
Failed tasks and costs
When a generation fails (content filter, provider error, timeout), the task ends with status: "failed", an error object explaining why, and a cost of 0. The amount held at submission is released in full. For budgeting this means your real spend is the sum of cost on successful tasks, which the task table above already tracks:
SELECT model, COUNT(*) AS videos, SUM(CAST(cost AS REAL)) AS spent
FROM tasks WHERE status = 'stored'
GROUP BY model;For per-second rates by model, see How much does AI video cost per second?.
Testing webhooks locally
Your laptop is not reachable from the internet, so use a tunnel (ngrok, Cloudflare Tunnel or similar) to expose localhost:8000 and pass the public URL as webhook_url. Then:
- Submit a short, cheap job, for example Seedance 2.5 at 480p for a few seconds.
- Watch the receiver log for the signed POST.
- Tamper with one byte of a saved payload and replay it with curl to confirm your receiver returns 401.
curl -X POST http://localhost:8000/hooks/video \
-H "Content-Type: application/json" \
-H "x-jantaris-signature: 0000" \
--data-binary @payload.jsonScaling the pipeline
Once the four parts work, scaling is mostly configuration:
- Concurrency. Submit from a queue with a worker limit and back off on 429.
- Mixed models. The same pipeline runs Seedance 2.5, Kling 3.0 and Veo 3.1 (Quality), plus image models and Mureka V9, because every model uses the same task object. Compare video models in Best AI video generation APIs in 2026.
- Budgets. Sum
estimated_costof queued tasks and refuse new submissions above a daily cap. - Balance alerts. Poll
GET /v1/balancehourly and alert below a threshold, so the queue never stalls on 402.
The reference docs are /docs/generate, /docs/tasks and /docs/webhooks.
FAQ
How do I verify a Jantaris webhook signature?
Compute HMAC-SHA256 of the raw request body using your webhook secret, hex-encode it, and compare it with the x-jantaris-signature header using a constant-time comparison such as hmac.compare_digest. Do it before parsing the JSON.
Should I use polling or webhooks for AI video?
Webhooks for production, because you get notified as soon as a task ends without constant requests. Keep a slow poller as a backup for missed deliveries. For quick scripts, polling GET /v1/tasks/{id} every few seconds is fine.
What does the webhook payload contain?
The same JSON as GET /v1/tasks/{id}: task id, status, model, output list with file URLs and dimensions, cost, error and timestamps.
How long are output URLs valid?
For a limited time only. Download each file to your own storage as soon as the task succeeds.
Am I charged if a video generation fails?
No. A task with status failed has a cost of 0, and the amount held on your balance at submission is released in full.
