0Pricing
AI Engineering Academy · レッスン

Server-Sent EventsによるFastAPIストリーミング

StreamingResponseとtext/event-streamのコンテンツタイプを使い、LLMのストリーミングレスポンスをブラウザクライアントに中継するFastAPIエンドポイントを構築します。

「Server-Sent EventsによるFastAPIストリーミング」はCoddyKit上の無料AI Engineering Academyレッスンです。 これはレッスン3/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Engineering Academy学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Engineering Academyコースには全4レッスンが含まれています。

このレッスンの一部はまだ翻訳されておらず、英語で表示されています。

Why Server-Sent Events for LLM Streaming

Server-Sent Events (SSE) is a W3C standard that lets a server push a stream of text events to a browser client over a single long-lived HTTP connection. Unlike WebSockets, SSE is one-directional (server to client), works over standard HTTP/1.1, automatically reconnects on disconnection, and requires no special browser library. These properties make it the ideal transport for streaming LLM tokens from a FastAPI backend to a web frontend.

SSE Wire Format

SSE sends text data formatted as a series of fields separated by newlines. Each event contains an optional event type field, a data field with the payload, and an optional id for reconnection. Events are separated by a blank line. For LLM streaming, send each token as a data: token_text\n\n line and a special data: [DONE]\n\n event at the end to signal stream completion.

# SSE wire format example
'''
data: The\n\n
data:  capital\n\n
data:  of\n\n
data:  France\n\n
data:  is\n\n
data:  Paris\n\n
data: [DONE]\n\n
'''

# Each 'data:' line is one event.
# The double newline (\n\n) terminates each event.
# The client receives these as EventSource message events.
# The content-type must be 'text/event-stream'.

StreamingResponse in FastAPI

FastAPI's StreamingResponse accepts an async generator that yields strings and streams them to the client. By setting the media_type to 'text/event-stream' and formatting each yielded string as an SSE event, you turn any async generator into a proper SSE stream. FastAPI handles connection lifecycle, flushing, and HTTP headers automatically.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI
import asyncio

app = FastAPI()
async_client = AsyncOpenAI()

async def token_generator(prompt: str):
    stream = await async_client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        stream=True,
    )
    async for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            yield f'data: {delta}\n\n'  # SSE format
    yield 'data: [DONE]\n\n'

@app.get('/stream')
async def stream_endpoint(prompt: str):
    return StreamingResponse(
        token_generator(prompt),
        media_type='text/event-stream',
        headers={'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no'},
    )

Important HTTP Headers for SSE

Three HTTP headers are critical for SSE to work correctly through proxies and CDNs. Cache-Control: no-cache prevents intermediaries from caching the stream. Connection: keep-alive keeps the TCP connection open. X-Accel-Buffering: no disables Nginx's response buffering, which would otherwise batch chunks and defeat the streaming effect. Without this last header, Nginx will buffer all output before forwarding to the browser.

from fastapi.responses import StreamingResponse

SSE_HEADERS = {
    'Cache-Control': 'no-cache',
    'Connection': 'keep-alive',
    'X-Accel-Buffering': 'no',   # disable nginx buffering
    'Access-Control-Allow-Origin': '*',  # CORS for cross-origin clients
}

@app.get('/chat')
async def chat_stream(prompt: str):
    return StreamingResponse(
        token_generator(prompt),
        media_type='text/event-stream',
        headers=SSE_HEADERS,
    )

Structured SSE Events with JSON Payloads

For richer streaming APIs, encode each event payload as JSON rather than raw text. This lets you include metadata alongside the token — for example, the token type (content vs tool call), a message ID, or a latency timestamp. The browser client parses each event's JSON and routes different event types to different UI components.

import json
import time

async def json_token_generator(prompt: str, session_id: str):
    t_start = time.perf_counter()
    stream = await async_client.chat.completions.create(
        model='gpt-4o-mini',
        messages=[{'role': 'user', 'content': prompt}],
        stream=True,
    )
    async for chunk in stream:
        delta = chunk.choices[0].delta.content
        if delta:
            payload = json.dumps({
                'type': 'token',
                'content': delta,
                'session_id': session_id,
                't_ms': round((time.perf_counter() - t_start) * 1000),
            })
            yield f'data: {payload}\n\n'
    # Send completion event
    yield f'data: {json.dumps({"type": "done", "session_id": session_id})}\n\n'

Consuming SSE in a Browser (JavaScript)

The browser-side EventSource API connects to an SSE endpoint and fires events as they arrive. For token streaming, listen for the default message event, parse the data as JSON or treat it as a raw string, and append each token to the DOM. Handle the [DONE] sentinel by closing the EventSource connection.

// Browser-side JavaScript
const prompt = 'Explain hybrid search in one paragraph.';
const url = '/stream?prompt=' + encodeURIComponent(prompt);

const source = new EventSource(url);
const output = document.getElementById('output');

source.onmessage = (event) => {
  if (event.data === '[DONE]') {
    source.close();  // stop listening
    return;
  }
  output.textContent += event.data;  // append each token
};

source.onerror = (err) => {
  console.error('SSE error:', err);
  source.close();
};

POST Requests with fetch for Streaming

EventSource only supports GET requests, which is limiting for complex prompts. For POST requests (sending a JSON body with conversation history), use the browser's fetch API with the Streams API to read the response body incrementally. This pattern is used by ChatGPT's web interface and most production LLM chat UIs.

// Browser-side: POST with fetch and ReadableStream
async function streamPost(messages) {
  const response = await fetch('/chat', {
    method: 'POST',
    headers: {'Content-Type': 'application/json'},
    body: JSON.stringify({messages}),
  });

  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  const output = document.getElementById('output');

  while (true) {
    const {done, value} = await reader.read();
    if (done) break;
    const text = decoder.decode(value, {stream: true});
    // Parse SSE lines
    for (const line of text.split('\n')) {
      if (line.startsWith('data: ') && line !== 'data: [DONE]') {
        output.textContent += line.slice(6);
      }
    }
  }
}

