FastAPI + Claude API 스트리밍 백엔드 — SSE·재시도·에러 복구 프로덕션 가이드

AI 백엔드를 구축하다 보면 결국 하나의 질문에 부딪힌다. “응답이 다 생성될 때까지 사용자를 기다리게 해도 괜찮을까?” 답은 대부분 “아니오”다. 특히 Claude 같은 언어 모델이 긴 텍스트를 생성할 때, 전체를 기다렸다가 한 번에 내려주는 방식은 UX를 죽인다.

직접 서비스에 연동해보면서 느낀 건, 스트리밍 자체는 어렵지 않다. 진짜 문제는 그 주변이다. 레이트 리밋에 걸렸을 때 어떻게 할 것인가, 에러를 어떻게 분류하고 각각 다르게 처리할 것인가, Nginx 뒤에서 SSE가 제대로 흐르게 하려면 어떤 헤더가 필요한가. 이 글은 FastAPI 0.136과 Anthropic SDK 0.97 기준으로 그 실전 패턴을 직접 구현하고 테스트한 결과를 정리한 것이다.

시작 전에 필요한 것

• Python 3.11 이상 (3.12 권장)
• Anthropic API 키 (ANTHROPIC_API_KEY)
• 기본 FastAPI / asyncio 개념

의존성은 네 가지만 필요하다:

pip install fastapi uvicorn anthropic httpx

Python 환경 구성이 처음이라면, uv로 Python AI 개발 환경 설정하는 방법을 먼저 보는 게 낫다. 가상환경과 의존성 충돌 문제를 깔끔하게 해결해준다.

Step 1: 프로젝트 구조와 기본 설정

먼저 디렉터리를 잡는다:

claude-streaming-api/
├── main.py          # FastAPI 앱 + 엔드포인트
├── retry.py         # 재시도 로직
├── .env             # API 키 (gitignore)
├── Dockerfile
└── docker-compose.yml

main.py의 기본 뼈대:

import os
import anthropic
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

app = FastAPI(title="Claude Streaming API", version="1.0.0")

client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))


class ChatRequest(BaseModel):
    message: str
    max_tokens: int = 1024
    system: str = "You are a helpful assistant."

Pydantic BaseModel로 요청 스키마를 정의하면 FastAPI가 자동으로 입력 검증과 OpenAPI 문서를 생성한다. 아래 이미지처럼 Swagger UI가 자동으로 생성된 것을 확인할 수 있다.

실제로 로컬에서 uvicorn main:app --reload를 실행하면 /docs에서 Swagger UI로 바로 테스트할 수 있다. 이 편리함이 FastAPI를 선택한 주된 이유 중 하나다.

Step 2: SSE 스트리밍 엔드포인트 구현

Server-Sent Events(SSE)는 HTTP 위에서 단방향 실시간 스트림을 보내는 가장 단순한 방법이다. WebSocket보다 구현이 간단하고, Claude처럼 서버에서 클라이언트로 텍스트를 흘려보내는 패턴에 딱 맞는다.

핵심은 FastAPI의 StreamingResponse와 Anthropic SDK의 stream() 컨텍스트 매니저를 조합하는 것이다:

import asyncio
import json
from typing import AsyncGenerator


async def stream_claude(request: ChatRequest) -> AsyncGenerator[str, None]:
    """Claude API 스트리밍 → SSE 이벤트 제너레이터"""
    try:
        with client.messages.stream(
            model="claude-opus-4-7-20251101",
            max_tokens=request.max_tokens,
            system=request.system,
            messages=[{"role": "user", "content": request.message}],
        ) as stream:
            for text in stream.text_stream:
                # SSE 포맷: "data: {...}\n\n"
                yield f"data: {json.dumps({'text': text, 'type': 'delta'})}\n\n"

            yield f"data: {json.dumps({'type': 'done'})}\n\n"

    except anthropic.RateLimitError:
        yield f"data: {json.dumps({'type': 'error', 'error': 'rate_limit', 'retry_after': 30})}\n\n"
    except anthropic.AuthenticationError:
        yield f"data: {json.dumps({'type': 'error', 'error': 'auth_error'})}\n\n"
    except Exception as e:
        yield f"data: {json.dumps({'type': 'error', 'error': 'unknown', 'message': str(e)})}\n\n"


@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
    return StreamingResponse(
        stream_claude(request),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",  # Nginx 버퍼링 해제 필수
        },
    )

