⟐ Dev Tools/2026-03-27Advanced

Gemini 3.1 Pro × Cloud Run: Building Production Serverless AI APIs

Deploy Gemini 3.1 Pro on Cloud Run with SSE streaming, auto-scaling, cold start optimization, and production monitoring — the definitive guide to building serverless AI APIs.

gemini-api²⁸⁴ cloud-run⁶ serverless³ production¹⁴⁰ streaming²⁸ deploy google-cloud⁶

✦ Premium Article

Where Scale-to-Zero Meets a Large Model

Gemini 3.1 Pro is Google's latest flagship model, featuring a 1-million-token context window and output speeds of 114 tokens per second. Serving this model reliably in production requires thoughtful infrastructure design.

Google Cloud Run is a container-based serverless platform that makes it an excellent choice for hosting AI APIs. It scales down to zero when idle — keeping costs minimal — and automatically scales out when traffic spikes.

Who This Guide Is For

Developers shipping Gemini-powered services to production
Backend engineers exploring Cloud Run for AI service deployment
Solo developers and startups looking to balance cost efficiency with scalability

Prerequisites

Basic Google Cloud project management
Docker and container fundamentals
Python (FastAPI) or Node.js basics

Architecture Design for Cloud Run × Gemini API

When serving a Gemini API through Cloud Run, the recommended architecture looks like this:

Client → Cloud Load Balancer → Cloud Run Service → Gemini API
                                      ↓
                               Cloud Logging / Monitoring
                                      ↓
                               Secret Manager (API Keys)

Why Cloud Run?

There are three compelling reasons to choose Cloud Run for this use case.

1. Scale to zero. When there are no incoming requests, you pay nothing. For indie developers and startups, eliminating fixed infrastructure costs is a game-changer.

2. Automatic scaling. Instances spin up and down based on concurrent request count. A sudden traffic spike from a viral post is handled with a single configuration setting.

3. Managed SSL and domains. Custom domains and SSL certificates are provided out of the box, minimizing infrastructure overhead.

✦

Thank you for reading this far.

Continue Reading

What follows includes implementation code, benchmarks, and practical content we hope you'll find useful. This site runs without ads — server and development costs are supported entirely by members like you. If it's been helpful, we'd be truly grateful for your support.

WHAT YOU'LL LEARN

✦Master stable SSE streaming patterns for the Gemini API running on Cloud Run

✦Learn production-critical tuning techniques including cold start optimization and minimum instance configuration

✦Build a comprehensive monitoring stack combining Cloud Logging, Error Reporting, and cost alerts

Secure payment via Stripe · Cancel anytime

✦

Unlock This Article

Get full access to the rest of this article. Buy once, read anytime. This site is ad-free — your support goes directly toward keeping it running.

Unlock all articles with Membership →

Project Setup

Let's start by building the FastAPI application that we'll deploy to Cloud Run.

# requirements.txt
fastapi==0.115.0
uvicorn[standard]==0.32.0
google-genai==1.14.0
sse-starlette==2.2.1
google-cloud-logging==3.11.0
google-cloud-secret-manager==2.21.0

# main.py
import os
import json
import logging
from contextlib import asynccontextmanager
 
from fastapi import FastAPI, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from sse_starlette.sse import EventSourceResponse
from google import genai
 
# --- Logging setup ---
if os.getenv("K_SERVICE"):  # Detect Cloud Run environment
    import google.cloud.logging
    client = google.cloud.logging.Client()
    client.setup_logging()
 
logger = logging.getLogger(__name__)
 
# --- Initialize Gemini client once at startup ---
gemini_client = None
 
@asynccontextmanager
async def lifespan(app: FastAPI):
    """Initialize the Gemini client on application startup"""
    global gemini_client
    api_key = os.environ.get("GEMINI_API_KEY")
    if not api_key:
        raise RuntimeError("GEMINI_API_KEY is not set")
    gemini_client = genai.Client(api_key=api_key)
    logger.info("Gemini client initialized successfully")
    yield
    logger.info("Application shutting down")
 
