a developer, I’m all the time in search of methods to make my functions extra dynamic and interactive. Customers at this time anticipate real-time options, corresponding to reside notifications, streaming updates, and dashboards that refresh routinely. The instrument that usually involves thoughts for net builders when contemplating these kind of functions is WebSockets, and it’s extremely highly effective.
There are occasions, although, when WebSockets will be overkill, and their full performance is usually not required. They supply a fancy, bi-directional communication channel, however many occasions, all I want is for the server to push updates to the consumer. For these frequent situations, a extra easy and stylish resolution that’s constructed proper into trendy net platforms is named Server-Despatched Occasions (SSE).
On this article, I’m going to introduce you to Server-Despatched Occasions. We’ll focus on what they’re, how they examine to WebSockets, and why they’re usually the right instrument for the job. Then, we’ll dive right into a sequence of sensible examples, utilizing Python and the FastAPI framework to construct real-time functions which are surprisingly easy but highly effective.
What are Server-Despatched Occasions (SSE)?
Server-Despatched Occasions is an internet expertise commonplace that permits a server to push knowledge to a consumer asynchronously as soon as an preliminary consumer connection has been established. It offers a one-way, server-to-client stream of knowledge over a single, long-lived HTTP connection. The consumer, usually an internet browser, subscribes to this stream and might react to the messages it receives.
Some key points of Server-Despatched Occasions embrace:
- Easy Protocol. SSE is a simple, text-based protocol. Occasions are simply chunks of textual content despatched over HTTP, making them simple to debug with commonplace instruments like curl.
- Commonplace HTTP. SSE works over common HTTP/HTTPS. This implies it’s usually extra appropriate with present firewalls and proxy servers.
- Computerized Reconnection. It is a killer function. If the connection to the server is misplaced, the browser’s EventSource API will routinely attempt to reconnect. You get this resilience without spending a dime, with out writing any further JavaScript code.
- One-Method Communication. SSE is strictly for server-to-client knowledge pushes. In the event you want full-duplex, client-to-server communication, WebSockets are the extra acceptable selection.
- Native Browser Assist. All trendy net browsers have built-in assist for Server-Despatched Occasions (SSE) via the EventSource interface, eliminating the necessity for client-side libraries.
Why SSE Issues/Widespread Use Instances
The first benefit of SSE is its simplicity. For a big class of real-time issues, it offers all the required performance with a fraction of the complexity of WebSockets, each on the server and the consumer. This implies sooner growth, simpler upkeep, and fewer issues that may go flawed.
SSE is an ideal match for any situation the place the server must provoke communication and ship updates to the consumer. For instance …
- Dwell Notification Techniques. Pushing notifications to a person when a brand new message arrives or an essential occasion happens.
- Actual-Time Exercise Feeds. Streaming updates to a person’s exercise feed, much like a Twitter or Fb timeline.
- Dwell Knowledge Dashboards. Sending steady updates for inventory tickers, sports activities scores, or monitoring metrics to a reside dashboard.
- Streaming Log Outputs. Displaying the reside log output from a long-running background course of immediately within the person’s browser.
- Progress Updates. Displaying the real-time progress of a file add, an information processing job, or some other long-running process initiated by the person.
That’s sufficient idea; let’s see simply how simple it’s to implement these concepts with Python.
Organising the Growth Atmosphere
We’ll utilise FastAPI, a contemporary and high-performance Python net framework. Its native assist for asyncio and streaming responses makes it an ideal match for implementing Server-Despatched Occasions. You’ll additionally want the Uvicorn ASGI server to run the appliance.
As ordinary, we’ll arrange a growth setting to maintain our initiatives separate. I counsel utilizing MiniConda for this, however be happy to make use of whichever instrument you’re accustomed to.
# Create and activate a brand new digital setting
(base) $ conda create -n sse-env python=3.13 -y
(base) $ activate sse-env
Now, set up the exterior libraries we’d like.
# Set up FastAPI and Uvicorn
(sse-env) $ pip set up fastapi uvicorn
That’s all of the setup we’d like. Now, we will begin coding.
Code Instance 1 — The Python Backend. A Easy SSE Endpoint
Let’s create our first SSE endpoint. It is going to ship a message with the present time to the consumer each second.
Create a file named app.py and kind the next into it.
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
import time
app = FastAPI()
# Enable requests from http://localhost:8080 (the place index.html is served)
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:8080"],
allow_methods=["GET"],
allow_headers=["*"],
)
def event_stream():
whereas True:
yield f"knowledge: The time is {time.strftime('%X')}nn"
time.sleep(1)
@app.get("/stream-time")
def stream():
return StreamingResponse(event_stream(), media_type="textual content/event-stream")
I hope you agree that this code is simple.
- We outline an event_stream() operate. This loop repeats endlessly, producing a string each second.
- The yielded string is formatted in accordance with the SSE spec: it should begin with knowledge: and finish with two newlines (nn).
- Our endpoint /stream-time returns a StreamingResponse, passing our generator to it and setting the media_type to textual content/event-stream. FastAPI handles the remainder, retaining the connection open and sending every yielded chunk to the consumer.
To run the code, don’t use the usual Python app.py command as you’d usually. As a substitute, do that.
(sse-env)$ uvicorn app:app --reload
INFO: Will look ahead to modifications in these directories: ['/home/tom']
INFO: Uvicorn operating on http://127.0.0.1:8000 (Press CTRL+C to give up)
INFO: Began reloader course of [4109269] utilizing WatchFiles
INFO: Began server course of [4109271]
INFO: Ready for utility startup.
INFO: Utility startup full.
Now, kind this handle into your browser …
http://127.0.0.1:8000/stream-time
… and it is best to see one thing like this.
The display screen ought to show an up to date time document each second.
Code instance 2. Actual-Time System Monitoring Dashboard
On this instance, we’ll monitor our PC or laptop computer’s CPU and reminiscence utilization in actual time.
Right here is the app.py code you want.
import asyncio
import json
import psutil
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
import datetime
# Outline app FIRST
app = FastAPI()
# Then add middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:8080"],
allow_methods=["GET"],
allow_headers=["*"],
)
async def system_stats_generator(request: Request):
whereas True:
if await request.is_disconnected():
print("Shopper disconnected.")
break
cpu_usage = psutil.cpu_percent()
memory_info = psutil.virtual_memory()
stats = {
"cpu_percent": cpu_usage,
"memory_percent": memory_info.%,
"memory_used_mb": spherical(memory_info.used / (1024 * 1024), 2),
"memory_total_mb": spherical(memory_info.complete / (1024 * 1024), 2)
}
yield f"knowledge: {json.dumps(stats)}nn"
await asyncio.sleep(1)
@app.get("/system-stats")
async def stream_system_stats(request: Request):
return StreamingResponse(system_stats_generator(request), media_type="textual content/event-stream")
@app.get("/", response_class=HTMLResponse)
async def read_root():
with open("index.html") as f:
return HTMLResponse(content material=f.learn())
This code constructs a real-time system monitoring service utilizing the FastAPI net framework. It creates an internet server that repeatedly tracks and broadcasts the host machine’s CPU and reminiscence utilization to any related net consumer.
First, it initialises a FastAPI utility and configures Cross-Origin Useful resource Sharing (CORS) middleware. This middleware is a safety function that’s explicitly configured right here to permit an internet web page served from http://localhost:8080 to make requests to this server, which is a standard requirement when the frontend and backend are developed individually.
The core of the appliance is the system_stats_generator asynchronous operate. This operate runs in an infinite loop, and in every iteration, it makes use of the psutil library to fetch the present CPU utilisation share and detailed reminiscence statistics, together with the share used, megabytes used, and complete megabytes. It packages this info right into a dictionary, converts it to a JSON string, after which yields it within the particular “textual content/event-stream” format (knowledge: …nn).
The usage of asyncio.sleep(1) introduces a one-second pause between updates, stopping the loop from consuming extreme assets. The operate can also be designed to detect when a consumer has disconnected and gracefully cease sending knowledge for that consumer.
The script defines two net endpoints. The @app.get(“/system-stats”) endpoint creates a StreamingResponse that initiates the system_stats_generator. When a consumer makes a GET request to this URL, it establishes a persistent connection, and the server begins streaming the system stats each second. The second endpoint, @app.get(“/”), serves a static HTML file named index.html as the principle web page. This HTML file would usually comprise the JavaScript code wanted to connect with the /system-stats stream and dynamically show the incoming efficiency knowledge on the net web page.
Now, right here is the up to date (index.html) front-end code.
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>System Monitor</title>
<type>
physique { font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, "Helvetica Neue", Arial, sans-serif; background-color: #f0f2f5; shade: #333; show: flex; justify-content: heart; align-items: heart; peak: 100vh; margin: 0; }
.dashboard { background-color: white; padding: 2rem; border-radius: 8px; box-shadow: 0 4px 12px rgba(0,0,0,0.1); width: 400px; text-align: heart; }
h1 { margin-top: 0; }
.metric { margin-bottom: 1.5rem; }
.metric-label { font-weight: daring; font-size: 1.2rem; margin-bottom: 0.5rem; }
.progress-bar { width: 100%; background-color: #e9ecef; border-radius: 4px; overflow: hidden; }
.progress-bar-fill { peak: 20px; background-color: #007bff; width: 0%; transition: width 0.5s ease-in-out; }
.metric-value { margin-top: 0.5rem; font-size: 1rem; shade: #555; }
</type>
</head>
<physique>
<div class="dashboard">
<h1>Actual-Time Server Monitor</h1>
<div class="metric">
<div class="metric-label">CPU Utilization</div>
<div class="progress-bar">
<div id="cpu-progress" class="progress-bar-fill"></div>
</div>
<div id="cpu-value" class="metric-value">0%</div>
</div>
<div class="metric">
<div class="metric-label">Reminiscence Utilization</div>
<div class="progress-bar">
<div id="mem-progress" class="progress-bar-fill" type="background-color: #28a745;"></div>
</div>
<div id="mem-value" class="metric-value">0% (0 / 0 MB)</div>
</div>
</div>
<script>
const cpuProgress = doc.getElementById('cpu-progress');
const cpuValue = doc.getElementById('cpu-value');
const memProgress = doc.getElementById('mem-progress');
const memValue = doc.getElementById('mem-value');
const eventSource = new EventSource('http://localhost:8000/system-stats');
eventSource.onmessage = operate(occasion) {
// Parse the JSON knowledge from the server
const stats = JSON.parse(occasion.knowledge);
// Replace CPU parts
cpuProgress.type.width = stats.cpu_percent + '%';
cpuValue.textContent = stats.cpu_percent.toFixed(2) + '%';
// Replace Reminiscence parts
memProgress.type.width = stats.memory_percent + '%';
memValue.textContent = `${stats.memory_percent.toFixed(2)}% (${stats.memory_used_mb} / ${stats.memory_total_mb} MB)`;
};
eventSource.onerror = operate(err) {
console.error("EventSource failed:", err);
cpuValue.textContent = "Connection Error";
memValue.textContent = "Connection Error";
};
</script>
</physique>
</html>
Run the app utilizing Uvicorn, as we did in Instance 1. Then, in a separate command window, kind the next to begin a Python server.
python3 -m http.server 8080
Now, open the URL http://localhost:8080/index.html in your browser, and you will note the output, which ought to replace repeatedly.

Code instance 3 — Background process progress bar
On this instance, we provoke a process and show a bar indicating the duty’s progress.
Up to date app.py
import asyncio
import json
import psutil
from fastapi import FastAPI, Request
from fastapi.responses import HTMLResponse, StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
import datetime
# Outline app FIRST
app = FastAPI()
# Then add middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:8080"],
allow_methods=["GET"],
allow_headers=["*"],
)
async def training_progress_generator(request: Request):
"""
Simulates a long-running AI coaching process and streams progress.
"""
total_epochs = 10
steps_per_epoch = 100
for epoch in vary(1, total_epochs + 1):
# Simulate some preliminary processing for the epoch
await asyncio.sleep(0.5)
for step in vary(1, steps_per_epoch + 1):
# Examine if consumer has disconnected
if await request.is_disconnected():
print("Shopper disconnected, stopping coaching process.")
return
# Simulate work
await asyncio.sleep(0.02)
progress = (step / steps_per_epoch) * 100
simulated_loss = (1 / epoch) * (1 - (step / steps_per_epoch)) + 0.1
progress_data = {
"epoch": epoch,
"total_epochs": total_epochs,
"progress_percent": spherical(progress, 2),
"loss": spherical(simulated_loss, 4)
}
# Ship a named occasion "progress"
yield f"occasion: progressndata: {json.dumps(progress_data)}nn"
# Ship a closing "full" occasion
yield f"occasion: completendata: Coaching full!nn"
@app.get("/stream-training")
async def stream_training(request: Request):
"""SSE endpoint to stream coaching progress."""
return StreamingResponse(training_progress_generator(request), media_type="textual content/event-stream")
@app.get("/", response_class=HTMLResponse)
async def read_root():
"""Serves the principle HTML web page."""
with open("index.html") as f:
return HTMLResponse(content material=f.learn())
The up to date index.html code is that this.
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Dwell Process Progress</title>
<type>
physique { font-family: sans-serif; text-align: heart; padding-top: 50px; }
.progress-container { width: 80%; max-width: 700px; margin: auto; }
#start-btn { font-size: 1.2rem; padding: 10px 20px; cursor: pointer; }
.progress-bar-outer { border: 1px stable #ccc; padding: 3px; border-radius: 5px; margin-top: 20px; }
.progress-bar-inner { background-color: #4CAF50; width: 0%; peak: 30px; text-align: heart; line-height: 30px; shade: white; border-radius: 3px; transition: width 0.1s linear; }
#status-text { margin-top: 10px; font-size: 1rem; shade: #555; peak: 2em; }
</type>
</head>
<physique>
<h1>AI Mannequin Coaching Simulation</h1>
<div class="progress-container">
<button id="start-btn">Begin Coaching</button>
<div id="progress-bar-outer" class="progress-bar-outer" type="show: none;">
<div id="progress-bar-inner" class="progress-bar-inner">0%</div>
</div>
<div id="status-text"></div>
</div>
<script>
const startBtn = doc.getElementById('start-btn');
const progressBarOuter = doc.getElementById('progress-bar-outer');
const progressBarInner = doc.getElementById('progress-bar-inner');
const statusText = doc.getElementById('status-text');
let eventSource;
startBtn.addEventListener('click on', () => {
startBtn.disabled = true;
progressBarOuter.type.show = 'block';
statusText.textContent = 'Initializing...';
// Shut any present connection
if (eventSource) {
eventSource.shut();
}
// Begin a brand new SSE connection
eventSource = new EventSource('http://localhost:8000/stream-training');
eventSource.addEventListener('progress', (e) => {
const knowledge = JSON.parse(e.knowledge);
const % = knowledge.progress_percent;
progressBarInner.type.width = % + '%';
progressBarInner.textContent = %.toFixed(0) + '%';
statusText.textContent = `Epoch: ${knowledge.epoch}/${knowledge.total_epochs} | Loss: ${knowledge.loss}`;
});
eventSource.addEventListener('full', (e) => {
statusText.textContent = e.knowledge;
progressBarInner.type.backgroundColor = '#007bff';
eventSource.shut(); // Shut the connection
startBtn.disabled = false;
});
eventSource.onerror = () => {
statusText.textContent = 'Connection error. Please attempt once more.';
eventSource.shut();
startBtn.disabled = false;
};
});
</script>
</physique>
</html>
Cease your present uvicorn and Python server processes in the event that they’re nonetheless operating, after which restart each.
Now, whenever you open the index.html web page, it is best to see a display screen with a button. Urgent the button will begin a dummy process, and a transferring bar will show the duty progress.

Code Instance 4— A Actual-Time Monetary Inventory Ticker
For our closing instance, we’ll create a simulated inventory ticker. The server will generate random worth updates for a number of inventory symbols and ship them utilizing named occasions, the place the occasion identify corresponds to the inventory image (e.g., occasion: AAPL, occasion: GOOGL). It is a highly effective sample for multiplexing completely different sorts of knowledge over a single SSE connection, permitting the consumer to deal with every stream independently.
Right here is the up to date app.py code you’ll want.
import asyncio
import json
import random
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
# Step 1: Create app first
app = FastAPI()
# Step 2: Add CORS to permit requests from http://localhost:8080
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:8080"],
allow_methods=["GET"],
allow_headers=["*"],
)
# Step 3: Simulated inventory costs
STOCKS = {
"AAPL": 150.00,
"GOOGL": 2800.00,
"MSFT": 300.00,
}
# Step 4: Generator to simulate updates
async def stock_ticker_generator(request: Request):
whereas True:
if await request.is_disconnected():
break
image = random.selection(listing(STOCKS.keys()))
change = random.uniform(-0.5, 0.5)
STOCKS[symbol] = max(0, STOCKS[symbol] + change)
replace = {
"image": image,
"worth": spherical(STOCKS[symbol], 2),
"change": spherical(change, 2)
}
# Ship named occasions so the browser can hear by image
yield f"occasion: {image}ndata: {json.dumps(replace)}nn"
await asyncio.sleep(random.uniform(0.5, 1.5))
# Step 5: SSE endpoint
@app.get("/stream-stocks")
async def stream_stocks(request: Request):
return StreamingResponse(stock_ticker_generator(request), media_type="textual content/event-stream")
And the up to date index.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Dwell Inventory Ticker</title>
<type>
physique { font-family: sans-serif; show: flex; justify-content: heart; padding-top: 50px; }
.ticker { show: flex; hole: 20px; }
.inventory { border: 1px stable #ccc; padding: 15px; border-radius: 5px; width: 150px; text-align: heart; }
.image { font-weight: daring; font-size: 1.5rem; }
.worth { font-size: 2rem; margin: 10px 0; }
.change { font-size: 1rem; }
.up { shade: inexperienced; }
.down { shade: crimson; }
</type>
</head>
<physique>
<div class="ticker">
<div id="AAPL" class="inventory">
<div class="image">AAPL</div>
<div class="worth">--.--</div>
<div class="change">-.--</div>
</div>
<div id="GOOGL" class="inventory">
<div class="image">GOOGL</div>
<div class="worth">--.--</div>
<div class="change">-.--</div>
</div>
<div id="MSFT" class="inventory">
<div class="image">MSFT</div>
<div class="worth">--.--</div>
<div class="change">-.--</div>
</div>
</div>
<script>
const eventSource = new EventSource('http://localhost:8000/stream-stocks');
operate updateStock(knowledge) {
const inventory = doc.getElementById(knowledge.image);
if (!inventory) return;
const priceEl = inventory.querySelector('.worth');
const changeEl = inventory.querySelector('.change');
priceEl.textContent = knowledge.worth.toFixed(2);
changeEl.textContent = knowledge.change.toFixed(2);
const className = knowledge.change >= 0 ? 'up' : 'down';
priceEl.className = 'worth ' + className;
changeEl.className = 'change ' + className;
}
['AAPL', 'GOOGL', 'MSFT'].forEach(image => {
eventSource.addEventListener(image, e => {
const stockData = JSON.parse(e.knowledge);
updateStock(stockData);
});
});
eventSource.onerror = operate(err) {
console.error("EventSource failed:", err);
};
</script>
</physique>
</html>
Cease then restart the uvicorn and Python processes as earlier than. This time, whenever you open http://localhost:8080/index.html in your browser, it is best to see a display screen like this, which is able to regularly replace the dummy costs of the three shares.

Abstract
On this article, I demonstrated that for a lot of real-time use instances, Server-Despatched Occasions supply an easier various to WebSockets. We mentioned the core rules of SSE, together with its one-way communication mannequin and computerized reconnection capabilities. By way of a sequence of hands-on examples utilizing Python and FastAPI, we noticed simply how simple it’s to construct highly effective real-time options. We coated:
- A easy Python back-end and SSE endpoint
- A reside system monitoring dashboard streaming structured JSON knowledge.
- An actual-time progress bar for a simulated long-running background process.
- A multiplexed inventory ticker utilizing named occasions to handle completely different knowledge streams.
Subsequent time you should push knowledge out of your server to a consumer, I encourage you to pause earlier than reaching for WebSockets. Ask your self for those who actually want bi-directional communication. If the reply is not any, then Server-Despatched Occasions are seemingly the extra easy, sooner, and extra strong resolution you’ve been in search of.