Wie man KI-Videogenerierung zu einer SaaS-App hinzufügt

Das Hinzufügen von Videogenerierung zu Ihrer App ist nicht dasselbe wie das Hinzufügen von Bildgenerierung. Der API-Aufruf gibt sofort eine Antwort zurück — aber das Video ist noch nicht fertig. Sie erhalten eine Task-ID und müssen so lange immer wieder fragen „Ist es fertig?“, bis es abgeschlossen ist.

Die meisten Entwickler stolpern darüber, wenn sie zum ersten Mal eine Video-API aufrufen, auf einen Response-Body mit einer Video-URL warten und stattdessen eine Task-ID erhalten. Dieser Leitfaden zeigt den kompletten Ablauf: eine Aufgabe einreichen, auf Ergebnisse pollen, Fehler behandeln und die Ausgabe speichern, bevor die URL abläuft.

Was Sie bauen werden

Einen Backend-Service, der einen Textprompt oder ein Bild annimmt, eine Videogenerierungs-Task einreicht, bis zur Fertigstellung pollt und die endgültige Video-URL zurückgibt. Sie arbeiten mit vier Modellen — Veo 3 Fast, Sora 2, Kling Video und Runway — alle über einen einzigen API-Schlüssel.

Voraussetzungen:

• Python 3.8+ oder Node.js 18+
• Ein CometAPI-Schlüssel
• Grundkenntnisse zu REST-APIs

Verstehen, warum die Videogenerierung anders ist

Bei der Bildgenerierung senden Sie eine Anfrage und erhalten das Bild in derselben Antwort zurück. Die Videogenerierung nutzt eine asynchrone Task-Queue:

1. Einreichung einer Generierungsanfrage → Rückgabe einer task_id
2. Alle paar Sekunden einen Status-Endpunkt pollen
3. Sobald der Status einen Endzustand erreicht, erhalten Sie die Video-URL
4. Das Video herunterladen und speichern — die URL ist temporär

Wenn Sie die Videogenerierung wie die Bildgenerierung behandeln und darauf warten, dass die erste Antwort Ihr Video enthält, wird Ihre Anfrage jedes Mal in ein Timeout laufen.

In einem produktiven Webservice sollte diese Polling-Schleife in einem Background-Worker (Celery, Bull oder ähnlich) laufen, nicht im Request-Handler. Die folgenden Beispiele verwenden synchrones Polling — gut für Skripte und Prototypen, aber nicht für die Bedienung gleichzeitiger Nutzer.

Ein Modell auswählen

Quelle**: CometAPI-Modellseiten, Mai 2026. Hinweis: „Sora 2“ ist CometAPIs Modell-Bezeichner — siehe deren Modellseite für Details zum zugrunde liegenden Modell.

• Veo 3 Fast unterstützt sowohl Text-to-Video als auch Image-to-Video. Pro Sekunde am günstigsten, ein guter Startpunkt.
• Sora 2 erzeugt Audio nativ zusammen mit dem Video — Dialoge, Umgebungsgeräusche und Effekte ohne separaten TTS-Schritt.
• Kling Video bietet negative_prompt, cfg_scale, Einstellungen für Kamerabewegung und einen pro‑Modus. Am meisten Kontrolle von den vieren.
• Runway ist über CometAPI nur Image-to-Video. Geben Sie ein statisches Bild und eine Bewegungsbeschreibung vor, und es animiert dieses.

Eine Veo-Task einreichen

Veo verwendet multipart/form-data. Nutzen Sie in Python requests files=, um es korrekt zu senden — data=dict sendet application/x-www-form-urlencoded, was nicht dasselbe ist:

