Jak dodać generowanie wideo z wykorzystaniem AI do aplikacji SaaS

Dodanie generowania wideo do Twojej aplikacji to nie to samo, co dodanie generowania obrazów. Wywołanie API zwraca odpowiedź natychmiast — ale wideo nie jest jeszcze gotowe. Otrzymujesz identyfikator zadania i musisz wciąż pytać „czy już gotowe?”, aż do zakończenia.

Większość deweloperów natrafia na to przy pierwszym wywołaniu API wideo: czeka na treść odpowiedzi z adresem URL do wideo, a w zamian dostaje identyfikator zadania. Ten przewodnik przeprowadzi Cię przez cały przepływ: wysyłanie zadania, odpytywanie o wynik, obsługę błędów oraz zapis wygenerowanego pliku zanim adres URL wygaśnie.

Co zbudujesz

Usługę backendową, która przyjmuje prompt tekstowy lub obraz, wysyła zadanie generowania wideo, odpytuje do momentu zakończenia i zwraca finalny adres URL wideo. Będziesz pracować z czterema modelami — Veo 3 Fast, Sora 2, Kling Video i Runway — wszystkie przez jeden klucz API.

Wymagania wstępne:

• Python 3.8+ lub Node.js 18+
• Klucz CometAPI
• Podstawowa znajomość REST API

Zrozum, dlaczego generowanie wideo jest inne

W przypadku generowania obrazów wysyłasz żądanie i otrzymujesz obraz w tej samej odpowiedzi. Generowanie wideo korzysta z asynchronicznej kolejki zadań:

1. Submit żądanie wygenerowania → otrzymujesz task_id
2. Poll endpoint statusu co kilka sekund
3. Gdy status osiągnie stan końcowy, otrzymujesz adres URL wideo
4. Download and store wideo — URL jest tymczasowy

Jeśli potraktujesz generowanie wideo jak generowanie obrazów i będziesz czekać, aż pierwsza odpowiedź zawiera wideo, Twoje żądanie za każdym razem wygaśnie.

W produkcyjnej usłudze webowej ta pętla odpytywania powinna działać w tle (Celery, Bull lub podobne), a nie w samym handlerze żądania. Przykłady poniżej używają synchronicznego odpytywania — w sam raz do skryptów i prototypów, ale nie do obsługi współbieżnych użytkowników.

Wybierz model

Source**: Strony modeli CometAPI, maj 2026. Uwaga: „Sora 2” to identyfikator modelu w CometAPI — zobacz ich stronę modelu po szczegóły dotyczące bazowego modelu.

• Veo 3 Fast obsługuje zarówno text-to-video, jak i image-to-video. Najtańszy w przeliczeniu na sekundę, dobry punkt startowy.
• Sora 2 generuje natywnie audio wraz z wideo — dialogi, dźwięki tła i efekty bez osobnego kroku TTS.
• Kling Video daje negative_prompt, cfg_scale, ustawienia ruchu kamery i tryb pro. Najwięcej kontroli z całej czwórki.
• Runway przez CometAPI działa tylko jako image-to-video. Podaj statyczny obraz i opis ruchu, a zaanimuje go.

Wyślij zadanie Veo

Veo używa multipart/form-data. Użyj files= w Python requests, aby wysłać poprawny format — data=dict wysyła application/x-www-form-urlencoded, co nie jest tym samym:

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}")

Odpytywanie o wynik

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}")

Użyj Kling Video dla większej kontroli

Kling ma inną strukturę endpointów i używa JSON. Zwróć uwagę, że końcowy status Kling to "succeed" (nie "succeeded") — to odpowiada faktycznemu formatowi odpowiedzi 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")

Source**: Dokumentacja Kling Video CometAPI

Animuj statyczny obraz w Runway