실제 SSE 응답 스트림을 curl로 테스트하면 이렇게 흐른다:

$ curl -sN -X POST http://localhost:8000/chat/stream \
       -H "Content-Type: application/json" \
       -d '{"message": "FastAPI와 Claude를 결합하는 방법을 알려줘"}'

data: {"type": "delta", "text": "FastAPI"}
data: {"type": "delta", "text": "와 "}
data: {"type": "delta", "text": "Claude"}
...
data: {"type": "done"}

SSE 이벤트의 포맷 규칙은 단순하다: data: 접두사 + JSON + 두 번의 줄바꿈(\n\n). 이 포맷을 지키면 브라우저의 EventSource API나 대부분의 SSE 클라이언트가 자동으로 파싱한다.

한 가지 주의할 점: anthropic.Anthropic() 클라이언트의 messages.stream()은 동기 컨텍스트 매니저다. 비동기 FastAPI 라우트 안에서 블로킹 없이 실행하려면 Anthropic SDK의 AsyncAnthropic을 쓰는 게 더 정확하다:

client = anthropic.AsyncAnthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))

async def stream_claude(request: ChatRequest) -> AsyncGenerator[str, None]:
    async with client.messages.stream(...) as stream:
        async for text in stream.text_stream:
            yield f"data: {json.dumps({'text': text, 'type': 'delta'})}\n\n"

AsyncAnthropic을 쓰면 uvicorn의 이벤트 루프를 블로킹하지 않는다. 트래픽이 많지 않은 초기 프로젝트에서는 동기 클라이언트도 문제없이 작동하지만, 프로덕션에서는 비동기 클라이언트를 쓰는 게 맞다.

Step 3: 에러 분류와 재시도 전략

AI API 에러를 전부 같은 방식으로 처리하면 안 된다. 에러마다 올바른 행동이 다르기 때문이다:

레이트 리밋과 네트워크 에러만 재시도하는 지수 백오프 함수:

MAX_RETRIES = 3
BASE_DELAY = 1.0  # seconds


async def call_with_retry(fn, *args, **kwargs):
    """지수 백오프 재시도 — rate_limit과 network_error만 재시도"""
    for attempt in range(MAX_RETRIES):
        try:
            return await fn(*args, **kwargs)
        except anthropic.RateLimitError as e:
            if attempt == MAX_RETRIES - 1:
                raise  # 마지막 시도에서도 실패하면 전파
            delay = BASE_DELAY * (2 ** attempt)
            print(f"[retry] rate_limit, waiting {delay}s (attempt {attempt + 1}/{MAX_RETRIES})")
            await asyncio.sleep(delay)
        except anthropic.APIConnectionError:
            if attempt == MAX_RETRIES - 1:
                raise
            await asyncio.sleep(BASE_DELAY * (2 ** attempt))
        except (anthropic.AuthenticationError, anthropic.BadRequestError):
            raise  # 재시도해도 의미 없는 에러는 즉시 전파

이 패턴을 직접 테스트했을 때, 2번 실패 후 3번째 시도에서 성공하는 flaky API를 시뮬레이션한 결과 Result: success (after 3 attempts)로 정상 동작을 확인했다.

솔직히 말하면, 재시도 로직에서 가장 신경 쓰이는 부분은 MAX_RETRIES와 BASE_DELAY 값이다. 레이트 리밋은 Anthropic의 플랜마다 다르고, 재시도 간격이 너무 짧으면 같은 레이트 리밋에 다시 걸린다. 나는 API 플랜에 따라 이 값을 환경 변수로 외부화하는 걸 권장한다.

Step 4: 헬스체크와 프로덕션 배포

Kubernetes나 ECS 같은 컨테이너 환경에서는 헬스체크 엔드포인트가 필수다:

import time


@app.get("/health")
async def health_check():
    """K8s readiness / liveness probe용"""
    return {"status": "ok", "timestamp": time.time()}

Docker 이미지:

FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

Nginx 리버스 프록시 설정에서 SSE가 제대로 흐르려면 버퍼링을 반드시 꺼야 한다:

location /chat/stream {
    proxy_pass         http://backend:8000;
    proxy_buffering    off;           # SSE 버퍼링 비활성화 필수
    proxy_cache        off;
    proxy_set_header   Connection     '';
    proxy_http_version 1.1;
    proxy_read_timeout 300s;          # 긴 스트리밍 세션 허용
    chunked_transfer_encoding on;
}

