Webhooks voor Afbeelding & Video

Overzicht

Het genereren van video’s kan enkele minuten duren, waardoor polling kostbaar is om open te houden. Kies voor webhooklevering voor een generatie en ElevenLabs stuurt een flows_generation-event naar je endpoint zodra de generatie de status completed of failed bereikt.

De eventpayload is de eindresponse van het bijbehorende GET-endpoint, dus een handler die de pollingresponse al begrijpt, heeft geen apart parsingpad nodig.

Voordat je begint

Webhooklevering gebruikt de webhooks waarop je workspace is geabonneerd voor generatie-events. Het instellen bestaat uit twee stappen: maak de webhook en abonneer deze vervolgens op het event.

{
  "events": ["flows"]
}

Voor het maken en abonneren van webhooks heb je de machtiging Webhooks Manage of workspacebeheerderrechten nodig. Eén event accepteert maximaal 10 webhooks; daarboven mislukt de aanvraag met too_many_webhooks.

Een generatie die webhooklevering aanvraagt terwijl geen webhook op generatie-events is geabonneerd, wordt geweigerd. Zo wordt er nooit een resultaat gegenereerd zonder bestemming.

Webhooklevering aanvragen

Voeg een webhook-object toe aan de create-request. Gebruik {"type": "all"} om te leveren aan elke webhook die is geabonneerd op generatie-events. Zo blijft de aanvraag stabiel wanneer webhooks worden toegevoegd of vervangen.

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

Om in plaats daarvan specifieke webhooks te targeten, stel je het veld webhook in op een lijst met ID’s. Elke ID moet behoren tot een webhook van de workspace die is geabonneerd op generatie-events.

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

De create-request valideert het doel voordat de generatie start en retourneert een fout wanneer levering niet mogelijk is:

Webhooklevering werkt goed samen met gekoppelde generaties: stel webhook in voor de laatste generatie en de hele keten draait server-side met één event aan het einde. Dit geldt ook wanneer de keten halverwege mislukt — de fout wordt doorgegeven aan de laatste generatie, die deze levert als een failed-event met de reden dependency_failed.

Webhookpayload

Een voltooide generatie levert de output-URL en het MIME-type:

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

Een mislukte generatie levert in plaats daarvan de foutcategorie en het bericht:

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

Vertak op data.status om te bepalen welke velden aanwezig zijn. De twee eindstatussen zijn de enige statussen die een webhook kan bevatten, omdat levering pas plaatsvindt wanneer een generatie is voltooid.

Het event verwerken

Een handler verifieert de handtekening, controleert het eventtype en vertakt vervolgens op data.status. Dit voorbeeld downloadt de output van een voltooide generatie en logt de reden van een mislukte generatie.

# 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 voorbeelden downloaden voor de beknoptheid binnen de request. Een grote video duurt lang genoeg dat dit de leveringstime-out kan overschrijden, dus stuur in productie de generatie-ID naar een queue en retourneer meteen 2xx. De ondertekende URL is ongeveer een uur geldig, wat ruim voldoende is voor een background worker.

De handtekening verifiëren

De handler hierboven roept construct_event / constructEvent aan. Hiermee worden de header ElevenLabs-Signature geverifieerd, de timestamp gevalideerd en de payload in één stap geparseerd. Verifieer altijd voordat je een event vertrouwt.

Het is belangrijk dat de listener alle binnenkomende webhooks valideert. Webhooks ondersteunen momenteel authenticatie via HMAC-handtekeningen. Stel HMAC-authenticatie als volgt in:

• Sla het gedeelde geheim dat is gegenereerd bij het maken van de webhook veilig op
• Verifieer de header ElevenLabs-Signature in je endpoint met de SDK

De JavaScript SDK biedt constructEvent; de Python SDK biedt construct_event met rawBody, sig_header en secret (in Python heten deze niet payload / signature). Beide verifiëren de handtekening, valideren de timestamp en parsen de JSON-payload.

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

Leveringsgedrag

Elke generatie levert precies één eindevent per gerichte webhook. De levering staat los van de generatie zelf: een webhook die mislukt of niet bereikbaar is, heeft geen invloed op het resultaat. Dit blijft beschikbaar via het GET-endpoint en in de lijstresponse.

Retourneer meteen een 2xx-status vanuit je handler. Herhaalde fouten schakelen een webhook automatisch uit, en een uitgeschakelde webhook zorgt ervoor dat volgende generaties die erop zijn gericht bij het maken worden geweigerd. Ontwerp de handler idempotent en gebruik de generatie-id om duplicaten te verwijderen.

Als een gemist resultaat niet acceptabel is voor je workflow, gebruik webhooks dan als de snelle route en vergelijk periodiek met flows.image.list of flows.video.list, gefilterd op status.

Volgende stappen