Python Tutorial: Build an Async Media Pipeline with Signed Webhooks

Media processing should not keep a Python web request open while a video is transcoded, an HLS package is assembled, audio is extracted, or transcripts are generated. A production application submits the work, stores the returned job ID, and handles the terminal result asynchronously. This tutorial builds that complete workflow using Python, Flask, Requests, psycopg, and PostgreSQL. It includes the part that is easiest to get subtly wrong: verifying the MediaRuntime webhook against the original request bytes before parsing its JSON.

What the Python application will do

The backend requests a short-lived upload target, uploads the source with the returned headers, creates an asynchronous job using the returned file_uri, and stores the job_id. MediaRuntime later calls a public HTTPS Flask route when the job becomes COMPLETED, FAILED, or REJECTED. The outgoing API request uses your server-side API key. The incoming webhook uses a different, view-once webhook secret. Neither credential belongs in browser code.

Python backend
    │
    ├── POST /v1/upload-url ─────────► MediaRuntime
    │◄─ upload_url, upload_headers, file_uri
    ├── PUT file bytes ──────────────► upload target
    ├── POST /v1/transcode ─────────► MediaRuntime
    │◄─ job_id, queued
    ├── persist job_id
    │
    ◄── signed terminal webhook ───── MediaRuntime
    ├── verify raw bytes and timestamp
    ├── insert event_id once
    └── update the application job

Before you begin

Create an API key and configure your public HTTPS webhook destination in the MediaRuntime account page. Save the webhook secret immediately when it is shown. Store the API key and webhook secret as two separate server-side secrets. Set MEDIARUNTIME_API_KEY, MEDIARUNTIME_WEBHOOK_SECRET, and DATABASE_URL in your runtime environment. MEDIARUNTIME_API_URL can be omitted when using https://mediaruntime.com.

Install the Python dependencies

python -m venv .venv
source .venv/bin/activate
pip install Flask requests "psycopg[binary]"

Step 1: upload a source and create the job

Request the upload target from your backend with X-API-Key. The response contains upload_url, upload_headers, and file_uri. Send every returned upload header with the PUT, but do not send your MediaRuntime API key to the upload URL. After the upload succeeds, submit the exact file_uri. It is an opaque reference owned by the service, so your application should never guess or construct its bucket path.

import os
from pathlib import Path

import requests

BASE_URL = os.getenv("MEDIARUNTIME_API_URL", "https://mediaruntime.com").rstrip("/")
API_KEY = os.environ["MEDIARUNTIME_API_KEY"]


def api_post(path: str, payload: dict) -> dict:
    response = requests.post(
        f"{BASE_URL}{path}",
        headers={
            "X-API-Key": API_KEY,
            "Content-Type": "application/json",
        },
        json=payload,
        timeout=30,
    )
    response.raise_for_status()
    return response.json()


def submit_media_job(file_path: str) -> dict:
    source = Path(file_path)
    target = api_post("/v1/upload-url", {
        "filename": source.name,
        "content_type": "video/mp4",
    })

    with source.open("rb") as media:
        upload = requests.put(
            target["upload_url"],
            headers=target["upload_headers"],
            data=media,
            timeout=(30, 600),
        )
    upload.raise_for_status()

    return api_post("/v1/transcode", {
        "file_url": target["file_uri"],
        "metadata": {
            "producer": "python-tutorial-api",
            "entity_id": "asset_0426",
            "media_type": "video",
        },
        "outputs": [{
            "type": "mp4",
            "preset": "mp4_720p_h264_aac",
            "path_suffix": "playback",
            "poster_time_sec": 2,
        }],
    })


job = submit_media_job("launch-trailer.mp4")
print(job["job_id"], job["status"])  # job_... queued

Persist the job ID immediately

Write job_id beside your application asset or workflow record before returning a processing state to the client. Do not wait for the media output inside the request that created the job. Use metadata for your own stable identifiers. MediaRuntime returns it under meta.request_metadata in the terminal event, while job_id remains the authoritative external job reference.

