support sse though get

src/examples/client/simpleStreamableHttp.ts

@@ -11,7 +11,8 @@ import {
   GetPromptResultSchema,
   ListResourcesRequest,
   ListResourcesResultSchema,
-  LoggingMessageNotificationSchema
+  LoggingMessageNotificationSchema,
+  ResourceListChangedNotificationSchema
 } from '../../types.js';
 
 async function main(): Promise<void> {
@@ -24,51 +25,80 @@ async function main(): Promise<void> {
   const transport = new StreamableHTTPClientTransport(
     new URL('http://localhost:3000/mcp')
   );
+  const supportsStandaloneSse = false;
 
   // Connect the client using the transport and initialize the server
   await client.connect(transport);
+  console.log('Connected to MCP server');
+  // Open a standalone SSE stream to receive server-initiated messages
+  console.log('Opening SSE stream to receive server notifications...');
+  try {
+    await transport.openSseStream();
+    const supportsStandaloneSse = false;
+    console.log('SSE stream established successfully. Waiting for notifications...');
+  }
+  catch (error) {
+    console.error('Failed to open SSE stream:', error);
+  }
+
+  // Set up notification handlers for server-initiated messages
   client.setNotificationHandler(LoggingMessageNotificationSchema, (notification) => {
     console.log(`Notification received: ${notification.params.level} - ${notification.params.data}`);
   });
-
-
-  console.log('Connected to MCP server');
-  // List available tools
-  const toolsRequest: ListToolsRequest = {
-    method: 'tools/list',
-    params: {}
-  };
-  const toolsResult = await client.request(toolsRequest, ListToolsResultSchema);
-  console.log('Available tools:', toolsResult.tools);
-
-  // Call the 'greet' tool
-  const greetRequest: CallToolRequest = {
-    method: 'tools/call',
-    params: {
-      name: 'greet',
-      arguments: { name: 'MCP User' }
-    }
-  };
-  const greetResult = await client.request(greetRequest, CallToolResultSchema);
-  console.log('Greeting result:', greetResult.content[0].text);
-
-  // Call the new 'multi-greet' tool
-  console.log('\nCalling multi-greet tool (with notifications)...');
-  const multiGreetRequest: CallToolRequest = {
-    method: 'tools/call',
-    params: {
-      name: 'multi-greet',
-      arguments: { name: 'MCP User' }
-    }
-  };
-  const multiGreetResult = await client.request(multiGreetRequest, CallToolResultSchema);
-  console.log('Multi-greet results:');
-  multiGreetResult.content.forEach(item => {
-    if (item.type === 'text') {
-      console.log(`- ${item.text}`);
-    }
+  client.setNotificationHandler(ResourceListChangedNotificationSchema, async (_) => {
+    console.log(`Resource list changed notification received!`);
+    const resourcesRequest: ListResourcesRequest = {
+      method: 'resources/list',
+      params: {}
+    };
+    const resourcesResult = await client.request(resourcesRequest, ListResourcesResultSchema);
+    console.log('Available resources count:', resourcesResult.resources.length);
   });
 
+  // List available tools
+  try {
+    const toolsRequest: ListToolsRequest = {
+      method: 'tools/list',
+      params: {}
+    };
+    const toolsResult = await client.request(toolsRequest, ListToolsResultSchema);
+    console.log('Available tools:', toolsResult.tools);
+
+    if (toolsResult.tools.length === 0) {
+      console.log('No tools available from the server');
+    } else {
+      // Call the 'greet' tool
+      const greetRequest: CallToolRequest = {
+        method: 'tools/call',
+        params: {
+          name: 'greet',
+          arguments: { name: 'MCP User' }
+        }
+      };
+      const greetResult = await client.request(greetRequest, CallToolResultSchema);
+      console.log('Greeting result:', greetResult.content[0].text);
+
+      // Call the new 'multi-greet' tool
+      console.log('\nCalling multi-greet tool (with notifications)...');
+      const multiGreetRequest: CallToolRequest = {
+        method: 'tools/call',
+        params: {
+          name: 'multi-greet',
+          arguments: { name: 'MCP User' }
+        }
+      };
+      const multiGreetResult = await client.request(multiGreetRequest, CallToolResultSchema);
+      console.log('Multi-greet results:');
+      multiGreetResult.content.forEach(item => {
+        if (item.type === 'text') {
+          console.log(`- ${item.text}`);
+        }
+      });
+    }
+  } catch (error) {
+    console.log(`Tools not supported by this server (${error})`);
+  }
+
   // List available prompts
   try {
     const promptsRequest: ListPromptsRequest = {
@@ -107,9 +137,11 @@ async function main(): Promise<void> {
   } catch (error) {
     console.log(`Resources not supported by this server (${error})`);
   }
+  if (supportsStandaloneSse) {
+    // Instead of closing immediately, keep the connection open to receive notifications
+    console.log('\nKeeping connection open to receive notifications. Press Ctrl+C to exit.');
+  }
 
-  // Close the connection
-  await client.close();
 }
 
 main().catch((error: unknown) => {

src/examples/server/standaloneSseWithGetStreamableHttp.ts

@@ -0,0 +1,136 @@
+import express, { Request, Response } from 'express';
+import { randomUUID } from 'node:crypto';
+import { McpServer } from '../../server/mcp.js';
+import { StreamableHTTPServerTransport } from '../../server/streamableHttp.js';
+import { ReadResourceResult } from '../../types.js';
+
+// Create an MCP server with implementation details
+const server = new McpServer({
+  name: 'resource-list-changed-notification-server',
+  version: '1.0.0',
+}, {
+  capabilities: {
+    resources: {
+      listChanged: true, // Support notifications for resource list changes
+    },
+  }
+});
+
+// Store transports by session ID to send notifications
+const transports: { [sessionId: string]: StreamableHTTPServerTransport } = {};
+
+const addResource = (name: string, content: string) => {
+  const uri = `https://mcp-example.com/dynamic/${encodeURIComponent(name)}`;
+  server.resource(
+    name,
+    uri,
+    { mimeType: 'text/plain', description: `Dynamic resource: ${name}` },
+    async (): Promise<ReadResourceResult> => {
+      return {
+        contents: [{ uri, text: content }],
+      };
+    }
+  );
+
+};
+
+addResource('example-resource', 'Initial content for example-resource');
+
+const resourceChangeInterval = setInterval(() => {
+  const name = randomUUID();
+  addResource(name, `Content for ${name}`);
+}, 5000); // Change resources every 5 seconds for testing
+
+const app = express();
+app.use(express.json());
+
+app.post('/mcp', async (req: Request, res: Response) => {
+  console.log('Received MCP request:', req.body);
+  try {
+    // Check for existing session ID
+    const sessionId = req.headers['mcp-session-id'] as string | undefined;
+    let transport: StreamableHTTPServerTransport;
+
+    if (sessionId && transports[sessionId]) {
+      // Reuse existing transport
+      transport = transports[sessionId];
+    } else if (!sessionId && isInitializeRequest(req.body)) {
+      // New initialization request
+      transport = new StreamableHTTPServerTransport({
+        sessionIdGenerator: () => randomUUID(),
+      });
+
+      // Connect the transport to the MCP server
+      await server.connect(transport);
+
+      await transport.handleRequest(req, res, req.body);
+
+      // Store the transport by session ID for future requests
+      if (transport.sessionId) {
+        transports[transport.sessionId] = transport;
+      }
+      return; // Already handled
+    } else {
+      // Invalid request - no session ID or not initialization request
+      res.status(400).json({
+        jsonrpc: '2.0',
+        error: {
+          code: -32000,
+          message: 'Bad Request: No valid session ID provided',
+        },
+        id: null,
+      });
+      return;
+    }
+
+    // Handle the request with existing transport
+    await transport.handleRequest(req, res, req.body);
+  } catch (error) {
+    console.error('Error handling MCP request:', error);
+    if (!res.headersSent) {
+      res.status(500).json({
+        jsonrpc: '2.0',
+        error: {
+          code: -32603,
+          message: 'Internal server error',
+        },
+        id: null,
+      });
+    }
+  }
+});
+
+// Handle GET requests for SSE streams (now using built-in support from StreamableHTTP)
+app.get('/mcp', async (req: Request, res: Response) => {
+  const sessionId = req.headers['mcp-session-id'] as string | undefined;
+  if (!sessionId || !transports[sessionId]) {
+    res.status(400).send('Invalid or missing session ID');
+    return;
+  }
+
+  console.log(`Establishing SSE stream for session ${sessionId}`);
+  const transport = transports[sessionId];
+  await transport.handleRequest(req, res);
+});
+
+// Helper function to detect initialize requests
+function isInitializeRequest(body: unknown): boolean {
+  if (Array.isArray(body)) {
+    return body.some(msg => typeof msg === 'object' && msg !== null && 'method' in msg && msg.method === 'initialize');
+  }
+  return typeof body === 'object' && body !== null && 'method' in body && body.method === 'initialize';
+}
+
+// Start the server
+const PORT = 3000;
+app.listen(PORT, () => {
+  console.log(`Server listening on port ${PORT}`);
+});
+
+// Handle server shutdown
+process.on('SIGINT', async () => {
+  console.log('Shutting down server...');
+  clearInterval(resourceChangeInterval);
+  await server.close();
+  process.exit(0);
+});

src/server/streamableHttp.test.ts

@@ -564,22 +564,123 @@ describe("StreamableHTTPServerTransport", () => {
       mockResponse.writeHead.mockClear();
     });
 
-    it("should reject GET requests for SSE with 405 Method Not Allowed", async () => {
+    it("should accept GET requests for SSE with proper Accept header", async () => {
       const req = createMockRequest({
         method: "GET",
         headers: {
-          "accept": "application/json, text/event-stream",
+          "accept": "text/event-stream",
           "mcp-session-id": transport.sessionId,
         },
       });
 
       await transport.handleRequest(req, mockResponse);
 
-      expect(mockResponse.writeHead).toHaveBeenCalledWith(405, expect.objectContaining({
-        "Allow": "POST, DELETE"
+      expect(mockResponse.writeHead).toHaveBeenCalledWith(200, expect.objectContaining({
+        "Content-Type": "text/event-stream",
+        "Cache-Control": "no-cache",
+        "Connection": "keep-alive",
+        "mcp-session-id": transport.sessionId,
       }));
-      expect(mockResponse.end).toHaveBeenCalledWith(expect.stringContaining('"jsonrpc":"2.0"'));
-      expect(mockResponse.end).toHaveBeenCalledWith(expect.stringContaining('Method not allowed'));
+    });
+
+    it("should reject GET requests without Accept: text/event-stream header", async () => {
+      const req = createMockRequest({
+        method: "GET",
+        headers: {
+          "accept": "application/json",
+          "mcp-session-id": transport.sessionId,
+        },
+      });
+
+      await transport.handleRequest(req, mockResponse);
+
+      expect(mockResponse.writeHead).toHaveBeenCalledWith(406);
+      expect(mockResponse.end).toHaveBeenCalledWith(expect.stringContaining('"message":"Not Acceptable: Client must accept text/event-stream"'));
+    });
+
+    it("should send server-initiated requests to GET SSE stream", async () => {
+      // Open a standalone SSE stream with GET
+      const req = createMockRequest({
+        method: "GET",
+        headers: {
+          "accept": "text/event-stream",
+          "mcp-session-id": transport.sessionId,
+        },
+      });
+
+      const sseResponse = createMockResponse();
+      await transport.handleRequest(req, sseResponse);
+
+      // Send a notification without a related request ID
+      const notification: JSONRPCMessage = {
+        jsonrpc: "2.0",
+        method: "notifications/resources/updated",
+        params: { uri: "someuri" }
+      };
+
+      await transport.send(notification);
+
+      // Verify notification was sent on SSE stream
+      expect(sseResponse.write).toHaveBeenCalledWith(
+        expect.stringContaining(`event: message\ndata: ${JSON.stringify(notification)}\n\n`)
+      );
+    });
+
+    it("should not close GET SSE stream after sending server requests or notifications", async () => {
+      // Open a standalone SSE stream
+      const req = createMockRequest({
+        method: "GET",
+        headers: {
+          "accept": "text/event-stream",
+          "mcp-session-id": transport.sessionId,
+        },
+      });
+
+      const sseResponse = createMockResponse();
+      await transport.handleRequest(req, sseResponse);
+
+      // Send multiple notifications
+      const notification1: JSONRPCMessage = { jsonrpc: "2.0", method: "event1", params: {} };
+      const notification2: JSONRPCMessage = { jsonrpc: "2.0", method: "event2", params: {} };
+
+      await transport.send(notification1);
+      await transport.send(notification2);
+
+      // Stream should remain open
+      expect(sseResponse.end).not.toHaveBeenCalled();
+    });
+
+    it("should reject second GET SSE stream for the same session", async () => {
+      // Open first SSE stream - should succeed
+      const req1 = createMockRequest({
+        method: "GET",
+        headers: {
+          "accept": "text/event-stream",
+          "mcp-session-id": transport.sessionId,
+        },
+      });
+
+      const sseResponse1 = createMockResponse();
+      await transport.handleRequest(req1, sseResponse1);
+
+      // Try to open a second SSE stream - should be rejected
+      const req2 = createMockRequest({
+        method: "GET",
+        headers: {
+          "accept": "text/event-stream",
+          "mcp-session-id": transport.sessionId,
+        },
+      });
+
+      const sseResponse2 = createMockResponse();
+      await transport.handleRequest(req2, sseResponse2);
+
+      // First stream should be good
+      expect(sseResponse1.writeHead).toHaveBeenCalledWith(200, expect.anything());
+
+      // Second stream should get 409 Conflict
+      expect(sseResponse2.writeHead).toHaveBeenCalledWith(409);
+      expect(sseResponse2.end).toHaveBeenCalledWith(expect.stringContaining('"message":"Conflict: Only one SSE stream is allowed per session"'));
     });
 
     it("should reject POST requests without proper Accept header", async () => {

src/server/streamableHttp.ts

@@ -68,6 +68,7 @@ export class StreamableHTTPServerTransport implements Transport {
   private _requestResponseMap: Map<RequestId, JSONRPCMessage> = new Map();
   private _initialized: boolean = false;
   private _enableJsonResponse: boolean = false;
+  private _sseStreamKey = "standalone-sse";
 
 
   sessionId?: string | undefined;
@@ -97,6 +98,8 @@ export class StreamableHTTPServerTransport implements Transport {
   async handleRequest(req: IncomingMessage, res: ServerResponse, parsedBody?: unknown): Promise<void> {
     if (req.method === "POST") {
       await this.handlePostRequest(req, res, parsedBody);
+    } else if (req.method === "GET") {
+      await this.handleGetRequest(req, res);
     } else if (req.method === "DELETE") {
       await this.handleDeleteRequest(req, res);
     } else {
@@ -105,12 +108,80 @@ export class StreamableHTTPServerTransport implements Transport {
   }
 
   /**
-   * Handles unsupported requests (GET, PUT, PATCH, etc.)
-   * For now we support only POST and DELETE requests. Support for GET for SSE connections will be added later.
+   * Handles GET requests for SSE stream
+   */
+  private async handleGetRequest(req: IncomingMessage, res: ServerResponse): Promise<void> {
+    // The client MUST include an Accept header, listing text/event-stream as a supported content type.
+    const acceptHeader = req.headers.accept;
+    if (!acceptHeader?.includes("text/event-stream")) {
+      res.writeHead(406).end(JSON.stringify({
+        jsonrpc: "2.0",
+        error: {
+          code: -32000,
+          message: "Not Acceptable: Client must accept text/event-stream"
+        },
+        id: null
+      }));
+      return;
+    }
+
+    // If an Mcp-Session-Id is returned by the server during initialization,
+    // clients using the Streamable HTTP transport MUST include it 
+    // in the Mcp-Session-Id header on all of their subsequent HTTP requests.
+    if (!this.validateSession(req, res)) {
+      return;
+    }
+
+    // The server MUST either return Content-Type: text/event-stream in response to this HTTP GET, 
+    // or else return HTTP 405 Method Not Allowed
+    const headers: Record<string, string> = {
+      "Content-Type": "text/event-stream",
+      "Cache-Control": "no-cache",
+      Connection: "keep-alive",
+    };
+
+    // After initialization, always include the session ID if we have one
+    if (this.sessionId !== undefined) {
+      headers["mcp-session-id"] = this.sessionId;
+    }
+    // The server MAY include a Last-Event-ID header in the response to this HTTP GET.
+    // Resumability will be supported in the future
+
+    // Check if there's already an active standalone SSE stream for this session
+    const existingStream = this._responseMapping.get(this._sseStreamKey);
+
+    if (existingStream !== undefined) {
+      // Only one GET SSE stream is allowed per session
+      res.writeHead(409).end(JSON.stringify({
+        jsonrpc: "2.0",
+        error: {
+          code: -32000,
+          message: "Conflict: Only one SSE stream is allowed per session"
+        },
+        id: null
+      }));
+      return;
+    }
+
+    res.writeHead(200, headers);
+
+    // Store the response for this request so we can use it for standalone server notifications
+    // This response doesn't have an associated request ID, so we'll use a special string to track it
+    this._responseMapping.set(this._sseStreamKey, res);
+
+    // Set up close handler for client disconnects
+    res.on("close", () => {
+      // Clean up resources associated with this connection
+      this._responseMapping.delete(this._sseStreamKey);
+    });
+  }
+
+  /**
+   * Handles unsupported requests (PUT, PATCH, etc.)
    */
   private async handleUnsupportedRequest(res: ServerResponse): Promise<void> {
     res.writeHead(405, {
-      "Allow": "POST, DELETE"
+      "Allow": "GET, POST, DELETE"
     }).end(JSON.stringify({
       jsonrpc: "2.0",
       error: {
@@ -379,8 +450,25 @@ export class StreamableHTTPServerTransport implements Transport {
       // If the message is a response, use the request ID from the message
       requestId = message.id;
     }
+
+    // Check if this message should be sent on the standalone SSE stream (no request ID)
+    // Ignore notifications from tools (which have relatedRequestId set)
+    // Those will be sent via dedicated response SSE streams
     if (requestId === undefined) {
-      throw new Error("No request ID provided for the message");
+      // For standalone SSE streams, we can only send requests and notifications
+      if ('result' in message || 'error' in message) {
+        throw new Error("Cannot send a response on a standalone SSE stream unless resuming a previous client request");
+      }
+
+      const standaloneStream = this._responseMapping.get(this._sseStreamKey);
+      if (standaloneStream === undefined) {
+        // The spec says the server MAY send messages on the stream, so it's ok to discard if no stream
+        return;
+      }
+
+      // Send the message to the standalone SSE stream
+      standaloneStream.write(`event: message\ndata: ${JSON.stringify(message)}\n\n`);
+      return;
     }
 
     // Get the response for this request
@@ -389,7 +477,6 @@ export class StreamableHTTPServerTransport implements Transport {
       throw new Error(`No connection established for request ID: ${String(requestId)}`);
     }
 
-
     if (!this._enableJsonResponse) {
       response.write(`event: message\ndata: ${JSON.stringify(message)}\n\n`);
     }