proxy_buffering off를 빠뜨리면 Nginx가 스트림을 전부 버퍼에 모은 뒤 한꺼번에 내보낸다. 그러면 스트리밍이 아니라 그냥 느린 응답이 된다. 이 설정은 처음 SSE를 Nginx 뒤에 붙이는 사람이 거의 반드시 한 번씩 겪는 문제다.

docker-compose로 로컬에서 전체 스택을 올릴 때:

version: '3.9'
services:
  api:
    build: .
    environment:
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
    ports:
      - "8000:8000"
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

Step 5: 클라이언트 연동: 브라우저 EventSource와 Python

브라우저 (JavaScript):

const source = new EventSource('/chat/stream');
// EventSource는 GET 전용 — POST 요청은 fetch + ReadableStream 필요

// POST로 SSE를 연동하려면 fetch 사용
const response = await fetch('/chat/stream', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({ message: '안녕하세요' }),
});

const reader = response.body.getReader();
const decoder = new TextDecoder();

while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  
  const text = decoder.decode(value);
  const lines = text.split('\n\n').filter(l => l.startsWith('data:'));
  
  for (const line of lines) {
    const data = JSON.parse(line.slice(6));
    if (data.type === 'delta') {
      outputElement.textContent += data.text;
    }
  }
}

Python (httpx):

import httpx
import json

async def stream_chat(message: str):
    async with httpx.AsyncClient() as client:
        async with client.stream(
            "POST",
            "http://localhost:8000/chat/stream",
            json={"message": message},
            timeout=60.0,
        ) as response:
            async for line in response.aiter_lines():
                if line.startswith("data:"):
                    event = json.loads(line[6:])
                    if event["type"] == "delta":
                        print(event["text"], end="", flush=True)

Vercel AI SDK를 사용하는 프론트엔드가 있다면 Vercel AI SDK로 Claude 스트리밍 에이전트 만들기를 참고하면 프론트엔드 연동을 더 빠르게 처리할 수 있다. useChat 훅이 SSE 파싱을 대신 해줘서 클라이언트 코드가 훨씬 단순해진다.

아쉬운 점과 실제로 막히는 부분

내가 이 스택을 실제 프로젝트에서 써보면서 느낀 한계를 솔직하게 정리한다.

첫째, 스트리밍과 프롬프트 캐싱의 조합이 까다롭다. Claude API의 프롬프트 캐싱은 입력 토큰 비용을 크게 줄여준다. 그런데 스트리밍과 캐싱을 동시에 쓸 때 캐시 히트 여부를 스트림 중간에 알 수가 없다. 스트리밍 완료 후 usage 객체에서 확인할 수 있지만, 실시간으로 캐시 상태를 반영해야 하는 UI라면 구현이 복잡해진다. Claude API 프롬프트 캐싱으로 비용 최적화하기에서 캐싱 전략을 미리 파악해두는 게 좋다.

둘째, uvicorn의 워커 수와 커넥션 관리가 생각보다 복잡하다. SSE는 연결을 오래 유지한다. --workers 4로 4개 워커를 쓴다면 동시에 최대 4개의 긴 스트리밍 연결이 가능하다. 실제 트래픽이 이를 초과하면 요청이 대기한다. K8s로 수평 확장하거나, gunicorn + uvicorn worker class 조합을 써야 한다.

셋째, 재시도 로직이 스트리밍 중간에 끼면 처리가 복잡하다. 스트리밍이 절반쯤 진행됐을 때 네트워크 에러가 나면 어떻게 할 것인가? 처음부터 다시 요청하면 클라이언트는 이미 받은 텍스트가 중복된다. 실용적인 해결책은 클라이언트 측에서 last-event-id를 관리하고, 서버가 이를 받아 이어서 생성하는 것인데, 이 구현은 이 글의 범위를 벗어난다.

이 패턴이 스트리밍 응답이 불필요한 대량 처리 시나리오에는 오버엔지니어링이다. 1,000개의 문서를 배치로 처리한다면 Anthropic Message Batches API가 훨씬 저렴하고 적합하다.

성능 모니터링: 스트리밍에서 무엇을 측정해야 하는가

스트리밍 API에서는 전통적인 응답 시간 외에 추가적인 지표가 필요하다. 내가 실제 서비스에서 모니터링하는 핵심 지표 세 가지:

