Merge branch 'resumability' into ihrpr/resume-long-running-requests

Author: ihrpr

src/client/streamableHttp.test.ts

@@ -164,7 +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["_startOrAuthSse"]()).resolves.not.toThrow("Failed to open SSE stream: Method Not Allowed");
+    await expect(transport["_startOrAuthSse"]({})).resolves.not.toThrow("Failed to open SSE stream: Method Not Allowed");
     // Check that GET was attempted
     expect(global.fetch).toHaveBeenCalledWith(
       expect.anything(),
@@ -208,7 +208,7 @@ describe("StreamableHTTPClientTransport", () => {
     transport.onmessage = messageSpy;
 
     await transport.start();
-    await transport["_startOrAuthSse"]();
+    await transport["_startOrAuthSse"]({});
 
     // Give time for the SSE event to be processed
     await new Promise(resolve => setTimeout(resolve, 50));
@@ -289,10 +289,10 @@ describe("StreamableHTTPClientTransport", () => {
     // Verify options were set correctly (checking implementation details)
     // Access private properties for testing
     const transportInstance = transport as unknown as {
-      _defaultReconnectionOptions: StreamableHTTPReconnectionOptions;
+      _reconnectionOptions: StreamableHTTPReconnectionOptions;
     };
-    expect(transportInstance._defaultReconnectionOptions.initialReconnectionDelay).toBe(1000);
-    expect(transportInstance._defaultReconnectionOptions.maxRetries).toBe(2);
+    expect(transportInstance._reconnectionOptions.initialReconnectionDelay).toBe(500);
+    expect(transportInstance._reconnectionOptions.maxRetries).toBe(5);
   });
 
   it("should pass lastEventId when reconnecting", async () => {
@@ -313,9 +313,9 @@ describe("StreamableHTTPClientTransport", () => {
     await transport.start();
     // Type assertion to access private method
     const transportWithPrivateMethods = transport as unknown as {
-      _startOrAuthSse: (lastEventId?: string) => Promise<void>
+      _startOrAuthSse: (options: { lastEventId?: string }) => Promise<void>
     };
-    await transportWithPrivateMethods._startOrAuthSse("test-event-id");
+    await transportWithPrivateMethods._startOrAuthSse({ lastEventId: "test-event-id" });
 
     // Verify fetch was called with the lastEventId header
     expect(fetchSpy).toHaveBeenCalled();
@@ -382,7 +382,7 @@ describe("StreamableHTTPClientTransport", () => {
 
     await transport.start();
 
-    await transport["_startOrAuthSse"]();
+    await transport["_startOrAuthSse"]({});
     expect((actualReqInit.headers as Headers).get("x-custom-header")).toBe("CustomValue");
 
     requestInit.headers["X-Custom-Header"] = "SecondCustomValue";

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;
 }
@@ -102,8 +119,8 @@ export class StreamableHTTPClientTransport implements Transport {
     this._url = url;
     this._requestInit = opts?.requestInit;
     this._authProvider = opts?.authProvider;
-    this._reconnectionOptions = opts?.reconnectionOptions || this._defaultReconnectionOptions;
     this._sessionId = opts?.sessionId;
+    this._reconnectionOptions = opts?.reconnectionOptions ?? DEFAULT_STREAMABLE_HTTP_RECONNECTION_OPTIONS;
   }
 
   private async _authThenStart(): Promise<void> {
@@ -123,7 +140,7 @@ export class StreamableHTTPClientTransport implements Transport {
       throw new UnauthorizedError();
     }
 
-    return await this._startOrAuthSse();
+    return await this._startOrAuthSse({ lastEventId: undefined });
   }
 
   private async _commonHeaders(): Promise<Headers> {
@@ -144,7 +161,9 @@ export class StreamableHTTPClientTransport implements Transport {
     );
   }
 
-  private async _startOrAuthSse(lastEventId?: string): Promise<void> {
+
+  private async _startOrAuthSse(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
@@ -187,19 +206,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
@@ -233,12 +242,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._startOrAuthSse(lastEventId).catch(error => {
+      this._startOrAuthSse({ 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);
@@ -253,7 +261,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
@@ -286,7 +294,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) {
@@ -343,7 +351,7 @@ export class StreamableHTTPClientTransport implements Transport {
       const { lastEventId, onLastEventIdUpdate } = options ?? {};
       if (lastEventId) {
         // If we have at last event ID, we need to reconnect the SSE stream
-        this._startOrAuthSse(lastEventId).catch(err => this.onerror?.(err));
+        this._startOrAuthSse({ lastEventId }).catch(err => this.onerror?.(err));
         return;
       }
 
@@ -390,7 +398,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._startOrAuthSse().catch(err => this.onerror?.(err));
+          this._startOrAuthSse({ lastEventId: undefined }).catch(err => this.onerror?.(err));
         }
         return;
       }

src/examples/server/simpleStreamableHttp.ts

@@ -12,18 +12,18 @@ class InMemoryEventStore implements EventStore {
   /**
    * Generates a unique event ID for a given stream ID
    */
-  generateEventId(streamId: string): string {
+  private generateEventId(streamId: string): string {
     return `${streamId}_${Date.now()}_${Math.random().toString(36).substring(2, 10)}`;
   }
 
-  getStreamIdFromEventId(eventId: string): string {
+  private getStreamIdFromEventId(eventId: string): string {
     const parts = eventId.split('_');
     return parts.length > 0 ? parts[0] : '';
   }
 
-
   /**
    * Stores an event with a generated event ID
+   * Implements EventStore.storeEvent
    */
   async storeEvent(streamId: string, message: JSONRPCMessage): Promise<string> {
     const eventId = this.generateEventId(streamId);
@@ -33,17 +33,26 @@ class InMemoryEventStore implements EventStore {
   }
 
   /**
-   * Retrieves events that occurred after a specific event
+   * Replays events that occurred after a specific event ID
+   * Implements EventStore.replayEventsAfter
    */
-  async getEventsAfter(lastEventId: string): Promise<Array<{ eventId: string, message: JSONRPCMessage }>> {
+  async replayEventsAfter(lastEventId: string, 
+    { send }: { send: (eventId: string, message: JSONRPCMessage) => Promise<void> }
+  ): Promise<string> {
     if (!lastEventId || !this.events.has(lastEventId)) {
-      return [];
+      console.log(`No events found for lastEventId: ${lastEventId}`);
+      return '';
     }
 
     // Extract the stream ID from the event ID
-    const streamId = lastEventId.split('_')[0];
-    const result: Array<{ eventId: string, message: JSONRPCMessage }> = [];
+    const streamId = this.getStreamIdFromEventId(lastEventId);
+    if (!streamId) {
+      console.log(`Could not extract streamId from lastEventId: ${lastEventId}`);
+      return '';
+    }
+
     let foundLastEvent = false;
+    let eventCount = 0;
 
     // Sort events by eventId for chronological ordering
     const sortedEvents = [...this.events.entries()].sort((a, b) => a[0].localeCompare(b[0]));
@@ -54,21 +63,21 @@ class InMemoryEventStore implements EventStore {
         continue;
       }
 
-      // Start collecting events after we find the lastEventId
+      // Start sending events after we find the lastEventId
       if (eventId === lastEventId) {
         foundLastEvent = true;
         continue;
       }
 
       if (foundLastEvent) {
-        result.push({ eventId, message });
+        await send(eventId, message);
+        eventCount++;
       }
     }
 
-    console.log(`Found ${result.length} events after ${lastEventId} for replay`);
-    return result;
+    console.log(`Replayed ${eventCount} events after ${lastEventId} for stream ${streamId}`);
+    return streamId;
   }
-
 }
 
 // Create an MCP server with implementation details

src/integration-tests/taskResumability.test.ts

@@ -14,11 +14,11 @@ import { z } from 'zod';
 class InMemoryEventStore implements EventStore {
   private events: Map<string, { streamId: string, message: JSONRPCMessage }> = new Map();
 
-  generateEventId(streamId: string): string {
+  private generateEventId(streamId: string): string {
     return `${streamId}_${Date.now()}_${Math.random().toString(36).substring(2, 10)}`;
   }
 
-  getStreamIdFromEventId(eventId: string): string {
+  private getStreamIdFromEventId(eventId: string): string {
     const parts = eventId.split('_');
     return parts.length > 0 ? parts[0] : '';
   }
@@ -29,14 +29,18 @@ class InMemoryEventStore implements EventStore {
     return eventId;
   }
 
-  async getEventsAfter(lastEventId: string): Promise<Array<{ eventId: string, message: JSONRPCMessage }>> {
+  async replayEventsAfter(lastEventId: string,
+    { send }: { send: (eventId: string, message: JSONRPCMessage) => Promise<void> }
+  ): Promise<string> {
     if (!lastEventId || !this.events.has(lastEventId)) {
-      return [];
+      return '';
     }
 
     // Extract the stream ID from the event ID
     const streamId = this.getStreamIdFromEventId(lastEventId);
-    const result: Array<{ eventId: string, message: JSONRPCMessage }> = [];
+    if (!streamId) {
+      return '';
+    }
     let foundLastEvent = false;
 
     // Sort events by eventId for chronological ordering
@@ -55,11 +59,11 @@ class InMemoryEventStore implements EventStore {
       }
 
       if (foundLastEvent) {
-        result.push({ eventId, message });
+        await send(eventId, message);
       }
     }
 
-    return result;
+    return streamId;
   }
 }
 

src/server/streamableHttp.test.ts

@@ -1,7 +1,7 @@
 import { createServer, type Server, IncomingMessage, ServerResponse } from "node:http";
 import { AddressInfo } from "node:net";
 import { randomUUID } from "node:crypto";
-import { EventStore, StreamableHTTPServerTransport } from "./streamableHttp.js";
+import { EventStore, StreamableHTTPServerTransport, EventId, StreamId } from "./streamableHttp.js";
 import { McpServer } from "./mcp.js";
 import { CallToolResult, JSONRPCMessage } from "../types.js";
 import { z } from "zod";
@@ -912,31 +912,25 @@ describe("StreamableHTTPServerTransport with resumability", () => {
 
   // Simple implementation of EventStore
   const eventStore: EventStore = {
-    generateEventId(streamId: string): string {
-      return `${streamId}_${randomUUID()}`;
-    },
-    getStreamIdFromEventId(eventId: string): string {
-      return eventId.split('_')[0]; // Extract stream ID from the event ID
-    },
+
     async storeEvent(streamId: string, message: JSONRPCMessage): Promise<string> {
-      const eventId = this.generateEventId(streamId);
+      const eventId = `${streamId}_${randomUUID()}`;
       storedEvents.set(eventId, { eventId, message });
       return eventId;
     },
 
-    async getEventsAfter(lastEventId: string): Promise<Array<{ eventId: string, message: JSONRPCMessage }>> {
-      const streamId = lastEventId.split('_')[0]; // Extract stream ID from the event ID
-      const result: Array<{ eventId: string, message: JSONRPCMessage }> = [];
-
+    async replayEventsAfter(lastEventId: EventId, { send }: {
+      send: (eventId: EventId, message: JSONRPCMessage) => Promise<void>
+    }): Promise<StreamId> {
+      const streamId = lastEventId.split('_')[0];
+      // Extract stream ID from the event ID
       // For test simplicity, just return all events with matching streamId that aren't the lastEventId
-      // This avoids issues with event ordering in tests
       for (const [eventId, { message }] of storedEvents.entries()) {
         if (eventId.startsWith(streamId) && eventId !== lastEventId) {
-          result.push({ eventId, message });
+          await send(eventId, message);
         }
       }
-
-      return result;
+      return streamId;
     },
   };