Server-Sent Events(SSE)로 실시간 데이터 보내기

브라우저가 기본으로 지원하는 EventSource API로 실시간으로 데이터를 받아보게 하려면, Server-Sent Events(SSE) 를 쓰면 돼요. 앞서 다룬 Stream JSON Lines와 비슷하지만 text/event-stream 형식을 사용하는데요, AI 채팅 스트리밍, 실시간 알림, 로그·옵저버빌리티 같은 곳에서 자주 쓰여요.

출처: FastAPI 공식 문서 - Server-Sent Events (SSE)

참고 — FastAPI 0.135.0에 추가된 기능이에요.

Server-Sent Events가 뭔가요

SSE는 HTTP를 통해 서버에서 클라이언트로 데이터를 스트리밍하는 표준이에요. 각 이벤트는 data, event, id, retry 같은 필드로 이뤄진 작은 텍스트 블록이고, 빈 줄로 구분돼요. 실제로는 이런 모양이에요.

data: {"name": "Portal Gun", "price": 999.99}

data: {"name": "Plumbus", "price": 32.99}

서버가 업데이트를 클라이언트로 밀어주는 경우, 즉 AI 채팅 스트리밍, 실시간 알림, 로그·옵저버빌리티 같은 데 주로 쓰여요. 바이너리 데이터(동영상·오디오)를 스트리밍하려면 Stream Data 고급 가이드를 확인해 보세요.

FastAPI에서 SSE 스트리밍하기

FastAPI에서 SSE를 스트리밍하려면 path operation 함수에서 yield를 쓰고 response_class=EventSourceResponse를 지정해요. EventSourceResponse는 fastapi.sse에서 가져와요.

from collections.abc import AsyncIterable, Iterable

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None


items = [
    Item(name="Plumbus", description="A multi-purpose household device."),
    Item(name="Portal Gun", description="A portal opening device."),
    Item(name="Meeseeks Box", description="A box that summons a Meeseeks."),
]


@app.get("/items/stream", response_class=EventSourceResponse)
async def sse_items() -> AsyncIterable[Item]:
    for item in items:
        yield item


@app.get("/items/stream-no-async", response_class=EventSourceResponse)
def sse_items_no_async() -> Iterable[Item]:
    for item in items:
        yield item


@app.get("/items/stream-no-annotation", response_class=EventSourceResponse)
async def sse_items_no_annotation():
    for item in items:
        yield item


@app.get("/items/stream-no-async-no-annotation", response_class=EventSourceResponse)
def sse_items_no_async_no_annotation():
    for item in items:
        yield item

yield되는 각 항목은 JSON으로 인코딩되어 SSE 이벤트의 data: 필드로 보내져요. 반환 타입을 AsyncIterable[Item]으로 선언하면 FastAPI가 Pydantic으로 데이터를 검증·문서화·직렬화해요. Pydantic이 직렬화를 Rust 쪽에서 처리하므로 반환 타입을 선언할 때 훨씬 높은 성능을 얻을 수 있어요.

비동기가 아닌 path operation 함수

async가 없는 일반 def 함수에서도 똑같이 yield를 쓸 수 있어요. FastAPI가 이벤트 루프를 막지 않도록 올바르게 실행해 줘요. 이 경우 올바른 반환 타입은 Iterable[Item]이에요. 반환 타입을 생략하면 FastAPI가 jsonable_encoder로 데이터를 변환해서 보내요.

`ServerSentEvent`로 필드 제어하기

event, id, retry, comment 같은 SSE 필드를 직접 설정해야 한다면, 일반 데이터 대신 ServerSentEvent 객체를 yield하면 돼요. ServerSentEvent도 fastapi.sse에서 가져와요.

from collections.abc import AsyncIterable

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse, ServerSentEvent
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    price: float


items = [
    Item(name="Plumbus", price=32.99),
    Item(name="Portal Gun", price=999.99),
    Item(name="Meeseeks Box", price=49.99),
]


