FastAPI 스트리밍 프록시

왜 프록시를 둘까

운영 앱이 Gemini 앞에 프록시를 두는 데는 두 가지 이유가 있어:

API 키를 숨겨. 브라우저는 자격 증명을 절대 보지 못해. 브라우저가 침해되어도 피해 범위가 "키로 Gemini를 직접 호출"에서 "자체 프록시를 호출"로 줄어들어.
비즈니스 로직을 더해. 인증, 호출 한도, 요청 검증, 응답 기록, 모델 선택을 서버 한곳에 모을 수 있어.

구현 패턴

FastAPI의 StreamingResponse와 httpx.AsyncClient.stream()을 함께 써. Gemini에 스트림을 열고 청크를 클라이언트로 전달한 뒤 끝나면 닫아. 작동하는 프록시는 약 30줄이면 충분해.

SSE 재전송과 그대로 전달하는 방식

설계는 두 가지야:

그대로 전달: Gemini의 바이트를 그대로 전달해. 가장 저렴하지만 클라이언트가 Gemini의 정확한 형식을 알아야 해.
자체 SSE 형식으로 다시 내보내기: 각 청크를 파싱해 text를 꺼내고 자체 data: {"text": "..."} 이벤트로 내보내. 손은 더 가지만 클라이언트를 Gemini의 변하는 스키마에서 떼어 놓을 수 있어.

Code

그대로 전달하는 프록시·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',
    )

자체 SSE 프로토콜로 다시 내보내기·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')

브라우저 — EventSource 소비자·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

두 번째 코드 블록처럼 다시 내보내는 프록시를 작성해. localhost:8000에서 실행하고, 텍스트 입력란과 버튼이 있는 작은 정적 HTML 페이지를 만들어 POST 요청을 보낸 뒤 스트리밍 답변을 토큰이 오는 대로 화면에 그려. 프록시에 청크당 50ms의 인공 지연을 넣고 UI가 부드럽게 갱신되는지 확인해. 청크가 실시간으로 와야지 한꺼번에 몰려오면 안 돼.