How to mock Server-Sent Events (SSE) in Python
A minimal HTTP server that streams Server-Sent Events to clients, perfect for testing and development.
Python code
44 linesfrom http.server import HTTPServer, BaseHTTPRequestHandler
import threading
import time
MESSAGES = iter([
"data: Hello world\n\n",
"data: Second message\n\n",
"event: custom\n",
"data: Custom event payload\n\n",
"data: Final message\n\n"
])
class SSEHandler(BaseHTTPRequestHandler):
def do_GET(self):
if self.path != "/events":
self.send_error(404)
return
self.send_response(200)
self.send_header("Content-Type", "text/event-stream")
self.send_header("Cache-Control", "no-cache")
self.send_header("Connection", "keep-alive")
self.end_headers()
try:
for msg in MESSAGES:
self.wfile.write(msg.encode())
self.wfile.flush()
time.sleep(1)
except BrokenPipeError:
pass
def log_message(self, format, *args):
pass
if __name__ == "__main__":
server = HTTPServer(("localhost", 8000), SSEHandler)
print("SSE mock running at http://localhost:8000/events")
print("Use curl: curl -N http://localhost:8000/events")
# Auto-stop after serving all messages
timer = threading.Timer(6, server.shutdown)
timer.start()
server.serve_forever()
Output
SSE mock running at http://localhost:8000/events
Use curl: curl -N http://localhost:8000/events
How it works
The handler sets Content-Type: text/event-stream and Cache-Control: no-cache to signal a streaming response. Each message is written with utf-8 encoding and flushed immediately to simulate real-time events. The MESSAGES iterator yields SSE-formatted strings, including named events and data frames. A timer auto-shuts down the server after all messages are sent to avoid hanging. log_message is suppressed to keep the console clean during development.
Common mistakes
- Forgetting the trailing blank line after each event, which is required by the SSE spec.
- Not flushing the write buffer, causing messages to be delayed or batched on the client.
- Using `print` instead of `self.wfile.write`, which doesn't send data over the HTTP connection.
Variations
- Use Flask with `Response(stream_with_context(...))` for a more framework-friendly mock.
- Make the server loop forever by removing the `Timer` shutdown, useful for interactive testing.
Real-world use cases
- Developing and testing a chat or notification UI without a real backend.
- Simulating stock price or sports scores stream to validate client-side event handling.
- Demoing an AI assistant's token-by-token response streaming to a web frontend.
Sponsored
More from API design & gRPC
- Build a Bulk Array POST Mock Server in Python medium
- Build a Mock REST API with PUT and GET in Python medium
- Convert Protobuf to JSON and Dict in Python easy
- Create a Data Helper in Python for gRPC-style APIs easy
- Format data in Python using dataclasses like gRPC messages easy
- Generate an OpenAPI Spec from Mock Routes in Python easy
Keep learning
Related tutorials and quizzes for this topic.