Webhook

Webhook cho phép Gemini API gửi thông báo theo thời gian thực đến máy chủ của bạn khi các Thao tác không đồng bộ hoặc Thao tác chạy trong thời gian dài (LRO) hoàn tất. Điều này giúp bạn không cần phải thăm dò API để biết thông tin cập nhật về trạng thái, giảm độ trễ và chi phí.

Webhook có sẵn cho các thao tác như Công việc theo lô, Hoạt động tương tác và tạo video.

Cách hoạt động

Thay vì thăm dò GET /operations nhiều lần để kiểm tra xem một công việc đã hoàn tất hay chưa, bạn có thể định cấu hình Webhook Gemini API để gửi yêu cầu POST qua HTTP đến URL trình nghe ngay khi một sự kiện được kích hoạt.

Gemini API hỗ trợ 2 cách định cấu hình webhook:

  • Webhook tĩnh: Điểm cuối ở cấp dự án được định cấu hình bằng Gemini WebhookService API. Phù hợp với các hoạt động tích hợp trên toàn cầu (ví dụ: thông báo cho Slack, đồng bộ hoá cơ sở dữ liệu, v.v.).
  • Webhook động: Ghi đè ở cấp yêu cầu bằng cách truyền URL webhook trong tải trọng cấu hình của một lệnh gọi công việc cụ thể. Phù hợp để định tuyến các công việc cụ thể đến các điểm cuối chuyên dụng.

Webhook tĩnh

Webhook tĩnh được đăng ký cho toàn bộ dự án và kích hoạt cho mọi sự kiện phù hợp.

Tạo webhook

Bạn có thể tạo điểm cuối bằng SDK hoặc REST API.

QUAN TRỌNG: Khi tạo webhook, API sẽ chỉ trả về một lầnbí mật ký. Bạn phải lưu trữ bí mật này một cách an toàn (ví dụ: trong các biến môi trường) để xác minh chữ ký sau này. Nếu mất bí mật ký, bạn sẽ phải xoay vòng bí mật đó.

Python

from google import genai

client = genai.Client()

webhook = client.webhooks.create(
    name="MyBatchWebhook",
    subscribed_events=["batch.succeeded", "batch.failed"],
    uri="https://my-api.com/gemini-callback",
)

# Store webhook.new_signing_secret securely
webhook_secret = webhook.new_signing_secret
print(f"Created webhook: {webhook.name}, {webhook.id}")

JavaScript

import { GoogleGenAI } from "@google/genai";

const client = new GoogleGenAI();

async function createWebhook() {
  const webhook = await client.webhooks.create({
    name: "MyBatchWebhook",
    subscribed_events: ["batch.succeeded", "batch.failed"],
    uri: "https://my-api.com/gemini-callback",
  });

  // Store webhook.signingSecret securely
  const webhookSecret = webhook.new_signing_secret;
  console.log(`Created webhook: ${webhook.name}, ${webhook.id}`);
}

createWebhook();

REST

curl -X POST \
  "https://generativelanguage.googleapis.com/v1/webhooks" \
  -H "Content-Type: application/json" \
  -H "x-goog-api-key: $GEMINI_API_KEY" \
  -d '{
    "name": "MyBatchWebhook",
    "uri": "https://my-api.com/gemini-callback",
    "subscribed_events": ["batch.succeeded", "batch.failed"]
  }'

Để biết thông tin chi tiết về cách thiết lập máy chủ để nhận dữ liệu, hãy xem phần Xử lý yêu cầu webhook.

Nhận webhook

Truy xuất thông tin chi tiết về một webhook cụ thể theo tên tài nguyên của webhook đó.

Python

from google import genai

client = genai.Client()

webhook = client.webhooks.get(id="<your_webhook_id>")

print(f"Webhook: {webhook.name}")
print(f"URI: {webhook.uri}")
print(f"Events: {webhook.subscribed_events}")

JavaScript

import { GoogleGenAI } from "@google/genai";

const client = new GoogleGenAI(); // Assumes process.env.GEMINI_API_KEY is set

