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, invalid_request, validation, model_output_validation, model_not_found, internal, unknown) using structured status codes first, with message-text fallback. Only transport-level concerns are classified here: application-domain failures belong to the calling project, which wraps classify_api_error with its own rules.
• 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.

Variable (with prefix)	Default	Description
_API_KEYS	—	Comma/newline/semicolon-separated list of keys.
_API_KEY	—	Single key (used if _API_KEYS is unset).
_MODEL	gemini-3.1-flash-lite	Primary model. May itself be a comma-separated list: the first entry is primary, the rest become fallbacks.
_FALLBACK_MODELS	—	Comma-separated fallback models, tried only when the primary is exhausted.
_PROXY_URL	—	Optional HTTP(S) proxy URL used only by Gemini API clients.
_RPM	15	Max requests per minute, per key.
_TPM	250000	Max tokens per minute, per key.
_RPD	500	Max requests per day, per key (rolling 24h window).
_TIMEOUT_MS	60000	Per-request HTTP timeout, in milliseconds.
_MAX_RETRIES	3	Max attempts for retryable errors.
_MAX_OUTPUT_TOKENS	2048	Default output token cap (overridable per call).
_TEMPERATURE	0.6	Sampling temperature (0–2).
_RETRY_BASE_SECONDS	2.0	Base delay for exponential backoff.
_RETRY_MAX_SECONDS	60.0	Maximum backoff delay.
_DEFAULT_COOLDOWN_SECONDS	5.0	Cooldown applied to a key after a limit/error with no hint.
_DAILY_QUOTA_COOLDOWN_SECONDS	10800.0	How long a (key, model) pair is parked after a per-day project quota.
_RESPONSE_SCHEMA_MODE	pydantic	pydantic sends the model type as response_schema; json_schema sends its raw JSON Schema.
_RETRY_CAPACITY_ERRORS_INDEFINITELY	false	If true, retry capacity (overload) errors without a retry limit.
_THINKING_LEVEL	не задан	Уровень размышлений Gemini 3 (minimal/low/medium/high); значение не валидируется и уходит модели как есть.
_THINKING_BUDGET	не задан	Устаревший потолок токенов размышлений (-1 — на усмотрение модели). Нельзя задавать вместе с _THINKING_LEVEL.

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:

Error	Scope	Duration
…PerDayPerProjectPerModel quota	the failing (key, model) pair	daily_quota_cooldown_seconds (hint is a lower bound)
RPM/TPM quota	the failing (key, model) pair	provider retryDelay, else 60 s
5xx / network	the failing (key, model) pair	provider retryDelay, else 5 s
401 / 403	the whole key, all its models	permanent (disable_key)
404 unknown model	the whole model, on every key	permanent (disable_model)

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. Block reasons never carry the provider's message text: a disabled slot reports describe_exception() output (exception type, category and HTTP status), because an SDK message may quote the request URL, its parameters or a fragment of the prompt:

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:

Category	Retryable	Notes
rate_limit	yes	Cools down the (key, model) pair; honors retryDelay hints. quota_scope / quota_id say whether the daily project quota was hit.
network	yes	Connection/timeout/TLS errors. A status reported by the API wins over the message text, so 503 Service temporarily unavailable is server, not network.
server	yes	5xx / overload. Capacity errors are retryable indefinitely only when opted in.
empty_response	yes	Model returned no text; retried on the next slot.
content_policy	mixed	RECITATION is retried, PROHIBITED_CONTENT is not; both hint should_split.
auth	no	Disables the key (all its models) and rotates to the next one.
invalid_request	no	Deterministic 400 INVALID_ARGUMENT: the request itself is malformed or over the model's input limit. Not retried and no cooldown — the key is healthy. Hints should_split, because an over-long prompt is the one cause a smaller request can fix.
validation	no	Gateway configuration errors (no API keys).
internal	no	Programming errors (TypeError, KeyError, …).
no_active_keys	no	Every key is disabled.
unknown	yes	Conservatively retried.

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, disable_model, key_label, restore_disabled_keys
• KeyState, KeyAvailability

Errors & classification

• classify_api_error, describe_exception, 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.