import requestsimport osfrom dotenv import load_dotenv​load_dotenv()​def submit_veo_task(prompt: str, size: str = "16x9") -> str:    """Submit a Veo 3 Fast text-to-video task. Returns task_id."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY environment variable is not set")​    response = requests.post(        "https://api.cometapi.com/v1/videos",        headers={"Authorization": f"Bearer {api_key}"},        files={            "prompt": (None, prompt),            "model": (None, "veo3-fast"),            "size": (None, size)        },        timeout=30    )    response.raise_for_status()    return response.json()["id"]​​task_id = submit_veo_task("A paper kite drifting above a wheat field on a windy afternoon")print(f"Task submitted: {task_id}")

Nach dem Ergebnis pollen

import time​def poll_veo_task(task_id: str, interval: int = 10, max_wait: int = 600) -> str:    """Poll until Veo task completes. Returns video URL."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY environment variable is not set")​    headers = {"Authorization": f"Bearer {api_key}"}    url = f"https://api.cometapi.com/v1/videos/{task_id}"    elapsed = 0​    while elapsed < max_wait:        response = requests.get(url, headers=headers, timeout=30)        response.raise_for_status()        result = response.json()        status = result.get("status")​        if status == "succeeded":            return result["output"][0]        elif status in ("failed", "cancelled"):            raise RuntimeError(                f"Task {task_id} failed with status '{status}': "                f"{result.get('error', 'no error detail returned')}"            )​        time.sleep(interval)        elapsed += interval​    raise TimeoutError(f"Task {task_id} did not complete within {max_wait} seconds")​​video_url = poll_veo_task(task_id)print(f"Video ready: {video_url}")

Kling Video für mehr Kontrolle verwenden

Kling hat eine andere Endpunktstruktur und nutzt JSON. Beachten Sie, dass Kling als terminalen Status-String „succeed“ (nicht „succeeded“) zurückgibt — das entspricht dem tatsächlichen Response-Format der API:

def submit_kling_task(prompt: str, duration: str = "5", mode: str = "std") -> str:    """Submit a Kling text-to-video task. Returns task_id."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY environment variable is not set")​    response = requests.post(        "https://api.cometapi.com/kling/v1/videos/text2video",        headers={            "Authorization": f"Bearer {api_key}",            "Content-Type": "application/json"        },        json={            "model_name": "kling-v1-6",            "prompt": prompt,            "negative_prompt": "blurry, low quality, watermark",            "cfg_scale": 0.5,            "mode": mode,         # "std" or "pro"            "aspect_ratio": "16:9",            "duration": duration  # "5" or "10"        },        timeout=30    )    response.raise_for_status()    return response.json()["data"]["task_id"]​​def poll_kling_task(task_id: str, interval: int = 10, max_wait: int = 600) -> str:    """Poll Kling task until complete. Returns video URL."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY environment variable is not set")​    headers = {"Authorization": f"Bearer {api_key}"}    url = f"https://api.cometapi.com/kling/v1/videos/text2video/{task_id}"    elapsed = 0​    while elapsed < max_wait:        response = requests.get(url, headers=headers, timeout=30)        response.raise_for_status()        result = response.json()        status = result["data"]["task_status"]​        if status == "succeed":  # Kling uses "succeed", not "succeeded"            return result["data"]["task_result"]["videos"][0]["url"]        elif status == "failed":            error_detail = result.get("data", {}).get("task_result", "no detail")            raise RuntimeError(                f"Kling task {task_id} failed: {error_detail}"            )​        time.sleep(interval)        elapsed += interval​    raise TimeoutError(f"Kling task {task_id} timed out after {max_wait}s")

Quelle**: CometAPI Kling Video Dokumentation

Ein statisches Bild mit Runway animieren

Runway ist nur Image-to-Video. Außerdem ist ein zusätzlicher Header (X-Runway-Version) erforderlich:

