Gemini API 现已支持 Webhooks：告别 Veo 和批处理流水线中的轮询

while not operation.done: sleep(5) 这段代码，凡是构建过视频生成或批处理任务的团队都不陌生。它能用，但它是反模式：阻塞线程、在状态检查上浪费 API 调用，并引入与轮询间隔成正比的人为延迟。随着今日 Google AI Studio 的发布，这段代码可以——也应该——被淘汰。

发布内容

Google 发布了 Gemini API Webhook 支持，覆盖三类事件：

• 视频生成完成 — video.generated，支持 Veo（包括 veo-3.1-generate-preview）
• 批处理任务完成 — 批量处理任务结束时的通知
• 智能体工作流需要关注 — 智能体流水线中 human-in-the-loop 的触发器

技术规范采用 Standard Webhooks（standardwebhooks.com）：HMAC 签名 payload 的 HTTP POST 请求，webhook-id + webhook-timestamp + webhook-signature 请求头，以及指数退避重试的至少一次交付语义。

两种注册模式

API 支持两种配置模式：

静态（项目级）： 通过 Google Cloud Console 或管理 API 配置。项目中的所有操作都向已注册的端点发送通知。适合拥有集中式接收器的生产流水线。

动态（按请求）： 在 API 调用中内联传递。每个任务可以指定自己的端点——适用于每个客户有独立接收器的多租户架构，或测试流程。

Python SDK：代码对比

之前（轮询）：

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

# 轮询循环——阻塞线程，浪费 API 调用
while not operation.done:
    time.sleep(10)
    operation = client.operations.get(operation)

video = operation.result.generated_videos[0]

之后（webhook）：

from google import genai
from google.genai import types

client = genai.Client()

# 动态注册：任务完成时通知其接收器
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://your-api.com/webhooks/gemini",
            subscribed_events=["video.generated"],
        )
    ),
)

# 立即返回——无阻塞
print(f"任务已启动: {operation.name}")

接收器获得带签名 payload 的 HTTP POST：

# FastAPI 接收器（最简示例）
from fastapi import FastAPI, Request, Header
import hmac, hashlib, base64

app = FastAPI()

WEBHOOK_SECRET = "your-secret-here"

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

    # 签名验证（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"]
        # 触发下游：保存到 GCS、通知用户等

    return {"ok": True}

为什么这改变了架构

轮询曾是一种便利的权宜之计，在原型阶段可行，但无法规模化。它在生产环境中制造的三个真实问题：

1. 线程阻塞。 在同步 Python 中，每个进行中的任务都占用一个等待中的线程。50 个并发生成任务意味着你需要 50 个工作线程——仅仅用来等待。

2. 人为延迟。 10 秒的轮询间隔意味着视频完成后平均再等 5 秒才能交付。使用 Webhook，通知在完成后毫秒内到达。

3. 重启时的脆弱性。 如果运行轮询循环的进程崩溃，操作状态就丢失了。使用 Webhook，接收器是无状态的——如果没收到 2xx，Google 会重新投递。

Webhook 不只是更快的通知——它是控制反转的契约。模型调用你；你不需要一直调用模型。

Standard Webhooks：对平台团队真正重要的细节

Gemini API 采用 Standard Webhooks 规范 不是小事。这意味着你已经为 Stripe 或 GitHub 事件编写的签名验证代码在这里同样适用——只需替换密钥和请求头字段名。

对于已有事件摄取基础设施（FastAPI/Express 接收器、前置 SQS 或 Pub/Sub 队列）的团队，集成 Gemini 无需新的验证代码。只需注册端点并为 Gemini 事件类型添加处理器。

对智能体流水线的影响

除了 Veo 和批处理任务，此次发布还涵盖第三类事件：agent.needs_attention——当智能体工作流到达需要人工干预的决策点时触发。

这正是生产环境中 human-in-the-loop 缺少的原语。此前，你必须从头构建审批队列（数据库轮询、WebSocket 或 long-poll HTTP）。有了原生事件，智能体主动推送通知给你——审核人员可以通过 API 响应，然后智能体继续执行。

对于构建多步骤审批流水线（KYC、合同、风险分析）的团队，这显著降低了编排层的复杂度。

延伸阅读