FastAPI POST Endpoint for Chat Streaming

For POST-based chat streaming, define a Pydantic model for the request body, accept a list of messages, and stream the LLM response. This enables passing full conversation history with each request, supporting multi-turn chat applications. The pattern is identical to GET streaming except you extract the prompt from the request body.

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

class ChatRequest(BaseModel):
    messages: list[dict]
    model: str = 'gpt-4o-mini'

@app.post('/chat')
async def chat_post(request: ChatRequest):
    async def generate():
        stream = await async_client.chat.completions.create(
            model=request.model,
            messages=request.messages,
            stream=True,
        )
        async for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                yield f'data: {delta}\n\n'
        yield 'data: [DONE]\n\n'

    return StreamingResponse(
        generate(),
        media_type='text/event-stream',
        headers=SSE_HEADERS,
    )

Handling Client Disconnects

When a browser user navigates away or closes the tab, the HTTP connection closes and FastAPI raises asyncio.CancelledError in the streaming generator. Always handle this to avoid leaving LLM streaming requests open and incurring unnecessary API costs. Wrap your generator in a try/except for CancelledError and cancel the OpenAI stream when detected.

from fastapi import Request

@app.get('/stream')
async def stream_with_disconnect(prompt: str, request: Request):
    async def generate_with_cancel():
        try:
            stream = await async_client.chat.completions.create(
                model='gpt-4o-mini',
                messages=[{'role': 'user', 'content': prompt}],
                stream=True,
            )
            async for chunk in stream:
                if await request.is_disconnected():
                    break  # client gone, stop generating
                delta = chunk.choices[0].delta.content
                if delta:
                    yield f'data: {delta}\n\n'
        except asyncio.CancelledError:
            pass  # client disconnected
        finally:
            yield 'data: [DONE]\n\n'

    return StreamingResponse(generate_with_cancel(), media_type='text/event-stream')

Adding Request Authentication

Production streaming endpoints must authenticate requests to prevent unauthorized LLM usage. Use FastAPI's Depends with an API key or JWT header check. Authentication happens before the generator starts, so the overhead is minimal and the stream only begins after the user is verified.

from fastapi import Header, HTTPException, Depends

VALID_API_KEYS = {'sk-demo-key-1', 'sk-demo-key-2'}

async def verify_api_key(x_api_key: str = Header(None)):
    if x_api_key not in VALID_API_KEYS:
        raise HTTPException(status_code=401, detail='Invalid API key')
    return x_api_key

@app.post('/chat')
async def authenticated_chat(
    request: ChatRequest,
    api_key: str = Depends(verify_api_key),
):
    async def generate():
        stream = await async_client.chat.completions.create(
            model=request.model,
            messages=request.messages,
            stream=True,
        )
        async for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                yield f'data: {delta}\n\n'
        yield 'data: [DONE]\n\n'

    return StreamingResponse(generate(), media_type='text/event-stream', headers=SSE_HEADERS)

Testing SSE Endpoints

Test streaming endpoints with FastAPI's TestClient in streaming mode. Use with client.stream('GET', '/stream', params={...}) as r and iterate over r.iter_lines() to receive SSE events. This lets you verify that tokens are formatted correctly, the DONE sentinel is sent, and error cases produce appropriate SSE error events.

from fastapi.testclient import TestClient

def test_sse_endpoint():
    with TestClient(app) as client:
        with client.stream('GET', '/stream', params={'prompt': 'Say hi'}) as r:
            assert r.status_code == 200
            assert 'text/event-stream' in r.headers['content-type']

            events = []
            for line in r.iter_lines():
                if line.startswith('data: '):
                    events.append(line[6:])

            assert events[-1] == '[DONE]'
            full_text = ''.join(e for e in events if e != '[DONE]')
            assert len(full_text) > 0

Quick Check

Test your understanding of FastAPI streaming with SSE from this lesson.

Lesson Recap

In this lesson you learned: Server-Sent Events is the standard HTTP transport for streaming LLM tokens to browser clients, StreamingResponse with text/event-stream turns any async generator into an SSE stream in FastAPI, and critical headers including X-Accel-Buffering and Cache-Control are required for correct behavior behind proxies. Handle client disconnects to avoid orphaned LLM API calls. Next up we tackle streaming responses that contain tool calls.

よくある質問

「Server-Sent EventsによるFastAPIストリーミング」レッスンは無料ですか？

はい。「Server-Sent EventsによるFastAPIストリーミング」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Engineering Academyコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Engineering Academyコースには全4レッスンが含まれています。

「Server-Sent EventsによるFastAPIストリーミング」で何を学びますか？

StreamingResponseとtext/event-streamのコンテンツタイプを使い、LLMのストリーミングレスポンスをブラウザクライアントに中継するFastAPIエンドポイントを構築します。 ブラウザで直接実行するハンズオンコードでAI Engineering Academyを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。

AI Engineering Academyを始めるのに経験は必要ですか？

事前経験は必要ありません。CoddyKitのAI Engineering Academyは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン3/4です。

「Server-Sent EventsによるFastAPIストリーミング」レッスンにはどのくらい時間がかかりますか？

ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。

このAI Engineering Academyレッスンでコードを書いて実行できますか？

はい。すべてのAI Engineering Academyレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。

このコースのすべてのレッスン

  1. トークンストリーミングを理解する
  2. Python SDKでストリームを利用する
  3. Server-Sent EventsによるFastAPIストリーミング
  4. ストリーミングレスポンスのツール呼び出しを処理する
← AI Engineering Academyに戻る