Skip to content

fix(core): add streamable HTTP SSE POST fallback #4344

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
cyphercodes wants to merge 2 commits into from
Closed

fix(core): add streamable HTTP SSE POST fallback #4344

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d456916
fix(core): add streamable HTTP SSE POST fallback
cyphercodes May 20, 2026
d0e8f9b
test(core): cover streamable HTTP retry failures
cyphercodes May 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
231 changes: 231 additions & 0 deletions packages/core/src/tools/mcp-client.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand All @@ -33,6 +34,24 @@ 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;
}

function createResponseWithCancelableBody(
body: string,
init: ResponseInit,
): { response: Response; cancel: ReturnType<typeof vi.fn> } {
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,
}));
Expand Down Expand Up @@ -293,6 +312,218 @@ describe('mcp-client', () => {
Authorization: 'derp',
});
});

@wenshao wenshao May 21, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Suggestion] Add test coverage for POST retry failure paths.

The three existing tests cover happy paths (400→POST ok, 405→POST ok, 5xx no-retry), but neither "POST retry also fails" branch is tested. These paths contain the most nuanced logic — asymmetric body cancellation and different return-value semantics for 400 vs 405.

Suggested additional tests:

  1. GET 400, POST retry also fails (e.g. 503) — verify the retry's response is returned and bodies are handled correctly
  2. GET 405, POST retry also fails (e.g. 400) — verify the original 405 response is returned (not the retry's response)

These pin down the asymmetric failure semantics and guard against regressions in the body-cancellation logic.

— qwen-latest-series-invite-beta-v34 via Qwen Code /review

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);
});

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', () => {
Expand Down
91 changes: 86 additions & 5 deletions packages/core/src/tools/mcp-client.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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<string, unknown>) => Promise<GetPromptResult>;
Expand Down Expand Up @@ -506,7 +512,8 @@ async function createTransportWithOAuth(
},
};

return new StreamableHTTPClientTransport(
return createStreamableHTTPClientTransport(
mcpServerName,
new URL(mcpServerConfig.httpUrl),
oauthTransportOptions,
);
Expand All @@ -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,

@wenshao wenshao May 21, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Suggestion] Keep the original 405 response body intact until you know it will not be returned. Right now this cancels the original GET response before the POST retry; if that retry also fails, the 405 branch returns the same original response with its body already canceled, so downstream error handling cannot read the original diagnostic body.

Suggested change
headers: retryHeaders,
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;

— gpt-5.5 via Qwen Code /review

};
delete retryInit.body;

@wenshao wenshao May 21, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[Suggestion] Wrap the retry fetchImpl in a try/catch and cancel the original response body before the retry.

Two robustness improvements:

  1. Cancel the 400/405 body before retrying — the error response body is never useful to the caller. Cancelling it early frees the TCP connection and simplifies post-retry cleanup.
  2. Add try/catch for network errors — if fetchImpl(url, retryInit) throws (DNS failure, TCP reset, TLS error), the original response.body currently leaks. The catch ensures it's always cancelled.
Suggested change
delete retryInit.body;
await response.body?.cancel();
let retryResponse: Response;
try {
retryResponse = await fetchImpl(url, retryInit);
} catch (err) {
throw err;
}
if (retryResponse.ok) {
return retryResponse;
}
if (response.status === 405) {
await retryResponse.body?.cancel();
return response;
}
return retryResponse;

— qwen-latest-series-invite-beta-v34 via Qwen Code /review


await response.body?.cancel();

const retryResponse = await fetchImpl(url, retryInit);
if (retryResponse.ok) {
return retryResponse;
}

if (response.status === 405) {
await retryResponse.body?.cancel();
return response;
}

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
Expand Down Expand Up @@ -1332,7 +1410,8 @@ export async function createTransport(
};

if (mcpServerConfig.httpUrl) {
return new StreamableHTTPClientTransport(
return createStreamableHTTPClientTransport(
mcpServerName,
new URL(mcpServerConfig.httpUrl),
transportOptions,
);
Expand All @@ -1358,7 +1437,8 @@ export async function createTransport(
authProvider: provider,
};
if (mcpServerConfig.httpUrl) {
return new StreamableHTTPClientTransport(
return createStreamableHTTPClientTransport(
mcpServerName,
new URL(mcpServerConfig.httpUrl),
transportOptions,
);
Expand Down Expand Up @@ -1430,7 +1510,8 @@ export async function createTransport(
};
}

return new StreamableHTTPClientTransport(
return createStreamableHTTPClientTransport(
mcpServerName,
new URL(mcpServerConfig.httpUrl),
transportOptions,
);
Expand Down
Loading