mirror of
https://github.com/QwenLM/qwen-code.git
synced 2026-07-10 01:29:17 +00:00
fix(core): tolerate unsupported Streamable HTTP GET SSE (#4521)
Fixes #4326
This commit is contained in:
JerryLee
GitHub
parent
2d8052b02c
commit
4e1b3827e9
2 changed files with 379 additions and 2 deletions
|
|
@ -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<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);
|
||||
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<typeof fetch>()
|
||||
.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<typeof fetch>()
|
||||
.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<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 the SDK-native GET SSE unsupported sentinel', async () => {
|
||||
const fetchFn = vi
|
||||
.fn<typeof fetch>()
|
||||
.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<typeof fetch>()
|
||||
.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<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' },
|
||||
});
|
||||
|
||||
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);
|
||||
});
|
||||
});
|
||||
|
||||
|
|
@ -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' };
|
||||
|
|
|
|||
|
|
@ -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<string | undefined> {
|
||||
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<string, unknown>) => Promise<GetPromptResult>;
|
||||
|
|
@ -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) {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue