Webhooks de Imagem e Vídeo

Visão geral

As gerações de vídeo podem levar vários minutos, o que torna caro manter o polling aberto. Habilite a entrega por webhook em uma geração, e a ElevenLabs envia um evento flows_generation ao seu endpoint quando a geração chega a completed ou failed.

A carga do evento é a resposta final do endpoint GET correspondente, então um manipulador que já entende a resposta de polling não precisa de um caminho de análise separado.

Antes de começar

A entrega por webhook usa os webhooks do seu workspace inscritos em eventos de geração. A configuração tem duas etapas: criar o webhook e depois inscrevê-lo no evento.

{
  "events": ["flows"]
}

Criar e inscrever webhooks exige a permissão Gerenciar webhooks ou ser administrador do workspace. Um único evento aceita até 10 webhooks; acima disso, a solicitação falha com too_many_webhooks.

Uma geração que solicita entrega por webhook quando não há nenhum webhook inscrito em eventos de geração é rejeitada, portanto um resultado nunca é gerado sem ter para onde ser entregue.

Solicitar entrega por webhook

Adicione um objeto webhook à solicitação de criação. Use {"type": "all"} para entregar a todos os webhooks inscritos em eventos de geração, o que mantém a solicitação estável à medida que webhooks são adicionados ou substituídos.

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 direcionar webhooks específicos, defina o campo webhook como uma lista de IDs. Cada ID deve ser de um dos webhooks do workspace inscritos em eventos de geração.

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

A solicitação de criação valida o destino antes de iniciar a geração e retorna um erro quando a entrega não seria possível:

A entrega por webhook funciona bem com gerações encadeadas: defina webhook na geração final, e toda a cadeia será executada no servidor com um único evento no final. Isso também vale quando a cadeia falha no meio do processo — a falha se propaga para a geração final, que a entrega como um evento failed com o motivo dependency_failed.

Carga do webhook

Uma geração concluída entrega a URL de saída e o 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"
  }
}

Uma geração com falha entrega a categoria e a mensagem de falha:

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

Verifique data.status para decidir quais campos estão presentes. Os dois status finais são os únicos que um webhook pode transportar, já que a entrega ocorre somente quando uma geração é concluída.

Processar o evento

Um manipulador verifica a assinatura, confere o tipo do evento e depois verifica data.status. Este exemplo baixa a saída de uma geração concluída e registra o motivo de uma falha.

# 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 os exemplos fazem o download durante a solicitação para manter a brevidade. Um vídeo grande pode levar tempo suficiente para ultrapassar o tempo limite de entrega, então, em produção, envie o ID da geração para uma fila e retorne 2xx imediatamente. A URL assinada é válida por cerca de uma hora, tempo suficiente para um worker em segundo plano.

Verificar a assinatura

O manipulador acima chama construct_event / constructEvent, que verifica o cabeçalho ElevenLabs-Signature, valida o timestamp e analisa a carga em uma única etapa. Sempre faça a verificação antes de confiar em um evento.

É importante que o listener valide todos os webhooks recebidos. Atualmente, os webhooks oferecem suporte à autenticação por assinaturas HMAC. Configure a autenticação HMAC:

• Armazenando com segurança o segredo compartilhado gerado na criação do webhook
• Verificando o cabeçalho ElevenLabs-Signature no seu endpoint usando o SDK

O SDK JavaScript disponibiliza constructEvent; o SDK Python disponibiliza construct_event com rawBody, sig_header e secret (em Python, eles não se chamam payload / signature). Ambos verificam a assinatura, validam o carimbo de data e hora e analisam o 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 de entrega

Cada geração entrega exatamente um evento final por webhook direcionado. A entrega é independente da própria geração: um webhook que falha ou está inacessível não afeta o resultado, que continua disponível no endpoint GET e na resposta de listagem.

Retorne um status 2xx rapidamente no seu manipulador. Falhas repetidas desativam automaticamente um webhook, e um webhook desativado faz com que gerações subsequentes que o direcionam sejam rejeitadas na criação. Projete o manipulador para ser idempotente e use o id da geração para eliminar duplicatas.

Para workflows em que um resultado perdido não é aceitável, trate os webhooks como o caminho rápido e faça reconciliações periódicas com flows.image.list ou flows.video.list, filtrando por status.

Próximas etapas