Revert "fix(client): await SSE stream open in connect() to prevent race condition"

This reverts commit d0eeff6.

#1521 (review)

Author: LucaButBoring

src/client/streamableHttp.ts

@@ -137,6 +137,7 @@ export class StreamableHTTPClientTransport implements Transport {
     private _lastUpscopingHeader?: string; // Track last upscoping header to prevent infinite upscoping.
     private _serverRetryMs?: number; // Server-provided retry delay from SSE retry field
     private _reconnectionTimeout?: ReturnType<typeof setTimeout>;
+    private _sseStreamOpened = false; // Track if SSE stream was successfully opened
 
     onclose?: () => void;
     onerror?: (error: Error) => void;
@@ -240,6 +241,7 @@ export class StreamableHTTPClientTransport implements Transport {
                 throw new StreamableHTTPError(response.status, `Failed to open SSE stream: ${response.statusText}`);
             }
 
+            this._sseStreamOpened = true;
             this._handleSseStream(response.body, options, true);
         } catch (error) {
             this.onerror?.(error as Error);
@@ -479,10 +481,19 @@ export class StreamableHTTPClientTransport implements Transport {
 
             // Handle session ID received during initialization
             const sessionId = response.headers.get('mcp-session-id');
+            const hadSessionId = this._sessionId !== undefined;
             if (sessionId) {
                 this._sessionId = sessionId;
             }
 
+            // If we just received a session ID for the first time and SSE stream is not open,
+            // try to open it now. This handles the case where the initial SSE connection
+            // during start() was rejected because the server wasn't initialized yet.
+            // See: https://github.com/modelcontextprotocol/typescript-sdk/issues/1167
+            if (sessionId && !hadSessionId && !this._sseStreamOpened) {
+                this._startOrAuthSse({ resumptionToken: undefined }).catch(err => this.onerror?.(err));
+            }
+
             if (!response.ok) {
                 const text = await response.text().catch(() => null);
 
@@ -561,12 +572,8 @@ export class StreamableHTTPClientTransport implements Transport {
                 // if the accepted notification is initialized, we start the SSE stream
                 // if it's supported by the server
                 if (isInitializedNotification(message)) {
-                    // Await the SSE stream opening so that connect() does not resolve
-                    // until the GET listener is established.  This prevents a race where
-                    // the server sends a request (e.g. roots/list) before the stream is ready.
-                    // Errors are swallowed here because _startOrAuthSse already reports
-                    // them via onerror, and the SSE stream is optional (server may 405).
-                    await this._startOrAuthSse({ resumptionToken: undefined }).catch(() => {});
+                    // Start without a lastEventId since this is a fresh connection
+                    this._startOrAuthSse({ resumptionToken: undefined }).catch(err => this.onerror?.(err));
                 }
                 return;
             }

test/client/streamableHttp.test.ts

@@ -101,8 +101,19 @@ describe('StreamableHTTPClientTransport', () => {
             headers: new Headers({ 'content-type': 'text/event-stream', 'mcp-session-id': 'test-session-id' })
         });
 
+        // Mock the SSE stream GET request that happens after receiving session ID
+        (global.fetch as Mock).mockResolvedValueOnce({
+            ok: false,
+            status: 405,
+            headers: new Headers(),
+            body: { cancel: vi.fn() }
+        });
+
         await transport.send(message);
 
+        // Allow the async SSE connection attempt to complete
+        await new Promise(resolve => setTimeout(resolve, 10));
+
         // Send a second message that should include the session ID
         (global.fetch as Mock).mockResolvedValueOnce({
             ok: true,
@@ -137,8 +148,19 @@ describe('StreamableHTTPClientTransport', () => {
             headers: new Headers({ 'content-type': 'text/event-stream', 'mcp-session-id': 'test-session-id' })
         });
 
+        // Mock the SSE stream GET request that happens after receiving session ID
+        (global.fetch as Mock).mockResolvedValueOnce({
+            ok: false,
+            status: 405,
+            headers: new Headers(),
+            body: { cancel: vi.fn() }
+        });
+
         await transport.send(message);
 
+        // Allow the async SSE connection attempt to complete
+        await new Promise(resolve => setTimeout(resolve, 10));
+
         expect(transport.sessionId).toBe('test-session-id');
 
         // Now terminate the session
@@ -178,8 +200,19 @@ describe('StreamableHTTPClientTransport', () => {
             headers: new Headers({ 'content-type': 'text/event-stream', 'mcp-session-id': 'test-session-id' })
         });
 
+        // Mock the SSE stream GET request that happens after receiving session ID
+        (global.fetch as Mock).mockResolvedValueOnce({
+            ok: false,
+            status: 405,
+            headers: new Headers(),
+            body: { cancel: vi.fn() }
+        });
+
         await transport.send(message);
 
+        // Allow the async SSE connection attempt to complete
+        await new Promise(resolve => setTimeout(resolve, 10));
+
         // Now terminate the session, but server responds with 405
         (global.fetch as Mock).mockResolvedValueOnce({
             ok: false,