From 583202f81aedfc0b39bc8d9ae82b7cea56bc7688 Mon Sep 17 00:00:00 2001 From: ihrpr Date: Tue, 15 Apr 2025 09:24:11 +0100 Subject: [PATCH] address client comments --- src/client/streamableHttp.ts | 53 ++++++++++++++++++++---------------- 1 file changed, 29 insertions(+), 24 deletions(-) diff --git a/src/client/streamableHttp.ts b/src/client/streamableHttp.ts index 9451e8d9..7bb88c0d 100644 --- a/src/client/streamableHttp.ts +++ b/src/client/streamableHttp.ts @@ -1,9 +1,16 @@ -import { log } from "node:console"; import { Transport } from "../shared/transport.js"; import { isJSONRPCNotification, JSONRPCMessage, JSONRPCMessageSchema } from "../types.js"; import { auth, AuthResult, OAuthClientProvider, UnauthorizedError } from "./auth.js"; import { EventSourceParserStream } from "eventsource-parser/stream"; +// Default reconnection options for StreamableHTTP connections +const DEFAULT_STREAMABLE_HTTP_RECONNECTION_OPTIONS: StreamableHTTPReconnectionOptions = { + initialReconnectionDelay: 1000, + maxReconnectionDelay: 30000, + reconnectionDelayGrowFactor: 1.5, + maxRetries: 2, +}; + export class StreamableHTTPError extends Error { constructor( public readonly code: number | undefined, @@ -13,6 +20,16 @@ export class StreamableHTTPError extends Error { } } +/** + * Options for starting or authenticating an SSE connection + */ +export interface StartSSEOptions { + /** + * The ID of the last received event, used for resuming a disconnected stream + */ + lastEventId?: string; +} + /** * Configuration options for reconnection behavior of the StreamableHTTPClientTransport. */ @@ -37,7 +54,7 @@ export interface StreamableHTTPReconnectionOptions { /** * Maximum number of reconnection attempts before giving up. - * Default is 0 (unlimited). + * Default is 2. */ maxRetries: number; } @@ -97,7 +114,7 @@ export class StreamableHTTPClientTransport implements Transport { this._url = url; this._requestInit = opts?.requestInit; this._authProvider = opts?.authProvider; - this._reconnectionOptions = opts?.reconnectionOptions || this._defaultReconnectionOptions; + this._reconnectionOptions = opts?.reconnectionOptions ?? DEFAULT_STREAMABLE_HTTP_RECONNECTION_OPTIONS; } private async _authThenStart(): Promise { @@ -117,7 +134,7 @@ export class StreamableHTTPClientTransport implements Transport { throw new UnauthorizedError(); } - return await this._startOrAuthStandaloneSSE(); + return await this._startOrAuthStandaloneSSE({ lastEventId: undefined }); } private async _commonHeaders(): Promise { @@ -138,7 +155,9 @@ export class StreamableHTTPClientTransport implements Transport { ); } - private async _startOrAuthStandaloneSSE(lastEventId?: string): Promise { + + private async _startOrAuthStandaloneSSE(options: StartSSEOptions): Promise { + const { lastEventId } = options; 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 @@ -181,19 +200,9 @@ export class StreamableHTTPClientTransport implements Transport { } } - // Default reconnection options - private readonly _defaultReconnectionOptions: StreamableHTTPReconnectionOptions = { - initialReconnectionDelay: 1000, - maxReconnectionDelay: 30000, - reconnectionDelayGrowFactor: 1.5, - maxRetries: 2, - }; - - // We no longer need global reconnection state as it will be maintained per stream /** - * Calculates the next reconnection delay using exponential backoff algorithm - * with jitter for more effective reconnections in high load scenarios. + * Calculates the next reconnection delay using backoff algorithm * * @param attempt Current reconnection attempt count for the specific stream * @returns Time to wait in milliseconds before next reconnection attempt @@ -227,12 +236,11 @@ export class StreamableHTTPClientTransport implements Transport { // Calculate next delay based on current attempt count const delay = this._getNextReconnectionDelay(attemptCount); - log(`Reconnection attempt ${attemptCount + 1} in ${delay}ms...`); // Schedule the reconnection setTimeout(() => { // Use the last event ID to resume where we left off - this._startOrAuthStandaloneSSE(lastEventId).catch(error => { + this._startOrAuthStandaloneSSE({ lastEventId }).catch(error => { this.onerror?.(new Error(`Failed to reconnect SSE stream: ${error instanceof Error ? error.message : String(error)}`)); // Schedule another attempt if this one failed, incrementing the attempt counter this._scheduleReconnection(lastEventId, attemptCount + 1); @@ -247,7 +255,7 @@ export class StreamableHTTPClientTransport implements Transport { let lastEventId: string | undefined; const processStream = async () => { - // this is the closest we can get to trying to cath network errors + // this is the closest we can get to trying to catch network errors // if something happens reader will throw try { // Create a pipeline: binary stream -> text decoder -> SSE parser @@ -279,7 +287,7 @@ export class StreamableHTTPClientTransport implements Transport { } } catch (error) { // Handle stream errors - likely a network disconnect - this.onerror?.(new Error(`SSE stream disconnected: ${error instanceof Error ? error.message : String(error)}`)); + this.onerror?.(new Error(`SSE stream disconnected: ${error}`)); // Attempt to reconnect if the stream disconnects unexpectedly and we aren't closing if (this._abortController && !this._abortController.signal.aborted) { @@ -375,7 +383,7 @@ export class StreamableHTTPClientTransport implements Transport { // if it's supported by the server if (isJSONRPCNotification(message) && message.method === "notifications/initialized") { // Start without a lastEventId since this is a fresh connection - this._startOrAuthStandaloneSSE().catch(err => this.onerror?.(err)); + this._startOrAuthStandaloneSSE({ lastEventId: undefined }).catch(err => this.onerror?.(err)); } return; } @@ -390,9 +398,6 @@ export class StreamableHTTPClientTransport implements Transport { if (hasRequests) { if (contentType?.includes("text/event-stream")) { - // Handle SSE stream responses for requests - // We use the same handler as standalone streams, which now supports - // reconnection with the last event ID this._handleSseStream(response.body); } else if (contentType?.includes("application/json")) { // For non-streaming servers, we might get direct JSON responses