이미지 & 비디오 웹훅

개요

비디오 생성에는 몇 분이 걸릴 수 있어 폴링 연결을 계속 유지하는 비용이 큽니다. 생성 요청에서 웹훅 전송을 선택하면 생성이 completed 또는 failed 상태가 되었을 때 ElevenLabs가 엔드포인트로 flows_generation 이벤트를 전송합니다.

이벤트 페이로드는 해당 GET 엔드포인트의 최종 응답이므로, 이미 폴링 응답을 처리하는 핸들러에는 별도의 파싱 경로가 필요하지 않습니다.

시작하기 전에

웹훅 전송은 워크스페이스에서 생성 이벤트를 구독하도록 설정한 웹훅을 사용합니다. 설정은 두 단계로 이루어집니다. 먼저 웹훅을 만들고, 그다음 이벤트를 구독합니다.

{
  "events": ["flows"]
}

웹훅을 만들고 구독하려면 웹훅 관리 권한 또는 워크스페이스 관리자 권한이 필요합니다. 단일 이벤트에는 최대 10개의 웹훅을 연결할 수 있으며, 이를 초과하면 요청이 too_many_webhooks로 실패합니다.

생성 이벤트를 구독한 웹훅이 없는데 웹훅 전송을 요청하면 생성 요청이 거부됩니다. 따라서 전송할 곳 없는 결과가 생성되는 일은 없습니다.

웹훅 전송 요청

생성 요청에 webhook 객체를 추가하세요. 생성 이벤트를 구독한 모든 웹훅에 전송하려면 {"type": "all"}을 사용하세요. 이렇게 하면 웹훅이 추가되거나 교체되어도 요청을 안정적으로 유지할 수 있습니다.

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

특정 웹훅만 대상으로 지정하려면 webhook 필드를 ID 목록으로 설정하세요. 각 ID는 생성 이벤트를 구독한 워크스페이스의 웹훅이어야 합니다.

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

생성 요청은 생성을 시작하기 전에 대상을 검증하며, 전송할 수 없는 경우 오류를 반환합니다.

웹훅 전송은 연결된 생성과 함께 사용하기 좋습니다. 최종 생성에 webhook을 설정하면 전체 체인이 서버 측에서 실행되고 마지막에 단일 이벤트가 전송됩니다. 체인이 중간에 실패하는 경우에도 동일하게 적용됩니다. 실패가 최종 생성까지 전파되고, 최종 생성은 dependency_failed 사유와 함께 이를 failed 이벤트로 전송합니다.

웹훅 페이로드

완료된 생성은 출력 URL과 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"
  }
}

실패한 생성은 대신 실패 카테고리와 메시지를 전송합니다.

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

data.status에 따라 분기하여 어떤 필드가 있는지 판단하세요. 웹훅은 생성이 완료될 때만 전송되므로 두 최종 상태만 포함할 수 있습니다.

이벤트 처리

핸들러는 서명을 검증하고 이벤트 유형을 확인한 다음 data.status에 따라 분기합니다. 이 예제는 완료된 생성의 출력을 다운로드하고 실패한 생성의 사유를 기록합니다.

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

두 예제 모두 간결성을 위해 요청 중에 다운로드합니다. 대용량 비디오는 전송 제한 시간을 초과할 만큼 오래 걸릴 수 있으므로, 프로덕션에서는 생성 ID를 큐에 전달하고 즉시 2xx를 반환하세요. 서명 URL은 약 1시간 동안 유효하므로 백그라운드 워커에 충분한 시간입니다.

서명 검증

위 핸들러는 construct_event / constructEvent를 호출합니다. 이 함수는 ElevenLabs-Signature 헤더를 검증하고 타임스탬프를 확인하며 페이로드를 한 번에 파싱합니다. 이벤트를 신뢰하기 전에 항상 검증하세요.

수신 측에서는 들어오는 모든 웹훅의 유효성을 검증하는 것이 중요합니다. 현재 웹훅은 HMAC 서명을 통한 인증을 지원합니다. 다음 방법으로 HMAC 인증을 설정하세요.

• 웹훅 생성 시 생성되는 공유 시크릿을 안전하게 저장합니다.
• SDK를 사용하여 엔드포인트에서 ElevenLabs-Signature 헤더를 검증합니다.

JavaScript SDK는 constructEvent를 제공하며, Python SDK는 rawBody, sig_header, **secret**을 사용하는 construct_event를 제공합니다(Python에서는 payload / signature이라는 이름을 사용하지 않습니다). 두 SDK 모두 서명을 검증하고 타임스탬프를 확인하며 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"}

전송 동작

각 생성은 대상 웹훅마다 정확히 하나의 최종 이벤트를 전송합니다. 전송은 생성 자체와 독립적입니다. 웹훅이 실패하거나 연결할 수 없어도 GET 엔드포인트와 목록 응답에서 계속 사용할 수 있는 결과에는 영향을 주지 않습니다.

핸들러에서 즉시 2xx 상태를 반환하세요. 반복된 실패는 웹훅을 자동으로 비활성화하며, 비활성화된 웹훅을 대상으로 하는 이후 생성은 생성 시점에 거부됩니다. 핸들러는 멱등적으로 설계하고 생성 id를 사용해 중복을 제거하세요.

결과 누락을 허용할 수 없는 워크플로에서는 웹훅을 빠른 경로로 처리하고, status를 기준으로 필터링하여 flows.image.list 또는 flows.video.list로 주기적으로 조정하세요.

다음 단계