Keeping the Antigravity Python API Stable in Production — Retries, Timeouts, and Circuit Breakers That Actually Work

Timeouts Are Not a Single Number

A lot of Python services have exactly one timeout in them, and it looks like httpx.AsyncClient(timeout=30). That's a trap. httpx actually exposes four distinct timeouts: connect, read, write, and pool. Gen AI calls, especially with long prompts or Thinking mode, can legitimately sit in the "read" phase for 30 to 120 seconds while the model generates. If you blanket-set everything to 30 seconds you'll be cutting off perfectly healthy requests.

The pattern I've settled on is to have different client instances with different timeout profiles, keyed on the model.

# timeouts.py — per-model timeout strategies
from dataclasses import dataclass
import httpx
 
@dataclass(frozen=True)
class ModelTimeouts:
    connect: float
    read: float
    write: float
    pool: float
 
    def to_httpx(self) -> httpx.Timeout:
        return httpx.Timeout(
            connect=self.connect,
            read=self.read,
            write=self.write,
            pool=self.pool,
        )
 
TIMEOUTS = {
    # Flash: small, fast. If read > 30s, something is wrong.
    "gemini-3-flash": ModelTimeouts(connect=5, read=30, write=10, pool=2),
    # Pro: can reason for longer — relax read.
    "gemini-3-pro": ModelTimeouts(connect=5, read=120, write=10, pool=2),
    # Thinking: allow even more headroom.
    "gemini-3-pro-thinking": ModelTimeouts(connect=5, read=240, write=10, pool=2),
}
 
def client_for(model: str) -> httpx.AsyncClient:
    t = TIMEOUTS.get(model)
    if t is None:
        raise ValueError(f"Unknown model: {model}")
    return httpx.AsyncClient(
        timeout=t.to_httpx(),
        limits=httpx.Limits(max_connections=100, max_keepalive_connections=20),
        http2=True,  # multiplexing reduces latency for concurrent calls
    )

The key idea here is don't share pools across models with wildly different latency profiles. If you use one Flash-tuned client for a Pro call, the long Pro read will block your entire pool and take Flash down with it.

Separately, always put a deadline on the entire user request, outside the HTTP client, with asyncio.timeout(). This is an independent concept from socket timeouts — it says "no matter how many retries we do, we give up after this long."

# handler.py — outer deadline for the user request
async def handle_user_request(user_input: str, model: str) -> str:
    async with asyncio.timeout(45):  # never make the user wait longer
        async with client_for(model) as client:
            request = _build_request(user_input, model)
            response = await call_with_retry(
                client, request, max_attempts=3, total_budget_sec=30,
            )
            return response.json()["candidates"][0]["content"]["parts"][0]["text"]

Note that the outer 45 seconds is slightly longer than the inner retry budget of 30. That gap is deliberate: it guarantees the retry loop always gets to finish naturally, rather than getting cut off mid-attempt by the outer deadline and leaving a confusing partial state behind.

Circuit Breakers — the Last Line of Defense Against Cascades

Retries look clever from the client's side ("we'll just try again") but cruel from the server's side ("you're piling on during my incident"). When the upstream API is genuinely unwell — not a single blip, but a real outage — every retry you make slows the recovery and burns your quota.

A circuit breaker is a tiny state machine that sits in front of the API call: when failures cross a threshold, it opens and fails fast, skipping the network entirely. After a cool-off period it goes half-open and lets one probe request through. If that succeeds it closes and traffic resumes.

A Minimal Async Circuit Breaker

For production I'd reach for pybreaker, but writing one yourself at least once is the single best way to internalize the behavior.

# circuit.py — a small async circuit breaker
from __future__ import annotations
import asyncio
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Awaitable, Callable, TypeVar
 
from ai_errors import RetryableAIError
 
T = TypeVar("T")
 
