Как добавить генерацию видео с помощью ИИ в SaaS-приложение

Добавить генерацию видео в приложение — не то же самое, что добавить генерацию изображений. Вызов API возвращается сразу, но видео ещё не готово. Вы получаете идентификатор задачи и должны продолжать спрашивать «уже готово?», пока не будет готово.

Большинство разработчиков сталкиваются с этим, когда впервые вызывают видео‑API, ждут в теле ответа URL видео и вместо этого получают идентификатор задачи. Это руководство показывает полный поток: отправку задачи, опрос результата, обработку сбоев и сохранение результата до истечения срока действия URL.

Что вы создадите

Сервис бэкенда, который принимает текстовый промпт или изображение, отправляет задачу генерации видео, опрашивает до завершения и возвращает финальный URL видео. Вы поработаете с четырьмя моделями — Veo 3 Fast, Sora 2, Kling Video и Runway — все через один ключ API.

Предварительные требования:

• Python 3.8+ или Node.js 18+
• Ключ CometAPI
• Базовое знакомство с REST API

Почему генерация видео отличается

При генерации изображений вы отправляете запрос и получаете изображение в том же ответе. Генерация видео использует асинхронную очередь задач:

1. Отправьте запрос на генерацию → получите task_id
2. Опросите endpoint статуса каждые несколько секунд
3. Когда статус достигнет терминального состояния, вы получите URL видео
4. Скачайте и сохраните видео — URL временный

Если относиться к генерации видео как к генерации изображений и ждать, что в первом ответе будет ваше видео, запрос будет каждый раз истекать по тайм‑ауту.

В промышленном веб‑сервисе этот цикл опроса должен выполняться в фоновом воркере (Celery, Bull или аналог), а не в обработчике запроса. Примеры ниже используют синхронный опрос — это подходит для скриптов и прототипов, но не для обслуживания конкурирующих пользователей.

Выберите модель

Source**: CometAPI model pages, May 2026. Note: "Sora 2" is CometAPI's model identifier — refer to their model page for the underlying model details.

• 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. В Python requests используйте files= — data=dict отправляет application/x-www-form-urlencoded, это не одно и то же:

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

Опрос результата

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 для большего контроля

У 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 docs

Анимируйте статическое изображение с помощью 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 docs

Сохраните видео до истечения срока действия URL

URL видео из API генерации — временные. Скачивайте файл сразу и сохраняйте там, где вы контролируете хранение:

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

В продакшене замените запись в локальный файл на загрузку в S3, Cloudflare R2 или выбранное хранилище. Паттерн со стримингом остаётся тем же — передавайте байты напрямую, не загружая всё видео в память.

Обработка сбоев

Source: CometAPI video generation docs

Версия для Node.js

Node.js 18+ включает нативные fetch и FormData. Этот пример охватывает все четыре модели:

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

Что дальше

Теперь у вас есть рабочий код для четырёх видеомоделей, цикл опроса с обработкой сбоев и шаг скачивания, который предотвращает потерю сгенерированного контента.

Следующая проблема, с которой сталкиваются многие разработчики: они захардкодили одну модель, и переключение на более дешёвую или быструю требует править несколько файлов. В следующей статье — как маршрутизировать запросы между моделями, не переписывая код.

Далее: Как переключаться между AI‑моделями, не переписывая ваш код

Вопросы и ответы

В: Почему в ответе API я получаю идентификатор задачи вместо видео?

Генерация видео асинхронна — моделям вроде Veo, Sora, Kling и Runway требуется 2–5 минут на рендеринг. API возвращает идентификатор задачи сразу, чтобы ваш запрос не истёк. Вы опрашиваете отдельный endpoint статуса, пока задача не перейдёт в терминальное состояние (succeeded, succeed, failed).

В: Как долго действителен URL сгенерированного видео?

URL видео из API генерации — временные. Скачивайте файл сразу после получения URL и храните у себя (S3, Cloudflare R2 и т. п.). Не храните только URL и не рассчитывайте, что он будет работать через несколько часов.

В: В чём разница между Veo 3 Fast и Kling Video?

Veo 3 Fast дешевле ($0.05/сек), быстрее и проще в вызове. 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). Отправляйте задачу в обработчике запроса, возвращайте клиенту идентификатор задачи и пусть воркер уведомит клиента, когда видео будет готово.