Como adicionar geração de vídeo com IA a um aplicativo SaaS

Adicionar geração de vídeo ao seu app não é o mesmo que adicionar geração de imagem. A chamada à API retorna imediatamente — mas o vídeo ainda não está pronto. Você recebe um task ID, e precisa continuar perguntando “já terminou?” até terminar.

A maioria dos desenvolvedores esbarra nisso na primeira chamada a uma API de vídeo: espera um corpo de resposta com a URL do vídeo e recebe um task ID no lugar. Este guia percorre todo o fluxo: envio da tarefa, polling pelos resultados, tratamento de falhas e armazenamento da saída antes que a URL expire.

O que você vai construir

Um serviço de backend que aceita um prompt de texto ou imagem, envia uma tarefa de geração de vídeo, faz polling até a conclusão e retorna a URL final do vídeo. Você trabalhará com quatro modelos — Veo 3 Fast, Sora 2, Kling Video e Runway — todos com uma única chave de API.

Pré-requisitos:

• Python 3.8+ ou Node.js 18+
• Uma chave da CometAPI
• Familiaridade básica com REST APIs

Entenda por que a geração de vídeo é diferente

Com geração de imagem, você envia uma solicitação e recebe a imagem na mesma resposta. Geração de vídeo usa uma fila de tarefas assíncronas:

1. Enviar uma solicitação de geração → recebe um task_id
2. Fazer polling de um endpoint de status a cada poucos segundos
3. Quando o status atinge um estado terminal, você recebe a URL do vídeo
4. Baixar e armazenar o vídeo — a URL é temporária

Se você tratar geração de vídeo como geração de imagem e esperar que a primeira resposta contenha seu vídeo, sua solicitação vai expirar sempre.

Em um serviço web de produção, esse loop de polling deve rodar em um worker de segundo plano (Celery, Bull ou similar), não no seu request handler. Os exemplos abaixo usam polling síncrono — ótimo para scripts e protótipos, mas não para lidar com usuários concorrentes.

Escolha um modelo

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 suporta texto-para-vídeo e imagem-para-vídeo. Mais barato por segundo, bom ponto de partida.
• Sora 2 gera áudio nativamente junto com o vídeo — diálogos, sons ambientes e efeitos sem uma etapa TTS separada.
• Kling Video oferece negative_prompt, cfg_scale, configurações de movimento de câmera e um modo pro. Mais controle entre os quatro.
• Runway é apenas imagem-para-vídeo via CometAPI. Forneça uma imagem estática e uma descrição de movimento, e ele a anima.

Envie uma tarefa Veo

Veo usa multipart/form-data. Use files= no requests do Python para enviar corretamente — data=dict envia application/x-www-form-urlencoded, que não é a mesma coisa:

import requestsimport osfrom dotenv import load_dotenv​load_dotenv()​def submit_veo_task(prompt: str, size: str = "16x9") -> str:    """Enviar uma tarefa de texto para vídeo do Veo 3 Fast. Retorna task_id."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("A variável de ambiente COMETAPI_KEY não está definida")​    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"Tarefa enviada: {task_id}")

Faça polling pelo resultado

import time​def poll_veo_task(task_id: str, interval: int = 10, max_wait: int = 600) -> str:    """Faz polling até a tarefa Veo ser concluída. Retorna a URL do vídeo."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("A variável de ambiente COMETAPI_KEY não está definida")​    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"Tarefa {task_id} falhou com status '{status}': "                f"{result.get('error', 'nenhum detalhe de erro retornado')}"            )​        time.sleep(interval)        elapsed += interval​    raise TimeoutError(f"Tarefa {task_id} não foi concluída em {max_wait} segundos")​​video_url = poll_veo_task(task_id)print(f"Vídeo pronto: {video_url}")

Use o Kling Video para mais controle

Kling tem uma estrutura de endpoint diferente e usa JSON. Note que a string de status terminal do Kling é "succeed" (não "succeeded") — isso condiz com o formato real de resposta da API:

