Project 20: Real-Time MCP Server with WebSocket Support
Project 20: Real-Time MCP Server with WebSocket Support
Project Overview
| Attribute | Value |
|---|---|
| File | P20-realtime-mcp-websocket.md |
| Main Programming Language | TypeScript |
| Alternative Programming Languages | Python (with asyncio), Go, Rust |
| Coolness Level | Level 4: Hardcore Tech Flex |
| Business Potential | 4. The “Open Core” Infrastructure |
| Difficulty | Level 4: Expert |
| Knowledge Area | MCP / Real-Time Communication / WebSockets |
| Software or Tool | MCP SDK, WebSocket, Server-Sent Events |
| Main Book | “High Performance Browser Networking” by Grigorik |
| Time Estimate | 2-3 weeks |
| Prerequisites | Projects 15-19 completed, WebSocket understanding, async programming |
What You Will Build
An MCP server using HTTP/WebSocket transport for real-time bidirectional communication. Supports: streaming responses, push notifications, live data updates, and long-running operations with progress reporting.
While stdio is great for local use, production deployments often need HTTP/WebSocket for remote access, multiple clients, and real-time updates. This project teaches network-based MCP.
Real World Outcome
You: Process all 10,000 images in the dataset
Claude: [Invokes mcp__remote__process_images via WebSocket]
Starting image processing...
Progress:
[=================> ] 42% (4,200 / 10,000)
Current: processing batch 42
Speed: 350 images/sec
ETA: 16 seconds
[Live updates streaming as they complete]
Processing complete!
- Total: 10,000 images
- Success: 9,847
- Errors: 153 (logged to errors.json)
- Duration: 28.5 seconds
The Core Question You Are Answering
“How do I build MCP servers that support real-time communication, streaming responses, and progress updates for long-running operations?”
Real-world AI workflows involve long-running operations. This project teaches you to build servers that keep users informed with real-time progress, streaming results, and push notifications.
Transport Comparison: stdio vs WebSocket
+------------------------------------------------------------------+
| TRANSPORT COMPARISON |
+------------------------------------------------------------------+
STDIO TRANSPORT
+------------------+ +------------------+
| Claude Code | | MCP Server |
| | | (local process) |
| stdin <--------+--------------------+---- stdout |
| stdout +--------+------------------->+---- stdin |
+------------------+ +------------------+
Pros:
- Simple to implement
- No network configuration
- Process isolation
Cons:
- Local only (same machine)
- One client per process
- No streaming during execution
WEBSOCKET TRANSPORT
+------------------+ Internet +------------------+
| Claude Code | | MCP Server |
| | | (remote) |
| Client <====+=========TCP=============+====> Server |
| | Full-duplex | |
+------------------+ WebSocket +------------------+
Pros:
- Remote access
- Multiple clients
- Bidirectional streaming
- Progress updates
- Push notifications
Cons:
- Network complexity
- Security considerations
- Connection management
WebSocket MCP Protocol Flow
+------------------------------------------------------------------+
| WEBSOCKET MCP FLOW |
+------------------------------------------------------------------+
Claude (Client) MCP Server
| |
| -------- WebSocket Upgrade ------> |
| <------- 101 Switching ---------- |
| |
| Connection established |
| |
| -------- tools/call ------------> |
| { tool: "process_batch", |
| args: { count: 10000 } } |
| |
| | Start processing
| | |
| <-------- progress ------------- | |
| { progress: 0, total: 10000 } | |
| | |
| <-------- progress ------------- | | ...processing
| { progress: 1000, total: 10000 } | |
| | |
| <-------- progress ------------- | | ...processing
| { progress: 5000, total: 10000 } | |
| | |
| <-------- progress ------------- | | ...processing
| { progress: 9000, total: 10000 } | |
| | v
| <-------- result --------------- | Complete
| { success: true, |
| processed: 10000 } |
| |
Message Types for Real-Time Communication
+------------------------------------------------------------------+
| MESSAGE TYPES |
+------------------------------------------------------------------+
1. REQUEST (Client -> Server)
{
"jsonrpc": "2.0",
"id": "req_001",
"method": "tools/call",
"params": {
"name": "process_batch",
"arguments": { "count": 10000 }
}
}
2. PROGRESS (Server -> Client)
{
"jsonrpc": "2.0",
"method": "progress",
"params": {
"requestId": "req_001",
"progress": 4200,
"total": 10000,
"message": "Processing batch 42",
"eta": 16
}
}
3. RESPONSE (Server -> Client)
{
"jsonrpc": "2.0",
"id": "req_001",
"result": {
"content": [{ "type": "text", "text": "..." }]
}
}
4. ERROR (Server -> Client)
{
"jsonrpc": "2.0",
"id": "req_001",
"error": {
"code": -32000,
"message": "Processing failed at item 4201",
"data": { "lastSuccessful": 4200 }
}
}
5. NOTIFICATION (Server -> Client, no response expected)
{
"jsonrpc": "2.0",
"method": "notification",
"params": {
"type": "system",
"message": "Server will restart in 5 minutes"
}
}
6. CANCEL (Client -> Server)
{
"jsonrpc": "2.0",
"method": "cancel",
"params": {
"requestId": "req_001"
}
}
7. HEARTBEAT (Bidirectional)
{
"jsonrpc": "2.0",
"method": "ping"
}
{
"jsonrpc": "2.0",
"method": "pong"
}
Concepts You Must Understand First
Stop and research these before coding:
1. WebSocket Protocol (RFC 6455)
+------------------------------------------------------------------+
| WEBSOCKET HANDSHAKE |
+------------------------------------------------------------------+
Client Request:
GET /mcp HTTP/1.1
Host: server.example.com
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Sec-WebSocket-Version: 13
Server Response:
HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
After handshake, bidirectional binary/text frames flow over TCP.
Key concepts:
- Full-duplex communication
- Message framing (text and binary)
- Ping/pong for keepalive
- Close handshake
2. MCP HTTP Transport
| Aspect | stdio | HTTP/WebSocket |
|---|---|---|
| Connection | Process pipes | TCP socket |
| Lifecycle | Process lifetime | Explicit connect/disconnect |
| Multiple clients | No | Yes |
| Streaming | Limited | Full support |
| Remote access | No | Yes |
Reference: MCP specification HTTP transport section
3. Async Programming Patterns
+------------------------------------------------------------------+
| ASYNC PATTERNS |
+------------------------------------------------------------------+
EVENT LOOP
+---------------------+
| Pending Tasks |
| [T1, T2, T3] |
+----------+----------+
|
v
+---------------------+
| Event Loop |
| while (tasks) { |
| task = next() |
| execute(task) |
| } |
+---------------------+
CONCURRENT PROCESSING (Non-blocking)
async function processBatch(items) {
for (const item of items) {
await processItem(item); // Yields to event loop
sendProgress(current, total);
}
}
PARALLEL PROCESSING (Multiple at once)
async function processAll(items) {
const chunks = chunkArray(items, 10);
await Promise.all(
chunks.map(chunk => processChunk(chunk))
);
}
Reference: “Fluent Python” Ch. 21, “High Performance Browser Networking” Ch. 17
Questions to Guide Your Design
Before implementing, think through these:
1. Streaming Patterns
| Question | Consideration |
|---|---|
| How do you stream partial results? | Chunked responses, progress events |
| How do you report progress? | Percentage, ETA, current item |
| How do you handle cancellation? | Cancel message, cleanup |
| What about backpressure? | Slow client, buffer management |
2. Connection Management
| Scenario | Strategy |
|---|---|
| Client disconnect | Cleanup, cancel in-progress operations |
| Reconnection | Resume or restart operations |
| Multiple clients | Broadcast, or per-client state |
| Server restart | Graceful shutdown, client notification |
3. Error Handling
| Error Type | Response |
|---|---|
| Network failure | Retry with exponential backoff |
| Partial completion | Report progress, offer resume |
| Timeout | Configurable, with cancellation |
| Server error | Structured error response |
Thinking Exercise
Design the Streaming Protocol
Consider a long-running image processing operation:
Request: Process 10,000 images
|
+-- Server receives request
|
+-- Server starts processing
| |
| +-- Every 100 images, send progress update
| | { progress: 100, total: 10000, rate: 350/sec }
| |
| +-- Every batch, yield to event loop
| | await new Promise(r => setImmediate(r))
| |
| +-- Check for cancellation
| if (cancelled) { cleanup(); return; }
|
+-- Processing complete
| { success: true, processed: 10000 }
|
+-- Or: Client sends cancel
{ method: "cancel", params: { requestId: "..." } }
Questions to consider:
- What message types do you need? (request, progress, result, error, cancel)
- How frequently should you send progress updates? (time-based vs item-based)
- What if the client disconnects mid-operation? (cleanup, rollback)
- How do you handle multiple concurrent operations? (request IDs)
The Interview Questions They Will Ask
- “How would you implement real-time progress updates for an AI tool?”
- WebSocket for bidirectional communication, progress events, cancellation support.
- “What is the difference between WebSocket and Server-Sent Events?”
- WebSocket: bidirectional, full-duplex
- SSE: server-to-client only, HTTP-based, simpler
- “How do you handle long-running operations in a service?”
- Async processing, progress reporting, cancellation, timeout handling.
- “What is backpressure and how do you handle it?”
- When producer is faster than consumer; buffering, flow control, or dropping.
- “How do you implement cancellation for async operations?”
- Cancellation tokens, periodic checks, cleanup on cancel.
Hints in Layers
Hint 1: Use a WebSocket Library
Use ws for Node.js or websockets for Python. Do not implement the protocol yourself:
import { WebSocketServer } from "ws";
const wss = new WebSocketServer({ port: 8080 });
wss.on("connection", (ws) => {
ws.on("message", (data) => {
const request = JSON.parse(data.toString());
// Handle request...
});
});
Hint 2: Define Message Types
Create clear TypeScript types for all message types:
type MCPMessage =
| { type: "request"; id: string; method: string; params: any }
| { type: "response"; id: string; result: any }
| { type: "progress"; requestId: string; progress: number; total: number }
| { type: "error"; id: string; error: { code: number; message: string } }
| { type: "cancel"; requestId: string };
Hint 3: Implement Progress Callbacks
For long operations, yield progress at regular intervals or batch completions:
async function processWithProgress(
items: any[],
onProgress: (current: number, total: number) => void
): Promise<void> {
for (let i = 0; i < items.length; i++) {
await processItem(items[i]);
if (i % 100 === 0) {
onProgress(i, items.length);
await yieldToEventLoop(); // Prevent blocking
}
}
onProgress(items.length, items.length);
}
function yieldToEventLoop(): Promise<void> {
return new Promise(resolve => setImmediate(resolve));
}
Hint 4: Add Cancellation Support
Support a cancel message that can abort in-progress operations:
const activeOperations: Map<string, { cancelled: boolean }> = new Map();
function handleCancel(requestId: string): void {
const operation = activeOperations.get(requestId);
if (operation) {
operation.cancelled = true;
}
}
async function longOperation(requestId: string): Promise<void> {
const state = { cancelled: false };
activeOperations.set(requestId, state);
try {
for (const item of items) {
if (state.cancelled) {
throw new Error("Operation cancelled");
}
await processItem(item);
}
} finally {
activeOperations.delete(requestId);
}
}
Books That Will Help
| Topic | Book | Chapter | Why It Helps |
|---|---|---|---|
| WebSockets | “High Performance Browser Networking” by Grigorik | Ch. 17 | Protocol details |
| Async Patterns | “Fluent Python” by Ramalho | Ch. 21 | Coroutines and async |
| Streaming | “Designing Data-Intensive Applications” | Ch. 11 | Stream processing |
| Network Programming | “TCP/IP Illustrated” by Stevens | Ch. 1-4 | TCP fundamentals |
Implementation Skeleton
import { WebSocketServer, WebSocket } from "ws";
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
const PORT = parseInt(process.env.PORT || "8080");
// Active connections and operations
const connections: Map<WebSocket, { id: string }> = new Map();
const activeOperations: Map<string, {
ws: WebSocket;
cancelled: boolean;
}> = new Map();
// Create WebSocket server
const wss = new WebSocketServer({ port: PORT });
wss.on("connection", (ws) => {
const connectionId = crypto.randomUUID();
connections.set(ws, { id: connectionId });
console.log(`Client connected: ${connectionId}`);
ws.on("message", async (data) => {
try {
const message = JSON.parse(data.toString());
await handleMessage(ws, message);
} catch (e) {
sendError(ws, null, -32700, "Parse error");
}
});
ws.on("close", () => {
console.log(`Client disconnected: ${connectionId}`);
connections.delete(ws);
// Cancel any active operations for this client
for (const [reqId, op] of activeOperations) {
if (op.ws === ws) {
op.cancelled = true;
}
}
});
// Send initialization
ws.send(JSON.stringify({
jsonrpc: "2.0",
method: "connected",
params: { connectionId }
}));
});
async function handleMessage(ws: WebSocket, message: any): Promise<void> {
const { id, method, params } = message;
switch (method) {
case "initialize":
ws.send(JSON.stringify({
jsonrpc: "2.0",
id,
result: {
protocolVersion: "2024-11-05",
capabilities: { tools: {}, streaming: true },
serverInfo: { name: "realtime-mcp", version: "1.0.0" }
}
}));
break;
case "tools/list":
ws.send(JSON.stringify({
jsonrpc: "2.0",
id,
result: {
tools: [
{
name: "process_batch",
description: "Process items with progress reporting",
inputSchema: {
type: "object",
properties: {
count: { type: "number", description: "Number of items" }
},
required: ["count"]
}
}
]
}
}));
break;
case "tools/call":
await handleToolCall(ws, id, params.name, params.arguments);
break;
case "cancel":
handleCancel(params.requestId);
ws.send(JSON.stringify({
jsonrpc: "2.0",
id,
result: { cancelled: true }
}));
break;
case "ping":
ws.send(JSON.stringify({ jsonrpc: "2.0", method: "pong" }));
break;
default:
sendError(ws, id, -32601, `Method not found: ${method}`);
}
}
async function handleToolCall(
ws: WebSocket,
requestId: string,
name: string,
args: any
): Promise<void> {
if (name === "process_batch") {
const { count } = args;
// Register operation for cancellation
activeOperations.set(requestId, { ws, cancelled: false });
try {
await processBatchWithProgress(
ws,
requestId,
count,
(progress, total, message) => {
// Send progress update
ws.send(JSON.stringify({
jsonrpc: "2.0",
method: "progress",
params: {
requestId,
progress,
total,
message,
percentage: Math.round((progress / total) * 100)
}
}));
}
);
// Send final result
ws.send(JSON.stringify({
jsonrpc: "2.0",
id: requestId,
result: {
content: [{
type: "text",
text: `Processing complete!\n- Total: ${count}\n- Success: ${count}`
}]
}
}));
} catch (e) {
if (e.message === "Operation cancelled") {
ws.send(JSON.stringify({
jsonrpc: "2.0",
id: requestId,
result: {
content: [{
type: "text",
text: "Operation was cancelled by user."
}]
}
}));
} else {
sendError(ws, requestId, -32000, e.message);
}
} finally {
activeOperations.delete(requestId);
}
} else {
sendError(ws, requestId, -32601, `Unknown tool: ${name}`);
}
}
async function processBatchWithProgress(
ws: WebSocket,
requestId: string,
count: number,
onProgress: (progress: number, total: number, message: string) => void
): Promise<void> {
const batchSize = 100;
const operation = activeOperations.get(requestId);
for (let i = 0; i < count; i += batchSize) {
// Check for cancellation
if (operation?.cancelled) {
throw new Error("Operation cancelled");
}
// Simulate processing
await simulateProcessing(batchSize);
// Report progress
const progress = Math.min(i + batchSize, count);
onProgress(
progress,
count,
`Processing batch ${Math.floor(i / batchSize) + 1}`
);
// Yield to event loop to handle other messages
await new Promise(resolve => setImmediate(resolve));
}
}
async function simulateProcessing(count: number): Promise<void> {
// Simulate work (replace with actual processing)
await new Promise(resolve => setTimeout(resolve, 50));
}
function handleCancel(requestId: string): void {
const operation = activeOperations.get(requestId);
if (operation) {
operation.cancelled = true;
console.log(`Cancelled operation: ${requestId}`);
}
}
function sendError(
ws: WebSocket,
id: string | null,
code: number,
message: string
): void {
ws.send(JSON.stringify({
jsonrpc: "2.0",
id,
error: { code, message }
}));
}
// Heartbeat to detect dead connections
setInterval(() => {
for (const [ws, conn] of connections) {
if (ws.readyState === WebSocket.OPEN) {
ws.send(JSON.stringify({ jsonrpc: "2.0", method: "ping" }));
}
}
}, 30000);
// Graceful shutdown
process.on("SIGTERM", () => {
console.log("Shutting down...");
// Notify all clients
for (const [ws] of connections) {
ws.send(JSON.stringify({
jsonrpc: "2.0",
method: "notification",
params: { type: "shutdown", message: "Server is shutting down" }
}));
ws.close(1001, "Server shutdown");
}
wss.close(() => {
console.log("Server closed");
process.exit(0);
});
});
console.log(`WebSocket MCP server listening on port ${PORT}`);
Learning Milestones
| Milestone | What It Proves | Verification |
|---|---|---|
| WebSocket connection works | You understand the transport | Client connects successfully |
| Progress updates stream | You can report long-running status | Progress bar updates in real-time |
| Cancellation works | Operations can be stopped | Cancel mid-processing stops it |
| Multiple clients work | Server handles concurrency | Two clients process simultaneously |
Core Challenges Mapped to Concepts
| Challenge | Concept | Book Reference |
|---|---|---|
| WebSocket lifecycle | Connection management | RFC 6455 |
| Streaming responses | Chunked transfer | “High Performance Browser Networking” |
| Progress reporting | Long-running operations | “Fluent Python” Ch. 21 |
| Connection resilience | Reconnection logic | “Designing Data-Intensive Applications” |
Extension Ideas
Once the basic server works, consider these enhancements:
- Add TLS/SSL support for secure WebSocket (wss://)
- Implement reconnection with operation resume
- Add broadcast notifications for multi-client updates
- Implement request queuing for fair resource sharing
- Add compression for large payloads
Common Pitfalls
- Not handling disconnects - Clean up operations on client disconnect
- Blocking the event loop - Use async operations and yield periodically
- Missing heartbeats - Detect dead connections with ping/pong
- No cancellation support - Long operations should be cancellable
- Ignoring backpressure - Monitor client’s ability to consume messages
Success Criteria
You have completed this project when:
- Your MCP server accepts WebSocket connections
- Tools can be invoked over WebSocket
- Long-running operations send progress updates
- Clients can cancel in-progress operations
- Multiple clients can connect simultaneously
- Client disconnects are handled gracefully
- The server performs graceful shutdown
- Heartbeats detect dead connections