SSE Response Streaming
Implement Server-Sent Events to stream JSON-RPC responses to clients.
Streaming Responses with SSE
Instead of sending a single JSON response, we can stream multiple events using Server-Sent Events. This is useful for long-running operations, progress updates, or sending multiple responses.
SSE Writer Class
Let's create a helper class to properly format and send SSE events:
"""
sse_writer.py - Server-Sent Events writer for HTTP responses.
"""
import json
class SSEWriter:
"""Write Server-Sent Events to an HTTP response."""
def __init__(self, wfile):
"""
Initialize with a writable file object.
Args:
wfile: The response write file (self.wfile from handler)
"""
self.wfile = wfile
self.event_id = 0
def send_event(self, data, event_type="message", event_id=None):
"""
Send a single SSE event.
Args:
data: The event data (will be JSON-encoded if dict)
event_type: The event type (default: "message")
event_id: Optional event ID (auto-incremented if None)
"""
# Auto-increment event ID if not provided
if event_id is None:
self.event_id += 1
event_id = self.event_id
# JSON-encode dict data
if isinstance(data, dict):
data = json.dumps(data)
# Build the event
lines = []
lines.append(f"event: {event_type}")
lines.append(f"id: {event_id}")
lines.append(f"data: {data}")
# Join with newlines and add blank line
event = "\n".join(lines) + "\n\n"
# Write and flush
self.wfile.write(event.encode("utf-8"))
self.wfile.flush()
def send_json_rpc(self, response):
"""
Send a JSON-RPC response as an SSE event.
Args:
response: A JSON-RPC response dict
"""
self.send_event(response, event_type="message")
def send_comment(self, text):
"""
Send an SSE comment (for keep-alive).
Args:
text: Comment text
"""
comment = f": {text}\n\n"
self.wfile.write(comment.encode("utf-8"))
self.wfile.flush()Using SSE in the Handler
Now integrate the SSE writer into your HTTP handler:
class MCPHTTPHandler(BaseHTTPRequestHandler):
"""Handle MCP requests with SSE support."""
def send_sse_headers(self):
"""Send headers for SSE response."""
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.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
def handle_sse_response(self, request):
"""Handle a request with SSE streaming response."""
# Send SSE headers
self.send_sse_headers()
# Create SSE writer
sse = SSEWriter(self.wfile)
# Process the request
method = request.get("method")
request_id = request.get("id")
params = request.get("params", {})
# Route and handle the request
try:
if method == "initialize":
response = self.handle_initialize(request_id, params)
elif method == "tools/list":
response = self.handle_tools_list(request_id)
elif method == "tools/call":
# Tools might send progress events
response = self.handle_tools_call(request_id, params, sse)
else:
response = {
"jsonrpc": "2.0",
"id": request_id,
"error": {
"code": -32601,
"message": f"Method not found: {method}"
}
}
# Send the final response
sse.send_json_rpc(response)
except Exception as e:
# Send error response
sse.send_json_rpc({
"jsonrpc": "2.0",
"id": request_id,
"error": {
"code": -32603,
"message": f"Internal error: {e}"
}
})Progress Notifications
For long-running tool calls, you can send progress notifications before the final result:
def handle_tools_call(self, request_id, params, sse):
"""Handle tools/call with progress streaming."""
tool_name = params.get("name", "")
arguments = params.get("arguments", {})
# For demonstration: a slow tool that reports progress
if tool_name == "slow_search":
query = arguments.get("query", "")
# Send progress notifications
for i in range(1, 4):
sse.send_json_rpc({
"jsonrpc": "2.0",
"method": "notifications/progress",
"params": {
"progressToken": request_id,
"progress": i * 25,
"total": 100,
"message": f"Searching... {i}/4 complete"
}
})
import time
time.sleep(0.5) # Simulate work
# Return final result
return {
"jsonrpc": "2.0",
"id": request_id,
"result": {
"content": [
{
"type": "text",
"text": f"Found 42 results for '{query}'"
}
]
}
}
# Unknown tool
return {
"jsonrpc": "2.0",
"id": request_id,
"error": {
"code": -32602,
"message": f"Unknown tool: {tool_name}"
}
}SSE Output Example
When streaming a slow tool call, the SSE output looks like this:
event: message
id: 1
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":1,"progress":25,"total":100,"message":"Searching... 1/4 complete"}}
event: message
id: 2
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":1,"progress":50,"total":100,"message":"Searching... 2/4 complete"}}
event: message
id: 3
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":1,"progress":75,"total":100,"message":"Searching... 3/4 complete"}}
event: message
id: 4
data: {"jsonrpc":"2.0","id":1,"result":{"content":[{"type":"text","text":"Found 42 results for 'example'"}]}}
For long-lived SSE connections, send periodic comments (lines starting
with :) to prevent timeouts. Comments are ignored by clients
but keep the connection alive.
Testing SSE Responses
Test SSE streaming with curl:
# Request with Accept header for SSE
curl -X POST http://localhost:8080/mcp \
-H "Content-Type: application/json" \
-H "Accept: text/event-stream" \
-d '{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": "slow_search",
"arguments": {"query": "example"}
}
}'
# Output shows streamed events in real-time:
# event: message
# id: 1
# data: {"jsonrpc":"2.0","method":"notifications/progress",...}
# ...Now let's implement session management to track client connections and maintain state between requests.