diff --git a/packages/components/nodes/tools/MCP/core.test.ts b/packages/components/nodes/tools/MCP/core.test.ts index e97e874516c..0af86e2ca2d 100644 --- a/packages/components/nodes/tools/MCP/core.test.ts +++ b/packages/components/nodes/tools/MCP/core.test.ts @@ -5,8 +5,41 @@ import { validateEnvironmentVariables, validateMCPServerConfig, sanitizeMCPToolDescription, - sanitizeMCPToolName + sanitizeMCPToolName, + toWebReadableResponse } from './core' +import { PassThrough } from 'node:stream' +import { Response } from 'node-fetch' + +describe('toWebReadableResponse', () => { + it('converts a node-fetch stream while preserving response metadata', async () => { + const body = new PassThrough() + body.end('data: connected\n\n') + const response = new Response(body, { + status: 200, + statusText: 'OK', + headers: { 'content-type': 'text/event-stream', 'x-test': 'preserved' } + }) + + expect(typeof response.body?.pipe).toBe('function') + expect('getReader' in (response.body as object)).toBe(false) + + const converted = toWebReadableResponse(response) + + expect(typeof converted.body?.getReader).toBe('function') + expect(converted.status).toBe(200) + expect(converted.statusText).toBe('OK') + expect(converted.headers.get('content-type')).toBe('text/event-stream') + expect(converted.headers.get('x-test')).toBe('preserved') + await expect(converted.text()).resolves.toBe('data: connected\n\n') + }) + + it('leaves non-success responses for EventSource to reject by status', () => { + const response = new Response('not found', { status: 404 }) + + expect(toWebReadableResponse(response)).toBe(response) + }) +}) describe('MCP Security Validations', () => { describe('validateCommandFlags', () => { diff --git a/packages/components/nodes/tools/MCP/core.ts b/packages/components/nodes/tools/MCP/core.ts index 09f60d9643a..152e64f5c5d 100644 --- a/packages/components/nodes/tools/MCP/core.ts +++ b/packages/components/nodes/tools/MCP/core.ts @@ -4,11 +4,37 @@ import { SSEClientTransport } from '@modelcontextprotocol/sdk/client/sse.js' import { StdioClientTransport, StdioServerParameters } from '@modelcontextprotocol/sdk/client/stdio.js' import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js' import { CallToolRequest, CallToolResultSchema, ListToolsResult, ListToolsResultSchema } from '@modelcontextprotocol/sdk/types.js' +import { Readable } from 'node:stream' +import type { Response as NodeFetchResponse } from 'node-fetch' import { checkDenyList, secureFetch } from '../../../src/httpSecurity' const DEFAULT_MCP_TOOL_DESCRIPTION_MAX_LENGTH = 1024 const DEFAULT_MCP_TOOL_NAME_MAX_LENGTH = 128 +export function toWebReadableResponse(response: NodeFetchResponse): globalThis.Response { + const body = response.body + // EventSource rejects non-200 responses before reading their bodies, and + // the Fetch Response constructor disallows bodies for statuses such as 204. + if (response.status !== 200 || !body || typeof body.pipe !== 'function') { + return response as unknown as globalThis.Response + } + + const headers: Record = {} + response.headers.forEach((value, key) => { + headers[key] = value + }) + + return new globalThis.Response(Readable.toWeb(body as Readable) as any, { + status: response.status, + statusText: response.statusText, + headers + }) +} + +async function secureFetchForSse(url: string, init?: Parameters[1]): Promise { + return toWebReadableResponse(await secureFetch(url, init)) +} + function getMCPToolDescriptionMaxLength(): number { const parsed = Number(process.env.CUSTOM_MCP_TOOL_DESCRIPTION_MAX_LENGTH) return Number.isFinite(parsed) && parsed > 0 ? parsed : DEFAULT_MCP_TOOL_DESCRIPTION_MAX_LENGTH @@ -145,10 +171,10 @@ export class MCPToolkit extends BaseToolkit { }, eventSourceInit: { fetch: async (url, init) => { - return secureFetch(url.toString(), { + return secureFetchForSse(url.toString(), { ...(init as any), headers - }) as any + }) } } }) @@ -156,7 +182,7 @@ export class MCPToolkit extends BaseToolkit { transport = new SSEClientTransport(baseUrl, { eventSourceInit: { fetch: async (url, init) => { - return secureFetch(url.toString(), init as any) as any + return secureFetchForSse(url.toString(), init as any) } } })