Search for the best AI video generator today and you land on a ranking that goes stale in about three weeks. Kling AI shipped Kling 4.0 in October 2026, and its launch announcement promised native 3 to 30 second clips, up to 15 reference assets and up to 10 keyframes. Runway’s developer pages, refreshed this week, now list Veo 3.1, Seedance 2.0, Wan 3.0 and Kling 3.0 behind one API key and one credit balance.
A ranking cannot keep up with that pace. A router can. This tutorial builds a small Python tool that reads a catalog of video models, filters it by what one specific shot needs (duration, resolution, audio, references, budget), ranks the survivors and falls down the list when a model fails or vanishes. You can run all of it offline in ten minutes, then point it at real providers. If you want an opinionated list first, read our best AI video generator ranking and the 12-step benchmark harness that scores output quality. This guide covers the other half: choosing and failing over at runtime.
The finished project is about 500 lines across seven files, ships with 12 passing tests, and costs nothing to try because it includes an offline mock provider. A real 8-second clip on a fast tier costs a little over a dollar at list prices, so a live smoke test against any AI video generator API is cheap too.
Don't miss new tech stories on Google
Add Tech Insider once in the Google app and our stories appear in your news suggestions.
Why the Best AI Video Generator Depends on the Shot
Ask five teams which AI video generator is best and you get five answers, and all of them can be right. A product teaser with spoken dialogue needs native audio. A silent looping background needs the cheapest 720p tier. A 25-second brand film needs a model that can run past 15 seconds in one pass. The best AI video generator for each of those jobs is a different model, and the answer shifts every time a vendor ships a new version.
Three failure modes show up again and again once a team moves from experiments to production. First, people pick one favorite model and use it for everything, overpaying on easy shots. Second, price comparisons quietly mix units. One provider bills per second, another per clip, a third in credits. Research gathered for this article put the problem plainly: prices are not directly comparable unless you fix resolution, duration, audio, quality tier, currency, endpoint and the billing unit first. Third, model IDs disappear. One October 2026 ranking on lumienai.com is titled “AI Video Generators Ranked for October 2026: Sora Is Gone”, and an open-source agent project had to ship a commit titled fix(google): video_generate fails after Veo 3.1 Preview retirement. Treat both as reported signs of how fast IDs move, not as a complete history.
A router answers all three. It keeps the facts about each model in one reviewed file, refuses to guess when a fact is missing, prices every request before sending it, and moves to the next candidate when a call fails. The quality judgment stays with you, which is where it belongs.
What You Will Build and What It Costs
The tool, called vidrouter, takes a job description and returns a ranked plan. Here is what each piece does:
- Catalog: a JSON file with one row per model. Each row records route, status, duration limits, audio, reference limits, pricing and the date someone last checked it.
- Filter: hard requirements first. A model that cannot do the duration, resolution, audio or reference count is out, and so is any model whose key fact is unconfirmed.
- Pricer: handles per-second and per-clip billing and refuses to quote a price it cannot support.
- Ranker: cheapest, quality or balanced policies, with a deterministic tiebreak.
- Runner: retries rate limits with backoff, skips retired models immediately and writes every attempt to a JSONL ledger.
- Drift check: compares your catalog with the model list a provider actually serves today.
Offline runs cost $0. A live test is cheap: eight seconds on a fast tier at the $0.15 per second list rate comes to $1.20. If you want to go deeper on spend tracking for a single vendor, our Gemini Omni Flash cost ledger tutorial shows a budget guard you can bolt onto this runner later.
Prerequisites and Versions
You need a terminal and basic Python. Everything below was tested on Linux, and it should work unchanged on macOS or WSL.
- Python 3.11 or newer (tested on 3.12.3). The project uses the standard-library
tomllib, which arrived in 3.11. - httpx for the HTTP adapter. The offline mock needs no network library at all.
- pytest 9.1.1 for the test suite.
- FFmpeg with ffprobe (any recent build), optional, for checking delivered clip length in the advanced section.
- A Runway Dev account and API key, optional, only for the live smoke test. The router also works with any provider you describe in
providers.toml. - Time: about 70 minutes, including reading the outputs.
The Video Model Lineup on October 27, 2026
The catalog you build in Step 4 starts from what could be confirmed on the day of writing. Every price below is a list price captured on October 27, 2026, and several come through Runway Dev, which now acts as a multi-provider API layer. Its documentation says one credit equals $0.01, so a 40-credit price is $0.40. See the Kling 3.0 page on Runway Dev for that credit math and the Standard, Pro and 4K tiers.
| Model | Route in the demo | List price used by the router | Not confirmed by our sources |
|---|---|---|---|
| Veo 3.1 | Runway Dev | $0.40 per second with audio | Per-resolution pricing on this route |
| Veo 3.1 Fast | Runway Dev | $0.15 per second with audio, $0.10 without | Reference image limits |
| Seedance 2.0 | Runway Dev | $0.14 (480p), $0.30 (720p), $0.40 (1080p), $1.50 (4K) per second | Audio support |
| Wan 3.0 | Runway Dev | $0.05 (480p), $0.10 (720p), $0.20 (1080p) per second | Audio, stable seed parameter |
| Wan 3.0 Prime | Runway Dev | $0.14 (720p), $0.28 (1080p) per second | Audio |
| Kling 3.0 | Runway Dev | $0.13 (720p), $0.17 (1080p), $0.43 (4K) per second, audio included | Reference support (lives on a separate Kling O3 model) |
| Gemini Omni Flash 1.1 | Runway Dev | $0.10 (720p), $0.15 (1080p), $0.30 (4K) per second | Audio |
| Kling 4.0 | Kling direct | No confirmed API price sheet | API identifier, price, rate limits |
| Hailuo 2.3 (older family) | MiniMax direct | $0.28 (768p, 6 s), $0.56 (768p, 10 s), $0.49 (1080p, 6 s) per clip | Still available, but no longer the newest MiniMax family |
What the sources confirm
Google’s own Veo page identifies Veo 3.1 as the current Veo model, covering text-to-video, image-to-video and first-and-last-frame control. Kling 4.0 is the newest Kling release, with Kling 4.0 Flash in limited early access since September 28. MiniMax announced its H3 video model on July 31, 2026, with text, image, video and audio context, native stereo sound, clips up to 15 seconds and output up to 2K. Note that MiniMax calls it H3, not Hailuo 3.0, even though secondary coverage uses both names.
What they do not confirm
Kling’s launch material says API access for enterprise customers and developers is coming, and as of October 27 official sources still show no confirmed price sheet or stable API identifier. One third-party announcement claims the Kling 4.0 API went live on October 22, which we could not verify. MiniMax’s pricing overview reportedly lists H3 at $0.08 for 768p and $0.13 for 2K, yet it also says video packages do not support H3 yet, so the billing unit needs checking before you trust those numbers. Your router has to handle exactly this kind of half-known model, which is why Step 4 gives every row a price_verified flag.
For a hands-on walk through one of these vendors alone, our Veo 3.1 API tutorial covers authentication and polling in detail. The router below treats that kind of call as just one adapter among several.
Step 1: Create the Project and Virtual Environment
Start with an empty folder and an isolated environment. The layout below keeps provider code, selection logic and tests apart, so you can swap any one of them without touching the others.
mkdir vidrouter && cd vidrouter
python3 -m venv .venv
source .venv/bin/activate
mkdir router tests
touch router/__init__.py
python --versionExpected output on a current Linux box:
Python 3.12.3If you see a version below 3.11, install a newer interpreter before going on. The TOML loader in Step 10 needs it.
Step 2: Pin Dependencies and Store Keys Safely
Pin the two third-party packages so a surprise upgrade cannot change behavior between your laptop and CI. Keys go in environment variables only, never in the catalog or the repository.
cat > requirements.txt <<'EOF'
httpx==0.28.1
pytest==9.1.1
EOF
pip install -r requirements.txt
# only needed for the live smoke test in Step 13
export RUNWAY_DEV_API_KEY="paste-your-key-here"
printf '.venv/\nledger.jsonl\n__pycache__/\n' > .gitignoreThe ledger file is in the ignore list on purpose. It records job hashes, model IDs and output URLs, and you rarely want that in version control.
Step 3: Define the Job Spec and the Model Entry
Two small dataclasses carry everything the router needs. JobSpec describes the shot. ModelEntry describes what a model can do. The detail that matters is the type of audio, max_refs and first_last: they are Optional, so None can mean “nobody has confirmed this” and stay distinct from False.
from dataclasses import dataclass
from typing import Optional
@dataclass(frozen=True)
class JobSpec:
prompt: str
duration_s: int
resolution: str = "720p"
needs_audio: bool = False
ref_images: int = 0
first_last_frame: bool = False
max_cost: Optional[float] = None
policy: str = "balanced" # cheapest | quality | balanced
allow_unknown: bool = False
allow_legacy: bool = False
@dataclass
class ModelEntry:
id: str
label: str
route: str
status: str # current | legacy
quality_tier: int # your own score, 1 to 3
min_s: int
max_s: int
audio: Optional[bool] # None means "not confirmed"
max_refs: Optional[int]
first_last: Optional[bool]
pricing: Optional[dict] # {"unit": "second" | "clip", "table": {...}}
price_verified: bool
snapshot: str # ISO date the row was last checked
notes: str = ""
@dataclass
class Plan:
candidates: list # list of (ModelEntry, estimated_cost), best first
rejected: dict # model id -> list of reasonsCreate router/models.py with that content. The quality_tier field is your own 1 to 3 score. Nothing in the router pretends it is an industry benchmark, and you should overwrite the starting values after you run your own blind tests (our guide on how to test AI image generators lays out a method that transfers to video).
Step 4: Build the Catalog With Verified Flags
The catalog is the single source of truth, and it is plain JSON so a reviewer can diff it in a pull request. Each row carries a snapshot date and a price_verified flag. Kling 4.0 has pricing set to null because no confirmed price sheet existed on October 27. MiniMax H3 carries a price but price_verified is false, and Hailuo 2.3 uses per-clip billing. For a broader menu of tools to seed the file, the text-to-video tools roundup lists more candidates.
[
{"id": "veo-3.1", "label": "Google Veo 3.1 (with audio)", "route": "runway_dev", "status": "current", "quality_tier": 3, "min_s": 1, "max_s": 10, "audio": true, "max_refs": null, "first_last": true, "pricing": {"unit": "second", "table": {"any": 0.40}}, "price_verified": true, "snapshot": "2026-10-27", "notes": "Flat $0.40/s with audio on Runway Dev"},
{"id": "veo-3.1-fast", "label": "Google Veo 3.1 Fast (with audio)", "route": "runway_dev", "status": "current", "quality_tier": 2, "min_s": 1, "max_s": 10, "audio": true, "max_refs": null, "first_last": true, "pricing": {"unit": "second", "table": {"any": 0.15}}, "price_verified": true, "snapshot": "2026-10-27", "notes": "$0.15/s with audio"},
{"id": "veo-3.1-fast-silent", "label": "Google Veo 3.1 Fast (no audio)", "route": "runway_dev", "status": "current", "quality_tier": 2, "min_s": 1, "max_s": 10, "audio": false, "max_refs": null, "first_last": true, "pricing": {"unit": "second", "table": {"any": 0.10}}, "price_verified": true, "snapshot": "2026-10-27", "notes": "$0.10/s without audio"},
{"id": "seedance-2.0", "label": "ByteDance Seedance 2.0", "route": "runway_dev", "status": "current", "quality_tier": 3, "min_s": 1, "max_s": 10, "audio": null, "max_refs": null, "first_last": null, "pricing": {"unit": "second", "table": {"480p": 0.14, "720p": 0.30, "1080p": 0.40, "4k": 1.50}}, "price_verified": true, "snapshot": "2026-10-27", "notes": "Audio support not confirmed in our sources"},
{"id": "wan-3.0", "label": "Alibaba Wan 3.0 (standard)", "route": "runway_dev", "status": "current", "quality_tier": 2, "min_s": 1, "max_s": 10, "audio": null, "max_refs": null, "first_last": null, "pricing": {"unit": "second", "table": {"480p": 0.05, "720p": 0.10, "1080p": 0.20}}, "price_verified": true, "snapshot": "2026-10-27", "notes": "Seed parameter not confirmed"},
{"id": "wan-3.0-prime", "label": "Alibaba Wan 3.0 Prime", "route": "runway_dev", "status": "current", "quality_tier": 2, "min_s": 1, "max_s": 10, "audio": null, "max_refs": null, "first_last": null, "pricing": {"unit": "second", "table": {"720p": 0.14, "1080p": 0.28}}, "price_verified": true, "snapshot": "2026-10-27", "notes": "Speed-optimized tier"},
{"id": "kling-3.0", "label": "Kling 3.0 (Standard/Pro/4K)", "route": "runway_dev", "status": "current", "quality_tier": 2, "min_s": 1, "max_s": 10, "audio": true, "max_refs": null, "first_last": null, "pricing": {"unit": "second", "table": {"720p": 0.13, "1080p": 0.17, "4k": 0.43}}, "price_verified": true, "snapshot": "2026-10-27", "notes": "Prices include audio"},
{"id": "gemini-omni-flash-1.1", "label": "Gemini Omni Flash 1.1", "route": "runway_dev", "status": "current", "quality_tier": 2, "min_s": 1, "max_s": 10, "audio": null, "max_refs": null, "first_last": null, "pricing": {"unit": "second", "table": {"720p": 0.10, "1080p": 0.15, "4k": 0.30}}, "price_verified": true, "snapshot": "2026-10-27", "notes": "Audio support not confirmed in our sources"},
{"id": "kling-4.0", "label": "Kling 4.0", "route": "kling_direct", "status": "current", "quality_tier": 3, "min_s": 3, "max_s": 30, "audio": null, "max_refs": 15, "first_last": true, "pricing": null, "price_verified": false, "snapshot": "2026-10-27", "notes": "No confirmed API price sheet yet"},
{"id": "minimax-h3", "label": "MiniMax H3 (Hailuo)", "route": "minimax_direct", "status": "current", "quality_tier": 3, "min_s": 4, "max_s": 15, "audio": true, "max_refs": null, "first_last": true, "pricing": {"unit": "second", "table": {"768p": 0.08, "2k": 0.13}}, "price_verified": false, "snapshot": "2026-10-27", "notes": "Billing unit unconfirmed: listed rates may be per clip"},
{"id": "hailuo-2.3", "label": "MiniMax Hailuo 2.3 (older family)", "route": "minimax_direct", "status": "legacy", "quality_tier": 1, "min_s": 6, "max_s": 10, "audio": null, "max_refs": null, "first_last": null, "pricing": {"unit": "clip", "table": {"768p:6": 0.28, "768p:10": 0.56, "1080p:6": 0.49}}, "price_verified": true, "snapshot": "2026-10-27", "notes": "Billed per clip, 6 or 10 seconds only"}
]One honesty note about this file. The min_s and max_s values are vendor limits only for Kling 4.0 (3 to 30 seconds) and MiniMax H3 (4 to 15 seconds). Everywhere else the demo uses a house cap of 1 to 10 seconds, meaning the longest clip this team is willing to request. Replace those caps with documented limits as you confirm them.
Step 5: Load, Validate and Age-Check the Catalog
Save the loader as router/catalog.py. A bad catalog row should fail at load time, not halfway through an expensive job. The loader checks required keys, duplicate IDs, status values and impossible duration ranges. A second function flags rows nobody has re-checked recently, which matters because list prices on this market change in weeks.
import json
from datetime import date
from pathlib import Path
from .models import ModelEntry
REQUIRED = {"id", "route", "status", "quality_tier", "min_s", "max_s", "snapshot"}
def load_catalog(path: str = "catalog.json") -> list[ModelEntry]:
rows = json.loads(Path(path).read_text())
seen, entries = set(), []
for row in rows:
missing = REQUIRED - row.keys()
if missing:
raise ValueError(f"{row.get('id', '?')}: missing {sorted(missing)}")
if row["id"] in seen:
raise ValueError(f"duplicate model id {row['id']}")
if row["status"] not in ("current", "legacy"):
raise ValueError(f"{row['id']}: bad status {row['status']!r}")
if row["min_s"] > row["max_s"]:
raise ValueError(f"{row['id']}: min_s > max_s")
seen.add(row["id"])
entries.append(ModelEntry(**row))
return entries
def stale_rows(entries: list[ModelEntry], today: date, max_age_days: int = 14) -> list[str]:
out = []
for e in entries:
age = (today - date.fromisoformat(e.snapshot)).days
if age > max_age_days:
out.append(f"{e.id} last checked {age} days ago")
return outTry the age check with a pretend date. A row last checked on October 27 is fine three days later and stale in late November:
catalog ok
STALE:
veo-3.1 last checked 24 days ago
veo-3.1-fast last checked 24 days ago
veo-3.1-fast-silent last checked 24 days agoThe CLI returns exit code 1 when anything is stale, so a nightly CI job can fail loudly. We wire that up in Step 13.
Step 6: Price a Clip, Per Second or Per Clip
Pricing is where most hand-rolled comparisons go wrong, whether the job is text to video AI or image-to-video. Veo and Wan bill by the second, with some rates varying by resolution. Hailuo 2.3 bills per finished clip at fixed durations. The function below handles both and returns None instead of guessing when it has no row for the requested resolution and duration.
from .models import JobSpec, ModelEntry, Plan
def estimate_cost(m: ModelEntry, job: JobSpec):
if not m.pricing:
return None
table = m.pricing["table"]
if m.pricing["unit"] == "second":
rate = table.get(job.resolution, table.get("any"))
return None if rate is None else round(rate * job.duration_s, 4)
return table.get(f"{job.resolution}:{job.duration_s}")Save the block above as router/select.py. A flat rate stored under the key any covers cases like Veo 3.1 on Runway Dev, where our sources list a single per-second figure. Test it quickly in a Python shell:
3.2
0.28
NoneThe last line prints None because Hailuo 2.3 has no 8-second clip price. Seedance 2.0 at 1080p for eight seconds is 8 times $0.40, or $3.20, matching its list price of 40 credits per second.
Step 7: Filter by Hard Requirements
Filtering decides who is allowed to compete. The function collects every reason a model fails instead of stopping at the first, so the CLI can print a full rejection list later. Missing facts reject a model unless the job explicitly sets allow_unknown.
def _flag(value, wanted, name, job, reasons):
if not wanted:
return
if value is None and not job.allow_unknown:
reasons.append(f"{name} unconfirmed")
elif value is False:
reasons.append(f"no {name}")
def check(m: ModelEntry, job: JobSpec):
reasons = []
if m.status == "legacy" and not job.allow_legacy:
reasons.append("legacy model")
if not m.min_s <= job.duration_s <= m.max_s:
reasons.append(f"duration outside {m.min_s}-{m.max_s}s")
_flag(m.audio, job.needs_audio, "audio", job, reasons)
_flag(m.first_last, job.first_last_frame, "first/last frame", job, reasons)
if job.ref_images:
if m.max_refs is None and not job.allow_unknown:
reasons.append("reference limit unconfirmed")
elif m.max_refs is not None and job.ref_images > m.max_refs:
reasons.append(f"max {m.max_refs} references")
cost = estimate_cost(m, job)
if cost is None:
reasons.append(f"no price for {job.resolution}/{job.duration_s}s")
elif not m.price_verified and not job.allow_unknown:
reasons.append("price unverified")
elif job.max_cost is not None and cost > job.max_cost:
reasons.append(f"over budget (${cost:.2f})")
return reasons, costAppend this block to router/select.py. Read the pricing branch closely. A model with no price, an unverified price or a price over budget never reaches the ranker. That single rule keeps Kling 4.0 and MiniMax H3 out of automated routing until a human confirms their numbers, even though both look attractive on paper.
Step 8: Rank the Survivors by Policy
Three policies cover most teams. cheapest sorts by cost and breaks ties on quality. quality does the reverse. balanced rewards your quality tier and subtracts half of the cost share relative to the priciest eligible option, which stops a premium model winning by a hair on a throwaway shot. The tier itself should come from a blind rating pass like the one in our image API benchmark.
def rank(job: JobSpec, catalog: list[ModelEntry]) -> Plan:
ok, rejected = [], {}
for m in catalog:
reasons, cost = check(m, job)
if reasons:
rejected[m.id] = reasons
else:
ok.append((m, cost))
top = max((c for _, c in ok), default=1) or 1
if job.policy == "cheapest":
ok.sort(key=lambda mc: (mc[1], -mc[0].quality_tier, mc[0].id))
elif job.policy == "quality":
ok.sort(key=lambda mc: (-mc[0].quality_tier, mc[1], mc[0].id))
else:
ok.sort(key=lambda mc: (-(mc[0].quality_tier / 3 - 0.5 * mc[1] / top), mc[0].id))
return Plan(candidates=ok, rejected=rejected)Append the ranker to the same file, then create cli.py in Step 13 to see output. Until then, here is what the plan looks like. Run the router for an 8-second, 1080p clip with audio and the cheapest policy. Only models with confirmed audio qualify:
# model est. cost route
1 veo-3.1-fast $1.20 runway_dev
2 kling-3.0 $1.36 runway_dev
3 veo-3.1 $3.20 runway_dev
rejected:
veo-3.1-fast-silent no audio
seedance-2.0 audio unconfirmed
wan-3.0 audio unconfirmed
wan-3.0-prime audio unconfirmed
gemini-omni-flash-1.1 audio unconfirmed
kling-4.0 audio unconfirmed; no price for 1080p/8s
minimax-h3 no price for 1080p/8s
hailuo-2.3 legacy model; audio unconfirmed; no price for 1080p/8sThree models survive: Veo 3.1 Fast at $1.20, Veo 3.1 at $3.20 and Kling 3.0, which is listed at $0.13 per second at 720p with audio. Allow unknowns and the pool widens:
# model est. cost route
1 seedance-2.0 $3.20 runway_dev
2 veo-3.1 $3.20 runway_dev
3 gemini-omni-flash-1.1 $1.20 runway_dev
4 veo-3.1-fast $1.20 runway_dev
5 kling-3.0 $1.36 runway_dev
6 wan-3.0 $1.60 runway_dev
7 wan-3.0-prime $2.24 runway_dev
rejected:| Model | 1080p rate per second | 8-second clip | Audio confirmed in our sources |
|---|---|---|---|
| Veo 3.1 Fast | $0.15 | $1.20 | Yes (rate is with audio) |
| Gemini Omni Flash 1.1 | $0.15 | $1.20 | No |
| Kling 3.0 (Pro tier) | $0.17 | Not listed | Yes (rate is with audio) |
| Wan 3.0 | $0.20 | $1.60 | No |
| Wan 3.0 Prime | $0.28 | $2.24 | No |
| Veo 3.1 | $0.40 | $3.20 | Yes (rate is with audio) |
| Seedance 2.0 | $0.40 | $3.20 | No |
This is the right place to stop and think about the title question. Among models with confirmed audio, the best AI video generator for this exact shot is Veo 3.1 Fast if price decides and Veo 3.1 if your tier scores decide. The router shows both and lets policy pick.
Step 9: Add Provider Errors and an Offline Mock
The runner needs to tell failures apart, because the right reaction differs. A rate limit deserves a retry with backoff. A retired model deserves an immediate skip. A content-policy rejection deserves the next candidate. ProviderError.kind carries that distinction, and MockProvider lets you rehearse every case offline.
import os
import tomllib
from pathlib import Path
class ProviderError(Exception):
def __init__(self, kind: str, message: str = ""):
super().__init__(f"{kind}: {message}")
self.kind = kind # retired | rate_limit | transient | policy | auth
class MockProvider:
"""Offline stand-in so the whole pipeline runs without keys or spend."""
def __init__(self, retired=(), flaky=()):
self.retired, self.flaky, self._jobs, self._hits = set(retired), set(flaky), {}, {}
def submit(self, model_id, job):
if model_id in self.retired:
raise ProviderError("retired", f"{model_id} is no longer served")
if model_id in self.flaky:
self._hits[model_id] = self._hits.get(model_id, 0) + 1
if self._hits[model_id] == 1:
raise ProviderError("rate_limit", "429 too many requests")
handle = f"mock-{model_id}-{len(self._jobs)}"
self._jobs[handle] = {"polls": 0, "seconds": job.duration_s}
return handle
def poll(self, handle):
job = self._jobs[handle]
job["polls"] += 1
if job["polls"] < 2:
return {"state": "pending"}
return {"state": "done", "url": f"https://example.invalid/{handle}.mp4", "billed_s": job["seconds"]}
def list_models(self):
return [m for m in ("veo-3.1", "veo-3.1-fast", "veo-3.1-fast-silent", "seedance-2.0",
"wan-3.0", "wan-3.0-prime", "kling-3.0", "gemini-omni-flash-1.1",
"kling-4.0", "minimax-h3", "hailuo-2.3") if m not in self.retired]Create router/providers.py with the block above first, then add the HTTP adapter from the next step to the same file. The mock returns a pending state on the first poll and a finished clip on the second, which exercises the polling loop without any network traffic.
Step 10: Add a Config-Driven HTTP Adapter
Every AI video generator API differs in paths, field names and status strings, and those details change. Rather than hard-code guesses, the adapter reads them from providers.toml, and you copy each value from the vendor’s own API documentation. The status mapping is the part worth studying: the adapter sorts HTTP errors into its own categories, so rate-limit responses become rate_limit, missing or gone models become retired, rejected requests become policy and server errors become transient. We used the same layered idea in the fallback chain for image generator APIs, and it carries over to video unchanged.
class HttpProvider:
"""Config-driven REST adapter. Endpoint paths and field names come from providers.toml."""
def __init__(self, cfg: dict):
self.cfg = cfg
key = os.environ.get(cfg["api_key_env"])
if not key:
raise ProviderError("auth", f"set {cfg['api_key_env']}")
self.headers = {"Authorization": f"Bearer {key}", **cfg.get("headers", {})}
def _client(self):
import httpx
return httpx.Client(base_url=self.cfg["base_url"], headers=self.headers, timeout=60)
def _raise(self, resp):
if resp.status_code == 429:
raise ProviderError("rate_limit", resp.text[:200])
if resp.status_code in (401, 403):
raise ProviderError("auth", resp.text[:200])
if resp.status_code in (404, 410):
raise ProviderError("retired", resp.text[:200])
if resp.status_code == 422:
raise ProviderError("policy", resp.text[:200])
if resp.status_code >= 500:
raise ProviderError("transient", resp.text[:200])
resp.raise_for_status()
def submit(self, model_id, job):
f = self.cfg["fields"]
body = {f["model"]: model_id, f["prompt"]: job.prompt,
f["duration"]: job.duration_s, f["resolution"]: job.resolution}
with self._client() as c:
r = c.post(self.cfg["submit_path"], json=body)
self._raise(r)
return r.json()[self.cfg["task_id_field"]]
def poll(self, handle):
with self._client() as c:
r = c.get(self.cfg["status_path"].format(task_id=handle))
self._raise(r)
data = r.json()
status = data[self.cfg["status_field"]]
if status in self.cfg["done_values"]:
return {"state": "done", "url": data[self.cfg["output_field"]],
"billed_s": data.get(self.cfg.get("billed_field", ""), None)}
if status in self.cfg["failed_values"]:
raise ProviderError("policy", str(data)[:200])
return {"state": "pending"}
def list_models(self):
with self._client() as c:
r = c.get(self.cfg["models_path"])
self._raise(r)
return [row[self.cfg["model_id_field"]] for row in r.json()[self.cfg["models_list_field"]]]
def load_providers(path: str = "providers.toml") -> dict:
cfg = tomllib.loads(Path(path).read_text())
return {name: HttpProvider(route) for name, route in cfg["routes"].items()}# Copy every value marked REPLACE from the provider's own API docs.
[routes.runway_dev]
base_url = "REPLACE_BASE_URL"
api_key_env = "RUNWAY_DEV_API_KEY"
submit_path = "REPLACE_SUBMIT_PATH"
status_path = "REPLACE_STATUS_PATH/{task_id}"
models_path = "REPLACE_MODELS_PATH"
task_id_field = "id"
status_field = "status"
output_field = "output_url"
billed_field = "billed_seconds"
model_id_field = "id"
models_list_field = "data"
done_values = ["SUCCEEDED"]
failed_values = ["FAILED", "CANCELLED"]
[routes.runway_dev.headers]
X-Runway-Version = "2026-06" # confirm the header name in the docs
[routes.runway_dev.fields]
model = "model"
prompt = "promptText"
duration = "duration"
resolution = "resolution"Every value starting with REPLACE is a placeholder on purpose. Fill them from the docs of the route you use. The one non-placeholder value is the API version header. Runway Dev’s documentation says you can pin version 2026-06 for stable behavior or use unsafe-latest to receive ongoing improvements. Pin the dated one for anything that runs unattended, and confirm the exact header name in the docs when you copy it.
Step 11: Run Jobs With Retries, Fallback and a Ledger
Save the runner as router/runner.py. It is deliberately short. It asks the ranker for a plan, walks the candidates in order and writes one JSONL line per attempt. Rate limits and transient errors retry with exponential backoff capped at 30 seconds. Retired, policy and auth errors break out of the retry loop at once and move to the next model, since repeating them wastes time and money.
import hashlib
import json
import time
from pathlib import Path
from .models import JobSpec
from .providers import ProviderError
from .select import rank
class RoutingFailed(Exception):
pass
def _log(path, row):
with open(path, "a") as fh:
fh.write(json.dumps(row) + "\n")
def run_job(job: JobSpec, catalog, providers, ledger="ledger.jsonl",
max_attempts=3, poll_every=2.0, timeout_s=600, sleep=time.sleep, now=time.time):
plan = rank(job, catalog)
job_id = hashlib.sha1(f"{job.prompt}|{job.duration_s}|{job.resolution}".encode()).hexdigest()[:10]
if not plan.candidates:
raise RoutingFailed(f"no eligible model: {plan.rejected}")
for model, est in plan.candidates:
provider = providers.get(model.route)
if provider is None:
_log(ledger, {"job": job_id, "model": model.id, "outcome": "no_provider"})
continue
for attempt in range(1, max_attempts + 1):
try:
handle = provider.submit(model.id, job)
deadline = now() + timeout_s
while True:
state = provider.poll(handle)
if state["state"] == "done":
break
if now() > deadline:
raise ProviderError("transient", "poll timeout")
sleep(poll_every)
billed = state.get("billed_s") or job.duration_s
row = {"job": job_id, "model": model.id, "outcome": "ok", "attempt": attempt,
"est_cost": est, "billed_s": billed, "url": state["url"]}
_log(ledger, row)
return row
except ProviderError as err:
_log(ledger, {"job": job_id, "model": model.id, "outcome": err.kind, "attempt": attempt})
if err.kind in ("retired", "policy", "auth"):
break
sleep(min(2 ** attempt, 30))
raise RoutingFailed(f"all {len(plan.candidates)} candidates failed, see {Path(ledger)}")Now simulate the scenario that hurts teams most: the cheapest model stops being served. Pass --retire veo-3.1-fast to the mock and watch the plan absorb it:
$ python cli.py run --mock --seconds 8 --res 1080p --audio --policy cheapest --retire veo-3.1-fast
{
"job": "1a087720fd",
"model": "kling-3.0",
"outcome": "ok",
"attempt": 1,
"est_cost": 1.36,
"billed_s": 8,
"url": "https://example.invalid/mock-kling-3.0-0.mp4"
}
$ cat ledger.jsonl
{"job": "1a087720fd", "model": "veo-3.1-fast", "outcome": "retired", "attempt": 1}
{"job": "1a087720fd", "model": "kling-3.0", "outcome": "ok", "attempt": 1, "est_cost": 1.36, "billed_s": 8, "url": "https://example.invalid/mock-kling-3.0-0.mp4"}The ledger shows one retired attempt on Veo 3.1 Fast, then a clean success on Kling 3.0 at the estimated $1.36. The job finished and nothing paged anyone. Because the ledger stores estimated cost and billed seconds side by side, you can reconcile it against the vendor’s invoice at month end.
Step 12: Detect Model Drift Before It Breaks a Job
Save the drift check as router/drift.py. Fallbacks protect a single job, but you also want to know the moment a catalog row goes stale. The drift check compares your catalog with the list of IDs each route reports today. It returns two lists: IDs you rely on that upstream no longer serves, and IDs upstream serves that you have never reviewed.
def detect_drift(catalog, live_by_route: dict) -> dict:
"""Compare catalog ids against what each route actually serves today."""
report = {"missing_upstream": [], "unlisted_upstream": []}
for route, live in live_by_route.items():
mine = {m.id for m in catalog if m.route == route}
live = set(live)
report["missing_upstream"] += sorted(f"{route}:{i}" for i in mine - live)
report["unlisted_upstream"] += sorted(f"{route}:{i}" for i in live - mine)
return reportRun it against the mock with Kling 3.0 pretend-removed:
{
"missing_upstream": [
"runway_dev:kling-3.0"
],
"unlisted_upstream": []
}Schedule the real version daily. When Google reportedly retired its Gemini API preview IDs for Veo 3.1 on October 22, the failure showed up downstream as broken video calls in projects that pinned the old names. A nightly drift report turns that into a Slack message you read at breakfast instead of a pager alert.
Step 13: Wire the CLI and Run the Tests
The CLI ties everything together with four commands: route prints a plan, run executes it, drift compares catalogs and check flags stale rows. Create cli.py in the project root.
import argparse
import json
import sys
from datetime import date
from router.catalog import load_catalog, stale_rows
from router.drift import detect_drift
from router.models import JobSpec
from router.providers import MockProvider, load_providers
from router.runner import RoutingFailed, run_job
from router.select import rank
def build_job(a) -> JobSpec:
return JobSpec(prompt=a.prompt, duration_s=a.seconds, resolution=a.res, needs_audio=a.audio,
ref_images=a.refs, first_last_frame=a.first_last, max_cost=a.max_cost,
policy=a.policy, allow_unknown=a.allow_unknown, allow_legacy=a.allow_legacy)
def main(argv=None):
p = argparse.ArgumentParser(prog="vidrouter")
p.add_argument("cmd", choices=["route", "run", "drift", "check"])
p.add_argument("--prompt", default="A slow dolly shot through a rainy neon market")
p.add_argument("--seconds", type=int, default=8)
p.add_argument("--res", default="1080p")
p.add_argument("--audio", action="store_true")
p.add_argument("--refs", type=int, default=0)
p.add_argument("--first-last", action="store_true")
p.add_argument("--max-cost", type=float)
p.add_argument("--policy", default="balanced", choices=["cheapest", "quality", "balanced"])
p.add_argument("--allow-unknown", action="store_true")
p.add_argument("--allow-legacy", action="store_true")
p.add_argument("--mock", action="store_true", help="use the offline mock provider")
p.add_argument("--retire", nargs="*", default=[], help="mock: model ids that return 'retired'")
p.add_argument("--catalog", default="catalog.json")
p.add_argument("--today", default=date.today().isoformat())
a = p.parse_args(argv)
catalog = load_catalog(a.catalog)
if a.cmd == "check":
stale = stale_rows(catalog, date.fromisoformat(a.today))
print("catalog ok" if not stale else "STALE:\n " + "\n ".join(stale))
return 1 if stale else 0
job = build_job(a)
if a.cmd == "route":
plan = rank(job, catalog)
print(f"{'#':<3}{'model':<24}{'est. cost':>10} route")
for i, (m, cost) in enumerate(plan.candidates, 1):
print(f"{i:<3}{m.id:<24}{'$%.2f' % cost:>10} {m.route}")
print("\nrejected:")
for mid, why in plan.rejected.items():
print(f" {mid:<24}{'; '.join(why)}")
return 0 if plan.candidates else 2
mock = MockProvider(retired=a.retire)
providers = {r: mock for r in {m.route for m in catalog}} if a.mock else load_providers()
if a.cmd == "drift":
live = {r: prov.list_models() for r, prov in providers.items()}
if a.mock:
live = {r: [i for i in mock.list_models() if any(m.id == i and m.route == r for m in catalog)]
for r in providers}
print(json.dumps(detect_drift(catalog, live), indent=2))
return 0
try:
print(json.dumps(run_job(job, catalog, providers, sleep=lambda s: None), indent=2))
except RoutingFailed as err:
print(f"FAILED: {err}", file=sys.stderr)
return 3
return 0
if __name__ == "__main__":
sys.exit(main())Then add the tests. They cover cost math, the unknown-versus-false distinction, budget limits, fallback on a retired model, retry on a rate limit, total failure and both maintenance checks.
from datetime import date
from router.catalog import load_catalog, stale_rows
from router.drift import detect_drift
from router.models import JobSpec
from router.providers import MockProvider
from router.runner import RoutingFailed, run_job
from router.select import estimate_cost, rank
import pytest
CAT = load_catalog("catalog.json")
BY_ID = {m.id: m for m in CAT}
def ids(plan):
return [m.id for m, _ in plan.candidates]
def test_per_second_cost():
job = JobSpec("x", 8, "1080p")
assert estimate_cost(BY_ID["seedance-2.0"], job) == 3.2
def test_per_clip_cost_needs_exact_duration():
assert estimate_cost(BY_ID["hailuo-2.3"], JobSpec("x", 6, "768p")) == 0.28
assert estimate_cost(BY_ID["hailuo-2.3"], JobSpec("x", 8, "768p")) is None
def test_audio_excludes_unconfirmed_by_default():
plan = rank(JobSpec("x", 8, "1080p", needs_audio=True, policy="cheapest"), CAT)
assert ids(plan) == ["veo-3.1-fast", "kling-3.0", "veo-3.1"]
assert plan.rejected["seedance-2.0"] == ["audio unconfirmed"]
def test_allow_unknown_widens_the_pool():
plan = rank(JobSpec("x", 8, "1080p", needs_audio=True, policy="cheapest", allow_unknown=True), CAT)
assert {"wan-3.0", "seedance-2.0"} <= set(ids(plan))
assert len(ids(plan)) > 3
def test_unpriced_and_unverified_models_never_route():
plan = rank(JobSpec("x", 8, "1080p", allow_unknown=True), CAT)
assert "kling-4.0" not in ids(plan)
strict = rank(JobSpec("x", 8, "768p"), CAT)
assert "minimax-h3" not in ids(strict)
assert "price unverified" in strict.rejected["minimax-h3"]
def test_legacy_blocked_unless_allowed():
assert "hailuo-2.3" not in ids(rank(JobSpec("x", 6, "768p"), CAT))
assert "hailuo-2.3" in ids(rank(JobSpec("x", 6, "768p", allow_legacy=True), CAT))
def test_budget_filter():
plan = rank(JobSpec("x", 8, "1080p", max_cost=1.5), CAT)
assert all(c <= 1.5 for _, c in plan.candidates)
def test_fallback_on_retired_model():
job = JobSpec("x", 8, "1080p", needs_audio=True, policy="cheapest")
mock = MockProvider(retired={"veo-3.1-fast"})
providers = {r: mock for r in {m.route for m in CAT}}
row = run_job(job, CAT, providers, ledger="/tmp/test-ledger.jsonl", sleep=lambda s: None)
assert row["model"] == "kling-3.0"
def test_retry_on_rate_limit_then_succeed():
job = JobSpec("x", 8, "1080p", needs_audio=True, policy="cheapest")
mock = MockProvider(flaky={"veo-3.1-fast"})
providers = {r: mock for r in {m.route for m in CAT}}
row = run_job(job, CAT, providers, ledger="/tmp/test-ledger.jsonl", sleep=lambda s: None)
assert row["model"] == "veo-3.1-fast" and row["attempt"] == 2
def test_all_fail_raises():
job = JobSpec("x", 8, "1080p", needs_audio=True)
mock = MockProvider(retired={"veo-3.1", "veo-3.1-fast", "kling-3.0"})
providers = {r: mock for r in {m.route for m in CAT}}
with pytest.raises(RoutingFailed):
run_job(job, CAT, providers, ledger="/tmp/test-ledger.jsonl", sleep=lambda s: None)
def test_drift_report():
live = {"runway_dev": ["veo-3.1", "brand-new-model"]}
rep = detect_drift(CAT, live)
assert "runway_dev:veo-3.1-fast" in rep["missing_upstream"]
assert rep["unlisted_upstream"] == ["runway_dev:brand-new-model"]
def test_stale_rows():
assert stale_rows(CAT, date(2026, 10, 30)) == []
assert len(stale_rows(CAT, date(2026, 12, 1))) == len(CAT)Run the suite:
............ [100%]
12 passed in 0.02sTwelve green tests means the selection, retry and drift logic behave as described. Then do a live smoke test when you are ready to spend a dollar. Fill in providers.toml, export your key and run a short, cheap shot:
python cli.py run --seconds 4 --res 720p --policy cheapest --max-cost 0.60Complete Working Project: Layout and a Full Run
Here is the final tree. If your listing matches this, you have the whole project:
vidrouter/
âââ catalog.json
âââ providers.toml
âââ requirements.txt
âââ cli.py
âââ router/
â âââ __init__.py
â âââ models.py
â âââ catalog.py
â âââ select.py
â âââ providers.py
â âââ runner.py
â âââ drift.py
âââ tests/
âââ test_router.pyOne more output worth reading is a failure. Ask for a 20-second clip with six reference images and allow unknowns. No model passes, because the only model that can run 20 seconds and take that many references, Kling 4.0, has no confirmed price:
# model est. cost route
rejected:
veo-3.1 duration outside 1-10s
veo-3.1-fast duration outside 1-10s
veo-3.1-fast-silent duration outside 1-10s
seedance-2.0 duration outside 1-10s
wan-3.0 duration outside 1-10s
wan-3.0-prime duration outside 1-10sThe empty plan with exit code 2 is the correct answer. It tells you exactly what to confirm next, instead of quietly sending a long job to a model that will bill you an unknown amount. When Kling publishes a price sheet, you edit one row, bump its snapshot date and the job routes. For wider context on where this fits among model tooling, browse our AI and machine learning coverage.
Common Pitfalls When Routing AI Video Models
These six mistakes cost teams the most money or the most debugging time:
- Comparing per-second to per-clip prices. Hailuo 2.3’s $0.28 for a 6-second 768p clip is a per-clip price, while most other listings are quoted per second, so the two are not comparable until you convert them at that exact duration. Normalize before you compare, and let the pricer return
Nonefor durations the vendor does not sell. We saw the same unit trap on image endpoints in our AI image generator API costs breakdown. - Treating unknown as yes. Our sources do not confirm audio on Seedance 2.0. Routing a dialogue shot to it on a hunch gives you a silent clip you paid for. Keep
Nonedistinct fromTrue. - Trusting an aggregator price as an official one. Third-party listings disagree. One comparison lists Wan 3.0 at $0.10 per second at 720p and another at $0.05, which suggests different providers, audio settings or discounts. Record which route each price came from.
- Floating on the latest API version. Runway Dev’s own docs offer a dated version for stability and an
unsafe-latestoption. Production code should pin the dated one and upgrade deliberately. - Retrying a dead model. A 404 will still be a 404 after three backoffs. Classify errors first, then decide whether a retry makes sense.
- Ignoring billing edge cases. One pricing report says Wan image-to-video requests may charge for input-video duration plus successful output seconds, with failed requests not billed. Log billed seconds from the response, not the seconds you asked for.
Troubleshooting: 10 Problems and Fixes
| Symptom | Likely cause | Fix |
|---|---|---|
ModuleNotFoundError: No module named 'router' | Running from the wrong folder, or router/__init__.py is missing | Run commands from the project root and create the empty __init__.py |
ModuleNotFoundError: tomllib | Python older than 3.11 | Install Python 3.11 or newer and recreate the virtual environment |
ValueError: duplicate model id | Two catalog rows share an id | Rename or merge the rows, then rerun |
| Plan is empty, exit code 2 | Every model was rejected by a hard requirement | Read the rejected list, confirm the missing fact in vendor docs or relax a flag |
| Everything shows “audio unconfirmed” | Rows have audio: null | Verify audio in docs and set true or false, or pass --allow-unknown for drafts |
ProviderError: auth: set RUNWAY_DEV_API_KEY | Environment variable not exported in this shell | Export the key again and confirm with echo ${RUNWAY_DEV_API_KEY:+set} |
| Every call returns “retired” | Wrong submit_path or base URL, so the server returns 404 | Compare both values with the docs, then run the drift command |
KeyError inside poll | Response field names differ from providers.toml | Print one raw response and fix status_field, output_field and done_values |
| Job loops until the timeout | The finished status string is not in done_values | Add the vendor’s exact success string and raise timeout_s for 4K jobs |
| Stale-catalog warning on every run | Snapshot dates were never refreshed | Re-verify each row against vendor pages, then update snapshot |
When a failure does not match this table, run the same job with --mock. If the mock succeeds, the logic is fine and the problem sits in the config or the network.
Advanced Tips for Production Use
- Feed tiers from your own scores. Replace the hand-set
quality_tierwith output from a blind rating run: fixed prompts, fixed references, and ratings collected before anyone sees the model name. - Add shot presets. Store named profiles such as
dialogue-teaserorsilent-loopin a small YAML or JSON file so producers pick a preset instead of seven flags. - Verify delivered length. Run
ffprobe -v error -show_entries format=duration -of csv=p=0 clip.mp4on every result and compare it withbilled_s. A mismatch is a billing question worth raising. - Use a circuit breaker per route. After three consecutive
transientfailures, skip a route for ten minutes instead of paying the retry delay on every job. - Prefer webhooks to polling when a provider offers them. Polling every two seconds across hundreds of jobs is a quiet way to hit rate limits.
- Keep a zero-cost lane. A local model is a useful last fallback for drafts. Our walkthrough on running Wan 2.2 locally in ComfyUI shows how to set one up, and you can describe it as another route in the catalog with a price of zero.
- Diff price snapshots. Commit
catalog.jsonchanges with a date in the message and let CI comment on any price that moved more than 20 percent.
FAQ: Picking and Routing the Best AI Video Generator
What is the best AI video generator right now?
There is no single winner. For a shot that needs confirmed native audio, Veo 3.1 Fast is a low-cost option in this catalog at $0.15 per second with audio, and Veo 3.1 costs $0.40 per second with audio. For long clips, Kling 4.0 supports 3 to 30 seconds natively, but its API pricing was not confirmed on October 27, 2026.
Can I run the router without paying for any API?
Yes. The --mock flag swaps every route for the offline provider, so selection, retries, fallback and the ledger all run with no key and no spend.
Is the Kling 4.0 API available?
Kling’s launch material says API access for enterprises and developers is coming. Runway Dev’s documentation lists Kling 4.0 among its models, but no official price sheet or stable identifier was confirmed on October 27. One third-party source claims an October 22 API launch. Until you confirm pricing, the router keeps it out of automated plans.
Why does the router refuse MiniMax H3?
Its catalog row has price_verified set to false because the pricing page lists H3 rates while also saying video packages do not support H3 yet. Confirm the billing unit in the API, update the row and the model joins the pool.
Do I have to use Runway Dev?
No. Runway Dev is simply the route that exposes several models behind one key, which keeps the demo short. Add any provider as a new section in providers.toml and point catalog rows at it.
How often should I refresh the catalog?
The age check flags rows after 14 days, and a weekly review fits how quickly this market moves. Launch weeks, such as Kling 4.0’s on October 15, justify an extra pass.
Does the router judge video quality?
No. It trusts the quality_tier you assign. Treat the starting values in this tutorial as placeholders and replace them with scores from your own blind tests.