TTFT (Time to First Token): 요청을 보낸 시점부터 첫 번째 토큰이 도착하는 데 걸리는 시간. 사용자가 “응답이 빠르다”고 느끼는 가장 직관적인 지표다. Claude API에서 TTFT는 보통 0.5초〜2초 사이다. 이 값이 5초를 넘기 시작하면 뭔가 잘못된 것이다.

TPS (Tokens Per Second): 스트리밍 중 초당 생성되는 토큰 수. 사용자가 텍스트를 읽는 속도는 보통 분당 200〜300단어인데, TPS가 너무 빠르면 UI가 텍스트를 따라잡지 못한다. 반대로 너무 느리면 “끊겨 보인다”는 피드백이 온다.

스트리밍 에러율: 전체 스트리밍 요청 중 done 이벤트 없이 종료된 비율. 클라이언트 측 연결 끊김과 서버 측 에러를 구분해서 추적해야 근본 원인을 찾을 수 있다.

간단한 미들웨어로 이 지표들을 로그로 남길 수 있다:

import time
from fastapi import Request


@app.middleware("http")
async def log_streaming_metrics(request: Request, call_next):
    if request.url.path == "/chat/stream":
        start = time.perf_counter()
        response = await call_next(request)
        elapsed = time.perf_counter() - start
        # 실제 스트리밍 완료 시간은 미들웨어에서 측정하기 어렵다
        # 제너레이터 내부에서 done 이벤트 전 시간 기록을 추천
        print(f"[metrics] stream_request elapsed={elapsed:.2f}s")
        return response
    return await call_next(request)

OpenTelemetry와 연결하면 분산 트레이싱으로 훨씬 정교하게 추적할 수 있지만, 시작 단계에서는 이 정도 로그도 충분히 유용하다. Anthropic SDK vs OpenAI SDK 비교에서 Anthropic SDK의 usage 객체를 통해 토큰 사용량을 어떻게 추적하는지 더 자세히 살펴볼 수 있다.

트러블슈팅 FAQ

Q: SSE가 클라이언트에 도착하지 않고 한꺼번에 온다

Nginx의 proxy_buffering off가 빠진 경우가 대부분이다. 또는 Content-Type: text/event-stream 헤더가 없으면 브라우저가 SSE로 인식하지 않는다.

Q: asyncio.CancelledError가 간헐적으로 발생한다

클라이언트가 스트리밍 도중 연결을 끊으면 FastAPI가 제너레이터를 취소한다. stream_claude 안에서 except asyncio.CancelledError: return을 처리해주면 깔끔하게 종료된다.

Q: RuntimeError: Event loop is closed 에러

동기 anthropic.Anthropic() 클라이언트를 비동기 컨텍스트에서 쓰면 발생할 수 있다. anthropic.AsyncAnthropic()으로 교체하는 게 근본 해결책이다.

Q: 레이트 리밋에 걸렸는데 재시도해도 계속 실패한다

BASE_DELAY가 너무 짧거나, 짧은 시간에 요청이 몰리는 버스트 트래픽이 원인이다. Anthropic의 공식 Rate Limits 페이지에서 플랜별 TPM/RPM 한도를 확인하고 BASE_DELAY를 최소 5초 이상으로 올리는 걸 권장한다.

프로덕션 체크리스트

배포 전에 확인해야 할 항목들을 정리했다:

보안:

• ANTHROPIC_API_KEY는 절대 코드에 하드코딩하지 않는다. 환경 변수 또는 AWS Secrets Manager, GCP Secret Manager 같은 시크릿 관리 서비스를 쓴다.
• /chat/stream 엔드포인트에 인증 미들웨어가 있는지 확인한다. API 키나 JWT 토큰 없이 퍼블릭으로 열면 외부에서 무제한으로 호출할 수 있다.
• CORS 설정이 필요 이상으로 열려 있지 않은지 점검한다.

성능:

• AsyncAnthropic 클라이언트를 앱 시작 시 한 번만 생성하고 재사용한다. 요청마다 새 클라이언트를 생성하면 커넥션 풀이 낭비된다.
• uvicorn --workers는 CPU 코어 수의 2배가 일반적인 시작점이다. 스트리밍 응답은 I/O 바운드이므로 멀티프로세스보다 asyncio 이벤트 루프가 더 효율적이다.
• GZIP 압축은 SSE와 궁합이 맞지 않는다. 청크 단위로 스트리밍되는 텍스트는 GZIP 압축의 효과가 거의 없고, 오히려 첫 청크 전달을 지연시킬 수 있다.

