Comment ajouter la génération vidéo 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 recevez 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 détaille le flux complet : soumettre une tâche, interroger pour les résultats, gérer les échecs et stocker la sortie avant l’expiration de l’URL.

Ce que vous allez construire

Un service backend qui accepte un prompt texte ou une image, soumet une tâche de génération vidéo, interroge jusqu’à son achèvement et 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
• Une 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 recevez l’image dans la même réponse. La génération vidéo utilise une file de tâches asynchrones :

1. Soumettre une requête de génération → recevoir un task_id
2. Interroger un endpoint de statut toutes les quelques secondes
3. Lorsque 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 de production, cette boucle d’interrogation doit s’exécuter dans un worker en arrière-plan (Celery, Bull, ou similaire), pas dans votre gestionnaire de requêtes. Les exemples ci-dessous utilisent une interrogation synchrone — convenable pour des scripts et 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 le texte-vers-vidéo et l’image-vers-vidéo. Le moins cher par seconde, bon point de départ.
• Sora 2 génère l’audio nativement avec la vidéo — dialogues, ambiances et effets sans étape TTS séparée.
• Kling Video vous donne 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, et il l’anime.

Soumettre une tâche Veo

Veo utilise multipart/form-data. Utilisez files= avec la bibliothèque Python requests 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}")

Interroger pour obtenir 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 vidéo des 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 en fichier local par un upload vers S3, Cloudflare R2, ou le stockage de votre choix. Le schéma de streaming reste le même — transférez 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 fetch et FormData nativement. 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 avez maintenant du code fonctionnel pour quatre modèles vidéo, une boucle d’interrogation qui gère les échecs, et une étape de téléchargement qui vous évite de perdre le contenu généré.

Le prochain problème que rencontrent la plupart des développeurs : ils ont codé en dur un 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.

Prochain : Comment basculer entre des modèles d’IA 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 mettent 2–5 minutes à rendre. L’API renvoie immédiatement un ID de tâche pour éviter l’expiration de votre requête. Vous interrogez un endpoint de statut séparé jusqu’à ce que la tâche atteigne un état terminal (succeeded, succeed, failed).

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

Les URL 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 espérant qu’elle fonctionne 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 vous offre davantage de contrôle : negative_prompt, cfg_scale, réglages de mouvement de caméra, et un mode qualité pro. Si vous devez affiner le rendu, utilisez Kling. Si vous avez besoin de vitesse et de faible coût, utilisez Veo 3 Fast.

Q : Puis-je générer une vidéo à partir d’une image au lieu d’un prompt texte ?

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 de prompts texte seuls 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 initialisation côté backend. Attendez quelques secondes et réessayez. Ce n’est pas une erreur. Le code d’interrogation ci-dessus le gère automatiquement.

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

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

Q : La boucle d’interrogation synchrone est-elle sûre à utiliser dans un serveur web ?

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