Webhooks.ipynb：使用Google Colab构建Webhook的完整指南

本教程面向有编程基础的开发者，深入讲解如何利用Gemini API的Webhooks功能实现异步任务完成时的实时通知。文章从背景出发，解释了Webhooks相较于轮询的优势，并详细介绍了静态与动态两种配置方式。通过Google Cloud Run函数搭建接收端点，实现签名验证，并演示了视频生成等长时任务的触发流程。教程还涵盖了Webhook的创建、列出、更新、删除等管理操作，以及动态Webhook的配置要点。最后提供了常见错误排查和生产环境优化建议，帮助开发者构建可靠、安全的Webhook服务。

技术背景

在AI应用中，许多操作如视频生成、批处理任务等耗时较长，传统做法是客户端轮询服务器状态，这不仅浪费资源，还增加延迟。Webhooks提供了一种事件驱动的解决方案：当任务完成时，服务器主动向预设的URL发送HTTP POST请求，实现实时通知。Gemini API的Webhooks支持静态（项目级）和动态（请求级）两种配置，适用于需要异步通知的场景，如批处理作业、交互式任务和视频生成。其主要限制包括：端点必须在几秒内响应2xx，否则会触发重试；签名密钥仅显示一次，需妥善保管；目前处于预览阶段，功能可能变化。

核心原理

Webhooks的核心原理是事件驱动：客户端在发起长时操作时，可配置一个回调URL（Webhook端点）。当操作完成时，Gemini API会向该URL发送一个HTTP POST请求，请求体包含事件类型和操作结果数据。为确保安全，请求带有签名头（webhook-id、webhook-signature、webhook-timestamp），接收方需使用签名密钥验证请求真实性。签名验证遵循Standard Webhooks规范，防止伪造请求。数据流为：客户端发起操作（如generate_videos）→ Gemini API处理 → 操作完成 → 发送POST到Webhook端点 → 端点验证签名并处理事件。

架构分析

系统架构分为三部分：客户端（Colab笔记本）、Gemini API服务、Webhook接收端点（Cloud Run函数）。客户端负责发起操作并管理Webhook配置；Gemini API负责执行长时任务并触发Webhook；Cloud Run函数作为无服务器端点，接收POST请求，验证签名，处理事件。设计决策包括：使用Cloud Run函数实现无服务器，易于扩展且成本低；签名验证使用standardwebhooks库，确保安全性；端点需快速响应（2xx）以避免重试。动态Webhook允许在请求级别覆盖URL，实现更灵活的路由。

完整实战教程

#@title Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

Webhooks

<a target="_blank" href="https://colab.research.google.com/github/google-gemini/cookbook/blob/main/quickstarts/Webhooks.ipynb">在 Colab 中打开</a>

Webhooks 允许你在特定事件发生时接收实时通知。本教程将演示如何使用 Gemini API 设置 webhook，以便在模型生成内容时接收通知。

设置

首先，安装必要的库并配置 API 密钥。

</a>

预览：Gemini API 中的 Webhooks 支持目前处于预览阶段。

Webhooks 允许 Gemini API 在异步或长时间运行的操作（LRO）完成时，将实时通知推送到您的服务器。这取代了轮询 API 以获取状态更新的需要，从而减少了延迟和开销。

Webhooks 可用于诸如批处理作业、交互和视频生成等操作。

工作原理

无需反复轮询 GET /operations 来检查作业是否完成，您可以配置 Gemini API Webhooks，在事件触发时立即向您的监听器 URL 发送 HTTP POST 请求。

Gemini API 支持两种配置 Webhooks 的方式：

* 静态 Webhooks：项目级端点。适用于全局集成（例如，通知 Slack、同步数据库等）。

* 动态 Webhooks：请求级覆盖，在特定作业调用的配置负载中传递 Webhook URL。适用于将特定作业路由到专用端点。

设置

安装 SDK

从 PyPI 安装 SDK。

%pip install -U -q google-genai # If you see an error related to google-auth version mismatch, re-run the cell

设置你的 API 密钥

要运行以下单元格，你的 API 密钥必须存储在名为 GEMINI_API_KEY 的 Colab 密钥中。如果你还没有 API 密钥，或者不确定如何创建 Colab 密钥，请参阅 [身份验证]

](../quickstarts/Authentication.ipynb) 查看演练。

from google.colab import userdata

GEMINI_API_KEY = userdata.get('GEMINI_API_KEY')

