Dodanie generowania wideo do Twojej aplikacji nie jest tym samym co dodanie generowania obrazów. Wywołanie API zwraca odpowiedź od razu — ale wideo nie jest jeszcze gotowe. Otrzymujesz identyfikator zadania (task ID) i musisz w kółko pytać „czy już skończone?”, aż faktycznie będzie.
Większość deweloperów trafia na to przy pierwszym wywołaniu API wideo: czeka na odpowiedź z adresem URL wideo, a dostaje identyfikator zadania. Ten przewodnik przeprowadzi Cię przez pełny przepływ: przesłanie zadania, odpytywanie o wynik, obsługę błędów oraz zapis wygenerowanego pliku zanim adres URL wygaśnie.
Co zbudujesz
Usługę backend, która przyjmuje tekstowy prompt lub obraz, przesyła zadanie generowania wideo, odpyta aż do zakończenia i zwróci końcowy adres URL wideo. Pracujesz z czterema modelami — Veo 3 Fast, Sora 2, Kling Video i Runway — wszystkie przez jeden klucz API.
Prerequisites:
- Python 3.8+ lub Node.js 18+
- Klucz CometAPI
- Podstawowa znajomość REST API
Zrozum, dlaczego generowanie wideo jest inne
W przypadku obrazów wysyłasz żądanie i otrzymujesz obraz w tej samej odpowiedzi. Generowanie wideo działa asynchronicznie z kolejką zadań:
- Wyślij żądanie generowania → otrzymasz
task_id - Odpytuj endpoint statusu co kilka sekund
- Gdy status osiągnie stan końcowy, dostaniesz adres URL wideo
- Pobierz i zapisz wideo — adres URL jest tymczasowy
Jeśli potraktujesz generowanie wideo jak generowanie obrazów i będziesz czekać, aż pierwszy response zawiera wideo, Twoje żądanie będzie za każdym razem przekraczać limit czasu.
W produkcyjnej usłudze webowej ta pętla odpytywania powinna działać w tle (Celery, Bull lub podobne), a nie w handlerze żądania. Przykłady poniżej używają synchronicznego odpytywania — w porządku do skryptów i prototypów, ale nie do obsługi wielu równoległych użytkowników.
Wybierz model
| Model | Dostawca | Max duration | Cena (via CometAPI) | Najlepsze zastosowania |
|---|---|---|---|---|
| Veo 3 Fast | 8 sec | $0.05/sec | Szybkie prototypowanie, shorty | |
| Sora 2 | OpenAI (via CometAPI model ID) | ~10 sec | $0.08/sec | Wysokiej jakości kreatywne krótkie |
| Kling Video | Kuaishou | 10 sec | $0.13–$2.64/task | Marketing, granularna kontrola |
| Runway Gen-3A Turbo | Runway | 5 or 10 sec | $0.32/task | Image-to-video, treści komercyjne |
Źródło**: strony modeli CometAPI, maj 2026. Uwaga: „Sora 2” to identyfikator modelu w CometAPI — szczegóły bazowego modelu znajdziesz na ich stronie modelu.
- Veo 3 Fast obsługuje zarówno text-to-video, jak i image-to-video. Najtańszy za sekundę, dobry punkt startowy.
- Sora 2 generuje audio natywnie wraz z wideo — dialog, dźwięki tła i efekty bez osobnego kroku TTS.
- Kling Video daje
negative_prompt,cfg_scale, ustawienia ruchu kamery i trybpro. Najwięcej kontroli z całej czwórki. - Runway jest w CometAPI wyłącznie image-to-video. Podajesz statyczny obraz i opis ruchu, a on go animuje.
Wyślij zadanie do Veo
Veo używa multipart/form-data. Użyj files= w requests dla Pythona, aby wysłać poprawnie — data=dict wysyła application/x-www-form-urlencoded, co nie jest tym samym:
import requestsimport osfrom dotenv import load_dotenvload_dotenv()def submit_veo_task(prompt: str, size: str = "16x9") -> str: """Wyślij zadanie Veo 3 Fast (text-to-video). Zwraca task_id.""" api_key = os.getenv("COMETAPI_KEY") if not api_key: raise ValueError("Zmienna środowiskowa COMETAPI_KEY nie jest ustawiona") 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("Papierowy latawiec unoszący się nad polem pszenicy w wietrzne popołudnie")print(f"Zadanie wysłane: {task_id}")
Odpytuj o wynik
import timedef poll_veo_task(task_id: str, interval: int = 10, max_wait: int = 600) -> str: """Odpytuj aż zadanie Veo zakończy się. Zwraca URL wideo.""" api_key = os.getenv("COMETAPI_KEY") if not api_key: raise ValueError("Zmienna środowiskowa COMETAPI_KEY nie jest ustawiona") 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"Zadanie {task_id} zakończyło się statusem '{status}': " f"{result.get('error', 'brak szczegółów błędu')}" ) time.sleep(interval) elapsed += interval raise TimeoutError(f"Zadanie {task_id} nie zakończyło się w ciągu {max_wait} sekund")video_url = poll_veo_task(task_id)print(f"Wideo gotowe: {video_url}")
Użyj Kling Video dla większej kontroli
Kling ma inny układ endpointów i używa JSON. Zwróć uwagę, że końcowy status w Kling to "succeed" (nie "succeeded") — odpowiada to rzeczywistemu formatowi odpowiedzi API:
def submit_kling_task(prompt: str, duration: str = "5", mode: str = "std") -> str: """Wyślij zadanie Kling (text-to-video). Zwraca task_id.""" api_key = os.getenv("COMETAPI_KEY") if not api_key: raise ValueError("Zmienna środowiskowa COMETAPI_KEY nie jest ustawiona") 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" lub "pro" "aspect_ratio": "16:9", "duration": duration # "5" lub "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: """Odpytuj zadanie Kling aż do zakończenia. Zwraca URL wideo.""" api_key = os.getenv("COMETAPI_KEY") if not api_key: raise ValueError("Zmienna środowiskowa COMETAPI_KEY nie jest ustawiona") 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 używa "succeed", nie "succeeded" return result["data"]["task_result"]["videos"][0]["url"] elif status == "failed": error_detail = result.get("data", {}).get("task_result", "brak szczegółów") raise RuntimeError( f"Zadanie Kling {task_id} zakończyło się niepowodzeniem: {error_detail}" ) time.sleep(interval) elapsed += interval raise TimeoutError(f"Zadanie Kling {task_id} przekroczyło limit czasu po {max_wait}s")
Źródło**:** Dokumentacja CometAPI Kling Video
Animuj statyczny obraz z Runway
Runway obsługuje wyłącznie 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: """Wyślij zadanie Runway image-to-video. Zwraca task_id.""" api_key = os.getenv("COMETAPI_KEY") if not api_key: raise ValueError("Zmienna środowiskowa COMETAPI_KEY nie jest ustawiona") 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, # musi to być stabilny adres HTTPS "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: """Odpytuj zadanie Runway. Zwraca URL wideo po zakończeniu.""" api_key = os.getenv("COMETAPI_KEY") if not api_key: raise ValueError("Zmienna środowiskowa COMETAPI_KEY nie jest ustawiona") 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": # Specyficzne dla CometAPI: zadanie wciąż się inicjalizuje, spróbuj ponownie za kilka sekund time.sleep(interval) elapsed += interval continue elif status == "succeeded": return result["output"][0] elif status in ("failed", "cancelled"): raise RuntimeError(f"Zadanie Runway {task_id} zakończyło się niepowodzeniem: {result.get('error', 'brak szczegółów')}") time.sleep(interval) elapsed += interval raise TimeoutError(f"Zadanie Runway {task_id} przekroczyło limit czasu po {max_wait}s")
Źródło**:** Dokumentacja CometAPI Runway
Zapisz wideo zanim adres URL wygaśnie
Adresy URL wideo z API generowania są tymczasowe. Pobierz plik od razu i zapisz go w miejscu, którym zarządzasz:
import requestsimport pathlibdef download_video(url: str, output_path: str) -> None: """Pobierz wideo z URL do pliku lokalnego strumieniowo.""" 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"Zapisano do {output_path}")# Pełny przepływtask_id = submit_veo_task("Timelapse chmur przemieszczających się nad panoramą miasta")video_url = poll_veo_task(task_id)download_video(video_url, "output/city_timelapse.mp4")
W produkcji zamień zapis lokalny na upload do S3, Cloudflare R2 lub wybranego magazynu. Wzorzec strumieniowania pozostaje taki sam — przekazuj bajty bezpośrednio, zamiast ładować całe wideo do pamięci.
Obsługa błędów
| Objaw | Prawdopodobna przyczyna | Rozwiązanie |
|---|---|---|
| Zadanie utknęło w queued przez 10+ min | Obciążenie serwera lub model niedostępny | Spróbuj innego modelu |
| task_not_exist przy pierwszym pollu Runway | Zadanie wciąż się inicjalizuje | Poczekaj 5 s i ponów — udokumentowane zachowanie CometAPI |
| failed bez komunikatu błędu | Prompt wywołał filtr treści | Przeformułuj prompt |
| URL wideo zwraca 403 | Adres wygasł przed pobraniem | Pobierz od razu po otrzymaniu adresu URL |
| Timeout po 10 min | Generowanie trwało zbyt długo | Zwiększ max_wait lub przejdź na Veo 3 Fast |
| Kling zwraca „succeed”, nie „succeeded” | API Kling używa niestandardowego statusu | To poprawne — patrz kod odpytywania Kling powyżej |
Źródło: Dokumentacja CometAPI dla generowania wideo
Wersja Node.js
Node.js 18+ ma natywnie fetch i FormData. Ten przykład obejmuje wszystkie cztery modele:
// Node.js 18+ — brak potrzeby dodatkowych paczekconst API_KEY = process.env.COMETAPI_KEY;if (!API_KEY) throw new Error('COMETAPI_KEY nie jest ustawiony');// --- 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(`Nie udało się wysłać Veo: ${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(`Błąd przy odpytywaniu: ${res.status}`); const result = await res.json(); if (result.status === 'succeeded') return result.output[0]; if (['failed', 'cancelled'].includes(result.status)) { throw new Error(`Zadanie ${taskId} nie powiodło się: ${result.error ?? 'brak szczegółów'}`); } await new Promise(r => setTimeout(r, intervalMs)); elapsed += intervalMs; } throw new Error(`Zadanie ${taskId} przekroczyło limit czasu`);}// --- 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(`Nie udało się wysłać do Kling: ${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(`Błąd odpytywania Kling: ${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(`Zadanie Kling ${taskId} nie powiodło się: ${JSON.stringify(result.data.task_result ?? 'brak szczegółów')}`); } await new Promise(r => setTimeout(r, intervalMs)); elapsed += intervalMs; } throw new Error(`Zadanie Kling ${taskId} przekroczyło limit czasu`);}// --- 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(`Nie udało się wysłać do Runway: ${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(`Błąd odpytywania Runway: ${res.status}`); const result = await res.json(); const status = result.status; if (status === 'task_not_exist') { // Specyficzne dla CometAPI: zadanie wciąż się inicjalizuje 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(`Zadanie Runway ${taskId} nie powiodło się: ${result.error ?? 'brak szczegółów'}`); } await new Promise(r => setTimeout(r, intervalMs)); elapsed += intervalMs; } throw new Error(`Zadanie Runway ${taskId} przekroczyło limit czasu`);}// Przykład użyciaconst taskId = await submitVeoTask('Papierowy latawiec unoszący się nad polem pszenicy');const videoUrl = await pollVeoTask(taskId);console.log('Wideo gotowe:', videoUrl);
Co dalej
Masz działający kod dla czterech modeli wideo, pętlę odpytywania obsługującą błędy i krok pobierania, który zapobiega utracie wygenerowanych treści.
Kolejny problem, na który trafia większość deweloperów: zahardcodowali jeden model i przełączenie 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
Q: 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 od razu, aby Twoje żądanie nie przekroczyło limitu czasu. Odpytujesz osobny endpoint statusu aż zadanie osiągnie stan końcowy (succeeded, succeed, failed).
Q: Jak długo wygenerowany adres URL wideo pozostaje ważny?
Adresy URL wideo z API generowania są tymczasowe. Pobierz plik natychmiast po otrzymaniu adresu i zapisz we własnym magazynie (S3, Cloudflare R2 itd.). Nie zapisuj samego URL i nie zakładaj, że zadziała po kilku godzinach.
Q: Czym różni się Veo 3 Fast od Kling Video?
Veo 3 Fast jest tańszy ($0.05/sec), szybszy i prostszy w wywołaniu. Kling Video daje większą kontrolę: negative_prompt, cfg_scale, ustawienia ruchu kamery i tryb jakości pro. Jeśli potrzebujesz precyzyjnego strojenia, użyj Kling. Jeśli zależy Ci na szybkości i niskim koszcie, wybierz Veo 3 Fast.
Q: 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 wyłącznie image-to-video — nie akceptuje samych promptów tekstowych przez CometAPI.
Q: Dlaczego Runway zwraca task_not_exist przy pierwszym odpytywaniu?
To udokumentowane zachowanie CometAPI — zadanie wciąż się inicjalizuje w backendzie. Poczekaj kilka sekund i spróbuj ponownie. To nie jest błąd. Kod odpytywania powyżej obsługuje to automatycznie.
Q: Dlaczego Kling używa "succeed" zamiast "succeeded"?
To faktyczny format odpowiedzi API Kling. To nie literówka. Veo i Runway używają "succeeded" — Kling używa "succeed". Jeśli budujesz zunifikowany wrapper do odpytywania, musisz obsłużyć oba ciągi.
Q: Czy synchroniczna pętla odpytywania jest bezpieczna do użycia w serwerze webowym?
Nie. Pętla odpytywania w tym przewodniku blokuje wątek na kilka minut. W prawdziwej usłudze webowej z równoległymi użytkownikami uruchom odpytywanie w procesie pracownika w tle (Celery dla Pythona, Bull dla Node.js). Wyślij zadanie w handlerze żądania, zwróć klientowi identyfikator zadania, a pracownik niech powiadomi klienta, gdy wideo będzie gotowe.