How to Use Server Sent Events in FastAPI: Simple Guide

To use Server Sent Events (SSE) in FastAPI, create an endpoint that returns a StreamingResponse with a generator yielding event data strings formatted as data: message\n\n. Set the response media type to text/event-stream so the client can receive real-time updates.

Syntax

In FastAPI, SSE is implemented by returning a StreamingResponse from an endpoint. The response streams data continuously using a generator function that yields strings formatted as SSE events.

• generator(): Yields event data strings with the prefix data: and double newline \n\n to separate events.
• StreamingResponse(generator(), media_type='text/event-stream'): Sends the streamed data with the correct content type for SSE.

python

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

app = FastAPI()

def event_generator():
    count = 0
    while True:
        yield f"data: Message {count}\n\n"
        count += 1
        await asyncio.sleep(1)

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

Example

This example shows a FastAPI app with an SSE endpoint that sends a message every second. The client can connect to /sse to receive a continuous stream of messages labeled "Message 0", "Message 1", and so on.

python

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

app = FastAPI()

async def event_generator():
    count = 0
    while True:
        yield f"data: Message {count}\n\n"
        count += 1
        await asyncio.sleep(1)

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

Output

When you open http://localhost:8000/sse in a browser or SSE client, you receive a stream like:
data: Message 0

data: Message 1

data: Message 2

... continuing every second

Common Pitfalls

• Not using async generator: SSE requires non-blocking code; use async def and await asyncio.sleep() to avoid blocking the event loop.
• Incorrect event format: Each event must start with data: and end with two newlines \n\n to be recognized by clients.
• Wrong media type: Always set media_type='text/event-stream' in StreamingResponse for SSE.
• Forgetting to flush: Yielding strings in the generator flushes data; do not buffer large chunks.

python

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

app = FastAPI()

# Wrong: blocking sleep and missing async
# def event_generator():
#     count = 0
#     while True:
#         yield f"data: Message {count}\n\n"
#         count += 1
#         time.sleep(1)  # blocks event loop

# Correct async generator
async def event_generator():
    count = 0
    while True:
        yield f"data: Message {count}\n\n"
        count += 1
        await asyncio.sleep(1)

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

Quick Reference

Key points for using SSE in FastAPI:

• Use StreamingResponse with media_type='text/event-stream'.
• Implement an async generator yielding strings formatted as data: message\n\n.
• Use await asyncio.sleep() for delays to keep the server responsive.
• Clients receive real-time updates without polling.

Key Takeaways

Use an async generator yielding properly formatted SSE strings with double newlines.

Return a StreamingResponse with media_type set to 'text/event-stream' for SSE endpoints.

Avoid blocking calls; use await asyncio.sleep() to keep the server responsive.

Each SSE event must start with 'data:' and end with two newlines to be valid.

Clients can listen to the SSE endpoint to receive continuous real-time messages.