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.

Medium Python 3.9+ Aug 9, 2026 API design & gRPC 14 views 0 copies

Python code

44 lines
Python 3.9+
from 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

stdout
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

  1. Use Flask with `Response(stream_with_context(...))` for a more framework-friendly mock.
  2. 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

Run this sample

Open the browser IDE to tweak the example and see results without installing anything.

Open editor

More from API design & gRPC

Related tutorials and quizzes for this topic.