FastAPI Streaming Proxy

Why proxy at all

Two reasons your production app probably wants a Gemini proxy in front:

Hide the API key. The browser never sees credentials. The blast radius of a compromised browser becomes "can call your proxy" instead of "can call Gemini directly with your key."
Add business logic. Auth, rate-limiting, request validation, response logging, model selection — all in one server-side place.

The pattern

FastAPI's StreamingResponse + httpx.AsyncClient.stream(). Open a stream to Gemini, forward chunks to the client, close when done. Total: ~30 lines for a working proxy.

SSE vs raw passthrough

Two designs:

Raw passthrough: forward Gemini's bytes directly. Cheapest, but client must speak Gemini's exact format.
Re-emit as your own SSE format: parse each chunk, extract text, emit your own data: {"text": "..."} events. More work but decouples client from Gemini's evolving schema.

Code

Raw passthrough proxy·python

import httpx
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse

app = FastAPI()
GEMINI_API_KEY = '...'  # from env in real code
BASE = 'https://generativelanguage.googleapis.com/v1beta'

@app.post('/v1beta/models/{model}:streamGenerateContent')
async def proxy_stream(model: str, request: Request):
    body = await request.body()
    url = f'{BASE}/models/{model}:streamGenerateContent?alt=sse'

    async def stream_gen():
        async with httpx.AsyncClient(timeout=120) as client:
            async with client.stream(
                'POST', url,
                headers={
                    'x-goog-api-key': GEMINI_API_KEY,
                    'Content-Type': 'application/json',
                },
                content=body,
            ) as response:
                async for chunk in response.aiter_bytes():
                    yield chunk

    return StreamingResponse(
        stream_gen(),
        media_type='text/event-stream',
    )

Re-emit as your own SSE protocol·python

import json
from google import genai
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()
client = genai.Client()  # uses env GEMINI_API_KEY

@app.post('/api/chat')
async def chat_stream(req: dict):
    prompt = req['prompt']

    async def stream_gen():
        try:
            async for chunk in await client.aio.models.generate_content_stream(
                model='gemini-2.5-flash', contents=prompt,
            ):
                if chunk.text:
                    payload = json.dumps({'type': 'text', 'data': chunk.text})
                    yield f'data: {payload}\n\n'
                if chunk.usage_metadata:
                    usage = {
                        'prompt': chunk.usage_metadata.prompt_token_count,
                        'completion': chunk.usage_metadata.candidates_token_count,
                    }
                    payload = json.dumps({'type': 'usage', 'data': usage})
                    yield f'data: {payload}\n\n'
            yield 'data: {"type": "done"}\n\n'
        except Exception as e:
            err = json.dumps({'type': 'error', 'data': str(e)})
            yield f'data: {err}\n\n'

    return StreamingResponse(stream_gen(), media_type='text/event-stream')

Browser side — EventSource consumer·typescript

const es = new EventSource('/api/chat?prompt=' + encodeURIComponent(prompt));

es.onmessage = (e) => {
  const msg = JSON.parse(e.data);
  if (msg.type === 'text') {
    appendToReply(msg.data);
  } else if (msg.type === 'usage') {
    showUsage(msg.data);
  } else if (msg.type === 'done') {
    es.close();
  } else if (msg.type === 'error') {
    showError(msg.data);
    es.close();
  }
};

Exercise

Build the re-emit proxy from the second code block. Run it on localhost:8000 and write a small static HTML page with a textarea + button that POSTs to it and renders the streamed reply token-by-token. Confirm chunks arrive in real time (not batched) by adding a 50ms artificial delay per chunk in the proxy and watching the UI update smoothly.