From e6b14e1fc3144d8efdb597290a3a95960fb679c0 Mon Sep 17 00:00:00 2001 From: JerryLee <223425819+Jerry2003826@users.noreply.github.com> Date: Tue, 26 May 2026 09:00:49 +1000 Subject: [PATCH 1/4] fix(core): tolerate unsupported streamable http get sse --- packages/core/src/tools/mcp-client.test.ts | 57 ++++++++++++++++++++++ packages/core/src/tools/mcp-client.ts | 52 +++++++++++++++++++- 2 files changed, 108 insertions(+), 1 deletion(-) diff --git a/packages/core/src/tools/mcp-client.test.ts b/packages/core/src/tools/mcp-client.test.ts index 2fe52d62f63..36fbc84d83c 100644 --- a/packages/core/src/tools/mcp-client.test.ts +++ b/packages/core/src/tools/mcp-client.test.ts @@ -16,6 +16,7 @@ import type { PromptRegistry } from '../prompts/prompt-registry.js'; import type { WorkspaceContext } from '../utils/workspaceContext.js'; import { addMCPStatusChangeListener, + createStreamableHttpCompatibilityFetch, createTransport, getAllMCPServerStatuses, getMCPServerStatus, @@ -273,6 +274,8 @@ describe('mcp-client', () => { expect(transport).toBeInstanceOf(StreamableHTTPClientTransport); // eslint-disable-next-line @typescript-eslint/no-explicit-any expect((transport as any)._url).toEqual(new URL('http://test-server')); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + expect((transport as any)._fetch).toEqual(expect.any(Function)); }); it('with headers', async () => { @@ -292,6 +295,60 @@ describe('mcp-client', () => { expect((transport as any)._requestInit?.headers).toEqual({ Authorization: 'derp', }); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + expect((transport as any)._fetch).toEqual(expect.any(Function)); + }); + + it('treats 400 from optional GET SSE stream as unsupported', async () => { + const fetchFn = vi + .fn() + .mockResolvedValue(new Response('bad method', { status: 400 })); + const fetchWithFallback = createStreamableHttpCompatibilityFetch( + 'spring-ai', + fetchFn, + ); + + const response = await fetchWithFallback('http://test-server/mcp', { + method: 'GET', + headers: { Accept: 'text/event-stream' }, + }); + + expect(fetchFn).toHaveBeenCalledTimes(1); + expect(response.status).toBe(405); + }); + + it('does not hide Streamable HTTP GET SSE server errors', async () => { + const fetchFn = vi + .fn() + .mockResolvedValue(new Response('server exploded', { status: 502 })); + const fetchWithFallback = createStreamableHttpCompatibilityFetch( + 'server-error', + fetchFn, + ); + + const response = await fetchWithFallback('http://test-server/mcp', { + method: 'GET', + headers: { Accept: 'text/event-stream' }, + }); + + expect(response.status).toBe(502); + }); + + it('does not rewrite non-SSE GET responses', async () => { + const fetchFn = vi + .fn() + .mockResolvedValue(new Response('bad request', { status: 400 })); + const fetchWithFallback = createStreamableHttpCompatibilityFetch( + 'plain-get', + fetchFn, + ); + + const response = await fetchWithFallback('http://test-server/mcp', { + method: 'GET', + headers: { Accept: 'application/json' }, + }); + + expect(response.status).toBe(400); }); }); diff --git a/packages/core/src/tools/mcp-client.ts b/packages/core/src/tools/mcp-client.ts index 52c82171afd..a742f7dd677 100644 --- a/packages/core/src/tools/mcp-client.ts +++ b/packages/core/src/tools/mcp-client.ts @@ -62,6 +62,49 @@ export const MCP_DEFAULT_TIMEOUT_MSEC = 10 * 60 * 1000; // default to 10 minutes const debugLogger = createDebugLogger('MCP'); +const STREAMABLE_HTTP_GET_SSE_FALLBACK_STATUSES = new Set([400, 405]); + +function isStreamableHttpGetSseRequest(init?: RequestInit): boolean { + const method = (init?.method ?? 'GET').toUpperCase(); + if (method !== 'GET') { + return false; + } + + const accept = new Headers(init?.headers).get('accept') ?? ''; + return accept + .split(',') + .some((value) => value.trim().toLowerCase() === 'text/event-stream'); +} + +export function createStreamableHttpCompatibilityFetch( + mcpServerName: string, + fetchFn: typeof fetch = globalThis.fetch.bind(globalThis), +): typeof fetch { + return async (input, init) => { + const response = await fetchFn(input, init); + if ( + !isStreamableHttpGetSseRequest(init) || + !STREAMABLE_HTTP_GET_SSE_FALLBACK_STATUSES.has(response.status) + ) { + return response; + } + + await response.body?.cancel().catch(() => { + // Best-effort body cleanup before returning a synthetic 405. + }); + debugLogger.warn( + `MCP server '${mcpServerName}' rejected the optional Streamable HTTP ` + + `GET SSE stream with HTTP ${response.status}; continuing without ` + + `the standalone GET stream. POST request streams remain enabled.`, + ); + + return new Response(null, { + status: 405, + statusText: 'Method Not Allowed', + }); + }; +} + export type DiscoveredMCPPrompt = Prompt & { serverName: string; invoke: (params: Record) => Promise; @@ -498,6 +541,7 @@ async function createTransportWithOAuth( if (mcpServerConfig.httpUrl) { // Create HTTP transport with OAuth token const oauthTransportOptions: StreamableHTTPClientTransportOptions = { + fetch: createStreamableHttpCompatibilityFetch(mcpServerName), requestInit: { headers: { ...mcpServerConfig.headers, @@ -1332,6 +1376,8 @@ export async function createTransport( }; if (mcpServerConfig.httpUrl) { + (transportOptions as StreamableHTTPClientTransportOptions).fetch = + createStreamableHttpCompatibilityFetch(mcpServerName); return new StreamableHTTPClientTransport( new URL(mcpServerConfig.httpUrl), transportOptions, @@ -1358,6 +1404,8 @@ export async function createTransport( authProvider: provider, }; if (mcpServerConfig.httpUrl) { + (transportOptions as StreamableHTTPClientTransportOptions).fetch = + createStreamableHttpCompatibilityFetch(mcpServerName); return new StreamableHTTPClientTransport( new URL(mcpServerConfig.httpUrl), transportOptions, @@ -1414,7 +1462,9 @@ export async function createTransport( } if (mcpServerConfig.httpUrl) { - const transportOptions: StreamableHTTPClientTransportOptions = {}; + const transportOptions: StreamableHTTPClientTransportOptions = { + fetch: createStreamableHttpCompatibilityFetch(mcpServerName), + }; // Set up headers with OAuth token if available if (hasOAuthConfig && accessToken) { From 2e80fb45fe5d141fa5844635b4a3c1a5e482cf6d Mon Sep 17 00:00:00 2001 From: JerryLee <223425819+Jerry2003826@users.noreply.github.com> Date: Tue, 26 May 2026 12:52:28 +1000 Subject: [PATCH 2/4] test(core): cover streamable http post passthrough --- packages/core/src/tools/mcp-client.test.ts | 89 ++++++++++++++++++++++ packages/core/src/tools/mcp-client.ts | 11 ++- 2 files changed, 99 insertions(+), 1 deletion(-) diff --git a/packages/core/src/tools/mcp-client.test.ts b/packages/core/src/tools/mcp-client.test.ts index 36fbc84d83c..6f172b5ffc1 100644 --- a/packages/core/src/tools/mcp-client.test.ts +++ b/packages/core/src/tools/mcp-client.test.ts @@ -12,6 +12,7 @@ import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/ import { afterEach, describe, expect, it, vi } from 'vitest'; import { AuthProviderType, type Config } from '../config/config.js'; import { GoogleCredentialProvider } from '../mcp/google-auth-provider.js'; +import { MCPOAuthProvider } from '../mcp/oauth-provider.js'; import type { PromptRegistry } from '../prompts/prompt-registry.js'; import type { WorkspaceContext } from '../utils/workspaceContext.js'; import { @@ -317,6 +318,25 @@ describe('mcp-client', () => { expect(response.status).toBe(405); }); + it('treats parameterized GET SSE Accept headers as unsupported', async () => { + const fetchFn = vi + .fn() + .mockResolvedValue(new Response('bad method', { status: 400 })); + const fetchWithFallback = createStreamableHttpCompatibilityFetch( + 'spring-ai', + fetchFn, + ); + + const response = await fetchWithFallback('http://test-server/mcp', { + method: 'GET', + headers: { + Accept: 'application/json, text/event-stream; charset=utf-8', + }, + }); + + expect(response.status).toBe(405); + }); + it('does not hide Streamable HTTP GET SSE server errors', async () => { const fetchFn = vi .fn() @@ -350,6 +370,23 @@ describe('mcp-client', () => { expect(response.status).toBe(400); }); + + it('does not rewrite POST responses', async () => { + const fetchFn = vi + .fn() + .mockResolvedValue(new Response('bad request', { status: 400 })); + const fetchWithFallback = createStreamableHttpCompatibilityFetch( + 'post-test', + fetchFn, + ); + + const response = await fetchWithFallback('http://test-server/mcp', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + }); + + expect(response.status).toBe(400); + }); }); describe('should connect via url', () => { @@ -531,6 +568,8 @@ describe('mcp-client', () => { // eslint-disable-next-line @typescript-eslint/no-explicit-any const authProvider = (transport as any)._authProvider; expect(authProvider).toBeInstanceOf(GoogleCredentialProvider); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + expect((transport as any)._fetch).toEqual(expect.any(Function)); }); it('should use GoogleCredentialProvider with SSE transport', async () => { @@ -569,6 +608,56 @@ describe('mcp-client', () => { ); }); }); + + describe('authenticated Streamable HTTP compatibility fetch', () => { + it('wires the compatibility fetch for OAuth httpUrl transports', async () => { + const getValidToken = vi.fn().mockResolvedValue('oauth-token'); + vi.mocked(MCPOAuthProvider).mockImplementation( + () => + ({ + getValidToken, + }) as unknown as MCPOAuthProvider, + ); + + const transport = await createTransport( + 'oauth-test-server', + { + httpUrl: 'http://test-server', + oauth: { + enabled: true, + clientId: 'client-id', + }, + }, + false, + ); + + expect(transport).toBeInstanceOf(StreamableHTTPClientTransport); + expect(getValidToken).toHaveBeenCalledWith('oauth-test-server', { + enabled: true, + clientId: 'client-id', + }); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + expect((transport as any)._fetch).toEqual(expect.any(Function)); + }); + + it('wires the compatibility fetch for service account httpUrl transports', async () => { + const transport = await createTransport( + 'service-account-test-server', + { + httpUrl: 'http://test-server', + authProviderType: AuthProviderType.SERVICE_ACCOUNT_IMPERSONATION, + targetAudience: 'client.apps.googleusercontent.com', + targetServiceAccount: + 'service-account@example-project.iam.gserviceaccount.com', + }, + false, + ); + + expect(transport).toBeInstanceOf(StreamableHTTPClientTransport); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + expect((transport as any)._fetch).toEqual(expect.any(Function)); + }); + }); }); describe('isEnabled', () => { const funcDecl = { name: 'myTool' }; diff --git a/packages/core/src/tools/mcp-client.ts b/packages/core/src/tools/mcp-client.ts index a742f7dd677..72a388cadc5 100644 --- a/packages/core/src/tools/mcp-client.ts +++ b/packages/core/src/tools/mcp-client.ts @@ -73,9 +73,18 @@ function isStreamableHttpGetSseRequest(init?: RequestInit): boolean { const accept = new Headers(init?.headers).get('accept') ?? ''; return accept .split(',') - .some((value) => value.trim().toLowerCase() === 'text/event-stream'); + .map((value) => value.split(';')[0].trim().toLowerCase()) + .some((type) => type === 'text/event-stream'); } +/** + * Wraps fetch to normalize Spring AI-style 400 responses to the SDK's + * unsupported sentinel for the optional Streamable HTTP GET SSE request. + * + * SDK coupling: `StreamableHTTPClientTransport._startOrAuthSse()` treats a + * 405 response as "GET SSE unsupported" and continues in POST-only mode. + * If the SDK changes that non-OK handling, update this wrapper in lockstep. + */ export function createStreamableHttpCompatibilityFetch( mcpServerName: string, fetchFn: typeof fetch = globalThis.fetch.bind(globalThis), From ba3e01e11046ecee5460704488baeea6f20233e8 Mon Sep 17 00:00:00 2001 From: JerryLee <223425819+Jerry2003826@users.noreply.github.com> Date: Tue, 26 May 2026 23:41:34 +1000 Subject: [PATCH 3/4] fix(core): leave streamable HTTP 405 responses intact --- packages/core/src/tools/mcp-client.test.ts | 20 ++++++++++++++++++++ packages/core/src/tools/mcp-client.ts | 2 +- 2 files changed, 21 insertions(+), 1 deletion(-) diff --git a/packages/core/src/tools/mcp-client.test.ts b/packages/core/src/tools/mcp-client.test.ts index 6f172b5ffc1..4742eb0e489 100644 --- a/packages/core/src/tools/mcp-client.test.ts +++ b/packages/core/src/tools/mcp-client.test.ts @@ -354,6 +354,26 @@ describe('mcp-client', () => { expect(response.status).toBe(502); }); + it('does not rewrite the SDK-native GET SSE unsupported sentinel', async () => { + const fetchFn = vi + .fn() + .mockResolvedValue( + new Response('method not allowed', { status: 405 }), + ); + const fetchWithFallback = createStreamableHttpCompatibilityFetch( + 'native-unsupported', + fetchFn, + ); + + const response = await fetchWithFallback('http://test-server/mcp', { + method: 'GET', + headers: { Accept: 'text/event-stream' }, + }); + + expect(response.status).toBe(405); + expect(await response.text()).toBe('method not allowed'); + }); + it('does not rewrite non-SSE GET responses', async () => { const fetchFn = vi .fn() diff --git a/packages/core/src/tools/mcp-client.ts b/packages/core/src/tools/mcp-client.ts index 72a388cadc5..ff065b686bc 100644 --- a/packages/core/src/tools/mcp-client.ts +++ b/packages/core/src/tools/mcp-client.ts @@ -62,7 +62,7 @@ export const MCP_DEFAULT_TIMEOUT_MSEC = 10 * 60 * 1000; // default to 10 minutes const debugLogger = createDebugLogger('MCP'); -const STREAMABLE_HTTP_GET_SSE_FALLBACK_STATUSES = new Set([400, 405]); +const STREAMABLE_HTTP_GET_SSE_FALLBACK_STATUSES = new Set([400]); function isStreamableHttpGetSseRequest(init?: RequestInit): boolean { const method = (init?.method ?? 'GET').toUpperCase(); From e466b6a9c796d64b3a3874e058b312569f012dcc Mon Sep 17 00:00:00 2001 From: JerryLee <223425819+Jerry2003826@users.noreply.github.com> Date: Wed, 27 May 2026 02:34:37 +1000 Subject: [PATCH 4/4] fix(core): preserve resumable streamable HTTP SSE errors --- packages/core/src/tools/mcp-client.test.ts | 85 +++++++++++++++++++++- packages/core/src/tools/mcp-client.ts | 73 ++++++++++++++++++- 2 files changed, 155 insertions(+), 3 deletions(-) diff --git a/packages/core/src/tools/mcp-client.test.ts b/packages/core/src/tools/mcp-client.test.ts index 4742eb0e489..8bc2c78749f 100644 --- a/packages/core/src/tools/mcp-client.test.ts +++ b/packages/core/src/tools/mcp-client.test.ts @@ -9,7 +9,7 @@ import * as ClientLib from '@modelcontextprotocol/sdk/client/index.js'; import { SSEClientTransport } from '@modelcontextprotocol/sdk/client/sse.js'; import * as SdkClientStdioLib from '@modelcontextprotocol/sdk/client/stdio.js'; import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js'; -import { afterEach, describe, expect, it, vi } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { AuthProviderType, type Config } from '../config/config.js'; import { GoogleCredentialProvider } from '../mcp/google-auth-provider.js'; import { MCPOAuthProvider } from '../mcp/oauth-provider.js'; @@ -33,6 +33,12 @@ import { import type { ToolRegistry } from './tool-registry.js'; const mockExistsSync = vi.hoisted(() => vi.fn(() => true)); +const mockDebugLogger = vi.hoisted(() => ({ + debug: vi.fn(), + error: vi.fn(), + info: vi.fn(), + warn: vi.fn(), +})); const ORIGINAL_ENV = process.env; vi.mock('node:fs', () => ({ @@ -43,8 +49,15 @@ vi.mock('@modelcontextprotocol/sdk/client/index.js'); vi.mock('@google/genai'); vi.mock('../mcp/oauth-provider.js'); vi.mock('../mcp/oauth-token-storage.js'); +vi.mock('../utils/debugLogger.js', () => ({ + createDebugLogger: vi.fn(() => mockDebugLogger), +})); describe('mcp-client', () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + afterEach(() => { vi.restoreAllMocks(); process.env = ORIGINAL_ENV; @@ -316,6 +329,53 @@ describe('mcp-client', () => { expect(fetchFn).toHaveBeenCalledTimes(1); expect(response.status).toBe(405); + expect(response.statusText).toBe('Method Not Allowed'); + expect(await response.text()).toBe(''); + }); + + it('omits response body diagnostics when the fallback body is empty', async () => { + const fetchFn = vi + .fn() + .mockResolvedValue(new Response(null, { status: 400 })); + const fetchWithFallback = createStreamableHttpCompatibilityFetch( + 'empty-body', + fetchFn, + ); + + const response = await fetchWithFallback('http://test-server/mcp', { + method: 'GET', + headers: { Accept: 'text/event-stream' }, + }); + + expect(response.status).toBe(405); + expect(mockDebugLogger.warn).toHaveBeenCalledWith( + expect.not.stringContaining('Response body:'), + ); + }); + + it('truncates Streamable HTTP GET SSE error body diagnostics', async () => { + const fetchFn = vi + .fn() + .mockResolvedValue(new Response('x'.repeat(1024), { status: 400 })); + const fetchWithFallback = createStreamableHttpCompatibilityFetch( + 'large-error-body', + fetchFn, + ); + + const response = await fetchWithFallback('http://test-server/mcp', { + method: 'GET', + headers: { Accept: 'text/event-stream' }, + }); + + expect(response.status).toBe(405); + expect(mockDebugLogger.warn).toHaveBeenCalledWith( + expect.stringContaining( + `Response body: ${JSON.stringify(`${'x'.repeat(512)}...`)}`, + ), + ); + expect(mockDebugLogger.warn).not.toHaveBeenCalledWith( + expect.stringContaining('x'.repeat(600)), + ); }); it('treats parameterized GET SSE Accept headers as unsupported', async () => { @@ -374,6 +434,29 @@ describe('mcp-client', () => { expect(await response.text()).toBe('method not allowed'); }); + it('does not rewrite resumable GET SSE errors with Last-Event-ID', async () => { + const fetchFn = vi + .fn() + .mockResolvedValue( + new Response('{"error":"invalid cursor"}', { status: 400 }), + ); + const fetchWithFallback = createStreamableHttpCompatibilityFetch( + 'resume-error', + fetchFn, + ); + + const response = await fetchWithFallback('http://test-server/mcp', { + method: 'GET', + headers: { + Accept: 'text/event-stream', + 'Last-Event-ID': 'event-123', + }, + }); + + expect(response.status).toBe(400); + expect(await response.text()).toBe('{"error":"invalid cursor"}'); + }); + it('does not rewrite non-SSE GET responses', async () => { const fetchFn = vi .fn() diff --git a/packages/core/src/tools/mcp-client.ts b/packages/core/src/tools/mcp-client.ts index ff065b686bc..e6b653d83e5 100644 --- a/packages/core/src/tools/mcp-client.ts +++ b/packages/core/src/tools/mcp-client.ts @@ -63,6 +63,68 @@ export const MCP_DEFAULT_TIMEOUT_MSEC = 10 * 60 * 1000; // default to 10 minutes const debugLogger = createDebugLogger('MCP'); const STREAMABLE_HTTP_GET_SSE_FALLBACK_STATUSES = new Set([400]); +const STREAMABLE_HTTP_GET_SSE_ERROR_BODY_LIMIT = 512; + +async function readResponseBodyExcerpt( + response: Response, +): Promise { + const reader = response.clone().body?.getReader(); + if (!reader) { + return undefined; + } + + const decoder = new TextDecoder(); + let body = ''; + let bytesRead = 0; + let truncated = false; + try { + while (bytesRead < STREAMABLE_HTTP_GET_SSE_ERROR_BODY_LIMIT) { + const { done, value } = await reader.read(); + if (done) { + body += decoder.decode(); + break; + } + + const remaining = STREAMABLE_HTTP_GET_SSE_ERROR_BODY_LIMIT - bytesRead; + if (value.byteLength > remaining) { + body += decoder.decode(value.subarray(0, remaining), { + stream: true, + }); + bytesRead += remaining; + truncated = true; + reader.cancel().catch(() => { + // Best-effort cleanup after collecting the bounded excerpt. + }); + break; + } + + body += decoder.decode(value, { stream: true }); + bytesRead += value.byteLength; + } + + if (bytesRead >= STREAMABLE_HTTP_GET_SSE_ERROR_BODY_LIMIT && !truncated) { + const { done } = await reader.read(); + if (!done) { + truncated = true; + reader.cancel().catch(() => { + // Best-effort cleanup after collecting the bounded excerpt. + }); + } + } + + body += decoder.decode(); + const excerpt = body.trim(); + if (!excerpt) { + return undefined; + } + return truncated || + excerpt.length > STREAMABLE_HTTP_GET_SSE_ERROR_BODY_LIMIT + ? `${excerpt.slice(0, STREAMABLE_HTTP_GET_SSE_ERROR_BODY_LIMIT)}...` + : excerpt; + } catch { + return undefined; + } +} function isStreamableHttpGetSseRequest(init?: RequestInit): boolean { const method = (init?.method ?? 'GET').toUpperCase(); @@ -70,7 +132,12 @@ function isStreamableHttpGetSseRequest(init?: RequestInit): boolean { return false; } - const accept = new Headers(init?.headers).get('accept') ?? ''; + const headers = new Headers(init?.headers); + if (headers.has('last-event-id')) { + return false; + } + + const accept = headers.get('accept') ?? ''; return accept .split(',') .map((value) => value.split(';')[0].trim().toLowerCase()) @@ -98,13 +165,15 @@ export function createStreamableHttpCompatibilityFetch( return response; } + const responseBody = await readResponseBodyExcerpt(response); await response.body?.cancel().catch(() => { // Best-effort body cleanup before returning a synthetic 405. }); debugLogger.warn( `MCP server '${mcpServerName}' rejected the optional Streamable HTTP ` + `GET SSE stream with HTTP ${response.status}; continuing without ` + - `the standalone GET stream. POST request streams remain enabled.`, + `the standalone GET stream. POST request streams remain enabled.` + + (responseBody ? ` Response body: ${JSON.stringify(responseBody)}` : ''), ); return new Response(null, {