WIP: Attempting to proxy streamable-http connections. Inspector still works fine with STDIO and SSE servers.

* In index.ts,
  - refactor transport webAppTransports to be a map with the session id as key and transport as value.

* Implement /mcp GET and POST endpoints using StreamableHTTPServerTransport and doing the new session in the POST (opposite from SSE) handler.

* In package.json
  - update the SDK to 1.10.2

* In useConnection.ts
  - import StreamableHTTPClientTransport
    - NOTE: while we NEED to do this, it causes useConnection.test.ts to fail with " ReferenceError: TransformStream is not defined"
  - in connect method
    - instantiate the appropriate transport

Author: cliffhall

client/src/lib/hooks/useConnection.ts

@@ -3,6 +3,7 @@ import {
   SSEClientTransport,
   SseError,
 } from "@modelcontextprotocol/sdk/client/sse.js";
+import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
 import {
   ClientNotification,
   ClientRequest,
@@ -286,6 +287,7 @@ export function useConnection({
         mcpProxyServerUrl.searchParams.append("args", args);
         mcpProxyServerUrl.searchParams.append("env", JSON.stringify(env));
         break;
+
       case "sse":
         mcpProxyServerUrl = new URL(`${getMCPProxyAddress(config)}/sse`);
         mcpProxyServerUrl.searchParams.append("url", sseUrl);
@@ -317,14 +319,24 @@ export function useConnection({
         headers[authHeaderName] = `Bearer ${token}`;
       }
 
-      const clientTransport = new SSEClientTransport(mcpProxyServerUrl as URL, {
+      // Create appropriate transport
+      const transportOptions = {
         eventSourceInit: {
-          fetch: (url, init) => fetch(url, { ...init, headers }),
+          fetch: (
+            url: string | URL | globalThis.Request,
+            init: RequestInit | undefined,
+          ) => fetch(url, { ...init, headers }),
         },
         requestInit: {
           headers,
         },
-      });
+      };
+      const clientTransport =
+        transportType === "streamable-http"
+          ? new StreamableHTTPClientTransport(mcpProxyServerUrl as URL, {
+              sessionId: undefined,
+            })
+          : new SSEClientTransport(mcpProxyServerUrl as URL, transportOptions);
 
       if (onNotification) {
         [

package-lock.json

@@ -41,7 +41,7 @@
     "@modelcontextprotocol/inspector-cli": "^0.10.2",
     "@modelcontextprotocol/inspector-client": "^0.10.2",
     "@modelcontextprotocol/inspector-server": "^0.10.2",
-    "@modelcontextprotocol/sdk": "^1.10.0",
+    "@modelcontextprotocol/sdk": "^1.10.2",
     "concurrently": "^9.0.1",
     "shell-quote": "^1.8.2",
     "spawn-rx": "^5.1.2",

package.json

@@ -12,15 +12,17 @@ import {
   StdioClientTransport,
   getDefaultEnvironment,
 } from "@modelcontextprotocol/sdk/client/stdio.js";
-import { Transport } from "@modelcontextprotocol/sdk/shared/transport.js";
-import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
 import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
+import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js";
+import { SSEServerTransport } from "@modelcontextprotocol/sdk/server/sse.js";
+import { Transport } from "@modelcontextprotocol/sdk/shared/transport.js";
 import express from "express";
 import { findActualExecutable } from "spawn-rx";
 import mcpProxy from "./mcpProxy.js";
+import { randomUUID } from "node:crypto";
 
 const SSE_HEADERS_PASSTHROUGH = ["authorization"];
-const STREAMABLE_HTTP_HEADERS_PASSTHROUGH = ["authorization"];
+const STREAMABLE_HTTP_HEADERS_PASSTHROUGH = ["authorization", "mcp-session-id"];
 
 const defaultEnvironment = {
   ...getDefaultEnvironment(),
@@ -38,7 +40,7 @@ const { values } = parseArgs({
 const app = express();
 app.use(cors());
 
-let webAppTransports: SSEServerTransport[] = [];
+const webAppTransports: Map<string, Transport> = new Map<string, Transport>(); // Transports by sessionId
 
 const createTransport = async (req: express.Request): Promise<Transport> => {
   const query = req.query;
@@ -130,71 +132,89 @@ const createTransport = async (req: express.Request): Promise<Transport> => {
 let backingServerTransport: Transport | undefined;
 
 app.get("/mcp", async (req, res) => {
+  const sessionId = req.headers["mcp-session-id"] as string;
+  console.log(`Received GET message for sessionId ${sessionId}`);
   try {
-    console.log("New streamable-http connection");
+    const transport = webAppTransports.get(
+      sessionId,
+    ) as StreamableHTTPServerTransport;
+    if (!transport) {
+      res.status(404).end("Session not found");
+      return;
+    } else {
+      await transport.handleRequest(req, res);
+    }
+  } catch (error) {
+    console.error("Error in /mcp route:", error);
+    res.status(500).json(error);
+  }
+});
 
+app.post("/mcp", async (req, res) => {
+  const sessionId = req.headers["mcp-session-id"] as string | undefined;
+  console.log(`Received POST message for sessionId ${sessionId}`);
+  if (!sessionId) {
     try {
-      await backingServerTransport?.close();
-      backingServerTransport = await createTransport(req);
-    } catch (error) {
-      if (error instanceof SseError && error.code === 401) {
-        console.error(
-          "Received 401 Unauthorized from MCP server:",
-          error.message,
-        );
-        res.status(401).json(error);
-        return;
+      console.log("New streamable-http connection");
+      try {
+        await backingServerTransport?.close();
+        backingServerTransport = await createTransport(req);
+      } catch (error) {
+        if (error instanceof SseError && error.code === 401) {
+          console.error(
+            "Received 401 Unauthorized from MCP server:",
+            error.message,
+          );
+          res.status(401).json(error);
+          return;
+        }
+
+        throw error;
       }
 
-      throw error;
-    }
-
-    console.log("Connected MCP client to backing server transport");
-
-    const webAppTransport = new SSEServerTransport("/mcp", res);
-    webAppTransports.push(webAppTransport);
-    console.log("Created web app transport");
-
-    await webAppTransport.start();
+      console.log("Connected MCP client to backing server transport");
 
-    if (backingServerTransport instanceof StdioClientTransport) {
-      backingServerTransport.stderr!.on("data", (chunk) => {
-        webAppTransport.send({
-          jsonrpc: "2.0",
-          method: "notifications/stderr",
-          params: {
-            content: chunk.toString(),
-          },
-        });
+      const webAppTransport = new StreamableHTTPServerTransport({
+        sessionIdGenerator: randomUUID,
+        onsessioninitialized: (sessionId) => {
+          webAppTransports.set(sessionId, webAppTransport);
+          console.log("Created streamable web app transport " + sessionId);
+        },
       });
-    }
-
-    mcpProxy({
-      transportToClient: webAppTransport,
-      transportToServer: backingServerTransport,
-    });
 
-    console.log("Set up MCP proxy");
-  } catch (error) {
-    console.error("Error in /sse route:", error);
-    res.status(500).json(error);
-  }
-});
+      await webAppTransport.start();
 
-app.post("/mcp", async (req, res) => {
-  try {
-    const sessionId = req.query.sessionId;
-    console.log(`Received message for sessionId ${sessionId}`);
+      mcpProxy({
+        transportToClient: webAppTransport,
+        transportToServer: backingServerTransport,
+      });
 
-    const transport = webAppTransports.find((t) => t.sessionId === sessionId);
-    if (!transport) {
-      res.status(404).end("Session not found");
-      return;
+      await (webAppTransport as StreamableHTTPServerTransport).handleRequest(
+        req,
+        res,
+        req.body,
+      );
+    } catch (error) {
+      console.error("Error in /mcp POST route:", error);
+      res.status(500).json(error);
+    }
+  } else {
+    try {
+      const transport = webAppTransports.get(
+        sessionId,
+      ) as StreamableHTTPServerTransport;
+      if (!transport) {
+        res.status(404).end("Transport not found for sessionId " + sessionId);
+      } else {
+        await (transport as StreamableHTTPServerTransport).handleRequest(
+          req,
+          res,
+        );
+      }
+    } catch (error) {
+      console.error("Error in /mcp route:", error);
+      res.status(500).json(error);
     }
-    await transport.handlePostMessage(req, res);
-  } catch (error) {
-    console.error("Error in /mcp route:", error);
-    res.status(500).json(error);
   }
 });
 
@@ -221,7 +241,7 @@ app.get("/stdio", async (req, res) => {
     console.log("Connected MCP client to backing server transport");
 
     const webAppTransport = new SSEServerTransport("/message", res);
-    webAppTransports.push(webAppTransport);
+    webAppTransports.set(webAppTransport.sessionId, webAppTransport);
 
     console.log("Created web app transport");
 
@@ -276,8 +296,7 @@ app.get("/sse", async (req, res) => {
     console.log("Connected MCP client to backing server transport");
 
     const webAppTransport = new SSEServerTransport("/message", res);
-    webAppTransports.push(webAppTransport);
-
+    webAppTransports.set(webAppTransport.sessionId, webAppTransport);
     console.log("Created web app transport");
 
     await webAppTransport.start();
@@ -299,7 +318,9 @@ app.post("/message", async (req, res) => {
     const sessionId = req.query.sessionId;
     console.log(`Received message for sessionId ${sessionId}`);
 
-    const transport = webAppTransports.find((t) => t.sessionId === sessionId);
+    const transport = webAppTransports.get(
+      sessionId as string,
+    ) as SSEServerTransport;
     if (!transport) {
       res.status(404).end("Session not found");
       return;