fix(mcp): harden SSE fallback for Bailian and --url overrides

Author: hopelynd

packages/commands/src/commands/mcp/call.ts

@@ -36,7 +36,8 @@ const CALL_FLAGS = {
   url: {
     type: "string",
     valueHint: "<url>",
-    description: "Override the MCP endpoint URL (for non-Bailian servers)",
+    description:
+      "Override the MCP endpoint URL (non-Bailian). Tries Streamable HTTP first, then classic SSE on the same URL.",
   },
 } satisfies FlagsDef;
 type CallFlags = ParsedFlags<typeof CALL_FLAGS>;

packages/commands/src/commands/mcp/tools.ts

@@ -16,7 +16,8 @@ export default defineCommand({
     url: {
       type: "string",
       valueHint: "<url>",
-      description: "Override the MCP endpoint URL (for non-Bailian servers)",
+      description:
+        "Override the MCP endpoint URL (non-Bailian). Tries Streamable HTTP first, then classic SSE on the same URL.",
     },
   },
   exampleArgs: [

packages/core/src/client/client.ts

@@ -171,8 +171,8 @@ export class Client {
   }
 
   /**
-   * Connect to a Bailian MCP: try Streamable HTTP, then SSE on 405+streamableHttp (except WebSearch).
-   * `urlOverride` maps to `--url` and uses Streamable only (no fallback).
+   * Connect to a Bailian MCP: try Streamable HTTP, then SSE on 405 (except WebSearch).
+   * `urlOverride` maps to `--url`: Streamable first, then classic SSE on the same URL (405/404).
    */
   connectBailianMcp(
     serverCode: string,

packages/core/src/client/index.ts

@@ -68,6 +68,7 @@ export {
   bailianMcpPath,
   bailianMcpSsePath,
   isStreamableHttpUnsupported,
+  isUrlOverrideSseFallbackCandidate,
   connectBailianMcpWithFallback,
 } from "./mcp.ts";
 export type { ServerSentEvent } from "./stream.ts";

packages/core/src/client/mcp-sse.ts

@@ -24,18 +24,25 @@ type PendingResolver = {
   reject: (reason: unknown) => void;
 };
 
+/** 用字符串键匹配 JSON-RPC id（兼容 number / string 回传）。 */
+function pendingKey(id: number | string): string {
+  return String(id);
+}
+
 export class McpSseClient {
   private sseUrl: string;
   private messageUrl: string | undefined;
   private nextId = 1;
   private deps: HttpDeps;
   private authToken: string | undefined;
   private abortController: AbortController | undefined;
-  private pending = new Map<number, PendingResolver>();
+  private pending = new Map<string, PendingResolver>();
   private endpointReady: Promise<void>;
   private resolveEndpoint: (() => void) | undefined;
   private rejectEndpoint: ((reason: unknown) => void) | undefined;
   private closed = false;
+  /** SSE GET 已结束（非主动 close）时置位，后续 RPC 立即失败。 */
+  private streamEnded = false;
 
   constructor(deps: HttpDeps, sseUrl: string, authToken?: string) {
     this.deps = deps;
@@ -87,12 +94,23 @@ export class McpSseClient {
     if (this.closed) return;
     this.closed = true;
     this.abortController?.abort();
+    this.failPending(new BailianError("MCP SSE session closed.", ExitCode.GENERAL));
+    this.messageUrl = undefined;
+  }
+
+  private failPending(reason: unknown): void {
     for (const [, waiter] of this.pending) {
-      waiter.reject(new BailianError("MCP SSE session closed.", ExitCode.GENERAL));
+      waiter.reject(reason);
     }
     this.pending.clear();
   }
 
+  private markStreamEnded(reason: BailianError): void {
+    this.streamEnded = true;
+    this.messageUrl = undefined;
+    this.failPending(reason);
+  }
+
   private async openSse(): Promise<void> {
     if (this.abortController) return;
 
@@ -145,10 +163,10 @@ export class McpSseClient {
               ExitCode.GENERAL,
             );
       this.rejectEndpoint?.(reason);
-      for (const [, waiter] of this.pending) {
-        waiter.reject(reason);
+      // consumeSse 在正常结束路径已 markStreamEnded；此处覆盖解析/读取异常。
+      if (!this.streamEnded) {
+        this.markStreamEnded(reason);
       }
-      this.pending.clear();
     });
 
     const timeoutMs = this.deps.settings.timeout * 1000;
@@ -167,7 +185,8 @@ export class McpSseClient {
     for await (const event of parseSSE(response)) {
       if (this.closed) break;
 
-      if (event.event === "endpoint" || (!event.event && !this.messageUrl)) {
+      // 规范要求首事件为 event: endpoint；不接受无名事件以免误把 JSON 当 URL。
+      if (event.event === "endpoint") {
         const raw = event.data.trim();
         if (!raw) continue;
         // Only accept same-origin message URLs so we never forward the Bearer token cross-origin.
@@ -178,21 +197,25 @@ export class McpSseClient {
         continue;
       }
 
+      // 缺省 event 类型在 SSE 中等同 message。
       if (event.event === "message" || event.event === undefined) {
         let payload: JsonRpcResponse;
         try {
           payload = JSON.parse(event.data) as JsonRpcResponse;
         } catch {
           continue;
         }
-        if (typeof payload.id !== "number") continue;
-        const waiter = this.pending.get(payload.id);
+        if (typeof payload.id !== "number" && typeof payload.id !== "string") continue;
+        const key = pendingKey(payload.id);
+        const waiter = this.pending.get(key);
         if (!waiter) continue;
-        this.pending.delete(payload.id);
+        this.pending.delete(key);
         waiter.resolve(payload);
       }
     }
 
+    if (this.closed) return;
+
     if (!this.messageUrl) {
       const error = new BailianError(
         "MCP SSE stream ended before endpoint event.",
@@ -201,10 +224,19 @@ export class McpSseClient {
       this.rejectEndpoint?.(error);
       throw error;
     }
+
+    // 已拿到 endpoint 后流仍结束：标记会话死亡并唤醒 pending；不再 throw，
+    // 避免 void consumeSse().catch 之外再冒出未处理 rejection。
+    this.markStreamEnded(new BailianError("MCP SSE stream ended unexpectedly.", ExitCode.GENERAL));
   }
 
   private async rpc(method: string, params?: Record<string, unknown>): Promise<unknown> {
+    if (this.closed || this.streamEnded) {
+      throw new BailianError("MCP SSE stream ended unexpectedly.", ExitCode.GENERAL);
+    }
+
     const id = this.nextId++;
+    const key = pendingKey(id);
     const body = {
       jsonrpc: "2.0" as const,
       id,
@@ -214,15 +246,20 @@ export class McpSseClient {
 
     const timeoutMs = this.deps.settings.timeout * 1000;
     const responsePromise = new Promise<JsonRpcResponse>((resolve, reject) => {
-      this.pending.set(id, { resolve, reject });
+      this.pending.set(key, { resolve, reject });
     });
+    // 流可能在 Promise.race 之前结束并 reject pending，先挂上 catch 避免 unhandledRejection。
+    void responsePromise.catch(() => undefined);
     const responseTimeout = cancellableTimeoutReject(
       timeoutMs,
       `MCP SSE timed out waiting for response to ${method}.`,
     );
 
     try {
       await this.postMessage(body);
+      if (this.closed || this.streamEnded) {
+        throw new BailianError("MCP SSE stream ended unexpectedly.", ExitCode.GENERAL);
+      }
       const data = await Promise.race([responsePromise, responseTimeout.promise]);
       if (data.error) {
         throw new BailianError(
@@ -232,7 +269,7 @@ export class McpSseClient {
       }
       return data.result;
     } catch (error) {
-      this.pending.delete(id);
+      this.pending.delete(key);
       throw error;
     } finally {
       responseTimeout.cancel();
@@ -249,6 +286,9 @@ export class McpSseClient {
   }
 
   private async postMessage(body: unknown): Promise<void> {
+    if (this.closed || this.streamEnded) {
+      throw new BailianError("MCP SSE stream ended unexpectedly.", ExitCode.GENERAL);
+    }
     if (!this.messageUrl) {
       throw new BailianError("MCP SSE message endpoint is not ready.", ExitCode.GENERAL);
     }

packages/core/src/client/mcp.ts

@@ -16,6 +16,7 @@ import { ExitCode } from "../errors/codes.ts";
 import type { HttpDeps } from "./http.ts";
 import { trackingHeaders } from "./headers.ts";
 import { McpSseClient } from "./mcp-sse.ts";
+import { parseSSE } from "./stream.ts";
 
 // ---- JSON-RPC 2.0 Types ----
 
@@ -28,7 +29,7 @@ interface JsonRpcRequest {
 
 interface JsonRpcResponse {
   jsonrpc: "2.0";
-  id: number;
+  id?: number | string | null;
   result?: unknown;
   error?: { code: number; message: string; data?: unknown };
 }
@@ -67,11 +68,22 @@ export function bailianMcpSsePath(serverCode: string): string {
   return `/api/v1/mcps/${serverCode}/sse`;
 }
 
-/** True when the error is a 405 that indicates Streamable HTTP is unsupported (SSE fallback). */
+/**
+ * True when Streamable HTTP is unsupported and classic SSE fallback should be tried.
+ * 以 HTTP 405 为准，不依赖服务端英文文案（避免文案变更导致降级失效）。
+ * Bailian 的 404（未开通）不在此列，避免误降级。
+ */
 export function isStreamableHttpUnsupported(error: unknown): boolean {
   if (!(error instanceof BailianError)) return false;
-  const message = error.message;
-  return /405\b/i.test(message) && /streamableHttp/i.test(message);
+  return /405\b/i.test(error.message);
+}
+
+/**
+ * `--url` 覆盖时的 SSE 降级条件（官方 backwards-compat：同 URL 上 405/404 后尝试 GET SSE）。
+ */
+export function isUrlOverrideSseFallbackCandidate(error: unknown): boolean {
+  if (!(error instanceof BailianError)) return false;
+  return /405\b/i.test(error.message) || /404\b/i.test(error.message);
 }
 
 export type McpConnectedClient = {
@@ -89,12 +101,16 @@ export type ConnectBailianMcpOptions = {
   /** Full classic SSE URL (/sse). */
   sseUrl: string;
   serverCode: string;
-  /** Explicit `--url` override: Streamable only, no SSE fallback. */
+  /**
+   * Explicit `--url` override: try Streamable on that URL first;
+   * on 405/404 fall back to classic SSE on the same URL.
+   */
   urlOverride?: string;
 };
 
 /**
- * Connect via Streamable HTTP first; on 405+streamableHttp (except WebSearch), fall back to SSE.
+ * Connect via Streamable HTTP first; on 405 (except WebSearch), fall back to SSE.
+ * `--url` uses the same URL for Streamable then classic SSE (official backwards-compat).
  * For WebSearch, rethrow the original error so commands can attach a re-activate hint.
  */
 export async function connectBailianMcpWithFallback(
@@ -103,9 +119,24 @@ export async function connectBailianMcpWithFallback(
   const { deps, authToken, httpUrl, sseUrl, serverCode, urlOverride } = options;
 
   if (urlOverride) {
-    const client = new McpClient(deps, urlOverride, authToken);
-    await client.initialize();
-    return { client, url: urlOverride };
+    const httpClient = new McpClient(deps, urlOverride, authToken);
+    try {
+      await httpClient.initialize();
+      return { client: httpClient, url: urlOverride };
+    } catch (error) {
+      if (!isUrlOverrideSseFallbackCandidate(error)) {
+        throw error;
+      }
+    }
+
+    const sseClient = new McpSseClient(deps, urlOverride, authToken);
+    try {
+      await sseClient.initialize();
+      return { client: sseClient, url: urlOverride };
+    } catch (error) {
+      sseClient.close();
+      throw error;
+    }
   }
 
   const httpClient = new McpClient(deps, httpUrl, authToken);
@@ -188,7 +219,7 @@ export class McpClient {
     };
 
     const response = await this.send(body);
-    const data = (await response.json()) as JsonRpcResponse;
+    const data = await this.readJsonRpcResponse(response, id);
 
     if (data.error) {
       throw new BailianError(
@@ -210,6 +241,44 @@ export class McpClient {
     await this.send(body);
   }
 
+  /**
+   * 按 Content-Type 读取 JSON-RPC 响应：支持 application/json 与 text/event-stream。
+   */
+  private async readJsonRpcResponse(
+    response: Response,
+    expectedId: number,
+  ): Promise<JsonRpcResponse> {
+    const contentType = response.headers.get("content-type") || "";
+    if (contentType.includes("text/event-stream")) {
+      return await this.readJsonRpcFromSse(response, expectedId);
+    }
+
+    return (await response.json()) as JsonRpcResponse;
+  }
+
+  private async readJsonRpcFromSse(
+    response: Response,
+    expectedId: number,
+  ): Promise<JsonRpcResponse> {
+    const expectedKey = String(expectedId);
+    for await (const event of parseSSE(response)) {
+      if (event.event && event.event !== "message") continue;
+      let payload: JsonRpcResponse;
+      try {
+        payload = JSON.parse(event.data) as JsonRpcResponse;
+      } catch {
+        continue;
+      }
+      if (payload.id == null) continue;
+      if (String(payload.id) !== expectedKey) continue;
+      return payload;
+    }
+    throw new BailianError(
+      "MCP SSE response stream ended without a matching JSON-RPC response.",
+      ExitCode.GENERAL,
+    );
+  }
+
   private async send(body: unknown): Promise<Response> {
     const headers: Record<string, string> = {
       "Content-Type": "application/json",