Webhooks de Imagen y Video

Descripción general

Las generaciones de vídeo pueden tardar varios minutos, por lo que mantener una consulta periódica abierta resulta costoso. Activa la entrega por webhook para una generación y ElevenLabs envía un evento flows_generation a tu ruta cuando la generación alcanza el estado completed o failed.

La carga útil del evento es la respuesta final de la ruta GET correspondiente, así que un gestor que ya entiende la respuesta de consulta periódica no necesita una ruta de análisis independiente.

Antes de empezar

La entrega por webhook utiliza los webhooks de tu espacio de trabajo suscritos a eventos de generación. Configurarlo requiere dos pasos: crear el webhook y suscribirlo al evento.

{
  "events": ["flows"]
}

Crear y suscribir webhooks requiere el permiso Gestionar webhooks o ser administrador del espacio de trabajo. Un único evento admite hasta 10 webhooks; a partir de ese límite, la solicitud falla con too_many_webhooks.

Una generación que solicita entrega por webhook cuando no hay ningún webhook suscrito a eventos de generación se rechaza, por lo que nunca se genera un resultado sin un lugar donde entregarlo.

Solicita la entrega por webhook

Añade un objeto webhook a la solicitud de creación. Usa {"type": "all"} para entregar a todos los webhooks suscritos a eventos de generación, lo que mantiene estable la solicitud aunque se añadan o sustituyan webhooks.

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

Para dirigirte a webhooks específicos, establece en su lugar el campo webhook como una lista de ID. Cada ID debe pertenecer a uno de los webhooks del espacio de trabajo suscritos a eventos de generación.

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

La solicitud de creación valida el destino antes de iniciar la generación y devuelve un error cuando no sería posible entregar el resultado:

La entrega por webhook combina bien con las generaciones en cadena: establece webhook en la generación final y toda la cadena se ejecutará en el servidor con un único evento al final. Esto también se cumple cuando la cadena falla a mitad de camino: el fallo se propaga hasta la generación final, que lo entrega como un evento failed con el motivo dependency_failed.

Carga útil del webhook

Una generación completada entrega la URL de salida y el 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 generación fallida entrega en su lugar la categoría y el mensaje del fallo:

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

Comprueba data.status para decidir qué campos están presentes. Los dos estados finales son los únicos que puede incluir un webhook, ya que la entrega solo se produce cuando termina una generación.

Gestiona el evento

Un gestor verifica la firma, comprueba el tipo de evento y, después, evalúa data.status. Este ejemplo descarga la salida de una generación completada y registra el motivo de una fallida.

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

Ambos ejemplos descargan dentro de la solicitud para simplificar. Un vídeo grande tarda lo suficiente como para superar el tiempo de espera de entrega, así que en producción envía el ID de generación a una cola y devuelve 2xx inmediatamente. La URL firmada es válida durante aproximadamente una hora, tiempo de sobra para un proceso en segundo plano.

Verifica la firma

El gestor anterior llama a construct_event / constructEvent, que verifica la cabecera ElevenLabs-Signature, valida la marca de tiempo y analiza la carga útil en un solo paso. Verifica siempre un evento antes de confiar en él.

Es importante que el listener valide todos los webhooks entrantes. Actualmente, los webhooks admiten autenticación mediante firmas HMAC. Configura la autenticación HMAC de esta forma:

• Almacena de forma segura el secreto compartido generado al crear el webhook
• Verifica la cabecera ElevenLabs-Signature en tu ruta de API mediante el SDK

El SDK de JavaScript incluye constructEvent; el SDK de Python incluye construct_event con rawBody, sig_header y secret (en Python no se llaman payload / signature). Ambos verifican la firma, validan la marca de tiempo y analizan la carga útil 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"}

Comportamiento de entrega

Cada generación entrega exactamente un evento final por cada webhook de destino. La entrega es independiente de la generación: un webhook que falla o no es accesible no afecta al resultado, que sigue disponible desde la ruta GET y en la respuesta de lista.

Devuelve enseguida un estado 2xx desde tu gestor. Los fallos repetidos desactivan automáticamente un webhook, y un webhook desactivado hace que las generaciones posteriores que lo tienen como destino se rechacen al crearlas. Diseña el gestor para que sea idempotente y usa el id de generación para eliminar duplicados.

En flujos de trabajo donde no sea aceptable perder un resultado, trata los webhooks como la vía rápida y reconcilia periódicamente con flows.image.list o flows.video.list, filtrando por status.

Siguientes pasos