Step 2: receive the webhook in Flask

MediaRuntime signs timestamp + '.' + event ID + '.' + the exact JSON request bytes with HMAC-SHA256. Read request.get_data before calling request.get_json. Parsing and re-encoding the body can change its bytes and invalidate a correct signature. The receiver also limits the timestamp age to five minutes and compares the digest with hmac.compare_digest.

X-Transcoder-Id: webhook_evt_job_1320c28b72104811b075a26a99496cf6
X-Transcoder-Timestamp: 1786435283
X-Transcoder-Signature: t=1786435283,v1=4a1f...
Content-Type: application/json

Make event IDs unique in PostgreSQL

Webhook delivery is retryable, so your receiver must be idempotent. Store event_id as a primary key. If the same verified event arrives again, acknowledge it without repeating the state change or downstream work.

CREATE TABLE media_jobs (
  id bigserial PRIMARY KEY,
  mediaruntime_job_id text NOT NULL UNIQUE,
  status text NOT NULL,
  bundle_url text,
  error jsonb,
  updated_at timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE mediaruntime_webhook_events (
  event_id text PRIMARY KEY,
  mediaruntime_job_id text NOT NULL,
  status text NOT NULL,
  payload jsonb NOT NULL,
  received_at timestamptz NOT NULL DEFAULT now()
);

Verify first, parse second, record once

import hashlib
import hmac
import json
import os
import time

import psycopg
from flask import Flask, Response, request

app = Flask(__name__)
app.config["MAX_CONTENT_LENGTH"] = 2 * 1024 * 1024
WEBHOOK_SECRET = os.environ["MEDIARUNTIME_WEBHOOK_SECRET"]
DATABASE_URL = os.environ["DATABASE_URL"]
TERMINAL_STATUSES = {"COMPLETED", "FAILED", "REJECTED"}


def parse_signature(header: str) -> dict[str, str]:
    parts = {}
    for item in (header or "").split(","):
        key, separator, value = item.strip().partition("=")
        if separator and key:
            parts[key] = value
    return parts


def valid_signature(raw_body: bytes) -> bool:
    event_id = request.headers.get("X-Transcoder-Id", "")
    timestamp_header = request.headers.get("X-Transcoder-Timestamp", "")
    parts = parse_signature(request.headers.get("X-Transcoder-Signature", ""))
    timestamp = parts.get("t", "")
    received = parts.get("v1", "")

    if not event_id or not timestamp.isdigit():
        return False
    if timestamp_header and timestamp_header != timestamp:
        return False
    if len(received) != 64 or any(c not in "0123456789abcdefABCDEF" for c in received):
        return False
    if abs(int(time.time()) - int(timestamp)) > 300:
        return False

    signed = timestamp.encode() + b"." + event_id.encode() + b"." + raw_body
    expected = hmac.new(
        WEBHOOK_SECRET.encode(), signed, hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(received.lower(), expected)


@app.post("/webhooks/mediaruntime")
def mediaruntime_webhook():
    raw_body = request.get_data(cache=False, as_text=False)
    if not valid_signature(raw_body):
        return Response(status=401)

    try:
        event = json.loads(raw_body)
    except (UnicodeDecodeError, json.JSONDecodeError):
        return Response(status=400)

    event_id = request.headers.get("X-Transcoder-Id", "")
    status = str(event.get("status", "")).upper()
    job_id = str(event.get("job_id", ""))
    if event.get("event_id") != event_id or not job_id:
        return Response(status=400)
    if status not in TERMINAL_STATUSES:
        return Response(status=400)

    try:
        with psycopg.connect(DATABASE_URL) as connection:
            with connection.cursor() as cursor:
                cursor.execute(
                    """
                    INSERT INTO mediaruntime_webhook_events
                      (event_id, mediaruntime_job_id, status, payload)
                    VALUES (%s, %s, %s, %s::jsonb)
                    ON CONFLICT (event_id) DO NOTHING
                    RETURNING event_id
                    """,
                    (event_id, job_id, status, json.dumps(event)),
                )
                inserted = cursor.fetchone()

                if inserted:
                    bundle_url = (
                        event.get("delivery", {})
                        .get("bundle", {})
                        .get("download", {})
                        .get("url")
                    )
                    error_json = (
                        json.dumps(event["error"])
                        if event.get("error") is not None
                        else None
                    )
                    cursor.execute(
                        """
                        UPDATE media_jobs
                           SET status = %s, bundle_url = %s,
                               error = %s::jsonb, updated_at = now()
                         WHERE mediaruntime_job_id = %s
                        """,
                        (status, bundle_url, error_json, job_id),
                    )
        return Response(status=204)
    except psycopg.Error:
        app.logger.exception("Unable to record MediaRuntime webhook")
        return Response(status=500)

Step 3: acknowledge only after durable storage

The example returns HTTP 204 only after PostgreSQL commits the verified event. If the insert fails, it returns 500 so the delivery is not falsely acknowledged. If the event already exists, ON CONFLICT performs no second state transition and the handler still returns 204. Keep this endpoint fast. Insert an outbox task in the same transaction when you need to copy large outputs, send notifications, update a search index, or call another service. A background worker should perform that slower work after the webhook response.

Read the completed output from the event

For a COMPLETED job, delivery.bundle.download.url contains the short-lived, job-scoped MediaRuntime bundle URL. Use it exactly as returned. If your product needs longer retention, schedule a worker to copy the required artifacts into storage you control before the delivery window ends.

bundle_url = (
    event.get("delivery", {})
    .get("bundle", {})
    .get("download", {})
    .get("url")
)

request_metadata = event.get("meta", {}).get("request_metadata", {})
asset_id = request_metadata.get("entity_id")

Handle COMPLETED, FAILED, and REJECTED

COMPLETED carries delivery and final usage information. FAILED means execution could not finish and may include an error that operators can inspect. REJECTED means the job was stopped before successful execution. All three are terminal; update your local record so the interface does not wait forever.

Python webhook mistakes to avoid

Do not call request.get_json before preserving the raw request body. Do not authenticate incoming events with the API key. Do not use == for digest comparison when hmac.compare_digest is available. Do not skip the timestamp window or the header-to-body event ID check. Do not log the webhook secret, signature, signed bundle token, or full sensitive payload. Log the event ID, job ID, terminal status, and your own trace identifier.

Test the unhappy paths

Run a real test-mode job and confirm that the first webhook updates one row and a repeated webhook does not duplicate work. Modify one body byte and expect 401. Use a timestamp older than five minutes and expect 401. Stop PostgreSQL temporarily and expect 500. Exercise FAILED and REJECTED as well as COMPLETED. Webhooks should drive normal completion, while a scheduled reconciliation task checks jobs that remain QUEUED or PROCESSING unexpectedly. That recovery path complements the webhook without replacing it with aggressive polling.

Python production checklist

1. Keep both credentials in server-side secret storage. 2. Send every returned upload header and use the exact file_uri. 3. Persist job_id before returning a processing response. 4. Read raw request bytes before parsing JSON. 5. Verify timestamp, event ID, and HMAC with compare_digest. 6. Enforce a unique event_id in durable storage. 7. Return 2xx only after commit and 5xx when durable recording fails. 8. Move slow work into an outbox or queue. 9. Handle every terminal status. 10. Reconcile unexpectedly old non-terminal jobs.

One lifecycle for every media recipe

The output recipe can produce an MP4, HLS stream, extracted audio, transcript, poster, GIF preview, image derivative, moderation report, or watermarked result. The Python application lifecycle remains the same: upload, submit, persist, verify, record, and react. Explore the complete endpoint and payload reference at https://mediaruntime.com/docs.