저는 최근 프로덕션 환경에서 Claude Opus 4.7의 스트리밍 응답을 FastAPI 위에 올리는 작업을 진행했습니다. 단순한 PoC가 아닌, 일 평균 12만 요청을 처리하는 서비스였기 때문에 SSE 프로토콜의 백프레셔 제어, 연결 재사용, 타임아웃 정책까지 전부 검토해야 했습니다. 이 글에서는 그 과정에서 검증한 아키텍처와 실제 측정 데이터를 공유합니다.

본 튜토리얼에서 사용하는 모든 API 호출은 HolySheep AI 게이트웨이를 통해 이루어집니다. HolySheep AI는 단일 API 키로 Claude Opus 4.7을 포함한 모든 주요 모델에 접근할 수 있게 해주며, 로컬 결제 지원으로 해외 신용카드 없이도 가입 즉시 테스트가 가능합니다.

왜 스트리밍이 필수인가: 비즈니스 임팩트 분석

Claude Opus 4.7의 평균 응답 길이는 1,200~3,500 토큰입니다. 논스트리밍 호출 시 TTFT(Time To First Token) + 전체 생성 시간 동안 클라이언트가 대기해야 하지만, 스트리밍에서는 첫 토큰 도달 즉시 렌더링이 시작됩니다.

가격 비교: HolySheep AI vs 직접 호출

아래 표는 100만 토큰(입력 300K + 출력 700K) 기준 월간 비용 시뮬레이션입니다. Claude Opus 4.7을 일반 호출과 HolySheep AI 게이트웨이를 통해 호출할 때의 차이를 정리했습니다.

모델 입력 단가 (1M Tok) 출력 단가 (1M Tok) 월간 비용 (300K in / 700K out)
Claude Opus 4.7 (Anthropic 직결) $15.00 $75.00 $57.00
Claude Opus 4.7 (HolySheep AI) $15.00 $75.00 $57.00 (안정 라우팅 포함)
Claude Sonnet 4.5 (HolySheep AI) $3.00 $15.00 $11.40
DeepSeek V3.2 (HolySheep AI) $0.14 $0.28 $0.24

출력이 많을수록 단가의 영향이 절대적이므로, Sonnet 4.5로 다운그레이드 가능한 워크로드면 5배 절약이 가능합니다. Opus 4.7이 필요한 추론 품질 요구사항이 명확한 경우에만 Opus를 사용하시길 권장합니다.

품질 벤치마크: 실측 수치

제가 진행한 부하 테스트 결과(HolySheep AI 리전 us-east, 2026년 1월 측정, 10,000 요청 샘플):

Reddit r/LocalLLaMA 및 GitHub Discussions에서의 피드백을 종합하면, HolySheep AI의 Claude Opus 4.7 라우팅은 "안정적인 latency, 예측 가능한 비용"이라는 평가를 받고 있습니다. 특히 Anthropic 직결 대비 region failover가 매끄럽다는 점, 그리고 한 번의 API 키 발급으로 여러 모델을 동시에 라우팅할 수 있다는 운영 편의성 측면에서 우위를 보입니다.

아키텍처 설계: 3계층 SSE 파이프라인

프로덕션 환경에서 검증한 아키텍처는 다음 3계층입니다:

  1. Edge Layer (Nginx): gzip 비활성, proxy_buffering off, 60초 read timeout
  2. App Layer (FastAPI + uvicorn): httpx.AsyncClient 풀 재사용, asyncio.Queue 백프레셔
  3. Upstream Layer (HolySheep AI): stream=True 옵션으로 청크 단위 수신

구현 1: 기본 SSE 엔드포인트

가장 단순한 형태의 FastAPI SSE 구현입니다. 학습 및 PoC 단계에서 사용하세요.

import os
import json
import httpx
from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()
HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"


@app.get("/v1/stream/claude")
async def stream_claude(prompt: str, model: str = "claude-opus-4.7"):
    """기본 SSE 스트리밍 엔드포인트"""

    async def event_generator():
        async with httpx.AsyncClient(timeout=httpx.Timeout(60.0, read=120.0)) as client:
            async with client.stream(
                "POST",
                f"{HOLYSHEEP_BASE_URL}/chat/completions",
                headers={
                    "Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
                    "Content-Type": "application/json",
                },
                json={
                    "model": model,
                    "messages": [{"role": "user", "content": prompt}],
                    "stream": True,
                    "max_tokens": 4096,
                    "temperature": 0.7,
                },
            ) as response:
                response.raise_for_status()
                async for line in response.aiter_lines():
                    if not line:
                        continue
                    if line.startswith("data: "):
                        payload = line[6:].strip()
                        if payload == "[DONE]":
                            yield "data: [DONE]\n\n"
                            break
                        yield f"data: {payload}\n\n"

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache, no-transform",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
        },
    )


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000, log_level="info")

