Skip to content

fix(core): tolerate unsupported Streamable HTTP GET SSE #4521

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

Merged
tanzhenxin merged 4 commits into from
Jun 2, 2026
Merged

fix(core): tolerate unsupported Streamable HTTP GET SSE #4521

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
e6b14e1
fix(core): tolerate unsupported streamable http get sse
Jerry2003826 May 25, 2026
2e80fb4
test(core): cover streamable http post passthrough
Jerry2003826 May 26, 2026
ba3e01e
fix(core): leave streamable HTTP 405 responses intact
Jerry2003826 May 26, 2026
e466b6a
fix(core): preserve resumable streamable HTTP SSE errors
Jerry2003826 May 26, 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
146 changes: 146 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 @@ -12,10 +12,12 @@ 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 {
addMCPStatusChangeListener,
createStreamableHttpCompatibilityFetch,
createTransport,
getAllMCPServerStatuses,
getMCPServerStatus,
Expand Down Expand Up @@ -273,6 +275,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 () => {
Expand All @@ -292,6 +296,96 @@ describe('mcp-client', () => {
expect((transport as any)._requestInit?.headers).toEqual({
Authorization: 'derp',
});
// eslint-disable-next-line @typescript-eslint/no-explicit-any

@wenshao wenshao May 26, 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] _fetch is asserted here and at line 277, but createStreamableHttpCompatibilityFetch is injected into 4 code paths (OAuth at mcp-client.ts:552, SERVICE_ACCOUNT_IMPERSONATION at :1388, GOOGLE_CREDENTIALS at :1416, plain httpUrl at :1473). The OAuth and SERVICE_ACCOUNT_IMPERSONATION paths have no test verifying the fetch was wired. A refactor that accidentally drops the fetch assignment in one branch would go undetected.

Consider adding expect((transport as any)._fetch).toEqual(expect.any(Function)) to the existing Google Credentials httpUrl test and new tests for the OAuth and Service Account Impersonation transport paths.

— qwen3.7-max via Qwen Code /review

@Jerry2003826 Jerry2003826 May 26, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Added in 2e80fb4. The test suite now verifies compatibility fetch wiring for OAuth Streamable HTTP, Service Account Impersonation Streamable HTTP, and Google Credentials Streamable HTTP transports, in addition to the existing plain httpUrl coverage.

@wenshao wenshao Jun 1, 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] The test suite covers all guard branches but leaves gaps in readResponseBodyExcerpt sub-branches:

  1. Empty body — if (!body) return undefined (line 74) is untested. Add new Response(null, { status: 400 }) with GET + SSE Accept.
  2. Truncation — body.length > limit (line 76) is untested. Add new Response('x'.repeat(1024), { status: 400 }) and verify the ... suffix.
  3. Synthetic response shape — tests assert response.status === 405 but not body or statusText. Add expect(await response.text()).toBe('') and expect(response.statusText).toBe('Method Not Allowed').

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

expect((transport as any)._fetch).toEqual(expect.any(Function));
});

it('treats 400 from optional GET SSE stream as unsupported', async () => {
const fetchFn = vi
.fn<typeof fetch>()
.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('treats parameterized GET SSE Accept headers as unsupported', async () => {
const fetchFn = vi
.fn<typeof fetch>()
.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<typeof fetch>()
.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<typeof fetch>()
.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' },
});

@wenshao wenshao May 26, 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] The three new tests cover GET+400, GET+502, and GET+non-SSE+400, but none verify that a POST request receiving a 400 response passes through unchanged. POST is the primary MCP JSON-RPC channel — a regression that accidentally rewrites POST responses would silently break all tool calls.

Suggested change
});
it('does not rewrite POST responses', async () => {
const fetchFn = vi
.fn<typeof fetch>()
.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);
});

— qwen3.7-max via Qwen Code /review

@Jerry2003826 Jerry2003826 May 26, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Added in ef4fe4e. The compatibility wrapper now has POST passthrough coverage: a POST request returning 400 remains 400, so the optional GET SSE fallback cannot accidentally rewrite the primary JSON-RPC channel.

expect(response.status).toBe(400);
});

it('does not rewrite POST responses', async () => {
const fetchFn = vi
.fn<typeof fetch>()
.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);
});
});

Expand Down Expand Up @@ -474,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 () => {
Expand Down Expand Up @@ -512,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' };
Expand Down
61 changes: 60 additions & 1 deletion packages/core/src/tools/mcp-client.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,58 @@ export const MCP_DEFAULT_TIMEOUT_MSEC = 10 * 60 * 1000; // default to 10 minutes

const debugLogger = createDebugLogger('MCP');

@wenshao wenshao May 26, 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] STREAMABLE_HTTP_GET_SSE_FALLBACK_STATUSES includes 405, but the SDK already handles 405 natively — _startOrAuthSse() checks response.status === 405 and returns silently without throwing. Including 405 here means the wrapper intercepts a perfectly valid 405 response, discards its body via cancel(), emits a misleading debugLogger.warn ("rejected the optional Streamable HTTP GET SSE stream with HTTP 405" — when 405 is actually the expected sentinel), and returns a redundant synthetic 405.

