Добавление генерации видео в ваше приложение — это не то же самое, что добавление генерации изображений. Вызов API возвращается сразу — но видео ещё не готово. Вы получаете ID задачи и должны продолжать спрашивать «оно готово?» до тех пор, пока не будет готово.
Большинство разработчиков сталкиваются с этим впервые, когда вызывают видео API, ждут в теле ответа URL видео и вместо этого получают ID задачи. В этом руководстве разобран полный поток: отправка задачи, опрос результатов, обработка сбоев и сохранение результата до истечения срока действия URL.
Что вы построите
Сервис бэкенда, который принимает текстовый промпт или изображение, отправляет задачу генерации видео, опрашивает до завершения и возвращает финальный URL видео. Вы будете работать с четырьмя моделями — Veo 3 Fast, Sora 2, Kling Video и Runway — всё через один API-ключ.
Требования:
- Python 3.8+ или Node.js 18+
- Ключ CometAPI
- Базовые знания REST API
Поймите, чем генерация видео отличается
При генерации изображений вы отправляете запрос и получаете изображение в том же ответе. Генерация видео использует асинхронную очередь задач:
- Отправьте запрос на генерацию → получите
task_id - Опросите endpoint статуса каждые несколько секунд
- Когда статус достигнет конечного состояния, вы получите URL видео
- Скачайте и сохраните видео — URL временный
Если вы будете обращаться с генерацией видео как с изображениями и ждать, что первый ответ будет содержать ваше видео, ваш запрос каждый раз будет истекать по таймауту.
В продакшн-сервисе веб-уровня этот цикл опроса должен выполняться в фоновом воркере (Celery, Bull и т. п.), а не в обработчике запроса. Примеры ниже используют синхронный опрос — он подходит для скриптов и прототипов, но не для одновременной обработки множества пользователей.
Выберите модель
| Model | Provider | Max duration | Price (via CometAPI) | Best for |
|---|---|---|---|---|
| Veo 3 Fast | 8 sec | $0.05/sec | Быстрое прототипирование, соцклипы | |
| Sora 2 | OpenAI (via CometAPI model ID) | ~10 sec | $0.08/sec | Короткие креативные ролики высокого качества |
| Kling Video | Kuaishou | 10 sec | $0.13–$2.64/task | Маркетинговый контент, тонкая настройка |
| Runway Gen-3A Turbo | Runway | 5 or 10 sec | $0.32/task | Image-to-video, коммерческий контент |
Source*: страницы моделей CometAPI, май 2026. Примечание: "Sora 2" — это* идентификатор модели в CometAPI — смотрите их страницу модели для подробностей об используемой модели.
- Veo 3 Fast поддерживает и text-to-video, и image-to-video. Самая низкая цена за секунду, хороший стартовый вариант.
- Sora 2 нативно генерирует аудио вместе с видео — диалоги, фоновые звуки и эффекты без отдельного шага TTS.
- Kling Video предоставляет
negative_prompt,cfg_scale, настройки движения камеры и режимpro. Больше всего контроля из четырёх. - Runway через CometAPI — только image-to-video. Дайте статичное изображение и описание движения — он его анимирует.
Отправьте задачу для Veo
Veo использует multipart/form-data. Используйте files= в Python requests, чтобы отправить корректно — data=dict отправляет application/x-www-form-urlencoded, что не одно и то же:
import requestsimport osfrom dotenv import load_dotenvload_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}")
Опросите результат
import timedef 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 для большего контроля
У Kling другая структура endpoint-ов и используется JSON. Обратите внимание, что конечная строка статуса у Kling — "succeed" (не "succeeded") — это соответствует фактическому формату ответа 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*:* Документация CometAPI по Kling Video
Анимируйте статичное изображение с Runway
Runway — только image-to-video. Ему также нужен дополнительный заголовок (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*:* Документация CometAPI по Runway
Сохраните видео до истечения срока действия URL
URL видео от API генерации — временные. Скачайте файл сразу и сохраните там, где вы контролируете хранение:
import requestsimport pathlibdef 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")
В продакшне замените запись в локальный файл на загрузку в S3, Cloudflare R2 или ваше хранилище. Паттерн стриминга остаётся прежним — направляйте байты напрямую, а не загружайте весь файл в память.
Обрабатывайте сбои
| Симптом | Вероятная причина | Решение |
|---|---|---|
| Задача застряла в queued 10+ мин | Нагрузка на сервер или модель недоступна | Повторите с другой моделью |
| task_not_exist при первом опросе Runway | Задача всё ещё инициализируется | Подождите 5 сек и повторите — задокументировано в CometAPI |
| failed без сообщения об ошибке | Промпт сработал на контент-фильтр | Переформулируйте промпт |
| URL видео возвращает 403 | URL истёк до скачивания | Скачивайте сразу же после получения URL |
| Таймаут через 10 мин | Генерация заняла слишком много времени | Увеличьте max_wait или переключитесь на Veo 3 Fast |
| Kling возвращает "succeed", а не "succeeded" | API Kling использует нестандартную строку статуса | Это корректно — смотрите код опроса Kling выше |
Источник: Документация CometAPI по генерации видео
Версия для Node.js
Node.js 18+ включает fetch и FormData нативно. Этот пример охватывает все четыре модели:
// Node.js 18+ — no extra packages neededconst 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);
Что дальше
Теперь у вас есть рабочий код для четырёх видео-моделей, цикл опроса с обработкой сбоев и шаг скачивания, который не даст потерять сгенерированный контент.
Следующая типичная проблема: вы захардкодили одну модель, и переключение на более дешёвую или быструю требует править несколько файлов. В следующей статье рассказывается, как маршрутизировать запросы между моделями без переписывания кода.
Далее: Как переключаться между моделями ИИ, не переписывая код
FAQ
В: Почему я получаю ID задачи вместо видео в ответе API?
Генерация видео — асинхронная: моделям вроде Veo, Sora, Kling и Runway требуется 2–5 минут на рендеринг. API сразу возвращает ID задачи, чтобы ваш запрос не истёк по таймауту. Вы опрашиваете отдельный endpoint статуса, пока задача не достигнет конечного состояния (succeeded, succeed, failed).
В: Как долго действует URL сгенерированного видео?
URL видео у подобных API — временные. Сразу скачайте файл после получения URL и сохраните в своём хранилище (S3, Cloudflare R2 и т. п.). Не храните только URL и не рассчитывайте, что он будет работать спустя часы.
В: В чём разница между Veo 3 Fast и Kling Video?
Veo 3 Fast дешевле ($0.05/sec), быстрее и проще в использовании. Kling Video даёт больше контроля: negative_prompt, cfg_scale, настройки движения камеры и режим качества pro. Нужна тонкая настройка — используйте Kling. Нужны скорость и низкая стоимость — Veo 3 Fast.
В: Могу ли я генерировать видео из изображения вместо текстового промпта?
Да. Veo поддерживает image-to-video с передачей файла input_reference. Kling поддерживает это через endpoint /kling/v1/videos/image2video с параметром image (URL или base64). Runway — только image-to-video, он не принимает чисто текстовые промпты через CometAPI.
В: Почему Runway возвращает task_not_exist при первом опросе?
Это задокументированное поведение CometAPI — задача ещё инициализируется в бэкенде. Подождите несколько секунд и повторите. Это не ошибка. Код опроса выше обрабатывает это автоматически.
В: Почему Kling использует "succeed" вместо "succeeded"?
Это фактический формат ответа API Kling. Это не опечатка. Veo и Runway используют "succeeded", Kling — "succeed". Если вы строите единый обёрточный опрос, нужно поддерживать обе строки.
В: Безопасно ли использовать синхронный цикл опроса в веб-сервере?
Нет. Такой цикл блокирует поток на минуты. В реальном веб-сервисе с несколькими пользователями запускайте опрос в фоновом воркере (Celery для Python, Bull для Node.js). В обработчике запросов отправляйте задачу, возвращайте клиенту ID задачи, а воркер пусть уведомляет клиента, когда видео будет готово.