fix(mcp-server): support concurrent sessions via per-session transports

kvz · claude · kvz · commit 10c6e48aae1c · 2026-03-26T09:09:52.000+01:00
The HTTP and Express handlers previously created a single
StreamableHTTPServerTransport shared by all clients. After the first
client initialized, subsequent clients received "Server already
initialized" because the transport tracks initialization state per
instance. This switches to the SDK-recommended pattern of creating a
new transport + server pair for each session, keyed by session ID.

Co-Authored-By: Claude Opus 4.6 &lt;noreply@anthropic.com&gt;
diff --git a/packages/mcp-server/src/express.ts b/packages/mcp-server/src/express.ts
@@ -1,9 +1,9 @@
 import { randomUUID } from 'node:crypto'
 import { StreamableHTTPServerTransport } from '@modelcontextprotocol/sdk/server/streamableHttp.js'
+import { isInitializeRequest } from '@modelcontextprotocol/sdk/types.js'
 import express from 'express'
 import type { TransloaditMcpHttpOptions } from './http.ts'
 import { isBasicAuthorized } from './http-helpers.ts'
-import { createMcpRequestHandler } from './http-request-handler.ts'
 import { getMetrics, getMetricsContentType } from './metrics.ts'
 import { createTransloaditMcpServer } from './server.ts'
 import { buildServerCard, serverCardPath } from './server-card.ts'
