Image & Video-Webhooks

Überblick

Videogenerierungen können mehrere Minuten dauern, wodurch fortlaufende Abfragen aufwendig sind. Aktivieren Sie die Webhook-Zustellung für eine Generierung, und ElevenLabs sendet 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 bereits die Abfrageantwort versteht, benötigt daher keinen separaten Parsing-Pfad.

Bevor Sie beginnen

Die Webhook-Zustellung verwendet 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 Workspace-Adminrechte. Ein Ereignis akzeptiert bis zu 10 Webhooks. Darüber hinaus schlägt die Anfrage mit too_many_webhooks fehl.

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

Webhook-Zustellung anfordern

Fügen Sie der Erstellungsanfrage ein webhook-Objekt hinzu. Verwenden Sie {"type": "all"}, um an jeden Webhook zuzustellen, 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 validiert das Ziel vor dem Start der Generierung 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 letzte Generierung, und die gesamte Kette läuft serverseitig mit einem einzigen Ereignis am Ende. Das gilt auch, wenn die Kette mittendrin fehlschlägt — der Fehler wird an die letzte Generierung weitergegeben, die ihn als failed-Ereignis mit dem Grund dependency_failed zustellt.

Webhook-Payload

Eine abgeschlossene Generierung übermittelt 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 übermittelt stattdessen 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 entscheiden, welche Felder vorhanden sind. Die beiden finalen Status sind die einzigen, die ein Webhook enthalten kann, da die Zustellung nur erfolgt, wenn eine Generierung endet.

Das 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 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. Dadurch 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 adressiertem Webhook. Die Zustellung ist unabhängig von der Generierung selbst: Ein fehlgeschlagener oder nicht erreichbarer Webhook beeinträchtigt das Ergebnis nicht. Es bleibt über den GET-Endpunkt und in der Listenantwort verfügbar.

Geben Sie in 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 adressieren, 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