Webhooki

Webhooki umożliwiają interfejsowi Gemini API wysyłanie powiadomień w czasie rzeczywistym na Twój serwer, gdy zakończą się operacje asynchroniczne lub długotrwałe (LRO). Zastępuje to konieczność odpytywania interfejsu API o aktualizacje stanu, co zmniejsza opóźnienia i obciążenie.

Webhooki są dostępne w przypadku operacji takich jak zadania wsadowe, interakcje i generowanie filmów.

Jak to działa

Zamiast wielokrotnie odpytywać GET /operations, czy zadanie zostało zakończone, możesz skonfigurować webhooki Gemini API tak, aby po wywołaniu zdarzenia natychmiast wysyłały żądanie HTTP POST na adres URL odbiornika.

Interfejs Gemini API obsługuje 2 sposoby konfigurowania webhooków:

  • Statyczne webhooki: punkty końcowe na poziomie projektu skonfigurowane za pomocą interfejsu WebhookService API Gemini. Ta opcja sprawdza się w przypadku integracji globalnych (np. powiadamiania Slacka, synchronizowania bazy danych itp.).
  • Dynamiczne webhooki: zastąpienia na poziomie żądania przekazujące adres URL webhooka w ładunku konfiguracyjnym konkretnego wywołania interfejsu Jobs API. Idealne rozwiązanie do kierowania określonych zadań do dedykowanych punktów końcowych.

Statyczne webhooki

Statyczne webhooki są rejestrowane dla całego projektu i wyzwalane w przypadku każdego pasującego zdarzenia.

Tworzenie webhooka

Punkty końcowe możesz tworzyć za pomocą pakietu SDK lub interfejsu API REST.

WAŻNE: podczas tworzenia webhooka interfejs API zwraca tajny klucz podpisywania tylko raz. Musisz bezpiecznie przechowywać ten klucz (np. w zmiennych środowiskowych), aby później weryfikować podpisy. Jeśli utracisz klucz tajny podpisywania, musisz go zrotować.

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"]
  }'

Więcej informacji o konfigurowaniu serwera do odbierania danych znajdziesz w sekcji Obsługa żądań webhooka.

Pobieranie webhooka

Pobieranie szczegółowych informacji o konkretnym webhooku na podstawie jego nazwy zasobu.

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"

Wyświetlanie listy webhooków

Wyświetla listę wszystkich skonfigurowanych webhooków w bieżącym projekcie z opcjonalnym podziałem na strony.

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"

Aktualizowanie webhooka

Aktualizowanie właściwości istniejącego webhooka, takich jak wyświetlana nazwa, docelowy URI lub subskrybowane zdarzenia.

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"]
  }'

Usuwanie webhooka

Usuń punkt końcowy webhooka z projektu. Spowoduje to zatrzymanie dostarczania przyszłych zdarzeń do tego punktu końcowego.

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"

Wykonaj rotację obiektu tajnego podpisywania

Zmień obiekt tajny podpisywania webhooka. Możesz skonfigurować, czy wcześniej aktywne klucze tajne mają być unieważniane natychmiast, czy po 24-godzinnym okresie prolongaty.

WAŻNE: nowy klucz tajny podpisywania jest zwracany tylko raz w momencie rotacji. Zanim zaktualizujesz logikę weryfikacji, zapisz go w bezpiecznym miejscu.

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

Obsługa żądań webhooka na serwerze

Gdy nastąpi zdarzenie, na które masz subskrypcję, Twój adres URL webhooka otrzyma żądanie HTTP POST. Aby uniknąć ponownej próby, punkt końcowy musi odpowiedzieć kodem stanu 2xx w ciągu kilku sekund. Aby zapewnić dostarczenie, interfejs Gemini API automatycznie ponawia nieudane żądania przez 24 godziny, stosując algorytm wzrastającego czasu do ponowienia.

Gemini ściśle przestrzega specyfikacji Standard Webhooks w przypadku nagłówków bezpieczeństwa. Sprawdź ładunek na serwerze, używając podpisów w podpisanym nagłówku i zapisanego statycznego tajnego klucza podpisywania. Informacje o ładunku znajdziesz w sekcji Koperta webhooka.

Oto przykład użycia platformy Flask w przypadku odbiornika 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.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 === "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");
});

Dynamiczne webhooki

Dynamiczne wywołania zwrotne umożliwiają powiązanie punktu końcowego wywołania zwrotnego z konkretną konfiguracją żądania, co jest idealne w przypadku kolejek orkiestracji agentów. Dynamiczne webhooki wykorzystują asymetryczne podpisy JWKS klucza publicznego zamiast symetrycznych kluczy tajnych.

