Add Streamable HTTP transport for MCP (protocol 2025-11-25)

APIbase · claude · APIbase · commit 3ddd619486ee · 2026-03-12T06:31:46.000Z
Primary /mcp endpoint now uses StreamableHTTPServerTransport with session
management via mcp-session-id header. SSE transport moved to /sse + /messages
for backward compatibility. Nginx configs updated for new routes. Smoke test
updated to test auth rejection via Streamable HTTP POST.

Co-Authored-By: Claude Opus 4.6 &lt;noreply@anthropic.com&gt;
diff --git a/nginx/apibase-host.conf b/nginx/apibase-host.conf
@@ -64,4 +64,29 @@ server {
         proxy_cache off;
         proxy_read_timeout 300s;
     }
+
+    # SSE transport (deprecated, backward compat) — no buffering
+    location /sse {
+        proxy_pass http://127.0.0.1:8880;
+        proxy_http_version 1.1;
+        proxy_set_header Host $host;
+        proxy_set_header X-Real-IP $remote_addr;
+        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+        proxy_set_header X-Forwarded-Proto $scheme;
+        proxy_set_header Connection '';
+        chunked_transfer_encoding off;
+        proxy_buffering off;
+        proxy_cache off;
+        proxy_read_timeout 300s;
+    }
+
+    location /messages {
+        proxy_pass http://127.0.0.1:8880;
+        proxy_http_version 1.1;
+        proxy_set_header Host $host;
+        proxy_set_header X-Real-IP $remote_addr;
+        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+        proxy_set_header X-Forwarded-Proto $scheme;
+        proxy_read_timeout 60s;
+    }
 }
diff --git a/nginx/nginx.conf b/nginx/nginx.conf
@@ -81,7 +81,7 @@ http {
             proxy_set_header X-Request-ID $req_id;
         }
 
