Video Generation at the API Level

Video generation is inherently async

A single video generation call takes 30 seconds to 5 minutes. Every reputable API is asynchronous: POST a job, get a task ID, poll (or listen on a webhook) for completion, download the result. Build this pattern into your product from day one — latency retrofits are expensive.

Sora 2 API (OpenAI)

OpenAI released Sora 2 via API in late 2025. Note: OpenAI announced April 2026 it will discontinue the Sora web/app experiences, with API discontinuation in September 2026 — so Sora is a short-term option, not a long-term bet. Structure code behind a provider interface.

from openai import OpenAI import time client = OpenAI() job = client.videos.generate( model="sora-2", prompt="A chef ladles broth into a ramen bowl, steam rising, 35mm film look. Camera dollies in slowly.", duration=10, # seconds resolution="1080p", aspect_ratio="16:9", ) # Poll for completion while True: status = client.videos.retrieve(job.id) if status.status == "completed": video_url = status.video_url break elif status.status == "failed": raise RuntimeError(status.error) time.sleep(5) # Download and store import urllib.request urllib.request.urlretrieve(video_url, "./ramen.mp4")Sora 2 via OpenAI Python SDK. Simple polling loop.

Veo 3.1 API (Google Vertex AI)

from google.cloud import aiplatform from google.protobuf import json_format from google.protobuf.struct_pb2 import Value aiplatform.init(project="my-project", location="us-central1") client = aiplatform.gapic.PredictionServiceClient() endpoint = client.endpoint_path( project="my-project", location="us-central1", endpoint="publishers/google/models/veo-3.1-generate-001", ) instance = json_format.ParseDict({ "prompt": "Cinematic drone shot over rice terraces in Bali at sunrise", "duration_seconds": 8, "resolution": "1080p", "aspect_ratio": "16:9", "generate_audio": True, }, Value()) # Long-running operation operation = client.predict_long_running( endpoint=endpoint, instances=[instance], ) result = operation.result(timeout=600) video_bytes = result.predictions[0]["video_bytes"]Veo 3.1 via Vertex AI. Uses long-running operations pattern.

Runway Gen-4.5 API

import requests import time RUNWAY_KEY = "rwa_" headers = {"Authorization": f"Bearer {RUNWAY_KEY}", "Content-Type": "application/json"} # Image-to-video (most consistent quality path) resp = requests.post( "https://api.dev.runwayml.com/v1/image_to_video", headers=headers, json={ "promptImage": "https://cdn.example.com/hero-frame.png", "promptText": "Camera slowly pulls back to reveal the full landscape", "model": "gen4_turbo", "duration": 10, "ratio": "1280:720", }, ) task_id = resp.json()["id"] while True: status = requests.get( f"https://api.dev.runwayml.com/v1/tasks/{task_id}", headers=headers ).json() if status["status"] == "SUCCEEDED": video_url = status["output"][0] break time.sleep(5)Runway Gen-4 image-to-video — the most reliable quality path.

Production pipeline pattern

1. Accept a job from a user. Store in a queue (Redis, SQS).
2. Worker picks up the job, calls the video API, stores the task_id.
3. Webhook handler or polling worker updates job status.
4. On success, download video to object storage (S3/R2).
5. Notify user (email, websocket, push).
6. Periodic cleanup — providers keep result URLs for hours/days, not forever.

Provider ergonomics compared

Provider abstraction layer

from abc import ABC, abstractmethod from dataclasses import dataclass @dataclass class VideoJob: prompt: str duration: int resolution: str ref_image_url: str | None = None class VideoProvider(ABC): @abstractmethod async def submit(self, job: VideoJob) -> str: # returns task_id @abstractmethod async def status(self, task_id: str) -> dict: @abstractmethod async def download(self, task_id: str) -> bytes: class SoraProvider(VideoProvider): class VeoProvider(VideoProvider): class RunwayProvider(VideoProvider): class KlingProvider(VideoProvider): def pick_provider(job: VideoJob, policy: str) -> VideoProvider: if policy == "cheapest_4k": return KlingProvider() if policy == "best_physics": return VeoProvider() if policy == "best_character_consistency": return RunwayProvider() return VeoProvider() # sensible defaultAbstract behind a provider interface. Video models consolidate fast.

Quality-control pipeline

1. Generate 3-5 candidates per shot in parallel.
2. Run each through a lightweight classifier (CLIP similarity to prompt; motion analysis; face-detection stability).
3. Pick top candidate automatically; queue lower candidates for human review.
4. On human rejection, feed feedback back as a refined prompt (closed-loop).