def submit_runway_task(image_url: str, motion_prompt: str, duration: int = 5) -> str:    """Submit a Runway image-to-video task. Returns task_id."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY environment variable is not set")​    response = requests.post(        "https://api.cometapi.com/runwayml/v1/image_to_video",        headers={            "Authorization": f"Bearer {api_key}",            "X-Runway-Version": "2024-11-06",            "Content-Type": "application/json"        },        json={            "model": "gen3a_turbo",            "promptImage": image_url,  # must be a stable HTTPS URL            "promptText": motion_prompt,            "duration": duration,            "ratio": "1280:720",            "watermark": False        },        timeout=30    )    response.raise_for_status()    return response.json()["id"]​​def poll_runway_task(task_id: str, interval: int = 5, max_wait: int = 600) -> str:    """Poll Runway task. Returns video URL when done."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("COMETAPI_KEY environment variable is not set")​    headers = {        "Authorization": f"Bearer {api_key}",        "X-Runway-Version": "2024-11-06"    }    url = f"https://api.cometapi.com/runwayml/v1/tasks/{task_id}"    elapsed = 0​    while elapsed < max_wait:        response = requests.get(url, headers=headers, timeout=30)        response.raise_for_status()        result = response.json()        status = result.get("status")​        if status == "task_not_exist":            # CometAPI-specific: task is still initializing, retry after a few seconds            time.sleep(interval)            elapsed += interval            continue        elif status == "succeeded":            return result["output"][0]        elif status in ("failed", "cancelled"):            raise RuntimeError(f"Runway task {task_id} failed: {result.get('error', 'no detail')}")​        time.sleep(interval)        elapsed += interval​    raise TimeoutError(f"Runway task {task_id} timed out after {max_wait}s")

Quelle**: CometAPI Runway Dokumentation

Speichern Sie das Video, bevor die URL abläuft

Video-URLs aus Generierungs-APIs sind temporär. Laden Sie die Datei sofort herunter und speichern Sie sie an einem Ort, den Sie kontrollieren:

import requestsimport pathlib​def download_video(url: str, output_path: str) -> None:    """Download video from URL to local file using streaming."""    out = pathlib.Path(output_path)    if out.parent != pathlib.Path("."):        out.parent.mkdir(parents=True, exist_ok=True)​    with requests.get(url, stream=True, timeout=60) as r:        r.raise_for_status()        with open(out, "wb") as f:            for chunk in r.iter_content(chunk_size=8192):                f.write(chunk)    print(f"Saved to {output_path}")​​# Full flowtask_id = submit_veo_task("A timelapse of clouds moving over a city skyline")video_url = poll_veo_task(task_id)download_video(video_url, "output/city_timelapse.mp4")

In der Produktion ersetzen Sie den lokalen Dateischreibvorgang durch einen Upload zu S3, Cloudflare R2 oder dem Speicher Ihrer Wahl. Das Streaming-Muster bleibt gleich — leiten Sie die Bytes direkt durch, statt das gesamte Video in den Speicher zu laden.

Fehler behandeln

Quelle: CometAPI Video-Generierungsdokumentation

Node.js-Version

Node.js 18+ enthält fetch und FormData nativ. Dieses Beispiel deckt alle vier Modelle ab:

