Zettelkasten

Python에서 SSE는 ASGI 환경에서만 운영 가능하다

·수정 2026.05.06·수정 1회
sseasgiwsgipythonfastapidjangostreaminghttp

요약

  • SSE 는 응답을 영원히 안 닫는 HTTP 스트림이라, 한 worker 가 한 요청에 sync 로 매여있는 WSGI 모델에선 connection 수만큼 worker 가 블록되어 사실상 운영 불가하다.
  • ASGI 는 single event loop 위에서 코루틴으로 다중화하므로 한 프로세스가 수천 SSE 연결을 동시에 들고 있을 수 있다 (Node.js event loop 와 동일 구조).
  • FastAPI 의 StreamingResponse + async generator 가 가장 깔끔하고, Django 도 4.2+ async view + uvicorn/daphne 로 동일하게 구현 가능하다.

본문

SSE 의 본질이 worker 모델을 정한다

SSE 는 응답을 안 닫고 서버가 텍스트 청크를 흘려보내는 단방향 HTTP 스트림 이다. Content-Type: text/event-stream 위에 event: ...\ndata: ...\n\n 텍스트 포맷이 전부고, TCP connection 한 개가 stream 이 살아있는 동안 계속 열려 있다.

이 특성이 동시성 모델 선택을 강제한다:

  • 한 connection 이 수십 초~수 분 점유됨
  • 점유 중 대부분의 시간은 대기 (SDK 호출, DB 조회, sleep)
  • 동시 connection 수가 곧 동시 처리량

WSGI 가 왜 사실상 불가능한가

WSGI (gunicorn sync workers + Flask/Django) 는 한 요청 = 한 worker thread/process 가 처리 끝까지 점유한다. SSE 는 응답이 안 끝나니까 worker 가 영구히 블록.

gunicorn --workers 4 --worker-class sync app:wsgi
↓
SSE connection 4개 = 모든 worker 점유 = 5번째 일반 요청도 못 받음

워커 수를 늘려도 RAM/CPU 가 비례해서 늘 뿐이고, 100명 동시 접속 = 100 worker 가 필요해지는 비용 구조. 게다가 sync IO 라 SDK 응답 기다리는 동안 worker 는 그냥 놀고 있음. 사실상 운영 불가.

gevent/eventlet 같은 그린스레드 worker 로 우회 가능하지만 monkey-patching 의 부작용 + 라이브러리 호환성 문제가 커서 권장되지 않음.

ASGI 가 답인 이유

ASGI 는 single event loop 위에서 코루틴이 await 지점에서 yield 하면 다른 코루틴이 실행 된다. SSE 가 SDK 응답을 기다리는 동안 다른 connection 이 처리됨. 한 프로세스가 수천 SSE connection 을 동시 처리 가능 — Node.js event loop 와 동일한 모델.

병목은 worker 수가 아니라:

  • 프로세스당 메모리
  • file descriptor 한도 (ulimit -n)

uvicorn 의 --workers 4 는 프로세스 4개를 띄우는 것 이지 한 프로세스 안에 worker 가 4개가 아니다. 각 프로세스는 단일 event loop.

FastAPI / Starlette — 가장 깔끔한 구현

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio, json

app = FastAPI()

async def event_stream(request: Request, prompt: str):
    yield f"event: init\ndata: {json.dumps({'sessionId': 'abc'})}\n\n"

    async for chunk in call_claude_sdk(prompt):
        if await request.is_disconnected():
            await abort_sdk()
            return
        payload = json.dumps({"text": chunk.text})
        yield f"event: partial\ndata: {payload}\n\n"

    yield "event: result\ndata: {}\n\n"

@app.post("/api/query")
async def query(request: Request, body: dict):
    return StreamingResponse(
        event_stream(request, body["prompt"]),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache, no-transform",
            "X-Accel-Buffering": "no",  # nginx 버퍼링 차단
        },
    )