async function getWebhook() {
  const webhook = await client.webhooks.get("<your_webhook_id>");

  console.log(`Webhook: ${webhook.name}`);
  console.log(`URI: ${webhook.uri}`);
  console.log(`Events: ${webhook.subscribed_events}`);
}

getWebhook();

REST

curl -X GET \
  "https://generativelanguage.googleapis.com/v1/webhooks/<your_webhook_id>" \
  -H "x-goog-api-key: $GEMINI_API_KEY"

Liệt kê webhook

Liệt kê tất cả webhook đã định cấu hình cho dự án hiện tại, có thể phân trang.

Python

from google import genai

client = genai.Client()

webhooks = client.webhooks.list()

for wh in webhooks:
    print(f"{wh.id}: {wh.name} -> {wh.uri}")

JavaScript

import { GoogleGenAI } from "@google/genai";

const client = new GoogleGenAI();

async function listWebhooks() {
  const webhooks = await client.webhooks.list();

  for (const wh of webhooks) {
    console.log(`${wh.id}: ${wh.name} -> ${wh.uri}`);
  }
}

listWebhooks();

REST

curl -X GET \
  "https://generativelanguage.googleapis.com/v1/webhooks" \
  -H "x-goog-api-key: $GEMINI_API_KEY"

Cập nhật webhook

Cập nhật các thuộc tính của webhook hiện có, chẳng hạn như tên hiển thị, URI mục tiêu hoặc sự kiện đã đăng ký.

Python

from google import genai

client = genai.Client()

updated_webhook = client.webhooks.update(
    id="<your_webhook_id>",
    subscribed_events=["batch.succeeded", "batch.failed", "batch.cancelled"],
)

print(f"Updated webhook: {updated_webhook.name}")

JavaScript

import { GoogleGenAI } from "@google/genai";

const client = new GoogleGenAI();

async function updateWebhook() {
  const updatedWebhook = await client.webhooks.update(
    "<your_webhook_id>",
    {
      subscribed_events: ["batch.succeeded", "batch.failed", "batch.cancelled"],
    }
  );

  console.log(`Updated webhook: ${updatedWebhook.name}`);
}

updateWebhook();

REST

curl -X PATCH \
  "https://generativelanguage.googleapis.com/v1/webhooks/<your_webhook_id>" \
  -H "Content-Type: application/json" \
  -H "x-goog-api-key: $GEMINI_API_KEY" \
  -d '{
    "subscribed_events": ["batch.succeeded", "batch.failed", "batch.cancelled"]
  }'

Xoá webhook

Xoá điểm cuối webhook khỏi dự án. Thao tác này sẽ dừng việc phân phối sự kiện trong tương lai đến điểm cuối đó.

Python

from google import genai

client = genai.Client()

client.webhooks.delete(id="<your_webhook_id>")

print("Webhook deleted.")

JavaScript

import { GoogleGenAI } from "@google/genai";

const client = new GoogleGenAI();

async function deleteWebhook() {
  await client.webhooks.delete("<your_webhook_id>");

  console.log("Webhook deleted.");
}

deleteWebhook();

REST

curl -X DELETE \
  "https://generativelanguage.googleapis.com/v1/webhooks/<your_webhook_id>" \
  -H "x-goog-api-key: $GEMINI_API_KEY"

Xoay vòng bí mật ký

Xoay vòng bí mật ký cho webhook. Bạn có thể định cấu hình để thu hồi ngay các bí mật đang hoạt động trước đó hoặc sau thời gian gia hạn 24 giờ.

QUAN TRỌNG: Bí mật ký mới sẽ chỉ được trả về một lần tại thời điểm xoay vòng. Hãy lưu trữ bí mật này một cách an toàn trước khi cập nhật logic xác minh.

Python

from google import genai
from google.genai import types

client = genai.Client()

response = client.webhooks.rotate_signing_secret(
    id="<your_webhook_id>",
    revocation_behavior="REVOKE_PREVIOUS_SECRETS_AFTER_H24",
)

# Store response.secret securely, then update your server's verification config
print("New signing secret generated. Update your server configuration.")

JavaScript

import { GoogleGenAI } from "@google/genai";

const client = new GoogleGenAI();

