feat: implement SEP-1699 SSE polling via server-side disconnect (#1129)

src/server/streamableHttp.test.ts

@@ -31,6 +31,7 @@ interface TestServerConfig {
     eventStore?: EventStore;
     onsessioninitialized?: (sessionId: string) => void | Promise<void>;
     onsessionclosed?: (sessionId: string) => void | Promise<void>;
+    retryInterval?: number;
 }
 
 /**
@@ -142,7 +143,8 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
             enableJsonResponse: config.enableJsonResponse ?? false,
             eventStore: config.eventStore,
             onsessioninitialized: config.onsessioninitialized,
-            onsessionclosed: config.onsessionclosed
+            onsessionclosed: config.onsessionclosed,
+            retryInterval: config.retryInterval
         });
 
         await mcpServer.connect(transport);
@@ -1427,6 +1429,78 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
             expect(reconnectText).toContain('Second notification from MCP server');
             expect(reconnectText).toContain('id: ');
         });
+
+        it('should store and replay multiple notifications sent while client is disconnected', async () => {
+            // Establish a standalone SSE stream
+            const sseResponse = await fetch(baseUrl, {
+                method: 'GET',
+                headers: {
+                    Accept: 'text/event-stream',
+                    'mcp-session-id': sessionId,
+                    'mcp-protocol-version': '2025-03-26'
+                }
+            });
+            expect(sseResponse.status).toBe(200);
+
+            const reader = sseResponse.body?.getReader();
+
+            // Send a notification to get an event ID
+            await mcpServer.server.sendLoggingMessage({ level: 'info', data: 'Initial notification' });
+
+            // Read the notification from the SSE stream
+            const { value } = await reader!.read();
+            const text = new TextDecoder().decode(value);
+
+            // Extract the event ID
+            const idMatch = text.match(/id: ([^\n]+)/);
+            expect(idMatch).toBeTruthy();
+            const lastEventId = idMatch![1];
+
+            // Close the SSE stream to simulate a disconnect
+            await reader!.cancel();
+
+            // Send MULTIPLE notifications while the client is disconnected
+            await mcpServer.server.sendLoggingMessage({ level: 'info', data: 'Missed notification 1' });
+            await mcpServer.server.sendLoggingMessage({ level: 'info', data: 'Missed notification 2' });
+            await mcpServer.server.sendLoggingMessage({ level: 'info', data: 'Missed notification 3' });
+
+            // Reconnect with the Last-Event-ID to get all missed messages
+            const reconnectResponse = await fetch(baseUrl, {
+                method: 'GET',
+                headers: {
+                    Accept: 'text/event-stream',
+                    'mcp-session-id': sessionId,
+                    'mcp-protocol-version': '2025-03-26',
+                    'last-event-id': lastEventId
+                }
+            });
+
+            expect(reconnectResponse.status).toBe(200);
+
+            // Read replayed notifications with a timeout
+            const reconnectReader = reconnectResponse.body?.getReader();
+            let allText = '';
+
+            // Read chunks until we have all 3 notifications or timeout
+            const readWithTimeout = async () => {
+                const timeout = setTimeout(() => reconnectReader!.cancel(), 2000);
+                try {
+                    while (!allText.includes('Missed notification 3')) {
+                        const { value, done } = await reconnectReader!.read();
+                        if (done) break;
+                        allText += new TextDecoder().decode(value);
+                    }
+                } finally {
+                    clearTimeout(timeout);
+                }
+            };
+            await readWithTimeout();
+
+            // Verify we received ALL notifications that were sent while disconnected
+            expect(allText).toContain('Missed notification 1');
+            expect(allText).toContain('Missed notification 2');
+            expect(allText).toContain('Missed notification 3');
+        });
     });
 
     // Test stateless mode
@@ -1517,6 +1591,219 @@ describe.each(zodTestMatrix)('$zodVersionLabel', (entry: ZodMatrixEntry) => {
         });
     });
 
+    // Test SSE priming events for POST streams
+    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; streamId: string }>();
+            return {
+                async storeEvent(streamId: string, message: JSONRPCMessage): Promise<string> {
+                    const eventId = `${streamId}::${Date.now()}_${randomUUID()}`;
+                    storedEvents.set(eventId, { eventId, message, streamId });
+                    return eventId;
+                },
+                async getStreamIdForEventId(eventId: string): Promise<string | undefined> {
+                    const event = storedEvents.get(eventId);
+                    return event?.streamId;
+                },
+                async replayEventsAfter(
+                    lastEventId: EventId,
+                    { send }: { send: (eventId: EventId, message: JSONRPCMessage) => Promise<void> }
+                ): Promise<StreamId> {
+                    const event = storedEvents.get(lastEventId);
+                    const streamId = event?.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);
+                        }
+                    }
+                    return streamId;
+                }
+            };
+        };
+
+        afterEach(async () => {
+            if (server && transport) {
+                await stopTestServer({ server, transport });
+            }
+        });
+
+        it('should send priming event with retry field on POST SSE stream', async () => {
+            const result = await createTestServer({
+                sessionIdGenerator: () => randomUUID(),
+                eventStore: createEventStore(),
+                retryInterval: 5000
+            });
+            server = result.server;
+            transport = result.transport;
+            baseUrl = result.baseUrl;
+            mcpServer = result.mcpServer;
+
+            // 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();
+
+            // Send a tool call request
+            const toolCallRequest: JSONRPCMessage = {
+                jsonrpc: '2.0',
+                id: 100,
+                method: 'tools/call',
+                params: { name: 'greet', arguments: { name: 'Test' } }
+            };
+
+            const postResponse = await fetch(baseUrl, {
+                method: 'POST',
+                headers: {
+                    '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(postResponse.status).toBe(200);
+            expect(postResponse.headers.get('content-type')).toBe('text/event-stream');
+
+            // Read the priming event
+            const reader = postResponse.body?.getReader();
+            const { value } = await reader!.read();
+            const text = new TextDecoder().decode(value);
+
+            // Verify priming event has id and retry field
+            expect(text).toContain('id: ');
+            expect(text).toContain('retry: 5000');
+            expect(text).toContain('data: ');
+        });
+
+        it('should send priming event without retry field when retryInterval is not configured', async () => {
+            const result = await createTestServer({
+                sessionIdGenerator: () => randomUUID(),
+                eventStore: createEventStore()
+                // No retryInterval
+            });
+            server = result.server;
+            transport = result.transport;
+            baseUrl = result.baseUrl;
+            mcpServer = result.mcpServer;
+
+            // 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();
+
+            // Send a tool call request
+            const toolCallRequest: JSONRPCMessage = {
+                jsonrpc: '2.0',
+                id: 100,
+                method: 'tools/call',
+                params: { name: 'greet', arguments: { name: 'Test' } }
+            };
+
+            const postResponse = await fetch(baseUrl, {
+                method: 'POST',
+                headers: {
+                    '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(postResponse.status).toBe(200);
+
+            // Read the priming event
+            const reader = postResponse.body?.getReader();
+            const { value } = await reader!.read();
+            const text = new TextDecoder().decode(value);
+
+            // 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 close POST SSE stream when closeSseStream is called', async () => {
+            const result = await createTestServer({
+                sessionIdGenerator: () => randomUUID(),
+                eventStore: createEventStore(),
+                retryInterval: 1000
+            });
+            server = result.server;
+            transport = result.transport;
+            baseUrl = result.baseUrl;
+            mcpServer = result.mcpServer;
+
+            // Track tool execution state
+            let toolResolve: () => void;
+            const toolPromise = new Promise<void>(resolve => {
+                toolResolve = resolve;
+            });
+
+            // Register a blocking tool
+            mcpServer.tool('blocking-tool', 'A blocking 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();
+
+            // Send a tool call request
+            const toolCallRequest: JSONRPCMessage = {
+                jsonrpc: '2.0',
+                id: 100,
+                method: 'tools/call',
+                params: { name: 'blocking-tool', arguments: {} }
+            };
+
+            const postResponse = await fetch(baseUrl, {
+                method: 'POST',
+                headers: {
+                    '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(postResponse.status).toBe(200);
+
+            const reader = postResponse.body?.getReader();
+
+            // Read the priming event
+            await reader!.read();
+
+            // Close the SSE stream
+            transport.closeSSEStream(100);
+
+            // Stream should now be closed
+            const { done } = await reader!.read();
+            expect(done).toBe(true);
+
+            // Clean up - resolve the tool promise
+            toolResolve!();
+        });
+    });
+
     // Test onsessionclosed callback
     describe('StreamableHTTPServerTransport onsessionclosed callback', () => {
         it('should call onsessionclosed callback when session is closed via DELETE', async () => {

src/server/streamableHttp.ts

@@ -35,6 +35,16 @@ export interface EventStore {
      */
     storeEvent(streamId: StreamId, message: JSONRPCMessage): Promise<EventId>;
 
+    /**
+     * Get the stream ID associated with a given event ID.
+     * @param eventId The event ID to look up
+     * @returns The stream ID, or undefined if not found
+     *
+     * Optional: If not provided, the SDK will use the streamId returned by
+     * replayEventsAfter for stream mapping.
+     */
+    getStreamIdForEventId?(eventId: EventId): Promise<StreamId | undefined>;
+
     replayEventsAfter(
         lastEventId: EventId,
         {
@@ -108,6 +118,13 @@ export interface StreamableHTTPServerTransportOptions {
      * Default is false for backwards compatibility.
      */
     enableDnsRebindingProtection?: boolean;
+
+    /**
+     * Retry interval in milliseconds to suggest to clients in SSE retry field.
+     * When set, the server will send a retry field in SSE priming events to control
+     * client reconnection timing for polling behavior.
+     */
+    retryInterval?: number;
 }
 
 /**
@@ -160,6 +177,7 @@ export class StreamableHTTPServerTransport implements Transport {
     private _allowedHosts?: string[];
     private _allowedOrigins?: string[];
     private _enableDnsRebindingProtection: boolean;
+    private _retryInterval?: number;
 
     sessionId?: string;
     onclose?: () => void;
@@ -175,6 +193,7 @@ export class StreamableHTTPServerTransport implements Transport {
         this._allowedHosts = options.allowedHosts;
         this._allowedOrigins = options.allowedOrigins;
         this._enableDnsRebindingProtection = options.enableDnsRebindingProtection ?? false;
+        this._retryInterval = options.retryInterval;
     }
 
     /**
@@ -249,6 +268,24 @@ export class StreamableHTTPServerTransport implements Transport {
         }
     }
 
+    /**
+     * Writes a priming event to establish resumption capability.
+     * Only sends if eventStore is configured (opt-in for resumability).
+     */
+    private async _maybeWritePrimingEvent(res: ServerResponse, streamId: string): Promise<void> {
+        if (!this._eventStore) {
+            return;
+        }
+
+        const primingEventId = await this._eventStore.storeEvent(streamId, {} as JSONRPCMessage);
+
+        let primingEvent = `id: ${primingEventId}\ndata: \n\n`;
+        if (this._retryInterval !== undefined) {
+            primingEvent = `id: ${primingEventId}\nretry: ${this._retryInterval}\ndata: \n\n`;
+        }
+        res.write(primingEvent);
+    }
+
     /**
      * Handles GET requests for SSE stream
      */
@@ -342,6 +379,41 @@ export class StreamableHTTPServerTransport implements Transport {
             return;
         }
         try {
+            // If getStreamIdForEventId is available, use it for conflict checking
+            let streamId: string | undefined;
+            if (this._eventStore.getStreamIdForEventId) {
+                streamId = await this._eventStore.getStreamIdForEventId(lastEventId);
+
+                if (!streamId) {
+                    res.writeHead(400).end(
+                        JSON.stringify({
+                            jsonrpc: '2.0',
+                            error: {
+                                code: -32000,
+                                message: 'Invalid event ID format'
+                            },
+                            id: null
+                        })
+                    );
+                    return;
+                }
+
+                // Check conflict with the SAME streamId we'll use for mapping
+                if (this._streamMapping.get(streamId) !== undefined) {
+                    res.writeHead(409).end(
+                        JSON.stringify({
+                            jsonrpc: '2.0',
+                            error: {
+                                code: -32000,
+                                message: 'Conflict: Stream already has an active connection'
+                            },
+                            id: null
+                        })
+                    );
+                    return;
+                }
+            }
+
             const headers: Record<string, string> = {
                 'Content-Type': 'text/event-stream',
                 'Cache-Control': 'no-cache, no-transform',
@@ -353,7 +425,8 @@ export class StreamableHTTPServerTransport implements Transport {
             }
             res.writeHead(200, headers).flushHeaders();
 
-            const streamId = await this._eventStore?.replayEventsAfter(lastEventId, {
+            // Replay events - returns the streamId for backwards compatibility
+            const replayedStreamId = await this._eventStore.replayEventsAfter(lastEventId, {
                 send: async (eventId: string, message: JSONRPCMessage) => {
                     if (!this.writeSSEEvent(res, message, eventId)) {
                         this.onerror?.(new Error('Failed replay events'));
@@ -361,7 +434,13 @@ export class StreamableHTTPServerTransport implements Transport {
                     }
                 }
             });
-            this._streamMapping.set(streamId, res);
+
+            this._streamMapping.set(replayedStreamId, res);
+
+            // Set up close handler for client disconnects
+            res.on('close', () => {
+                this._streamMapping.delete(replayedStreamId);
+            });
 
             // Add error handler for replay stream
             res.on('error', error => {
@@ -547,6 +626,8 @@ export class StreamableHTTPServerTransport implements Transport {
                     }
 
                     res.writeHead(200, headers);
+
+                    await this._maybeWritePrimingEvent(res, streamId);
                 }
                 // Store the response for this request to send messages back through this connection
                 // We need to track by request ID to maintain the connection
@@ -709,6 +790,22 @@ export class StreamableHTTPServerTransport implements Transport {
         this.onclose?.();
     }
 
+    /**
+     * Close an SSE stream for a specific request, triggering client reconnection.
+     * Use this to implement polling behavior during long-running operations -
+     * client will reconnect after the retry interval specified in the priming event.
+     */
+    closeSSEStream(requestId: RequestId): void {
+        const streamId = this._requestToStreamMapping.get(requestId);
+        if (!streamId) return;
+
+        const stream = this._streamMapping.get(streamId);
+        if (stream) {
+            stream.end();
+            this._streamMapping.delete(streamId);
+        }
+    }
+
     async send(message: JSONRPCMessage, options?: { relatedRequestId?: RequestId }): Promise<void> {
         let requestId = options?.relatedRequestId;
         if (isJSONRPCResponse(message) || isJSONRPCError(message)) {