0Pricing
AI Engineering Academy · Aula

Streaming no FastAPI com eventos enviados pelo servidor

Crie um endpoint do FastAPI que encaminhe respostas de streaming do LLM para um cliente no navegador usando StreamingResponse e o tipo de conteúdo text/event-stream.

Streaming no FastAPI com eventos enviados pelo servidor é uma aula grátis de AI Engineering Academy no CoddyKit. Esta é a aula 3 de 4. Você pode ler a aula completa abaixo gratuitamente — depois pratica ao vivo no navegador com um editor de código integrado e um tutor de IA 24/7. Faz parte do caminho de aprendizado de AI Engineering Academy, e seu progresso é sincronizado entre a web e o app CoddyKit. O curso de AI Engineering Academy inclui 4 aulas no total.

Partes desta aula ainda não foram traduzidas e aparecem em inglês.

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.

Perguntas Frequentes

A aula “Streaming no FastAPI com eventos enviados pelo servidor” é grátis?

Sim — o texto completo de “Streaming no FastAPI com eventos enviados pelo servidor” é grátis para ler aqui na web. Para praticá-la interativamente (um editor de código integrado e um tutor de IA 24/7) e desbloquear o restante do curso de AI Engineering Academy, atualize para CoddyKit PRO. O curso de AI Engineering Academy inclui 4 aulas no total.

O que vou aprender em “Streaming no FastAPI com eventos enviados pelo servidor”?

Crie um endpoint do FastAPI que encaminhe respostas de streaming do LLM para um cliente no navegador usando StreamingResponse e o tipo de conteúdo text/event-stream. Você pratica AI Engineering Academy com código prático que executa diretamente no navegador, e um tutor de IA 24/7 responde suas dúvidas enquanto trabalha na aula.

Preciso ter experiência prévia para começar AI Engineering Academy?

Nenhuma experiência prévia é necessária. AI Engineering Academy no CoddyKit é estruturado para alunos iniciantes até avançados, então você pode começar aqui ou desde o início e aprender no seu ritmo. Esta é a aula 3 de 4.

Quanto tempo leva a aula “Streaming no FastAPI com eventos enviados pelo servidor”?

A maioria das aulas CoddyKit leva cerca de 5–10 minutos. Cada uma é compacta e interativa, então você faz progresso constante e retoma exatamente de onde parou entre web e app.

Posso escrever e executar código nesta aula de AI Engineering Academy?

Sim. Cada aula de AI Engineering Academy inclui um editor de código integrado, então você escreve e executa código real direto no navegador e recebe feedback de IA instantaneamente — nenhuma configuração local necessária.

Todas as aulas deste curso

  1. Entendendo o streaming de tokens
  2. Consumindo streams com o SDK para Python
  3. Streaming no FastAPI com eventos enviados pelo servidor
  4. Lidando com chamadas de ferramentas em respostas em streaming
← Voltar para AI Engineering Academy