初始化 SDK 客户端

使用你的 API 密钥初始化客户端：

import os
from google import genai
from google.genai import errors, types

client = genai.Client(api_key=GEMINI_API_KEY)

步骤 1：创建 Webhook 端点

本示例使用 Google Cloud Run 函数，它允许你轻松部署一个无服务器函数，用作我们的 Webhook 端点。

前往 https://console.cloud.google.com/run/
在“编写函数”下点击 NodeJS
输入名称，如“gemini-webhook-receiver”，并在“身份验证”下选择“允许公开访问”
点击“部署”。如果这是你首次部署 Cloud Run 函数，你可能需要按提示启用 Cloud Build 等 API。

（其管理界面 https://console.cloud.google.com/run/ 有日志观察页面，你可以在那里查看请求）

实现 Webhook 签名验证：在“源代码”选项卡中，更新 package.json 的依赖项，添加：

"dependencies": {
"standardwebhooks": "^1.0.0"
}

将函数入口点更新为 geminiWebhook

并将 /src/index.js 中的代码替换为以下代码，然后点击“保存并重新部署”

const functions = require('@google-cloud/functions-framework');
const { Webhook } = require('standardwebhooks');

/**
* Google Cloud Function to handle Gemini API Webhooks
*/
functions.http('geminiWebhook', async (req, res) => {
// Respond with a 200 for non-POST requests to indicate that the function is running
if (req.method !== 'POST') {
return res.status(200).send('Worker is running');
}

// Print headers and body to stdout
console.log('--- Webhook Headers ---');
console.log(JSON.stringify(req.headers, null, 2));
console.log('--- Webhook Body ---');
console.log(JSON.stringify(req.body, null, 2));

const payload = JSON.stringify(req.body);

// Extract specific headers for Standard Webhooks verification
// This avoids issues with extra headers added by Cloud Run/Functions proxy
const headers = {
'webhook-id': req.headers['webhook-id'],
'webhook-signature': req.headers['webhook-signature'],
'webhook-timestamp': req.headers['webhook-timestamp'],
};

const secret = process.env.WEBHOOK_SIGNING_SECRET;

try {
const wh = new Webhook(secret);
// Verify the webhook signature using only the relevant headers
const event = wh.verify(payload, headers);

console.log(`Event type: ${event.type}, data: ${JSON.stringify(event.data)}`);

if (event.type === 'batch.succeeded') {
console.log(`Batch completed! ID: ${event.data.id}`);
if (event.data.output_file_uri) {
console.log(`Batch file: ${event.data.output_file_uri}`);
}
} else if (event.type === 'video.generated') {
console.log(`Video generated! URI: ${event.data.video_uri}`);
}

res.status(200).json({ status: 'received' });
} catch (err) {
console.error('Webhook verification failed:', err.message);
res.status(400).json({ error: 'Signature invalid' });
}
});

当您订阅的事件发生时，您的 webhook URL 将收到一个 HTTP POST 请求。您的端点必须在几秒内以 2xx 状态码响应，以避免重试。为确保送达，Gemini API 会在 24 小时内使用指数退避自动重试失败的请求。

TODO：将其改为使用你自己的 webhook 端点！

# TODO: Copy and paste the URL for the function you just deployed above
WEBHOOK_ENDPOINT="https://your-app.run.app"  # @param {type:"string"}

检查端点是否正常工作。这个示例端点仅对 GET 请求返回一段文本。

import requests

r = requests.get(WEBHOOK_ENDPOINT)
print(r.text)

步骤 2：创建 Webhook 并将密钥添加到 Cloud Run 函数

重要提示：创建 Webhook 时，API 会返回一个签名密钥，且仅显示一次。请将下面的密钥添加到你的 Cloud Run 函数中。

步骤

在 Google Cloud Console 中打开你的函数，点击“修改并部署新修订版本”。
然后，在“容器”>“变量和密钥”选项卡中，点击“添加变量”。
然后，输入密钥名称“WEBHOOK_SIGNING_SECRET”，并设置其值如下。
然后点击“部署”。

webhook = client.webhooks.create(
    subscribed_events=["video.generated"],
    name="video_generation_webhook",
    uri=WEBHOOK_ENDPOINT,
)

# Store webhook.signing_secret securely (e.g., in your env vars)
os.environ["WEBHOOK_SIGNING_SECRET"] = webhook.new_signing_secret

