How to beginner · 3 min read

How to use Server-Sent Events with FastAPI

Q: How to use Server-Sent Events with FastAPI

Use FastAPI's StreamingResponse to implement Server-Sent Events (SSE) by streaming a generator that yields properly formatted text/event-stream messages. This enables real-time one-way communication from server to client efficiently in Python web apps.

Quick answer

Use FastAPI's StreamingResponse to implement Server-Sent Events (SSE) by streaming a generator that yields properly formatted text/event-stream messages. This enables real-time one-way communication from server to client efficiently in Python web apps.

PREREQUISITES

Python 3.8+
pip install fastapi uvicorn
Basic knowledge of async Python

Setup

Install fastapi and uvicorn to run the FastAPI server. Use the following command:

bash

pip install fastapi uvicorn

output

Collecting fastapi
Collecting uvicorn
Successfully installed fastapi uvicorn

Step by step

Create a FastAPI app that streams Server-Sent Events by returning a StreamingResponse with media_type="text/event-stream". The generator yields SSE-formatted strings with event data.

python

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio
import os

app = FastAPI()

async def event_generator():
    for i in range(1, 6):
        # SSE format: data: message\n\n
        yield f"data: Message {i}\n\n"
        await asyncio.sleep(1)

@app.get("/stream")
async def stream():
    return StreamingResponse(event_generator(), media_type="text/event-stream")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="127.0.0.1", port=8000)

output

INFO:     Started server process [12345]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

Common variations

Use async def generators for non-blocking streaming.
Send JSON data by formatting data: {json_string}\n\n.
Use different models or APIs inside the generator to stream AI completions.
Integrate with OpenAI SDK streaming by yielding chunks as SSE messages.

python

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import OpenAI
import os

app = FastAPI()
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])

async def openai_stream():
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Say hello in parts."}],
        stream=True
    )
    async for chunk in response:
        content = chunk.choices[0].delta.content
        if content:
            yield f"data: {content}\n\n"

@app.get("/ai-stream")
async def ai_stream():
    return StreamingResponse(openai_stream(), media_type="text/event-stream")

output

INFO:     Started server process [12346]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

Troubleshooting

If the client does not receive events, ensure the media_type is set to text/event-stream.
Check that the generator yields strings ending with \n\n to delimit events.
Use browser developer tools to verify SSE connection and events.
For CORS issues, configure FastAPI middleware accordingly.

✅

Key Takeaways

Use FastAPI's StreamingResponse with media_type="text/event-stream" to implement SSE.
Yield properly formatted SSE messages ending with double newline \n\n from an async generator.
Integrate AI streaming by yielding chunks from OpenAI SDK's streaming responses as SSE events.
Test SSE endpoints with browsers or tools like curl to verify real-time streaming.
Configure CORS and event formatting carefully to avoid common SSE connection issues.

Verified 2026-04 · gpt-4o-mini

Verify ↗