class State(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"
 
@dataclass
class CircuitBreaker:
    failure_threshold: int = 5
    recovery_timeout_sec: float = 30.0
    state: State = State.CLOSED
    _failure_count: int = 0
    _opened_at: float | None = None
    _lock: asyncio.Lock = field(default_factory=asyncio.Lock)
 
    async def call(self, fn: Callable[[], Awaitable[T]]) -> T:
        async with self._lock:
            if self.state is State.OPEN:
                assert self._opened_at is not None
                if time.monotonic() - self._opened_at >= self.recovery_timeout_sec:
                    self.state = State.HALF_OPEN
                else:
                    raise RetryableAIError("Circuit is OPEN — fast-failing")
 
        try:
            result = await fn()
        except Exception:
            async with self._lock:
                self._failure_count += 1
                if (
                    self.state is State.HALF_OPEN
                    or self._failure_count >= self.failure_threshold
                ):
                    self.state = State.OPEN
                    self._opened_at = time.monotonic()
            raise
        else:
            async with self._lock:
                self._failure_count = 0
                self.state = State.CLOSED
                self._opened_at = None
            return result

Wrap the retry helper from the previous section with this breaker and you get "don't pile on during an outage" behavior for almost free:

breaker = CircuitBreaker(failure_threshold=5, recovery_timeout_sec=30.0)
 
async def protected_call() -> str:
    async def _do() -> str:
        async with client_for("gemini-3-pro") as client:
            request = _build_request(user_input, "gemini-3-pro")
            response = await call_with_retry(client, request)
            return response.json()["candidates"][0]["content"]["parts"][0]["text"]
 
    return await breaker.call(_do)

Three implementation notes. Count final failures, not retry intermediates — otherwise the breaker will open far too aggressively because each retried failure will look like a brand-new incident. In the code above this happens naturally: only the last exception from call_with_retry reaches the breaker. Scope breakers per-model (or per-key) — if Flash is having a bad day, you don't want to knock out Pro as well. And tune recovery_timeout_sec from real data: I start at 30 seconds, watch actual recovery times in the metrics, and set the timeout to roughly 1.5× the observed average. Too short means thrashing; too long means you're down when the API has long since recovered.

Streaming Responses That Survive Mid-Flight Disconnects

Streaming is a huge UX win, but it opens up a new failure mode: what do you do when the connection drops 100 tokens into a 1,000-token generation? Restarting from scratch is both costly and jarring for the user.

Because LLMs are non-deterministic, you can't generally resume exactly where you left off. The two pragmatic options are:

• Treat the partial result as complete. Return what you have, move on. Works well for chat UIs.
• Use the partial result as prompt context and ask the model to continue. Works well when the expected output is long (think structured documents).

The second option only works if you've been checkpointing the partial output as it arrives.

# streaming.py — partial-response checkpointing for streams
from dataclasses import dataclass, field
from typing import AsyncIterator
from google import genai
 
@dataclass
class StreamCheckpoint:
    produced: str = ""
    finished: bool = False
    error: Exception | None = field(default=None, repr=False)
 
async def stream_with_checkpoint(
    client: genai.Client, model: str, prompt: str, checkpoint: StreamCheckpoint
) -> AsyncIterator[str]:
    """Stream tokens while accumulating them into checkpoint.produced."""
    try:
        async for chunk in await client.aio.models.generate_content_stream(
            model=model, contents=prompt,
        ):
            text = chunk.text or ""
            checkpoint.produced += text
            yield text
        checkpoint.finished = True
    except Exception as exc:
        checkpoint.error = exc
        raise
 
async def resume_if_interrupted(
    client: genai.Client, model: str, original_prompt: str
) -> str:
    cp = StreamCheckpoint()
    try:
        async for _ in stream_with_checkpoint(client, model, original_prompt, cp):
            pass
        return cp.produced
    except Exception:
        if not cp.produced:
            raise  # nothing arrived — a plain retry will do
        resume_prompt = (
            f"{original_prompt}\n\n"
            f"[The following is a partial response you started generating. "
            f"Continue naturally from where it left off, without repeating "
            f"any of it.]\n\n{cp.produced}"
        )
        cp2 = StreamCheckpoint()
        async for _ in stream_with_checkpoint(client, model, resume_prompt, cp2):
            pass
        return cp.produced + cp2.produced

The only real rule here is don't throw away what you've already shown the user. The checkpoint object is shared with the UI layer, so whatever got rendered on the client's screen is what the server continues from. For a deeper look at streaming patterns (including how to render tokens cleanly in the browser), I wrote a companion piece in Designing Antigravity streaming APIs in Python — real-time output without the pitfalls.

Rate Limits and Cost Budgets Belong in the Same Layer

The single costliest bug I've seen in production AI services looked like this: a retry loop kept re-firing an expensive model for eight hours because nobody had bounded it in currency. The retry-count cap existed. The currency cap did not.

The trick I reach for most often is a shared bucket that tracks RPM and currency simultaneously. RPM recovers with time; currency doesn't. Representing that asymmetry explicitly in your code prevents a whole class of accidents.

# budget.py — unified RPM + daily cost bucket
import asyncio
import time
from dataclasses import dataclass
 
@dataclass
class BudgetBucket:
    rpm_limit: int
    daily_usd_limit: float
    _tokens: float = 0.0
    _spent_today_usd: float = 0.0
    _last_refill: float = 0.0
    _lock: asyncio.Lock = None
 
    def __post_init__(self):
        self._tokens = float(self.rpm_limit)
        self._last_refill = time.monotonic()
        self._lock = asyncio.Lock()
 
    async def acquire(self, estimated_usd: float) -> None:
        async with self._lock:
            # Refill RPM tokens (rpm_limit tokens per 60s).
            now = time.monotonic()
            elapsed = now - self._last_refill
            self._tokens = min(
                self.rpm_limit,
                self._tokens + elapsed * (self.rpm_limit / 60.0),
            )
            self._last_refill = now
 
            # Currency cap: waiting doesn't help, so raise immediately.
            if self._spent_today_usd + estimated_usd > self.daily_usd_limit:
                raise RuntimeError(
                    f"Daily budget exceeded: "
                    f"{self._spent_today_usd:.2f} + {estimated_usd:.2f} "
                    f"> {self.daily_usd_limit:.2f}"
                )
 
            # RPM cap: waiting does help.
            while self._tokens < 1.0:
                sleep_for = (1.0 - self._tokens) * 60.0 / self.rpm_limit
                await asyncio.sleep(sleep_for)
                now = time.monotonic()
                elapsed = now - self._last_refill
                self._tokens = min(
                    self.rpm_limit,
                    self._tokens + elapsed * (self.rpm_limit / 60.0),
                )
                self._last_refill = now
 
            self._tokens -= 1.0
            self._spent_today_usd += estimated_usd

Estimate the cost before you call — it's usually fine to use input tokens × model rate — and then correct after the response returns, using the actual output token count. That gives you both a safe pre-check and accurate post-hoc accounting. For distributed deployments (multi-instance, multi-region), the bucket has to move out of process; a Redis-backed variant is sketched in Antigravity + Upstash Redis — edge cache and rate limiting that scales.

Observability — If You Can't See It, You Can't Fix It

Everything above only matters if, when it fails, you can figure out why it failed. A single log line saying "Timeout" is useless at 3 AM. The attributes I insist on capturing on every single call are:

• A request ID (linkable to the user's upstream trace ID)
• Model name, prompt hash (not the prompt itself — treat it as PII)
• Attempt number and per-attempt outcome
• Token usage: input, output, thinking (if applicable)
• Measured latencies: DNS, connect, TTFB, total
• Circuit breaker state transitions
• Estimated cost in your billing currency

In 2026 the obvious home for this is OpenTelemetry. Emit it as structured span attributes and send it to Grafana/Tempo/Loki or a hosted equivalent. Asking the Antigravity agent for "an OTel decorator that wraps Gen AI SDK calls" will generate a reasonable skeleton you can tune — I walk through a full build-out in Antigravity × OpenTelemetry — an AI observability pipeline from scratch.

If you only have time for three dashboards, make them: error type breakdown per minute, p50/p95 latency by model, and circuit breaker open count. Almost every incident I've investigated was legible from those three panels in the first 60 seconds. Everything else is gravy.

Retries and Side Effects — the Idempotency Problem

"A read-only call is safe to retry" is true as far as it goes, but the moment your pipeline writes anything — to a database, a queue, Slack, a payment processor — blanket retries become dangerous. Retrying a network failure is fine. Retrying a call that already succeeded (but whose response you didn't receive) means running the side effect twice.

I've personally shipped this bug. A batch job that summarized incoming tickets with Gen AI and then posted the summary to Slack hit a 429, retried, and the retry re-ran the whole block — including the Slack post that had already succeeded the first time. The channel ended up with duplicated summaries at suspicious intervals, and the root cause sat undiagnosed for two days because the AI call itself looked healthy in the logs.

There are two standard defenses. The first is an idempotency key: derive a stable key from (user, payload, timestamp bucket), and have your side-effect layer refuse to execute twice for the same key. The second is transactional separation: put only the AI call inside the retry loop, and let the side effect live outside, running exactly once per successful AI result.

# idempotency.py — minimal "run this work exactly once" helper
import hashlib
from typing import Awaitable, Callable, TypeVar
 
T = TypeVar("T")
 
class IdempotencyStore:
    """In production this is Redis or your DB — this shows the shape only."""
    def __init__(self):
        self._cache: dict[str, object] = {}
 
    async def run_once(self, key: str, fn: Callable[[], Awaitable[T]]) -> T:
        if key in self._cache:
            return self._cache[key]  # type: ignore[return-value]
        result = await fn()
        self._cache[key] = result
        return result
 
def request_idempotency_key(user_id: str, payload: str) -> str:
    return hashlib.sha256(f"{user_id}:{payload}".encode()).hexdigest()

Separating the AI call from the side effect is usually just a matter of await granularity. Keep the retry-wrapped AI call in one block, then hand the result to a second block that uses run_once with the idempotency key. Even if the outer request is re-driven, the side effect executes exactly once.

Testing Resilience the Same Way Production Hits It

Resilience features are functions that only run when something goes wrong. That means happy-path unit tests tell you almost nothing about whether they work. The only way to be sure is to inject the failures and observe the response.

The bare minimum I write for any service that talks to an LLM API:

• A test where the mock returns 503 twice and then 200 — asserts that call_with_retry eventually succeeds and that it attempted exactly three times
• A test where the mock always returns 400 — asserts zero retries and an immediate PermanentAIError
• A test where the mock returns 429 with Retry-After: 2 — asserts the elapsed sleep time is at least 2 seconds
• A test that fires five consecutive 503s — asserts the circuit breaker transitions to OPEN

# test_resilience.py — failure injection tests with pytest-asyncio + respx
import pytest
import respx
import httpx
 
from resilient_client import call_with_retry
 
@pytest.mark.asyncio
@respx.mock
async def test_retries_on_503_then_succeeds():
    route = respx.post("https://example.test/gen").mock(
        side_effect=[
            httpx.Response(503, text="Service Unavailable"),
            httpx.Response(503, text="Service Unavailable"),
            httpx.Response(200, json={"ok": True}),
        ]
    )
    async with httpx.AsyncClient() as client:
        req = client.build_request("POST", "https://example.test/gen")
        res = await call_with_retry(client, req, max_attempts=5)
    assert res.status_code == 200
    assert route.call_count == 3
 
@pytest.mark.asyncio
@respx.mock
async def test_no_retry_on_400():
    respx.post("https://example.test/gen").mock(
        return_value=httpx.Response(400, text="Bad Request")
    )
    async with httpx.AsyncClient() as client:
        req = client.build_request("POST", "https://example.test/gen")
        with pytest.raises(Exception):
            await call_with_retry(client, req, max_attempts=5)

These tests look boring, and that's the point. Boring, reliable failure-injection tests catch the regressions that production would otherwise catch for you at 3 AM. I also recommend having one end-to-end integration test that hits a staging endpoint configured to inject a 503 on every third request — it's the only way to verify that the full stack (retry + breaker + budget + observability) composes correctly under real network conditions.

One more practical tip: when you write these tests, don't reach for time.sleep or asyncio.sleep patching to fake out backoff delays. Let the delays actually happen (with small values — initial=0.05, max=0.2) so the test exercises the real clock path. Otherwise you end up with tests that pass because the scheduling code is mocked away, and real 429 sleeps in production go untested.

Common Mistakes That Quietly Burn Production Services

The scaffolding above gets you to "not on fire." Here's the short list of details I see teams miss even after they have the scaffolding:

• Skipping jitter. Exponential backoff without jitter is synchronized retry. Several of your instances will magically coordinate to take the API down on its way back up.
• Catching every exception as retryable. Classic trap. Prompt mistakes that return 400 get retried forever, quietly wasting money.
• Ignoring Retry-After. 429 is a conversation, not a slap. The header tells you exactly when to come back; honor it.
• Forgetting the outer deadline. Your retry budget is 30s, but there's no asyncio.timeout(45) around the whole request, so one unlucky user waits five minutes.
• Over-sharing connection pools. One global AsyncClient for every model mixes short and long requests; the long ones hold pool slots and choke the short ones.
• Logging only failures. Drop success logs and you lose the denominator; your "error rate" becomes a number of errors with no context.
• Per-worker circuit breakers. If you have 16 workers and each owns its own breaker, the first breaker to open doesn't stop the others from piling on.

Pre-Deploy Checklist

Before I push anything I run through this list. If even one item is "probably fine," I stop and investigate.

• I injected 429, 503, timeouts, and connection resets in local tests and confirmed each took the expected branch
• The outer asyncio.timeout is strictly longer than the inner retry budget, with a clear safety margin
• I triggered the circuit breaker by forcing five consecutive failures and watched it open, then recover
• The daily cost bucket fails closed in a test that intentionally exceeds the limit
• On a simulated 429, the client slept for exactly the Retry-After value (plus tiny jitter) and no more
• My dashboards show: error type over time, latency per model, breaker transitions
• I have a separate API key for chaos testing so I can safely induce failures in staging

One Concrete Next Step

If this article leaves you with energy for exactly one change, make it this: replace any except Exception in your AI call sites with a dispatch on RetryableAIError vs PermanentAIError. Nothing else in this article — not backoff, not breakers, not budgets — matters as much as this split. It takes less than an hour and removes the overwhelming majority of the "why did my service do that?" questions before they even form.

For a broader view of how resilience fits alongside concurrency and streaming design, my piece on Designing async concurrent pipelines for Antigravity + Python picks up where this one leaves off. May your service sleep through its nights, quietly answering requests while you do the same.