From 4e1b3827e9c554d50ae4b7186ecde163f1f2ac4e Mon Sep 17 00:00:00 2001 From: JerryLee <223425819+Jerry2003826@users.noreply.github.com> Date: Tue, 2 Jun 2026 12:09:01 +1000 Subject: [PATCH] fix(core): tolerate unsupported Streamable HTTP GET SSE (#4521) Fixes #4326 --- packages/core/src/tools/mcp-client.test.ts | 251 ++++++++++++++++++++- packages/core/src/tools/mcp-client.ts | 130 ++++++++++- 2 files changed, 379 insertions(+), 2 deletions(-) diff --git a/packages/core/src/tools/mcp-client.test.ts b/packages/core/src/tools/mcp-client.test.ts index 2fe52d62f6..8bc2c78749 100644 --- a/packages/core/src/tools/mcp-client.test.ts +++ b/packages/core/src/tools/mcp-client.test.ts @@ -9,13 +9,15 @@ 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'; import type { PromptRegistry } from '../prompts/prompt-registry.js'; import type { WorkspaceContext } from '../utils/workspaceContext.js'; import { addMCPStatusChangeListener, + createStreamableHttpCompatibilityFetch, createTransport, getAllMCPServerStatuses, getMCPServerStatus, @@ -31,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', () => ({ @@ -41,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; @@ -273,6 +288,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 +309,186 @@ 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); + 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 () => { + 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() + .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 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 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() + .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); + }); + + 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); }); }); @@ -474,6 +671,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 () => { @@ -512,6 +711,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 52c82171af..e6b653d83e 100644 --- a/packages/core/src/tools/mcp-client.ts +++ b/packages/core/src/tools/mcp-client.ts @@ -62,6 +62,127 @@ 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(); + if (method !== 'GET') { + return false; + } + + 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()) + .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), +): 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; + } + + 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.` + + (responseBody ? ` Response body: ${JSON.stringify(responseBody)}` : ''), + ); + + return new Response(null, { + status: 405, + statusText: 'Method Not Allowed', + }); + }; +} + export type DiscoveredMCPPrompt = Prompt & { serverName: string; invoke: (params: Record) => Promise; @@ -498,6 +619,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 +1454,8 @@ export async function createTransport( }; if (mcpServerConfig.httpUrl) { + (transportOptions as StreamableHTTPClientTransportOptions).fetch = + createStreamableHttpCompatibilityFetch(mcpServerName); return new StreamableHTTPClientTransport( new URL(mcpServerConfig.httpUrl), transportOptions, @@ -1358,6 +1482,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 +1540,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) {