Server-Sent Events with Python FastAPI

24 Jul 2025 ·

Server-Sent Events with Python FastAPI

Here's your article converted to Markdown:

Server-Sent Events with Python FastAPI

Nanda Gopal Pattanayak Mar 9, 2024

SSE Communication Model

alt text

When we need to stream data continuously from the server to the client, we usually consider one of three options: polling, WebSockets, or Server-Sent Events (SSE). Each has pros and cons:

  • Polling can overload the server when no new data is available.
  • WebSockets support bidirectional communication but are harder to scale.
  • SSE is unidirectional — only server to client.

SSE in a Nutshell

  • Unidirectional: Server sends data to frontend. No request is needed for every update.
  • Text-based: Each message is plain text, ending with two newlines (\n\n) to denote event separation.
  • Event-typed: Messages can have an event: field to handle different event types.
  • Auto-reconnect: Clients automatically reconnect if the connection is interrupted.

Use Cases

  • Streaming live logs
  • Updating live locations
  • Real-time dashboard updates

Backend with FastAPI

We have a JSON file with coordinates. We want to stream one coordinate every second.

Sample waypoints.json:

[
    { "lat": 22.09769, "lng": 87.24068 },
    { "lat": 22.09776, "lng": 87.24075 },
    { "lat": 22.09784, "lng": 87.24082 },
    { "lat": 22.09811, "lng": 87.24098 }
]

FastAPI Code:

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
import json, uvicorn
from asyncio import sleep

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

async def waypoints_generator():
    waypoints = open('waypoints.json')
    waypoints = json.load(waypoints)
    for waypoint in waypoints[0: 10]:
        data = json.dumps(waypoint)
        yield f"event: locationUpdate\ndata: {data}\n\n"
        await sleep(1)

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

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

Frontend

HTML:

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <title>Live Vehicle Coordinates</title>
</head>
<body>
    <div id="coordinates-container">
        <h2>Live Vehicle Coordinates</h2>
        <div id="coordinates"></div>
    </div>
    <script src="script.js"></script>
</body>
</html>

JavaScript (script.js):

const coordinatesElement = document.getElementById('coordinates');
let coords;

const eventSource = new EventSource('http://127.0.0.1:8000/get-waypoints');

eventSource.onopen = () => {
    console.log('EventSource connected');
    coordinatesElement.innerText = '';
}

eventSource.addEventListener('locationUpdate', function (event) {
    coords = JSON.parse(event.data);
    console.log('LocationUpdate', coords);
    updateCoordinates(coords);
});

eventSource.onerror = (error) => {
    console.error('EventSource failed', error);
    eventSource.close();
}

function updateCoordinates(coordinates) {
    const paragraph = document.createElement('p');
    paragraph.textContent = `Latitude: ${coordinates.lat}, Longitude: ${coordinates.lng}`;
    coordinatesElement.appendChild(paragraph);
}

Notes

  • SSE responses must have Content-Type: text/event-stream.
  • Each message ends with \n\n.
  • Use StreamingResponse in FastAPI to send incremental responses.
  • Browsers retry the connection automatically unless explicitly closed.
  • Retry interval can be set using retry: directive from the server.

Sample Response in Postman

Text-based events:

event: locationUpdate
data: {"lat":22.09769,"lng":87.24068}

event: locationUpdate
data: {"lat":22.09776,"lng":87.24075}

Conclusion

SSE is efficient for unidirectional real-time communication. It consumes fewer resources compared to WebSockets and is easy to implement for use cases like logs, notifications, and tracking. But proper unit testing is key.

Resources