// Node.js 18+ — no extra packages needed​const API_KEY = process.env.COMETAPI_KEY;if (!API_KEY) throw new Error('COMETAPI_KEY is not set');​// --- Veo 3 Fast ---async function submitVeoTask(prompt, size = '16x9') {  const form = new FormData();  form.append('prompt', prompt);  form.append('model', 'veo3-fast');  form.append('size', size);​  const res = await fetch('https://api.cometapi.com/v1/videos', {    method: 'POST',    headers: { 'Authorization': `Bearer ${API_KEY}` },    body: form  });  if (!res.ok) throw new Error(`Veo submit failed: ${res.status}`);  return (await res.json()).id;}​async function pollVeoTask(taskId, intervalMs = 10000, maxWaitMs = 600000) {  let elapsed = 0;  while (elapsed < maxWaitMs) {    const res = await fetch(`https://api.cometapi.com/v1/videos/${taskId}`, {      headers: { 'Authorization': `Bearer ${API_KEY}` }    });    if (!res.ok) throw new Error(`Poll failed: ${res.status}`);    const result = await res.json();​    if (result.status === 'succeeded') return result.output[0];    if (['failed', 'cancelled'].includes(result.status)) {      throw new Error(`Task ${taskId} failed: ${result.error ?? 'no detail'}`);    }    await new Promise(r => setTimeout(r, intervalMs));    elapsed += intervalMs;  }  throw new Error(`Task ${taskId} timed out`);}​// --- Kling Video ---async function submitKlingTask(prompt, duration = '5', mode = 'std') {  const res = await fetch('https://api.cometapi.com/kling/v1/videos/text2video', {    method: 'POST',    headers: {      'Authorization': `Bearer ${API_KEY}`,      'Content-Type': 'application/json'    },    body: JSON.stringify({      model_name: 'kling-v1-6',      prompt,      negative_prompt: 'blurry, low quality, watermark',      cfg_scale: 0.5,      mode,      aspect_ratio: '16:9',      duration    })  });  if (!res.ok) throw new Error(`Kling submit failed: ${res.status}`);  return (await res.json()).data.task_id;}​async function pollKlingTask(taskId, intervalMs = 10000, maxWaitMs = 600000) {  let elapsed = 0;  while (elapsed < maxWaitMs) {    const res = await fetch(      `https://api.cometapi.com/kling/v1/videos/text2video/${taskId}`,      { headers: { 'Authorization': `Bearer ${API_KEY}` } }    );    if (!res.ok) throw new Error(`Kling poll failed: ${res.status}`);    const result = await res.json();    const status = result.data.task_status;​    if (status === 'succeed') return result.data.task_result.videos[0].url;    if (status === 'failed') {      throw new Error(`Kling task ${taskId} failed: ${JSON.stringify(result.data.task_result ?? 'no detail')}`);    }    await new Promise(r => setTimeout(r, intervalMs));    elapsed += intervalMs;  }  throw new Error(`Kling task ${taskId} timed out`);}​// --- Runway (image-to-video) ---async function submitRunwayTask(imageUrl, motionPrompt, duration = 5) {  const res = await fetch('https://api.cometapi.com/runwayml/v1/image_to_video', {    method: 'POST',    headers: {      'Authorization': `Bearer ${API_KEY}`,      'X-Runway-Version': '2024-11-06',      'Content-Type': 'application/json'    },    body: JSON.stringify({      model: 'gen3a_turbo',      promptImage: imageUrl,      promptText: motionPrompt,      duration,      ratio: '1280:720',      watermark: false    })  });  if (!res.ok) throw new Error(`Runway submit failed: ${res.status}`);  return (await res.json()).id;}​async function pollRunwayTask(taskId, intervalMs = 5000, maxWaitMs = 600000) {  let elapsed = 0;  while (elapsed < maxWaitMs) {    const res = await fetch(      `https://api.cometapi.com/runwayml/v1/tasks/${taskId}`,      { headers: { 'Authorization': `Bearer ${API_KEY}`, 'X-Runway-Version': '2024-11-06' } }    );    if (!res.ok) throw new Error(`Runway poll failed: ${res.status}`);    const result = await res.json();    const status = result.status;​    if (status === 'task_not_exist') {      // CometAPI-specific: task still initializing      await new Promise(r => setTimeout(r, intervalMs));      elapsed += intervalMs;      continue;    }    if (status === 'succeeded') return result.output[0];    if (['failed', 'cancelled'].includes(status)) {      throw new Error(`Runway task ${taskId} failed: ${result.error ?? 'no detail'}`);    }    await new Promise(r => setTimeout(r, intervalMs));    elapsed += intervalMs;  }  throw new Error(`Runway task ${taskId} timed out`);}​// Usage exampleconst taskId = await submitVeoTask('A paper kite drifting above a wheat field');const videoUrl = await pollVeoTask(taskId);console.log('Video ready:', videoUrl);

