Update streamableHttp custom fetch test

Author: cliffhall

src/client/sse.test.ts

@@ -262,6 +262,36 @@ describe("SSEClientTransport", () => {
       expect(lastServerRequest.headers.authorization).toBe(authToken);
     });
 
+    it("uses custom fetch implementation from options", async () => {
+      const authToken = "Bearer custom-token";
+
+      const fetchWithAuth = jest.fn((url: string | URL, init?: RequestInit) => {
+        const headers = new Headers(init?.headers);
+        headers.set("Authorization", authToken);
+        return fetch(url.toString(), { ...init, headers });
+      });
+
+      transport = new SSEClientTransport(resourceBaseUrl, {
+        fetch: fetchWithAuth,
+      });
+
+      await transport.start();
+
+      expect(lastServerRequest.headers.authorization).toBe(authToken);
+
+      // Send a message to verify fetchWithAuth used for POST as well
+      const message: JSONRPCMessage = {
+        jsonrpc: "2.0",
+        id: "1",
+        method: "test",
+        params: {},
+      };
+
+      await transport.send(message);
+
+      expect(fetchWithAuth).toHaveBeenCalled();
+    });
+
     it("passes custom headers to fetch requests", async () => {
       const customHeaders = {
         Authorization: "Bearer test-token",

src/client/sse.ts

@@ -3,6 +3,8 @@ import { Transport } from "../shared/transport.js";
 import { JSONRPCMessage, JSONRPCMessageSchema } from "../types.js";
 import { auth, AuthResult, extractResourceMetadataUrl, OAuthClientProvider, UnauthorizedError } from "./auth.js";
 
+export type FetchLike = (url: string | URL, init?: RequestInit) => Promise<Response>;
+
 export class SseError extends Error {
   constructor(
     public readonly code: number | undefined,
@@ -47,6 +49,11 @@ export type SSEClientTransportOptions = {
    * Customizes recurring POST requests to the server.
    */
   requestInit?: RequestInit;
+
+  /**
+   * Custom fetch implementation used for all network requests.
+   */
+  fetch?: FetchLike;
 };
 
 /**
@@ -62,6 +69,7 @@ export class SSEClientTransport implements Transport {
   private _eventSourceInit?: EventSourceInit;
   private _requestInit?: RequestInit;
   private _authProvider?: OAuthClientProvider;
+  private _fetch?: FetchLike;
   private _protocolVersion?: string;
 
   onclose?: () => void;
@@ -77,6 +85,7 @@ export class SSEClientTransport implements Transport {
     this._eventSourceInit = opts?.eventSourceInit;
     this._requestInit = opts?.requestInit;
     this._authProvider = opts?.authProvider;
+    this._fetch = opts?.fetch;
   }
 
   private async _authThenStart(): Promise<void> {
@@ -117,7 +126,7 @@ export class SSEClientTransport implements Transport {
   }
 
   private _startOrAuth(): Promise<void> {
-    const fetchImpl = (this?._eventSourceInit?.fetch || fetch) as typeof fetch
+    const fetchImpl = (this?._eventSourceInit?.fetch || this._fetch || fetch) as typeof fetch
     return new Promise((resolve, reject) => {
       this._eventSource = new EventSource(
         this._url.href,
@@ -242,7 +251,7 @@ export class SSEClientTransport implements Transport {
         signal: this._abortController?.signal,
       };
 
-      const response = await fetch(this._endpoint, init);
+      const response = await (this._fetch ?? fetch)(this._endpoint, init);
       if (!response.ok) {
         if (response.status === 401 && this._authProvider) {
 

src/client/streamableHttp.test.ts

@@ -443,6 +443,35 @@ describe("StreamableHTTPClientTransport", () => {
     expect(errorSpy).toHaveBeenCalled();
   });
 
+  it("uses custom fetch implementation", async () => {
+    const authToken = "Bearer custom-token";
+
+    const fetchWithAuth = jest.fn((url: string | URL, init?: RequestInit) => {
+      const headers = new Headers(init?.headers);
+      headers.set("Authorization", authToken);
+      return (global.fetch as jest.Mock)(url, { ...init, headers });
+    });
+
+    (global.fetch as jest.Mock)
+      .mockResolvedValueOnce(
+        new Response(null, { status: 200, headers: { "content-type": "text/event-stream" } })
+      )
+      .mockResolvedValueOnce(new Response(null, { status: 202 }));
+
+    transport = new StreamableHTTPClientTransport(new URL("http://localhost:1234/mcp"), { fetch: fetchWithAuth });
+
+    await transport.start();
+    await (transport as unknown as { _startOrAuthSse: (opts: any) => Promise<void> })._startOrAuthSse({});
+
+    await transport.send({ jsonrpc: "2.0", method: "test", params: {}, id: "1" } as JSONRPCMessage);
+
+    expect(fetchWithAuth).toHaveBeenCalled();
+    for (const call of (global.fetch as jest.Mock).mock.calls) {
+      const headers = call[1].headers as Headers;
+      expect(headers.get("Authorization")).toBe(authToken);
+    }
+  });
+
 
   it("should always send specified custom headers", async () => {
     const requestInit = {

src/client/streamableHttp.ts

@@ -3,6 +3,8 @@ import { isInitializedNotification, isJSONRPCRequest, isJSONRPCResponse, JSONRPC
 import { auth, AuthResult, extractResourceMetadataUrl, OAuthClientProvider, UnauthorizedError } from "./auth.js";
 import { EventSourceParserStream } from "eventsource-parser/stream";
 
+export type FetchLike = (url: string | URL, init?: RequestInit) => Promise<Response>;
+
 // Default reconnection options for StreamableHTTP connections
 const DEFAULT_STREAMABLE_HTTP_RECONNECTION_OPTIONS: StreamableHTTPReconnectionOptions = {
   initialReconnectionDelay: 1000,
@@ -99,6 +101,11 @@ export type StreamableHTTPClientTransportOptions = {
    */
   requestInit?: RequestInit;
 
+  /**
+   * Custom fetch implementation used for all network requests.
+   */
+  fetch?: FetchLike;
+
   /**
    * Options to configure the reconnection behavior.
    */
@@ -122,6 +129,7 @@ export class StreamableHTTPClientTransport implements Transport {
   private _resourceMetadataUrl?: URL;
   private _requestInit?: RequestInit;
   private _authProvider?: OAuthClientProvider;
+  private _fetch?: FetchLike;
   private _sessionId?: string;
   private _reconnectionOptions: StreamableHTTPReconnectionOptions;
   private _protocolVersion?: string;
@@ -138,6 +146,7 @@ export class StreamableHTTPClientTransport implements Transport {
     this._resourceMetadataUrl = undefined;
     this._requestInit = opts?.requestInit;
     this._authProvider = opts?.authProvider;
+    this._fetch = opts?.fetch;
     this._sessionId = opts?.sessionId;
     this._reconnectionOptions = opts?.reconnectionOptions ?? DEFAULT_STREAMABLE_HTTP_RECONNECTION_OPTIONS;
   }
@@ -200,7 +209,7 @@ export class StreamableHTTPClientTransport implements Transport {
         headers.set("last-event-id", resumptionToken);
       }
 
-      const response = await fetch(this._url, {
+      const response = await (this._fetch ?? fetch)(this._url, {
         method: "GET",
         headers,
         signal: this._abortController?.signal,
@@ -414,7 +423,7 @@ export class StreamableHTTPClientTransport implements Transport {
         signal: this._abortController?.signal,
       };
 
-      const response = await fetch(this._url, init);
+      const response = await (this._fetch ?? fetch)(this._url, init);
 
       // Handle session ID received during initialization
       const sessionId = response.headers.get("mcp-session-id");
@@ -520,7 +529,7 @@ export class StreamableHTTPClientTransport implements Transport {
         signal: this._abortController?.signal,
       };
 
-      const response = await fetch(this._url, init);
+      const response = await (this._fetch ?? fetch)(this._url, init);
 
       // We specifically handle 405 as a valid response according to the spec,
       // meaning the server does not support explicit session termination