Have you ever had your Antigravity integration run perfectly on your laptop, then collapse under rate-limit errors the moment you push it to production? I shipped my first Antigravity-powered side project with barely more than the SDK quickstart, and a late-night traffic spike chained 429s into timeouts into a rollback. The SDK itself was fine — I just forgot the seatbelts.
The Antigravity Python SDK is delightfully easy: client.models.generate_content() and you are off. But deploying that single call to real users is a bit like merging onto a highway without fastening anything. In this article I walk through the smallest set of safeguards I reach for on every new production deployment: retries, timeouts, and a circuit breaker.
Why a bare SDK call breaks under real traffic
Antigravity API failures fall into three rough buckets. The first is transient errors — 429 rate limits, the occasional 5xx, a brief network blip. These usually succeed if you wait a second or two. The second is long-running calls where a heavy prompt or cold model takes much longer than your usual budget; without a timeout, your request handler just waits forever. The third is persistent outages — a real incident on the provider side, or a misconfigured API key — where retrying will not help at all.
The SDK default is "call once, raise on error," which is naked against all three. The remedy is the same three safeguards, stacked in a specific order:
- Retries (absorb transient errors)
- Timeouts (cap a single call)
- Circuit breakers (stop hammering an outage)
Safeguard 1: Retries with exponential backoff and jitter
Start with tenacity. Do not write a flat "sleep 3 seconds and try again" loop — use exponential backoff with jitter. Without jitter, every process that hit the incident retries at the exact same moment, which actively slows down the service's recovery.
# retry.py — wrap Antigravity calls in a retry decorator
from google import genai
from google.genai import errors as genai_errors
from tenacity import (
retry, stop_after_attempt, wait_exponential_jitter, retry_if_exception_type
)
client = genai.Client() # reads GOOGLE_API_KEY from the environment
# Only retry truly transient errors. Auth failures should fail fast.
TRANSIENT = (
genai_errors.ServerError, # 5xx
genai_errors.ResourceExhausted, # 429 rate limit
TimeoutError,
)
@retry(
retry=retry_if_exception_type(TRANSIENT),
stop=stop_after_attempt(4), # up to 4 attempts
wait=wait_exponential_jitter(initial=1, max=30), # 1s -> 2s -> 4s -> 8s (+jitter)
reraise=True,
)
def generate_with_retry(prompt: str) -> str:
resp = client.models.generate_content(
model="gemini-2.5-flash",
contents=prompt,
)
return resp.text
if __name__ == "__main__":
print(generate_with_retry("Describe Antigravity retry strategy in one sentence."))Expected behavior: a transient 429 or 5xx triggers up to four attempts with exponential backoff, and the original exception is re-raised if none succeed. Auth failures like PermissionDenied are deliberately not in TRANSIENT, so they fail fast instead of hammering Google with a dead API key.
It is tempting to retry everything, but in my experience permanent errors should fail fast. Retrying them four times just makes your user wait longer for the same dead end.
Safeguard 2: Timeouts to cap a single call
Next, cap the duration of a single call. The SDK accepts a timeout in http_options, which is your inner boundary. I like to layer a second asyncio.wait_for on top so that the application has its own hard deadline even if the SDK decides to retry internally.
# timeout.py — layered timeouts
import asyncio
from google import genai
from google.genai import types
client = genai.Client()
async def generate_with_timeout(prompt: str, deadline_sec: float = 20.0) -> str:
"""Inner SDK timeout + outer asyncio deadline."""
async def _call() -> str:
resp = await client.aio.models.generate_content(
model="gemini-2.5-flash",
contents=prompt,
config=types.GenerateContentConfig(
http_options=types.HttpOptions(timeout=15_000), # milliseconds
),
)
return resp.text
# Outer app-level deadline (slightly larger than SDK timeout)
return await asyncio.wait_for(_call(), timeout=deadline_sec)
async def main():
try:
text = await generate_with_timeout("Three 2026 AI trends in one line each.")
print(text)
except asyncio.TimeoutError:
print("⏱ timeout: returning fallback response")
asyncio.run(main())Expected behavior: the SDK fires its own 15-second timeout first; if anything still hangs past 20 seconds, asyncio.wait_for raises TimeoutError. Catching that in the calling code lets you serve a cached reply or a static fallback instead of a spinner of death.
It feels generous to give calls a long deadline, but a shorter timeout plus a graceful fallback is actually kinder — you avoid exhausting your connection pool and blocking other users. I use roughly "10–20s for interactive" and "60s for batch" as starting points.
Safeguard 3: A circuit breaker to stop chain failures
The third safeguard is pybreaker. When the provider is genuinely down, retries do not help — they just burn your CPU and connection pool. A circuit breaker watches for consecutive failures, opens after a threshold (skipping calls entirely), waits a cooldown, then cautiously lets a single probe through.
# breaker.py — retries + circuit breaker
import pybreaker
from google import genai
from google.genai import errors as genai_errors
from tenacity import retry, stop_after_attempt, wait_exponential_jitter
client = genai.Client()
# Open after 5 consecutive failures; try again 60s later
breaker = pybreaker.CircuitBreaker(
fail_max=5,
reset_timeout=60,
exclude=[genai_errors.PermissionDenied], # auth errors should not trip the breaker
)
@breaker
@retry(stop=stop_after_attempt(3),
wait=wait_exponential_jitter(initial=1, max=10),
reraise=True)
def generate_safe(prompt: str) -> str:
resp = client.models.generate_content(
model="gemini-2.5-flash",
contents=prompt,
)
return resp.text
def answer(prompt: str) -> str:
try:
return generate_safe(prompt)
except pybreaker.CircuitBreakerError:
return "Our AI service is busy right now. Please try again in a moment."
except Exception as e:
return f"Temporary error ({type(e).__name__})"Expected behavior: after five consecutive failures the breaker opens, and further calls raise CircuitBreakerError instantly (no network call at all). After 60 seconds the breaker enters half-open, allows one probe, and closes again on success. Without this, a real outage means every request still pays the full retry cost.
One caveat: on serverless platforms (Cloud Run, Vercel) the breaker state is per-instance and will not be shared. For a truly robust setup, swap the in-memory state for Redis with pybreaker.CircuitRedisStorage.
How the three safeguards interact in a real incident
It helps to walk through a concrete timeline. Imagine Google pushes a bad region rollout and your Antigravity calls start returning 5xx at 11:47 PM local time.
- The first failed request hits the inner SDK timeout because the upstream is slow, not a clean error.
asyncio.wait_forcatches it and surfaces aTimeoutError. tenacitysees a transient error type and schedules the next attempt with a ~1 second pause plus a small jitter. Another process on the same server randomly picks 1.3 seconds, so the two do not sync up.- After two attempts still fail, the user sees an immediate fallback string (because you wrapped the call with your own try/except). Meanwhile,
pybreakerkeeps counting. - Once five failures land within the breaker window, it flips to OPEN. Every subsequent request skips the network entirely and returns the fallback in milliseconds. Your application CPU stays flat.
- Google restores the region at 12:03 AM. At the next natural retry window the breaker sends a single probe request. It succeeds, the breaker closes, and traffic resumes without any human intervention.
Without this stack, the same incident would have meant sixteen minutes of 30-second spinners, pager alerts, and a long manual warm-up. With it, you usually do not even wake up.
Observability: if you cannot see it, you cannot trust it
A safeguard you cannot observe is only slightly better than no safeguard at all. Two small additions make the difference between "I think retries are working" and "I can prove it":
- Structured logs on every retry.
tenacityexposesbefore_sleep=before_sleep_log(logger, logging.WARNING). Ship those logs with the prompt hash and the attempt number so you can grep for retry storms. - Breaker state metric. Expose the breaker's
current_stateas a Prometheus gauge (or a one-line write to any time-series backend). Alerting on "breaker stayed OPEN for more than 5 minutes" catches real outages without blowing up your pager for normal traffic bumps.
I have a personal rule: if I add resilience code, I add the log and the metric in the same pull request. Otherwise I am just adding confidence-without-evidence, which is how production surprises happen.
Stacking the three safeguards
Each safeguard is useful on its own, but the canonical production layout is breaker → retries → timeout, outermost to innermost. The breaker gates whether you even try; retries absorb transient failures when the breaker is closed; and each attempt is capped by a timeout so no single call can run away.
- Outer: circuit breaker (detects full outages)
- Middle: retries with exponential backoff (absorbs blips)
- Inner: SDK timeout + asyncio deadline (caps each attempt)
If you run an Antigravity-powered service for more than a few weeks, you will see sporadic 429s at peak, occasional 5xxs during upstream maintenance, and the odd 40-second response that should have been ten. Any of these three safeguards, layered correctly, reduces those events from "outage" to "barely noticeable."
For a deeper systems view, see the Antigravity Python SDK production master guide, and for diagnosing specific failures see Antigravity Python SDK error diagnosis and fixes. If you are brand new to the SDK, start with the Antigravity Python × Gemini API quickstart.
One concrete next step
Pick one Antigravity call in your current codebase and add just the tenacity retry decorator today. Five extra lines is often enough to eliminate the majority of nightly rate-limit errors from your logs. Once you trust that layer, add the timeout, then the circuit breaker — in that order.