app = FastAPI(title="Gemini AI API", lifespan=lifespan)
 
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Restrict appropriately in production
    allow_methods=["POST"],
    allow_headers=["*"],
)

The key design decision here is initializing the Gemini client once during the lifespan event. Initializing it per-request adds connection overhead that compounds quickly under load.

Implementing SSE Streaming

Getting SSE streaming right on Cloud Run requires careful attention to response buffering.

# main.py (continued)
from pydantic import BaseModel, Field
from typing import Optional
 
class GenerateRequest(BaseModel):
    prompt: str = Field(..., min_length=1, max_length=30000)
    model: str = Field(default="gemini-3.1-pro-preview")
    max_tokens: Optional[int] = Field(default=8192, ge=1, le=65000)
    temperature: Optional[float] = Field(default=0.7, ge=0.0, le=2.0)
    stream: bool = Field(default=True)
 
@app.post("/v1/generate")
async def generate(req: GenerateRequest):
    """Call the Gemini API and return SSE streaming response"""
    if req.stream:
        return EventSourceResponse(
            stream_generate(req),
            media_type="text/event-stream",
            headers={
                "Cache-Control": "no-cache",
                "X-Accel-Buffering": "no",  # Disable Nginx buffering
            },
        )
    else:
        return await batch_generate(req)
 
async def stream_generate(req: GenerateRequest):
    """SSE streaming generator"""
    try:
        response = gemini_client.models.generate_content_stream(
            model=req.model,
            contents=req.prompt,
            config=genai.types.GenerateContentConfig(
                max_output_tokens=req.max_tokens,
                temperature=req.temperature,
            ),
        )
 
        total_tokens = 0
        for chunk in response:
            if chunk.text:
                total_tokens += len(chunk.text.split())
                yield {
                    "event": "chunk",
                    "data": json.dumps({
                        "text": chunk.text,
                        "done": False,
                    }),
                }
 
        # Completion signal
        yield {
            "event": "done",
            "data": json.dumps({
                "text": "",
                "done": True,
                "usage": {"approximate_tokens": total_tokens},
            }),
        }
 
    except Exception as e:
        logger.error(f"Stream generation failed: {e}", exc_info=True)
        yield {
            "event": "error",
            "data": json.dumps({"error": str(e)}),
        }
 
async def batch_generate(req: GenerateRequest):
    """Non-streaming (batch) response"""
    try:
        response = gemini_client.models.generate_content(
            model=req.model,
            contents=req.prompt,
            config=genai.types.GenerateContentConfig(
                max_output_tokens=req.max_tokens,
                temperature=req.temperature,
            ),
        )
        return {
            "text": response.text,
            "usage": {
                "prompt_tokens": response.usage_metadata.prompt_token_count,
                "output_tokens": response.usage_metadata.candidates_token_count,
            },
        }
    except Exception as e:
        logger.error(f"Batch generation failed: {e}", exc_info=True)
        raise HTTPException(status_code=500, detail=str(e))

Critical SSE Considerations

The most important detail when running SSE on Cloud Run is disabling response buffering. Without the X-Accel-Buffering: no header, Cloud Run's reverse proxy will buffer response chunks and deliver them in batches — defeating the purpose of streaming.

Also watch the Cloud Run request timeout (default: 300 seconds). Generating 65,000 tokens of output can take several minutes. You can extend this up to 3,600 seconds.

Dockerfile and Cold Start Optimization

Cold start time directly impacts user experience. The following Dockerfile uses multi-stage builds and pre-installed dependencies to minimize startup time.

# ---- Build stage ----
FROM python:3.12-slim AS builder
 
WORKDIR /build
COPY requirements.txt .
RUN pip install --no-cache-dir --target=/build/deps -r requirements.txt
 
# ---- Runtime stage ----
FROM python:3.12-slim
 
# Security: run as non-root user
RUN groupadd -r appuser && useradd -r -g appuser appuser
 
