gemini-gateway 0.1.0

Shared Google Gemini API gateway with multi-key rate limiting and retries.

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, internal, unknown) using structured status codes first, with message-text fallback.
• Typed structured output — pass a Pydantic model and get a validated instance back.
• 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 or a local .env file.
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-2.0-flash"
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-2.0-flash	Model name.
_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	120000	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.2	Sampling temperature (0–2).
_RETRY_BASE_SECONDS	5.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.
_RETRY_CAPACITY_ERRORS_INDEFINITELY	false	If true, retry capacity (overload) errors without a retry limit.

Example .env (see .env.example):

GEMINI_API_KEYS=key1,key2,key3
GEMINI_MODEL=gemini-2.0-flash
GEMINI_RPM=15
GEMINI_TPM=250000
GEMINI_RPD=500
GEMINI_MAX_OUTPUT_TOKENS=2048

GeminiGatewayConfig.from_env() calls load_dotenv() by default, which reads a .env file into os.environ. Pass load_dotenv_file=False if you need strict control over the environment.

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-2.0-flash",
    api_keys=("key1", "key2"),
    requests_per_minute=10,
    tokens_per_minute=200_000,
    requests_per_day=400,
)
gateway = GeminiGateway(config)

How rate limiting works

Limits are enforced per key by MultiKeyRateLimiter before a request is sent. On each call the limiter picks the next eligible key in round-robin order, skipping any key that is disabled, cooling down, or would exceed its RPM, TPM, or RPD window. If every key is momentarily unavailable, the limiter sleeps until the soonest one frees up rather than overshooting quota.

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

Retries and error handling

Each failed attempt is classified by classify_api_error:

Category	Retryable	Notes
rate_limit	yes	Triggers a key cooldown; honors retryDelay hints.
network	yes	Connection/timeout/TLS errors.
server	yes	5xx / overload. Capacity errors are retryable indefinitely only when opted in.
auth	no	Disables the key and rotates to the next available one.
validation	no	Bad/empty/invalid output; raised to the caller.
internal	no	Programming errors (TypeError, KeyError, …).
unknown	yes	Conservatively retried.

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"), generate_text, generate_text_result, generate_json, generate_json_result
• GeminiGatewayConfig — from_env(prefix=..., env=..., load_dotenv_file=...)

Result types

• GeminiTextResult, GeminiJsonResult, GeminiUsage

Rate limiting

• MultiKeyRateLimiter, KeyState

Errors & classification

• classify_api_error, ApiErrorInfo
• GeminiGatewayError (base), NoApiKeysError, 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 mypy src
python -m pytest
python -m build
python -m twine check dist/*

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.