Real-Time Data Streaming via MCP SSE

Real-Time Data Streaming via MCP SSE

In traditional synchronous API communication, a client issues a request and blocks while the server calculates and returns the complete result payload. In autonomous AI agent workflows, this synchronous paradigm rapidly collapses. Enterprise agent operations—such as compiling large multi-service container images, executing complex SQL analytical aggregations across billions of rows, or running end-to-end regression test suites—frequently require between thirty seconds and several minutes to complete.

When an MCP server blocks its main event loop or leaves an HTTP socket hanging without ongoing feedback, AI hosts (such as Claude Desktop, Cursor, or the OpenAI Codex CLI) encounter internal transport timeouts (typically bounded between thirty and sixty seconds). The host drops the connection, terminates the agent turn, and triggers severe model hallucination loops.

Resolving this latency bottleneck requires asynchronous, non-blocking streaming transports over Server-Sent Events (SSE) coupled with MCP JSON-RPC 2.0 Progress Notifications (notifications/progress).

This technical guide details the asynchronous streaming architecture, provides verified multi-host configuration recipes, demonstrates wire-level progress notification frames, and implements a production TypeScript server using Express and BullMQ.

1. Streaming Architecture: Asynchronous Task Decoupling

To prevent long-running tasks from freezing the server's communication channel, the MCP gateway decouples request reception from background execution using an asynchronous worker queue:

Synchronous Blocking Anti-Pattern (High Failure Rate):
Host Agent ──> [tools/call: long_task] ──> (Blocks 90s) ──> Gateway Timeout (504)!

Decoupled Streaming Architecture:
┌────────────────────────────────────────────────────────┐
│ HOST AI AGENT (Claude Desktop / Cursor / Codex CLI)    │
│ Sends tools/call with params._meta.progressToken: 991  │
└──────────────────────────┬─────────────────────────────┘
                           │ 1. POST /messages (JSON-RPC)
                           ▼
┌────────────────────────────────────────────────────────┐
│ REMOTE MCP SSE GATEWAY (Node.js / Express)             │
│ • Accepts request, enqueues background job             │
│ • Returns initial acknowledgement immediately          │
└─────────────┬────────────────────────────▲─────────────┘
              │ 2. Enqueue Job             │ 3. Stream SSE Events
              ▼                            │
┌──────────────────────────┐  ┌────────────┴─────────────┐
│ REDIS BULLMQ TASK QUEUE  │  │ SSE NOTIFICATION STREAM  │
│ Background Worker Daemon │  │ • notifications/progress │
│ • Compiles assets        │  │   (25% - Compiling)      │
│ • Runs test suite        │  │ • notifications/progress │
│ • Dispatches updates     │  │   (80% - Linking)        │
└──────────────────────────┘  │ • Final JSON-RPC Result  │
                              └──────────────────────────┘

The streaming lifecycle operates in four stages:

1. ▸Request Reception: The host transmits a tools/call payload that includes an optional _meta.progressToken.
2. ▸Immediate Task Offloading: The server pushes the work item onto a Redis task queue (e.g., BullMQ) and maintains the open Server-Sent Events stream.
3. ▸Progress Streaming: As the background worker advances, it publishes status updates back to the MCP gateway, which emits notifications/progress frames across the open SSE stream.
4. ▸Final Result Delivery: Upon job completion, the server transmits the standard JSON-RPC response frame containing the final content output.

2. Prerequisites and Environment Matrix

Before configuring an SSE streaming MCP server, verify your development environment satisfies the runtime requirements:

3. Multi-Host Configuration Snippets

Configure your AI assistant host to connect with remote streaming MCP servers over HTTP Server-Sent Events.

Claude Desktop Configuration

Configure Claude Desktop to connect to your streaming server endpoint:

{
  "mcpServers": {
    "streaming-worker": {
      "url": "https://mcp-streaming.internal.example.com/sse",
      "headers": {
        "Authorization": "Bearer STREAMING_SERVICE_TOKEN_PLACEHOLDER"
      }
    }
  }
}

Cursor IDE Configuration (.cursor/mcp.json)

Inside Cursor, configure the remote SSE server:

{
  "mcpServers": {
    "streaming-worker": {
      "url": "/sse"
    }
  }
}

OpenAI Codex CLI Configuration (~/.codex/config.toml)

When utilizing the OpenAI Codex CLI:

[mcp.servers.streaming_worker]
url = "/sse"
bearer_token = "STREAMING_SERVICE_TOKEN_PLACEHOLDER"

4. Step-by-Step Server Implementation

Step 1: Express SSE Server and Progress Emitter

Below is a complete TypeScript server that handles SSE connections and dispatches non-blocking progress notifications:

// streaming-server.ts
import express from 'express';
import { Server } from '@modelcontextprotocol/sdk/server/index.js';
import { SSEServerTransport } from '@modelcontextprotocol/sdk/server/sse.js';
import { CallToolRequestSchema, ListToolsRequestSchema } from '@modelcontextprotocol/sdk/types.js';