def submit_kling_task(prompt: str, duration: str = "5", mode: str = "std") -> str:    """Enviar uma tarefa Kling de texto para vídeo. Retorna task_id."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("A variável de ambiente COMETAPI_KEY não está definida")​    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" ou "pro"            "aspect_ratio": "16:9",            "duration": duration  # "5" ou "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:    """Fazer polling da tarefa Kling até concluir. Retorna a URL do vídeo."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("A variável de ambiente COMETAPI_KEY não está definida")​    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 usa "succeed", não "succeeded"            return result["data"]["task_result"]["videos"][0]["url"]        elif status == "failed":            error_detail = result.get("data", {}).get("task_result", "sem detalhes")            raise RuntimeError(                f"Tarefa Kling {task_id} falhou: {error_detail}"            )​        time.sleep(interval)        elapsed += interval​    raise TimeoutError(f"Tarefa Kling {task_id} expirou após {max_wait}s")

Source**: CometAPI Kling Video docs

Anime uma imagem estática com Runway

Runway é apenas imagem-para-vídeo. Também exige um cabeçalho extra (X-Runway-Version):

def submit_runway_task(image_url: str, motion_prompt: str, duration: int = 5) -> str:    """Enviar uma tarefa Runway de imagem para vídeo. Retorna task_id."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("A variável de ambiente COMETAPI_KEY não está definida")​    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,  # deve ser uma URL HTTPS estável            "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:    """Fazer polling da tarefa Runway. Retorna a URL do vídeo quando terminar."""    api_key = os.getenv("COMETAPI_KEY")    if not api_key:        raise ValueError("A variável de ambiente COMETAPI_KEY não está definida")​    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":            # Específico do CometAPI: a tarefa ainda está inicializando, tente novamente após alguns segundos            time.sleep(interval)            elapsed += interval            continue        elif status == "succeeded":            return result["output"][0]        elif status in ("failed", "cancelled"):            raise RuntimeError(f"Tarefa Runway {task_id} falhou: {result.get('error', 'sem detalhes')}")​        time.sleep(interval)        elapsed += interval​    raise TimeoutError(f"Tarefa Runway {task_id} expirou após {max_wait}s")

Source**: CometAPI Runway docs

Salve o vídeo antes que a URL expire

As URLs de vídeo das APIs de geração são temporárias. Baixe o arquivo imediatamente e armazene em um local sob seu controle:

import requestsimport pathlib​def download_video(url: str, output_path: str) -> None:    """Baixar o vídeo da URL para um arquivo local usando 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"Salvo em {output_path}")​​# Fluxo completotask_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")

Em produção, troque a gravação de arquivo local por um upload para S3, Cloudflare R2 ou o armazenamento de sua preferência. O padrão de streaming permanece o mesmo — direcione os bytes diretamente em vez de carregar o vídeo inteiro na memória.

Trate falhas

Source: CometAPI video generation docs

Versão para Node.js

Node.js 18+ inclui fetch e FormData nativamente. Este exemplo cobre os quatro modelos:

