Webhooks

Webhooks

Webhook를 이용하면 비동기 또는 장기 실행 작업(Long-Running Operations, LRO)이 완료됐을 때 Gemini API가 실시간 알림을 서버로 push해 줘요. 상태 업데이트를 위해 API를 폴링할 필요가 없어져 지연과 오버헤드가 줄어들어요.

Webhook는 Batch 작업, Interactions, 비디오 생성 같은 작업에 사용할 수 있어요.

출처: 원문

본문

작동 방식

작업이 끝났는지 확인하기 위해 GET /operations를 반복 폴링하는 대신, Gemini API Webhook를 구성해 이벤트 트리거 발생 즉시 리스너 URL로 HTTP POST 요청을 보낼 수 있어요.

Gemini API는 웹훅을 구성하는 두 가지 방법을 지원해요.

• 정적 웹훅: Gemini WebhookService API로 구성하는 프로젝트 수준 엔드포인트. 전역 통합(Slack 알림, DB 동기화 등)에 좋아요.
• 동적 웹훅: 특정 작업 호출의 구성 페이로드에서 웹훅 URL을 전달하는 요청 수준 오버라이드. 특정 작업을 전용 엔드포인트로 라우팅하는 데 이상적이에요.

정적 웹훅

정적 웹훅은 전체 프로젝트에 대해 등록되고 일치하는 모든 이벤트에 대해 트리거돼요.

웹훅 만들기

SDK나 REST API로 엔드포인트를 만들 수 있어요.

중요: 웹훅을 만들 때 API는 서명 시크릿(signing secret)을 한 번만 반환해요. 나중에 서명을 검증하려면 이를 안전하게(예: 환경 변수) 저장해야 해요. 서명 시크릿을 잃으면 회전시켜야 해요.

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

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();

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

서버가 데이터를 받도록 설정하는 방법은 웹훅 요청 처리 섹션을 참고하세요.

웹훅 가져오기

리소스 이름으로 특정 웹훅의 세부 정보를 검색해요.

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

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();

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

웹훅 나열

현재 프로젝트에 구성된 모든 웹훅을 선택적 페이지네이션과 함께 나열해요.

from google import genai

client = genai.Client()

webhooks = client.webhooks.list()

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

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();

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

웹훅 업데이트

표시 이름, 대상 URI, 구독 이벤트 같은 기존 웹훅의 속성을 업데이트해요.

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

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();

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

웹훅 삭제

프로젝트에서 웹훅 엔드포인트를 제거해요. 이후 해당 엔드포인트로의 이벤트 전달이 중지돼요.

from google import genai

client = genai.Client()

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

print("Webhook deleted.")

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

const client = new GoogleGenAI();

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

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

deleteWebhook();

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

서명 시크릿 회전

웹훅의 서명 시크릿을 회전해요. 이전에 활성화된 시크릿을 즉시 폐기할지 24시간 유예 기간 후 폐기할지 구성할 수 있어요.

중요: 새 서명 시크릿은 회전 시점에 한 번만 반환돼요. 검증 로직을 업데이트하기 전에 안전하게 저장하세요.

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

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();

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

서버에서 웹훅 요청 처리

구독 중인 이벤트가 발생하면 웹훅 URL이 HTTP POST 요청을 받아요. 엔드포인트는 재시도를 피하려면 몇 초 안에 2xx 상태 코드로 응답해야 해요. 전달을 보장하기 위해 Gemini API는 실패한 요청을 지수 백오프로 24시간 동안 자동 재시도해요.

Gemini는 보안 헤더에 대해 Standard Webhooks 사양을 엄격히 따르고 있어요. 서버에서 서명된 헤더 서명과 저장된 정적 서명 시크릿으로 페이로드를 검증하세요. 페이로드 정보는 Webhook 봉투 섹션을 참고하세요.

HTTP 리스너에 Flask를 사용하는 예제는 다음과 같아요.

# 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)

// 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");
});

동적 웹훅

동적 웹훅을 사용하면 웹훅 엔드포인트를 특정 요청 구성에 바인딩할 수 있어요. 에이전트 오케스트레이션 큐에 이상적이에요. 동적 웹훅은 대칭 시크릿 대신 비대칭 공개 키 JWKS 서명을 활용해요.

동적 요청 제출

비동기 작업(예: Batch 생성)을 트리거할 때 webhook_config를 추가해요.

from google import genai
from google.genai import types

client = genai.Client()

file_batch_job = client.batches.create(
    model="gemini-3.8-flash",
    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"}
        }
    }
)

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

const client = new GoogleGenAI();

async function createBatchWithWebhook() {
  const fileBatchJob = await client.batches.create({
    model: "gemini-3.8-flash",
    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"}
      },
    },
  });
}

curl -X POST \
  "https://generativelanguage.googleapis.com/v1/models/gemini-3.8-flash:batchCreate" \
  -H "Content-Type: application/json" \
  -H "x-goog-api-key: YOUR_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"}
      }
    }
  }'

동적 서명 검증(JWKS)

동적 웹훅 요청은 JSON Web Token(JWT) 서명을 발행해요. 리스너는 서명을 추출하고 Google 공개 인증서 엔드포인트로 검증해야 해요.

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

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" });
    }
  );
});

Webhook 봉투

대역폭 혼잡을 피하기 위해 Gemini 웹훅은 데이터 전달에 thin payload 모델을 사용해요. 전달은 원시 출력 파일 자체 대신 상태 세부 정보와 결과 포인터를 담은 스냅샷을 보내요.

예시 페이로드 형식:

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

이벤트 카탈로그 참조

지원 작업에 대해 다음 이벤트가 트리거돼요.

이벤트 유형	트리거	페이로드 항목(data)
batch.succeeded	처리가 성공적으로 완료됨.	id, output_file_uri
batch.cancelled	사용자가 요청 취소	id
batch.expired	24시간 내에 배치가 처리(완료)되지 않음	id
batch.failed	배치 작업 실패(시스템 또는 검증 오류).	id, error_code, error_message
interaction.requires_action	함수 호출, 사용자가 뭔가 해야 함	id
interaction.completed	interactions API의 LRO 성공	id
interaction.failed	interactions API의 LRO 실패(시스템 또는 검증 오류).	id, error_code, error_message
interaction.cancelled	interactions API의 LRO 취소	id
video.generated	비디오 생성 LRO 완료.	id, output_file_uri, file_name

모범 사례

안정적이고 확장 가능한 운영을 위해:

• 엄격한 재생 방지 검사: 모든 요청은 webhook-timestamp 헤더를 담아요. 이 타임스탬프를 항상 서버 구성 레이어에서 검증해 5분보다 오래된 페이로드를 거부하세요(재생 공격 완화).
• 비동기 처리: 유효한 서명 감지 시 즉시 2xx OK로 응답하고 파싱 작업은 내부적으로 대기시켜요. 리스너 대기 시간이 길면 전달 재시도가 트리거돼요.
• 중복 처리: 표준 웹훅은 "최소 1회(At-least-once)" 전달해요. 일관된 webhook-id 헤더로 높은 혼잡 흐름에서 잠재적 중복을 처리하세요.

다음 단계

• Batch API: 웹훅을 활용해 대용량 엔드포인트를 자동화하세요.

더 알아보기 (Learn more)

• Batch API
• Interactions API
• Video generation