StreamingResponse 가 async generator 를 받아서 청크가 yield 될 때마다 ASGI http.response.body 메시지로 흘려보낸다.

Django 4.2+ async view

async def view + StreamingHttpResponse + uvicorn/daphne 로 ASGI 띄우기.

from django.http import StreamingHttpResponse
import json

async def query_view(request):
    async def gen():
        yield "event: init\ndata: {}\n\n"
        async for chunk in call_claude_sdk(...):
            yield f"event: partial\ndata: {json.dumps({'text': chunk.text})}\n\n"
    return StreamingHttpResponse(gen(), content_type="text/event-stream")

함정:

  • gunicorn sync worker 그대로 두면 의미 없음 — uvicorn/daphne 로 마이그레이션 필요
  • Django ORM 은 기본 sync 라 sync_to_async() 또는 5.0+ 의 async ORM (aget, acreate, afilter) 사용. sync ORM 을 async view 안에서 그대로 호출하면 SynchronousOnlyOperation 예외
  • DRF 는 async view 지원이 늦어서 raw Django view 로 SSE 만 따로 빼는 경우가 많음
  • 미들웨어 도 async-aware 한지 확인 필요. sync 미들웨어는 자동 감싸지지만 성능 저하

Channels 는 SSE 만을 위해선 오버킬

Django Channels 는 WebSocket 위주로 설계됐지만 HTTP+ASGI 도 지원. SSE 만 필요하면 도입 비용이 커서 권장되지 않고, 이미 Channels 쓰는 프로젝트면 자연스러운 선택.

Node.js 와 1:1 매핑되는 패턴

Node (Fastify) Python (ASGI)
reply.raw.write(...) yield "event: ...\n\n"
request.raw.on("close") await request.is_disconnected() 또는 asyncio.CancelledError catch
setInterval(keepalive, 15000) 별도 task 로 주기 yield 또는 asyncio.wait_for
AbortController.abort() task.cancel() → CancelledError
동시성 카운터 asyncio.Semaphore(MAX_CONCURRENT)
for await (msg of conversation) async for msg in conversation

asyncio.CancelledError 가 Node 의 AbortError 와 의미적으로 동일. SDK 호출부를 try/finally 로 감싸서 cleanup 보장.

Disconnect 감지의 두 패턴

폴링 방식 — 매 청크마다 is_disconnected() 호출:

async for chunk in sdk():
    if await request.is_disconnected():
        return
    yield ...

간단하지만 청크 간격이 길면 끊김 감지가 늦음.

예외 기반 — yield 중 클라이언트 끊김이 감지되면 asyncio.CancelledError 가 던져짐:

try:
    async for chunk in sdk():
        yield ...
except asyncio.CancelledError:
    await abort_sdk()
    raise

이쪽이 더 정확하고 즉각적. 보통 둘을 섞어 씀.

운영시 체크리스트

  • nginx 등 프록시: proxy_buffering off; + X-Accel-Buffering: no 헤더 — 안 그러면 청크가 모였다가 한 번에 도착해서 스트리밍이 무력화됨
  • 15초 keepalive 코멘트 (: keepalive\n\n) — idle proxy 가 끊는 거 방지
  • HTTP/1.1 의 브라우저당 도메인 6 connection 제한 → SSE 6개 띄우면 7번째 stall. HTTP/2 로 해결됨 (multiplex)

검증 질문

  • "ASGI 안 쓰고 SSE 가능한가?" → 기술적으론 WSGI 도 가능하지만 connection 수만큼 worker 점유라 운영 불가
  • "왜 FastAPI 가 자주 추천되나?" → 처음부터 ASGI 기반 + StreamingResponse 가 async generator 를 그대로 받음
  • "Django 로도 정말 되나?" → 4.2+ async view + ASGI server (uvicorn/daphne) 면 가능. 단 gunicorn sync 그대로 두면 의미 없음

참고

이 문서를 참조하는 노트 (2)

함께 읽기 좋은 글