WORKDIR /app
COPY --from=builder /build/deps /usr/local/lib/python3.12/site-packages/
COPY main.py .
 
# Cloud Run automatically sets the PORT environment variable
ENV PORT=8080
ENV PYTHONUNBUFFERED=1
 
# Health check endpoint
HEALTHCHECK --interval=30s --timeout=5s \
  CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8080/health')"
 
USER appuser
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080", "--workers", "1"]

Three Strategies to Minimize Cold Starts

1. Set minimum instances. For services with predictable traffic, set --min-instances=1 or higher. You'll pay for always-on capacity, but cold starts are completely eliminated.

gcloud run deploy gemini-api \
  --min-instances=1 \
  --max-instances=10 \
  --concurrency=80 \
  --cpu=2 \
  --memory=1Gi \
  --timeout=600

2. Enable CPU boost. Cloud Run's "startup CPU boost" temporarily increases CPU allocation during startup, cutting initialization time significantly.

gcloud run deploy gemini-api \
  --cpu-boost

3. Minimize image size. Using python:3.12-slim as the base and avoiding unnecessary packages keeps pull time short. Aim for a final image size under 200MB.

API Key Management with Secret Manager

Never embed Gemini API keys in container images or source code. Cloud Run provides native integration with Secret Manager.

# Create the secret
echo -n "YOUR_GEMINI_API_KEY" | \
  gcloud secrets create gemini-api-key --data-file=-
 
# Mount it in your Cloud Run service
gcloud run deploy gemini-api \
  --set-secrets="GEMINI_API_KEY=gemini-api-key:latest"

This automatically populates the GEMINI_API_KEY environment variable with the latest secret value. When you rotate keys, just update the secret version — no redeployment needed.

For a deep dive into production security patterns, see our Gemini API Production Security Guide.

Building a Production Monitoring Stack

Request Logging with Cloud Logging

Cloud Run automatically forwards stdout to Cloud Logging. The google-cloud-logging library we configured earlier enables structured log entries.

# Add this middleware to main.py
import time
 
