Webhooks för Image & Video

Översikt

Videogenereringar kan ta flera minuter, vilket gör det kostsamt att hålla polling öppen. Välj webhook-leverans för en generering så skickar ElevenLabs en flows_generation-händelse till din slutpunkt när genereringen får statusen completed eller failed.

Händelsens payload är det slutliga svaret från motsvarande GET-slutpunkt, så en hanterare som redan förstår polling-svaret behöver ingen separat sökväg för tolkning.

Innan du börjar

Webhook-leverans använder de webhooks som din arbetsyta prenumererar med på genereringshändelser. Att konfigurera en sker i två steg: skapa webhooken och prenumerera sedan på händelsen.

{
  "events": ["flows"]
}

Att skapa och prenumerera på webhooks kräver behörigheten Webhooks Manage eller att du är arbetsyteadministratör. En enskild händelse kan ha upp till 10 webhooks; därefter misslyckas begäran med too_many_webhooks.

En generering som begär webhook-leverans när ingen webhook prenumererar på genereringshändelser avvisas, så ett resultat genereras aldrig utan någonstans att leverera det.

Begär webhook-leverans

Lägg till ett webhook-objekt i begäran för att skapa. Använd {"type": "all"} för att leverera till varje webhook som prenumererar på genereringshändelser, så förblir begäran stabil när webhooks läggs till eller ersätts.

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 du i stället vill rikta in dig på specifika webhooks anger du fältet webhook som en lista med ID:n. Varje ID måste tillhöra någon av arbetsytans webhooks som prenumererar på genereringshändelser.

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

Begäran för att skapa validerar målet innan genereringen startar och returnerar ett fel när leverans inte skulle vara möjlig:

Webhook-leverans fungerar bra med kedjade genereringar: ange webhook för den sista genereringen så körs hela kedjan på serversidan med en enda händelse i slutet. Detta gäller även om kedjan misslyckas halvvägs – felet fortplantas till den sista genereringen, som levererar det som en failed-händelse med orsaken dependency_failed.

Webhook-payload

En slutförd generering levererar utdata-URL:en och MIME-typen:

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

En misslyckad generering levererar i stället felkategorin och meddelandet:

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

Förgrena på data.status för att avgöra vilka fält som finns. De två slutliga statusarna är de enda som en webhook kan innehålla, eftersom leverans endast sker när en generering är klar.

Hantera händelsen

En hanterare verifierar signaturen, kontrollerar händelsetypen och förgrenar sedan på data.status. Det här exemplet laddar ned utdata från en slutförd generering och loggar orsaken för en misslyckad.

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

Båda exemplen laddar ned under begäran för enkelhetens skull. En stor video kan ta så lång tid att detta överskrider leveranstimeouten, så i produktion bör du lämna genererings-ID:t till en kö och returnera 2xx omedelbart. Den signerade URL:en är giltig i ungefär en timme, vilket ger gott om tid för en bakgrundsarbetare.

Verifiera signaturen

Hanteraren ovan anropar construct_event / constructEvent, vilket verifierar huvudet ElevenLabs-Signature, validerar tidsstämpeln och tolkar payloaden i ett steg. Verifiera alltid innan du litar på en händelse.

Det är viktigt att mottagaren validerar alla inkommande webhooks. Webhooks har för närvarande stöd för autentisering via HMAC-signaturer. Konfigurera HMAC-autentisering genom att:

• Lagra den delade hemligheten som genereras när webhooken skapas på ett säkert sätt
• Verifiera headern ElevenLabs-Signature i din endpoint med hjälp av SDK:n

JavaScript-SDK:n exponerar constructEvent; Python-SDK:n exponerar construct_event med rawBody, sig_header och secret (dessa heter inte payload / signature i Python). Båda verifierar signaturen, validerar tidsstämpeln och tolkar JSON-payloaden.

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

Leveransbeteende

Varje generering levererar exakt en slutlig händelse per riktad webhook. Leveransen är oberoende av själva genereringen: en webhook som misslyckas eller inte kan nås påverkar inte resultatet, som fortsätter att vara tillgängligt från GET-slutpunkten och i listsvar.

Returnera snabbt status 2xx från din hanterare. Upprepade fel inaktiverar automatiskt en webhook, och en inaktiverad webhook gör att efterföljande genereringar som riktas till den avvisas när de skapas. Utforma hanteraren så att den är idempotent och använd genereringens id för att ta bort dubbletter.

För workflows där ett missat resultat inte är acceptabelt bör du behandla webhooks som den snabba vägen och stämma av regelbundet med flows.image.list eller flows.video.list, filtrerat på status.

Nästa steg