gemini-gateway

A small, dependency-light Python gateway for the Google Gemini API that adds the production concerns you usually end up writing yourself: multi-key rotation, client-side rate limiting, bounded retries, error classification, and typed structured output — behind a tiny, stable API.

It is deliberately scoped to reusable API plumbing. It contains no application-specific logic, so the same gateway can be shared across projects by giving each one its own environment-variable prefix.

Table of contents

• Why
• Features
• Requirements
• Installation
• Quick start
• Structured JSON output
• Result metadata
• Configuration
  • Environment variables
  • Per-project prefixes
  • Configuring in code
• How rate limiting works
• Retries and error handling
• Testing your own code
• Public API
• Development
• License

Why

Calling Gemini directly works until it doesn't: a key hits its per-minute quota, the backend returns a transient 503, or you want JSON back as a validated object instead of a string you have to parse by hand. gemini-gateway wraps the official google-genai client and handles those cases consistently, without leaking provider details into your application code.

Features

• Multi-key rotation — supply several API keys; requests are spread across them and a key is automatically skipped while it is cooling down or disabled.
• Client-side rate limiting — enforces requests-per-minute (RPM), tokens-per-minute (TPM), and requests-per-day (RPD) limits before sending, so you stay under quota instead of reacting to 429s.
• Bounded retries with backoff — transient failures are retried with exponential backoff and honor the API's retryDelay hint when present. Retry counts are bounded by default; unbounded retries on capacity errors are strict opt-in.
• Error classification — exceptions are categorized (rate_limit, network, server, auth, validation, model_output_validation, model_not_found, internal, unknown) using structured status codes first, with message-text fallback.
• Typed structured output — pass a Pydantic model and get a validated instance back. A response that does not parse is retried like any other transient failure, because the usual cause is JSON truncated by max_output_tokens.
• Usage metadata — optional result objects expose input/output token counts, the model used, and which key served the request.
• Testable by design — the network client, clock, and sleep function are all injectable, so your tests never touch the real API.

Requirements

• Python 3.10+
• A Google Gemini API key (get one here)

Installation

pip install gemini-gateway

Quick start

from gemini_gateway import GeminiGateway

# Reads GEMINI_API_KEY / GEMINI_API_KEYS (and other GEMINI_* vars) from the
# environment. Loads a local .env file first when python-dotenv is installed
# (pip install gemini-gateway[dotenv]).
gateway = GeminiGateway.from_env()

text = gateway.generate_text("Write a one-sentence product tagline for a coffee shop.")
print(text)

The minimum configuration is a single API key:

export GEMINI_API_KEY="your-key"

Structured JSON output

Pass a Pydantic model and receive a validated instance. The gateway requests JSON from Gemini using the model as the response schema and validates the result for you.

from pydantic import BaseModel
from gemini_gateway import GeminiGateway


class Product(BaseModel):
    name: str
    tagline: str
    price_usd: float


gateway = GeminiGateway.from_env()

product = gateway.generate_json(
    "Invent a fictional coffee product as JSON with name, tagline, price_usd.",
    Product,
    max_output_tokens=512,
)

print(product.name, product.price_usd)  # fully typed

If Gemini returns malformed or non-conforming JSON, a Pydantic ValidationError (a subclass of ValueError) is raised.

Result metadata

Use the *_result variants when you need token usage, the model name, or which key served the request (useful for logging and cost tracking).

result = gateway.generate_text_result("Summarize the theory of relativity.")

print(result.text)
print(result.usage.input_tokens, result.usage.output_tokens)
print(result.model)  # e.g. "gemini-3.1-flash-lite"
print(result.api_key_label)  # e.g. "key-2"

generate_json_result(...) returns the same metadata with a validated payload field instead of text.

Configuration

Environment variables

All variables are prefixed (default prefix: GEMINI). Only an API key is required; everything else has a sensible default.

Example environment variables:

GEMINI_API_KEYS=key1,key2,key3
GEMINI_MODEL=gemini-3.5-flash-lite
GEMINI_FALLBACK_MODELS=gemini-3.1-flash-lite
GEMINI_PROXY_URL=http://proxy.example:18888
GEMINI_RPM=15
GEMINI_TPM=250000
GEMINI_RPD=500
GEMINI_MAX_OUTPUT_TOKENS=2048

GeminiGateway.from_env() / GeminiGatewayConfig.from_env() load a local .env file when present if python-dotenv is installed (pip install gemini-gateway[dotenv]). Without that extra, they only read process environment variables. Pass load_dotenv_file=False to skip loading, or load_dotenv_file=True to require it (raises if python-dotenv is missing). Providing an explicit env= mapping never loads .env.

Per-project prefixes

Because the gateway is meant to be shared, each project can isolate its configuration with a custom prefix:

from gemini_gateway import GeminiGateway, GeminiGatewayConfig

config = GeminiGatewayConfig.from_env(prefix="PUBLISHER_GEMINI")
gateway = GeminiGateway(config)

This reads PUBLISHER_GEMINI_API_KEYS, PUBLISHER_GEMINI_MODEL, and so on.

Configuring in code

You can skip the environment entirely and build a config directly. Invalid values are rejected at construction time with a clear ValueError.

from gemini_gateway import GeminiGateway, GeminiGatewayConfig

config = GeminiGatewayConfig(
    model="gemini-3.1-flash-lite",
    api_keys=("key1", "key2"),
    requests_per_minute=10,
    tokens_per_minute=200_000,
    requests_per_day=400,
)
gateway = GeminiGateway(config)

Model fallback

fallback_models turns extra models into overflow capacity for the primary one:

config = GeminiGatewayConfig(
    model="gemini-3.5-flash-lite",
    fallback_models=("gemini-3.1-flash-lite",),
    api_keys=("key1", "key2"),
)

