SDK xAI et polling automatique

Leçon réécrite le 2 novembre 2026 d’après la documentation vidéo de xAI et le code du SDK xai-sdk 1.21.0 : les méthodes s’appellent client.video.generate et client.video.extend (l’édition passe par generate avec video_url), le résultat expose url, le délai se règle par un timedelta et l’intervalle de polling par défaut est d’une seconde. await exige le client asynchrone AsyncClient, et l’exception VideoGenerationError s’importe de xai_sdk.video. L’ancienne version appelait des méthodes et un module d’exceptions qui n’existent pas. Sources : Video Generation · Video Editing · Video Extension · xai-sdk-python.

Simplifier le polling avec les SDK

La leçon précédente vous a fait écrire une boucle de polling à la main. C’est un exercice utile pour comprendre le cycle de vie d’une génération, mais c’est aussi une trentaine de lignes à maintenir dans chaque projet, avec sa gestion de timeout, ses cas d’échec et ses conversions de statut. Le SDK Python officiel de xAI (pip install xai-sdk) encapsule tout cela : vous soumettez la requête, il s’occupe de l’attente.

Un appel unique au lieu d’une boucle

La méthode client.video.generate regroupe la soumission, le suivi et la récupération. Elle ne rend la main qu’une fois la vidéo prête, ou lève une exception si quelque chose s’est mal passé.

import os

import xai_sdk

client = xai_sdk.Client(api_key=os.getenv("XAI_API_KEY"))

# Génération text-to-video avec polling automatique
video = client.video.generate(
    prompt="Un lever de soleil sur les Alpes, time-lapse accéléré",
    model="grok-imagine-video-1.5",
    duration=5,
    aspect_ratio="16:9",
    resolution="720p",
)

print(video.url)
print(f"Durée : {video.duration} s, modèle : {video.model}")

Derrière ces quelques lignes, le SDK effectue exactement ce que vous codiez précédemment : la soumission de la requête, l’interrogation du statut toutes les secondes par défaut, puis, selon l’issue, le retour du résultat ou la levée d’une exception.

Un réglage mérite votre attention dès la mise en service : par défaut, le SDK patiente jusqu’à 10 minutes avant d’abandonner. C’est confortable pour un script de production nocturne, beaucoup trop long pour une requête web à laquelle un utilisateur attend une réponse. Les paramètres timeout et interval, deux timedelta, règlent la borne et le rythme :

from datetime import timedelta

video = client.video.generate(
    prompt="Votre description",
    model="grok-imagine-video-1.5",
    duration=10,
    timeout=timedelta(minutes=5),   # 5 minutes au plus
    interval=timedelta(seconds=5),  # un contrôle toutes les 5 secondes
)

Si vous voulez garder la main sur l’attente — pour relayer une progression, par exemple —, client.video.start soumet la requête et renvoie son request_id, que client.video.get interroge ensuite.

Traiter les erreurs à la source

Le SDK distingue les échecs, ce qui vous évite d’analyser des chaînes de caractères pour savoir ce qui s’est passé :

from xai_sdk.video import VideoGenerationError

try:
    video = client.video.generate(
        prompt="Votre description",
        model="grok-imagine-video-1.5",
        duration=5,
    )
    print(f"Vidéo prête : {video.url}")
except VideoGenerationError as e:
    print(f"Génération échouée [{e.code}] : {e.message}")
except TimeoutError:
    print("Délai dépassé : augmentez le timeout ou simplifiez la demande")

Chaque erreur appelle une réaction différente, et les confondre conduit à des boucles de retry inutiles ou, pire, à des générations facturées en pure perte.

Le SDK passe par gRPC : une limite de débit dépassée n’y apparaît pas comme un code HTTP 429, mais comme une grpc.RpcError de code RESOURCE_EXHAUSTED.

Édition et extension, mêmes facilités

L’édition n’a pas de méthode propre : c’est generate avec une vidéo source, passée en video_url (vidéo de 8,7 secondes au plus, sortie de même durée, 720p au plus).

video = client.video.generate(
    prompt="Transforme la scène en style anime japonais",
    model="grok-imagine-video",
    video_url="https://exemple.com/video.mp4",
)

