AI Engineering Academy · Lezione

Streaming in FastAPI con Server-Sent Events

Costruisca un endpoint FastAPI che inoltri le risposte LLM in streaming a un client browser usando StreamingResponse e il tipo di contenuto text/event-stream.

Lezione 3 di 413 passaggi

Streaming in FastAPI con Server-Sent Events è una lezione AI Engineering Academy gratuita su CoddyKit. Questa è la lezione 3 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Engineering Academy, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Engineering Academy include 4 lezioni in totale.

Perché utilizzare Server-Sent Events per lo streaming LLM

Server-Sent Events (SSE) è uno standard W3C che consente a un server di inviare uno stream di eventi testuali a un client browser tramite un'unica connessione HTTP di lunga durata. A differenza dei WebSocket, SSE è unidirezionale (dal server al client), funziona sul normale protocollo HTTP/1.1, si riconnette automaticamente in caso di disconnessione e non richiede librerie browser specifiche. Queste caratteristiche lo rendono il trasporto ideale per inviare in streaming i token LLM da un backend FastAPI a un frontend web.

Formato SSE sul filo

SSE invia dati testuali formattati come una serie di campi separati da nuove righe. Ogni evento contiene un campo di tipo event facoltativo, un campo data con il payload e un id facoltativo per la riconnessione. Gli eventi sono separati da una riga vuota. Per lo streaming LLM, invii ogni token come una riga data: token_text\n\n e, al termine, un evento speciale data: [DONE]\n\n per segnalare la conclusione dello stream.

# 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

StreamingResponse di FastAPI accetta un generatore asincrono che restituisce stringhe e le invia in streaming al client. Impostando media_type su 'text/event-stream' e formattando ogni stringa restituita come un evento SSE, trasforma qualsiasi generatore asincrono in un vero stream SSE. FastAPI gestisce automaticamente il ciclo di vita della connessione, il flush e le intestazioni HTTP.

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'},
    )

Intestazioni HTTP importanti per SSE

Tre intestazioni HTTP sono fondamentali affinché SSE funzioni correttamente attraverso proxy e CDN. Cache-Control: no-cache impedisce agli intermediari di memorizzare nella cache lo stream. Connection: keep-alive mantiene aperta la connessione TCP. X-Accel-Buffering: no disabilita il buffering delle risposte di Nginx, che altrimenti raggrupperebbe i chunk vanificando l'effetto dello streaming. Senza quest'ultima intestazione, Nginx memorizzerebbe tutto l'output prima di inoltrarlo al 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,
    )

Eventi SSE strutturati con payload JSON

Per API di streaming più ricche, codifichi il payload di ogni evento come JSON anziché come testo non elaborato. In questo modo può includere metadati insieme al token, ad esempio il tipo di token (contenuto o chiamata a uno strumento), un ID del messaggio o un timestamp della latenza. Il client browser analizza il JSON di ogni evento e indirizza i diversi tipi di evento a componenti UI differenti.

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'

Consumo di SSE in un browser (JavaScript)

L'API browser-side EventSource si connette a un endpoint SSE e genera eventi man mano che arrivano. Per lo streaming dei token, ascolti l'evento predefinito message, analizzi i dati come JSON o li tratti come una stringa non elaborata e aggiunga ogni token al DOM. Gestisca il sentinel [DONE] chiudendo la connessione EventSource.

// 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();
};

Richieste POST con fetch per lo streaming

EventSource supporta solo richieste GET, il che è limitante per i prompt complessi. Per le richieste POST, che inviano un corpo JSON con la cronologia della conversazione, utilizzi l'API browser fetch insieme alla Streams API per leggere progressivamente il corpo della risposta. Questo pattern è utilizzato dall'interfaccia web di ChatGPT e dalla maggior parte delle UI di chat LLM in produzione.

// 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);
      }
    }
  }
}

Endpoint POST FastAPI per lo streaming della chat