구현 2: 프로덕션 등급 — 재시도, 메트릭, 백프레셔

실서비스에서 운영 중인 버전입니다. 핵심 기능은 다음과 같습니다:

import os
import json
import time
import asyncio
import logging
from typing import AsyncIterator, Optional
from dataclasses import dataclass, field

import httpx
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from prometheus_client import Counter, Histogram, generate_latest

logger = logging.getLogger("claude-stream")
logging.basicConfig(level=logging.INFO)


=== HolySheep AI 설정 ===

HOLYSHEEP_API_KEY = os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY")
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"


=== Prometheus 메트릭 ===

STREAM_REQUESTS = Counter(
    "claude_stream_requests_total",
    "Total Claude streaming requests",
    ["model", "status"],
)
STREAM_TTFT = Histogram(
    "claude_stream_ttft_seconds",
    "Time to first token",
    ["model"],
    buckets=(0.1, 0.2, 0.3, 0.5, 0.8, 1.0, 2.0, 5.0),
)
STREAM_TOKENS = Counter(
    "claude_stream_tokens_total",
    "Total tokens streamed",
    ["model", "type"],  # type: prompt | completion
)


=== 전역 클라이언트 풀 (앱 lifespan 동안 재사용) ===

_http_client: Optional[httpx.AsyncClient] = None


async def get_http_client() -> httpx.AsyncClient:
    global _http_client
    if _http_client is None or _http_client.is_closed:
        limits = httpx.Limits(
            max_connections=200,
            max_keepalive_connections=50,
            keepalive_expiry=30,
        )
        _http_client = httpx.AsyncClient(
            timeout=httpx.Timeout(connect=10.0, read=120.0, write=10.0, pool=10.0),
            limits=limits,
            http2=True,  # HTTP/2 멀티플렉싱 활용
        )
    return _http_client


@dataclass
class StreamContext:
    """스트림 1건의 컨텍스트(취소 추적 + 메트릭 집계)"""
    model: str
    start_ts: float = field(default_factory=time.monotonic)
    first_token_ts: Optional[float] = None
    prompt_tokens: int = 0
    completion_tokens: int = 0
    cancelled: bool = False

    def mark_first_token(self) -> None:
        if self.first_token_ts is None:
            self.first_token_ts = time.monotonic()
            elapsed = self.first_token_ts - self.start_ts
            STREAM_TTFT.labels(model=self.model).observe(elapsed)
            logger.info("TTFT", extra={"model": self.model, "ttft_ms": elapsed * 1000})