@app.get("/items/stream", response_class=EventSourceResponse)
async def stream_items() -> AsyncIterable[ServerSentEvent]:
    yield ServerSentEvent(comment="stream of item updates")
    for i, item in enumerate(items):
        yield ServerSentEvent(data=item, event="item_update", id=str(i + 1), retry=5000)

data 필드는 항상 JSON으로 인코딩돼요. Pydantic 모델을 포함해 JSON으로 직렬화할 수 있는 어떤 값이든 넘길 수 있어요.

원시 데이터 보내기

JSON 인코딩 없이 데이터를 보내야 한다면 data 대신 raw_data를 써요. 미리 포맷된 텍스트, 로그 줄, 또는 [DONE] 같은 특별한 상태를 뜻하는 "센티널(sentinel)" 값을 보낼 때 유용해요.

from collections.abc import AsyncIterable

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse, ServerSentEvent

app = FastAPI()


@app.get("/logs/stream", response_class=EventSourceResponse)
async def stream_logs() -> AsyncIterable[ServerSentEvent]:
    logs = [
        "2025-01-01 INFO  Application started",
        "2025-01-01 DEBUG Connected to database",
        "2025-01-01 WARN  High memory usage detected",
    ]
    for log_line in logs:
        yield ServerSentEvent(raw_data=log_line)

data와 raw_data는 상호 배타적이에요. 각 ServerSentEvent에는 둘 중 하나만 설정할 수 있어요.

`Last-Event-ID`로 이어받기

브라우저가 연결이 끊긴 후 다시 연결하면, 마지막으로 받은 id를 Last-Event-ID 헤더로 보내요. 이걸 헤더 파라미터로 읽어서 클라이언트가 중단된 지점부터 스트림을 재개할 수 있어요.

from collections.abc import AsyncIterable
from typing import Annotated

from fastapi import FastAPI, Header
from fastapi.sse import EventSourceResponse, ServerSentEvent
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    price: float


items = [
    Item(name="Plumbus", price=32.99),
    Item(name="Portal Gun", price=999.99),
    Item(name="Meeseeks Box", price=49.99),
]


@app.get("/items/stream", response_class=EventSourceResponse)
async def stream_items(
    last_event_id: Annotated[int | None, Header()] = None,
) -> AsyncIterable[ServerSentEvent]:
    start = last_event_id + 1 if last_event_id is not None else 0
    for i, item in enumerate(items):
        if i < start:
            continue
        yield ServerSentEvent(data=item, id=str(i))

POST로 SSE 보내기

SSE는 GET뿐 아니라 어떤 HTTP 메서드와도 동작해요. MCP처럼 POST로 SSE를 스트리밍하는 프로토콜에서 유용해요.

from collections.abc import AsyncIterable

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse, ServerSentEvent
from pydantic import BaseModel

app = FastAPI()


class Prompt(BaseModel):
    text: str


@app.post("/chat/stream", response_class=EventSourceResponse)
async def stream_chat(prompt: Prompt) -> AsyncIterable[ServerSentEvent]:
    words = prompt.text.split()
    for word in words:
        yield ServerSentEvent(data=word, event="token")
    yield ServerSentEvent(raw_data="[DONE]", event="done")

기술적인 세부 사항

FastAPI는 몇 가지 SSE 모범 사례를 기본으로 구현해 두었어요. 그래서 특별히 뭘 하지 않아도 동작해요.

메시지가 없을 때 15초마다 keep-alive ping 코멘트를 보내서 일부 프록시가 연결을 끊지 못하게 해요 (HTML 스펙 권고에 따름).
스트림이 캐시되지 않도록 Cache-Control: no-cache 헤더를 설정해요.
Nginx 같은 프록시에서 버퍼링되지 않도록 특별한 X-Accel-Buffering: no 헤더를 설정해요.

더 알아보기 (Learn more)

Stream JSON Lines — JSON Lines를 통한 스트리밍 기초
Stream Data (고급) — 바이너리 데이터 스트리밍
헤더 파라미터 — Last-Event-ID 같은 헤더를 읽는 방법

Server-Sent Events(SSE)로 실시간 데이터 보내기