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 de API retorna imediatamente — mas o vídeo ainda não está pronto. Você recebe um task ID e precisa continuar perguntando “já terminou?” até que termine.

A maioria dos desenvolvedores esbarra nisso na primeira vez que chama uma API de vídeo, espera um corpo de resposta com uma URL de vídeo e recebe um task ID em vez disso. Este guia percorre o fluxo completo: enviar uma tarefa, fazer polling pelos resultados, lidar com falhas e armazenar a saída antes que a URL expire.

O que você irá construir

Um serviço 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 APIs REST

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

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

1. Envie uma requisição de geração → receba um task_id
2. Faça polling em um endpoint de status a cada poucos segundos
3. Quando o status atingir um estado terminal, você recebe a URL do vídeo
4. Baixe e armazene o vídeo — a URL é temporária

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

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

Escolha um modelo

Source*: páginas de modelos da CometAPI, maio de 2026. Observação: "Sora 2" é o* identificador do modelo na CometAPI — consulte a página do modelo para detalhes sobre o modelo subjacente.

• O Veo 3 Fast suporta text-to-video e image-to-video. Mais barato por segundo, bom ponto de partida.
• O Sora 2 gera áudio nativamente junto com o vídeo — diálogo, som ambiente e efeitos sem uma etapa separada de TTS.
• O Kling Video oferece negative_prompt, cfg_scale, configurações de movimento de câmera e modo pro. Maior controle entre os quatro.
• O Runway é somente image-to-video via CometAPI. Forneça uma imagem estática e uma descrição de movimento, e ele a anima.

Envie uma tarefa do Veo

O 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("Um papagaio de papel pairando sobre um campo de trigo em uma tarde ventosa")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:    """Consultar até a tarefa do 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"A tarefa {task_id} falhou com o status '{status}': "                f"{result.get('error', 'nenhum detalhe de erro retornado')}"            )​        time.sleep(interval)        elapsed += interval​    raise TimeoutError(f"A 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

O Kling tem uma estrutura de endpoint diferente e usa JSON. Observe que o status terminal do Kling é "succeed" (não "succeeded") — isso corresponde ao formato real da resposta da API:

def submit_kling_task(prompt: str, duration: str = "5", mode: str = "std") -> str:    """Enviar uma tarefa de texto para vídeo do Kling. 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": "borrado, baixa qualidade, marca d'água",            "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 do 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"A tarefa Kling {task_id} falhou: {error_detail}"            )​        time.sleep(interval)        elapsed += interval​    raise TimeoutError(f"A tarefa Kling {task_id} expirou após {max_wait}s")

Source*:* CometAPI Kling Video docs

Anime uma imagem estática com o Runway

O Runway é somente image-to-video. Ele também requer um cabeçalho extra (X-Runway-Version):

def submit_runway_task(image_url: str, motion_prompt: str, duration: int = 5) -> str:    """Enviar uma tarefa do 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 do Runway. Retorna a URL do vídeo quando concluir."""    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 da 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"A tarefa Runway {task_id} falhou: {result.get('error', 'sem detalhes')}")​        time.sleep(interval)        elapsed += interval​    raise TimeoutError(f"A 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("Um timelapse de nuvens se movendo sobre o horizonte de uma cidade")video_url = poll_veo_task(task_id)download_video(video_url, "output/city_timelapse.mp4")

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

Lide com falhas

Source: CometAPI video generation docs

Versão em Node.js

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

// Node.js 18+ — sem pacotes extras necessários​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(`Envio do Veo falhou: ${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 ao consultar: ${res.status}`);    const result = await res.json();​    if (result.status === 'succeeded') return result.output[0];    if (['failed', 'cancelled'].includes(result.status)) {      throw new Error(`A tarefa ${taskId} falhou: ${result.error ?? 'sem detalhes'}`);    }    await new Promise(r => setTimeout(r, intervalMs));    elapsed += intervalMs;  }  throw new Error(`A 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: 'borrado, baixa qualidade, marca d\'água',      cfg_scale: 0.5,      mode,      aspect_ratio: '16:9',      duration    })  });  if (!res.ok) throw new Error(`Envio do Kling falhou: ${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 ao consultar 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(`A tarefa Kling ${taskId} falhou: ${JSON.stringify(result.data.task_result ?? 'sem detalhes')}`);    }    await new Promise(r => setTimeout(r, intervalMs));    elapsed += intervalMs;  }  throw new Error(`A 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(`Envio do Runway falhou: ${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 ao consultar Runway: ${res.status}`);    const result = await res.json();    const status = result.status;​    if (status === 'task_not_exist') {      // Específico da 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(`A tarefa Runway ${taskId} falhou: ${result.error ?? 'sem detalhes'}`);    }    await new Promise(r => setTimeout(r, intervalMs));    elapsed += intervalMs;  }  throw new Error(`A tarefa Runway ${taskId} expirou`);}​// Exemplo de usoconst taskId = await submitVeoTask('Um papagaio de papel pairando sobre um campo de trigo');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 lida com 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 alternar para uma opção mais barata ou rápida significa mexer em vários arquivos. O próximo artigo aborda como rotear solicitações entre modelos sem reescrever seu código.

Próximo: Como alternar entre modelos de IA sem reescrever seu código

FAQ

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

A 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 sua requisição não expirar. Você faz polling em 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 armazenamento (S3, Cloudflare R2, etc.). Não armazene apenas a URL esperando que funcione horas depois.

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

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

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

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

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

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 da resposta da API do Kling. Não é um erro de digitação. Veo e Runway usam "succeeded" — o Kling usa "succeed". Se você estiver construindo um wrapper unificado de polling, precisará lidar com 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 minutos. Em um serviço web real com usuários concorrentes, execute o polling em um worker em background (Celery para Python, Bull para Node.js). Envie a tarefa no request handler, retorne o task ID ao cliente e deixe o worker notificá-lo quando o vídeo estiver pronto.