async function rotateSigningSecret() {
  const response = await client.webhooks.rotateSigningSecret(
    "<your_webhook_id>",
    {
      revocation_behavior: "REVOKE_PREVIOUS_SECRETS_AFTER_H24",
    }
  );

  // Store response.secret securely, then update your server's verification config
  console.log("New signing secret generated. Update your server configuration.");
}

rotateSigningSecret();

REST

curl -X POST \
  "https://generativelanguage.googleapis.com/v1/webhooks/<your_webhook_id>/rotate_secret" \
  -H "Content-Type: application/json" \
  -H "x-goog-api-key: $GEMINI_API_KEY" \
  -d '{
    "revocation_behavior": "REVOKE_PREVIOUS_SECRETS_AFTER_H24"
  }'

Xử lý yêu cầu webhook trên máy chủ

Khi một sự kiện mà bạn đã đăng ký xảy ra, URL webhook của bạn sẽ nhận được yêu cầu POST qua HTTP. Điểm cuối của bạn phải phản hồi bằng mã trạng thái 2xx trong vòng vài giây để tránh thử lại. Để đảm bảo phân phối, Gemini API sẽ tự động thử lại các yêu cầu không thành công trong 24 giờ bằng cách sử dụng thuật toán trì hoãn theo cấp số nhân.

Gemini tuân thủ nghiêm ngặt thông số kỹ thuật Webhook tiêu chuẩn cho các tiêu đề bảo mật. Xác minh tải trọng trên máy chủ của bạn bằng chữ ký tiêu đề đã ký và bí mật ký tĩnh đã lưu trữ. Hãy xem phần Vỏ webhook để biết thông tin về tải trọng.

Sau đây là ví dụ về cách sử dụng Flask cho trình nghe HTTP:

Python

# pip install flask standardwebhooks
import os
from flask import Flask, request, jsonify
# Standard verification wrapper for Standard Webhook Headers
from standardwebhooks.webhooks import Webhook, WebhookVerificationError

app = Flask(__name__)

SIGNING_SECRET = os.environ.get('WEBHOOK_SIGNING_SECRET')

@app.route('/gemini-callback', methods=['POST'])
def gemini_callback():
    payload = request.get_data(as_text=True)
    headers = request.headers

    try:
        wh = Webhook(SIGNING_SECRET)
        event = wh.verify(payload, headers)
    except WebhookVerificationError as e:
        return jsonify({"error": "Signature invalid"}), 400

    # Process thin payload contents
    if event.get("type") == "batch.succeeded":
        print(f"Batch completed! ID: {event['data']['id']}")
        if event["data"].get("output_file_uri"):
            # For batch jobs with input file
            print(f"Batch file: {event['data']['output_file_uri']}")
    elif event.get("type") == "interaction.completed":
        print(f"Interaction completed! ID: {event['data']['id']}")
    elif event.get("type") == "video.generated":
        print(f"Video generated! URI: {event['data']['output_file_uri']}")

    return jsonify({"status": "received"}), 200

if __name__ == "__main__":
    app.run(port=8000)

JavaScript

// npm install standardwebhooks
import { Webhook } from "standardwebhooks";
import express from "express";

const app = express();
const client = new GoogleGenAI({ webhookSecret: process.env.WEBHOOK_SIGNING_SECRET });

// Don't use express.json() because signature verification needs the raw text body
app.use(express.text({ type: "application/json" }));

app.post("/gemini-callback", async (req, res) => {
  const payload = await req.text();
        const headers: Record<string, string> = {};
        req.headers.forEach((value, key) => {
            headers[key] = value;
        });

        try {
            const wh = new Webhook(process.env.WEBHOOK_SIGNING_SECRET);
            const event = wh.verify(payload, headers) as Record<string, any>;
    console.log(`Event type: ${event.type}, data: ${JSON.stringify(event.data)}`);

            // Process thin payload contents
            if (event.type === "batch.succeeded") {
                console.log(`Batch completed! ID: ${event.data.id}`);
                if (event.data.output_file_uri) {
                    // For batch jobs with input file
                    console.log(`Batch file: ${event.data.output_file_uri}`);
                }
            } else if (event.type === "interaction.completed") {
                console.log(`Interaction completed! ID: ${event.data.id}`);
            } else if (event.type === "video.generated") {
                console.log(`Video generated! URI: ${event.data.output_file_uri}`);
            }

            res.status(200).json({ status: "received" });
        } catch (e) {
            console.error("Webhook verification failed:", e);
            res.status(400).send("Invalid signature");
        }
});

