Gemini API agora tem webhooks: fim do polling em pipelines Veo e batch

O padrão de polling while not operation.done: sleep(5) é um código que todo time que constrói com geração de vídeo ou batch jobs conhece bem. Funciona, mas é um antipadrão: bloqueia threads, desperdiça tokens de API em verificações de estado, e tem latência artificial proporcional ao intervalo de polling. Com o anúncio de hoje do Google AI Studio, esse código pode — e deve — ser aposentado.

O que foi anunciado

O Google lançou suporte a webhooks na Gemini API, cobrindo três classes de eventos:

• Geração de vídeo concluída — video.generated com Veo (incluindo veo-3.1-generate-preview)
• Batch job finalizado — notificação ao término de jobs de processamento em lote
• Workflow agêntico precisa de atenção — gatilho para human-in-the-loop em pipelines de agentes

A especificação técnica é a Standard Webhooks (standardwebhooks.com): requisições HTTP POST com payload assinado por HMAC, cabeçalho webhook-id + webhook-timestamp + webhook-signature, e semântica de entrega at-least-once com retry exponencial.

Dois modos de registro

A API suporta dois modos de configuração:

Estático (nível de projeto): configurado via Google Cloud Console ou API de administração. Todas as operações do projeto disparam notificações para o endpoint registrado. Ideal para pipelines de produção com receptor centralizado.

Dinâmico (por requisição): passado inline na chamada da API. Permite que cada job indique seu próprio endpoint — útil em arquiteturas multi-tenant onde cada cliente tem seu próprio receiver ou em flows de teste.

SDK Python: como fica o código

Antes (polling):

import time
from google import genai

client = genai.Client()
operation = client.models.generate_videos(
    model="veo-3.1-generate-preview",
    prompt="time-lapse of a server room at night",
)

# Loop de polling — bloqueia a thread, desperdiça chamadas
while not operation.done:
    time.sleep(10)
    operation = client.operations.get(operation)

video = operation.result.generated_videos[0]

Depois (webhook):

from google import genai
from google.genai import types

client = genai.Client()

# Registro dinâmico: o job notifica seu receptor ao concluir
operation = client.models.generate_videos(
    model="veo-3.1-generate-preview",
    prompt="time-lapse of a server room at night",
    config=types.GenerateVideoConfig(
        webhook=types.WebhookConfig(
            uri="https://sua-api.com/webhooks/gemini",
            subscribed_events=["video.generated"],
        )
    ),
)

# Retorna imediatamente — sem bloqueio
print(f"Job iniciado: {operation.name}")

O receptor recebe um HTTP POST com payload assinado:

# FastAPI receiver (exemplo mínimo)
from fastapi import FastAPI, Request, Header
import hmac, hashlib, base64

app = FastAPI()

WEBHOOK_SECRET = "seu-secret-aqui"

@app.post("/webhooks/gemini")
async def handle_gemini_event(
    request: Request,
    webhook_signature: str = Header(None),
):
    body = await request.body()

    # Verificação de assinatura (Standard Webhooks)
    expected = base64.b64encode(
        hmac.new(WEBHOOK_SECRET.encode(), body, hashlib.sha256).digest()
    ).decode()

    if not hmac.compare_digest(expected, webhook_signature.split(",")[-1]):
        return {"error": "invalid signature"}, 401

    payload = await request.json()
    event_type = payload["type"]  # "video.generated"

    if event_type == "video.generated":
        video_uri = payload["data"]["uri"]
        # dispara downstream: salva no GCS, notifica usuário, etc.

    return {"ok": True}

Por que isso muda a arquitetura

O polling era um hack de conveniência que funcionava para prototipagem mas não escalava. Três problemas reais que ele cria em produção:

1. Threads bloqueadas. Em Python síncrono, cada job em andamento segura uma thread esperando. Com 50 gerações simultâneas, você precisa de 50 workers apenas para ficar esperando.

2. Latência artificial. Intervalo de 10 segundos significa que você entrega o vídeo, em média, 5 segundos depois do necessário. Com webhooks, a notificação chega em milissegundos após a conclusão.

3. Fragilidade em reinicializações. Se o processo que está em loop cair, você perde o estado da operação. Com webhooks, o receptor é stateless — o Google reentrega se não receber 2xx.

Webhook não é só uma notificação mais rápida — é um contrato de inversão de controle. O modelo te chama; você não precisa ficar chamando o modelo.

Standard Webhooks: o detalhe que importa para times de plataforma

O fato de a Gemini API ter aderido à Standard Webhooks spec não é detalhe menor. Significa que o mesmo código de validação de assinatura que você já usa para eventos do Stripe ou do GitHub funciona aqui — basta trocar o secret e o campo de cabeçalho.

Para times que já têm infraestrutura de ingestão de eventos (um receiver FastAPI/Express, fila SQS ou Pub/Sub na frente), integrar Gemini não exige código novo de verificação. É só registrar o endpoint e adicionar o handler para os event types do Gemini.

Implicações para pipelines agênticos

Além de Veo e batch jobs, o anúncio cobre um terceiro evento: agent.needs_attention — disparado quando um workflow agêntico atinge um ponto de decisão que requer intervenção humana.

Esse é o primitivo que faltava para human-in-the-loop em produção. Antes, você precisava implementar uma fila de aprovação do zero (polling de banco de dados, WebSocket, ou long-polling HTTP). Com o evento nativo, o agente empurra a notificação para você — e o reviewer humano pode responder via API antes que o agente continue.

Para quem está construindo pipelines de aprovação multi-step (KYC, contratos, análise de risco), isso reduz significativamente a complexidade da camada de orquestração.

Leia também