feat: support streamable http transport

package.json

@@ -34,7 +34,7 @@
     }
   },
   "dependencies": {
-    "@modelcontextprotocol/sdk": "^1.6.1",
+    "@modelcontextprotocol/sdk": "^1.10.1",
     "commander": "^13.1.0",
     "playwright": "1.53.0-alpha-1745357020000",
     "yaml": "^2.7.1",

src/program.ts

@@ -14,18 +14,13 @@
  * limitations under the License.
  */
 
-import http from 'http';
-
 import { program } from 'commander';
-import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
-import { SSEServerTransport } from '@modelcontextprotocol/sdk/server/sse.js';
-
 
 import { createServer } from './index';
 import { ServerList } from './server';
 
-import assert from 'assert';
 import { ToolCapability } from './tools/tool';
+import { startHttpTransport, startStdioTransport } from './transport';
 
 const packageJSON = require('../package.json');
 
@@ -53,12 +48,10 @@ program
       }));
       setupExitWatchdog(serverList);
 
-      if (options.port) {
-        startSSEServer(+options.port, options.host || 'localhost', serverList);
-      } else {
-        const server = await serverList.create();
-        await server.connect(new StdioServerTransport());
-      }
+      if (options.port)
+        startHttpTransport(+options.port, options.host, serverList);
+      else
+        await startStdioTransport(serverList);
     });
 
 function setupExitWatchdog(serverList: ServerList) {
@@ -74,64 +67,3 @@ function setupExitWatchdog(serverList: ServerList) {
 }
 
 program.parse(process.argv);
-
-function startSSEServer(port: number, host: string, serverList: ServerList) {
-  const sessions = new Map<string, SSEServerTransport>();
-  const httpServer = http.createServer(async (req, res) => {
-    if (req.method === 'POST') {
-      const searchParams = new URL(`http://localhost${req.url}`).searchParams;
-      const sessionId = searchParams.get('sessionId');
-      if (!sessionId) {
-        res.statusCode = 400;
-        res.end('Missing sessionId');
-        return;
-      }
-      const transport = sessions.get(sessionId);
-      if (!transport) {
-        res.statusCode = 404;
-        res.end('Session not found');
-        return;
-      }
-
-      await transport.handlePostMessage(req, res);
-      return;
-    } else if (req.method === 'GET') {
-      const transport = new SSEServerTransport('/sse', res);
-      sessions.set(transport.sessionId, transport);
-      const server = await serverList.create();
-      res.on('close', () => {
-        sessions.delete(transport.sessionId);
-        serverList.close(server).catch(e => console.error(e));
-      });
-      await server.connect(transport);
-      return;
-    } else {
-      res.statusCode = 405;
-      res.end('Method not allowed');
-    }
-  });
-
-  httpServer.listen(port, host, () => {
-    const address = httpServer.address();
-    assert(address, 'Could not bind server socket');
-    let url: string;
-    if (typeof address === 'string') {
-      url = address;
-    } else {
-      const resolvedPort = address.port;
-      let resolvedHost = address.family === 'IPv4' ? address.address : `[${address.address}]`;
-      if (resolvedHost === '0.0.0.0' || resolvedHost === '[::]')
-        resolvedHost = host === 'localhost' ? 'localhost' : resolvedHost;
-      url = `http://${resolvedHost}:${resolvedPort}`;
-    }
-    console.log(`Listening on ${url}`);
-    console.log('Put this in your client config:');
-    console.log(JSON.stringify({
-      'mcpServers': {
-        'playwright': {
-          'url': `${url}/sse`
-        }
-      }
-    }, undefined, 2));
-  });
-}

src/transport.ts

@@ -0,0 +1,127 @@
+/**
+ * Copyright (c) Microsoft Corporation.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import http from 'node:http';
+import assert from 'node:assert';
+import crypto from 'node:crypto';
+
+import { ServerList } from './server';
+import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
+import { SSEServerTransport } from '@modelcontextprotocol/sdk/server/sse.js';
+import { StreamableHTTPServerTransport } from '@modelcontextprotocol/sdk/server/streamableHttp.js';
+
+export async function startStdioTransport(serverList: ServerList) {
+  const server = await serverList.create();
+  await server.connect(new StdioServerTransport());
+}
+
+async function handleSSE(req: http.IncomingMessage, res: http.ServerResponse, url: URL, serverList: ServerList, sessions: Map<string, SSEServerTransport>) {
+  if (req.method === 'POST') {
+    const sessionId = url.searchParams.get('sessionId');
+    if (!sessionId) {
+      res.statusCode = 400;
+      return res.end('Missing sessionId');
+    }
+
+    const transport = sessions.get(sessionId);
+    if (!transport) {
+      res.statusCode = 404;
+      return res.end('Session not found');
+    }
+
+    return await transport.handlePostMessage(req, res);
+  } else if (req.method === 'GET') {
+    const transport = new SSEServerTransport('/sse', res);
+    sessions.set(transport.sessionId, transport);
+    const server = await serverList.create();
+    res.on('close', () => {
+      sessions.delete(transport.sessionId);
+      serverList.close(server).catch(e => console.error(e));
+    });
+    return await server.connect(transport);
+  }
+
+  res.statusCode = 405;
+  res.end('Method not allowed');
+}
+
+async function handleStreamable(req: http.IncomingMessage, res: http.ServerResponse, serverList: ServerList, sessions: Map<string, StreamableHTTPServerTransport>) {
+  const sessionId = req.headers['mcp-session-id'] as string | undefined;
+  if (sessionId) {
+    const transport = sessions.get(sessionId);
+    if (!transport) {
+      res.statusCode = 404;
+      res.end('Session not found');
+      return;
+    }
+    return await transport.handleRequest(req, res);
+  }
+
+  if (req.method === 'POST') {
+    const transport = new StreamableHTTPServerTransport({
+      sessionIdGenerator: () => crypto.randomUUID(),
+      onsessioninitialized: sessionId => {
+        sessions.set(sessionId, transport);
+      }
+    });
+    transport.onclose = () => {
+      if (transport.sessionId)
+        sessions.delete(transport.sessionId);
+    };
+    const server = await serverList.create();
+    await server.connect(transport);
+    return await transport.handleRequest(req, res);
+  }
+
+  res.statusCode = 400;
+  res.end('Invalid request');
+}
+
+export function startHttpTransport(port: number, hostname: string | undefined, serverList: ServerList) {
+  const sseSessions = new Map<string, SSEServerTransport>();
+  const streamableSessions = new Map<string, StreamableHTTPServerTransport>();
+  const httpServer = http.createServer(async (req, res) => {
+    const url = new URL(`http://localhost${req.url}`);
+    if (url.pathname.startsWith('/mcp'))
+      await handleStreamable(req, res, serverList, streamableSessions);
+    else
+      await handleSSE(req, res, url, serverList, sseSessions);
+  });
+  httpServer.listen(port, hostname, () => {
+    const address = httpServer.address();
+    assert(address, 'Could not bind server socket');
+    let url: string;
+    if (typeof address === 'string') {
+      url = address;
+    } else {
+      const resolvedPort = address.port;
+      let resolvedHost = address.family === 'IPv4' ? address.address : `[${address.address}]`;
+      if (resolvedHost === '0.0.0.0' || resolvedHost === '[::]')
+        resolvedHost = 'localhost';
+      url = `http://${resolvedHost}:${resolvedPort}`;
+    }
+    console.log(`Listening on ${url}`);
+    console.log('Put this in your client config:');
+    console.log(JSON.stringify({
+      'mcpServers': {
+        'playwright': {
+          'url': `${url}/sse`
+        }
+      }
+    }, undefined, 2));
+    console.log('If your client supports streamable HTTP, you can use the /mcp endpoint instead.');
+  });
+}

tests/sse.spec.ts

@@ -16,27 +16,45 @@
 
 import { spawn } from 'node:child_process';
 import path from 'node:path';
-import { test } from './fixtures';
+import { test as baseTest } from './fixtures';
+import { expect } from 'playwright/test';
 
-test('sse transport', async () => {
-  const cp = spawn('node', [path.join(__dirname, '../cli.js'), '--port', '0'], { stdio: 'pipe' });
-  try {
-    let stdout = '';
-    const url = await new Promise<string>(resolve => cp.stdout?.on('data', data => {
-      stdout += data.toString();
-      const match = stdout.match(/Listening on (http:\/\/.*)/);
-      if (match)
-        resolve(match[1]);
-    }));
+const test = baseTest.extend<{ serverEndpoint: string }>({
+  serverEndpoint: async ({}, use) => {
+    const cp = spawn('node', [path.join(__dirname, '../cli.js'), '--port', '0'], { stdio: 'pipe' });
+    try {
+      let stdout = '';
+      const url = await new Promise<string>(resolve => cp.stdout?.on('data', data => {
+        stdout += data.toString();
+        const match = stdout.match(/Listening on (http:\/\/.*)/);
+        if (match)
+          resolve(match[1]);
+      }));
 
-    // need dynamic import b/c of some ESM nonsense
-    const { SSEClientTransport } = await import('@modelcontextprotocol/sdk/client/sse.js');
-    const { Client } = await import('@modelcontextprotocol/sdk/client/index.js');
-    const transport = new SSEClientTransport(new URL(url));
-    const client = new Client({ name: 'test', version: '1.0.0' });
-    await client.connect(transport);
-    await client.ping();
-  } finally {
-    cp.kill();
-  }
+      await use(url);
+    } finally {
+      cp.kill();
+    }
+  },
+});
+
+test('sse transport', async ({ serverEndpoint }) => {
+  // need dynamic import b/c of some ESM nonsense
+  const { SSEClientTransport } = await import('@modelcontextprotocol/sdk/client/sse.js');
+  const { Client } = await import('@modelcontextprotocol/sdk/client/index.js');
+  const transport = new SSEClientTransport(new URL(serverEndpoint));
+  const client = new Client({ name: 'test', version: '1.0.0' });
+  await client.connect(transport);
+  await client.ping();
+});
+
+test('streamable http transport', async ({ serverEndpoint }) => {
+  // need dynamic import b/c of some ESM nonsense
+  const { StreamableHTTPClientTransport } = await import('@modelcontextprotocol/sdk/client/streamableHttp.js');
+  const { Client } = await import('@modelcontextprotocol/sdk/client/index.js');
+  const transport = new StreamableHTTPClientTransport(new URL('/mcp', serverEndpoint));
+  const client = new Client({ name: 'test', version: '1.0.0' });
+  await client.connect(transport);
+  await client.ping();
+  expect(transport.sessionId, 'has session support').toBeDefined();
 });