// Node.js 18+ — nenhum pacote extra necessário​const API_KEY = process.env.COMETAPI_KEY;if (!API_KEY) throw new Error('COMETAPI_KEY não está definida');​// --- 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(`Falha ao enviar 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(`Falha no polling: ${res.status}`);    const result = await res.json();​    if (result.status === 'succeeded') return result.output[0];    if (['failed', 'cancelled'].includes(result.status)) {      throw new Error(`Tarefa ${taskId} falhou: ${result.error ?? 'sem detalhes'}`);    }    await new Promise(r => setTimeout(r, intervalMs));    elapsed += intervalMs;  }  throw new Error(`Tarefa ${taskId} expirou`);}​// --- 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(`Falha ao enviar 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(`Falha no polling do 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(`Tarefa Kling ${taskId} falhou: ${JSON.stringify(result.data.task_result ?? 'sem detalhes')}`);    }    await new Promise(r => setTimeout(r, intervalMs));    elapsed += intervalMs;  }  throw new Error(`Tarefa Kling ${taskId} expirou`);}​// --- 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(`Falha ao enviar 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(`Falha no polling do Runway: ${res.status}`);    const result = await res.json();    const status = result.status;​    if (status === 'task_not_exist') {      // Específico do CometAPI: tarefa ainda inicializando      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(`Tarefa Runway ${taskId} falhou: ${result.error ?? 'sem detalhes'}`);    }    await new Promise(r => setTimeout(r, intervalMs));    elapsed += intervalMs;  }  throw new Error(`Tarefa Runway ${taskId} expirou`);}​// Exemplo de usoconst taskId = await submitVeoTask('A paper kite drifting above a wheat field');const videoUrl = await pollVeoTask(taskId);console.log('Vídeo pronto:', videoUrl);

O que vem a seguir

Agora você tem código funcional para quatro modelos de vídeo, um loop de polling que trata falhas e uma etapa de download que evita perder conteúdo gerado.

O próximo problema que a maioria dos desenvolvedores encontra: fixaram um único modelo, e mudar para uma opção mais barata ou mais rápida exige alterar vários arquivos. O próximo artigo cobre como rotear solicitações entre modelos sem reescrever seu código.

Próximo: How to Switch Between AI Models Without Rewriting Your Code

Perguntas frequentes

P: Por que recebo um task ID em vez de um vídeo na resposta da API?

Geração de vídeo é assíncrona — modelos como Veo, Sora, Kling e Runway levam de 2 a 5 minutos para renderizar. A API retorna um task ID imediatamente para que sua solicitação não expire. Você faz polling de um endpoint de status separado até a tarefa atingir um estado terminal (succeeded, succeed, failed).

P: Por quanto tempo uma URL de vídeo gerado permanece válida?

As URLs de vídeo de APIs de geração são temporárias. Baixe o arquivo imediatamente após obter a URL e armazene em seu próprio storage (S3, Cloudflare R2, etc.). Não armazene a URL esperando que funcione horas depois.

P: Qual a diferença entre Veo 3 Fast e Kling Video?

Veo 3 Fast é mais barato ($0.05/sec), mais rápido e mais simples de chamar. Kling Video oferece mais controle: negative_prompt, cfg_scale, configurações de movimento de câmera e um modo pro de qualidade. Se você precisa ajustar o resultado, use Kling. Se precisa de velocidade e baixo custo, use Veo 3 Fast.

P: Posso gerar vídeo a partir de uma imagem em vez de um prompt de texto?

Sim. Veo suporta imagem-para-vídeo passando um arquivo input_reference. Kling suporta via endpoint /kling/v1/videos/image2video com um parâmetro image (URL ou base64). Runway é apenas imagem-para-vídeo — não aceita prompts somente de texto via CometAPI.

P: Por que o Runway retorna task_not_exist na primeira sondagem?

Este é um comportamento documentado da CometAPI — a tarefa ainda está inicializando no backend. Aguarde alguns segundos e tente novamente. Não é um erro. O código de polling acima trata isso automaticamente.

P: Por que o Kling usa "succeed" em vez de "succeeded"?

Esse é o formato real de resposta da API do Kling. Não é erro de digitação. Veo e Runway usam "succeeded" — Kling usa "succeed". Se você estiver construindo um wrapper de polling unificado, precisará tratar ambas as strings.

P: O loop de polling síncrono é seguro para usar em um servidor web?

Não. O loop de polling neste guia bloqueia a thread por vários minutos. Em um serviço web real com usuários concorrentes, rode o polling em um worker de segundo plano (Celery para Python, Bull para Node.js). Envie a tarefa no request handler, retorne o task ID ao cliente e deixe o worker notificar o cliente quando o vídeo estiver pronto.