Webhook receiver that queues finished SKU videos for human approval

A Format run on the Sume API is unattended, so a human approval step has to sit after the run, not inside it. The simplest pattern is a webhook receiver that verifies each format.run.terminal delivery and writes the finished video into a review table with state pending. Nothing is published until a person flips that row. This is a pattern built on the documented webhook, not a Sume feature, and it works for a bulk queue because each item carries its own communication.webhook_url.

What the delivery gives you

The run webhook docs describe one signed POST per terminal run. Headers are x-sume-webhook-timestamp and x-sume-webhook-signature (sume-v1=<hex>), plus a secret fingerprint. The signature is HMAC-SHA256 over <timestamp>.<raw_body>, and timestamps more than five minutes off should be refused. request_id and run_id are equal and stable across retries, which makes them the dedupe key.

outcome is ok, degraded or error. Only ok means a finished video with output. A bulk queue has no webhook of its own and may report completed while items failed, per the bulk runs docs, so the per-item webhook is where you learn about each SKU.

A receiver core you can run

This stdlib-only core refuses an empty secret, checks the five-minute window, compares in constant time, and inserts with insert or ignore so a retried delivery does not queue twice. The __main__ block signs a sample body and shows an empty secret failing, then a duplicate being ignored.

import hashlib, hmac, json, sqlite3, time

db = sqlite3.connect(":memory:")
db.execute("create table review(run_id text primary key, url text, state text)")


def verify(secret, raw, ts, sig, now=None):
    if not secret or not ts or not sig:
        return False
    if abs((now or time.time()) - int(ts)) > 300:
        return False
    mac = hmac.new(secret.encode(), ts.encode() + b"." + raw, hashlib.sha256)
    return hmac.compare_digest("sume-v1=" + mac.hexdigest(), sig)


def handle(event):
    if event["outcome"] != "ok" or not event["payload"]:
        return "needs-attention"
    url = event["payload"]["primary_output_url"]
    cur = db.execute("insert or ignore into review values (?,?,?)",
                     (event["run_id"], url, "pending"))
    return "queued" if cur.rowcount else "duplicate"


if __name__ == "__main__":
    ev = {"outcome": "ok", "run_id": "r1", "payload": {"primary_output_url": "u"}}
    raw, ts = json.dumps(ev).encode(), str(int(time.time()))
    sig = "sume-v1=" + hmac.new(b"k", ts.encode() + b"." + raw, hashlib.sha256).hexdigest()
    print(verify("", raw, ts, sig), verify("k", raw, ts, sig))
    print(handle(ev), handle(ev))

Wire it to your approval step

Put verify in front of your HTTP handler and read the raw bytes before parsing JSON, as the Formats cookbook does. Return 2xx quickly, then let a person or a script move rows from pending to approved or rejected.

Two limits matter. The sketch above treats only ok with a payload as reviewable; a production version should also fetch the receipt when payload is null. And approval is your policy: the run is already paid for when the row lands, so pair this with a retry plan for failed SKUs.