From 18e4981bd3d0fde152659b95d4250828872718f8 Mon Sep 17 00:00:00 2001 From: ihrpr Date: Fri, 11 Apr 2025 14:06:51 +0100 Subject: [PATCH] move get sse stream from a separate method to automatic connection when notifications/initialized received --- src/client/streamableHttp.test.ts | 11 ++++---- src/client/streamableHttp.ts | 30 +++++++++------------ src/examples/client/simpleStreamableHttp.ts | 21 +++------------ src/server/streamableHttp.test.ts | 2 +- 4 files changed, 23 insertions(+), 41 deletions(-) diff --git a/src/client/streamableHttp.test.ts b/src/client/streamableHttp.test.ts index 40f22139..1572e912 100644 --- a/src/client/streamableHttp.test.ts +++ b/src/client/streamableHttp.test.ts @@ -164,8 +164,7 @@ describe("StreamableHTTPClientTransport", () => { // We expect the 405 error to be caught and handled gracefully // This should not throw an error that breaks the transport await transport.start(); - await expect(transport.openSseStream()).rejects.toThrow("Failed to open SSE stream: Method Not Allowed"); - + await expect(transport["_startOrAuthStandaloneSSE"]()).resolves.not.toThrow("Failed to open SSE stream: Method Not Allowed"); // Check that GET was attempted expect(global.fetch).toHaveBeenCalledWith( expect.anything(), @@ -209,7 +208,7 @@ describe("StreamableHTTPClientTransport", () => { transport.onmessage = messageSpy; await transport.start(); - await transport.openSseStream(); + await transport["_startOrAuthStandaloneSSE"](); // Give time for the SSE event to be processed await new Promise(resolve => setTimeout(resolve, 50)); @@ -295,7 +294,7 @@ describe("StreamableHTTPClientTransport", () => { }); await transport.start(); - await transport.openSseStream(); + await transport["_startOrAuthStandaloneSSE"](); await new Promise(resolve => setTimeout(resolve, 50)); // Now simulate attempting to reconnect @@ -306,7 +305,7 @@ describe("StreamableHTTPClientTransport", () => { body: null }); - await transport.openSseStream(); + await transport["_startOrAuthStandaloneSSE"](); // Check that Last-Event-ID was included const calls = (global.fetch as jest.Mock).mock.calls; @@ -366,7 +365,7 @@ describe("StreamableHTTPClientTransport", () => { await transport.start(); - await transport.openSseStream(); + await transport["_startOrAuthStandaloneSSE"](); expect((actualReqInit.headers as Headers).get("x-custom-header")).toBe("CustomValue"); requestInit.headers["X-Custom-Header"] = "SecondCustomValue"; diff --git a/src/client/streamableHttp.ts b/src/client/streamableHttp.ts index ae54bb26..ea69ee77 100644 --- a/src/client/streamableHttp.ts +++ b/src/client/streamableHttp.ts @@ -1,5 +1,5 @@ import { Transport } from "../shared/transport.js"; -import { JSONRPCMessage, JSONRPCMessageSchema } from "../types.js"; +import { isJSONRPCNotification, JSONRPCMessage, JSONRPCMessageSchema } from "../types.js"; import { auth, AuthResult, OAuthClientProvider, UnauthorizedError } from "./auth.js"; import { EventSourceParserStream } from "eventsource-parser/stream"; @@ -126,6 +126,12 @@ export class StreamableHTTPClientTransport implements Transport { return await this._authThenStart(); } + // 405 indicates that the server does not offer an SSE stream at GET endpoint + // This is an expected case that should not trigger an error + if (response.status === 405) { + return; + } + throw new StreamableHTTPError( response.status, `Failed to open SSE stream: ${response.statusText}`, @@ -243,6 +249,12 @@ export class StreamableHTTPClientTransport implements Transport { // If the response is 202 Accepted, there's no body to process if (response.status === 202) { + // if the accepted notification is initialized, we start the SSE stream + // if it's supported by the server + if (isJSONRPCNotification(message) && message.method === "notifications/initialized") { + // We don't need to handle 405 here anymore as it's handled in _startOrAuthStandaloneSSE + this._startOrAuthStandaloneSSE().catch(err => this.onerror?.(err)); + } return; } @@ -279,20 +291,4 @@ export class StreamableHTTPClientTransport implements Transport { throw error; } } - - /** - * Opens SSE stream to receive messages from the server. - * - * This allows the server to push messages to the client without requiring the client - * to first send a request via HTTP POST. Some servers may not support this feature. - * If authentication is required but fails, this method will throw an UnauthorizedError. - */ - async openSseStream(): Promise { - if (!this._abortController) { - throw new Error( - "StreamableHTTPClientTransport not started! Call connect() before openSseStream().", - ); - } - await this._startOrAuthStandaloneSSE(); - } } diff --git a/src/examples/client/simpleStreamableHttp.ts b/src/examples/client/simpleStreamableHttp.ts index 027db881..c23c17f8 100644 --- a/src/examples/client/simpleStreamableHttp.ts +++ b/src/examples/client/simpleStreamableHttp.ts @@ -25,20 +25,10 @@ async function main(): Promise { const transport = new StreamableHTTPClientTransport( new URL('http://localhost:3000/mcp') ); - let supportsStandaloneSse = false; - - // Connect the client using the transport and initialize the server + + // Connect the client using the transport and initialize the server await client.connect(transport); console.log('Connected to MCP server'); - console.log('Opening SSE stream to receive server notifications...'); - try { - await transport.openSseStream(); - supportsStandaloneSse = true; - 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) => { @@ -136,11 +126,8 @@ async function main(): Promise { } 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.'); - } - + // Keep the connection open to receive notifications + console.log('\nKeeping connection open to receive notifications. Press Ctrl+C to exit.'); } main().catch((error: unknown) => { diff --git a/src/server/streamableHttp.test.ts b/src/server/streamableHttp.test.ts index d6f7cb8a..85fcae2f 100644 --- a/src/server/streamableHttp.test.ts +++ b/src/server/streamableHttp.test.ts @@ -578,7 +578,7 @@ describe("StreamableHTTPServerTransport", () => { expect(mockResponse.writeHead).toHaveBeenCalledWith(200, expect.objectContaining({ "Content-Type": "text/event-stream", - "Cache-Control": "no-cache", + "Cache-Control": "no-cache, no-transform", "Connection": "keep-alive", "mcp-session-id": transport.sessionId, }));