@@ -15,28 +15,16 @@ export type TransloaditMcpExpressOptions = TransloaditMcpHttpOptions & {
 export const createTransloaditMcpExpressRouter = async (
   options: TransloaditMcpExpressOptions = {},
 ) => {
-  const server = createTransloaditMcpServer(options)
-  const transport = new StreamableHTTPServerTransport({
-    sessionIdGenerator: options.sessionIdGenerator ?? (() => randomUUID()),
-    allowedOrigins: options.allowedOrigins,
-    allowedHosts: options.allowedHosts,
-    enableDnsRebindingProtection: options.enableDnsRebindingProtection,
-  })
+  const sessionIdGenerator = options.sessionIdGenerator ?? (() => randomUUID())
 
-  await server.connect(transport)
+  // Per-session transport map: each MCP client gets its own transport + server pair.
+  const transports = new Map<string, StreamableHTTPServerTransport>()
 
   const router = express.Router()
   const routePath = options.path ?? '/mcp'
   const metricsPath =
     options.metricsPath === false ? undefined : (options.metricsPath ?? '/metrics')
   const metricsAuth = options.metricsAuth
-  const handler = createMcpRequestHandler(transport, {
-    allowedOrigins: options.allowedOrigins,
-    mcpToken: options.mcpToken,
-    path: { expectedPath: routePath, allowRoot: true },
-    logger: options.logger,
-    redactSecrets: [options.mcpToken, options.authKey, options.authSecret],
-  })
 
   const serverCardJson = JSON.stringify(
     buildServerCard(routePath, { authKey: options.authKey, authSecret: options.authSecret }),
@@ -77,8 +65,60 @@ export const createTransloaditMcpExpressRouter = async (
     sendServerCard(res, false)
   })
 
-  router.all(routePath, (req, res) => {
-    void handler(req, res)
+  router.all(routePath, async (req: express.Request, res: express.Response) => {
+    const sessionId = req.headers['mcp-session-id'] as string | undefined
+    let transport: StreamableHTTPServerTransport | undefined
+
+    if (sessionId) {
+      transport = transports.get(sessionId)
+      if (!transport) {
+        res.status(404).json({
+          jsonrpc: '2.0',
+          error: { code: -32000, message: 'Session not found' },
+          id: null,
+        })
+        return
+      }
+    } else if (req.method === 'POST' && isInitializeRequest(req.body)) {
+      // New initialization request — create a new transport + server pair.
+      transport = new StreamableHTTPServerTransport({
+        sessionIdGenerator,
+        allowedOrigins: options.allowedOrigins,
+        allowedHosts: options.allowedHosts,
+        enableDnsRebindingProtection: options.enableDnsRebindingProtection,
+        onsessioninitialized: (sid) => {
+          transports.set(sid, transport!)
+        },
+      })
+
+      transport.onclose = () => {
+        const sid = transport!.sessionId
+        if (sid) {
+          transports.delete(sid)
+        }
+      }
+
+      const server = createTransloaditMcpServer(options)
+      await server.connect(transport)
+    } else if (req.method === 'POST') {
+      res.status(400).json({
+        jsonrpc: '2.0',
+        error: { code: -32600, message: 'Bad Request: No valid session ID provided' },
+        id: null,
+      })
+      return
+    }
+
+    if (!transport) {
+      res.status(400).json({
+        jsonrpc: '2.0',
+        error: { code: -32600, message: 'Bad Request: No valid session ID provided' },
+        id: null,
+      })
+      return
+    }
+
+    await transport.handleRequest(req, res, req.body)
   })
 
   if (metricsPath) {
diff --git a/packages/mcp-server/src/http.ts b/packages/mcp-server/src/http.ts
@@ -1,14 +1,15 @@
 import { randomUUID } from 'node:crypto'
 import type { IncomingMessage, ServerResponse } from 'node:http'
 import { StreamableHTTPServerTransport } from '@modelcontextprotocol/sdk/server/streamableHttp.js'
+import { isInitializeRequest } from '@modelcontextprotocol/sdk/types.js'
 import type { SevLogger } from '@transloadit/sev-logger'
 import {
   applyCorsHeaders,
+  isAuthorized,
   isBasicAuthorized,
   normalizePath,
   parsePathname,
 } from './http-helpers.ts'
-import { createMcpRequestHandler } from './http-request-handler.ts'
 import { getMetrics, getMetricsContentType } from './metrics.ts'
 import type { TransloaditMcpServerOptions } from './server.ts'
 import { createTransloaditMcpServer } from './server.ts'
@@ -35,31 +36,38 @@ export type TransloaditMcpHttpHandler = ((
 
 const defaultPath = '/mcp'
 
+/** Read the full request body and JSON-parse it so `isInitializeRequest` can inspect the payload. */
+async function readJsonBody(req: IncomingMessage): Promise<unknown> {
+  return new Promise((resolve, reject) => {
+    const chunks: Buffer[] = []
+    req.on('data', (chunk: Buffer) => chunks.push(chunk))
+    req.on('end', () => {
+      const raw = Buffer.concat(chunks).toString('utf8')
+      if (!raw) {
+        resolve(undefined)
+        return
+      }
+      try {
+        resolve(JSON.parse(raw))
+      } catch {
+        resolve(undefined)
+      }
+    })
+    req.on('error', reject)
+  })
+}
+
 export const createTransloaditMcpHttpHandler = async (
   options: TransloaditMcpHttpOptions = {},
 ): Promise<TransloaditMcpHttpHandler> => {
-  const server = createTransloaditMcpServer(options)
-  const transport = new StreamableHTTPServerTransport({
-    sessionIdGenerator: options.sessionIdGenerator ?? (() => randomUUID()),
-    allowedOrigins: options.allowedOrigins,
-    allowedHosts: options.allowedHosts,
-    enableDnsRebindingProtection: options.enableDnsRebindingProtection,
-  })
-
-  await server.connect(transport)
-
   const expectedPath = options.path ?? defaultPath
   const metricsPath =
     options.metricsPath === false ? undefined : normalizePath(options.metricsPath ?? '/metrics')
   const metricsAuth = options.metricsAuth
+  const sessionIdGenerator = options.sessionIdGenerator ?? (() => randomUUID())
 
-  const mcpHandler = createMcpRequestHandler(transport, {
-    allowedOrigins: options.allowedOrigins,
-    mcpToken: options.mcpToken,
-    path: { expectedPath },
-    logger: options.logger,
-    redactSecrets: [options.mcpToken, options.authKey, options.authSecret],
-  })
+  // Per-session transport map: each MCP client gets its own transport + server pair.
+  const transports = new Map<string, StreamableHTTPServerTransport>()
 
   const serverCardJson = JSON.stringify(
     buildServerCard(expectedPath, { authKey: options.authKey, authSecret: options.authSecret }),
@@ -69,7 +77,6 @@ export const createTransloaditMcpHttpHandler = async (
     const pathname = normalizePath(parsePathname(req.url, expectedPath))
 
     if (pathname === serverCardPath) {
-      // Public discovery endpoint for registries; always allow CORS (optionally restricted by allowedOrigins).
       if (!applyCorsHeaders(req, res, options.allowedOrigins)) {
         return
       }
@@ -115,11 +122,131 @@ export const createTransloaditMcpHttpHandler = async (
       return
     }
 
-    await mcpHandler(req, res)
+    if (pathname !== normalizePath(expectedPath)) {
+      res.statusCode = 404
+      res.end('Not Found')
+      return
+    }
+
+    if (!applyCorsHeaders(req, res, options.allowedOrigins)) {
+      return
+    }
+
+    if (req.method === 'OPTIONS') {
+      res.statusCode = 204
+      res.end()
+      return
+    }
+
+    if (options.mcpToken && !isAuthorized(req, options.mcpToken)) {
+      res.statusCode = 401
+      res.setHeader('WWW-Authenticate', 'Bearer')
+      res.end('Unauthorized')
+      return
+    }
+
+    // Bare GETs without the SSE Accept header are not valid MCP requests (the
+    // Streamable HTTP spec requires Accept: text/event-stream for GET).  Return
+    // a friendly JSON status so directory health-probes see a 200 instead of 406.
+    const accept = req.headers.accept ?? ''
+    if (req.method === 'GET' && !accept.includes('text/event-stream')) {
+      res.statusCode = 200
+      res.setHeader('Content-Type', 'application/json')
+      res.end(
+        JSON.stringify({
+          name: 'Transloadit MCP Server',
+          status: 'ok',
+          docs: 'https://transloadit.com/docs/sdks/mcp-server/',
+        }),
+      )
+      return
+    }
+
+    // Route request to the correct per-session transport.
+    const sessionId = req.headers['mcp-session-id'] as string | undefined
+    let transport: StreamableHTTPServerTransport | undefined
+
+    if (sessionId) {
+      transport = transports.get(sessionId)
+      if (!transport) {
+        res.statusCode = 404
+        res.setHeader('Content-Type', 'application/json')
+        res.end(
+          JSON.stringify({
+            jsonrpc: '2.0',
+            error: { code: -32000, message: 'Session not found' },
+            id: null,
+          }),
+        )
+        return
+      }
+    }
+
+    // For POST requests without a session, read the body to check for initialization.
+    let parsedBody: unknown
+    if (req.method === 'POST' && !transport) {
+      parsedBody = await readJsonBody(req)
+      if (isInitializeRequest(parsedBody)) {
+        transport = new StreamableHTTPServerTransport({
+          sessionIdGenerator,
+          allowedOrigins: options.allowedOrigins,
+          allowedHosts: options.allowedHosts,
+          enableDnsRebindingProtection: options.enableDnsRebindingProtection,
+          onsessioninitialized: (sid) => {
+            transports.set(sid, transport!)
+          },
+        })
+
+        transport.onclose = () => {
+          const sid = transport!.sessionId
+          if (sid) {
+            transports.delete(sid)
+          }
+        }
+
+        const server = createTransloaditMcpServer(options)
+        await server.connect(transport)
+      } else {
+        res.statusCode = 400
+        res.setHeader('Content-Type', 'application/json')
+        res.end(
+          JSON.stringify({
+            jsonrpc: '2.0',
+            error: { code: -32600, message: 'Bad Request: No valid session ID provided' },
+            id: null,
+          }),
+        )
+        return
+      }
+    }
+
+    if (!transport) {
+      res.statusCode = 400
+      res.setHeader('Content-Type', 'application/json')
+      res.end(
+        JSON.stringify({
+          jsonrpc: '2.0',
+          error: { code: -32600, message: 'Bad Request: No valid session ID provided' },
+          id: null,
+        }),
+      )
+      return
+    }
+
+    try {
+      await transport.handleRequest(req, res, parsedBody)
+    } catch (error) {
+      if (!res.headersSent) {
+        res.statusCode = 500
+        res.end('Internal Server Error')
+      }
+    }
   }) as TransloaditMcpHttpHandler
 
   handler.close = async () => {
-    await transport.close()
+    const closePromises = [...transports.values()].map((t) => t.close())
+    await Promise.all(closePromises)
+    transports.clear()
   }
 
   return handler
diff --git a/packages/mcp-server/test/e2e/concurrent-sessions.test.ts b/packages/mcp-server/test/e2e/concurrent-sessions.test.ts
@@ -0,0 +1,37 @@
+import { expect, test } from 'vitest'
+import { createHttpClient, startHttpServer } from './http-server.ts'
+import { parseToolPayload } from './mcp-client.ts'
+
+test('streamable http: supports concurrent sessions', async () => {
+  const server = await startHttpServer()
+
+  try {
+    // Create two independent MCP clients (each sends its own initialize request).
+    const session1 = await createHttpClient(server.url)
+    const session2 = await createHttpClient(server.url)
+
+    try {
+      // Both sessions should be able to call tools independently.
+      const [robots1, robots2] = await Promise.all([
+        session1.client.callTool({
+          name: 'transloadit_list_robots',
+          arguments: { limit: 1 },
+        }),
+        session2.client.callTool({
+          name: 'transloadit_list_robots',
+          arguments: { limit: 1 },
+        }),
+      ])
+
+      expect(parseToolPayload(robots1).status).toBe('ok')
+      expect(parseToolPayload(robots2).status).toBe('ok')
+    } finally {
+      await session1.transport.close()
+      await session1.client.close()
+      await session2.transport.close()
+      await session2.client.close()
+    }
+  } finally {
+    await server.close()
+  }
+})
