diff --git a/src/client/streamableHttp.test.ts b/src/client/streamableHttp.test.ts index e4ba7b1d..ef35cacc 100644 --- a/src/client/streamableHttp.test.ts +++ b/src/client/streamableHttp.test.ts @@ -10,7 +10,8 @@ describe("StreamableHTTPClientTransport", () => { jest.spyOn(global, "fetch"); }); - afterEach(() => { + afterEach(async () => { + await transport.close().catch(() => { }); jest.clearAllMocks(); }); @@ -191,4 +192,155 @@ describe("StreamableHTTPClientTransport", () => { expect(messageSpy).toHaveBeenCalledWith(responseMessage); }); + + it("should attempt initial GET connection and handle 405 gracefully", async () => { + // Mock the server not supporting GET for SSE (returning 405) + (global.fetch as jest.Mock).mockResolvedValueOnce({ + ok: false, + status: 405, + statusText: "Method Not Allowed" + }); + + await transport.start(); + + // Check that GET was attempted + expect(global.fetch).toHaveBeenCalledWith( + expect.anything(), + expect.objectContaining({ + method: "GET", + headers: expect.any(Headers) + }) + ); + + // Verify transport still works after 405 + (global.fetch as jest.Mock).mockResolvedValueOnce({ + ok: true, + status: 202, + headers: new Headers() + }); + + await transport.send({ jsonrpc: "2.0", method: "test", params: {} } as JSONRPCMessage); + expect(global.fetch).toHaveBeenCalledTimes(2); + }); + + it("should handle successful initial GET connection for SSE", async () => { + // Set up readable stream for SSE events + const encoder = new TextEncoder(); + const stream = new ReadableStream({ + start(controller) { + // Send a server notification via SSE + const event = 'event: message\ndata: {"jsonrpc": "2.0", "method": "serverNotification", "params": {}}\n\n'; + controller.enqueue(encoder.encode(event)); + } + }); + + // Mock successful GET connection + (global.fetch as jest.Mock).mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ "content-type": "text/event-stream" }), + body: stream + }); + + const messageSpy = jest.fn(); + transport.onmessage = messageSpy; + + await transport.start(); + + // Give time for the SSE event to be processed + await new Promise(resolve => setTimeout(resolve, 50)); + + expect(messageSpy).toHaveBeenCalledWith( + expect.objectContaining({ + jsonrpc: "2.0", + method: "serverNotification", + params: {} + }) + ); + }); + + it("should handle multiple concurrent SSE streams", async () => { + // Mock two POST requests that return SSE streams + const makeStream = (id: string) => { + const encoder = new TextEncoder(); + return new ReadableStream({ + start(controller) { + const event = `event: message\ndata: {"jsonrpc": "2.0", "result": {"id": "${id}"}, "id": "${id}"}\n\n`; + controller.enqueue(encoder.encode(event)); + } + }); + }; + + (global.fetch as jest.Mock) + .mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ "content-type": "text/event-stream" }), + body: makeStream("request1") + }) + .mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ "content-type": "text/event-stream" }), + body: makeStream("request2") + }); + + const messageSpy = jest.fn(); + transport.onmessage = messageSpy; + + // Send two concurrent requests + await Promise.all([ + transport.send({ jsonrpc: "2.0", method: "test1", params: {}, id: "request1" }), + transport.send({ jsonrpc: "2.0", method: "test2", params: {}, id: "request2" }) + ]); + + // Give time for SSE processing + await new Promise(resolve => setTimeout(resolve, 50)); + + // 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" }) + ); + }); + + it("should include last-event-id header when resuming a broken connection", async () => { + // First make a successful connection that provides an event ID + const encoder = new TextEncoder(); + const stream = new ReadableStream({ + start(controller) { + const event = 'id: event-123\nevent: message\ndata: {"jsonrpc": "2.0", "method": "serverNotification", "params": {}}\n\n'; + controller.enqueue(encoder.encode(event)); + controller.close(); + } + }); + + (global.fetch as jest.Mock).mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ "content-type": "text/event-stream" }), + body: stream + }); + + await transport.start(); + await new Promise(resolve => setTimeout(resolve, 50)); + + // Now simulate attempting to reconnect + (global.fetch as jest.Mock).mockResolvedValueOnce({ + ok: true, + status: 200, + headers: new Headers({ "content-type": "text/event-stream" }), + body: null + }); + + await transport.start(); + + // Check that Last-Event-ID was included + const calls = (global.fetch as jest.Mock).mock.calls; + const lastCall = calls[calls.length - 1]; + expect(lastCall[1].headers.get("last-event-id")).toBe("event-123"); + }); }); \ No newline at end of file diff --git a/src/client/streamableHttp.ts b/src/client/streamableHttp.ts index ffe1b5b4..9fa65c85 100644 --- a/src/client/streamableHttp.ts +++ b/src/client/streamableHttp.ts @@ -1,7 +1,7 @@ import { Transport } from "../shared/transport.js"; import { JSONRPCMessage, JSONRPCMessageSchema } from "../types.js"; import { auth, AuthResult, OAuthClientProvider, UnauthorizedError } from "./auth.js"; -import { EventSource, type ErrorEvent, type EventSourceInit } from "eventsource"; +import { type ErrorEvent } from "eventsource"; export class StreamableHTTPError extends Error { constructor( @@ -34,12 +34,7 @@ export type StreamableHTTPClientTransportOptions = { authProvider?: OAuthClientProvider; /** - * Customizes the initial SSE request to the server (the request that begins the stream). - */ - eventSourceInit?: EventSourceInit; - - /** - * Customizes recurring POST requests to the server. + * Customizes HTTP requests to the server. */ requestInit?: RequestInit; }; @@ -50,10 +45,9 @@ export type StreamableHTTPClientTransportOptions = { * for receiving messages. */ export class StreamableHTTPClientTransport implements Transport { - private _eventSource?: EventSource; + private _activeStreams: Map> = new Map(); private _abortController?: AbortController; private _url: URL; - private _eventSourceInit?: EventSourceInit; private _requestInit?: RequestInit; private _authProvider?: OAuthClientProvider; private _sessionId?: string; @@ -68,7 +62,6 @@ export class StreamableHTTPClientTransport implements Transport { opts?: StreamableHTTPClientTransportOptions, ) { this._url = url; - this._eventSourceInit = opts?.eventSourceInit; this._requestInit = opts?.requestInit; this._authProvider = opts?.authProvider; } @@ -109,61 +102,59 @@ export class StreamableHTTPClientTransport implements Transport { return headers; } - private _startOrAuth(): Promise { - return new Promise((resolve, reject) => { - // Try to open an SSE stream with GET to listen for server messages + private async _startOrAuth(): 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 - - const headers: HeadersInit = { - 'Accept': 'text/event-stream' - }; + const commonHeaders = await this._commonHeaders(); + const headers = new Headers(commonHeaders); + headers.set('Accept', 'text/event-stream'); // Include Last-Event-ID header for resumable streams if (this._lastEventId) { - headers['last-event-id'] = this._lastEventId; + headers.set('last-event-id', this._lastEventId); } - fetch(this._url, { + const response = await fetch(this._url, { method: 'GET', - headers - }).then(response => { - if (response.status === 405) { - // Server doesn't support GET for SSE, which is allowed by the spec - // We'll rely on SSE responses to POST requests for communication - resolve(); - return; - } - - if (!response.ok) { - if (response.status === 401 && this._authProvider) { - // Need to authenticate - this._authThenStart().then(resolve, reject); - return; - } - - const error = new Error(`Failed to open SSE stream: ${response.status} ${response.statusText}`); - reject(error); - this.onerror?.(error); - return; - } - - // Successful connection, handle the SSE stream - this._handleSseStream(response.body); - resolve(); - }).catch(error => { - reject(error); - this.onerror?.(error); + headers, + signal: this._abortController?.signal, }); - }); + + if (response.status === 405) { + // 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}`); + this.onerror?.(error); + throw error; + } + + // Successful connection, handle the SSE stream as a standalone listener + const streamId = `initial-${Date.now()}`; + this._handleSseStream(response.body, streamId); + } catch (error) { + this.onerror?.(error as Error); + throw error; + } } async start() { - if (this._eventSource) { + if (this._activeStreams.size > 0) { throw new Error( "StreamableHTTPClientTransport already started! If using Client class, note that connect() calls start() automatically.", ); } + this._abortController = new AbortController(); return await this._startOrAuth(); } @@ -182,8 +173,18 @@ export class StreamableHTTPClientTransport implements Transport { } async close(): Promise { + // Close all active streams + for (const reader of this._activeStreams.values()) { + try { + reader.cancel(); + } catch (error) { + this.onerror?.(error as Error); + } + } + this._activeStreams.clear(); + + // Abort any pending requests this._abortController?.abort(); - this._eventSource?.close(); // If we have a session ID, send a DELETE request to explicitly terminate the session if (this._sessionId) { @@ -257,17 +258,25 @@ export class StreamableHTTPClientTransport implements Transport { return; } - // Determine if the message contains any requests - const containsRequests = Array.isArray(message) - ? message.some(msg => 'method' in msg && 'id' in msg) - : ('method' in message && 'id' in message); + // Get original message(s) for detecting request IDs + const messages = Array.isArray(message) ? message : [message]; - if (containsRequests) { - // Check if we got text/event-stream in response - if so, handle SSE stream - const contentType = response.headers.get("content-type"); + // Extract IDs from request messages for tracking responses + const requestIds = messages.filter(msg => 'method' in msg && 'id' in msg) + .map(msg => 'id' in msg ? msg.id : undefined) + .filter(id => id !== undefined); + + // If we have request IDs and an SSE response, create a unique stream ID + const hasRequests = requestIds.length > 0; + + // Check the response type + const contentType = response.headers.get("content-type"); + + if (hasRequests) { if (contentType?.includes("text/event-stream")) { - // Handle SSE stream to receive responses - this._handleSseStream(response.body); + // For streaming responses, create a unique stream ID based on request IDs + const streamId = `req-${requestIds.join('-')}-${Date.now()}`; + this._handleSseStream(response.body, streamId); } else if (contentType?.includes("application/json")) { // For non-streaming servers, we might get direct JSON responses const data = await response.json(); @@ -286,13 +295,14 @@ export class StreamableHTTPClientTransport implements Transport { } } - private _handleSseStream(stream: ReadableStream | null): void { + private _handleSseStream(stream: ReadableStream | null, streamId: string): void { if (!stream) { return; } // Set up stream handling for server-sent events const reader = stream.getReader(); + this._activeStreams.set(streamId, reader); const decoder = new TextDecoder(); let buffer = ''; @@ -300,7 +310,11 @@ export class StreamableHTTPClientTransport implements Transport { try { while (true) { const { done, value } = await reader.read(); - if (done) break; + if (done) { + // Stream closed by server + this._activeStreams.delete(streamId); + break; + } buffer += decoder.decode(value, { stream: true }); @@ -326,22 +340,27 @@ export class StreamableHTTPClientTransport implements Transport { } // Update last event ID if provided by server + // As per spec: the ID MUST be globally unique across all streams within that session if (id) { this._lastEventId = id; } // Handle message event - if (eventType === 'message' && data) { - try { - const message = JSONRPCMessageSchema.parse(JSON.parse(data)); - this.onmessage?.(message); - } catch (error) { - this.onerror?.(error as Error); + if (data) { + // Default event type is 'message' per SSE spec if not specified + if (!eventType || eventType === 'message') { + try { + const message = JSONRPCMessageSchema.parse(JSON.parse(data)); + this.onmessage?.(message); + } catch (error) { + this.onerror?.(error as Error); + } } } } } } catch (error) { + this._activeStreams.delete(streamId); this.onerror?.(error as Error); } };