L’extension prolonge une séquence de 2 à 15 secondes d’un segment de 2 à 10 secondes, et renvoie la vidéo complète :

video = client.video.extend(
    prompt="Continue la scène avec un panoramique vers la droite",
    model="grok-imagine-video",
    video_url="https://exemple.com/video.mp4",
    duration=5,
)

Notez la différence avec l’API REST : le SDK prend la vidéo en paramètre video_url, là où le corps JSON attend un objet video avec sa propriété url.

Ne jamais laisser une vidéo dans le cloud

Le SDK ne change rien au caractère temporaire des URL de résultat. Si votre code se contente d’afficher video.url, vous avez payé une génération dont il ne restera bientôt plus rien. La bonne pratique consiste à enchaîner génération et téléchargement dans la même fonction, de sorte qu’il devienne impossible d’oublier l’étape.

from pathlib import Path

import httpx


async def generer_et_sauvegarder(client, prompt, fichier_sortie, **options):
    """Génère une vidéo et la sauvegarde localement (client : xai_sdk.AsyncClient)."""
    video = await client.video.generate(
        prompt=prompt,
        model="grok-imagine-video-1.5",
        **options,
    )

    # Téléchargement immédiat
    async with httpx.AsyncClient() as http:
        reponse = await http.get(video.url)
        reponse.raise_for_status()
        Path(fichier_sortie).write_bytes(reponse.content)

    print(f"Vidéo sauvegardée : {fichier_sortie}")
    return fichier_sortie

Pour un stockage sans transit par votre serveur, le paramètre storage_options de la génération fait déposer la vidéo dans la Files API de xAI, avec une expiration de votre choix.

Générer plusieurs vidéos de front

Avec le client asynchrone, chaque appel est une coroutine, et asyncio.gather lance un lot entier en parallèle. L’option return_exceptions=True est ici décisive : sans elle, une seule requête refusée ferait échouer l’ensemble du lot et vous perdriez les résultats déjà obtenus.

import asyncio
import os

import xai_sdk


async def generation_parallele(requetes):
    """Génère plusieurs vidéos en parallèle."""
    client = xai_sdk.AsyncClient(api_key=os.getenv("XAI_API_KEY"))
    taches = [
        client.video.generate(
            prompt=req["prompt"],
            model="grok-imagine-video-1.5",
            duration=req.get("duration", 5),
            aspect_ratio=req.get("aspect_ratio", "16:9"),
        )
        for req in requetes
    ]

    resultats = await asyncio.gather(*taches, return_exceptions=True)

    for i, res in enumerate(resultats, 1):
        if isinstance(res, Exception):
            print(f"Requête {i} échouée : {res}")
        else:
            print(f"Requête {i} terminée : {res.url}")

    return resultats

Deux garde-fous encadrent cette parallélisation. Les soumissions consomment votre limite de débit, comptée en requêtes par seconde selon le palier de votre équipe (10 au palier d’entrée pour les modèles vidéo) : pour un gros lot, bornez la concurrence avec un asyncio.Semaphore. Rappelez-vous aussi ce que le parallélisme fait et ne fait pas : il n’accélère aucune génération individuelle, il réduit seulement le temps total du lot.

Faut-il abandonner le polling manuel ?

Pas systématiquement. La comparaison suivante montre où chaque approche prend l’avantage.

Le polling manuel garde sa pertinence dans deux situations : quand vous avez besoin d’un contrôle fin, par exemple pour relayer la progression vers une interface temps réel ou pour l’intégrer à un système de files existant, et quand votre contexte de déploiement vous interdit d’ajouter une dépendance supplémentaire.

Points clés à retenir

• client.video.generate soumet, attend et renvoie la vidéo (url, duration, model) ; timeout de 10 minutes et contrôle chaque seconde par défaut
• L’édition passe par generate avec video_url, l’extension par client.video.extend
• VideoGenerationError (de xai_sdk.video), TimeoutError et RuntimeError distinguent les échecs ; la limite de débit arrive en grpc.RpcError
• Téléchargez toujours la vidéo immédiatement après la génération (URL temporaires)
• Le client asynchrone et asyncio.gather parallélisent un lot, dans la limite de débit de votre palier