Only 400 needs normalization (Spring AI's non-standard response). Removing 405 avoids spurious warnings for servers that correctly signal "GET SSE unsupported" and keeps the wrapper narrowly scoped to its stated purpose.

Suggested change
const STREAMABLE_HTTP_GET_SSE_FALLBACK_STATUSES = new Set([400]);

— qwen3.7-max via Qwen Code /review

@Jerry2003826 Jerry2003826 May 26, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Handled in ba3e01e. The compatibility wrapper now only normalizes the Spring-AI-style 400 response; SDK-native GET SSE 405 responses pass through unchanged. Added a regression that preserves the original 405 response body/status so the wrapper cannot emit a redundant warning for the SDK sentinel.

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(',')
.map((value) => value.split(';')[0].trim().toLowerCase())

@wenshao wenshao May 26, 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] isStreamableHttpGetSseRequest does not check for the Last-Event-ID header. The SDK's _startOrAuthSse() sends Last-Event-ID during GET SSE resumption (verified in SDK source streamableHttp.js:84-89). If a server supports GET SSE but returns 400 for an expired/invalid Last-Event-ID, the wrapper silently rewrites to 405, causing the SDK to permanently abandon the GET stream for that session — losing all server-initiated push notifications.

Suggested change
.map((value) => value.split(';')[0].trim().toLowerCase())
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');

— qwen3.7-max via Qwen Code /review

@Jerry2003826 Jerry2003826 May 26, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Fixed in 672d761. GET SSE requests carrying Last-Event-ID are now left untouched, so resumable-stream 400 responses preserve their original status/body instead of being normalized to the unsupported-GET sentinel. Added a regression for that header path.

.some((type) => type === 'text/event-stream');
}

@wenshao wenshao May 26, 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] No documentation explains why 405 is the sentinel value. The SDK's StreamableHTTPClientTransport._startOrAuthSse() treats response.status === 405 as "server does not support the optional GET SSE stream" and silently falls back to POST-only mode. If the SDK changes this behavior, this wrapper silently breaks with no compile-time or test-time signal.

Consider adding a JSDoc that names the SDK coupling:

/**
 * Wraps fetch to normalize 400→405 for the optional GET SSE request.
 *
 * WHY 405: The SDK's _startOrAuthSse() treats status 405 as "GET SSE
 * unsupported" and falls back to POST-only. Some servers (Spring AI)
 * return 400 instead.
 *
 * SDK COUPLING: Pin to the SDK version in package.json. If the SDK
 * changes its non-OK handling, this wrapper must be updated in lockstep.
 */

— qwen3.7-max via Qwen Code /review

@Jerry2003826 Jerry2003826 May 26, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Documented in ef4fe4e. The wrapper JSDoc now names the StreamableHTTPClientTransport._startOrAuthSse() coupling and explains why 405 is the unsupported GET SSE sentinel.


/**
* 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;
}

await response.body?.cancel().catch(() => {

@wenshao wenshao May 26, 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] The original 400 response body — which may contain diagnostic payloads like {"error":"invalid_token"} or {"error":"unsupported_media_type"} — is cancelled before logging. The debugLogger.warn() only records the numeric status code. Operators investigating why an MCP server degraded to POST-only mode lose the server's explanation for the 400.

Suggested change
await response.body?.cancel().catch(() => {
const bodyText = await response.text().catch(() => '<unreadable>');
debugLogger.warn(
`MCP server '${mcpServerName}' rejected the optional Streamable HTTP ` +
`GET SSE stream with HTTP ${response.status} (${bodyText}); continuing without ` +
`the standalone GET stream. POST request streams remain enabled.`,
);

— qwen3.7-max via Qwen Code /review

@Jerry2003826 Jerry2003826 May 26, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Fixed in 672d761. The compatibility wrapper now reads a bounded clone of the 400 response body before cancelling the original stream and includes that excerpt in the debug warning.

// 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<string, unknown>) => Promise<GetPromptResult>;
Expand Down Expand Up @@ -498,6 +550,7 @@ async function createTransportWithOAuth(
if (mcpServerConfig.httpUrl) {
// Create HTTP transport with OAuth token
const oauthTransportOptions: StreamableHTTPClientTransportOptions = {
fetch: createStreamableHttpCompatibilityFetch(mcpServerName),
requestInit: {
headers: {
...mcpServerConfig.headers,
Expand Down Expand Up @@ -1332,6 +1385,8 @@ export async function createTransport(
};

if (mcpServerConfig.httpUrl) {
(transportOptions as StreamableHTTPClientTransportOptions).fetch =
createStreamableHttpCompatibilityFetch(mcpServerName);
return new StreamableHTTPClientTransport(
new URL(mcpServerConfig.httpUrl),
transportOptions,
Expand All @@ -1358,6 +1413,8 @@ export async function createTransport(
authProvider: provider,
};
if (mcpServerConfig.httpUrl) {
(transportOptions as StreamableHTTPClientTransportOptions).fetch =
createStreamableHttpCompatibilityFetch(mcpServerName);
return new StreamableHTTPClientTransport(
new URL(mcpServerConfig.httpUrl),
transportOptions,
Expand Down Expand Up @@ -1414,7 +1471,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) {
Expand Down
Loading