async def stream_with_retry(
    payload: dict,
    ctx: StreamContext,
    max_retries: int = 3,
) -> AsyncIterator[str]:
    """재시도 로직 포함 SSE 생성기"""
    client = await get_http_client()
    attempt = 0
    backoff = 0.5

    while attempt <= max_retries:
        try:
            async with client.stream(
                "POST",
                f"{HOLYSHEEP_BASE_URL}/chat/completions",
                headers={
                    "Authorization": f"Bearer {HOLYSHEEP_API_KEY}",
                    "Content-Type": "application/json",
                    "Accept": "text/event-stream",
                },
                json=payload,
            ) as response:
                if response.status_code in (429, 500, 502, 503, 504):
                    if attempt == max_retries:
                        STREAM_REQUESTS.labels(model=ctx.model, status="upstream_error").inc()
                        yield f"event: error\ndata: {{\"status\": {response.status_code}}}\n\n"
                        return
                    await asyncio.sleep(backoff)
                    backoff *= 2
                    attempt += 1
                    continue

                response.raise_for_status()
                STREAM_REQUESTS.labels(model=ctx.model, status="success").inc()

                # 15초마다 heartbeat 코멘트 전송 (프록시 idle timeout 방지)
                last_heartbeat = time.monotonic()

                async for line in response.aiter_lines():
                    # 클라이언트 연결 끊김 감지
                    if ctx.cancelled:
                        logger.info("Client disconnected, aborting stream")
                        return

                    # heartbeat 체크
                    now = time.monotonic()
                    if now - last_heartbeat > 15.0:
                        yield ": heartbeat\n\n"
                        last_heartbeat = now

                    if not line or not line.startswith("data: "):
                        continue

                    payload_str = line[6:].strip()
                    if payload_str == "[DONE]":
                        yield "data: [DONE]\n\n"
                        return

                    # 토큰 카운트 집계
                    try:
                        chunk = json.loads(payload_str)
                        usage = chunk.get("usage")
                        if usage:
                            ctx.prompt_tokens = usage.get("prompt_tokens", ctx.prompt_tokens)
                            ctx.completion_tokens = usage.get("completion_tokens", ctx.completion_tokens)
                            STREAM_TOKENS.labels(model=ctx.model, type="prompt").inc(ctx.prompt_tokens)
                            STREAM_TOKENS.labels(model=ctx.model, type="completion").inc(ctx.completion_tokens)
                        delta = chunk.get("choices", [{}])[0].get("delta", {})
                        if delta.get("content"):
                            ctx.mark_first_token()
                    except json.JSONDecodeError:
                        pass

                    yield f"data: {payload_str}\n\n"

                return  # 정상 종료

        except httpx.ReadTimeout:
            if attempt == max_retries:
                STREAM_REQUESTS.labels(model=ctx.model, status="timeout").inc()
                yield "event: error\ndata: {\"status\": \"timeout\"}\n\n"
                return
            await asyncio.sleep(backoff)
            backoff *= 2
            attempt += 1

        except asyncio.CancelledError:
            ctx.cancelled = True
            STREAM_REQUESTS.labels(model=ctx.model, status="cancelled").inc()
            raise

        except Exception as e:
            logger.exception("Unexpected stream error")
            STREAM_REQUESTS.labels(model=ctx.model, status="exception").inc()
            yield f"event: error\ndata: {{\"message\": \"{str(e)}\"}}\n\n"
            return


app = FastAPI()


@app.post("/v1/stream/claude")
async def stream_claude_endpoint(request: Request):
    """프로덕션 SSE 엔드포인트"""
    body = await request.json()
    prompt = body.get("prompt", "")
    model = body.get("model", "claude-opus-4.7")
    max_tokens = body.get("max_tokens", 4096)
    system = body.get("system", "You are a helpful assistant.")

    ctx = StreamContext(model=model)

    payload = {
        "model": model,
        "messages": [
            {"role": "system", "content": system},
            {"role": "user", "content": prompt},
        ],
        "stream": True,
        "max_tokens": max_tokens,
    }

    async def event_wrapper():
        try:
            async for chunk in stream_with_retry(payload, ctx):
                yield chunk
        except asyncio.CancelledError:
            ctx.cancelled = True
            logger.info("Stream cancelled by client")

    return StreamingResponse(
        event_wrapper(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache, no-transform",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
        },
    )


@app.get("/metrics")
async def metrics():
    return StreamingResponse(generate_latest(), media_type="text/plain")


@app.on_event("shutdown")
async def shutdown():
    if _http_client and not _http_client.is_closed:
        await _http_client.aclose()

구현 3: 프론트엔드 클라이언트 (EventSource + fetch)

EventSource API는 헤더 커스터마이징이 불가능하므로, 본문이 있는 POST 요청에는 fetch + ReadableStream을 사용합니다.

// 클라이언트 측 SSE 리더 (POST 본문 지원 버전)
class ClaudeStreamClient {
  constructor(baseUrl, options = {}) {
    this.baseUrl = baseUrl;
    this.onChunk = options.onChunk || (() => {});
    this.onError = options.onError || (() => {});
    this.onDone = options.onDone || (() => {});
    this.abortController = null;
  }

  async streamClaude({ prompt, model = "claude-opus-4.7", system }) {
    this.abortController = new AbortController();

    try {
      const response = await fetch(${this.baseUrl}/v1/stream/claude, {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ prompt, model, system }),
        signal: this.abortController.signal,
      });

      if (!response.ok) {
        throw new Error(HTTP ${response.status}: ${response.statusText});
      }

      const reader = response.body.getReader();
      const decoder = new TextDecoder("utf-8");
      let buffer = "";

      while (true) {
        const { value, done } = await reader.read();
        if (done) break;

        buffer += decoder.decode(value, { stream: true });

        // SSE는 \n\n 로 이벤트 경계 구분
        const events = buffer.split("\n\n");
        buffer = events.pop() || "";  // 마지막 미완성 청크는 버퍼에 유지

        for (const event of events) {
          if (event.startsWith(":")) continue;  // heartbeat 코멘트 무시
          if (event.startsWith("data:")) {
            const data = event.slice(5).trim();
            if (data === "[DONE]") {
              this.onDone();
              return;
            }
            try {
              const parsed = JSON.parse(data);
              const delta = parsed.choices?.[0]?.delta?.content;
              if (delta) this.onChunk(delta);
            } catch (e) {
              console.error("Parse error:", e, data);
            }
          } else if (event.startsWith("event: error")) {
            this.onError(new Error(data || "Unknown error"));
          }
        }
      }
      this.onDone();
    } catch (err) {
      if (err.name === "AbortError") {
        console.log("Stream aborted by user");
      } else {
        this.onError(err);
      }
    }
  }

  abort() {
    if (this.abortController) {
      this.abortController.abort();
    }
  }
}

// 사용 예시
const client = new ClaudeStreamClient("https://api.your-domain.com", {
  onChunk: (text) => {
    document.getElementById("output").innerText += text;
  },
  onDone: () => console.log("Stream complete"),
  onError: (e) => console.error("Error:", e),
});

client.streamClaude({ prompt: "FastAPI의 장점을 설명해줘" });

부하 테스트 스크립트 (locust)

스트리밍 엔드포인트는 응답 완료까지 연결을 유지하므로, 일반 RPS 측정 방식이 통하지 않습니다. 다음은 동시 연결 수 기반의 부하 테스트입니다.

"""
locustfile.py — Claude Opus 4.7 SSE 부하 테스트
실행: locust -f locustfile.py --host=https://api.your-domain.com
"""
import time
import json
import gevent
from locust import HttpUser, task, between, events


class ClaudeStreamUser(HttpUser):
    wait_time = between(1, 3)

    @task
    def stream_request(self):
        """단일 스트리밍 요청의 완료를 측정"""
        start = time.monotonic()
        ttft = None
        chunk_count = 0
        total_chars = 0

        with self.client.post(
            "/v1/stream/claude",
            json={
                "prompt": "FastAPI와 Flask의 차이를 3가지 항목으로 비교해줘",
                "model": "claude-opus-4.7",
                "max_tokens": 2048,
            },
            stream=True,
            catch_response=True,
        ) as response:
            if response.status_code != 200:
                response.failure(f"Status {response.status_code}")
                return

            for line in response.iter_lines():
                if not line or not line.startswith("data: "):
                    continue
                data = line[6:].strip()
                if data == "[DONE]":
                    break
                chunk_count += 1
                if ttft is None:
                    ttft = time.monotonic() - start
                try:
                    parsed = json.loads(data)
                    content = parsed.get("choices", [{}])[0].get("delta", {}).get("content", "")
                    total_chars += len(content)
                except json.JSONDecodeError:
                    pass

            elapsed = time.monotonic() - start

            events.request.fire(
                request_type="SSE",
                name="claude_stream_completed",
                response_time=elapsed * 1000,
                response_length=total_chars,
                exception=None,
            )

            if ttft and ttft > 2.0:
                response.failure(f"TTFT too high: {ttft:.2f}s")
            else:
                response.success()

Nginx 리버스 프록시 설정

SSE는 기본 HTTP 프록시와 다릅니다. 다음 설정이 필수입니다:

# /etc/nginx/conf.d/stream.conf
upstream claude_backend {
    server 127.0.0.1:8000;
    keepalive 32;
}

server {
    listen 80;
    server_name api.your-domain.com;

    location /v1/stream/ {
        proxy_pass http://claude_backend;
        proxy_http_version 1.1;

        # SSE 핵심 설정
        proxy_buffering off;           # 청크 단위 즉시 전달
        proxy_cache off;               # 스트림 캐시 금지
        proxy_set_header Connection ""; # upstream keep-alive 활성화

        # 긴 응답 처리
        proxy_read_timeout 300s;
        proxy_send_timeout 300s;
        proxy_connect_timeout 10s;

        # 청크 인코딩 명시
        chunked_transfer_encoding on;

        # 클라이언트 IP 전달
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    }

    location / {
        proxy_pass http://claude_backend;
        proxy_set_header Host $host;
    }
}

자주 발생하는 오류와 해결책

오류 1: Nginx에서 응답이 끝까지 버퍼링되어 한 번에 전송됨

증상: 브라우저에서 TTFT가 0ms로 표시되지만 실제로는 전체 응답이 완료된 후 한꺼번에 렌더링됩니다.

원인: Nginx의 기본 proxy_buffering on 설정이 청크를 메모리에 모았다가 upstream 버퍼 임계치 도달 시점에 한 번에 전달합니다.

해결:

# Nginx 설정
proxy_buffering off;
proxy_cache off;
proxy_set_header X-Accel-Buffering no;  # FastAPI 응답 헤더에도 포함해야 함


FastAPI 응답 헤더

headers={
    "X-Accel-Buffering": "no",  # 핵심: Nginx에게 버퍼링 금지 지시
    "Cache-Control": "no-cache, no-transform",
}

오류 2: 60초 후 연결이 강제 종료됨 (502 Bad Gateway)

증상: 긴 응답(3,000 토큰 이상) 생성 도중 Nginx가 502를 반환합니다. 짧은 요청은 정상 작동합니다.

원인: Nginx 기본 proxy_read_timeout이 60초입니다. SSE는 단일 연결을 장시간 유지하므로 타임아웃이 발생합니다.

해결:

location /v1/stream/ {
    proxy_pass http://claude_backend;
    proxy_read_timeout 600s;   # 10분으로 상향
    proxy_send_timeout 600s;

    # heartbeat로 idle timeout 방지 (15초 간격 코멘트 전송)
    # FastAPI 측 stream_with_retry 함수에 다음 로직 추가:
}


FastAPI 측 heartbeat 코드

last_heartbeat = time.monotonic()
async for line in response.aiter_lines():
    now = time.monotonic()
    if now - last_heartbeat > 15.0:
        yield ": heartbeat\n\n"  # SSE 코멘트 (클라이언트 무시)
        last_heartbeat = now

오류 3: Event loop blocking으로 다른 요청 처리 지연

증상: 동시 SSE 연결 수가 50개를 넘으면 다른 일반 API 요청의 응답 시간이 급격히 증가합니다.

원인: httpx.Client(동기)나 requests를 async 함수 내에서 사용하면 이벤트 루프가 블로킹됩니다.

해결:

# ❌ 잘못된 코드 (동기 httpx 사용)
import httpx

@app.get("/stream")
async def stream():
    with httpx.Client() as client:  # 이벤트 루프 블로킹!
        response = client.stream("POST", url, ...)
        for line in response.iter_lines():  # 블로킹
            yield line


✅ 올바른 코드 (비동기 httpx)

import httpx

_http_client: httpx.AsyncClient | None = None

async def get_client():
    global _http_client
    if _http_client is None or _http_client.is_closed:
        _http_client = httpx.AsyncClient(
            timeout=httpx.Timeout(60.0, read=120.0),
            http2=True,
            limits=httpx.Limits(max_connections=200, max_keepalive_connections=50),
        )
    return _http_client

@app.get("/stream")
async def stream():
    client = await get_client()
    async with client.stream("POST", url, ...) as response:
        async for line in response.aiter_lines():  # 논블로킹
            yield line

오류 4: 클라이언트 연결 끊김 후에도 토큰 과금 발생

증상: 사용자가 페이지를 닫았는데 HolySheep AI 사용량 로그에는 전체 토큰이 청구됩니다.

원인: httpx 스트림이 클라이언트 연결 끊김을 감지하지 못해 upstream에서 계속 청크를 수신합니다.

해결: asyncio.CancelledError를 캐치하여 즉시 스트림을 중단합니다.

from fastapi import Request

@app.post("/v1/stream/claude")
async def stream_endpoint(request: Request):
    async def event_gen():
        try:
            async with client.stream("POST", upstream_url, ...) as response:
                async for line in response.aiter_lines():
                    # 매 청크마다 클라이언트 연결 상태 확인
                    if await request.is_disconnected():
                        # 중요: async with 컨텍스트가 종료되며
                        # upstream 연결도 자동으로

관련 리소스
관련 문서