@app.middleware("http")
async def log_requests(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    duration = time.time() - start
 
    logger.info(
        "API request completed",
        extra={
            "json_fields": {
                "method": request.method,
                "path": request.url.path,
                "status": response.status_code,
                "duration_ms": round(duration * 1000, 2),
                "model": request.state.model if hasattr(request.state, "model") else "unknown",
            }
        },
    )
    return response

Leveraging Error Reporting

Exceptions in Cloud Run are automatically aggregated in Cloud Error Reporting. You get error grouping, notifications, and trend analysis with zero additional configuration.

However, Gemini API rate limit errors (429) are expected behavior. To keep them from cluttering Error Reporting, handle them explicitly with retry logic:

from google.api_core.exceptions import ResourceExhausted
 
async def stream_generate_with_retry(req: GenerateRequest, max_retries: int = 3):
    """Retry with exponential backoff"""
    for attempt in range(max_retries):
        try:
            async for chunk in stream_generate(req):
                yield chunk
            return
        except ResourceExhausted:
            if attempt < max_retries - 1:
                wait = 2 ** attempt
                logger.warning(f"Rate limited, retrying in {wait}s (attempt {attempt + 1})")
                await asyncio.sleep(wait)
            else:
                yield {
                    "event": "error",
                    "data": json.dumps({"error": "Rate limit exceeded. Please try again later."}),
                }

For comprehensive error handling strategies, check out the Gemini API Error Handling & Retry Patterns Guide.

Setting Up Cost Alerts

Cloud Run billing is based on request processing time multiplied by CPU and memory allocation. Combined with Gemini API costs, you'll want alerts to catch unexpected spending.

# Create a budget alert (notify when exceeding $50/month)
gcloud billing budgets create \
  --billing-account=YOUR_BILLING_ACCOUNT \
  --display-name="Gemini API Cloud Run Budget" \
  --budget-amount=50USD \
  --threshold-rule=percent=80 \
  --threshold-rule=percent=100

Performance Tuning

Optimizing Concurrency

Cloud Run's concurrency setting determines how many requests a single instance handles simultaneously. Since Gemini API calls are I/O-bound, you can set relatively high values (50–100).

# Recommended settings
gcloud run deploy gemini-api \
  --concurrency=80 \
  --cpu=2 \
  --memory=1Gi

Keep in mind that streaming requests hold connections open for extended periods, which increases per-instance memory consumption at high concurrency. Use load testing to find your optimal value.

Response Caching

Caching responses for identical prompts dramatically reduces API costs and latency.

from functools import lru_cache
import hashlib
 
# In-memory cache (non-streaming only)
_cache = {}
 
def get_cache_key(prompt: str, model: str, temperature: float) -> str:
    """Use prompt hash as cache key"""
    content = f"{model}:{temperature}:{prompt}"
    return hashlib.sha256(content.encode()).hexdigest()
 
async def cached_generate(req: GenerateRequest):
    """Cached generation (only when temperature=0)"""
    if req.temperature == 0:
        key = get_cache_key(req.prompt, req.model, req.temperature)
        if key in _cache:
            logger.info("Cache hit")
            return _cache[key]
 
    result = await batch_generate(req)
 
    if req.temperature == 0:
        _cache[key] = result
 
    return result

For production environments, consider using Redis (Cloud Memorystore) as your cache backend. It enables cache sharing across instances and supports automatic TTL-based expiration.

CI/CD Pipeline

Use Cloud Build to automate deployment on every push to GitHub.

# cloudbuild.yaml
steps:
  # Run tests
  - name: 'python:3.12-slim'
    entrypoint: 'bash'
    args:
      - '-c'
      - |
        pip install -r requirements.txt
        pip install pytest httpx
        pytest tests/ -v
 
  # Build and push Docker image
  - name: 'gcr.io/cloud-builders/docker'
    args: ['build', '-t', 'gcr.io/$PROJECT_ID/gemini-api:$SHORT_SHA', '.']
 
  - name: 'gcr.io/cloud-builders/docker'
    args: ['push', 'gcr.io/$PROJECT_ID/gemini-api:$SHORT_SHA']
 
  # Deploy to Cloud Run
  - name: 'gcr.io/cloud-builders/gcloud'
    args:
      - 'run'
      - 'deploy'
      - 'gemini-api'
      - '--image=gcr.io/$PROJECT_ID/gemini-api:$SHORT_SHA'
      - '--region=asia-northeast1'
      - '--platform=managed'
      - '--min-instances=1'
      - '--max-instances=10'
      - '--concurrency=80'
      - '--cpu=2'
      - '--memory=1Gi'
      - '--timeout=600'
      - '--cpu-boost'
      - '--set-secrets=GEMINI_API_KEY=gemini-api-key:latest'
 
options:
  logging: CLOUD_LOGGING_ONLY

Blue-Green Deployments

Cloud Run supports revision-based deployments. You can gradually shift traffic to new revisions to minimize risk.

# Deploy new revision with no traffic
gcloud run deploy gemini-api \
  --image=gcr.io/PROJECT_ID/gemini-api:v2 \
  --no-traffic
 
# Route 10% of traffic to the new revision
gcloud run services update-traffic gemini-api \
  --to-revisions=gemini-api-v2=10
 
# If everything looks good, shift to 100%
gcloud run services update-traffic gemini-api \
  --to-latest

Summary

Cloud Run paired with Gemini 3.1 Pro is an excellent combination for building serverless AI APIs that balance cost efficiency with scalability. By combining the SSE streaming, cold start optimization, Secret Manager integration, production monitoring, and CI/CD pipeline patterns covered in this guide, you can build robust AI APIs that scale from solo projects to enterprise workloads.

Thank You for Reading

Gemini Lab is ad-free, supported entirely by members like you. We publish practical guides daily with implementation code, benchmarks, and production-ready patterns. If you've found it useful, we'd love to have you on board.