Build an Event-Driven Gemini API Pipeline with Webhooks Instead of Polling

On June 2026, Google added event-driven Webhooks to the Gemini API, so the Batch API and long-running operations can call your server when they finish instead of making you ask.

AI TECH CODING

June 29, 20266 min read11 sectionsBy Ahmed Abdullah

Build an Event-Driven Gemini API Pipeline with Webhooks Instead of Polling

Introduction

On June 2026, Google added event-driven Webhooks to the Gemini API, so the Batch API and long-running operations can call your server when they finish instead of making you ask. You will build a webhook receiver in Python with FastAPI, register it with the Gemini API, fire a batch job that notifies it, verify that the request actually came from Google, and process the results the moment they land. By the end you have a pipeline with zero polling loops.

What polling actually costs you in the Gemini Batch API

The Gemini Batch API takes a large set of requests, runs them asynchronously, and returns results later, sometimes minutes later, sometimes hours. Before webhooks, the only way to know it was done was to ask in a loop.

python

import time
from google import genai
client = genai.Client(api_key="YOUR_GEMINI_API_KEY")
job = client.batches.get(name="batches/abc123")
while job.state.name not in ("JOB_STATE_SUCCEEDED", "JOB_STATE_FAILED"):
time.sleep(30) # guess an interval, hope it is right
job = client.batches.get(name="batches/abc123")
print("done", job.state.name)

Look at what that loop is. A guess at an interval, a process that has to stay alive doing nothing, and a result you learn about up to thirty seconds after it was ready. Run a hundred batch jobs and you are running a hundred of these. The polling interval is a tax you pay in latency, and the idle process is a tax you pay in compute. Neither tax buys you anything.

What a webhook is in the Gemini API

A webhook inverts the question. Instead of your code asking Google, Google calls your code. You register a public HTTPS endpoint once. When a batch job changes state, the Gemini API sends an HTTP POST to that endpoint with an event payload describing what happened. Your server reacts.

The event carries the operation name, the new state, and a signature header so you can prove the call came from Google and not from someone who guessed your URL. That signature is the part most people skip, and it is the part that matters most.

Stand up a webhook receiver with FastAPI

Start with the endpoint that receives the event. It does one thing: accept the POST, read the raw body, and hand off. Keep the handler fast, because the Gemini API expects a quick 200 and will retry if you stall.

python

# receiver.py
import os
from fastapi import FastAPI, Request, Header, HTTPException
app = FastAPI()
WEBHOOK_SECRET = os.environ["GEMINI_WEBHOOK_SECRET"]
@app.post("/gemini/webhook")
async def gemini_webhook(
request: Request,
x_goog_signature: str | None = Header(default=None),
):
raw = await request.body()
if not x_goog_signature:
raise HTTPException(status_code=401, detail="missing signature")
# signature check comes in a later step; accept the shape for now
event = await request.json()
print("event:", event.get("eventType"), event.get("operation"))
return {"received": True}

This runs with uvicorn receiver:app --port 8080. For local development put it behind a tunnel so Google can reach it, then point the webhook at the public URL. The endpoint is deliberately boring. A receiver that does real work inline is a receiver that times out.

Register the webhook with the Gemini API

Now tell the Gemini API where to call. You register the endpoint once and get back a webhook resource you reference when you submit jobs.

python

# register.py
from google import genai
client = genai.Client(api_key="YOUR_GEMINI_API_KEY")
hook = client.webhooks.create(
url="https://your-domain.com/gemini/webhook",
events=["batch.completed", "batch.failed"],
secret="YOUR_GEMINI_WEBHOOK_SECRET", # used to sign every delivery
)
print("registered webhook:", hook.name)

The secret you set here is the same value your receiver reads from GEMINI_WEBHOOK_SECRET. Google signs every delivery with it. Subscribe only to the events you handle, batch.completed and batch.failed, so your endpoint is never woken for state changes it does not care about.

Submit a Batch job that targets the webhook

With the webhook registered, a batch job points at it through its notification config. The job runs asynchronously and your code moves on.

python

# submit.py
from google import genai
client = genai.Client(api_key="YOUR_GEMINI_API_KEY")
requests = [
{"contents": [{"parts": [{"text": f"Summarize ticket #{i}"}]}]}
for i in range(500)
]
job = client.batches.create(
model="gemini-3.5-flash",
src=requests,
config={"notification": {"webhook": hook.name}},
)
print("submitted:", job.name) # no loop, no sleep, no waiting

