So fügen Sie KI-Videogenerierung zu einer SaaS-App hinzu

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 wiederholt fragen „Ist es fertig?“, bis es soweit ist.

Die meisten Entwickler stoßen darauf, wenn sie zum ersten Mal eine Video-API aufrufen, auf eine Antwort mit einer Video-URL warten und stattdessen eine Task-ID zurückbekommen. Dieser Leitfaden führt durch den gesamten Ablauf: Aufgabe einreichen, Ergebnisse pollen, Fehler behandeln und die Ausgabe speichern, bevor die URL abläuft.

Was Sie erstellen werden

Ein Backend-Service, der eine Textaufforderung oder ein Bild entgegennimmt, eine Videogenerierungsaufgabe 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 in 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 verwendet eine asynchrone Aufgabenwarteschlange:

1. Einreichen einer Generierungsanfrage → Sie erhalten eine task_id
2. Pollen eines Status-Endpunkts alle paar Sekunden
3. Sobald der Status einen Terminalzustand erreicht, erhalten Sie die Video-URL
4. Herunterladen und Speichern des Videos — die URL ist temporär

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

In einem produktiven Webdienst sollte diese Polling-Schleife in einem Background-Worker (Celery, Bull oder ähnlich) und nicht in Ihrem Request-Handler laufen. Die folgenden Beispiele verwenden synchrones Polling — für Skripte und Prototypen in Ordnung, aber nicht zum Handling gleichzeitiger Benutzer.

Wählen Sie ein Modell

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-zu-Video als auch Bild-zu-Video. Am günstigsten pro Sekunde, guter Einstiegspunkt.
• Sora 2 generiert Audio nativ zusammen mit dem Video — Dialog, Umgebungsgeräusche und Effekte ohne separaten TTS-Schritt.
• Kling Video bietet negative_prompt, cfg_scale, Einstellungen für Kamerabewegungen und einen pro-Modus. Am meisten Kontrolle der vier.
• Runway ist über CometAPI nur Bild-zu-Video. Geben Sie ein statisches Bild und eine Bewegungsbeschreibung an, und es animiert dieses.

Eine Veo-Aufgabe einreichen

Veo verwendet multipart/form-data. Verwenden 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}")

Ergebnis abfragen

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 verwendet JSON. Beachten Sie, dass der Terminal-Statusstring bei Kling "succeed" (nicht "succeeded") lautet — das entspricht dem tatsächlichen API-Antwortformat:

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 ausschließlich Bild-zu-Video. Außerdem ist ein zusätzlicher Header erforderlich (X-Runway-Version):

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

Video speichern, bevor die URL abläuft

Video-URLs von Generierungs-APIs sind temporär. Laden Sie die Datei sofort herunter und speichern Sie sie an einem von Ihnen kontrollierten Ort:

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 Produktion ersetzen Sie den lokalen Schreibvorgang durch einen Upload zu S3, Cloudflare R2 oder einem Speicher Ihrer Wahl. Das Streaming-Muster bleibt gleich — leiten Sie die Bytes direkt weiter, anstatt 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 umfasst alle vier Modelle:

// 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 generierte Inhalte verloren gehen.

Das nächste Problem, auf das die meisten Entwickler stoßen: Sie haben ein Modell hart codiert, und das Umschalten auf eine günstigere oder schnellere Option bedeutet, mehrere Dateien anzupassen. Der nächste Artikel behandelt, wie Sie Anfragen über Modelle hinweg routen, ohne Ihren Code umzuschreiben.

Weiter: So wechseln Sie zwischen KI-Modellen, ohne Ihren Code umzuschreiben

FAQ

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

Videogenerierung ist asynchron — Modelle wie Veo, Sora, Kling und Runway benötigen 2–5 Minuten zum Rendern. Die API gibt sofort eine Task-ID zurück, damit Ihre Anfrage nicht abläuft. Sie pollen einen separaten Status-Endpunkt, bis die Aufgabe einen Terminalzustand (succeeded, succeed, failed) erreicht.

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

Video-URLs von 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 etc.). Speichern Sie nicht nur die URL 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/s), schneller und einfacher aufzurufen. Kling Video bietet mehr Kontrolle: negative_prompt, cfg_scale, Einstellungen für Kamerabewegung und einen pro-Qualitätsmodus. Wenn Sie die Ausgabe feinjustieren müssen, verwenden Sie Kling. Wenn Sie Geschwindigkeit und niedrige Kosten brauchen, verwenden Sie Veo 3 Fast.

F: Kann ich ein Video aus einem Bild statt aus einem Text-Prompt generieren?

Ja. Veo unterstützt Bild-zu-Video, indem eine input_reference-Datei übergeben wird. Kling unterstützt es über den Endpunkt /kling/v1/videos/image2video mit einem image-Parameter (URL oder Base64). Runway ist ausschließlich Bild-zu-Video — es akzeptiert über CometAPI keine reinen Textprompts.

F: Warum gibt Runway beim ersten Polling task_not_exist zurück?

Dies ist dokumentiertes CometAPI-Verhalten — die Aufgabe initialisiert noch im Backend. 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 Antwortformat 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 Strings unterstützen.

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

Nein. Die Polling-Schleife in diesem Leitfaden blockiert den Thread minutenlang. In einem echten Webservice mit gleichzeitigen Benutzern 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, wenn das Video bereit ist.