app.listen(8000, () => {
  console.log("Webhook server is running on port 8000");
});

Webhook động

Webhook động cho phép bạn liên kết điểm cuối webhook với cấu hình yêu cầu cụ thể, phù hợp với hàng đợi điều phối tác nhân. Webhook động tận dụng chữ ký JWKS khoá công khai không đối xứng thay vì bí mật đối xứng.

Gửi yêu cầu động

Thêm webhook_config khi kích hoạt công việc không đồng bộ (ví dụ: tạo Lô).

Python

# This will only work for SDK newer than 2.0.0
from google import genai

client = genai.Client()

response = client.interactions.create(
    model='gemini-3.5-flash',
    input='Tell me a short joke about programming.',
    background=True, # Required when webhook_config is specified
    webhook_config={
        'uris': ["https://my-api.com/gemini-webhook-dynamic"],
        'user_metadata': {"job_group": "nightly-eval", "priority": "high"}
    }
)

print(f"Interaction created! ID: {response.id}")
print(f"Status: {response.status}")

JavaScript

// This will only work for SDK newer than 2.0.0
import { GoogleGenAI } from "@google/genai";

const client = new GoogleGenAI();

async function createInteractionWithWebhook() {
  const response = await client.interactions.create({
    model: "gemini-3.5-flash",
    input: "Tell me a short joke about programming.",
    background: true, // Required when webhook_config is specified
    webhook_config: {
      uris: ["https://my-api.com/gemini-webhook-dynamic"],
      user_metadata: { job_group: "nightly-eval", priority: "high" },
    },
  });

  console.log(`Interaction created! ID: ${response.id}`);
  console.log(`Status: ${response.status}`);
}

createInteractionWithWebhook();

REST

# Specifies the API revision to avoid breaking changes when they become default
curl -X POST \
  "https://generativelanguage.googleapis.com/v1beta/interactions" \
  -H "Content-Type: application/json" \
  -H "x-goog-api-key: $GEMINI_API_KEY" \
  -d '{
    "model": "gemini-3.5-flash",
    "input": "Tell me a short joke about programming.",
    "background": true,
    "webhook_config": {
      "uris": ["https://my-api.com/gemini-webhook-dynamic"],
      "user_metadata": {"job_group": "nightly-eval", "priority": "high"}
    }
  }'

Xác minh chữ ký động (JWKS)

Các yêu cầu webhook động phát ra chữ ký Mã thông báo web JSON (JWT). Trình nghe của bạn phải trích xuất chữ ký và xác minh chữ ký đó bằng các điểm cuối chứng chỉ công khai của Google.

Python

import jwt
import requests
from flask import Flask, request, jsonify

app = Flask(__name__)

# Google public cert list endpoint
JWKS_URI = "https://generativelanguage.googleapis.com/.well-known/jwks.json"

def load_google_public_key(kid):
    response = requests.get(JWKS_URI).json()
    for key_item in response.get('keys', []):
        if key_item.get('kid') == kid:
            # Convert JWK to Cert wrapper
            return jwt.algorithms.RSAAlgorithm.from_jwk(key_item)
    return None

@app.route('/gemini-webhook-dynamic', methods=['POST'])
def dynamic_handler():
    payload = request.get_data(as_text=True)
    headers = request.headers

    token = headers.get('Webhook-Signature')
    if not token:
        return jsonify({"error": "No signature header"}), 400

    try:
        # Extract kid from JWT header
        unverified_headers = jwt.get_unverified_header(token)
        pub_key = load_google_public_key(unverified_headers.get('kid'))

        if not pub_key:
            return jsonify({"error": "Key cert not found"}), 400

        # Verify Signature against expected audience (e.g., your project client ID)
        event = jwt.decode(
            token,
            pub_key,
            algorithms=["RS256"],
            audience="your-configured-audience"
        )
    except Exception as e:
        return jsonify({"error": "Invalid Dynamic signature", "details": str(e)}), 400

    print("Verified Dynamic payload success.")
    return jsonify({"status": "received"}), 200