-        # --- MCP Routes (SSE, no buffering) ---
+        # --- MCP Routes (Streamable HTTP + SSE, no buffering) ---
         location /mcp {
             proxy_pass http://api_backend;
             proxy_set_header Host $host;
@@ -95,6 +95,28 @@ http {
             proxy_cache off;
         }
 
+        # --- SSE transport (deprecated, backward compat) ---
+        location /sse {
+            proxy_pass http://api_backend;
+            proxy_set_header Host $host;
+            proxy_set_header X-Real-IP $remote_addr;
+            proxy_set_header X-Forwarded-Proto $scheme;
+            proxy_set_header X-Request-ID $req_id;
+            proxy_set_header Connection '';
+            proxy_http_version 1.1;
+            chunked_transfer_encoding off;
+            proxy_buffering off;
+            proxy_cache off;
+        }
+
+        location /messages {
+            proxy_pass http://api_backend;
+            proxy_set_header Host $host;
+            proxy_set_header X-Real-IP $remote_addr;
+            proxy_set_header X-Forwarded-Proto $scheme;
+            proxy_set_header X-Request-ID $req_id;
+        }
+
         # --- Static Files (§12.51) ---
         location /.well-known/ {
             root /var/www/static;
diff --git a/scripts/smoke-test.sh b/scripts/smoke-test.sh
@@ -143,10 +143,14 @@ else
 fi
 
 # ---------------------------------------------------------------------------
-# 7. Auth rejection — GET /mcp without Authorization → 401
+# 7. Auth rejection — POST /mcp initialize without Authorization → 401
 # ---------------------------------------------------------------------------
 echo -n "7/8 Auth rejection..."
 AUTH_CODE=$(curl -s -o /dev/null -w "%{http_code}" \
+  -X POST \
+  -H "Content-Type: application/json" \
+  -H "Accept: application/json, text/event-stream" \
+  -d '{"jsonrpc":"2.0","method":"initialize","id":1,"params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"test","version":"1.0.0"}}}' \
   "$API_URL/mcp" 2>/dev/null || echo "000")
 if [ "$AUTH_CODE" = "401" ]; then
   pass
diff --git a/src/mcp/server.ts b/src/mcp/server.ts
@@ -1,47 +1,223 @@
 /**
  * MCP server factory and Express route handlers (§12.42, §6.14, §12.1).
  *
- * SSE transport only (Phase 1 — no WebSocket per §12.42, §16).
- * One McpServer + SSEServerTransport per SSE connection (SDK 1:1 design).
- * All tool calls route through the full 13-stage pipeline.
+ * Dual transport:
+ *   - Streamable HTTP (primary, protocol 2025-11-25) — /mcp
+ *   - SSE (deprecated, backward compat, protocol 2024-11-05) — /sse + /messages
+ *
+ * One McpServer + Transport per session. All tool calls route through the
+ * full 13-stage pipeline.
  */
 
 import express from 'express';
 import { randomUUID } from 'node:crypto';
 import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js';
 import { SSEServerTransport } from '@modelcontextprotocol/sdk/server/sse.js';
+import { StreamableHTTPServerTransport } from '@modelcontextprotocol/sdk/server/streamableHttp.js';
+import { isInitializeRequest } from '@modelcontextprotocol/sdk/types.js';
 import { logger } from '../config/logger';
 import { registerTools } from './tool-adapter';
 
-/** Active SSE sessions: sessionId → transport */
-const sessions = new Map<string, SSEServerTransport>();
+/** Active sessions: sessionId → transport (both transport types) */
+const sessions = new Map<string, SSEServerTransport | StreamableHTTPServerTransport>();
+
+/**
+ * Extract Bearer API key from Authorization header.
+ */
+function extractApiKey(req: express.Request): string | null {
+  const authHeader = req.headers.authorization;
+  if (!authHeader || !authHeader.startsWith('Bearer ')) {
+    return null;
+  }
+  return authHeader.slice(7);
+}
 
 /**
- * Create Express router for MCP endpoint.
+ * Create Express router for MCP endpoints.
+ *
+ * Streamable HTTP (primary):
+ *   POST   /mcp — JSON-RPC messages (initialize creates session)
+ *   GET    /mcp — SSE subscription for server-initiated notifications
+ *   DELETE /mcp — close session
  *
- * GET  /mcp              — SSE stream (sends endpoint event with sessionId)
- * POST /mcp?sessionId=x  — JSON-RPC messages routed to the correct transport
+ * SSE (deprecated, backward compat):
+ *   GET  /sse      — SSE stream
+ *   POST /messages — JSON-RPC messages
  */
 export function createMcpRouter(): express.Router {
   const router = express.Router();
 
-  // -------------------------------------------------------------------------
-  // GET /mcp — SSE stream
-  // -------------------------------------------------------------------------
-  router.get('/mcp', (req: express.Request, res: express.Response) => {
-    const authHeader = req.headers.authorization;
-    if (!authHeader || !authHeader.startsWith('Bearer ')) {
+  // =========================================================================
+  // Streamable HTTP transport — /mcp (primary, protocol 2025-11-25)
+  // =========================================================================
+
+  // --- POST /mcp: JSON-RPC messages ---
+  router.post('/mcp', async (req: express.Request, res: express.Response) => {
+    try {
+      const sessionId = req.headers['mcp-session-id'] as string | undefined;
+
+      // Existing session: route to stored transport
+      if (sessionId) {
+        const transport = sessions.get(sessionId);
+        if (!transport || !(transport instanceof StreamableHTTPServerTransport)) {
+          res.status(404).json({
+            jsonrpc: '2.0',
+            error: { code: -32000, message: 'Session not found' },
+            id: null,
+          });
+          return;
+        }
+        await transport.handleRequest(req, res, req.body);
+        return;
+      }
+
+      // New session: must be an initialize request
+      if (!isInitializeRequest(req.body)) {
+        res.status(400).json({
+          jsonrpc: '2.0',
+          error: {
+            code: -32600,
+            message: 'Bad Request: first request must be an initialize request',
+          },
+          id: null,
+        });
+        return;
+      }
+
+      const apiKey = extractApiKey(req);
+      if (!apiKey) {
+        res.status(401).json({
+          error: 'unauthorized',
+          message: 'Missing or invalid Authorization header. Expected: Bearer <api_key>',
+        });
+        return;
+      }
+
+      const requestId = (req.headers['x-request-id'] as string) || randomUUID();
+
+      const transport = new StreamableHTTPServerTransport({
+        sessionIdGenerator: () => randomUUID(),
+        onsessioninitialized: (sid: string) => {
+          sessions.set(sid, transport);
+          logger.info(
+            { request_id: requestId, session_id: sid },
+            'MCP Streamable HTTP session created',
+          );
+        },
+        onsessionclosed: (sid: string) => {
+          sessions.delete(sid);
+          logger.info(
+            { request_id: requestId, session_id: sid },
+            'MCP Streamable HTTP session closed',
+          );
+        },
+      });
+
+      transport.onerror = (error: Error) => {
+        logger.error(
+          { request_id: requestId, err: error },
+          'MCP Streamable HTTP transport error',
+        );
+      };
+
+      const mcpServer = new McpServer(
+        { name: 'APIbase', version: '1.0.0' },
+        { capabilities: { tools: {} } },
+      );
+
+      registerTools(mcpServer, apiKey, requestId);
+      await mcpServer.connect(transport);
+
+      await transport.handleRequest(req, res, req.body);
+    } catch (error) {
+      logger.error({ err: error }, 'MCP Streamable HTTP POST error');
+      if (!res.headersSent) {
+        res.status(500).json({ error: 'internal_error', message: 'MCP request handling failed' });
+      }
+    }
+  });
+
+  // --- GET /mcp: SSE subscription for server-initiated notifications ---
+  router.get('/mcp', async (req: express.Request, res: express.Response) => {
+    try {
+      const sessionId = req.headers['mcp-session-id'] as string | undefined;
+      if (!sessionId) {
+        res.status(400).json({
+          error: 'bad_request',
+          message: 'Missing mcp-session-id header',
+        });
+        return;
+      }
+
+      const transport = sessions.get(sessionId);
+      if (!transport || !(transport instanceof StreamableHTTPServerTransport)) {
+        res.status(404).json({
+          jsonrpc: '2.0',
+          error: { code: -32000, message: 'Session not found' },
+          id: null,
+        });
+        return;
+      }
+
+      await transport.handleRequest(req, res);
+    } catch (error) {
+      logger.error({ err: error }, 'MCP Streamable HTTP GET error');
+      if (!res.headersSent) {
+        res.status(500).json({ error: 'internal_error', message: 'MCP SSE subscription failed' });
+      }
+    }
+  });
+
+  // --- DELETE /mcp: close session ---
+  router.delete('/mcp', async (req: express.Request, res: express.Response) => {
+    try {
+      const sessionId = req.headers['mcp-session-id'] as string | undefined;
+      if (!sessionId) {
+        res.status(400).json({
+          error: 'bad_request',
+          message: 'Missing mcp-session-id header',
+        });
+        return;
+      }
+
+      const transport = sessions.get(sessionId);
+      if (!transport || !(transport instanceof StreamableHTTPServerTransport)) {
+        res.status(404).json({
+          jsonrpc: '2.0',
+          error: { code: -32000, message: 'Session not found' },
+          id: null,
+        });
+        return;
+      }
+
+      await transport.handleRequest(req, res);
+      sessions.delete(sessionId);
+    } catch (error) {
+      logger.error({ err: error }, 'MCP Streamable HTTP DELETE error');
+      if (!res.headersSent) {
+        res.status(500).json({ error: 'internal_error', message: 'MCP session close failed' });
+      }
+    }
+  });
+
+  // =========================================================================
+  // SSE transport — /sse + /messages (deprecated, backward compat)
+  // =========================================================================
+
+  // --- GET /sse: establish SSE stream ---
+  router.get('/sse', (req: express.Request, res: express.Response) => {
+    const apiKey = extractApiKey(req);
+    if (!apiKey) {
       res.status(401).json({
         error: 'unauthorized',
         message: 'Missing or invalid Authorization header. Expected: Bearer <api_key>',
       });
       return;
     }
 
-    const apiKey = authHeader.slice(7);
     const requestId = (req.headers['x-request-id'] as string) || randomUUID();
 
-    const transport = new SSEServerTransport('/mcp', res);
+    const transport = new SSEServerTransport('/messages', res);
     const sessionId = transport.sessionId;
 
     const mcpServer = new McpServer(
@@ -54,7 +230,6 @@ export function createMcpRouter(): express.Router {
     sessions.set(sessionId, transport);
     logger.info({ request_id: requestId, session_id: sessionId }, 'MCP SSE session created');
 
-    // Cleanup on connection close
     const cleanup = (): void => {
       sessions.delete(sessionId);
       logger.info({ request_id: requestId, session_id: sessionId }, 'MCP SSE session closed');
@@ -82,10 +257,8 @@ export function createMcpRouter(): express.Router {
     });
   });
 
-  // -------------------------------------------------------------------------
-  // POST /mcp — JSON-RPC messages
-  // -------------------------------------------------------------------------
-  router.post('/mcp', (req: express.Request, res: express.Response) => {
+  // --- POST /messages: JSON-RPC messages for SSE transport ---
+  router.post('/messages', (req: express.Request, res: express.Response) => {
     const sessionId = req.query.sessionId as string | undefined;
     if (!sessionId) {
       res.status(400).json({
@@ -96,7 +269,7 @@ export function createMcpRouter(): express.Router {
     }
 
     const transport = sessions.get(sessionId);
-    if (!transport) {
+    if (!transport || !(transport instanceof SSEServerTransport)) {
       res.status(400).json({
         error: 'bad_request',
         message: 'Unknown or expired sessionId',
