Step 3: Session Management and SSE

In this step, we will delve into how session management and Server-Sent Events (SSE) are implemented in the MCP Server using Node-RED.

Session Management

Session management is crucial for maintaining state and handling multiple client connections. The following JavaScript code snippet from a function node demonstrates how sessions are managed:

// Get or init connection map
let seenIPs = global.get("seenIPs") || {};

function generateUUID() {
  return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, function (c) {
    const r = Math.random() * 16 | 0,
      v = c === 'x' ? r : (r & 0x3 | 0x8);
    return v.toString(16);
  });
}

const sessionId = generateUUID();
const endpoint = `/sse/message?sessionId=${sessionId}`;

// Store session (optional)
let sessions = global.get("sse_sessions") || {};
sessions[sessionId] = {
  createdAt: new Date().toISOString(),
  sseRes: msg.res._res
};
global.set("sse_sessions", sessions);

Server-Sent Events (SSE)

SSE is used to push updates to clients in real-time. Here's how SSE is set up in the MCP Server:

// Set headers
msg.res._res.setHeader("Content-Type", "text/event-stream");
msg.res._res.setHeader("Cache-Control", "no-cache");
msg.res._res.setHeader("Connection", "keep-alive");
msg.res._res.setHeader("access-control-max-age", "86400");
msg.res._res.setHeader("Access-Control-Allow-Origin", "*");
msg.res._res.setHeader("Access-Control-Allow-Methods", "GET, POST, OPTIONS");
msg.res._res.setHeader("Access-Control-Allow-Headers", "Authorization, *");

// Send initial SSE message with endpoint info
msg.res._res.write(`event: endpoint\ndata: ${msg.endpoint}\n\n`);

JSON Export

Below is the JSON export for this step, which you can import into Node-RED to replicate the setup:

{
  "id": "3321740562069c7c",
  "type": "tab",
  "label": "MCP Server",
  "nodes": [
    {
      "id": "c882531a6a2da6a2",
      "type": "http in",
      "z": "3321740562069c7c",
      "name": "",
      "url": "/sse",
      "method": "get",
      "upload": false,
      "swaggerDoc": "",
      "x": 100,
      "y": 60,
      "wires": [
        [
          "2a76a8cf4603effc",
          "2f4cbc634ee2512c"
        ]
      ]
    },
    {
      "id": "2f4cbc634ee2512c",
      "type": "function",
      "z": "3321740562069c7c",
      "name": "function 877",
      "func": "// Get or init connection map\nlet seenIPs = global.get(\"seenIPs\") || {};\n\nfunction generateUUID() {\n  return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, function (c) {\n    const r = Math.random() * 16 | 0,\n      v = c === 'x' ? r : (r & 0x3 | 0x8);\n    return v.toString(16);\n  });\n}\n\nconst sessionId = generateUUID();\nconst endpoint = `/sse/message?sessionId=${sessionId}`;\n\n// Store session (optional)\nlet sessions = global.get(\"sse_sessions\") || {};\nsessions[sessionId] = {\n  createdAt: new Date().toISOString(),\n  sseRes: msg.res._res\n};\nglobal.set(\"sse_sessions\", sessions);\n\n// Set headers\nmsg.res._res.setHeader(\"Content-Type\", \"text/event-stream\");\nmsg.res._res.setHeader(\"Cache-Control\", \"no-cache\");\nmsg.res._res.setHeader(\"Connection\", \"keep-alive\");\nmsg.res._res.setHeader(\"access-control-max-age\", \"86400\");\nmsg.res._res.setHeader(\"Access-Control-Allow-Origin\", \"*\");\nmsg.res._res.setHeader(\"Access-Control-Allow-Methods\", \"GET, POST, OPTIONS\");\nmsg.res._res.setHeader(\"Access-Control-Allow-Headers\", \"Authorization, *\");\n\n// Send initial SSE message with endpoint info\nmsg.res._res.write(`event: endpoint\\ndata: ${msg.endpoint}\\n\\n`);\n\nconst messagePoller = setInterval(() => {\n  const currentSessions = global.get(\"sse_sessions\") || {};\n  const session = currentSessions[sessionId];\n\n  if (!session) {\n    clearInterval(messagePoller);\n    return;\n  }\n\n  session.messages = session.messages || [];\n\n  for (let i = 0; i < session.messages.length; i++) {\n    const msgObj = session.messages[i];\n    if (!msgObj.sent) {\n      session.sseRes.write(`event: message\\ndata: ${JSON.stringify(msgObj.data)}\\n\\n`);\n      msgObj.sent = true;\n    }\n  }\n\n  // Save session state\n  currentSessions[sessionId] = session;\n  global.set(\"sse_sessions\", currentSessions);\n}, 1000);\n\n\nreturn null;",
      "outputs": 1,
      "timeout": 0,
      "noerr": 0,
      "initialize": "",
      "finalize": "",
      "libs": [],
      "x": 290,
      "y": 60,
      "wires": [
        [
          "2a76a8cf4603effc"
        ]
      ]
    }
  ]
}