Five hundred requests go to Gemini 3.5 Flash in one batch, and the call returns immediately. There is no while loop after this line. The next time your code hears about this job is when the webhook fires.

Verify the signature so only Google can call you

A public endpoint is an open door. Anyone who learns the URL can POST fake completion events. The signature closes the door. Google signs the raw request body with your secret using HMAC-SHA256, and you recompute it.

python

# verify.py
import hashlib
import hmac
def verify_signature(raw_body: bytes, signature: str, secret: str) -> bool:
expected = hmac.new(
key=secret.encode("utf-8"),
msg=raw_body,
digestmod=hashlib.sha256,
).hexdigest()
# constant-time compare, never use ==
return hmac.compare_digest(expected, signature)

The detail that bites people is hmac.compare_digest. A plain == comparison leaks timing information that lets an attacker recover the signature byte by byte. Use the constant-time compare. It is one function call and it is not optional.

Gemini API Webhooks vs Polling

Dimension	Polling (batches.get loop)	Webhooks
Who initiates	Your code asks repeatedly	Gemini API calls you
Latency to result	Up to one poll interval	Immediate on completion
Idle compute	A process stays alive waiting	None between events
Scaling to many jobs	One loop per job	One endpoint for all jobs
Failure handling	You build ret/timeout logic	Delivery retries built in
Setup cost	None	A public endpoint plus signature check

The only column where polling wins is setup cost, and you pay that once.

What this does not do

This pipeline does not give you ordering guarantees. Webhook deliveries can arrive out of order, and batch.failed for one job may land before batch.completed for another. It does not handle duplicate deliveries on its own either, because Google retries on a non-200, so the same completion event can arrive twice. Treat the handler as idempotent: key your save_summary writes on request_id so a replay overwrites rather than duplicates. And a webhook is useless if your endpoint is unreachable. Behind a flaky tunnel or a sleeping laptop, deliveries fail and you are back to polling, except now you are polling for the deliveries you missed.

The full working example

python

# app.py -> run with: uvicorn app:app --port 8080
import hashlib
import hmac
import os
from fastapi import FastAPI, Request, Header, HTTPException
from google import genai
app = FastAPI()
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
SECRET = os.environ["GEMINI_WEBHOOK_SECRET"]
_seen: set[str] = set() # naive idempotency for the demo
def verify_signature(raw: bytes, signature: str) -> bool:
expected = hmac.new(SECRET.encode(), raw, hashlib.sha256).hexdigest()
return hmac.compare_digest(expected, signature)
def handle_completed(op_name: str):
job = client.batches.get(name=op_name)
if job.state.name != "JOB_STATE_SUCCEEDED":
return
for item in client.batches.list_results(name=op_name):
if item.request_id in _seen:
continue
_seen.add(item.request_id)
text = item.response.candidates[0].content.parts[0].text
print(f"[{item.request_id}] {text[:80]}")
@app.post("/gemini/webhook")
async def webhook(request: Request, x_goog_signature: str = Header(default="")):
raw = await request.body()
if not verify_signature(raw, x_goog_signature):
raise HTTPException(status_code=401, detail="bad signature")
event = await request.json()
if event.get("eventType") == "batch.completed":
handle_completed(event["operation"])
return {"received": True}
@app.post("/submit")
def submit():
reqs = [{"contents": [{"parts": [{"text": f"Summarize ticket #{i}"}]}]}
for i in range(500)]
job = client.batches.create(
model="gemini-3.5-flash",
src=reqs,
config={"notification": {"webhook": "webhooks/your-id"}},
)
return {"submitted": job.name}

Set GEMINI_API_KEY and GEMINI_WEBHOOK_SECRET, register the webhook once with register.py, expose port 8080 over HTTPS, then POST to /submit. The batch runs, and /gemini/webhook fires when it is done.

When to reach for this

Use webhooks the moment you have more than one batch job in flight or any long-running Gemini API operation where the result is not needed in the same request. For a single synchronous call that returns in two seconds, keep the simple await and skip all of this. The webhook earns its setup cost when the waiting would otherwise be a process you are paying to keep awake.

Keep reading from the journal.

June 29, 2026

Run LLM-Generated Code Safely with Cloudflare codemode and Dynamic Workers

Cloudflare's Agents SDK v0.16.1, shipped June 16, 2026, includes code mode: a way to let a model write a single program and execute it inside a sandboxed Dynamic Worker instead of making one tool call at a time.