Webhook di Immagini e Video

Panoramica

La generazione di video può richiedere diversi minuti, rendendo costoso mantenere aperto il polling. Configura una generazione per la consegna tramite webhook e ElevenLabs invierà un evento flows_generation al tuo endpoint quando la generazione raggiunge lo stato completed o failed.

Il payload dell’evento corrisponde alla risposta finale del relativo endpoint GET, quindi un handler che comprende già la risposta del polling non richiede un percorso di parsing separato.

Prima di iniziare

La consegna tramite webhook usa i webhook del tuo workspace iscritti agli eventi di generazione. La configurazione richiede due passaggi: crea il webhook, quindi iscrivilo all’evento.

{
  "events": ["flows"]
}

Per creare e iscrivere webhook sono necessari l’autorizzazione Gestione webhook o il ruolo di amministratore del workspace. Un singolo evento accetta fino a 10 webhook; oltre questo limite, la richiesta fallisce con too_many_webhooks.

Una generazione che richiede la consegna tramite webhook quando nessun webhook è iscritto agli eventi di generazione viene rifiutata, così non viene mai generato un risultato senza una destinazione a cui consegnarlo.

Richiedi la consegna tramite webhook

Aggiungi un oggetto webhook alla richiesta di creazione. Usa {"type": "all"} per consegnare a ogni webhook iscritto agli eventi di generazione: in questo modo la richiesta resta invariata quando i webhook vengono aggiunti o sostituiti.

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

Per indirizzare webhook specifici, imposta invece il campo webhook su un elenco di ID. Ogni ID deve corrispondere a uno dei webhook del workspace iscritti agli eventi di generazione.

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

La richiesta di creazione convalida la destinazione prima di avviare la generazione e restituisce un errore quando la consegna non sarebbe possibile:

La consegna tramite webhook si abbina bene alle generazioni concatenate: imposta webhook sulla generazione finale e l’intera catena viene eseguita lato server con un solo evento alla fine. Vale anche quando la catena fallisce a metà: l’errore si propaga alla generazione finale, che lo consegna come evento failed con motivo dependency_failed.

Payload del webhook

Una generazione completata fornisce l’URL di output e il tipo 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"
  }
}

Una generazione non riuscita fornisce invece la categoria e il messaggio dell’errore:

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

Verifica data.status per stabilire quali campi sono presenti. I due stati finali sono gli unici che un webhook può contenere, poiché la consegna avviene solo al termine di una generazione.

Gestisci l’evento

Un handler verifica la firma, controlla il tipo di evento e poi dirama in base a data.status. Questo esempio scarica l’output di una generazione completata e registra il motivo di una generazione non riuscita.

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

Entrambi gli esempi scaricano durante la richiesta per semplicità. Un video di grandi dimensioni richiede abbastanza tempo da superare il timeout di consegna, quindi in produzione invia l’ID della generazione a una coda e restituisci subito 2xx. L’URL firmato è valido per circa un’ora, un tempo più che sufficiente per un worker in background.

Verifica la firma

L’handler sopra chiama construct_event / constructEvent, che verifica l’header ElevenLabs-Signature, convalida il timestamp e analizza il payload in un solo passaggio. Verifica sempre un evento prima di considerarlo attendibile.

È importante che il listener convalidi tutti i webhook in arrivo. I webhook supportano attualmente l’autenticazione tramite firme HMAC. Configura l’autenticazione HMAC:

• Archiviando in modo sicuro il segreto condiviso generato alla creazione del webhook
• Verificando l’header ElevenLabs-Signature nel tuo endpoint tramite l’SDK

L’SDK JavaScript espone constructEvent; l’SDK Python espone construct_event con rawBody, sig_header e secret (in Python non si chiamano payload / signature). Entrambi verificano la firma, convalidano il timestamp e analizzano il payload 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"}

Comportamento di consegna

Ogni generazione consegna esattamente un evento finale per ogni webhook di destinazione. La consegna è indipendente dalla generazione: un webhook che fallisce o non è raggiungibile non influisce sul risultato, che resta disponibile dall’endpoint GET e nella risposta dell’elenco.

Restituisci tempestivamente uno stato 2xx dal tuo handler. Errori ripetuti disabilitano automaticamente un webhook e un webhook disabilitato fa sì che le generazioni successive che lo usano come destinazione vengano rifiutate al momento della creazione. Progetta l’handler in modo che sia idempotente e usa l’id della generazione per eliminare i duplicati.

Per workflow in cui non è accettabile perdere un risultato, considera i webhook come percorso rapido e riconcilia periodicamente con flows.image.list o flows.video.list, filtrando in base a status.

Passaggi successivi