0Pricing
AI Engineering Academy · Lección

Streaming en FastAPI con eventos enviados por el servidor

Cree un endpoint de FastAPI que redirija las respuestas del LLM en streaming a un cliente de navegador mediante StreamingResponse y el tipo de contenido text/event-stream.

Streaming en FastAPI con eventos enviados por el servidor es una lección gratuita de AI Engineering Academy en CoddyKit. Esta es la lección 3 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Engineering Academy, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Engineering Academy incluye 4 lecciones en total.

Partes de esta lección aún no han sido traducidas y se muestran en 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.

Preguntas frecuentes

¿La lección «Streaming en FastAPI con eventos enviados por el servidor» es gratis?

Sí — el texto completo de «Streaming en FastAPI con eventos enviados por el servidor» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Engineering Academy, actualiza a CoddyKit PRO. El curso de AI Engineering Academy incluye 4 lecciones en total.

¿Qué aprenderé en «Streaming en FastAPI con eventos enviados por el servidor»?

Cree un endpoint de FastAPI que redirija las respuestas del LLM en streaming a un cliente de navegador mediante StreamingResponse y el tipo de contenido text/event-stream. Practicas AI Engineering Academy con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.

¿Necesito experiencia previa para empezar AI Engineering Academy?

No se requiere experiencia previa. AI Engineering Academy en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 3 de 4.

¿Cuánto tiempo toma la lección «Streaming en FastAPI con eventos enviados por el servidor»?

La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.

¿Puedo escribir y ejecutar código en esta lección de AI Engineering Academy?

Sí. Cada lección de AI Engineering Academy incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.

Todas las lecciones de este curso

  1. Comprensión de la transmisión de tokens
  2. Consumo de streams con el SDK de Python
  3. Streaming en FastAPI con eventos enviados por el servidor
  4. Gestión de llamadas a herramientas en respuestas en streaming
← Volver a AI Engineering Academy