Streaming in FastAPI mit Server-Sent Events
Erstellen Sie einen FastAPI-Endpunkt, der gestreamte LLM-Antworten mithilfe von StreamingResponse und dem Content-Typ text/event-stream an einen Browser-Client weiterleitet.
Streaming in FastAPI mit Server-Sent Events ist eine kostenlose AI Engineering Academy-Lektion auf CoddyKit. Dies ist Lektion 3 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des AI Engineering Academy-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der AI Engineering Academy-Kurs umfasst insgesamt 4 Lektionen.
Warum Server-Sent Events für LLM-Streaming
Server-Sent Events (SSE) sind ein W3C-Standard, mit dem ein Server über eine einzige langlebige HTTP-Verbindung einen Stream von Textereignissen an einen Browser-Client senden kann. Anders als WebSockets sind SSE unidirektional (vom Server zum Client), funktionieren über standardmäßiges HTTP/1.1, stellen die Verbindung bei einer Unterbrechung automatisch wieder her und benötigen keine spezielle Browserbibliothek. Diese Eigenschaften machen SSE zum idealen Transport für das Streaming von LLM-Token von einem FastAPI-Backend an ein Web-Frontend.
SSE-Drahtformat
SSE sendet Textdaten, die als Reihe von durch Zeilenumbrüche getrennten Feldern formatiert sind. Jedes Ereignis enthält ein optionales Feld event für den Typ, ein Feld data mit der Nutzlast und eine optionale id für die Wiederverbindung. Ereignisse werden durch eine Leerzeile getrennt. Beim LLM-Streaming senden Sie jedes Token als Zeile data: token_text\n\n und am Ende ein spezielles Ereignis data: [DONE]\n\n, um den Abschluss des Streams zu signalisieren.
# 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
Die StreamingResponse von FastAPI akzeptiert einen asynchronen Generator, der Zeichenketten liefert, und streamt diese an den Client. Indem Sie media_type auf 'text/event-stream' setzen und jede gelieferte Zeichenkette als SSE-Ereignis formatieren, verwandeln Sie jeden asynchronen Generator in einen korrekten SSE-Stream. FastAPI übernimmt den Lebenszyklus der Verbindung, das Leeren des Puffers und die HTTP-Header automatisch.
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'},
)Wichtige HTTP-Header für SSE
Drei HTTP-Header sind entscheidend dafür, dass SSE über Proxys und CDNs korrekt funktioniert. Cache-Control: no-cache verhindert, dass zwischengeschaltete Komponenten den Stream zwischenspeichern. Connection: keep-alive hält die TCP-Verbindung offen. X-Accel-Buffering: no deaktiviert das Antwort-Puffern von Nginx, das Chunks andernfalls bündeln und den Streaming-Effekt zunichtemachen würde. Ohne diesen letzten Header puffert Nginx die gesamte Ausgabe, bevor sie an den Browser weitergeleitet wird.
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,
)Strukturierte SSE-Ereignisse mit JSON-Nutzlasten
Für umfangreichere Streaming-APIs sollten Sie die Nutzlast jedes Ereignisses als JSON statt als reinen Text kodieren. Dadurch können Sie Metadaten zusammen mit dem Token übertragen, beispielsweise den Tokentyp (Inhalt oder Tool-Aufruf), eine Nachrichten-ID oder einen Latenzzeitstempel. Der Browser-Client analysiert das JSON jedes Ereignisses und leitet die verschiedenen Ereignistypen an unterschiedliche UI-Komponenten weiter.
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'SSE in einem Browser verwenden (JavaScript)
Die browserseitige EventSource-API stellt eine Verbindung zu einem SSE-Endpunkt her und löst Ereignisse aus, sobald sie eintreffen. Beim Token-Streaming warten Sie auf das standardmäßige message-Ereignis, analysieren die Daten als JSON oder behandeln sie als rohe Zeichenkette und hängen jedes Token an das DOM an. Behandeln Sie das Sentinel [DONE], indem Sie die EventSource-Verbindung schließen.
// 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-Anfragen mit fetch für Streaming
EventSource unterstützt nur GET-Anfragen, was bei komplexen Prompts eine Einschränkung darstellt. Für POST-Anfragen, bei denen Sie einen JSON-Body mit dem Gesprächsverlauf senden, verwenden Sie die fetch-API des Browsers zusammen mit der Streams API, um den Antwort-Body schrittweise zu lesen. Dieses Muster wird von der Weboberfläche von ChatGPT und den meisten produktiven LLM-Chat-UIs verwendet.
// 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-Endpunkt für Chat-Streaming
Für POST-basiertes Chat-Streaming definieren Sie ein Pydantic-Modell für den Anfrage-Body, akzeptieren eine Nachrichtenliste und streamen die LLM-Antwort. Dadurch können Sie bei jeder Anfrage den vollständigen Gesprächsverlauf übergeben und so Chat-Anwendungen mit mehreren Dialogrunden unterstützen. Das Muster ist identisch mit GET-Streaming, außer dass Sie den Prompt aus dem Anfrage-Body extrahieren.
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,
)Umgang mit getrennten Clients
Wenn ein Benutzer im Browser weg navigiert oder den Tab schließt, wird die HTTP-Verbindung geschlossen und FastAPI löst im Streaming-Generator asyncio.CancelledError aus. Behandeln Sie dies immer, damit keine LLM-Streaming-Anfragen offen bleiben und unnötige API-Kosten verursachen. Schließen Sie Ihren Generator in einen try/except-Block für CancelledError ein und brechen Sie den OpenAI-Stream ab, sobald der Fehler erkannt wird.
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')Anfrageauthentifizierung hinzufügen
Produktive Streaming-Endpunkte müssen Anfragen authentifizieren, um eine unbefugte Nutzung von LLMs zu verhindern. Verwenden Sie FastAPI's Depends mit einem API-Schlüssel oder einer Prüfung des JWT-Headers. Die Authentifizierung erfolgt, bevor der Generator startet, sodass der Mehraufwand minimal ist und der Stream erst beginnt, nachdem der Benutzer verifiziert wurde.
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)SSE-Endpunkte testen
Testen Sie Streaming-Endpunkte mit FastAPI's TestClient im Streaming-Modus. Verwenden Sie with client.stream('GET', '/stream', params={...}) as r und iterieren Sie über r.iter_lines(), um SSE-Ereignisse zu empfangen. So können Sie überprüfen, ob Token korrekt formatiert werden, das DONE-Sentinel gesendet wird und Fehlerfälle geeignete SSE-Fehlerereignisse erzeugen.
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) > 0Schnelltest
Testen Sie Ihr Verständnis des FastAPI-Streamings mit SSE aus dieser Lektion.
Zusammenfassung der Lektion
In dieser Lektion haben Sie gelernt: Server-Sent Events ist der standardmäßige HTTP-Transport, um LLM-Token an Browser-Clients zu streamen, StreamingResponse mit text/event-stream wandelt jeden asynchronen Generator in FastAPI in einen SSE-Stream um, und wichtige Header wie X-Accel-Buffering und Cache-Control sind für das korrekte Verhalten hinter Proxys erforderlich. Behandeln Sie die Trennung von Clients, um verwaiste LLM-API-Aufrufe zu vermeiden. Als Nächstes beschäftigen wir uns mit Streaming-Antworten, die Tool-Aufrufe enthalten.
Häufig gestellte Fragen
Ist die Lektion „Streaming in FastAPI mit Server-Sent Events“ kostenlos?
Ja — der vollständige Text von „Streaming in FastAPI mit Server-Sent Events“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des AI Engineering Academy-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der AI Engineering Academy-Kurs umfasst insgesamt 4 Lektionen.
Was lerne ich in „Streaming in FastAPI mit Server-Sent Events“?
Erstellen Sie einen FastAPI-Endpunkt, der gestreamte LLM-Antworten mithilfe von StreamingResponse und dem Content-Typ text/event-stream an einen Browser-Client weiterleitet. Du übst AI Engineering Academy mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.
Brauche ich Erfahrung, um AI Engineering Academy zu starten?
Keine Vorkenntnisse erforderlich. AI Engineering Academy auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 3 von 4.
Wie lange dauert die Lektion „Streaming in FastAPI mit Server-Sent Events“?
Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.
Kann ich in dieser AI Engineering Academy-Lektion Code schreiben und ausführen?
Ja. Jede AI Engineering Academy-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.
Alle Lektionen in diesem Kurs
- Token-Streaming verstehen
- Streams mit dem Python-SDK verarbeiten
- Streaming in FastAPI mit Server-Sent Events
- Tool-Aufrufe in gestreamten Antworten verarbeiten