Server-Sent Events (SSE)

Meneruskan token LLM ke browser secara real-time memakai Server-Sent Events bawaan FastAPI.

SSE vs WebSocket

Bentuk paling sederhana

from collections.abc import AsyncIterable
from fastapi import FastAPI
from fastapi.sse import EventSourceResponse

app = FastAPI()

@app.post("/chat/stream", response_class=EventSourceResponse)
async def chat_stream(req: ChatRequest) -> AsyncIterable[str]:
    async with async_client.messages.stream(
        model="claude-opus-5",
        max_tokens=16000,
        messages=[{"role": "user", "content": req.pesan}],
    ) as stream:
        async for teks in stream.text_stream:
            yield teks

Yang diurus FastAPI untukmu:

• Header Content-Type: text/event-stream
• Cache-Control: no-cache
• X-Accel-Buffering: no — mencegah nginx menahan potongan data
• Ping keep-alive berkala supaya koneksi tidak diputus perantara

Event terstruktur

import json
from fastapi.sse import ServerSentEvent

@app.post("/chat/stream", response_class=EventSourceResponse)
async def chat_stream(req: ChatRequest) -> AsyncIterable[ServerSentEvent]:
    async with async_client.messages.stream(...) as stream:
        async for event in stream:
            if event.type == "content_block_delta" and event.delta.type == "text_delta":
                yield ServerSentEvent(
                    event="token",
                    data=json.dumps({"teks": event.delta.text}),
                )

        final = await stream.get_final_message()
        yield ServerSentEvent(
            event="selesai",
            data=json.dumps({
                "token_masuk": final.usage.input_tokens,
                "token_keluar": final.usage.output_tokens,
                "biaya": hitung_biaya(final.usage),
            }),
        )

Sisi browser

const res = await fetch("/chat/stream", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ pesan: teks, session_id: sid }),
});

const reader = res.body.pipeThrough(new TextDecoderStream()).getReader();
while (true) {
  const { value, done } = await reader.read();
  if (done) break;
  tampilkan(value);
}

Menangani error di tengah stream

@app.post("/chat/stream", response_class=EventSourceResponse)
async def chat_stream(req: ChatRequest) -> AsyncIterable[ServerSentEvent]:
    try:
        async with async_client.messages.stream(...) as stream:
            async for teks in stream.text_stream:
                yield ServerSentEvent(event="token", data=teks)
    except anthropic.RateLimitError:
        yield ServerSentEvent(event="error", data="layanan sedang sibuk")
    except Exception:
        logger.exception("stream gagal")
        yield ServerSentEvent(event="error", data="terjadi kesalahan")

Klien memutus koneksi

from fastapi import Request

@app.post("/chat/stream", response_class=EventSourceResponse)
async def chat_stream(req: ChatRequest, request: Request) -> AsyncIterable[str]:
    async with async_client.messages.stream(...) as stream:
        async for teks in stream.text_stream:
            if await request.is_disconnected():
                logger.info("klien memutus, hentikan generasi")
                break
            yield teks

Ini menghemat uang nyata: kalau pengguna menutup tab di tengah jawaban panjang, tidak ada gunanya terus membayar token yang tidak akan pernah dibaca.

Sambung ulang dengan Last-Event-ID

@app.get("/kejadian", response_class=EventSourceResponse)
async def kejadian(request: Request) -> AsyncIterable[ServerSentEvent]:
    terakhir = request.headers.get("last-event-id")
    mulai = int(terakhir) + 1 if terakhir else 0

    for i, item in enumerate(ambil_kejadian(sejak=mulai), start=mulai):
        yield ServerSentEvent(id=str(i), data=json.dumps(item))

Browser mengirim header Last-Event-ID secara otomatis saat menyambung ulang. Berguna untuk feed peristiwa; untuk jawaban chat, biasanya lebih sederhana memulai ulang saja.

Di belakang nginx

location /chat/stream {
    proxy_pass http://api:8000;
    proxy_buffering off;         # ← WAJIB, kalau tidak stream ditahan
    proxy_cache off;
    proxy_read_timeout 300s;
    proxy_set_header Connection '';
    proxy_http_version 1.1;
}

Checkpoint Fase 5

from contextlib import asynccontextmanager
from collections.abc import AsyncIterable
from fastapi import FastAPI, Request
from fastapi.sse import EventSourceResponse, ServerSentEvent
import anthropic, json

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.llm = anthropic.AsyncAnthropic()
    yield
    await app.state.llm.close()

app = FastAPI(lifespan=lifespan)

@app.post("/chat/stream", response_class=EventSourceResponse)
async def chat_stream(req: ChatRequest, request: Request) -> AsyncIterable[ServerSentEvent]:
    client = request.app.state.llm
    async with client.messages.stream(
        model="claude-opus-5", max_tokens=16000,
        messages=[{"role": "user", "content": req.pesan}],
    ) as stream:
        async for teks in stream.text_stream:
            if await request.is_disconnected():
                break
            yield ServerSentEvent(event="token", data=teks)

        final = await stream.get_final_message()
        yield ServerSentEvent(
            event="selesai",
            data=json.dumps({"token_keluar": final.usage.output_tokens}),
        )