Comment ajouter la génération de vidéos par IA à une application SaaS

Ajouter la génération vidéo à votre application n’est pas la même chose qu’ajouter la génération d’images. L’appel d’API renvoie immédiatement — mais la vidéo n’est pas encore prête. Vous obtenez un ID de tâche, et vous devez continuer à demander « est-ce terminé ? » jusqu’à ce que ça le soit.

La plupart des développeurs rencontrent cela la première fois qu’ils appellent une API vidéo, attendent un corps de réponse avec une URL de vidéo et reçoivent à la place un ID de tâche. Ce guide présente le flux complet : soumettre une tâche, sonder les résultats, gérer les échecs et stocker la sortie avant l’expiration de l’URL.

Ce que vous allez créer

Un service backend qui accepte une invite textuelle ou une image, soumet une tâche de génération vidéo, sonde jusqu’à l’achèvement, puis renvoie l’URL finale de la vidéo. Vous travaillerez avec quatre modèles — Veo 3 Fast, Sora 2, Kling Video et Runway — tous via une seule clé d’API.

Prérequis :

• Python 3.8+ ou Node.js 18+
• Une clé CometAPI
• Familiarité de base avec les API REST

Comprendre pourquoi la génération vidéo est différente

Avec la génération d’images, vous envoyez une requête et vous obtenez l’image dans la même réponse. La génération vidéo utilise une file de tâches asynchrone :

1. Soumettre une requête de génération → récupérer un task_id
2. Sonder une API de statut toutes les quelques secondes
3. Quand le statut atteint un état terminal, vous obtenez l’URL de la vidéo
4. Télécharger et stocker la vidéo — l’URL est temporaire

Si vous traitez la génération vidéo comme la génération d’images et attendez que la première réponse contienne votre vidéo, votre requête expirera à chaque fois.

Dans un service web en production, cette boucle de sondage doit s’exécuter dans un worker d’arrière-plan (Celery, Bull, ou similaire), pas dans votre gestionnaire de requêtes. Les exemples ci-dessous utilisent un sondage synchrone — parfait pour des scripts et des prototypes, mais pas pour gérer des utilisateurs concurrents.

Choisir un modèle

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 prend en charge texte-vers-vidéo et image-vers-vidéo. Le moins cher par seconde, bon point de départ.
• Sora 2 génère nativement l’audio en même temps que la vidéo — dialogues, ambiance et effets sans étape TTS séparée.
• Kling Video vous offre negative_prompt, cfg_scale, des réglages de mouvement de caméra et un mode pro. Le plus de contrôle des quatre.
• Runway est uniquement image-vers-vidéo via CometAPI. Donnez-lui une image statique et une description du mouvement, il l’anime.

Soumettre une tâche Veo

Veo utilise multipart/form-data. Utilisez files= avec la bibliothèque requests en Python pour l’envoyer correctement — data=dict envoie application/x-www-form-urlencoded, ce qui n’est pas la même chose :

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

Sonder le résultat

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

Utiliser Kling Video pour plus de contrôle

Kling a une structure d’endpoint différente et utilise JSON. Notez que la chaîne de statut terminal de Kling est "succeed" (et non "succeeded") — cela correspond au format de réponse réel de l’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

Animer une image statique avec Runway

Runway est uniquement image-vers-vidéo. Il nécessite également un en-tête supplémentaire (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

Enregistrer la vidéo avant l’expiration de l’URL

Les URL de vidéo fournies par les API de génération sont temporaires. Téléchargez le fichier immédiatement et stockez-le dans un emplacement que vous contrôlez :

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 production, remplacez l’écriture locale par un envoi vers S3, Cloudflare R2 ou le stockage de votre choix. Le modèle de streaming reste identique — acheminez les octets directement plutôt que de charger toute la vidéo en mémoire.

Gérer les échecs

Source: CometAPI video generation docs

Version Node.js

Node.js 18+ inclut nativement fetch et FormData. Cet exemple couvre les quatre modèles :

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

Et ensuite

Vous disposez maintenant de code fonctionnel pour quatre modèles vidéo, d’une boucle de sondage qui gère les échecs et d’une étape de téléchargement qui vous évite de perdre du contenu généré.

Le prochain problème que rencontrent la plupart des développeurs : ils ont codé en dur un seul modèle, et passer à une option moins chère ou plus rapide implique de modifier plusieurs fichiers. Le prochain article explique comment router les requêtes entre les modèles sans réécrire votre code.

À suivre : Comment passer d’un modèle d’IA à un autre sans réécrire votre code

FAQ

Q : Pourquoi est-ce que je reçois un ID de tâche au lieu d’une vidéo dans la réponse de l’API ?

La génération vidéo est asynchrone — des modèles comme Veo, Sora, Kling et Runway prennent 2 à 5 minutes pour le rendu. L’API renvoie immédiatement un ID de tâche afin que votre requête n’expire pas. Vous sondez un endpoint de statut séparé jusqu’à ce que la tâche atteigne un état terminal (succeeded, succeed, failed).

Q : Combien de temps l’URL d’une vidéo générée reste-t-elle valide ?

Les URL de vidéo des API de génération sont temporaires. Téléchargez le fichier immédiatement après avoir obtenu l’URL et stockez-le dans votre propre stockage (S3, Cloudflare R2, etc.). Ne stockez pas l’URL en supposant qu’elle fonctionnera encore des heures plus tard.

Q : Quelle est la différence entre Veo 3 Fast et Kling Video ?

Veo 3 Fast est moins cher ($0.05/sec), plus rapide et plus simple à appeler. Kling Video offre davantage de contrôle : negative_prompt, cfg_scale, réglages de mouvement de caméra et un mode qualité pro. Si vous avez besoin d’affiner le rendu, utilisez Kling. Si vous cherchez la vitesse et le faible coût, utilisez Veo 3 Fast.

Q : Puis-je générer une vidéo à partir d’une image plutôt que d’une invite textuelle ?

Oui. Veo prend en charge l’image-vers-vidéo en passant un fichier input_reference. Kling le prend en charge via l’endpoint /kling/v1/videos/image2video avec un paramètre image (URL ou base64). Runway est uniquement image-vers-vidéo — il n’accepte pas les invites texte seules via CometAPI.

Q : Pourquoi Runway renvoie-t-il task_not_exist au premier sondage ?

C’est un comportement documenté de CometAPI — la tâche est encore en cours d’initialisation côté backend. Attendez quelques secondes et réessayez. Ce n’est pas une erreur. Le code de sondage ci-dessus le gère automatiquement.

Q : Pourquoi Kling utilise "succeed" au lieu de "succeeded" ?

C’est le format de réponse réel de l’API Kling. Ce n’est pas une faute de frappe. Veo et Runway utilisent "succeeded" — Kling utilise "succeed". Si vous bâtissez un wrapper de sondage unifié, vous devrez gérer les deux chaînes.

Q : La boucle de sondage synchrone est-elle sûre à utiliser dans un serveur web ?

Non. La boucle de sondage de ce guide bloque le thread pendant plusieurs minutes. Dans un véritable service web avec des utilisateurs concurrents, exécutez le sondage dans un worker d’arrière-plan (Celery pour Python, Bull pour Node.js). Soumettez la tâche dans le gestionnaire de requêtes, renvoyez l’ID de tâche au client, et laissez le worker notifier le client quand la vidéo est prête.