Runway działa tylko jako image-to-video. Wymaga też dodatkowego nagłówka (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")

Source**: Dokumentacja Runway CometAPI

Zapisz wideo, zanim URL wygaśnie

Adresy URL do wideo z API generujących są tymczasowe. Pobierz plik natychmiast i zapisz go w miejscu, które kontrolujesz:

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")

W produkcji zamień zapis do pliku lokalnego na upload do S3, Cloudflare R2 lub innego wybranego magazynu. Wzorzec strumieniowania pozostaje ten sam — przesyłaj bajty bezpośrednio, zamiast ładować całe wideo do pamięci.

Obsługa błędów

Źródło: Dokumentacja generowania wideo CometAPI

Wersja Node.js

Node.js 18+ zawiera natywnie fetch i FormData. Ten przykład obejmuje wszystkie cztery modele:

// 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);

Co dalej

Masz działający kod dla czterech modeli wideo, pętlę odpytywania obsługującą błędy oraz krok pobierania, który chroni przed utratą wygenerowanej treści.

Następny problem, na jaki trafia większość deweloperów: zakodowali na sztywno jeden model, a przejście na tańszą lub szybszą opcję wymaga zmian w wielu plikach. Następny artykuł omawia, jak kierować żądania między modelami bez przepisywania kodu.

Dalej: Jak przełączać się między modelami AI bez przepisywania kodu

FAQ

P: Dlaczego w odpowiedzi API dostaję identyfikator zadania zamiast wideo?

Generowanie wideo jest asynchroniczne — modele takie jak Veo, Sora, Kling i Runway renderują 2–5 minut. API zwraca identyfikator zadania natychmiast, aby Twoje żądanie nie wygasło. Odpytujesz osobny endpoint statusu, aż zadanie osiągnie stan końcowy (succeeded, succeed, failed).

P: Jak długo wygenerowany URL wideo jest ważny?

Adresy URL wideo z API generujących są tymczasowe. Pobierz plik natychmiast po otrzymaniu adresu i zapisz go we własnym magazynie (S3, Cloudflare R2 itp.). Nie zapisuj samego URL z oczekiwaniem, że zadziała za kilka godzin.

P: Jaka jest różnica między Veo 3 Fast a Kling Video?

Veo 3 Fast jest tańszy ($0.05/sec), szybszy i prostszy w wywołaniu. Kling Video daje więcej kontroli: negative_prompt, cfg_scale, ustawienia ruchu kamery i tryb jakości pro. Jeśli chcesz precyzyjnie kształtować wynik, użyj Kling. Jeśli zależy Ci na szybkości i niskim koszcie, wybierz Veo 3 Fast.

P: Czy mogę wygenerować wideo z obrazu zamiast z promptu tekstowego?

Tak. Veo obsługuje image-to-video przez przekazanie pliku input_reference. Kling obsługuje to przez endpoint /kling/v1/videos/image2video z parametrem image (URL lub base64). Runway jest tylko image-to-video — nie przyjmuje promptów tekstowych przez CometAPI.

P: Dlaczego Runway zwraca task_not_exist przy pierwszym sprawdzaniu?

To udokumentowane zachowanie CometAPI — zadanie wciąż się inicjalizuje w backendzie. Poczekaj kilka sekund i spróbuj ponownie. To nie jest błąd. Powyższy kod odpytywania obsługuje to automatycznie.

P: Dlaczego Kling używa "succeed" zamiast "succeeded"?

Taki jest rzeczywisty format odpowiedzi API Kling. To nie literówka. Veo i Runway używają "succeeded" — Kling używa "succeed". Jeśli tworzysz ujednolicony wrapper odpytywania, musisz obsłużyć oba ciągi.

P: Czy synchroniczna pętla odpytywania jest bezpieczna do użycia w serwerze WWW?

Nie. Pętla odpytywania w tym poradniku blokuje wątek na kilka minut. W prawdziwej usłudze uruchom odpytywanie w workerze w tle (Celery dla Pythona, Bull dla Node.js). Wyślij zadanie w handlerze żądania, zwróć klientowi identyfikator zadania i pozwól workerowi powiadomić klienta, gdy wideo będzie gotowe.