const app = express();
const server = new Server(
  { name: 'streaming-mcp-server', version: '1.0.0' },
  { capabilities: { tools: {}, logging: {} } }
);

server.setRequestHandler(ListToolsRequestSchema, async () => {
  return {
    tools: [
      {
        name: 'build_and_deploy_service',
        description: 'Compile container image and deploy to staging with live progress feedback.',
        inputSchema: {
          type: 'object',
          properties: {
            service_name: { type: 'string' }
          },
          required: ['service_name']
        }
      }
    ]
  };
});

server.setRequestHandler(CallToolRequestSchema, async (request, extra) => {
  const { name, arguments: args } = request.params;
  const progressToken = (request.params as any)._meta?.progressToken;

  if (name === 'build_and_deploy_service') {
    const serviceName = args?.service_name as string;

    // Simulate multi-stage background build with progress notifications
    const stages = [
      { step: 1, total: 4, msg: 'Resolving npm dependencies...' },
      { step: 2, total: 4, msg: 'Compiling TypeScript bundle...' },
      { step: 3, total: 4, msg: 'Building Docker container image...' },
      { step: 4, total: 4, msg: 'Pushing image to private registry...' }
    ];

    for (const stage of stages) {
      await new Promise(resolve => setTimeout(resolve, 1500));

      if (progressToken && extra?.sendNotification) {
        // Emit standardized MCP progress notification
        await extra.sendNotification({
          method: 'notifications/progress',
          params: {
            progressToken,
            progress: stage.step,
            total: stage.total,
            message: stage.msg
          }
        });
      }
    }

    return {
      content: [
        {
          type: 'text',
          text: `Service '${serviceName}' successfully built and deployed to staging cluster.`
        }
      ]
    };
  }

  throw new Error(`Tool not found: ${name}`);
});

let transport: SSEServerTransport | null = null;

app.get('/sse', async (req, res) => {
  transport = new SSEServerTransport('/messages', res);
  await server.connect(transport);
});

app.post('/messages', async (req, res) => {
  if (transport) {
    await transport.handlePostMessage(req, res);
  } else {
    res.status(503).send('SSE transport not initialized');
  }
});

app.listen(3000, () => {
  console.log('Streaming MCP Server listening on port 3000');
});

Step 2: Wire Frame Event Flow

When the agent initiates the task, it supplies the progressToken:

{
  "jsonrpc": "2.0",
  "id": 84,
  "method": "tools/call",
  "params": {
    "name": "build_and_deploy_service",
    "arguments": { "service_name": "auth-service" },
    "_meta": { "progressToken": "job-token-84" }
  }
}

While execution progresses, the server emits streaming frames across the SSE socket:

event: message
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"job-token-84","progress":2,"total":4,"message":"Compiling TypeScript bundle..."}}

The host UI renders a live progress bar without dropping the transport connection.

5. Common Errors and Production Troubleshooting

Error 1: Intermediate HTTP Proxy Timeout Drops

• ▸Symptom: Long-running requests fail after exactly 60 seconds with 504 Gateway Timeout through Cloudflare or NGINX.
• ▸Root Cause: Intermediary load balancers close idle HTTP streams that do not emit data packets.
• ▸Resolution: Configure the server to transmit periodic SSE comment heartbeats (: keep-alive\n\n) every 15 seconds to keep intermediate sockets open.

Error 2: Missing progressToken Graceful Degradation

• ▸Symptom: The server throws runtime exceptions when attempting to send progress notifications to older clients.
• ▸Root Cause: The client did not supply params._meta.progressToken during tool invocation.
• ▸Resolution: Always wrap notification calls in an existence check (if (progressToken) ...). If the token is absent, execute the task to completion without emitting progress events.

Error 3: Event Loop Latency Spikes during Heavy Local Computations

• ▸Symptom: The MCP server becomes unresponsive to incoming health checks while running image processing tasks.
• ▸Root Cause: Synchronous CPU-bound tasks blocked the main Node.js event loop.
• ▸Resolution: Offload intensive computational workloads to Node.js worker_threads or external background job processors like BullMQ.

6. External Citations and Security Guardrails

Official Specifications and References

• ▸Model Context Protocol Specification (opens in a new tab) — Official standard for JSON-RPC 2.0 progress notifications and SSE transport mechanics.
• ▸WHATWG Server-Sent Events Standard (opens in a new tab) — Technical standard for unidirectional HTTP event streaming.
• ▸BullMQ Asynchronous Task Queue (opens in a new tab) — Production architecture for Redis-backed background worker distribution.

Security and Governance Guardrails

When exposing streaming SSE endpoints:

1. ▸Authenticate the Initial SSE Handshake: Ensure requests to /sse validate bearer tokens or mutual TLS certificates prior to opening streaming sockets.
2. ▸Enforce Rate Limits per Progress Token: Limit progress notifications to a maximum of five events per second to prevent flooding client WebSocket bridges.