diff --git a/src/client/streamableHttp.test.ts b/src/client/streamableHttp.test.ts index 8a1b9044..0dc582d4 100644 --- a/src/client/streamableHttp.test.ts +++ b/src/client/streamableHttp.test.ts @@ -161,7 +161,10 @@ describe("StreamableHTTPClientTransport", () => { statusText: "Method Not Allowed" }); + // 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'); // Check that GET was attempted expect(global.fetch).toHaveBeenCalledWith( @@ -206,6 +209,7 @@ describe("StreamableHTTPClientTransport", () => { transport.onmessage = messageSpy; await transport.start(); + await transport.openSseStream(); // Give time for the SSE event to be processed await new Promise(resolve => setTimeout(resolve, 50)); @@ -233,7 +237,7 @@ describe("StreamableHTTPClientTransport", () => { (global.fetch as jest.Mock) .mockResolvedValueOnce({ - ok: true, + ok: true, status: 200, headers: new Headers({ "content-type": "text/event-stream" }), body: makeStream("request1") @@ -255,16 +259,21 @@ describe("StreamableHTTPClientTransport", () => { ]); // Give time for SSE processing - await new Promise(resolve => setTimeout(resolve, 50)); + await new Promise(resolve => setTimeout(resolve, 100)); // Both streams should have delivered their messages expect(messageSpy).toHaveBeenCalledTimes(2); - expect(messageSpy).toHaveBeenCalledWith( - expect.objectContaining({ result: { id: "request1" }, id: "request1" }) - ); - expect(messageSpy).toHaveBeenCalledWith( - expect.objectContaining({ result: { id: "request2" }, id: "request2" }) - ); + + // Verify received messages without assuming specific order + expect(messageSpy.mock.calls.some(call => { + const msg = call[0]; + return msg.id === "request1" && msg.result?.id === "request1"; + })).toBe(true); + + expect(messageSpy.mock.calls.some(call => { + const msg = call[0]; + return msg.id === "request2" && msg.result?.id === "request2"; + })).toBe(true); }); it("should include last-event-id header when resuming a broken connection", async () => { @@ -286,6 +295,7 @@ describe("StreamableHTTPClientTransport", () => { }); await transport.start(); + await transport.openSseStream(); await new Promise(resolve => setTimeout(resolve, 50)); // Now simulate attempting to reconnect @@ -296,7 +306,7 @@ describe("StreamableHTTPClientTransport", () => { body: null }); - await transport.start(); + await transport.openSseStream(); // Check that Last-Event-ID was included const calls = (global.fetch as jest.Mock).mock.calls; diff --git a/src/client/streamableHttp.ts b/src/client/streamableHttp.ts index 38da5bcb..82cd61ec 100644 --- a/src/client/streamableHttp.ts +++ b/src/client/streamableHttp.ts @@ -1,13 +1,11 @@ import { Transport } from "../shared/transport.js"; import { JSONRPCMessage, JSONRPCMessageSchema } from "../types.js"; import { auth, AuthResult, OAuthClientProvider, UnauthorizedError } from "./auth.js"; -import { type ErrorEvent } from "eventsource"; import { EventSourceMessage, EventSourceParserStream } from 'eventsource-parser/stream'; export class StreamableHTTPError extends Error { constructor( public readonly code: number | undefined, message: string | undefined, - public readonly event: ErrorEvent, ) { super(`Streamable HTTP error: ${message}`); } @@ -83,7 +81,7 @@ export class StreamableHTTPClientTransport implements Transport { throw new UnauthorizedError(); } - return await this._startOrAuth(); + return await this._startOrAuthStandaloneSSE(); } private async _commonHeaders(): Promise { @@ -102,7 +100,7 @@ export class StreamableHTTPClientTransport implements Transport { return headers; } - private async _startOrAuth(): Promise { + private async _startOrAuthStandaloneSSE(): Promise { try { // Try to open an initial SSE stream with GET to listen for server messages // This is optional according to the spec - server may not support it @@ -121,25 +119,22 @@ export class StreamableHTTPClientTransport implements Transport { signal: this._abortController?.signal, }); - if (response.status === 405 || response.status === 404) { - // Server doesn't support GET for SSE, which is allowed by the spec - // We'll rely on SSE responses to POST requests for communication - return; - } - if (!response.ok) { if (response.status === 401 && this._authProvider) { // Need to authenticate return await this._authThenStart(); } - const error = new Error(`Failed to open SSE stream: ${response.status} ${response.statusText}`); + const error = new StreamableHTTPError( + response.status, + `Failed to open SSE stream: ${response.statusText}`, + ); this.onerror?.(error); throw error; } // Successful connection, handle the SSE stream as a standalone listener - const streamId = `initial-${Date.now()}`; + const streamId = `standalone-sse-${Date.now()}`; this._handleSseStream(response.body, streamId); } catch (error) { this.onerror?.(error as Error); @@ -147,6 +142,53 @@ export class StreamableHTTPClientTransport implements Transport { } } + private _handleSseStream(stream: ReadableStream | null, streamId: string): void { + if (!stream) { + return; + } + + // Create a pipeline: binary stream -> text decoder -> SSE parser + const eventStream = stream + .pipeThrough(new TextDecoderStream()) + .pipeThrough(new EventSourceParserStream()); + + const reader = eventStream.getReader(); + this._activeStreams.set(streamId, reader); + + const processStream = async () => { + try { + while (true) { + const { done, value: event } = await reader.read(); + if (done) { + this._activeStreams.delete(streamId); + break; + } + + // Update last event ID if provided + if (event.id) { + this._lastEventId = event.id; + } + + // Handle message events (default event type is undefined per docs) + // or explicit 'message' event type + if (!event.event || event.event === 'message') { + try { + const message = JSONRPCMessageSchema.parse(JSON.parse(event.data)); + this.onmessage?.(message); + } catch (error) { + this.onerror?.(error as Error); + } + } + } + } catch (error) { + this._activeStreams.delete(streamId); + this.onerror?.(error as Error); + } + }; + + processStream(); + } + async start() { if (this._activeStreams.size > 0) { throw new Error( @@ -155,7 +197,6 @@ export class StreamableHTTPClientTransport implements Transport { } this._abortController = new AbortController(); - return await this._startOrAuth(); } /** @@ -271,50 +312,17 @@ export class StreamableHTTPClientTransport implements Transport { } } - private _handleSseStream(stream: ReadableStream | null, streamId: string): void { - if (!stream) { - return; + /** + * 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) { + this._abortController = new AbortController(); } - - // Create a pipeline: binary stream -> text decoder -> SSE parser - const eventStream = stream - .pipeThrough(new TextDecoderStream()) - .pipeThrough(new EventSourceParserStream()); - - const reader = eventStream.getReader(); - this._activeStreams.set(streamId, reader); - - const processStream = async () => { - try { - while (true) { - const { done, value: event } = await reader.read(); - if (done) { - this._activeStreams.delete(streamId); - break; - } - - // Update last event ID if provided - if (event.id) { - this._lastEventId = event.id; - } - - // Handle message events (default event type is undefined per docs) - // or explicit 'message' event type - if (!event.event || event.event === 'message') { - try { - const message = JSONRPCMessageSchema.parse(JSON.parse(event.data)); - this.onmessage?.(message); - } catch (error) { - this.onerror?.(error as Error); - } - } - } - } catch (error) { - this._activeStreams.delete(streamId); - this.onerror?.(error as Error); - } - }; - - processStream(); + await this._startOrAuthStandaloneSSE(); } } \ No newline at end of file diff --git a/src/server/streamableHttp.ts b/src/server/streamableHttp.ts index 5aff0c45..b0fcce6d 100644 --- a/src/server/streamableHttp.ts +++ b/src/server/streamableHttp.ts @@ -359,13 +359,9 @@ export class StreamableHTTPServerTransport implements Transport { async send(message: JSONRPCMessage, options?: { relatedRequestId?: RequestId }): Promise { let requestId = options?.relatedRequestId; - let shouldCloseConnection = false; if ('result' in message || 'error' in message) { // If the message is a response, use the request ID from the message requestId = message.id; - // This is a response to the original request, we can close the stream - // after sending all related responses - shouldCloseConnection = true; } if (requestId === undefined) { throw new Error("No request ID provided for the message"); @@ -380,8 +376,8 @@ export class StreamableHTTPServerTransport implements Transport { sseResponse.write( `event: message\ndata: ${JSON.stringify(message)}\n\n`, ); - - if (shouldCloseConnection) { + // After all JSON-RPC responses have been sent, the server SHOULD close the SSE stream. + if ('result' in message || 'error' in message) { this._sseResponseMapping.delete(requestId); // Only close the connection if it's not needed by other requests const canCloseConnection = ![...this._sseResponseMapping.entries()].some(([id, res]) => res === sseResponse && id !== requestId);