Real-Time Data Streaming via MCP SSE and WebSockets

sequenceDiagram
    autonumber
    participant Host as AI Host (Cursor / Claude Code)
    participant Server as Remote MCP Server (SSE Endpoint)
    participant Queue as Redis Task Queue (BullMQ)
    participant Worker as Background Worker Daemon

    Host->>Server: POST /messages [tools/call: build_and_deploy_service]
    Note over Host,Server: Includes meta.progressToken: "job-991"
    Server->>Queue: Enqueue Build Job
    Server-->>Host: 202 Accepted / Initial Response
    
    Queue->>Worker: Dequeue & Execute Build
    loop During Execution (Every 2 seconds)
        Worker->>Queue: Publish Progress Event
        Queue->>Server: Progress Callback
        Server->>Host: SSE Event: notifications/progress (35% - Compiling)
        Server->>Host: SSE Event: notifications/progress (78% - Packaging)
    end
    
    Worker->>Queue: Job Complete (Artifact URL)
    Queue->>Server: Result Ready
    Server->>Host: SSE Event: tools/call Result [Completed]

{
  "jsonrpc": "2.0",
  "id": "build-req-102",
  "method": "tools/call",
  "params": {
    "name": "execute_cloud_build",
    "arguments": {
      "service_name": "billing-api",
      "environment": "staging"
    },
    "_meta": {
      "progressToken": "token-xyz-8812"
    }
  }
}

event: message
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"token-xyz-8812","progress":25,"total":100,"message":"Docker image build: Layer 3/8 compiled"}}

event: message
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"token-xyz-8812","progress":70,"total":100,"message":"Running database migration dry-run"}}

event: message
data: {"jsonrpc":"2.0","id":"build-req-102","result":{"content":[{"type":"text","text":"Build and deployment successful: Version 2.14.0 deployed to staging.\nArtifact: registry.enterprise.internal/billing-api:2.14.0"}]}}

import Fastify, { FastifyRequest, FastifyReply } from 'fastify';
import { Queue, Worker, QueueEvents } from 'bullmq';
import Redis from 'ioredis';

const app = Fastify({ logger: true });
const redisConnection = new Redis(process.env.REDIS_URL || 'redis://');

// 1. Task Queue Definition
const taskQueue = new Queue('mcp-long-tasks', { connection: redisConnection });
const queueEvents = new QueueEvents('mcp-long-tasks', { connection: redisConnection });

// Track active client SSE response streams by sessionId
const activeStreams = new Map<string, FastifyReply>();

// 2. Persistent SSE Endpoint
app.get('/sse', async (request: FastifyRequest, reply: FastifyReply) => {
  reply.raw.setHeader('Content-Type', 'text/event-stream');
  reply.raw.setHeader('Cache-Control', 'no-cache');
  reply.raw.setHeader('Connection', 'keep-alive');
  reply.raw.flushHeaders();

  const sessionId = 'sess_' + Math.random().toString(36).substring(2, 9);
  activeStreams.set(sessionId, reply);

  // Send initial endpoint registration event
  reply.raw.write(`event: endpoint\ndata: /messages?sessionId=${sessionId}\n\n`);

  request.raw.on('close', () => {
    activeStreams.delete(sessionId);
    console.log(`SSE session disconnected: ${sessionId}`);
  });
});

// 3. Asynchronous Tool Dispatcher Route
app.post('/messages', async (request: FastifyRequest<{ Querystring: { sessionId: string } }>, reply: FastifyReply) => {
  const { sessionId } = request.query;
  const body = request.body as any;

  if (body.method !== 'tools/call') {
    return reply.send({ jsonrpc: '2.0', id: body.id, error: { code: -32601, message: 'Method not supported' } });
  }

  const sseReply = activeStreams.get(sessionId);
  if (!sseReply) {
    return reply.status(400).send({ error: 'SSE session stream not active' });
  }

  const progressToken = body.params?._meta?.progressToken;

  // Enqueue long-running task to background worker
  const job = await taskQueue.add('execute_long_task', {
    toolName: body.params.name,
    args: body.params.arguments,
    requestId: body.id,
    progressToken,
    sessionId
  });

  // Acknowledge receipt immediately to HTTP caller
  reply.status(202).send({ status: 'queued', jobId: job.id });
});

// 4. Listen for Background Worker Progress & Completion Events
queueEvents.on('progress', ({ jobId, data }: { jobId: string; data: any }) => {
  const sseReply = activeStreams.get(data.sessionId);
  if (sseReply && data.progressToken) {
    const notification = {
      jsonrpc: '2.0',
      method: 'notifications/progress',
      params: {
        progressToken: data.progressToken,
        progress: data.progress,
        total: 100,
        message: data.message
      }
    };
    sseReply.raw.write(`event: message\ndata: ${JSON.stringify(notification)}\n\n`);
  }
});

queueEvents.on('completed', ({ jobId, returnvalue }: { jobId: string; returnvalue: any }) => {
  const sseReply = activeStreams.get(returnvalue.sessionId);
  if (sseReply) {
    const responsePayload = {
      jsonrpc: '2.0',
      id: returnvalue.requestId,
      result: {
        content: [{ type: 'text', text: returnvalue.resultText }]
      }
    };
    sseReply.raw.write(`event: message\ndata: ${JSON.stringify(responsePayload)}\n\n`);
  }
});

// 5. Worker Implementation (Runs in isolated process or thread pool)
const worker = new Worker('mcp-long-tasks', async (job) => {
  const { toolName, args, progressToken, sessionId } = job.data;

  // Step 1: 25% Progress
  await new Promise((r) => setTimeout(r, 2000));
  await job.updateProgress({ progress: 25, message: 'Validating dependencies', progressToken, sessionId });

  // Step 2: 75% Progress
  await new Promise((r) => setTimeout(r, 4000));
  await job.updateProgress({ progress: 75, message: 'Compiling project artifacts', progressToken, sessionId });

  // Step 3: Complete
  await new Promise((r) => setTimeout(r, 2000));
  return {
    requestId: job.data.requestId,
    sessionId,
    resultText: `Task '${toolName}' completed successfully in 8.0s.`
  };
}, { connection: redisConnection });

app.listen({ port: 8080, host: '0.0.0.0' }, () => {
  console.log('Streaming MCP Server running on port 8080');
});

// SSE Heartbeat Timer
setInterval(() => {
  for (const [sessionId, reply] of activeStreams.entries()) {
    // SSE comments start with a colon ':' and are ignored by JSON-RPC parsers
    reply.raw.write(': ping keep-alive\n\n');
  }
}, 15000);