Cómo añadir generación de video con IA a una aplicación SaaS

Agregar generación de video a tu app no es lo mismo que agregar generación de imágenes. La llamada a la API devuelve una respuesta de inmediato, pero el video aún no está listo. Recibes un ID de tarea y tienes que seguir preguntando “¿ya terminó?” hasta que finalice.

La mayoría de los desarrolladores se topa con esto la primera vez que llama a una API de video: esperan un cuerpo de respuesta con una URL de video y, en su lugar, reciben un ID de tarea. Esta guía recorre el flujo completo: envío de una tarea, sondeo de resultados, manejo de fallos y almacenamiento de la salida antes de que la URL caduque.

Qué vas a construir

Un servicio backend que acepta un prompt de texto o una imagen, envía una tarea de generación de video, la sondea hasta que se complete y devuelve la URL final del video. Trabajarás con cuatro modelos — Veo 3 Fast, Sora 2, Kling Video y Runway — todos con una sola clave de API.

Requisitos previos:

• Python 3.8+ o Node.js 18+
• Una clave de CometAPI
• Familiaridad básica con APIs REST

Entiende por qué la generación de video es diferente

Con la generación de imágenes, envías una solicitud y recibes la imagen en la misma respuesta. La generación de video usa una cola de tareas asíncronas:

1. Enviar una solicitud de generación → recibes un task_id
2. Sondear un endpoint de estado cada pocos segundos
3. Cuando el estado llega a uno terminal, obtienes la URL del video
4. Descargar y almacenar el video — la URL es temporal

Si tratas la generación de video como la de imágenes y esperas que la primera respuesta contenga tu video, tu solicitud agotará el tiempo de espera siempre.

En un servicio web de producción, este bucle de sondeo debe ejecutarse en un worker en segundo plano (Celery, Bull o similar), no en tu controlador de solicitudes. Los ejemplos a continuación usan sondeo sincrónico — válido para scripts y prototipos, pero no para manejar usuarios concurrentes.

Elige un modelo

Source**: Páginas de modelos de CometAPI, mayo de 2026. Nota: "Sora 2" es el identificador de modelo de CometAPI — consulta su página del modelo para los detalles del modelo subyacente.

• Veo 3 Fast admite text-to-video e image-to-video. El más barato por segundo, buen punto de partida.
• Sora 2 genera audio de forma nativa junto al video — diálogo, ambiente y efectos sin un paso TTS por separado.
• Kling Video te da negative_prompt, cfg_scale, ajustes de movimiento de cámara y un modo pro. El mayor control de los cuatro.
• Runway es solo image-to-video vía CometAPI. Dale una imagen estática y una descripción de movimiento, y la anima.

Envía una tarea de Veo

Veo usa multipart/form-data. Usa files= en requests de Python para enviarlo correctamente — data=dict envía application/x-www-form-urlencoded, que no es lo mismo:

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

Sondea el resultado

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

Usa Kling Video para más control

Kling tiene una estructura de endpoints diferente y usa JSON. Ten en cuenta que el estado terminal de Kling es "succeed" (no "succeeded") — esto coincide con el formato real de respuesta de la 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

Anima una imagen estática con Runway

Runway es solo image-to-video. También requiere una cabecera extra (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

Guarda el video antes de que la URL caduque

Las URLs de video de las APIs de generación son temporales. Descarga el archivo de inmediato y guárdalo en un lugar que controles:

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

En producción, sustituye la escritura a archivo local por una subida a S3, Cloudflare R2 o el almacenamiento de tu elección. El patrón de streaming sigue siendo el mismo: canaliza los bytes directamente en lugar de cargar el video completo en memoria.

Maneja los fallos

Source: CometAPI video generation docs

Versión Node.js

Node.js 18+ incluye fetch y FormData de forma nativa. Este ejemplo cubre los cuatro modelos:

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

Qué sigue

Ahora tienes código funcional para cuatro modelos de video, un bucle de sondeo que maneja fallos y un paso de descarga que evita perder contenido generado.

El siguiente problema que la mayoría encuentra: han fijado un solo modelo, y cambiar a una opción más barata o rápida implica tocar varios archivos. El próximo artículo cubre cómo enrutar solicitudes entre modelos sin reescribir tu código.

Siguiente: Cómo cambiar entre modelos de IA sin reescribir tu código

Preguntas frecuentes

P: ¿Por qué recibo un ID de tarea en lugar de un video en la respuesta de la API?

La generación de video es asíncrona — modelos como Veo, Sora, Kling y Runway tardan 2–5 minutos en renderizar. La API devuelve un ID de tarea inmediatamente para que tu solicitud no agote el tiempo de espera. Sondeas un endpoint de estado por separado hasta que la tarea llegue a un estado terminal (succeeded, succeed, failed).

P: ¿Cuánto tiempo permanece válida la URL de un video generado?

Las URLs de video de las APIs de generación son temporales. Descarga el archivo inmediatamente después de obtener la URL y guárdalo en tu propio almacenamiento (S3, Cloudflare R2, etc.). No guardes la URL esperando que funcione horas después.

P: ¿Cuál es la diferencia entre Veo 3 Fast y Kling Video?

Veo 3 Fast es más barato ($0.05/sec), más rápido y más simple de invocar. Kling Video te da más control: negative_prompt, cfg_scale, ajustes de movimiento de cámara y un modo de calidad pro. Si necesitas afinar el resultado, usa Kling. Si necesitas velocidad y bajo costo, usa Veo 3 Fast.

P: ¿Puedo generar video desde una imagen en lugar de un prompt de texto?

Sí. Veo admite image-to-video pasando un archivo input_reference. Kling lo admite vía el endpoint /kling/v1/videos/image2video con un parámetro image (URL o base64). Runway es solo image-to-video — no acepta prompts solo de texto vía CometAPI.

P: ¿Por qué Runway devuelve task_not_exist en el primer sondeo?

Es un comportamiento documentado de CometAPI — la tarea aún se está inicializando en el backend. Espera unos segundos y reintenta. No es un error. El código de sondeo anterior lo maneja automáticamente.

P: ¿Por qué Kling usa "succeed" en lugar de "succeeded"?

Es el formato real de respuesta de la API de Kling. No es un error tipográfico. Veo y Runway usan "succeeded" — Kling usa "succeed". Si estás construyendo un wrapper de sondeo unificado, tendrás que manejar ambas cadenas.

P: ¿Es seguro usar el bucle de sondeo sincrónico en un servidor web?

No. El bucle de sondeo de esta guía bloquea el hilo durante minutos. En un servicio web real con usuarios concurrentes, ejecuta el sondeo en un worker en segundo plano (Celery para Python, Bull para Node.js). Envía la tarea en el handler de la solicitud, devuelve el ID de tarea al cliente y deja que el worker notifique al cliente cuando el video esté listo.