From d456916f371439243c2e8299176e9c7df7e74ba6 Mon Sep 17 00:00:00 2001 From: cyphercodes Date: Wed, 20 May 2026 06:51:19 +0300 Subject: [PATCH 1/2] fix(core): add streamable HTTP SSE POST fallback --- packages/core/src/tools/mcp-client.test.ts | 115 +++++++++++++++++++++ packages/core/src/tools/mcp-client.ts | 91 +++++++++++++++- 2 files changed, 201 insertions(+), 5 deletions(-) diff --git a/packages/core/src/tools/mcp-client.test.ts b/packages/core/src/tools/mcp-client.test.ts index 2fe52d62f63..08377ab9070 100644 --- a/packages/core/src/tools/mcp-client.test.ts +++ b/packages/core/src/tools/mcp-client.test.ts @@ -9,6 +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 type { FetchLike } from '@modelcontextprotocol/sdk/shared/transport.js'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { AuthProviderType, type Config } from '../config/config.js'; import { GoogleCredentialProvider } from '../mcp/google-auth-provider.js'; @@ -33,6 +34,12 @@ import type { ToolRegistry } from './tool-registry.js'; const mockExistsSync = vi.hoisted(() => vi.fn(() => true)); const ORIGINAL_ENV = process.env; +function getStreamableHttpFetch( + transport: StreamableHTTPClientTransport, +): FetchLike { + return (transport as unknown as { _fetch: FetchLike })._fetch; +} + vi.mock('node:fs', () => ({ existsSync: mockExistsSync, })); @@ -293,6 +300,114 @@ describe('mcp-client', () => { Authorization: 'derp', }); }); + + it('retries Streamable HTTP SSE GET 400 responses with POST', async () => { + const fetchSpy = vi + .spyOn(globalThis, 'fetch') + .mockResolvedValueOnce( + new Response('bad request', { + status: 400, + statusText: 'Bad Request', + }), + ) + .mockResolvedValueOnce( + new Response('', { + status: 200, + headers: { 'content-type': 'text/event-stream' }, + }), + ); + const transport = (await createTransport( + 'spring-ai-server', + { + httpUrl: 'http://test-server/mcp', + }, + false, + )) as StreamableHTTPClientTransport; + + const response = await getStreamableHttpFetch(transport)( + new URL('http://test-server/mcp'), + { + method: 'GET', + headers: { + Accept: 'text/event-stream', + 'mcp-session-id': 'session-1', + }, + }, + ); + + expect(response.status).toBe(200); + expect(fetchSpy).toHaveBeenCalledTimes(2); + const retryInit = fetchSpy.mock.calls[1]?.[1]; + expect(retryInit?.method).toBe('POST'); + const retryHeaders = new Headers(retryInit?.headers); + expect(retryHeaders.get('accept')).toBe( + 'text/event-stream, application/json', + ); + expect(retryHeaders.get('mcp-session-id')).toBe('session-1'); + }); + + it('retries Streamable HTTP SSE GET 405 responses with POST', async () => { + const fetchSpy = vi + .spyOn(globalThis, 'fetch') + .mockResolvedValueOnce( + new Response('method not allowed', { + status: 405, + statusText: 'Method Not Allowed', + }), + ) + .mockResolvedValueOnce( + new Response('', { + status: 200, + headers: { 'content-type': 'text/event-stream' }, + }), + ); + const transport = (await createTransport( + 'method-fallback-server', + { + httpUrl: 'http://test-server/mcp', + }, + false, + )) as StreamableHTTPClientTransport; + + const response = await getStreamableHttpFetch(transport)( + new URL('http://test-server/mcp'), + { + method: 'GET', + headers: { Accept: 'text/event-stream' }, + }, + ); + + expect(response.status).toBe(200); + expect(fetchSpy).toHaveBeenCalledTimes(2); + expect(fetchSpy.mock.calls[1]?.[1]?.method).toBe('POST'); + }); + + it('does not retry Streamable HTTP SSE GET 5xx responses', async () => { + const fetchSpy = vi.spyOn(globalThis, 'fetch').mockResolvedValueOnce( + new Response('server error', { + status: 502, + statusText: 'Bad Gateway', + }), + ); + const transport = (await createTransport( + 'server-error', + { + httpUrl: 'http://test-server/mcp', + }, + false, + )) as StreamableHTTPClientTransport; + + const response = await getStreamableHttpFetch(transport)( + new URL('http://test-server/mcp'), + { + method: 'GET', + headers: { Accept: 'text/event-stream' }, + }, + ); + + expect(response.status).toBe(502); + expect(fetchSpy).toHaveBeenCalledTimes(1); + }); }); describe('should connect via url', () => { diff --git a/packages/core/src/tools/mcp-client.ts b/packages/core/src/tools/mcp-client.ts index 52c82171afd..a22ce38cdc9 100644 --- a/packages/core/src/tools/mcp-client.ts +++ b/packages/core/src/tools/mcp-client.ts @@ -10,7 +10,10 @@ import { SSEClientTransport } from '@modelcontextprotocol/sdk/client/sse.js'; import { StdioClientTransport } from '@modelcontextprotocol/sdk/client/stdio.js'; import type { StreamableHTTPClientTransportOptions } from '@modelcontextprotocol/sdk/client/streamableHttp.js'; import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js'; -import type { Transport } from '@modelcontextprotocol/sdk/shared/transport.js'; +import type { + FetchLike, + Transport, +} from '@modelcontextprotocol/sdk/shared/transport.js'; import type { GetPromptResult, JSONRPCMessage, @@ -62,6 +65,9 @@ export const MCP_DEFAULT_TIMEOUT_MSEC = 10 * 60 * 1000; // default to 10 minutes const debugLogger = createDebugLogger('MCP'); +const STREAMABLE_HTTP_SSE_ACCEPT = 'text/event-stream'; +const STREAMABLE_HTTP_POST_ACCEPT = 'text/event-stream, application/json'; + export type DiscoveredMCPPrompt = Prompt & { serverName: string; invoke: (params: Record) => Promise; @@ -506,7 +512,8 @@ async function createTransportWithOAuth( }, }; - return new StreamableHTTPClientTransport( + return createStreamableHTTPClientTransport( + mcpServerName, new URL(mcpServerConfig.httpUrl), oauthTransportOptions, ); @@ -531,6 +538,77 @@ async function createTransportWithOAuth( } } +function isStreamableHttpSseGet(init?: RequestInit): boolean { + const method = init?.method?.toUpperCase() ?? 'GET'; + if (method !== 'GET') { + return false; + } + + const acceptHeader = new Headers(init?.headers).get('accept'); + return acceptHeader?.includes(STREAMABLE_HTTP_SSE_ACCEPT) ?? false; +} + +function createStreamableHttpFallbackFetch( + mcpServerName: string, + baseFetch?: FetchLike, +): FetchLike { + let hasWarned = false; + const fetchImpl = baseFetch ?? fetch; + + return async (url, init) => { + const response = await fetchImpl(url, init); + if ( + !isStreamableHttpSseGet(init) || + (response.status !== 400 && response.status !== 405) + ) { + return response; + } + + if (!hasWarned) { + hasWarned = true; + debugLogger.warn( + `MCP server '${mcpServerName}' rejected the spec-compliant Streamable HTTP SSE GET with HTTP ${response.status}; retrying with POST for compatibility. ` + + `This usually indicates a spec-divergent server (for example Spring AI 1.1.x). Please update the server to support GET SSE when possible.`, + ); + } + + const retryHeaders = new Headers(init?.headers); + retryHeaders.set('accept', STREAMABLE_HTTP_POST_ACCEPT); + + const retryInit: RequestInit = { + ...init, + method: 'POST', + headers: retryHeaders, + }; + delete retryInit.body; + + const retryResponse = await fetchImpl(url, retryInit); + if (retryResponse.ok) { + await response.body?.cancel(); + return retryResponse; + } + + if (response.status === 405) { + await retryResponse.body?.cancel(); + return response; + } + + await response.body?.cancel(); + return retryResponse; + }; +} + +function createStreamableHTTPClientTransport( + mcpServerName: string, + url: URL, + options: StreamableHTTPClientTransportOptions = {}, +): StreamableHTTPClientTransport { + return new StreamableHTTPClientTransport(url, { + ...options, + fetch: createStreamableHttpFallbackFetch(mcpServerName, options.fetch), + }); +} + /** * Discovers tools from all configured MCP servers and registers them with the tool registry. * It orchestrates the connection and discovery process for each server defined in the @@ -1332,7 +1410,8 @@ export async function createTransport( }; if (mcpServerConfig.httpUrl) { - return new StreamableHTTPClientTransport( + return createStreamableHTTPClientTransport( + mcpServerName, new URL(mcpServerConfig.httpUrl), transportOptions, ); @@ -1358,7 +1437,8 @@ export async function createTransport( authProvider: provider, }; if (mcpServerConfig.httpUrl) { - return new StreamableHTTPClientTransport( + return createStreamableHTTPClientTransport( + mcpServerName, new URL(mcpServerConfig.httpUrl), transportOptions, ); @@ -1430,7 +1510,8 @@ export async function createTransport( }; } - return new StreamableHTTPClientTransport( + return createStreamableHTTPClientTransport( + mcpServerName, new URL(mcpServerConfig.httpUrl), transportOptions, ); From d0e8f9ba125d9c978ee109fcee425a759be31602 Mon Sep 17 00:00:00 2001 From: cyphercodes Date: Thu, 21 May 2026 08:38:26 +0300 Subject: [PATCH 2/2] test(core): cover streamable HTTP retry failures --- packages/core/src/tools/mcp-client.test.ts | 116 +++++++++++++++++++++ packages/core/src/tools/mcp-client.ts | 4 +- 2 files changed, 118 insertions(+), 2 deletions(-) diff --git a/packages/core/src/tools/mcp-client.test.ts b/packages/core/src/tools/mcp-client.test.ts index 08377ab9070..7624865c134 100644 --- a/packages/core/src/tools/mcp-client.test.ts +++ b/packages/core/src/tools/mcp-client.test.ts @@ -40,6 +40,18 @@ function getStreamableHttpFetch( return (transport as unknown as { _fetch: FetchLike })._fetch; } +function createResponseWithCancelableBody( + body: string, + init: ResponseInit, +): { response: Response; cancel: ReturnType } { + const response = new Response(body, init); + const cancel = vi.fn().mockResolvedValue(undefined); + Object.defineProperty(response, 'body', { + value: { cancel }, + }); + return { response, cancel }; +} + vi.mock('node:fs', () => ({ existsSync: mockExistsSync, })); @@ -408,6 +420,110 @@ describe('mcp-client', () => { expect(response.status).toBe(502); expect(fetchSpy).toHaveBeenCalledTimes(1); }); + + it('returns the retry response when GET 400 and POST retry fails', async () => { + const original = createResponseWithCancelableBody('bad request', { + status: 400, + statusText: 'Bad Request', + }); + const retry = createResponseWithCancelableBody('unavailable', { + status: 503, + statusText: 'Service Unavailable', + }); + const fetchSpy = vi + .spyOn(globalThis, 'fetch') + .mockResolvedValueOnce(original.response) + .mockResolvedValueOnce(retry.response); + const transport = (await createTransport( + 'retry-fails-server', + { + httpUrl: 'http://test-server/mcp', + }, + false, + )) as StreamableHTTPClientTransport; + + const response = await getStreamableHttpFetch(transport)( + new URL('http://test-server/mcp'), + { + method: 'GET', + headers: { Accept: 'text/event-stream' }, + }, + ); + + expect(response).toBe(retry.response); + expect(response.status).toBe(503); + expect(fetchSpy).toHaveBeenCalledTimes(2); + expect(original.cancel).toHaveBeenCalledTimes(1); + expect(retry.cancel).not.toHaveBeenCalled(); + }); + + it('returns the original response when GET 405 and POST retry fails', async () => { + const original = createResponseWithCancelableBody( + 'method not allowed', + { + status: 405, + statusText: 'Method Not Allowed', + }, + ); + const retry = createResponseWithCancelableBody('bad request', { + status: 400, + statusText: 'Bad Request', + }); + const fetchSpy = vi + .spyOn(globalThis, 'fetch') + .mockResolvedValueOnce(original.response) + .mockResolvedValueOnce(retry.response); + const transport = (await createTransport( + 'method-retry-fails-server', + { + httpUrl: 'http://test-server/mcp', + }, + false, + )) as StreamableHTTPClientTransport; + + const response = await getStreamableHttpFetch(transport)( + new URL('http://test-server/mcp'), + { + method: 'GET', + headers: { Accept: 'text/event-stream' }, + }, + ); + + expect(response).toBe(original.response); + expect(response.status).toBe(405); + expect(fetchSpy).toHaveBeenCalledTimes(2); + expect(original.cancel).toHaveBeenCalledTimes(1); + expect(retry.cancel).toHaveBeenCalledTimes(1); + }); + + it('cancels the original response body when the POST retry throws', async () => { + const original = createResponseWithCancelableBody('bad request', { + status: 400, + statusText: 'Bad Request', + }); + const retryError = new Error('network reset'); + const fetchSpy = vi + .spyOn(globalThis, 'fetch') + .mockResolvedValueOnce(original.response) + .mockRejectedValueOnce(retryError); + const transport = (await createTransport( + 'retry-throws-server', + { + httpUrl: 'http://test-server/mcp', + }, + false, + )) as StreamableHTTPClientTransport; + + await expect( + getStreamableHttpFetch(transport)(new URL('http://test-server/mcp'), { + method: 'GET', + headers: { Accept: 'text/event-stream' }, + }), + ).rejects.toThrow('network reset'); + + expect(fetchSpy).toHaveBeenCalledTimes(2); + expect(original.cancel).toHaveBeenCalledTimes(1); + }); }); describe('should connect via url', () => { diff --git a/packages/core/src/tools/mcp-client.ts b/packages/core/src/tools/mcp-client.ts index a22ce38cdc9..d8e76b8d3cf 100644 --- a/packages/core/src/tools/mcp-client.ts +++ b/packages/core/src/tools/mcp-client.ts @@ -582,9 +582,10 @@ function createStreamableHttpFallbackFetch( }; delete retryInit.body; + await response.body?.cancel(); + const retryResponse = await fetchImpl(url, retryInit); if (retryResponse.ok) { - await response.body?.cancel(); return retryResponse; } @@ -593,7 +594,6 @@ function createStreamableHttpFallbackFetch( return response; } - await response.body?.cancel(); return retryResponse; }; }