Webhooks Image & Vidéo

Vue d’ensemble

La génération de vidéos peut prendre plusieurs minutes, ce qui rend l’attente active coûteuse. Activez la diffusion par webhook pour une génération, et ElevenLabs envoie un événement flows_generation à votre endpoint lorsque la génération atteint l’état completed ou failed.

La charge utile de l’événement correspond à la réponse finale de l’endpoint GET associé. Un gestionnaire qui comprend déjà la réponse d’attente active n’a donc pas besoin d’une logique d’analyse distincte.

Avant de commencer

La diffusion par webhook utilise les webhooks auxquels votre Workspace est abonné pour les événements de génération. Sa configuration se fait en deux étapes : créez le webhook, puis abonnez-le à l’événement.

{
  "events": ["flows"]
}

La création et l’abonnement de webhooks nécessitent l’autorisation Webhooks Manage ou le rôle d’administrateur du Workspace. Un même événement accepte jusqu’à 10 webhooks. Au-delà, la requête échoue avec too_many_webhooks.

Une génération qui demande une diffusion par webhook alors qu’aucun webhook n’est abonné aux événements de génération est rejetée. Un résultat n’est donc jamais généré sans destination de livraison.

Demander une diffusion par webhook

Ajoutez un objet webhook à la requête de création. Utilisez {"type": "all"} pour diffuser vers tous les webhooks abonnés aux événements de génération. Ainsi, la requête reste stable lorsque des webhooks sont ajoutés ou remplacés.

from elevenlabs import VideoGenerationRequest_Veo31FastGenerate001, WebhookTarget_All
generation = elevenlabs.flows.video.create(
    request=VideoGenerationRequest_Veo31FastGenerate001(
        prompt="A corgi rides a tiny surfboard across a sunlit wave at golden hour, cinematic",
        duration_secs=8,
        webhook=WebhookTarget_All(),
    )
)

Pour cibler des webhooks spécifiques, définissez plutôt le champ webhook sur une liste d’identifiants. Chaque identifiant doit correspondre à l’un des webhooks du Workspace abonné aux événements de génération.

{
  "webhook": {
    "type": "ids",
    "ids": ["Q8mVr2LpXcT4nB6yJdKw"]
  }
}

La requête de création valide la cible avant de lancer la génération et renvoie une erreur lorsque la diffusion ne serait pas possible :

La diffusion par webhook s’associe bien aux générations enchaînées : définissez webhook sur la génération finale, et toute la chaîne s’exécute côté serveur avec un seul événement à la fin. Cela s’applique également si la chaîne échoue en cours de route : l’échec se propage à la génération finale, qui le diffuse sous forme d’événement failed avec le motif dependency_failed.

Charge utile du webhook

Une génération terminée diffuse l’URL de sortie et le type MIME :

{
  "type": "flows_generation",
  "event_timestamp": 1739721600,
  "data": {
    "id": "JWr5N6X9ZTqf8jD2LmQb",
    "status": "completed",
    "content_url": "https://storage.googleapis.com/generations/JWr5N6X9ZTqf8jD2LmQb",
    "content_mime_type": "video/mp4"
  }
}

Une génération échouée diffuse à la place la catégorie et le message d’échec :

{
  "type": "flows_generation",
  "event_timestamp": 1739721600,
  "data": {
    "id": "JWr5N6X9ZTqf8jD2LmQb",
    "status": "failed",
    "failure_reason": "timeout",
    "error_message": "Timed out while processing. You were not charged for this generation."
  }
}

Utilisez data.status pour déterminer les champs présents. Les deux états finaux sont les seuls qu’un webhook peut contenir, car la diffusion n’a lieu qu’à la fin d’une génération.

Gérer l’événement

Un gestionnaire vérifie la signature, contrôle le type d’événement, puis agit selon data.status. Cet exemple télécharge la sortie d’une génération terminée et consigne le motif d’une génération échouée.