Models are tried in order, and a model is only skipped when no key can serve it right now. So the gateway exhausts gemini-3.5-flash-lite on every key before sending anything to gemini-3.1-flash-lite. GeminiTextResult.model / GeminiJsonResult.model report which model actually answered, and the key label becomes key-1/gemini-3.5-flash-lite so logs stay unambiguous.

Three failures move the chain forward without waiting:

• Quota (429) — the slot is out of allowance, another one is not.
• Overload (5xx, including the 503 … high demand a hot new model returns all day) — the next model is tried at once, and the switch does not count against max_retries. Otherwise a chain longer than max_retries could never reach its last models.
• Unknown model (404 … is not found for API version) — the model is dropped from the chain for the rest of the process, so one typo in FALLBACK_MODELS costs a single request instead of every retry.

A client-side timeout or a broken connection is not one of them: it says nothing about the model, so it keeps the normal backoff instead of replaying the same failure once per slot.

How rate limiting works

Limits are enforced per (key, model) pair by MultiKeyRateLimiter before a request is sent. Google grants free-tier quota per Cloud project and per model, so the pair is the smallest independent quota bucket: exhausting the daily quota of one model on one key leaves every other key — and every other model of the same key — untouched.

On each call the limiter walks models in configured order and keys in round-robin order, skipping any slot that is disabled, cooling down, or would exceed its RPM, TPM, or RPD window. If every slot is momentarily unavailable, the limiter sleeps until the soonest one frees up rather than overshooting quota.

Cooldown scope follows the error:

availability() returns a secret-free KeyAvailability snapshot — slot counts, the nearest next_available_at, and per-slot block reasons using labels only, so it is safe to log:

availability = gateway.rate_limiter.availability()
if not availability.has_available_key:
    logger.warning("waiting %.0fs: %s", availability.wait_seconds(), availability.describe())

Token accounting uses an estimate of prompt_length / 4 + max_output_tokens when you don't pass an explicit token_budget. An explicit token_budget must satisfy 1 <= token_budget <= tokens_per_minute. RPD is tracked as a rolling 24-hour window — an API-safe approximation, not a calendar-day reset at the provider's midnight.

Driving your own retry loop

Callers that own batching, persistence or splitting logic can use the limiter directly and skip the built-in retry loop:

state = gateway.rate_limiter.acquire(token_budget=12_000)
try:
    result = gateway.generate_json_on_slot(state, prompt, BatchOut, max_output_tokens=4096)
except Exception as exc:
    error_info = classify_api_error(exc)
    if error_info.should_cooldown_key:
        gateway.rate_limiter.cooldown_key_after_error(state.key, error_info, model=state.model)
    raise

Retries and error handling

Each failed attempt is classified by classify_api_error:

ApiErrorInfo.should_split is advisory: it tells a batching caller that retrying a smaller request is likely to help.

Retries use exponential backoff (retry_base_delay_seconds * 2**attempt, capped at retry_max_delay_seconds) and respect any server-provided delay hint. When all retries are exhausted, a GeminiRetriesExhaustedError is raised with the last underlying exception chained.

from gemini_gateway import GeminiGateway, GeminiRetriesExhaustedError, NoApiKeysError

try:
    gateway = GeminiGateway.from_env()
    text = gateway.generate_text("Hello!")
except NoApiKeysError:
    ...  # no keys configured
except GeminiRetriesExhaustedError as exc:
    ...  # transient failures persisted past max_retries; exc.__cause__ has details

Testing your own code

The network client, clock, and sleep function are injectable, so you can drive the gateway deterministically without any network access:

from types import SimpleNamespace
from gemini_gateway import GeminiGateway, GeminiGatewayConfig


class FakeModels:
    def generate_content(self, **kwargs):
        return SimpleNamespace(text="stubbed", usage_metadata=None)


class FakeClient:
    models = FakeModels()


config = GeminiGatewayConfig(model="gemini-test", api_keys=("k",))
gateway = GeminiGateway(
    config,
    client_factory=lambda api_key, timeout_ms: FakeClient(),
    sleep_fn=lambda _seconds: None,  # no real waiting
)

assert gateway.generate_text("hi") == "stubbed"

Public API

Everything below is exported from the top-level gemini_gateway package.

Entry points

• GeminiGateway — from_env(prefix="GEMINI", load_dotenv_file=None), generate_text, generate_text_result, generate_json, generate_json_result, generate_text_on_slot, generate_json_on_slot, count_tokens, rate_limiter
• GeminiGatewayConfig — from_env(prefix=..., env=..., load_dotenv_file=...), models

Result types

• GeminiTextResult, GeminiJsonResult, GeminiUsage

Rate limiting

• MultiKeyRateLimiter — acquire, has_ready_key, availability, cooldown_key, cooldown_key_after_error, disable_key, restore_disabled_keys
• KeyState, KeyAvailability

Errors & classification

• classify_api_error, ApiErrorInfo
• QUOTA_SCOPE_PER_KEY_DAILY, QUOTA_SCOPE_PER_KEY_WINDOW
• GeminiGatewayError (base), NoApiKeysError, AllApiKeysDisabledError, GeminiRetriesExhaustedError

The package ships a py.typed marker, so type checkers see the annotations.

Development

python -m pip install -e ".[dev]"
python -m ruff check .
python -m ruff format --check .
python -m pyright
python -m pytest
python -m build
python -m twine check dist/*

Type checking is done by Pyright in strict mode for src (configured in [tool.pyright]); tests are excluded to avoid mock-related false positives. Suppressions (# pyright: ignore[...]) must be narrow and explain why the code is valid.

Tests use fake clients and never call the real Gemini API. Contributions should keep the public API small and typed, and avoid adding application-specific logic.

License

Apache-2.0. See LICENSE.