Wie geht es weiter

Sie haben nun funktionierenden Code für vier Videomodelle, eine Polling-Schleife, die Fehler behandelt, und einen Download-Schritt, der verhindert, dass Sie generierte Inhalte verlieren.

Das nächste Problem, auf das die meisten Entwickler stoßen: Sie haben ein Modell hart verdrahtet, und ein Wechsel zu einer günstigeren oder schnelleren Option erfordert Änderungen an mehreren Dateien. Der nächste Artikel erklärt, wie Sie Anfragen über Modelle routen, ohne Ihren Code neu zu schreiben.

Als Nächstes: Wie Sie zwischen KI‑Modellen wechseln, ohne Ihren Code umzuschreiben

FAQ

F: Warum erhalte ich in der API-Antwort eine Task-ID statt eines Videos?

Die Videogenerierung ist asynchron — Modelle wie Veo, Sora, Kling und Runway rendern 2–5 Minuten. Die API gibt sofort eine Task-ID zurück, damit Ihre Anfrage nicht in ein Timeout läuft. Sie pollen einen separaten Status-Endpunkt, bis die Aufgabe einen Endzustand (succeeded, succeed, failed) erreicht.

F: Wie lange bleibt eine generierte Video-URL gültig?

Video-URLs aus Generierungs-APIs sind temporär. Laden Sie die Datei unmittelbar nach Erhalt der URL herunter und speichern Sie sie in Ihrem eigenen Speicher (S3, Cloudflare R2 usw.). Bewahren Sie nicht nur die URL auf und erwarten Sie, dass sie Stunden später noch funktioniert.

F: Was ist der Unterschied zwischen Veo 3 Fast und Kling Video?

Veo 3 Fast ist günstiger ($0.05/sec), schneller und einfacher aufzurufen. Kling Video gibt Ihnen mehr Kontrolle: negative_prompt, cfg_scale, Einstellungen für Kamerabewegungen und einen pro‑Qualitätsmodus. Wenn Sie die Ausgabe fein steuern müssen, verwenden Sie Kling. Wenn Sie Geschwindigkeit und niedrige Kosten benötigen, verwenden Sie Veo 3 Fast.

F: Kann ich ein Video aus einem Bild statt aus einem Textprompt erzeugen?

Ja. Veo unterstützt Image-to-Video durch Übergabe einer input_reference‑Datei. Kling unterstützt es über den Endpunkt /kling/v1/videos/image2video mit einem image‑Parameter (URL oder Base64). Runway ist ausschließlich Image-to-Video — es akzeptiert über CometAPI keine reinen Textprompts.

F: Warum liefert Runway beim ersten Polling task_not_exist?

Dies ist dokumentiertes Verhalten bei CometAPI — die Aufgabe wird im Backend noch initialisiert. Warten Sie ein paar Sekunden und versuchen Sie es erneut. Es ist kein Fehler. Der oben stehende Polling-Code behandelt dies automatisch.

F: Warum verwendet Kling "succeed" statt "succeeded"?

Das ist das tatsächliche Response-Format der Kling-API. Es ist kein Tippfehler. Veo und Runway verwenden "succeeded" — Kling verwendet "succeed". Wenn Sie einen einheitlichen Polling-Wrapper bauen, müssen Sie beide Zeichenketten berücksichtigen.

F: Ist die synchrone Polling-Schleife in einem Webserver sicher?

Nein. Die Polling-Schleife in diesem Leitfaden blockiert den Thread mehrere Minuten lang. In einem echten Webservice mit gleichzeitigen Nutzern führen Sie das Polling in einem Background-Worker aus (Celery für Python, Bull für Node.js). Reichen Sie die Aufgabe im Request-Handler ein, geben Sie die Task-ID an den Client zurück, und lassen Sie den Worker den Client benachrichtigen, sobald das Video bereit ist.