관찰 가능성:

• done 이벤트에서 토큰 사용량을 추출해 로그로 남긴다. 예상보다 입력 토큰이 많이 쓰이는 요청 패턴을 조기에 발견할 수 있다.
• 에러 타입별 카운터를 Prometheus나 CloudWatch에 올린다. rate_limit 에러가 급증하면 플랜 업그레이드 시점을 알 수 있다.

이 체크리스트를 보면서 “다 알고 있는 내용인데 실제 배포 때 놓치는 경우가 많다”고 느꼈다면, 그게 정상이다. 나도 처음 배포할 때 CORS와 GZIP 설정에서 한 번씩 실수했다.

언제 쓰고, 언제 피해야 하는가

스트리밍 백엔드를 무조건 SSE + FastAPI로 가는 게 정답은 아니다. 직접 써본 기준으로 선택 기준을 정리한다.

이 스택이 제값을 하는 경우:

• Python 팀이 이미 있고 새 언어 스택 도입 비용을 피하고 싶을 때
• 스트리밍이 핵심 UX 요소인 AI 채팅, 코드 생성, 문서 작성 서비스
• OpenAPI 문서 자동화와 Pydantic 검증이 필요한 팀
• 기존 FastAPI 또는 Django REST 백엔드에 AI 기능을 점진적으로 추가하는 상황

피하는 게 나은 경우:

• 응답을 한 번에 받아도 UX에 지장이 없는 짧은 분류·추출 작업. 이때는 그냥 단일 요청-응답이 코드도 단순하고 디버깅도 쉽다.
• 1,000건 이상의 문서를 일괄 처리하는 배치 작업. 스트리밍은 의미가 없고 Anthropic Message Batches API가 비용 면에서 절반 수준이다.
• 양방향 실시간 상호작용(타이핑 인디케이터, 동시 편집)이 필요한 경우. SSE는 단방향이므로 WebSocket이 맞다.
• 로컬·온프레미스 환경에서 외부 API 호출 자체가 막혀 있을 때. 이 경우 자체 호스팅 모델이 먼저다. 자체 호스팅 옵션은 Ollama와 FastAPI로 프로덕션 배포하기에서 다뤘다.

요컨대 “긴 출력 + 실시간 표시”라는 두 조건이 동시에 성립할 때만 이 패턴의 복잡도가 정당화된다. 한쪽이라도 빠지면 더 단순한 방법이 있다.

1차 출처와 참고 자료

이 글의 코드는 다음 공식 문서를 기준으로 작성·검증했다. 버전이 올라가면 동작이 바뀔 수 있으니 구현 전에 한 번씩 확인하는 걸 권장한다.

• FastAPI 공식 문서 — Custom Response / StreamingResponse: StreamingResponse의 제너레이터 동작과 취소 처리에 대한 공식 설명.
• Anthropic — Streaming Messages: Claude API의 SSE 스트리밍 이벤트 구조와 SDK별 사용법.
• MDN — Using server-sent events: SSE 이벤트 포맷(data:, event:, id:, retry:)과 EventSource API의 표준 정의.

타입 안전한 요청 스키마를 더 엄격하게 가져가고 싶다면 Pydantic AI로 타입 세이프 에이전트 만들기도 같이 보면 도움이 된다.

솔직히 말하면, 이 스택이 모든 상황에서 최선은 아니다. Node.js 팀이라면 Vercel AI SDK가 더 빠르고, 대규모 실시간 연결이 필요하다면 WebSocket이나 gRPC Streaming이 더 나은 선택일 수 있다. 하지만 Python AI 백엔드를 빠르게 올리고 싶다면, 이 패턴은 내가 실제로 검증한 가장 실용적인 출발점이다.

이 글에서 구현한 모든 코드는 FastAPI 0.136과 Anthropic SDK 0.97 기준이다. Anthropic SDK는 버전 업데이트가 잦으므로, AsyncAnthropic의 stream() 컨텍스트 매니저 API가 변경됐다면 공식 문서를 먼저 확인하는 게 좋다. Anthropic SDK vs OpenAI SDK 개발자 경험 비교를 읽으면 SDK의 변경 이력과 설계 철학을 이해하는 데 도움이 된다.

다음 단계로는 프롬프트 캐싱 적용으로 비용을 줄이고, 스트리밍 응답에 OpenTelemetry 트레이싱을 붙여 레이턴시와 토큰 사용량을 가시화하는 작업을 추천한다.