fix(mcp): only fall back to legacy SSE transport on a 4xx Streamable HTTP error (#7154)

* fix(mcp): only fall back to legacy SSE transport on a 4xx Streamable HTTP error

The MCP client caught any Streamable HTTP connection error and immediately
retried with the legacy HTTP+SSE transport. For a Streamable-HTTP-only server,
a failure unrelated to protocol support (network error, 5xx, or an error after
a successful initialize) was masked by a misleading SSE error such as
"Non-200 status code (405)", hiding the real root cause.

Follow the MCP spec's backwards-compatibility flow: fall back to SSE only when
the server rejects the Streamable HTTP request with a 4xx (e.g. 404/405),
which signals it does not speak Streamable HTTP. Other errors now surface
as-is, and when the SSE fallback also fails both errors are reported.

* test(mcp): cover conditional SSE fallback behavior

Update the fallback test to use a 4xx StreamableHTTPError and add cases for:
non-HTTP errors and 5xx not triggering fallback, and both errors being
surfaced when the SSE fallback also fails.

* perf: code

* perf: code

---------

Co-authored-by: archer <545436317@qq.com>

packages/service/core/app/mcp.ts

 import { Client } from '@modelcontextprotocol/sdk/client/index.js';
 import { SSEClientTransport } from '@modelcontextprotocol/sdk/client/sse.js';
-import { StreamableHTTPClientTransport } from '@modelcontextprotocol/sdk/client/streamableHttp.js';
+import {
+  StreamableHTTPClientTransport,
+  StreamableHTTPError
+} from '@modelcontextprotocol/sdk/client/streamableHttp.js';
 import type { AppSchemaType } from '@fastgpt/global/core/app/type';
 import { type McpToolConfigType } from '@fastgpt/global/core/app/tool/mcpTool/type';
 import { retryFn } from '@fastgpt/global/common/system/utils';
@@ -20,6 +23,23 @@ export const assertMCPUrlNotInternal = async (url: string) => {
   }
 };
 
+const shouldFallbackToSSE = (error: unknown): boolean => {
+  return (
+    error instanceof StreamableHTTPError &&
+    typeof error.code === 'number' &&
+    error.code >= 400 &&
+    error.code < 500
+  );
+};
+
+const getErrorMessage = (error: unknown) => {
+  if (error instanceof Error) {
+    return error.message;
+  }
+
+  return String(error);
+};
+
 export class MCPClient {
   private client: Client;
   private url: string;
@@ -63,36 +83,50 @@ export class MCPClient {
         }
       });
       await this.client.connect(transport);
