Build a small typed Python client for a video API

Updated 2026-10-02

You do not need an SDK to call a video API from Python, and VideoRouter's documentation itself points to plain requests against the base URL, noting that the OpenAI client has no first-class video methods. What you do want is a thin class that owns the awkward parts once: authentication, timeouts, retry policy, polling with a deadline, and downloading the result. This page gives a complete ~100-line version and explains the decisions in it so you can change them.

Design constraints that shape the code

• Creation is billed once. The job is charged when it is created, from the requested duration. So retrying a create request is only safe when you know no job was made.
• Polling is free. Poll liberally, but with a deadline and jitter.
• Errors have a stable envelope. {"error": {"message", "type", "code"}}, so you can branch on status and code.
• Output URLs are delivery links. Download them promptly rather than storing the URL.

The retry rules come from the error-handling cookbook: retry 429 after Retry-After, retry 5xx with backoff, never retry 400, 401, 402 or 403, and never blindly retry a connection error on creation.

The client

from __future__ import annotations

import os
import random
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Optional

import requests

BASE = "https://videorouter.sh/api/v1"
RETRY_STATUS = {429, 500, 502, 503, 504}


class VideoApiError(Exception):
    def __init__(self, status: int, code: Optional[str], message: str):
        super().__init__(f"{status} {code}: {message}")
        self.status, self.code, self.message = status, code, message


class AmbiguousCreate(Exception):
    # Connection dropped on create: a job may or may not exist.


class JobFailed(Exception):
    pass


@dataclass
class Job:
    id: str
    status: str
    raw: dict[str, Any] = field(repr=False, default_factory=dict)

    @property
    def done(self) -> bool:
        return self.status in ("completed", "failed")

    @property
    def url(self) -> Optional[str]:
        data = self.raw.get("data") or []
        return data[0]["url"] if data else None


class VideoClient:
    def __init__(self, api_key: Optional[str] = None, base: str = BASE,
                 max_attempts: int = 5, timeout: tuple[float, float] = (5, 30)):
        key = api_key or os.environ["LLMR_API_KEY"]
        self.base, self.max_attempts, self.timeout = base, max_attempts, timeout
        self.s = requests.Session()
        self.s.headers["Authorization"] = f"Bearer {key}"

    def _request(self, method: str, path: str, *, retry_conn: bool, **kw) -> dict:
        for attempt in range(self.max_attempts):
            try:
                r = self.s.request(method, f"{self.base}{path}",
                                   timeout=self.timeout, **kw)
            except (requests.ConnectionError, requests.Timeout):
                if not retry_conn:
                    raise AmbiguousCreate(path)
                self._sleep(attempt)
                continue
            if r.ok:
                return r.json()
            err = (r.json().get("error") or {}) if r.content else {}
            if r.status_code not in RETRY_STATUS or attempt == self.max_attempts - 1:
                raise VideoApiError(r.status_code, err.get("code"), err.get("message", r.text))
            wait = float(r.headers.get("Retry-After") or 0)
            self._sleep(attempt, wait)
        raise VideoApiError(503, "exhausted", "retries exhausted")

    @staticmethod
    def _sleep(attempt: int, floor: float = 0.0) -> None:
        time.sleep(max(floor, min(60, 2 ** attempt)) + random.uniform(0, 1))

    def create(self, model: str, prompt: str, duration_secs: int = 5, **extra) -> Job:
        body = {"model": model, "prompt": prompt, "duration_secs": duration_secs, **extra}
        j = self._request("POST", "/videos", json=body, retry_conn=False)
        return Job(j["id"], j["status"], j)

    def get(self, job_id: str) -> Job:
        j = self._request("GET", f"/videos/{job_id}", retry_conn=True)
        return Job(j["id"], j["status"], j)

    def wait(self, job_id: str, deadline_s: float = 900, interval_s: float = 5) -> Job:
        end = time.monotonic() + deadline_s
        while time.monotonic() < end:
            job = self.get(job_id)
            if job.done:
                if job.status == "failed":
                    raise JobFailed(str(job.raw.get("error")))
                return job
            time.sleep(interval_s + random.uniform(0, 1.5))
        raise TimeoutError(f"{job_id} not finished after {deadline_s}s")

    def download(self, job: Job, dest: Path) -> Path:
        with requests.get(job.url, stream=True, timeout=(5, 60)) as r:
            r.raise_for_status()
            with open(dest, "wb") as f:
                for chunk in r.iter_content(1 << 20):
                    f.write(chunk)
        return dest

Using it

c = VideoClient()
job = c.create("kling-v3.0-std", "a paper airplane gliding over a city", 5,
               aspect_ratio="16:9")
job = c.wait(job.id)
c.download(job, Path("airplane.mp4"))

The model id there is one documented on the video generation docs and the quickstart; any other id from the catalog works the same way. Extra keyword arguments such as start_image_url, resolution or a provider dict go straight into the body.

Why create and get retry differently

This is the one choice that matters. get is read-only, so it is safe to retry on connection errors. create is not: if the connection drops after the server accepted the job, you cannot tell whether a job exists, and a second attempt could be a second charge. The client raises AmbiguousCreate instead of guessing. What you do next is a policy for your application. A conservative choice is to record the request, alert, and let a human or a reconciliation job check the dashboard for a matching job before resubmitting.

The status-code retry is different: a 429 or 5xx is an actual response, so the server told you the create did not go through. The documentation says that a failure where every upstream host failed is not billed, which is what makes those retries safe.

Things to add before production

• Your own job row. Persist an id, the payload and a state before calling create, and store the returned job id after. That row is how you resolve an ambiguous create. The production pipeline guide shows the worker layout.
• Typed errors per code. Map 402 spend_cap_exceeded and insufficient_credits to different exceptions, because they page different people.
• Deduplication. Hash model, prompt, inputs and duration, and refuse to submit the same hash twice inside a short window.
• Size verification. Unsupported resolution or aspect ratio is ignored rather than rejected, so probe the downloaded file's dimensions.
• Async variant. If you run inside an event loop, port the same structure to httpx.AsyncClient and asyncio.sleep; the retry rules do not change.

Testing it without spending

Unit-test the retry logic by injecting a fake Session that returns a scripted sequence (429 with a header, 503, then 200) and assert the number of calls and sleeps. For the integration path, use a throwaway key with a small monthly cap and a one-model allow-list, as described in the spend controls guide, so a bug costs a 402 rather than an invoice. When you are ready, create a key and run the snippet above.

Frequently asked questions

Keep reading

• Async Video Jobs Explained — Polling, Timeouts and Retries
• Video API Authentication and Spend Controls: Keys, Caps, Limits
• Video API Error Handling Cookbook: Retries, Backoff, Duplicates
• Production Video Generation Pipeline: Queue, Workers, Python