Enable distributed mode

[ZJONSSON]

Enabling distributed MCP server

Motivation and Context

The current implementation of streamableHttps relies on a single node managing all requests or using sticky sessions on a loadbalancer to ensure the same user always hits the same machine. This particular setup faces some problems if the sticky node becomes unavailable for some reason.

This approach makes minimal changes to make fully distributed setup relatively easy.

For a distributed mode you will have to create new transport for any inbound connection. Prior to initializing the transport you need to validate mcp-session-id against a db of open session (or just ignore if undefined = new session).

Initialize transport with optional sessionId argument (if defined, the transport will assume session has already been initialized). Additionally define disableLocalSse: true.

Listen to responseSse events from transport and publish them to a global pub/sub
Listen to the global pub/sub for events matching the current sessionId and emit them as sse locally.

A pseudo example would be something like:

async (req, res) => { 
  const sessionId = req.headers['mcp-session-id'];
  if (sessionId) {
    const validSession = await cache.get(`session:${sessionId}`);
    if (!validSession) {
      return req.status(400).send('Invalid session');
    }
  }
  const server = new Server({ ....}). // dynamically create server
  const transport = new Transport({
    sessionId,
    disableLocalSse: true,
    sessionIdGenerator: () => randomUUID(),
    onsessioninitialized: (sessionId) => {
      this.events('responseSse', data => pubsub.publish(`sse:${sessionId}`, JSON.stringify(data))),
    },
  });
  await server.connect(transport);
  transport.handleRequest(req, res, req.body);
  
  if (sessionId) {
    pubsub.subscribe(`sse:${sessionId}`, (data) => transport.events.emit('sse', JSON.parse(data)));
  }

  req.on('close', () => pubsub.unsubscribe(`sse:${sessionId}`));
}

How Has This Been Tested?

Tested locally - this PR is a draft, if approach makes sense we should add some tests

Breaking Changes

None

Types of changes

• Bug fix (non-breaking change which fixes an issue)
• New feature (non-breaking change which adds functionality)
• Breaking change (fix or feature that would cause existing functionality to change)
• Documentation update

Checklist

• I have read the MCP Documentation
• My code follows the repository's style guidelines
• New and existing tests pass locally
• I have added appropriate error handling
• I have added or updated documentation as needed

Additional context