JavaScript

import { GoogleGenAI } from "@google/genai";
import express from "express";
import jwt from "jsonwebtoken";
import jwksClient from "jwks-rsa";

const app = express();
app.use(express.text({ type: 'application/json' }));

const client = jwksClient({
  jwksUri: "https://generativelanguage.googleapis.com/.well-known/jwks.json"
});

function getKey(header, callback) {
  client.getSigningKey(header.kid, (err, key) => {
    const signingKey = key.getPublicKey();
    callback(null, signingKey);
  });
}

app.post('/gemini-webhook-dynamic', (req, res) => {
  const token = req.headers['webhook-signature'];

  if (!token) {
    return res.status(400).json({ error: "No signature header" });
  }

  jwt.verify(
    token,
    getKey,
    {
      algorithms: ["RS256"],
      audience: "your-configured-audience"
    },
    (err, decoded) => {
      if (err) {
        return res.status(400).json({ error: "Invalid Dynamic signature", details: err.message });
      }

      console.log("Verified Dynamic payload success.");
      res.status(200).json({ status: "received" });
    }
  );
});

Vỏ webhook

Để tránh tình trạng tắc nghẽn băng thông, webhook Gemini sử dụng mô hình tải trọng mỏng để phân phối dữ liệu. Các hoạt động phân phối sẽ gửi ảnh chụp nhanh chứa thông tin chi tiết về trạng thái và con trỏ đến kết quả, thay vì chính tệp đầu ra thô.

Sau đây là ví dụ về định dạng tải trọng:

{
  "type": "batch.succeeded",
  "version": "v1",
  "timestamp": "2026-01-22T12:00:00Z",
  "data": {
    "id": "batch_123456",
    "output_file_uri": "gs://my-bucket/results.jsonl"
  }
}

Tài liệu tham khảo về danh mục sự kiện

Các sự kiện sau đây được kích hoạt cho các công việc được hỗ trợ:

Loại sự kiện Trigger Mục tải trọng (data)
batch.succeeded Đã xử lý xong. id, output_file_uri
batch.cancelled Người dùng đã huỷ yêu cầu id
batch.expired Lô chưa được xử lý (hoàn tất) trong khung thời gian 24 giờ id
batch.failed Công việc theo lô không thành công (lỗi hệ thống hoặc lỗi xác thực). id, error_code, error_message
interaction.requires_action Lệnh gọi hàm, người dùng cần thực hiện thao tác id
interaction.completed LRO trong API hoạt động tương tác thành công id
interaction.failed LRO trong API hoạt động tương tác không thành công (lỗi hệ thống hoặc lỗi xác thực). id, error_code, error_message
interaction.cancelled LRO trong API hoạt động tương tác đã huỷ id
video.generated Đã hoàn tất LRO tạo video. id, output_file_uri, file_name

Các phương pháp hay nhất

Để đảm bảo hoạt động đáng tin cậy và có khả năng mở rộng:

  • Kiểm tra nghiêm ngặt khả năng bảo vệ khỏi phát lại: Tất cả yêu cầu đều mang webhook-timestamp tiêu đề. Luôn xác thực dấu thời gian này trên lớp cấu hình máy chủ để từ chối các tải trọng cũ hơn 5 phút (để giảm thiểu các cuộc tấn công phát lại).
  • Xử lý không đồng bộ: Phản hồi ngay lập tức bằng 2xx OK khi phát hiện chữ ký hợp lệ và xếp hàng các thao tác phân tích cú pháp nội bộ. Thời gian giữ trình nghe kéo dài sẽ kích hoạt chu kỳ thử lại phân phối.
  • Xử lý việc loại bỏ dữ liệu trùng lặp: Webhook tiêu chuẩn phân phối "Ít nhất một lần". Sử dụng tiêu đề webhook-id nhất quán để xử lý các bản sao tiềm ẩn trong các luồng tắc nghẽn cao hơn.

Tiếp theo là gì?

  • Batch API: Sử dụng webhook để tự động hoá các điểm cuối có khối lượng lớn.