Per lo streaming della chat basato su POST, definisca un modello Pydantic per il corpo della richiesta, accetti un elenco di messaggi e trasmetta in streaming la risposta dell'LLM. In questo modo può inviare l'intera cronologia della conversazione a ogni richiesta, supportando applicazioni di chat multi-turno. Il pattern è identico allo streaming GET, con la differenza che il prompt viene estratto dal corpo della richiesta.

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,
    )

Gestione delle disconnessioni del client

Quando un utente del browser cambia pagina o chiude la scheda, la connessione HTTP si chiude e FastAPI solleva asyncio.CancelledError nel generatore di streaming. Gestisca sempre questo caso per evitare di lasciare aperte richieste di streaming LLM e di sostenere costi API inutili. Racchiuda il generatore in un blocco try/except per CancelledError e annulli lo stream OpenAI quando rileva la disconnessione.

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')

Aggiunta dell'autenticazione delle richieste

Gli endpoint di streaming in produzione devono autenticare le richieste per impedire l'uso non autorizzato dell'LLM. Utilizzi Depends di FastAPI insieme a una chiave API o a un controllo dell'intestazione JWT. L'autenticazione avviene prima dell'avvio del generatore, quindi il sovraccarico è minimo e lo stream inizia solo dopo la verifica dell'utente.

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)

Test degli endpoint SSE

Testi gli endpoint di streaming con TestClient di FastAPI in modalità streaming. Utilizzi with client.stream('GET', '/stream', params={...}) as r e iteri su r.iter_lines() per ricevere gli eventi SSE. In questo modo può verificare che i token siano formattati correttamente, che venga inviato il sentinel DONE e che i casi di errore producano eventi SSE di errore appropriati.

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

Verifica rapida

Verifichi la sua comprensione dello streaming FastAPI con SSE trattato in questa lezione.

Rispondi nella lezione interattiva

Riepilogo della lezione

In questa lezione ha imparato che Server-Sent Events è il trasporto HTTP standard per inviare in streaming i token LLM ai client browser, che StreamingResponse con text/event-stream trasforma qualsiasi generatore asincrono in uno stream SSE in FastAPI e che le intestazioni fondamentali, tra cui X-Accel-Buffering e Cache-Control, sono necessarie per un comportamento corretto dietro i proxy. Gestisca le disconnessioni del client per evitare chiamate API LLM orfane. Nel prossimo argomento affronteremo le risposte in streaming che contengono chiamate a strumenti.

Gratis per iniziare

Impara Python con un tutor IA — gratis

Scrivi ed esegui vero codice nel tuo browser, ricevi aiuto istantaneo da un tutor IA disponibile 24/7, e riprendi da dove hai lasciato sul web o nell'app.

Corsi: 30
Lezioni: 120

Inizia a imparare gratis Visualizza il corso completo

Domande Frequenti

La lezione «Streaming in FastAPI con Server-Sent Events» è gratuita?

Sì — il testo completo di «Streaming in FastAPI con Server-Sent Events» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Engineering Academy, passa a CoddyKit PRO. Il corso AI Engineering Academy include 4 lezioni in totale.

Cosa imparerò in «Streaming in FastAPI con Server-Sent Events»?

Costruisca un endpoint FastAPI che inoltri le risposte LLM in streaming a un client browser usando StreamingResponse e il tipo di contenuto text/event-stream. Eserciti AI Engineering Academy con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.

Ho bisogno di esperienza per iniziare AI Engineering Academy?

Non è richiesta alcuna esperienza precedente. AI Engineering Academy su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 3 di 4.

Quanto tempo richiede la lezione «Streaming in FastAPI con Server-Sent Events»?

La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.

Posso scrivere ed eseguire codice in questa lezione AI Engineering Academy?

Sì. Ogni lezione AI Engineering Academy include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.

Tutte le lezioni di questo corso

Comprendere lo streaming dei token
Consumare gli stream con l'SDK Python
Streaming in FastAPI con Server-Sent Events
Gestire le chiamate agli strumenti nelle risposte in streaming

← Torna a AI Engineering Academy