-    } catch (error) {
-      await this.client.connect(
-        new SSEClientTransport(new URL(this.url), {
-          requestInit: {
-            headers: this.headers
-          },
-          eventSourceInit: {
-            fetch: (url, init) => {
-              const mergedHeaders: Record<string, string> = {
-                ...this.headers
-              };
-
-              if (init?.headers) {
-                if (init.headers instanceof Headers) {
-                  init.headers.forEach((value, key) => {
-                    mergedHeaders[key] = value;
-                  });
-                } else if (typeof init.headers === 'object') {
-                  Object.assign(mergedHeaders, init.headers);
+    } catch (streamableError: any) {
+      if (!shouldFallbackToSSE(streamableError)) {
+        logger.info('Streamable HTTP error', streamableError);
+        throw streamableError;
+      }
+
+      try {
+        await this.client.connect(
+          new SSEClientTransport(new URL(this.url), {
+            requestInit: {
+              headers: this.headers
+            },
+            eventSourceInit: {
+              fetch: (url, init) => {
+                const mergedHeaders: Record<string, string> = {
+                  ...this.headers
+                };
+
+                if (init?.headers) {
+                  if (init.headers instanceof Headers) {
+                    init.headers.forEach((value, key) => {
+                      mergedHeaders[key] = value;
+                    });
+                  } else if (typeof init.headers === 'object') {
+                    Object.assign(mergedHeaders, init.headers);
+                  }
                 }
-              }
 
-              return fetch(url, {
-                ...init,
-                headers: mergedHeaders
-              });
+                return fetch(url, {
+                  ...init,
+                  headers: mergedHeaders
+                });
+              }
             }
-          }
-        })
-      );
+          })
+        );
+      } catch (sseError: any) {
+        logger.info('SSE error', sseError);
+        throw new Error(
+          `MCP connection failed. Streamable HTTP: ${getErrorMessage(
+            streamableError
+          )}; SSE: ${getErrorMessage(sseError)}`
+        );
+      }
     }
 
     this.client.onerror = (error) => {

packages/service/test/core/app/mcp.test.ts

@@ -18,6 +18,7 @@ vi.mock('@fastgpt/service/core/app/schema', () => ({
   }
 }));
 
+import { StreamableHTTPError } from '@modelcontextprotocol/sdk/client/streamableHttp.js';
 import { MCPClient, assertMCPUrlNotInternal, getMCPChildren } from '@fastgpt/service/core/app/mcp';
 import type { AppSchemaType } from '@fastgpt/global/core/app/type';
 
@@ -423,13 +424,13 @@ describe('MCPClient', () => {
   });
 
   describe('getConnection', () => {
-    it('should fallback to SSE when StreamableHTTP fails', async () => {
+    it('should fallback to SSE when server rejects Streamable HTTP with a 4xx', async () => {
       const mcpClient = new MCPClient(config);
       const client = getPrivateClient(mcpClient);
-      // First connect (StreamableHTTP) fails, second (SSE) succeeds
+      // StreamableHTTP rejected with 405 (server speaks legacy SSE), SSE succeeds
       client.connect = vi
         .fn()
-        .mockRejectedValueOnce(new Error('streamable failed'))
+        .mockRejectedValueOnce(new StreamableHTTPError(405, 'Method Not Allowed'))
         .mockResolvedValueOnce(undefined);
 
       const result = await (mcpClient as any).getConnection();
@@ -437,12 +438,39 @@ describe('MCPClient', () => {
       expect(result).toBe(client);
     });
 
-    it('should reject when both transports fail', async () => {
+    it('should not fallback to SSE on a non-HTTP (e.g. network) error', async () => {
       const mcpClient = new MCPClient(config);
       const client = getPrivateClient(mcpClient);
-      client.connect = vi.fn().mockRejectedValue(new Error('all failed'));
+      client.connect = vi.fn().mockRejectedValue(new Error('network unreachable'));
 
-      await expect((mcpClient as any).getConnection()).rejects.toThrow('all failed');
+      await expect((mcpClient as any).getConnection()).rejects.toThrow('network unreachable');
+      // Original error surfaces as-is, SSE transport is not attempted
+      expect(client.connect).toHaveBeenCalledTimes(1);
+    });
+
+    it('should not fallback to SSE when Streamable HTTP fails with a 5xx', async () => {
+      const mcpClient = new MCPClient(config);
+      const client = getPrivateClient(mcpClient);
+      client.connect = vi
+        .fn()
+        .mockRejectedValue(new StreamableHTTPError(500, 'Internal Server Error'));
+
+      await expect((mcpClient as any).getConnection()).rejects.toThrow('Internal Server Error');
+      expect(client.connect).toHaveBeenCalledTimes(1);
+    });
+
+    it('should surface both errors when the SSE fallback also fails', async () => {
+      const mcpClient = new MCPClient(config);
+      const client = getPrivateClient(mcpClient);
+      client.connect = vi
+        .fn()
+        .mockRejectedValueOnce(new StreamableHTTPError(404, 'Not Found'))
+        .mockRejectedValueOnce(new Error('SSE handshake failed'));
+
+      await expect((mcpClient as any).getConnection()).rejects.toThrow(
+        /Streamable HTTP:.*Not Found.*SSE:.*SSE handshake failed/s
+      );
+      expect(client.connect).toHaveBeenCalledTimes(2);
     });
 
     it('should return client on StreamableHTTP success', async () => {

pro @ 722e7f70

-Subproject commit 465869779095ca30d14a2852c220dd2d896a6c67
+Subproject commit 722e7f70a59c2614e05143f62f50230d6efb641b