fix(mcp): adapt secure SSE responses to web streams

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', () => {

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<string, string> = {}
+    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<typeof secureFetch>[1]): Promise<globalThis.Response> {
+    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,18 +171,18 @@ export class MCPToolkit extends BaseToolkit {
                         },
                         eventSourceInit: {
                             fetch: async (url, init) => {
-                                return secureFetch(url.toString(), {
+                                return secureFetchForSse(url.toString(), {
                                     ...(init as any),
                                     headers
-                                }) as any
+                                })
                             }
                         }
                     })
                 } else {
                     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)
                             }
                         }
                     })