Image-&-Video-Webhooks

Überblick

Die Videogenerierung kann mehrere Minuten dauern. Daher ist kontinuierliches Polling teuer. Aktivieren Sie die Webhook-Zustellung für eine Generierung. ElevenLabs sendet dann ein flows_generation-Ereignis an Ihren Endpunkt, sobald die Generierung completed oder failed erreicht.

Die Ereignis-Payload entspricht der finalen Antwort des zugehörigen GET-Endpunkts. Ein Handler, der die Polling-Antwort bereits verarbeitet, benötigt daher keinen separaten Parsing-Pfad.

Vorbereitung

Die Webhook-Zustellung nutzt die Webhooks, die Ihr Workspace für Generierungsereignisse abonniert hat. Die Einrichtung besteht aus zwei Schritten: Erstellen Sie den Webhook und abonnieren Sie dann das Ereignis.

{
  "events": ["flows"]
}

Zum Erstellen und Abonnieren von Webhooks benötigen Sie die Berechtigung Webhooks Manage oder müssen Workspace-Admin sein. Ein Ereignis akzeptiert bis zu 10 Webhooks. Darüber hinaus schlägt die Anfrage mit too_many_webhooks fehl.

Eine Generierung, die eine Webhook-Zustellung anfordert, obwohl kein Webhook für Generierungsereignisse abonniert ist, wird abgelehnt. So wird kein Ergebnis generiert, das nicht zugestellt werden kann.

Webhook-Zustellung anfordern

Fügen Sie der Erstellungsanfrage ein webhook-Objekt hinzu. Mit {"type": "all"} erfolgt die Zustellung an jeden Webhook, der Generierungsereignisse abonniert hat. So bleibt die Anfrage stabil, wenn Webhooks hinzugefügt oder ersetzt werden.

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(),
    )
)

Um stattdessen bestimmte Webhooks anzusprechen, setzen Sie das Feld webhook auf eine Liste von IDs. Jede ID muss zu einem Webhook des Workspaces gehören, der Generierungsereignisse abonniert hat.

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

Die Erstellungsanfrage prüft das Ziel, bevor die Generierung startet, und gibt einen Fehler zurück, wenn eine Zustellung nicht möglich wäre:

Die Webhook-Zustellung eignet sich gut für verkettete Generierungen: Setzen Sie webhook für die finale Generierung. Die gesamte Kette läuft dann serverseitig, mit einem einzigen Ereignis am Ende. Das gilt auch, wenn die Kette zwischendurch fehlschlägt — der Fehler wird an die finale Generierung weitergegeben, die ihn als failed-Ereignis mit dem Grund dependency_failed zustellt.

Webhook-Payload

Eine abgeschlossene Generierung liefert die Ausgabe-URL und den MIME-Typ:

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

Eine fehlgeschlagene Generierung liefert stattdessen die Fehlerkategorie und Meldung:

{
  "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."
  }
}

Prüfen Sie data.status, um zu bestimmen, welche Felder vorhanden sind. Die beiden finalen Status sind die einzigen, die ein Webhook enthalten kann, da die Zustellung erst erfolgt, wenn eine Generierung abgeschlossen ist.

Ereignis verarbeiten

Ein Handler verifiziert die Signatur, prüft den Ereignistyp und verzweigt dann anhand von data.status. Dieses Beispiel lädt die Ausgabe einer abgeschlossenen Generierung herunter und protokolliert den Grund bei einer fehlgeschlagenen.

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

Beide Beispiele laden der Kürze halber innerhalb der Anfrage herunter. Ein großes Video kann lange genug dauern, um das Zustellungs-Timeout zu überschreiten. Übergeben Sie die Generierungs-ID daher in der Produktion an eine Queue und geben Sie sofort 2xx zurück. Die signierte URL ist etwa eine Stunde gültig, was für einen Background-Worker ausreichend ist.

Signatur verifizieren

Der obige Handler ruft construct_event / constructEvent auf. Damit werden der Header ElevenLabs-Signature verifiziert, der Zeitstempel validiert und die Payload in einem Schritt geparst. Verifizieren Sie immer, bevor Sie einem Ereignis vertrauen.

Der Listener muss alle eingehenden Webhooks validieren. Webhooks unterstützen derzeit die Authentifizierung über HMAC-Signaturen. So richten Sie die HMAC-Authentifizierung ein:

• Speichern Sie das beim Erstellen des Webhooks generierte gemeinsame Geheimnis sicher.
• Verifizieren Sie den Header ElevenLabs-Signature in Ihrem Endpunkt mithilfe des SDK.

Das JavaScript-SDK stellt constructEvent bereit, das Python-SDK construct_event mit rawBody, sig_header und secret (diese heißen in Python nicht payload / signature). Beide verifizieren die Signatur, validieren den Zeitstempel und parsen die JSON-Nutzlast.

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

Zustellungsverhalten

Jede Generierung liefert genau ein finales Ereignis pro Ziel-Webhook. Die Zustellung ist unabhängig von der Generierung selbst: Ein Webhook, der fehlschlägt oder nicht erreichbar ist, beeinflusst das Ergebnis nicht. Dieses bleibt über den GET-Endpunkt und in der Listenantwort verfügbar.

Geben Sie von Ihrem Handler zeitnah einen 2xx-Status zurück. Wiederholte Fehler deaktivieren einen Webhook automatisch, und ein deaktivierter Webhook führt dazu, dass nachfolgende Generierungen, die ihn ansprechen, beim Erstellen abgelehnt werden. Gestalten Sie den Handler idempotent und verwenden Sie die Generierungs-id zur Deduplizierung.

Wenn ein verpasstes Ergebnis nicht akzeptabel ist, behandeln Sie Webhooks als schnellen Pfad und gleichen Sie regelmäßig mit flows.image.list oder flows.video.list ab, gefiltert nach status.

Nächste Schritte