Come aggiungere la generazione di video con IA a un'app SaaS

Aggiungere la generazione di video alla tua app non è la stessa cosa che aggiungere la generazione di immagini. La chiamata all’API restituisce subito la risposta — ma il video non è ancora pronto. Ricevi un ID attività e devi continuare a chiedere «è pronto?» finché non lo è.

La maggior parte degli sviluppatori ci inciampa la prima volta che chiama un’API video: attende nel body della risposta un URL del video e invece riceve un ID attività. Questa guida illustra l’intero flusso: inviare un’attività, effettuare il polling per i risultati, gestire gli errori e archiviare l’output prima che l’URL scada.

What you'll build

Un servizio backend che accetta un prompt testuale o un’immagine, invia un’attività di generazione video, effettua il polling finché non è completata e restituisce l’URL finale del video. Lavorerai con quattro modelli — Veo 3 Fast, Sora 2, Kling Video e Runway — tutti tramite una singola chiave API.

Prerequisites:

• Python 3.8+ o Node.js 18+
• Una chiave CometAPI
• Familiarità di base con le API REST

Understand why video generation is different

Con la generazione di immagini, invii una richiesta e ricevi l’immagine nella stessa risposta. La generazione di video usa una coda di attività asincrona:

1. Submit una richiesta di generazione → ricevi un task_id
2. Poll un endpoint di stato ogni pochi secondi
3. Quando lo stato raggiunge uno stato terminale, ottieni l’URL del video
4. Download and store il video — l’URL è temporaneo

Se tratti la generazione di video come quella di immagini e aspetti che la prima risposta contenga il tuo video, la richiesta andrà in timeout ogni volta.

In un servizio web in produzione, questo ciclo di polling dovrebbe essere eseguito in un worker in background (Celery, Bull o simili), non nell’handler della richiesta. Gli esempi sotto usano polling sincrono — va bene per script e prototipi, ma non per gestire utenti concorrenti.

Choose a model

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 supporta sia text-to-video sia image-to-video. Il più economico al secondo, un buon punto di partenza.
• Sora 2 genera audio nativamente insieme al video — dialoghi, suoni ambientali ed effetti senza un passaggio TTS separato.
• Kling Video offre negative_prompt, cfg_scale, impostazioni di movimento camera e una modalità pro. Il maggior controllo tra i quattro.
• Runway è solo da immagine a video tramite CometAPI. Fornisci un’immagine statica e una descrizione del movimento, e la anima.

Submit a Veo task

Veo usa multipart/form-data. Usa files= in Python requests per inviarlo correttamente — data=dict invia application/x-www-form-urlencoded, che non è la stessa cosa:

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

Poll for the result

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

Use Kling Video for more control

Kling ha una struttura di endpoint diversa e usa JSON. Nota che lo stato terminale di Kling è "succeed" (non "succeeded") — questo corrisponde al formato di risposta effettivo dell’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

Animate a static image with Runway

Runway è solo da immagine a video. Richiede anche un header 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

Save the video before the URL expires

Gli URL video delle API di generazione sono temporanei. Scarica il file immediatamente e conservalo in un luogo sotto il tuo controllo:

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

In produzione, sostituisci la scrittura su file locale con un upload su S3, Cloudflare R2 o lo storage che preferisci. Il pattern di streaming rimane lo stesso — invia i byte direttamente invece di caricare l’intero video in memoria.

Handle failures

Source: CometAPI video generation docs

Node.js version

Node.js 18+ include fetch e FormData nativamente. Questo esempio copre tutti e quattro i modelli:

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

What's next

Ora hai codice funzionante per quattro modelli video, un ciclo di polling che gestisce gli errori e un passaggio di download che ti evita di perdere i contenuti generati.

Il prossimo problema che la maggior parte degli sviluppatori incontra: hanno fissato un unico modello, e passare a un’opzione più economica o più veloce significa toccare più file. Il prossimo articolo spiega come instradare le richieste tra modelli senza riscrivere il tuo codice.

Next: How to Switch Between AI Models Without Rewriting Your Code

FAQ

Q: Why do I get a task ID instead of a video in the API response?

La generazione video è asincrona — modelli come Veo, Sora, Kling e Runway impiegano 2–5 minuti per il rendering. L’API restituisce subito un ID attività per evitare il timeout della richiesta. Effettui il polling di un endpoint di stato separato finché l’attività non raggiunge uno stato terminale (succeeded, succeed, failed).

Q: How long does a generated video URL stay valid?

Gli URL dei video generati sono temporanei. Scarica il file immediatamente dopo aver ottenuto l’URL e archivialo nel tuo storage (S3, Cloudflare R2, ecc.). Non salvare l’URL aspettandoti che funzioni ore dopo.

Q: What's the difference between Veo 3 Fast and Kling Video?

Veo 3 Fast è più economico ($0.05/sec), più veloce e più semplice da chiamare. Kling Video offre più controllo: negative_prompt, cfg_scale, impostazioni di movimento della camera e una modalità di qualità pro. Se devi affinare l’output, usa Kling. Se ti servono velocità e basso costo, usa Veo 3 Fast.

Q: Can I generate video from an image instead of a text prompt?

Sì. Veo supporta da immagine a video passando un file input_reference. Kling lo supporta tramite l’endpoint /kling/v1/videos/image2video con un parametro image (URL o base64). Runway è solo da immagine a video — non accetta prompt solo testuali tramite CometAPI.

Q: Why does Runway return task_not_exist on the first poll?

È un comportamento documentato di CometAPI — l’attività si sta ancora inizializzando sul backend. Attendi qualche secondo e riprova. Non è un errore. Il codice di polling sopra lo gestisce automaticamente.

Q: Why does Kling use "succeed" instead of "succeeded"?

È il formato di risposta effettivo dell’API di Kling. Non è un refuso. Veo e Runway usano "succeeded" — Kling usa "succeed". Se stai costruendo un wrapper di polling unificato, dovrai gestire entrambe le stringhe.

Q: Is the synchronous polling loop safe to use in a web server?

No. Il ciclo di polling in questa guida blocca il thread per diversi minuti. In un servizio web reale con utenti concorrenti, esegui il polling in un worker in background (Celery per Python, Bull per Node.js). Invia l’attività nell’handler della richiesta, restituisci l’ID attività al client e lascia che il worker notifichi il client quando il video è pronto.