Webhooks Image & Vidéo

Présentation

Les générations de vidéos peuvent prendre plusieurs minutes, ce qui rend leur interrogation régulière coûteuse. Activez la livraison 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 est la réponse finale de l’endpoint GET correspondant. Un gestionnaire qui comprend déjà la réponse d’interrogation n’a donc besoin d’aucune logique d’analyse distincte.

Avant de commencer

La livraison par webhook utilise les webhooks de votre Workspace abonnés aux événements de génération. La 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 Gérer les webhooks, 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 livraison 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 la livraison par webhook

Ajoutez un objet webhook à la requête de création. Utilisez {"type": "all"} pour livrer le résultat à chaque webhook abonné aux événements de génération. La requête reste ainsi 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 comme une liste d’ID. Chaque ID doit correspondre à l’un des webhooks du Workspace abonnés 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 livraison est impossible :

La livraison 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 reste vrai si la chaîne échoue en cours de route : l’échec se propage jusqu’à la génération finale, qui le transmet en tant qu’événement failed avec le motif dependency_failed.

Charge utile du webhook

Une génération terminée fournit 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 fournit à 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 quels champs sont présents. Les deux états finaux sont les seuls qu’un webhook peut contenir, car la livraison a lieu uniquement lorsqu’une génération se termine.

Traiter l’événement

Un gestionnaire vérifie la signature, contrôle le type d’événement, puis utilise data.status. Cet exemple télécharge la sortie d’une génération terminée et journalise 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"}

Les deux exemples effectuent le téléchargement dans la requête par souci de concision. Une vidéo volumineuse peut prendre suffisamment longtemps pour dépasser le délai de livraison. En production, transmettez donc l’ID de génération à une file d’attente et renvoyez immédiatement un code 2xx. L’URL signée reste valide environ une heure, ce qui laisse largement le temps à un processus d’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 lui faire confiance.

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 livraison

Chaque génération fournit exactement un événement final par webhook ciblé. La livraison 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 état 2xx depuis votre gestionnaire. Des échecs répétés désactivent automatiquement un webhook, et un webhook désactivé entraîne le rejet, lors de leur création, des générations suivantes qui le ciblent. Concevez le gestionnaire de manière idempotente et utilisez l’id de la génération pour dédupliquer.

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

Étapes suivantes