feat: add closeSseStream API for POST SSE polling (SEP-1699)

- Remove priming from GET streams (only POST streams per spec)
- Replace closeStandaloneSseStream() with closeSseStream(requestId)
- Add validation that retryInterval requires eventStore
- Add tests for POST SSE priming events and stream closure
- Restore original resumability tests without priming expectations

Author: felixweinberger

src/server/streamableHttp.test.ts

@@ -1361,11 +1361,6 @@ describe('StreamableHTTPServerTransport with resumability', () => {
         // Read from the stream
         const reader = sseResponse.body?.getReader();
 
-        // Read and validate the priming event
-        const { value: primingValue } = await reader!.read();
-        const primingText = new TextDecoder().decode(primingValue);
-        expect(primingText).toContain('id: ');
-
         // Send a notification that should be stored with an event ID
         const notification: JSONRPCMessage = {
             jsonrpc: '2.0',
@@ -1413,12 +1408,6 @@ describe('StreamableHTTPServerTransport with resumability', () => {
 
         // Read the notification from the SSE stream
         const reader = sseResponse.body?.getReader();
-
-        // Read and validate the priming event
-        const { value: primingValue } = await reader!.read();
-        const primingText = new TextDecoder().decode(primingValue);
-        expect(primingText).toContain('id: ');
-
         const { value } = await reader!.read();
         const text = new TextDecoder().decode(value);
 
@@ -1549,29 +1538,76 @@ describe('StreamableHTTPServerTransport in stateless mode', () => {
     });
 });
 
-// Test SSE priming events
-describe('StreamableHTTPServerTransport SSE priming events', () => {
+// Test retryInterval validation
+describe('StreamableHTTPServerTransport retryInterval validation', () => {
+    it('should throw error when retryInterval is configured without eventStore', () => {
+        expect(() => {
+            new StreamableHTTPServerTransport({
+                sessionIdGenerator: () => randomUUID(),
+                retryInterval: 5000
+                // No eventStore
+            });
+        }).toThrow('retryInterval requires eventStore to be configured for resumability');
+    });
+
+    it('should not throw when retryInterval is configured with eventStore', async () => {
+        const storedEvents = new Map<string, { eventId: string; message: JSONRPCMessage }>();
+        const eventStore: EventStore = {
+            async storeEvent(streamId: string, message: JSONRPCMessage): Promise<string> {
+                const eventId = `${streamId}_${randomUUID()}`;
+                storedEvents.set(eventId, { eventId, message });
+                return eventId;
+            },
+            async replayEventsAfter(
+                lastEventId: EventId,
+                { send: _send }: { send: (eventId: EventId, message: JSONRPCMessage) => Promise<void> }
+            ): Promise<StreamId> {
+                const streamId = lastEventId.split('_')[0];
+                return streamId;
+            }
+        };
+
+        expect(() => {
+            new StreamableHTTPServerTransport({
+                sessionIdGenerator: () => randomUUID(),
+                eventStore,
+                retryInterval: 5000
+            });
+        }).not.toThrow();
+    });
+});
+
+// Test SSE priming events for POST streams (SEP-1699)
+describe('StreamableHTTPServerTransport POST SSE priming events', () => {
     let server: Server;
     let transport: StreamableHTTPServerTransport;
     let baseUrl: URL;
     let sessionId: string;
+    let mcpServer: McpServer;
 
     // Simple eventStore for priming event tests
     const createEventStore = (): EventStore => {
-        const storedEvents = new Map<string, { eventId: string; message: JSONRPCMessage }>();
+        const storedEvents = new Map<string, { eventId: string; message: JSONRPCMessage; streamId: string }>();
         return {
             async storeEvent(streamId: string, message: JSONRPCMessage): Promise<string> {
-                const eventId = `${streamId}_${randomUUID()}`;
-                storedEvents.set(eventId, { eventId, message });
+                const eventId = `${streamId}::${Date.now()}_${randomUUID()}`;
+                storedEvents.set(eventId, { eventId, message, streamId });
                 return eventId;
             },
             async replayEventsAfter(
                 lastEventId: EventId,
                 { send }: { send: (eventId: EventId, message: JSONRPCMessage) => Promise<void> }
             ): Promise<StreamId> {
-                const streamId = lastEventId.split('_')[0];
-                for (const [eventId, { message }] of storedEvents.entries()) {
-                    if (eventId.startsWith(streamId) && eventId !== lastEventId) {
+                const streamId = lastEventId.split('::')[0];
+                const eventsToReplay: Array<[string, { message: JSONRPCMessage }]> = [];
+                for (const [eventId, data] of storedEvents.entries()) {
+                    if (data.streamId === streamId && eventId > lastEventId) {
+                        eventsToReplay.push([eventId, data]);
+                    }
+                }
+                eventsToReplay.sort(([a], [b]) => a.localeCompare(b));
+                for (const [eventId, { message }] of eventsToReplay) {
+                    if (Object.keys(message).length > 0) {
                         await send(eventId, message);
                     }
                 }
@@ -1586,7 +1622,7 @@ describe('StreamableHTTPServerTransport SSE priming events', () => {
         }
     });
 
-    it('should send priming event with retry field when retryInterval is configured', async () => {
+    it('should send priming event with retry field on POST SSE stream', async () => {
         const result = await createTestServer({
             sessionIdGenerator: () => randomUUID(),
             eventStore: createEventStore(),
@@ -1595,27 +1631,42 @@ describe('StreamableHTTPServerTransport SSE priming events', () => {
         server = result.server;
         transport = result.transport;
         baseUrl = result.baseUrl;
+        mcpServer = result.mcpServer;
+
+        // Register a tool that we can call
+        mcpServer.tool('test-tool', 'A test tool', {}, async () => {
+            return { content: [{ type: 'text', text: 'Tool result' }] };
+        });
 
         // Initialize to get session ID
         const initResponse = await sendPostRequest(baseUrl, TEST_MESSAGES.initialize);
         sessionId = initResponse.headers.get('mcp-session-id') as string;
         expect(sessionId).toBeDefined();
 
-        // Open a standalone SSE stream
-        const sseResponse = await fetch(baseUrl, {
-            method: 'GET',
+        // Send a POST request that will return SSE stream
+        const toolCallRequest: JSONRPCMessage = {
+            jsonrpc: '2.0',
+            id: 1,
+            method: 'tools/call',
+            params: { name: 'test-tool', arguments: {} }
+        };
+
+        const postResponse = await fetch(baseUrl, {
+            method: 'POST',
             headers: {
-                Accept: 'text/event-stream',
+                'Content-Type': 'application/json',
+                Accept: 'text/event-stream, application/json',
                 'mcp-session-id': sessionId,
                 'mcp-protocol-version': '2025-03-26'
-            }
+            },
+            body: JSON.stringify(toolCallRequest)
         });
 
-        expect(sseResponse.status).toBe(200);
-        expect(sseResponse.headers.get('content-type')).toBe('text/event-stream');
+        expect(postResponse.status).toBe(200);
+        expect(postResponse.headers.get('content-type')).toBe('text/event-stream');
 
         // Read the priming event
-        const reader = sseResponse.body?.getReader();
+        const reader = postResponse.body?.getReader();
         const { value } = await reader!.read();
         const text = new TextDecoder().decode(value);
 
@@ -1634,91 +1685,111 @@ describe('StreamableHTTPServerTransport SSE priming events', () => {
         server = result.server;
         transport = result.transport;
         baseUrl = result.baseUrl;
+        mcpServer = result.mcpServer;
+
+        mcpServer.tool('test-tool', 'A test tool', {}, async () => {
+            return { content: [{ type: 'text', text: 'Tool result' }] };
+        });
 
         // Initialize to get session ID
         const initResponse = await sendPostRequest(baseUrl, TEST_MESSAGES.initialize);
         sessionId = initResponse.headers.get('mcp-session-id') as string;
         expect(sessionId).toBeDefined();
 
-        // Open a standalone SSE stream
-        const sseResponse = await fetch(baseUrl, {
-            method: 'GET',
+        // Send a POST request
+        const toolCallRequest: JSONRPCMessage = {
+            jsonrpc: '2.0',
+            id: 1,
+            method: 'tools/call',
+            params: { name: 'test-tool', arguments: {} }
+        };
+
+        const postResponse = await fetch(baseUrl, {
+            method: 'POST',
             headers: {
-                Accept: 'text/event-stream',
+                'Content-Type': 'application/json',
+                Accept: 'text/event-stream, application/json',
                 'mcp-session-id': sessionId,
                 'mcp-protocol-version': '2025-03-26'
-            }
+            },
+            body: JSON.stringify(toolCallRequest)
         });
 
-        expect(sseResponse.status).toBe(200);
+        expect(postResponse.status).toBe(200);
 
         // Read the priming event
-        const reader = sseResponse.body?.getReader();
-        const { value: primingValue } = await reader!.read();
-        const primingText = new TextDecoder().decode(primingValue);
-
-        // Priming event should have id field for resumability but NOT retry field
-        expect(primingText).toContain('id: ');
-        expect(primingText).toContain('data: ');
-        expect(primingText).not.toContain('retry:');
-
-        // Verify priming event has ID
-        expect(primingText.match(/id: ([^\n]+)/)).toBeTruthy();
-
-        // Send a notification
-        const notification: JSONRPCMessage = {
-            jsonrpc: '2.0',
-            method: 'notifications/message',
-            params: { level: 'info', data: 'Test notification' }
-        };
-        await transport.send(notification);
-
-        // Read the notification
-        const { value: notifValue } = await reader!.read();
-        const notifText = new TextDecoder().decode(notifValue);
+        const reader = postResponse.body?.getReader();
+        const { value } = await reader!.read();
+        const text = new TextDecoder().decode(value);
 
-        // Notification should have content and no retry field
-        expect(notifText).toContain('Test notification');
-        expect(notifText).not.toContain('retry:');
+        // Priming event should have id field but NOT retry field
+        expect(text).toContain('id: ');
+        expect(text).toContain('data: ');
+        expect(text).not.toContain('retry:');
     });
 
-    it('should include event ID in priming event for resumability', async () => {
+    it('should close POST SSE stream when closeSseStream is called', async () => {
         const result = await createTestServer({
             sessionIdGenerator: () => randomUUID(),
             eventStore: createEventStore(),
-            retryInterval: 2000
+            retryInterval: 1000
         });
         server = result.server;
         transport = result.transport;
         baseUrl = result.baseUrl;
+        mcpServer = result.mcpServer;
+
+        // Register a tool that will hang until we close the stream
+        let toolResolve: () => void;
+        const toolPromise = new Promise<void>(resolve => {
+            toolResolve = resolve;
+        });
+
+        mcpServer.tool('slow-tool', 'A slow tool', {}, async () => {
+            await toolPromise;
+            return { content: [{ type: 'text', text: 'Done' }] };
+        });
 
         // Initialize to get session ID
         const initResponse = await sendPostRequest(baseUrl, TEST_MESSAGES.initialize);
         sessionId = initResponse.headers.get('mcp-session-id') as string;
+        expect(sessionId).toBeDefined();
 
-        // Open SSE stream
-        const sseResponse = await fetch(baseUrl, {
-            method: 'GET',
+        // Send a POST request for the slow tool
+        const toolCallRequest: JSONRPCMessage = {
+            jsonrpc: '2.0',
+            id: 42,
+            method: 'tools/call',
+            params: { name: 'slow-tool', arguments: {} }
+        };
+
+        const postResponse = await fetch(baseUrl, {
+            method: 'POST',
             headers: {
-                Accept: 'text/event-stream',
+                'Content-Type': 'application/json',
+                Accept: 'text/event-stream, application/json',
                 'mcp-session-id': sessionId,
                 'mcp-protocol-version': '2025-03-26'
-            }
+            },
+            body: JSON.stringify(toolCallRequest)
         });
 
-        // Read priming event
-        const reader = sseResponse.body?.getReader();
-        const { value } = await reader!.read();
-        const text = new TextDecoder().decode(value);
+        expect(postResponse.status).toBe(200);
 
-        // Extract event ID from priming event
-        const idMatch = text.match(/id: ([^\n]+)/);
-        expect(idMatch).toBeTruthy();
-        const eventId = idMatch![1];
+        const reader = postResponse.body?.getReader();
+
+        // Read the priming event
+        await reader!.read();
+
+        // Close the SSE stream for this request
+        transport.closeSseStream(42);
+
+        // Stream should now be closed
+        const { done } = await reader!.read();
+        expect(done).toBe(true);
 
-        // Event ID should be present and non-empty
-        expect(eventId).toBeTruthy();
-        expect(eventId.length).toBeGreaterThan(0);
+        // Clean up - resolve the tool promise
+        toolResolve!();
     });
 });