Przesyłanie dynamicznej prośby

Dodaj webhook_config podczas wywoływania zadania asynchronicznego (np. tworzenia Batch).

Python

from google import genai
from google.genai import types

client = genai.Client()

file_batch_job = client.batches.create(
    model="gemini-3-flash-preview",
    src="files/uploaded_file_id",
    config={
        "display_name": "My Setup",
        "webhook_config": {
            "uris": ["https://my-api.com/gemini-webhook-dynamic"],
            "user_metadata":{"job_group": "nightly-eval", "priority": "high"}
        }
    }
)

JavaScript

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

const client = new GoogleGenAI();

async function createBatchWithWebhook() {
  const fileBatchJob = await client.batches.create({
    model: "gemini-3-flash-preview",
    src: "files/uploaded_file_id",
    config: {
      displayName: "My Setup",
      webhookConfig: {
        uris: ["https://my-api.com/gemini-webhook-dynamic"],
        user_metadata: {"job_group": "nightly-eval", "priority": "high"}
      },
    },
  });
}

REST

curl -X POST \
  "https://generativelanguage.googleapis.com/v1/models/gemini-3-flash-preview:batchCreate" \
  -H "Content-Type: application/json" \
  -H "x-goog-api-key: $GEMINI_API_KEY" \
  -d '{
    "src": "files/uploaded_file_id",
    "config": {
      "display_name": "My Setup",
      "webhook_config": {
        "uris": ["https://my-api.com/gemini-webhook-dynamic"],
        "user_metadata": {"job_group": "nightly-eval", "priority": "high"}
      }
    }
  }'

Weryfikowanie podpisów dynamicznych (JWKS)

Żądania dynamicznych webhooków emitują podpis tokena sieciowego JSON (JWT). Odbiornik musi wyodrębnić podpis i zweryfikować go za pomocą punktów końcowych certyfikatu publicznego 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" });
    }
  );
});

Koperta webhooka

Aby uniknąć przeciążenia przepustowości, webhooki Gemini używają modelu cienki ładunek do dostarczania danych. Dostawy wysyłają migawkę zawierającą szczegóły stanu i wskaźniki wyników, a nie sam plik wyjściowy.

Oto przykładowy format ładunku:

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

Informacje o katalogu zdarzeń

W przypadku zadań pomocniczych wywoływane są te zdarzenia:

Typ zdarzenia Wyzwalacz Element ładunku (data)
batch.succeeded Przetwarzanie zostało zakończone. id, output_file_uri
batch.cancelled Użytkownik anulował prośbę id
batch.expired Partia nie została przetworzona (zakończona) w ciągu 24 godzin id
batch.failed Zadanie wsadowe zakończyło się niepowodzeniem (błąd systemu lub weryfikacji). id, error_code, error_message
interaction.requires_action Wywołanie funkcji, użytkownik musi coś zrobić id
interaction.completed Długotrwała operacja w interfejsie API interakcji zakończyła się powodzeniem id
interaction.failed Nie udało się wykonać operacji LRO w interfejsie API interakcji (błąd systemu lub weryfikacji). id, error_code, error_message
interaction.cancelled Anulowanie długotrwałej operacji w interfejsie API interakcji id
video.generated Zakończono długotrwałą operację generowania filmu. id, output_file_uri, file_name

Sprawdzone metody

Aby zapewnić niezawodne i skalowalne działanie:

  • Ścisła ochrona przed powtórzeniem: wszystkie żądania zawierają webhook-timestampnagłówek. Zawsze sprawdzaj ten sygnaturę czasową w warstwie konfiguracji serwera, aby odrzucać ładunki starsze niż 5 minut (w celu ograniczenia ataków typu replay).
  • Przetwarzanie asynchroniczne: odpowiadaj za pomocą 2xx OK natychmiast po wykryciu prawidłowego podpisu i kolejkuj wewnętrznie operacje analizowania. Długi czas oczekiwania słuchacza spowoduje ponowienie próby dostarczenia.
  • Obsługa duplikatów: standardowe webhooki dostarczają dane „co najmniej raz”. Używaj spójnego nagłówka webhook-id, aby obsługiwać potencjalne duplikaty w przypadku większego natężenia ruchu.

Co dalej?

  • Batch API: Używaj webhooków do automatyzacji punktów końcowych o dużej liczbie wywołań.