# server.py
import os
import requests
from dotenv import load_dotenv
from elevenlabs.client import ElevenLabs
from elevenlabs.errors import BadRequestError
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
load_dotenv()
app = FastAPI()
elevenlabs = ElevenLabs(api_key=os.getenv("ELEVENLABS_API_KEY"))
WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET")
@app.post("/webhook/flows")
async def receive_generation(request: Request):
    payload = await request.body()
    signature = request.headers.get("elevenlabs-signature")
    try:
        event = elevenlabs.webhooks.construct_event(
            rawBody=payload.decode("utf-8"),
            sig_header=signature,
            secret=WEBHOOK_SECRET,
        )
    except BadRequestError:
        return JSONResponse(content={"error": "Invalid signature"}, status_code=401)
    # construct_event returns a parsed dict, not an object with attributes.
    if event.get("type") != "flows_generation":
        return {"status": "ignored"}
    generation = event["data"]
    if generation["status"] == "completed":
        media = requests.get(generation["content_url"]).content
        with open(f"{generation['id']}.mp4", "wb") as f:
            f.write(media)
    else:
        print(f"Generation {generation['id']} failed: {generation['failure_reason']}")
    return {"status": "received"}

Par souci de concision, les deux exemples téléchargent le fichier pendant la requête. Une grande vidéo peut prendre suffisamment longtemps pour dépasser le délai de diffusion. En production, placez plutôt l’identifiant de génération dans une file d’attente et renvoyez immédiatement un statut 2xx. L’URL signée reste valide environ une heure, ce qui laisse largement le temps à un worker en arrière-plan.

Vérifier la signature

Le gestionnaire ci-dessus appelle construct_event / constructEvent, qui vérifie l’en-tête ElevenLabs-Signature, valide l’horodatage et analyse la charge utile en une étape. Vérifiez toujours un événement avant de vous y fier.

Il est important que le récepteur valide tous les webhooks entrants. Les webhooks prennent actuellement en charge l’authentification par signatures HMAC. Configurez l’authentification HMAC en :

• Stockant de manière sécurisée le secret partagé généré lors de la création du webhook
• Vérifiant l’en-tête ElevenLabs-Signature dans votre endpoint à l’aide du SDK

Le SDK JavaScript expose constructEvent ; le SDK Python expose construct_event avec rawBody, sig_header et secret (ils ne s’appellent pas payload / signature en Python). Les deux vérifient la signature, valident l’horodatage et analysent la charge utile JSON.

from dotenv import load_dotenv
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from elevenlabs.client import ElevenLabs
from elevenlabs.errors import BadRequestError
import os
load_dotenv()
app = FastAPI()
elevenlabs = ElevenLabs(
    api_key=os.getenv("ELEVENLABS_API_KEY"),
)
WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET")
@app.post("/webhook")
async def receive_message(request: Request):
    payload = await request.body()
    signature = request.headers.get("elevenlabs-signature")
    try:
        event = elevenlabs.webhooks.construct_event(
            rawBody=payload.decode("utf-8"),
            sig_header=signature,
            secret=WEBHOOK_SECRET,
        )
    except BadRequestError as e:
        return JSONResponse(content={"error": "Invalid signature"}, status_code=401)
    # construct_event returns a dict (parsed JSON), not an object with attributes
    if event.get("type") == "post_call_transcription":
        print(f"Received transcription: {event.get('data')}")
    return {"status": "received"}

Comportement de diffusion

Chaque génération diffuse exactement un événement final par webhook ciblé. La diffusion est indépendante de la génération elle-même : un webhook qui échoue ou est inaccessible n’affecte pas le résultat, qui reste disponible depuis l’endpoint GET et dans la réponse de liste.

Renvoyez rapidement un statut 2xx depuis votre gestionnaire. Des échecs répétés désactivent automatiquement un webhook, et un webhook désactivé entraîne le rejet, à la création, des générations ultérieures qui le ciblent. Concevez le gestionnaire de façon idempotente et utilisez l’id de génération pour dédupliquer.

Pour les flux où aucun résultat manqué n’est acceptable, considérez les webhooks comme le chemin rapide et rapprochez périodiquement les résultats avec flows.image.list ou flows.video.list, en filtrant selon status.

Étapes suivantes