modelcontextprotocol/typescript-sdk
Watch
1
0
Fork
You've already forked typescript-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

address client comments

Browse source
This commit is contained in:
ihrpr 2025-04-15 09:24:11 +01:00
commit 583202f81a
No known key found for this signature in database
1 changed files with 29 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

53
src/client/streamableHttp.ts
View file

@ -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<void> {
@ -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<Headers> {
@ -138,7 +155,9 @@ export class StreamableHTTPClientTransport implements Transport {
);
}
private async _startOrAuthStandaloneSSE(lastEventId?: string): Promise<void> {
private async _startOrAuthStandaloneSSE(options: StartSSEOptions): Promise<void> {
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