# print(f"WEBHOOK_SIGNING_SECRET={webhook.new_signing_secret}")
# # Or write it to a file:
# with open(".env", "a") as f:
#     f.write(f"WEBHOOK_SIGNING_SECRET={webhook.new_signing_secret}\n")

print(f"Created webhook: {webhook.name}")

列出 Webhook

列出当前项目配置的所有 Webhook，支持可选的分页。

client.webhooks.list()

Get/Ping/Delete/Update

发送 Webhook 请求

client.webhooks.ping(id=webhook.id)

获取一个 webhook

wh = client.webhooks.get(id=webhook.id)
wh

这会删除该 webhook。设置 delete=True 来运行它：

delete = False
if delete:
    client.webhooks.delete(id=webhook.id)

这就是更新 webhook 的方法。设置 update=True 即可运行它：

update = False
if update:
    wh = client.webhooks.update(
        id=webhook.id,
        subscribed_events=["video.generated", "batch.succeeded"],
        name="video_and_batch_webhook",
    )

步骤 4：发送 Gemini API 请求

为你的 webhook 所订阅的长时间运行操作发送一个请求。

prompt = """A stunning drone view of the Grand Canyon during a flamboyant sunset that highlights the canyon's colors. The drone slowly flies towards the sun then accelerates, dives and flies inside the canyon."""

operation = client.models.generate_videos(
    model="veo-3.1-generate-preview",
    prompt=prompt,
)

现在，你的 webhook 不再需要轮询 GET /operations，而是在视频生成完成后收到一个 HTTP POST 请求。

步骤 5：检查 Worker 日志

在 Cloud Run 控制台的 可观测性 > 日志 标签页中查看日志

日志中应显示 Webhook 负载和函数的输出，在本例中应显示生成的视频的 URL。

动态 Webhook 配置

你还可以使用动态 webhook 配置，但为此，你需要更新你的 webhook 验证。

有关实现细节，请参阅文档：

https://ai.google.dev/gemini-api/docs/webhooks#dynamic-webhooks
https://ai.google.dev/gemini-api/docs/webhooks#verify-dynamic-signatures

后续步骤：

Webhooks 文档：阅读完整的 Webhooks 指南，了解高级配置选项、支持的事件类型和最佳实践。
Batch API：Webhooks 与 Batch API 配合使用效果很好，可以处理大量请求并在完成时收到通知。
视频生成：将 Webhooks 与 Veo 结合使用，在生成的视频准备好时收到通知。
Cloud Run 函数：了解有关 Cloud Run 函数的更多信息。
标准 Webhooks：Gemini API 使用标准 Webhooks 规范进行签名验证，确保安全且可互操作的 webhook 传递。

常见问题

常见错误：1) 签名验证失败：检查WEBHOOK_SIGNING_SECRET是否与创建Webhook时返回的密钥一致，注意密钥仅显示一次；确保提取的headers正确，避免代理添加额外头。2) 端点未收到请求：确认Webhook已创建且订阅了正确事件；检查Cloud Run函数是否部署成功且允许公开访问；查看日志确认请求是否到达。3) 响应超时：确保端点快速返回2xx，否则会触发重试。4) 依赖版本冲突：安装google-genai时若出现google-auth版本错误，重新运行安装单元。排查方法：利用Cloud Run日志查看请求和错误；使用ping方法测试Webhook；检查网络和防火墙设置。

企业应用扩展

生产环境优化：1) 使用密钥管理服务（如Google Cloud Secret Manager）存储签名密钥，避免硬编码。2) 实现幂等处理，防止重复事件导致重复操作。3) 增加监控和告警，跟踪Webhook失败率和延迟。4) 考虑使用消息队列（如Pub/Sub）缓冲事件，提高可靠性。5) 动态Webhook可结合请求上下文路由到不同处理逻辑。6) 对于高吞吐场景，可水平扩展Cloud Run实例。7) 定期轮换签名密钥，增强安全性。8) 利用Standard Webhooks规范的其他特性，如重放保护。

总结

内容来源与验证

内容等级：PREMIUM_A
质量等级：A
难度：中级
验证等级：V1
原文作者：
原文日期：
来源许可：Apache-2.0
来源版本：4dc9d8b9b8aab6af84987a8a8a104e0f5ff86893
查看原文

继续系统学习

站内还有更多完整实战教程

加入会员，解锁全部教程、完整代码与持续更新内容。

加入会员