From e5167cbe2004e12cedbcaa3d2ba444f13ef80273 Mon Sep 17 00:00:00 2001 From: Cliff Hall Date: Tue, 1 Jul 2025 13:20:28 -0400 Subject: [PATCH] Update streamableHttp custom fetch test --- src/client/sse.test.ts | 30 ++++++++++++++++++++++++++++++ src/client/sse.ts | 13 +++++++++++-- src/client/streamableHttp.test.ts | 29 +++++++++++++++++++++++++++++ src/client/streamableHttp.ts | 15 ++++++++++++--- 4 files changed, 82 insertions(+), 5 deletions(-) diff --git a/src/client/sse.test.ts b/src/client/sse.test.ts index 3cb4e8a3..c446b9e9 100644 --- a/src/client/sse.test.ts +++ b/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", diff --git a/src/client/sse.ts b/src/client/sse.ts index 2546d508..5ca724b1 100644 --- a/src/client/sse.ts +++ b/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; + 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 { @@ -117,7 +126,7 @@ export class SSEClientTransport implements Transport { } private _startOrAuth(): Promise { - 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) { diff --git a/src/client/streamableHttp.test.ts b/src/client/streamableHttp.test.ts index 11dfe7d4..be70b8b4 100644 --- a/src/client/streamableHttp.test.ts +++ b/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 })._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 = { diff --git a/src/client/streamableHttp.ts b/src/client/streamableHttp.ts index 73078442..b73f67b9 100644 --- a/src/client/streamableHttp.ts +++ b/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; + // 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