Merge branch 'main' into feat/coding-plan-usage

2026-09-14 19:49:23 +08:00 · 2026-08-14 17:05:53 +08:00
parent 9133b6bdd1 f1b6cacd7f
commit 5f1c97940d
14 changed files with 1740 additions and 84 deletions
@@ -1,11 +1,11 @@
-import { BailianError } from "bailian-cli-core";
+import { BailianError, isStreamableHttpUnsupported } from "bailian-cli-core";
 import { mcpMarketplaceDetailPage } from "bailian-cli-runtime";

 /** Detect MCP-not-activated / invalid 404 errors (CLI-wrapped server message). */
 export function isMcpNotActivated(error: unknown): boolean {
  if (!(error instanceof BailianError)) return false;
  const message = error.message;
-  if (!/MCP request failed:\s*404\b/i.test(message)) return false;
+  if (!/^MCP request failed:\s*404\b/i.test(message)) return false;
  return /未开通|MCP不存在|MCP_IS_INVALID/i.test(message);
 }

@@ -26,14 +26,28 @@ export function mcpActivateHint(serverCode: string): string {
 /**
 * For not-activated errors, keep the original message / exitCode and append a hint only.
 * Do not replace the server error message.
+ * WebSearch + 405 streamableHttp: do not fall back; attach a re-activate / upgrade hint.
 */
 export function rethrowWithMcpActivateHint(error: unknown, serverCode: string): never {
-  if (isMcpNotActivated(error) && error instanceof BailianError && !error.hint) {
+  if (!(error instanceof BailianError) || error.hint) {
+    throw error;
+  }
+
+  if (isMcpNotActivated(error)) {
    throw new BailianError(error.message, error.exitCode, mcpActivateHint(serverCode), {
      cause: error,
      api: error.api,
      rawResponse: error.rawResponse,
    });
  }
+
+  if (serverCode === "WebSearch" && isStreamableHttpUnsupported(error)) {
+    throw new BailianError(error.message, error.exitCode, mcpActivateHint(serverCode), {
+      cause: error,
+      api: error.api,
+      rawResponse: error.rawResponse,
+    });
+  }
+
  throw error;
 }
@@ -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>;
@@ -114,14 +115,14 @@ export default defineCommand({
    const { serverCode, toolName } = parseTarget(flags.target);
    const toolArgs = buildToolArgs(flags);

-    const url = flags.url || ctx.client.url(bailianMcpPath(serverCode));
+    const previewUrl = flags.url || ctx.client.url(bailianMcpPath(serverCode));
    const format = detectOutputFormat(settings.output);

    if (settings.dryRun) {
      emitResult(
        {
          server: serverCode,
-          url,
+          url: previewUrl,
          tool: toolName,
          arguments: toolArgs,
        },
@@ -130,13 +131,14 @@ export default defineCommand({
      return;
    }

-    const client = ctx.client.mcp(url);
+    let client: { close?(): void } | undefined;
    try {
-      await client.initialize();
-      const result = await client.callTool(toolName, toolArgs);
+      const connected = await ctx.client.connectBailianMcp(serverCode, flags.url);
+      client = connected.client;
+      const result = await connected.client.callTool(toolName, toolArgs);

      if (result.isError) {
-        const errText = result.content.map((c) => c.text || "").join("\n");
+        const errText = result.content.map((contentItem) => contentItem.text || "").join("\n");
        throw new BailianError(`Tool error: ${errText}`);
      }

@@ -146,6 +148,8 @@ export default defineCommand({
        rethrowWithMcpActivateHint(error, serverCode);
      }
      throw error;
+    } finally {
+      client?.close?.();
    }
  },
 });
@@ -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: [
@@ -28,24 +29,27 @@ export default defineCommand({
    const { settings, flags } = ctx;
    const code = flags.server;

-    const url = flags.url || ctx.client.url(bailianMcpPath(code));
+    const previewUrl = flags.url || ctx.client.url(bailianMcpPath(code));
    const format = detectOutputFormat(settings.output);

    if (settings.dryRun) {
-      emitResult({ server: code, url, action: "tools/list" }, format);
+      emitResult({ server: code, url: previewUrl, action: "tools/list" }, format);
      return;
    }

-    const client = ctx.client.mcp(url);
+    let client: { close?(): void } | undefined;
    try {
-      await client.initialize();
-      const tools = await client.listTools();
-      emitResult({ server: code, url, tools }, format);
+      const connected = await ctx.client.connectBailianMcp(code, flags.url);
+      client = connected.client;
+      const tools = await connected.client.listTools();
+      emitResult({ server: code, url: connected.url, tools }, format);
    } catch (error) {
      if (!flags.url) {
        rethrowWithMcpActivateHint(error, code);
      }
      throw error;
+    } finally {
+      client?.close?.();
    }
  },
 });
@@ -26,6 +26,12 @@ describe("mcp-activate-hint", () => {
      false,
    );
    expect(isMcpNotActivated(new Error("MCP不存在或未开通"))).toBe(false);
+    // Nested wrapper phrase must not match (anchored at start).
+    expect(
+      isMcpNotActivated(
+        new BailianError("MCP error (-32000): MCP request failed: 404 Not Found - 未开通"),
+      ),
+    ).toBe(false);
  });

  test("hint 含对应 server 的 MCP 广场深链", () => {
@@ -38,6 +44,36 @@ describe("mcp-activate-hint", () => {
    expect(mcpActivateHint("WebSearch")).toMatch(/SSE|Streamable HTTP/i);
  });

+  test("WebSearch + 405 streamableHttp 补重开通 hint", () => {
+    const original = new BailianError(
+      "MCP request failed: 405 Method Not Allowed - current mcp not support streamableHttp",
+      ExitCode.GENERAL,
+    );
+    try {
+      rethrowWithMcpActivateHint(original, "WebSearch");
+      expect.unreachable("should throw");
+    } catch (error) {
+      expect(error).toBeInstanceOf(BailianError);
+      const wrapped = error as BailianError;
+      expect(wrapped.message).toBe(original.message);
+      expect(wrapped.hint).toMatch(/SSE|Streamable HTTP|Activate|re-activate/i);
+      expect(wrapped.hint).toContain(mcpMarketplaceDetailPage("WebSearch"));
+    }
+  });
+
+  test("非 WebSearch 的 405 streamableHttp 不补 hint（由 fallback 处理）", () => {
+    const original = new BailianError(
+      "MCP request failed: 405 Method Not Allowed - current mcp not support streamableHttp",
+      ExitCode.GENERAL,
+    );
+    try {
+      rethrowWithMcpActivateHint(original, "WebParser");
+      expect.unreachable("should throw");
+    } catch (error) {
+      expect(error).toBe(original);
+    }
+  });
+
  test("rethrow 保留原 message，补 hint", () => {
    const serverCode = "market-cmapi00073529";
    const original = new BailianError(
@@ -5,7 +5,13 @@ import { ExitCode } from "../errors/codes.ts";
 import { request, requestJson, type HttpDeps, type RequestOpts } from "./http.ts";
 import { buildAcsCanonicalQuery, signAcsRequest, type AcsQueryParams } from "./acs.ts";
 import { imageFileToDataUri, isLocalFile, resolveFileUrl } from "../files/upload.ts";
-import { McpClient } from "./mcp.ts";
+import {
+  bailianMcpPath,
+  bailianMcpSsePath,
+  connectBailianMcpWithFallback,
+  McpClient,
+  type McpConnectedClient,
+} from "./mcp.ts";
 import { callConsoleGateway } from "../console/gateway.ts";
 import { refreshAccessToken } from "../auth/refresh-token.ts";
 import { maskToken } from "../utils/token.ts";
@@ -164,6 +170,25 @@ export class Client {
    return new McpClient(this.http, url, this.deps.apiCred?.token);
  }

+  /**
+   * 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,
+    urlOverride?: string,
+  ): Promise<{ client: McpConnectedClient; url: string }> {
+    this.requireApi();
+    return connectBailianMcpWithFallback({
+      deps: this.http,
+      authToken: this.deps.apiCred?.token,
+      httpUrl: this.url(bailianMcpPath(serverCode)),
+      sseUrl: this.url(bailianMcpSsePath(serverCode)),
+      serverCode,
+      urlOverride,
+    });
+  }
+
  async console<T>(api: string, data: Record<string, unknown>): Promise<T> {
    if (!this.deps.consoleCred) {
      throw new BailianError("This command needs a console access token.", ExitCode.AUTH);
@@ -69,7 +69,19 @@ export {
  type AcsQueryParams,
  type AcsSignConfig,
 } from "./acs.ts";
-export type { McpTool, McpToolResult } from "./mcp.ts";
-export { McpClient, bailianMcpPath } from "./mcp.ts";
+export type {
+  McpTool,
+  McpToolResult,
+  McpConnectedClient,
+  ConnectBailianMcpOptions,
+} from "./mcp.ts";
+export {
+  McpClient,
+  bailianMcpPath,
+  bailianMcpSsePath,
+  isStreamableHttpUnsupported,
+  isUrlOverrideSseFallbackCandidate,
+  connectBailianMcpWithFallback,
+} from "./mcp.ts";
 export type { ServerSentEvent } from "./stream.ts";
 export { parseSSE } from "./stream.ts";
@@ -0,0 +1,474 @@
+/**
+ * MCP classic HTTP+SSE client (protocol 2024-11-05 transport).
+ *
+ * Flow: GET /sse → endpoint event → POST JSON-RPC to message URL;
+ * responses arrive as SSE `message` events matched by JSON-RPC id.
+ */
+
+import { BailianError } from "../errors/base.ts";
+import { ExitCode } from "../errors/codes.ts";
+import type { HttpDeps } from "./http.ts";
+import { trackingHeaders } from "./headers.ts";
+import type { McpTool, McpToolResult } from "./mcp.ts";
+import { parseSSE } from "./stream.ts";
+
+interface JsonRpcResponse {
+  jsonrpc: "2.0";
+  id?: number | string | null;
+  result?: unknown;
+  error?: { code: number; message: string; data?: unknown };
+}
+
+type PendingResolver = {
+  resolve: (value: JsonRpcResponse) => void;
+  reject: (reason: unknown) => void;
+};
+
+/** Match JSON-RPC ids with string keys (number or string echo from server). */
+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<string, PendingResolver>();
+  private endpointReady: Promise<void>;
+  private resolveEndpoint: (() => void) | undefined;
+  private rejectEndpoint: ((reason: unknown) => void) | undefined;
+  private closed = false;
+  /** Set when the SSE GET ends without an intentional close(); later RPCs fail fast. */
+  private streamEnded = false;
+
+  constructor(deps: HttpDeps, sseUrl: string, authToken?: string) {
+    this.deps = deps;
+    this.sseUrl = sseUrl;
+    this.authToken = authToken;
+    this.endpointReady = new Promise<void>((resolve, reject) => {
+      this.resolveEndpoint = resolve;
+      this.rejectEndpoint = reject;
+    });
+  }
+
+  /** Open the SSE session and run initialize / notifications/initialized. */
+  async initialize(): Promise<void> {
+    if (!this.authToken) {
+      throw new BailianError("This command needs a model-domain API key.", ExitCode.AUTH);
+    }
+
+    await this.openSse();
+
+    const result = await this.rpc("initialize", {
+      protocolVersion: "2025-03-26",
+      capabilities: {},
+      clientInfo: {
+        name: this.deps.identity.clientName,
+        version: this.deps.identity.version,
+      },
+    });
+
+    if (this.deps.settings.verbose) {
+      console.error(`[MCP SSE] Session initialized`);
+      console.error(`[MCP SSE] Server: ${JSON.stringify(result)}`);
+    }
+
+    await this.notify("notifications/initialized");
+  }
+
+  async listTools(): Promise<McpTool[]> {
+    const result = (await this.rpc("tools/list")) as { tools: McpTool[] };
+    return result.tools || [];
+  }
+
+  async callTool(name: string, args: Record<string, unknown>): Promise<McpToolResult> {
+    const result = (await this.rpc("tools/call", { name, arguments: args })) as McpToolResult;
+    return result;
+  }
+
+  /** Abort the hanging GET /sse so the CLI process can exit. */
+  close(): void {
+    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(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;
+
+    // One abortController for header/error-body wait; clear timer before the long-lived stream.
+    this.abortController = new AbortController();
+    const timeoutMs = this.deps.settings.timeout * 1000;
+    let headerTimedOut = false;
+    const headerTimer = setTimeout(() => {
+      headerTimedOut = true;
+      this.abortController?.abort();
+    }, timeoutMs);
+
+    const headers: Record<string, string> = {
+      Accept: "text/event-stream",
+      "User-Agent": `${this.deps.identity.clientName}/${this.deps.identity.version}`,
+      ...trackingHeaders(this.deps.identity),
+    };
+    if (this.authToken) {
+      headers["Authorization"] = `Bearer ${this.authToken}`;
+    }
+
+    if (this.deps.settings.verbose) {
+      console.error(`> GET ${this.sseUrl}`);
+    }
+
+    let response: Response;
+    try {
+      response = await fetch(this.sseUrl, {
+        method: "GET",
+        headers,
+        signal: this.abortController.signal,
+      });
+    } catch (error) {
+      clearTimeout(headerTimer);
+      // Allow a later initialize() to openSse again on this instance.
+      this.abortController = undefined;
+      if (this.closed) {
+        throw new BailianError("MCP SSE session closed.", ExitCode.GENERAL);
+      }
+      if (headerTimedOut) {
+        throw new BailianError("MCP SSE timed out waiting for response headers.", ExitCode.TIMEOUT);
+      }
+      // Rethrow fetch failures so runtime can surface errno (e.g. ENOTFOUND) in JSON/text.
+      throw error;
+    }
+
+    if (this.deps.settings.verbose) {
+      console.error(`< ${response.status} ${response.statusText}`);
+    }
+
+    if (!response.ok) {
+      // Keep headerTimer until error body is read (or times out).
+      let errMsg = `MCP request failed: ${response.status} ${response.statusText}`;
+      try {
+        const errBody = await response.text();
+        if (errBody) errMsg += ` - ${errBody.slice(0, 500)}`;
+      } catch (error) {
+        clearTimeout(headerTimer);
+        this.abortController = undefined;
+        if (this.closed) {
+          throw new BailianError("MCP SSE session closed.", ExitCode.GENERAL);
+        }
+        if (headerTimedOut) {
+          throw new BailianError(
+            "MCP SSE timed out reading error response body.",
+            ExitCode.TIMEOUT,
+          );
+        }
+        throw new BailianError(errMsg, ExitCode.GENERAL, undefined, { cause: error });
+      }
+      clearTimeout(headerTimer);
+      this.abortController = undefined;
+      // Do not rejectEndpoint — openSse never awaits endpointReady on this path.
+      throw new BailianError(errMsg, ExitCode.GENERAL);
+    }
+
+    clearTimeout(headerTimer);
+
+    void this.consumeSse(response).catch((error) => {
+      if (this.closed) return;
+      const reason =
+        error instanceof BailianError
+          ? error
+          : new BailianError(
+              `MCP SSE stream failed: ${error instanceof Error ? error.message : String(error)}`,
+              ExitCode.GENERAL,
+            );
+      this.rejectEndpoint?.(reason);
+      // consumeSse already markStreamEnded on a clean end; cover parse/read failures here.
+      if (!this.streamEnded) {
+        this.markStreamEnded(reason);
+      }
+    });
+
+    const endpointTimeout = cancellableTimeoutReject(
+      timeoutMs,
+      "MCP SSE timed out waiting for endpoint event.",
+    );
+    try {
+      await Promise.race([this.endpointReady, endpointTimeout.promise]);
+    } finally {
+      endpointTimeout.cancel();
+    }
+  }
+
+  private async consumeSse(response: Response): Promise<void> {
+    for await (const event of parseSSE(response)) {
+      if (this.closed) break;
+
+      // Spec requires event: endpoint; ignore unnamed events so JSON is not treated as a 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.
+        this.messageUrl = resolveSameOriginMessageUrl(this.sseUrl, raw);
+        this.resolveEndpoint?.();
+        this.resolveEndpoint = undefined;
+        this.rejectEndpoint = undefined;
+        continue;
+      }
+
+      // Omitted SSE event type defaults to "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" && typeof payload.id !== "string") continue;
+        const key = pendingKey(payload.id);
+        const waiter = this.pending.get(key);
+        if (!waiter) continue;
+        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.",
+        ExitCode.GENERAL,
+      );
+      this.rejectEndpoint?.(error);
+      throw error;
+    }
+
+    // After endpoint: mark dead and wake pending; don't throw (avoid unhandledRejection).
+    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,
+      method,
+      ...(params ? { params } : {}),
+    };
+
+    const timeoutMs = this.deps.settings.timeout * 1000;
+    const responsePromise = new Promise<JsonRpcResponse>((resolve, reject) => {
+      this.pending.set(key, { resolve, reject });
+    });
+    // Stream may end and reject pending before Promise.race; attach catch to avoid 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(
+          `MCP error (${data.error.code}): ${data.error.message}`,
+          ExitCode.GENERAL,
+        );
+      }
+      return data.result;
+    } catch (error) {
+      this.pending.delete(key);
+      throw error;
+    } finally {
+      responseTimeout.cancel();
+    }
+  }
+
+  private async notify(method: string, params?: Record<string, unknown>): Promise<void> {
+    const body = {
+      jsonrpc: "2.0" as const,
+      method,
+      ...(params ? { params } : {}),
+    };
+    await this.postMessage(body);
+  }
+
+  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);
+    }
+
+    const headers: Record<string, string> = {
+      "Content-Type": "application/json",
+      Accept: "application/json, text/event-stream",
+      "User-Agent": `${this.deps.identity.clientName}/${this.deps.identity.version}`,
+      ...trackingHeaders(this.deps.identity),
+    };
+    // Bearer is only sent to a messageUrl that already passed the same-origin check.
+    if (this.authToken) {
+      headers["Authorization"] = `Bearer ${this.authToken}`;
+    }
+
+    if (this.deps.settings.verbose) {
+      console.error(`> POST ${this.messageUrl}`);
+      console.error(`> Method: ${(body as { method?: string }).method}`);
+    }
+
+    const timeoutMs = this.deps.settings.timeout * 1000;
+    // Combine per-RPC timeout with session abort so close() cancels in-flight POSTs.
+    const requestSignal = createLinkedAbortSignal(timeoutMs, this.abortController?.signal);
+    let res: Response;
+    try {
+      try {
+        res = await fetch(this.messageUrl, {
+          method: "POST",
+          headers,
+          body: JSON.stringify(body),
+          signal: requestSignal.signal,
+        });
+      } catch (error) {
+        if (this.closed) {
+          throw new BailianError("MCP SSE session closed.", ExitCode.GENERAL);
+        }
+        throw error;
+      }
+
+      if (this.deps.settings.verbose) {
+        console.error(`< ${res.status} ${res.statusText}`);
+      }
+
+      if (!res.ok) {
+        // Keep signal until error body is read (same class of bug as GET openSse).
+        let errMsg = `MCP request failed: ${res.status} ${res.statusText}`;
+        try {
+          const errBody = await res.text();
+          if (errBody) errMsg += ` - ${errBody.slice(0, 500)}`;
+        } catch (error) {
+          if (this.closed) {
+            throw new BailianError("MCP SSE session closed.", ExitCode.GENERAL);
+          }
+          if (requestSignal.timedOut) {
+            throw new BailianError(
+              "MCP SSE timed out reading error response body.",
+              ExitCode.TIMEOUT,
+            );
+          }
+          throw new BailianError(errMsg, ExitCode.GENERAL, undefined, { cause: error });
+        }
+        throw new BailianError(errMsg, ExitCode.GENERAL);
+      }
+    } finally {
+      requestSignal.cleanup();
+    }
+  }
+}
+
+/** Resolve the SSE endpoint data to an absolute URL and require same origin as sseUrl. */
+export function resolveSameOriginMessageUrl(sseUrl: string, endpointData: string): string {
+  let resolved: URL;
+  let base: URL;
+  try {
+    base = new URL(sseUrl);
+    resolved = new URL(endpointData, sseUrl);
+  } catch {
+    throw new BailianError(
+      `MCP SSE endpoint is not a valid URL: ${endpointData}`,
+      ExitCode.GENERAL,
+    );
+  }
+  if (resolved.origin !== base.origin) {
+    throw new BailianError(
+      `MCP SSE endpoint origin mismatch: expected ${base.origin}, got ${resolved.origin}`,
+      ExitCode.GENERAL,
+    );
+  }
+  return resolved.toString();
+}
+
+/**
+ * Cancellable timeout rejection: after Promise.race settles, call cancel()
+ * to clear the timer and avoid unhandledRejection.
+ */
+function cancellableTimeoutReject(
+  timeoutMs: number,
+  message: string,
+): { promise: Promise<never>; cancel: () => void } {
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  const promise = new Promise<never>((_, reject) => {
+    timer = setTimeout(() => {
+      timer = undefined;
+      reject(new BailianError(message, ExitCode.TIMEOUT));
+    }, timeoutMs);
+  });
+  // Swallow late rejects after cancel to avoid unhandledRejection.
+  void promise.catch(() => undefined);
+
+  return {
+    promise,
+    cancel: () => {
+      if (timer !== undefined) {
+        clearTimeout(timer);
+        timer = undefined;
+      }
+    },
+  };
+}
+
+/** Timeout + optional parent abort without AbortSignal.any (Node 18). */
+function createLinkedAbortSignal(
+  timeoutMs: number,
+  parentSignal?: AbortSignal,
+): { signal: AbortSignal; cleanup: () => void; timedOut: boolean } {
+  const controller = new AbortController();
+  const state = { timedOut: false };
+  const timeout = setTimeout(() => {
+    state.timedOut = true;
+    controller.abort();
+  }, timeoutMs);
+  const abortFromParent = () => controller.abort(parentSignal?.reason);
+  const cleanup = () => {
+    clearTimeout(timeout);
+    parentSignal?.removeEventListener("abort", abortFromParent);
+  };
+
+  if (parentSignal?.aborted) abortFromParent();
+  else parentSignal?.addEventListener("abort", abortFromParent, { once: true });
+  controller.signal.addEventListener("abort", cleanup, { once: true });
+
+  return {
+    signal: controller.signal,
+    cleanup,
+    get timedOut() {
+      return state.timedOut;
+    },
+  };
+}
@@ -15,6 +15,8 @@ import { BailianError } from "../errors/base.ts";
 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 ----

@@ -27,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 };
 }
@@ -61,6 +63,101 @@ export function bailianMcpPath(serverCode: string): string {
  return `/api/v1/mcps/${serverCode}/mcp`;
 }

+/** Classic SSE path: `/api/v1/mcps/<serverCode>/sse`. */
+export function bailianMcpSsePath(serverCode: string): string {
+  return `/api/v1/mcps/${serverCode}/sse`;
+}
+
+/**
+ * True when Streamable HTTP is unsupported and classic SSE fallback should be tried.
+ * Anchored to HTTP wrapper text only (not JSON-RPC / nested copies). Bailian 404 excluded.
+ */
+export function isStreamableHttpUnsupported(error: unknown): boolean {
+  if (!(error instanceof BailianError)) return false;
+  return /^MCP request failed:\s*405\b/i.test(error.message);
+}
+
+/**
+ * SSE fallback for `--url` (official backwards-compat: same URL, HTTP 405/404 then GET SSE).
+ */
+export function isUrlOverrideSseFallbackCandidate(error: unknown): boolean {
+  if (!(error instanceof BailianError)) return false;
+  return /^MCP request failed:\s*(405|404)\b/i.test(error.message);
+}
+
+export type McpConnectedClient = {
+  initialize(): Promise<void>;
+  listTools(): Promise<McpTool[]>;
+  callTool(name: string, args: Record<string, unknown>): Promise<McpToolResult>;
+  close?(): void;
+};
+
+export type ConnectBailianMcpOptions = {
+  deps: HttpDeps;
+  authToken: string | undefined;
+  /** Full Streamable HTTP URL (/mcp). */
+  httpUrl: string;
+  /** Full classic SSE URL (/sse). */
+  sseUrl: string;
+  serverCode: string;
+  /**
+   * 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 (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(
+  options: ConnectBailianMcpOptions,
+): Promise<{ client: McpConnectedClient; url: string }> {
+  const { deps, authToken, httpUrl, sseUrl, serverCode, urlOverride } = options;
+
+  if (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);
+  try {
+    await httpClient.initialize();
+    return { client: httpClient, url: httpUrl };
+  } catch (error) {
+    if (!isStreamableHttpUnsupported(error) || serverCode === "WebSearch") {
+      throw error;
+    }
+  }
+
+  const sseClient = new McpSseClient(deps, sseUrl, authToken);
+  try {
+    await sseClient.initialize();
+    return { client: sseClient, url: sseUrl };
+  } catch (error) {
+    sseClient.close();
+    throw error;
+  }
+}
+
 // ---- MCP Client ----

 export class McpClient {
@@ -121,7 +218,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(
@@ -143,6 +240,44 @@ export class McpClient {
    await this.send(body);
  }

+  /**
+   * Read a JSON-RPC response by Content-Type: application/json or 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",
@@ -7,6 +7,71 @@ export interface ServerSentEvent {
  id?: string;
 }

+/** Normalize CRLF/CR to LF; hold a trailing `\r` so a split CRLF is not double-broken. */
+function takeNormalizedSseLines(buffer: string): { lines: string[]; rest: string } {
+  let text = buffer;
+  let holdTrailingCr = false;
+  if (text.endsWith("\r")) {
+    holdTrailingCr = true;
+    text = text.slice(0, -1);
+  }
+
+  text = text.replace(/\r\n/g, "\n").replace(/\r/g, "\n");
+  const parts = text.split("\n");
+  const incomplete = parts.pop() ?? "";
+  return {
+    lines: parts,
+    rest: holdTrailingCr ? `${incomplete}\r` : incomplete,
+  };
+}
+
+function applySseLine(
+  line: string,
+  event: Partial<ServerSentEvent>,
+  maxBuffer: number,
+): { event: Partial<ServerSentEvent>; completed?: ServerSentEvent } {
+  if (line === "") {
+    if (event.data === undefined) {
+      return { event: {} };
+    }
+    return {
+      event: {},
+      completed: { data: event.data, event: event.event, id: event.id },
+    };
+  }
+
+  if (line.startsWith(":")) {
+    return { event };
+  }
+
+  const colonIndex = line.indexOf(":");
+  if (colonIndex === -1) {
+    return { event };
+  }
+
+  const field = line.slice(0, colonIndex);
+  const fieldValue = line.slice(colonIndex + 1).trimStart();
+  const nextEvent: Partial<ServerSentEvent> = { ...event };
+
+  switch (field) {
+    case "data":
+      nextEvent.data =
+        nextEvent.data !== undefined ? `${nextEvent.data}\n${fieldValue}` : fieldValue;
+      if (nextEvent.data.length > maxBuffer) {
+        throw new BailianError("SSE event exceeded the maximum buffer size.", ExitCode.GENERAL);
+      }
+      break;
+    case "event":
+      nextEvent.event = fieldValue;
+      break;
+    case "id":
+      nextEvent.id = fieldValue;
+      break;
+  }
+
+  return { event: nextEvent };
+}
+
 export async function* parseSSE(response: Response): AsyncGenerator<ServerSentEvent> {
  const reader = response.body?.getReader();
  if (!reader) return;
@@ -14,69 +79,55 @@ export async function* parseSSE(response: Response): AsyncGenerator<ServerSentEv
  const decoder = new TextDecoder();
  let buffer = "";

-  // Guard against a hostile or malfunctioning stream that never emits a newline
-  // (or builds a single absurdly large event): bound the in-memory buffer so the
-  // parser cannot be driven to exhaust process memory.
  const MAX_SSE_BUFFER = 16 * 1024 * 1024; // 16 MiB

  try {
+    // Keep partial event fields across chunks.
+    let event: Partial<ServerSentEvent> = {};
+
    while (true) {
      const { done, value } = await reader.read();
-      if (done) break;
+      if (done) {
+        // EOF: treat any held `\r` as a line ending.
+        if (buffer.length > 0) {
+          const finalText = buffer.replace(/\r\n/g, "\n").replace(/\r/g, "\n");
+          const parts = finalText.split("\n");
+          buffer = parts.pop() ?? "";
+          for (const line of parts) {
+            const applied = applySseLine(line, event, MAX_SSE_BUFFER);
+            event = applied.event;
+            if (applied.completed) {
+              yield applied.completed;
+            }
+          }
+        }
+        break;
+      }

      buffer += decoder.decode(value, { stream: true });
      if (buffer.length > MAX_SSE_BUFFER) {
        throw new BailianError("SSE stream exceeded the maximum buffer size.", ExitCode.GENERAL);
      }

-      const lines = buffer.split("\n");
-      buffer = lines.pop() || "";
-
-      let event: Partial<ServerSentEvent> = {};
+      const { lines, rest } = takeNormalizedSseLines(buffer);
+      buffer = rest;

      for (const line of lines) {
-        if (line === "") {
-          if (event.data !== undefined) {
-            yield { data: event.data, event: event.event, id: event.id };
-          }
-          event = {};
-          continue;
-        }
-
-        if (line.startsWith(":")) continue; // comment
-
-        const colonIndex = line.indexOf(":");
-        if (colonIndex === -1) continue;
-
-        const field = line.slice(0, colonIndex);
-        const value = line.slice(colonIndex + 1).trimStart();
-
-        switch (field) {
-          case "data":
-            event.data = event.data !== undefined ? `${event.data}\n${value}` : value;
-            if (event.data.length > MAX_SSE_BUFFER) {
-              throw new BailianError(
-                "SSE event exceeded the maximum buffer size.",
-                ExitCode.GENERAL,
-              );
-            }
-            break;
-          case "event":
-            event.event = value;
-            break;
-          case "id":
-            event.id = value;
-            break;
+        const applied = applySseLine(line, event, MAX_SSE_BUFFER);
+        event = applied.event;
+        if (applied.completed) {
+          yield applied.completed;
        }
      }
    }

-    // Flush remaining
-    if (buffer.trim() && buffer.includes("data:")) {
-      const colonIndex = buffer.indexOf(":");
-      if (colonIndex !== -1) {
-        yield { data: buffer.slice(colonIndex + 1).trimStart() };
-      }
+    // Legacy EOF flush: apply trailing field line and dispatch with event/id intact.
+    if (buffer.length > 0) {
+      const applied = applySseLine(buffer, event, MAX_SSE_BUFFER);
+      event = applied.event;
+    }
+    if (event.data !== undefined) {
+      yield { data: event.data, event: event.event, id: event.id };
    }
  } finally {
    reader.releaseLock();
@@ -0,0 +1,736 @@
+import { expect, test } from "vite-plus/test";
+import type { Identity, Settings } from "../src/index.ts";
+import {
+  BailianError,
+  bailianMcpPath,
+  bailianMcpSsePath,
+  connectBailianMcpWithFallback,
+  isStreamableHttpUnsupported,
+  isUrlOverrideSseFallbackCandidate,
+  McpClient,
+} from "../src/index.ts";
+import { McpSseClient, resolveSameOriginMessageUrl } from "../src/client/mcp-sse.ts";
+
+function testDeps(overrides?: Partial<Settings>): { identity: Identity; settings: Settings } {
+  return {
+    identity: {
+      binName: "bl",
+      version: "0.0.0-test",
+      npmPackage: "bailian-cli",
+      clientName: "bailian-cli",
+    },
+    settings: {
+      output: "json",
+      outputExplicit: true,
+      timeout: 5,
+      verbose: false,
+      quiet: true,
+      dryRun: false,
+      telemetry: true,
+      ...overrides,
+    },
+  };
+}
+
+function jsonRpcResult(id: number | string, result: unknown): string {
+  return `event:message\ndata:${JSON.stringify({ jsonrpc: "2.0", id, result })}\n\n`;
+}
+
+function requestUrl(input: string | URL | Request): string {
+  if (typeof input === "string") return input;
+  if (input instanceof URL) return input.href;
+  return input.url;
+}
+
+test("bailianMcp 路径与 isStreamableHttpUnsupported", () => {
+  expect(bailianMcpPath("WebParser")).toBe("/api/v1/mcps/WebParser/mcp");
+  expect(bailianMcpSsePath("WebParser")).toBe("/api/v1/mcps/WebParser/sse");
+
+  expect(
+    isStreamableHttpUnsupported(
+      new BailianError(
+        "MCP request failed: 405 Method Not Allowed - current mcp not support streamableHttp",
+      ),
+    ),
+  ).toBe(true);
+  expect(
+    isStreamableHttpUnsupported(new BailianError("MCP request failed: 405 Method Not Allowed")),
+  ).toBe(true);
+  expect(isStreamableHttpUnsupported(new BailianError("MCP request failed: 404 Not Found"))).toBe(
+    false,
+  );
+  expect(isStreamableHttpUnsupported(new Error("405 streamableHttp"))).toBe(false);
+  // JSON-RPC business 405 must not trigger HTTP transport fallback
+  expect(isStreamableHttpUnsupported(new BailianError("MCP error (405): Method Not Allowed"))).toBe(
+    false,
+  );
+  // Nested wrapper phrase in a JSON-RPC message must not trigger fallback.
+  expect(
+    isStreamableHttpUnsupported(
+      new BailianError("MCP error (-32000): MCP request failed: 405 Method Not Allowed"),
+    ),
+  ).toBe(false);
+
+  expect(
+    isUrlOverrideSseFallbackCandidate(new BailianError("MCP request failed: 404 Not Found")),
+  ).toBe(true);
+  expect(
+    isUrlOverrideSseFallbackCandidate(
+      new BailianError("MCP request failed: 405 Method Not Allowed"),
+    ),
+  ).toBe(true);
+  expect(isUrlOverrideSseFallbackCandidate(new BailianError("MCP error (404): not found"))).toBe(
+    false,
+  );
+  expect(
+    isUrlOverrideSseFallbackCandidate(
+      new BailianError("MCP error (-32000): MCP request failed: 404 Not Found"),
+    ),
+  ).toBe(false);
+});
+
+test("resolveSameOriginMessageUrl：同源通过、跨域拒绝", () => {
+  expect(
+    resolveSameOriginMessageUrl(
+      "https://example.test/api/v1/mcps/WebParser/sse",
+      "/api/v1/mcps/WebParser/message?sessionId=x",
+    ),
+  ).toBe("https://example.test/api/v1/mcps/WebParser/message?sessionId=x");
+
+  expect(() =>
+    resolveSameOriginMessageUrl(
+      "https://example.test/api/v1/mcps/WebParser/sse",
+      "https://evil.example/steal",
+    ),
+  ).toThrow(/origin mismatch/i);
+});
+
+test("connectBailianMcpWithFallback：成功走 Streamable；405 降级 SSE", async () => {
+  const originalFetch = globalThis.fetch;
+
+  // Streamable success path
+  globalThis.fetch = async (input, init) => {
+    const body = typeof init?.body === "string" ? JSON.parse(init.body) : {};
+    if (requestUrl(input).includes("/sse")) {
+      return new Response("should not hit sse", { status: 500 });
+    }
+    if (body.method === "notifications/initialized") {
+      return new Response(null, { status: 200 });
+    }
+    return new Response(JSON.stringify({ jsonrpc: "2.0", id: body.id, result: {} }), {
+      status: 200,
+    });
+  };
+
+  try {
+    const connected = await connectBailianMcpWithFallback({
+      deps: testDeps(),
+      authToken: "sk-test",
+      httpUrl: "https://example.test/api/v1/mcps/WebParser/mcp",
+      sseUrl: "https://example.test/api/v1/mcps/WebParser/sse",
+      serverCode: "WebParser",
+    });
+    expect(connected.url).toContain("/mcp");
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+
+  // Bare HTTP 405 (no streamableHttp body text) → SSE
+  let sseController: ReadableStreamDefaultController<Uint8Array> | undefined;
+  const encoder = new TextEncoder();
+  const urls: string[] = [];
+
+  globalThis.fetch = async (input, init) => {
+    const url = requestUrl(input);
+    urls.push(`${init?.method ?? "GET"} ${url}`);
+
+    if (url.endsWith("/mcp")) {
+      return new Response("Method Not Allowed", {
+        status: 405,
+        statusText: "Method Not Allowed",
+      });
+    }
+
+    if (url.endsWith("/sse") && (init?.method ?? "GET") === "GET") {
+      const stream = new ReadableStream<Uint8Array>({
+        start(controller) {
+          sseController = controller;
+          controller.enqueue(
+            encoder.encode(
+              "event:endpoint\ndata:/api/v1/mcps/WebParser/message?sessionId=test-session\n\n",
+            ),
+          );
+        },
+      });
+      return new Response(stream, {
+        status: 200,
+        headers: { "Content-Type": "text/event-stream" },
+      });
+    }
+
+    if (url.includes("/message")) {
+      const body = typeof init?.body === "string" ? JSON.parse(init.body) : {};
+      queueMicrotask(() => {
+        if (body.id != null && sseController) {
+          sseController.enqueue(encoder.encode(jsonRpcResult(body.id, {})));
+        }
+      });
+      return new Response(null, { status: 200 });
+    }
+
+    return new Response("unexpected", { status: 500 });
+  };
+
+  try {
+    const connected = await connectBailianMcpWithFallback({
+      deps: testDeps(),
+      authToken: "sk-test",
+      httpUrl: "https://example.test/api/v1/mcps/WebParser/mcp",
+      sseUrl: "https://example.test/api/v1/mcps/WebParser/sse",
+      serverCode: "WebParser",
+    });
+    expect(connected.url).toContain("/sse");
+    expect(urls.some((entry) => entry.includes("GET ") && entry.includes("/sse"))).toBe(true);
+    connected.client.close?.();
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});
+
+test("connectBailianMcpWithFallback：WebSearch 不降级；urlOverride 同 URL 降级 SSE；404 不降级 Bailian 路径", async () => {
+  const originalFetch = globalThis.fetch;
+  const urls: string[] = [];
+
+  globalThis.fetch = async (input) => {
+    urls.push(requestUrl(input));
+    return new Response("current mcp not support streamableHttp", {
+      status: 405,
+      statusText: "Method Not Allowed",
+    });
+  };
+
+  try {
+    await expect(
+      connectBailianMcpWithFallback({
+        deps: testDeps(),
+        authToken: "sk-test",
+        httpUrl: "https://example.test/api/v1/mcps/WebSearch/mcp",
+        sseUrl: "https://example.test/api/v1/mcps/WebSearch/sse",
+        serverCode: "WebSearch",
+      }),
+    ).rejects.toBeInstanceOf(BailianError);
+    expect(urls.some((url) => url.includes("/sse"))).toBe(false);
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+
+  // urlOverride: after POST 405, fall back with GET SSE on the same URL
+  urls.length = 0;
+  let sseController: ReadableStreamDefaultController<Uint8Array> | undefined;
+  const encoder = new TextEncoder();
+  const overrideUrl = "https://custom.example/mcp";
+
+  globalThis.fetch = async (input, init) => {
+    const url = requestUrl(input);
+    const method = init?.method ?? "GET";
+    urls.push(`${method} ${url}`);
+
+    if (method === "POST" && url === overrideUrl) {
+      return new Response("Method Not Allowed", {
+        status: 405,
+        statusText: "Method Not Allowed",
+      });
+    }
+
+    if (method === "GET" && url === overrideUrl) {
+      const stream = new ReadableStream<Uint8Array>({
+        start(controller) {
+          sseController = controller;
+          controller.enqueue(encoder.encode("event:endpoint\ndata:/message?sessionId=x\n\n"));
+        },
+      });
+      return new Response(stream, {
+        status: 200,
+        headers: { "Content-Type": "text/event-stream" },
+      });
+    }
+
+    if (url.includes("/message")) {
+      const body = typeof init?.body === "string" ? JSON.parse(init.body) : {};
+      queueMicrotask(() => {
+        if (body.id != null && sseController) {
+          sseController.enqueue(encoder.encode(jsonRpcResult(body.id, {})));
+        }
+      });
+      return new Response(null, { status: 200 });
+    }
+
+    return new Response("unexpected", { status: 500 });
+  };
+
+  try {
+    const connected = await connectBailianMcpWithFallback({
+      deps: testDeps(),
+      authToken: "sk-test",
+      httpUrl: "https://example.test/api/v1/mcps/WebParser/mcp",
+      sseUrl: "https://example.test/api/v1/mcps/WebParser/sse",
+      serverCode: "WebParser",
+      urlOverride: overrideUrl,
+    });
+    expect(connected.url).toBe(overrideUrl);
+    expect(urls.some((entry) => entry.startsWith(`GET ${overrideUrl}`))).toBe(true);
+    connected.client.close?.();
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+
+  globalThis.fetch = async () =>
+    new Response("MCP不存在或未开通", { status: 404, statusText: "Not Found" });
+
+  try {
+    await expect(
+      connectBailianMcpWithFallback({
+        deps: testDeps(),
+        authToken: "sk-test",
+        httpUrl: "https://example.test/api/v1/mcps/WebParser/mcp",
+        sseUrl: "https://example.test/api/v1/mcps/WebParser/sse",
+        serverCode: "WebParser",
+      }),
+    ).rejects.toMatchObject({ message: expect.stringContaining("404") });
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});
+
+test("McpSseClient：流结束后立刻失败 pending（不干等到 timeout）", async () => {
+  const originalFetch = globalThis.fetch;
+  const encoder = new TextEncoder();
+
+  globalThis.fetch = async (input, init) => {
+    const url = requestUrl(input);
+    if ((init?.method ?? "GET") === "GET" || url.endsWith("/sse")) {
+      // Close the stream immediately after the endpoint event
+      const stream = new ReadableStream<Uint8Array>({
+        start(controller) {
+          controller.enqueue(
+            encoder.encode("event:endpoint\ndata:/api/v1/mcps/WebParser/message?sessionId=x\n\n"),
+          );
+          controller.close();
+        },
+      });
+      return new Response(stream, {
+        status: 200,
+        headers: { "Content-Type": "text/event-stream" },
+      });
+    }
+    return new Response(null, { status: 200 });
+  };
+
+  try {
+    const client = new McpSseClient(
+      testDeps({ timeout: 5 }),
+      "https://example.test/sse",
+      "sk-test",
+    );
+    const started = Date.now();
+    await expect(client.initialize()).rejects.toThrow(/stream ended unexpectedly/i);
+    expect(Date.now() - started).toBeLessThan(2000);
+    client.close();
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});
+
+test("McpSseClient：string JSON-RPC id 可匹配；仅认 event:endpoint", async () => {
+  const originalFetch = globalThis.fetch;
+  let sseController: ReadableStreamDefaultController<Uint8Array> | undefined;
+  const encoder = new TextEncoder();
+
+  globalThis.fetch = async (input, init) => {
+    const url = requestUrl(input);
+    if ((init?.method ?? "GET") === "GET" || url.endsWith("/sse")) {
+      const stream = new ReadableStream<Uint8Array>({
+        start(controller) {
+          sseController = controller;
+          // Untyped events must not be treated as endpoint
+          controller.enqueue(
+            encoder.encode(`data:${JSON.stringify({ jsonrpc: "2.0", id: 99, result: {} })}\n\n`),
+          );
+          controller.enqueue(
+            encoder.encode("event:endpoint\ndata:/api/v1/mcps/WebParser/message?sessionId=x\n\n"),
+          );
+        },
+      });
+      return new Response(stream, {
+        status: 200,
+        headers: { "Content-Type": "text/event-stream" },
+      });
+    }
+
+    if (url.includes("/message")) {
+      const body = typeof init?.body === "string" ? JSON.parse(init.body) : {};
+      queueMicrotask(() => {
+        if (body.id != null && sseController) {
+          // Echo id as a string
+          sseController.enqueue(encoder.encode(jsonRpcResult(String(body.id), {})));
+        }
+      });
+      return new Response(null, { status: 200 });
+    }
+
+    return new Response("unexpected", { status: 500 });
+  };
+
+  try {
+    const client = new McpSseClient(testDeps(), "https://example.test/sse", "sk-test");
+    await client.initialize();
+    client.close();
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});
+
+test("McpClient：支持 text/event-stream 响应体", async () => {
+  const originalFetch = globalThis.fetch;
+
+  globalThis.fetch = async (_input, init) => {
+    const body = typeof init?.body === "string" ? JSON.parse(init.body) : {};
+    if (body.method === "notifications/initialized") {
+      return new Response(null, { status: 202 });
+    }
+    const sse = `event: message\ndata: ${JSON.stringify({
+      jsonrpc: "2.0",
+      id: body.id,
+      result: {
+        protocolVersion: "2025-03-26",
+        capabilities: {},
+        serverInfo: { name: "x", version: "0" },
+      },
+    })}\n\n`;
+    return new Response(sse, {
+      status: 200,
+      headers: { "Content-Type": "text/event-stream" },
+    });
+  };
+
+  try {
+    const client = new McpClient(testDeps(), "https://example.test/mcp", "sk-test");
+    await client.initialize();
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});
+
+test("McpSseClient.close 可中止挂起 GET", async () => {
+  const originalFetch = globalThis.fetch;
+  let aborted = false;
+
+  globalThis.fetch = async (_input, init) => {
+    const signal = init?.signal;
+    if (signal) {
+      signal.addEventListener("abort", () => {
+        aborted = true;
+      });
+    }
+    const stream = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(
+          new TextEncoder().encode(
+            "event:endpoint\ndata:/api/v1/mcps/WebParser/message?sessionId=x\n\n",
+          ),
+        );
+      },
+    });
+    return new Response(stream, {
+      status: 200,
+      headers: { "Content-Type": "text/event-stream" },
+    });
+  };
+
+  try {
+    const client = new McpSseClient(testDeps(), "https://example.test/sse", "sk-test");
+    const initPromise = client.initialize().catch(() => undefined);
+    await new Promise((resolve) => setTimeout(resolve, 20));
+    client.close();
+    await initPromise;
+    expect(aborted).toBe(true);
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});
+
+test("McpSseClient：等待响应头受 --timeout 约束", async () => {
+  const originalFetch = globalThis.fetch;
+
+  globalThis.fetch = async (_input, init) => {
+    const signal = init?.signal;
+    return new Promise((_resolve, reject) => {
+      if (!signal) {
+        reject(new Error("missing signal"));
+        return;
+      }
+      if (signal.aborted) {
+        reject(new DOMException("This operation was aborted.", "AbortError"));
+        return;
+      }
+      signal.addEventListener(
+        "abort",
+        () => reject(new DOMException("This operation was aborted.", "AbortError")),
+        { once: true },
+      );
+    });
+  };
+
+  try {
+    const client = new McpSseClient(
+      testDeps({ timeout: 1 }),
+      "https://example.test/sse",
+      "sk-test",
+    );
+    const started = Date.now();
+    await expect(client.initialize()).rejects.toThrow(/timed out waiting for response headers/i);
+    expect(Date.now() - started).toBeLessThan(2500);
+    client.close();
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});
+
+test("McpSseClient：非 2xx 不产生 unhandledRejection", async () => {
+  const originalFetch = globalThis.fetch;
+  const unhandled: unknown[] = [];
+  const onUnhandled = (reason: unknown) => {
+    unhandled.push(reason);
+  };
+  process.on("unhandledRejection", onUnhandled);
+
+  globalThis.fetch = async () =>
+    new Response("boom", { status: 500, statusText: "Internal Server Error" });
+
+  try {
+    const client = new McpSseClient(testDeps(), "https://example.test/sse", "sk-test");
+    await expect(client.initialize()).rejects.toThrow(/MCP request failed:\s*500/i);
+    await new Promise((resolve) => setTimeout(resolve, 30));
+    expect(unhandled).toEqual([]);
+    client.close();
+  } finally {
+    process.off("unhandledRejection", onUnhandled);
+    globalThis.fetch = originalFetch;
+  }
+});
+
+test("McpSseClient：非 2xx 读 body 仍受 --timeout 约束", async () => {
+  const originalFetch = globalThis.fetch;
+
+  globalThis.fetch = async (_input, init) => {
+    const signal = init?.signal;
+    return {
+      ok: false,
+      status: 500,
+      statusText: "Internal Server Error",
+      async text() {
+        return new Promise<string>((_resolve, reject) => {
+          if (!signal) {
+            reject(new Error("missing signal"));
+            return;
+          }
+          if (signal.aborted) {
+            reject(new DOMException("This operation was aborted.", "AbortError"));
+            return;
+          }
+          signal.addEventListener(
+            "abort",
+            () => reject(new DOMException("This operation was aborted.", "AbortError")),
+            { once: true },
+          );
+        });
+      },
+    } as Response;
+  };
+
+  try {
+    const client = new McpSseClient(
+      testDeps({ timeout: 1 }),
+      "https://example.test/sse",
+      "sk-test",
+    );
+    const started = Date.now();
+    await expect(client.initialize()).rejects.toThrow(/timed out reading error response body/i);
+    expect(Date.now() - started).toBeLessThan(2500);
+    client.close();
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});
+
+test("McpSseClient：fetch 失败抛出原始 TypeError（保留 ENOTFOUND）", async () => {
+  const originalFetch = globalThis.fetch;
+  const root = Object.assign(new Error("getaddrinfo ENOTFOUND example.test"), {
+    code: "ENOTFOUND",
+  });
+  const fetchFailed = new TypeError("fetch failed", { cause: root });
+
+  globalThis.fetch = async () => {
+    throw fetchFailed;
+  };
+
+  try {
+    const client = new McpSseClient(testDeps(), "https://example.test/sse", "sk-test");
+    const error = await client.initialize().catch((reason: unknown) => reason);
+    expect(error).toBe(fetchFailed);
+    expect((error as TypeError & { cause?: NodeJS.ErrnoException }).cause?.code).toBe("ENOTFOUND");
+    client.close();
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});
+
+test("McpSseClient：fetch 失败后同实例可重新 openSse", async () => {
+  const originalFetch = globalThis.fetch;
+  let attempt = 0;
+  let sseController: ReadableStreamDefaultController<Uint8Array> | undefined;
+  const encoder = new TextEncoder();
+
+  globalThis.fetch = async (input, init) => {
+    const url = requestUrl(input);
+    const method = init?.method ?? "GET";
+    if (method === "GET" || url.endsWith("/sse")) {
+      attempt += 1;
+      if (attempt === 1) {
+        throw new TypeError("fetch failed");
+      }
+      const stream = new ReadableStream<Uint8Array>({
+        start(controller) {
+          sseController = controller;
+          controller.enqueue(encoder.encode("event: endpoint\ndata: /message\n\n"));
+        },
+      });
+      return new Response(stream, {
+        status: 200,
+        headers: { "Content-Type": "text/event-stream" },
+      });
+    }
+    const body = typeof init?.body === "string" ? JSON.parse(init.body) : {};
+    queueMicrotask(() => {
+      if (body.id != null && sseController) {
+        sseController.enqueue(encoder.encode(jsonRpcResult(body.id, {})));
+      }
+    });
+    return new Response("{}", { status: 200, headers: { "Content-Type": "application/json" } });
+  };
+
+  try {
+    const client = new McpSseClient(testDeps(), "https://example.test/sse", "sk-test");
+    await expect(client.initialize()).rejects.toThrow(/fetch failed/i);
+    await client.initialize();
+    client.close();
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});
+
+test("McpSseClient：close 可中止进行中的 POST", async () => {
+  const originalFetch = globalThis.fetch;
+  let postAborted = false;
+  const encoder = new TextEncoder();
+
+  globalThis.fetch = async (input, init) => {
+    const url = requestUrl(input);
+    const method = init?.method ?? "GET";
+    if (method === "GET" || url.endsWith("/sse")) {
+      const stream = new ReadableStream<Uint8Array>({
+        start(controller) {
+          controller.enqueue(encoder.encode("event: endpoint\ndata: /message\n\n"));
+        },
+      });
+      return new Response(stream, {
+        status: 200,
+        headers: { "Content-Type": "text/event-stream" },
+      });
+    }
+
+    const signal = init?.signal;
+    return new Promise((_resolve, reject) => {
+      if (!signal) {
+        reject(new Error("missing signal"));
+        return;
+      }
+      const onAbort = () => {
+        postAborted = true;
+        reject(new DOMException("This operation was aborted.", "AbortError"));
+      };
+      if (signal.aborted) {
+        onAbort();
+        return;
+      }
+      signal.addEventListener("abort", onAbort, { once: true });
+    });
+  };
+
+  try {
+    const client = new McpSseClient(
+      testDeps({ timeout: 5 }),
+      "https://example.test/sse",
+      "sk-test",
+    );
+    const initPromise = client.initialize();
+    await new Promise((resolve) => setTimeout(resolve, 30));
+    client.close();
+    await expect(initPromise).rejects.toThrow(/session closed|aborted/i);
+    expect(postAborted).toBe(true);
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});
+
+test("McpSseClient：POST 非 2xx 读 body 仍受 --timeout 约束", async () => {
+  const originalFetch = globalThis.fetch;
+  const encoder = new TextEncoder();
+
+  globalThis.fetch = async (input, init) => {
+    const url = requestUrl(input);
+    const method = init?.method ?? "GET";
+    if (method === "GET" || url.endsWith("/sse")) {
+      const stream = new ReadableStream<Uint8Array>({
+        start(controller) {
+          controller.enqueue(encoder.encode("event: endpoint\ndata: /message\n\n"));
+        },
+      });
+      return new Response(stream, {
+        status: 200,
+        headers: { "Content-Type": "text/event-stream" },
+      });
+    }
+
+    const signal = init?.signal;
+    const body = new ReadableStream<Uint8Array>({
+      start(controller) {
+        if (!signal) return;
+        const onAbort = () => {
+          try {
+            controller.error(new DOMException("This operation was aborted.", "AbortError"));
+          } catch {
+            /* ignore */
+          }
+        };
+        if (signal.aborted) onAbort();
+        else signal.addEventListener("abort", onAbort, { once: true });
+      },
+    });
+    return new Response(body, { status: 500, statusText: "Internal Server Error" });
+  };
+
+  try {
+    const client = new McpSseClient(
+      testDeps({ timeout: 1 }),
+      "https://example.test/sse",
+      "sk-test",
+    );
+    const started = Date.now();
+    await expect(client.initialize()).rejects.toThrow(/timed out reading error response body/i);
+    expect(Date.now() - started).toBeLessThan(2500);
+    client.close();
+  } finally {
+    globalThis.fetch = originalFetch;
+  }
+});
@@ -0,0 +1,79 @@
+import { expect, test } from "vite-plus/test";
+import { parseSSE } from "../src/client/stream.ts";
+
+async function collectEvents(
+  chunks: string[],
+): Promise<Array<{ data: string; event?: string; id?: string }>> {
+  const encoder = new TextEncoder();
+  const stream = new ReadableStream<Uint8Array>({
+    start(controller) {
+      for (const chunk of chunks) {
+        controller.enqueue(encoder.encode(chunk));
+      }
+      controller.close();
+    },
+  });
+  const response = new Response(stream, {
+    headers: { "Content-Type": "text/event-stream" },
+  });
+  const events: Array<{ data: string; event?: string; id?: string }> = [];
+  for await (const event of parseSSE(response)) {
+    events.push(event);
+  }
+  return events;
+}
+
+test("parseSSE：单 chunk 完整事件保持原行为", async () => {
+  const events = await collectEvents([
+    'event: message\ndata: {"ok":true}\nid: 1\n\ndata: plain\n\n',
+  ]);
+  expect(events).toEqual([{ data: '{"ok":true}', event: "message", id: "1" }, { data: "plain" }]);
+});
+
+test("parseSSE：多行 data 与注释保持原行为", async () => {
+  const events = await collectEvents([": keep-alive\ndata: line1\ndata: line2\n\n"]);
+  expect(events).toEqual([{ data: "line1\nline2" }]);
+});
+
+test("parseSSE：跨 chunk 保留 event 类型", async () => {
+  const events = await collectEvents(["event: endpoint\n", "data: /message?sessionId=abc\n\n"]);
+  expect(events).toEqual([{ data: "/message?sessionId=abc", event: "endpoint" }]);
+});
+
+test("parseSSE：跨 chunk 保留 id，且多事件连续正确", async () => {
+  const events = await collectEvents([
+    "id: a\nevent: message\n",
+    'data: {"n":1}\n\n',
+    "event: message\ndata: ",
+    '{"n":2}\n\n',
+  ]);
+  expect(events).toEqual([
+    { data: '{"n":1}', event: "message", id: "a" },
+    { data: '{"n":2}', event: "message" },
+  ]);
+});
+
+test("parseSSE：CRLF 行尾可解析 endpoint", async () => {
+  const events = await collectEvents(["event: endpoint\r\ndata: /message\r\n\r\n"]);
+  expect(events).toEqual([{ data: "/message", event: "endpoint" }]);
+});
+
+test("parseSSE：纯 CR 行尾可解析 endpoint", async () => {
+  const events = await collectEvents(["event: endpoint\rdata: /message\r\r"]);
+  expect(events).toEqual([{ data: "/message", event: "endpoint" }]);
+});
+
+test("parseSSE：跨 chunk 的 CRLF（\\r|\\n）不丢事件", async () => {
+  const events = await collectEvents(["event: endpoint\r", "\ndata: /message\r\n\r\n"]);
+  expect(events).toEqual([{ data: "/message", event: "endpoint" }]);
+});
+
+test("parseSSE：EOF without blank line keeps event type", async () => {
+  const events = await collectEvents(["event: endpoint\ndata: /message"]);
+  expect(events).toEqual([{ data: "/message", event: "endpoint" }]);
+});
+
+test("parseSSE：EOF data-only flush keeps prior behavior", async () => {
+  const events = await collectEvents(["data: plain"]);
+  expect(events).toEqual([{ data: "plain" }]);
+});
@@ -78,11 +78,12 @@ function fromFetchFailed(err: TypeError): BailianError {
  if (causeMsg && causeMsg !== code) detailParts.push(causeMsg);
  const detail = detailParts.length > 0 ? detailParts.join(": ") : "unknown cause";

+  // Prefer the errno (ENOTFOUND, …) so JSON toJSON() exposes cause.code for agents.
  return new BailianError(
    `Network request failed: ${detail}`,
    ExitCode.NETWORK,
    pickNetworkHint(code),
-    { cause: err },
+    { cause: cause ?? err },
  );
 }

@@ -0,0 +1,85 @@
+import { ExitCode } from "bailian-cli-core";
+import { expect, test } from "vite-plus/test";
+import { handleError } from "../src/error-handler.ts";
+
+test("handleError: fetch failed JSON includes cause.code from errno", () => {
+  const previousOutput = process.env.DASHSCOPE_OUTPUT;
+  process.env.DASHSCOPE_OUTPUT = "json";
+
+  let stderr = "";
+  const originalWrite = process.stderr.write.bind(process.stderr);
+  const originalExit = process.exit;
+  process.stderr.write = ((chunk: string | Uint8Array) => {
+    stderr += String(chunk);
+    return true;
+  }) as typeof process.stderr.write;
+  process.exit = ((code?: number) => {
+    throw new Error(`process.exit:${code ?? 0}`);
+  }) as typeof process.exit;
+
+  const root = Object.assign(new Error("getaddrinfo ENOTFOUND example.invalid"), {
+    code: "ENOTFOUND",
+  });
+  const fetchFailed = new TypeError("fetch failed", { cause: root });
+
+  try {
+    expect(() => handleError(fetchFailed, "bl")).toThrow(
+      new RegExp(`process\\.exit:${ExitCode.NETWORK}`),
+    );
+    const payload = JSON.parse(stderr.trim()) as {
+      error: { code: number; message: string; cause?: { message: string; code?: string } };
+    };
+    expect(payload.error.code).toBe(ExitCode.NETWORK);
+    expect(payload.error.message).toMatch(/ENOTFOUND/);
+    expect(payload.error.cause).toEqual({
+      message: root.message,
+      code: "ENOTFOUND",
+    });
+  } finally {
+    process.stderr.write = originalWrite;
+    process.exit = originalExit;
+    if (previousOutput === undefined) {
+      delete process.env.DASHSCOPE_OUTPUT;
+    } else {
+      process.env.DASHSCOPE_OUTPUT = previousOutput;
+    }
+  }
+});
+
+test("handleError: fetch failed without nested cause still maps to NETWORK", () => {
+  const previousOutput = process.env.DASHSCOPE_OUTPUT;
+  process.env.DASHSCOPE_OUTPUT = "json";
+
+  let stderr = "";
+  const originalWrite = process.stderr.write.bind(process.stderr);
+  const originalExit = process.exit;
+  process.stderr.write = ((chunk: string | Uint8Array) => {
+    stderr += String(chunk);
+    return true;
+  }) as typeof process.stderr.write;
+  process.exit = ((code?: number) => {
+    throw new Error(`process.exit:${code ?? 0}`);
+  }) as typeof process.exit;
+
+  const fetchFailed = new TypeError("fetch failed");
+
+  try {
+    expect(() => handleError(fetchFailed, "bl")).toThrow(
+      new RegExp(`process\\.exit:${ExitCode.NETWORK}`),
+    );
+    const payload = JSON.parse(stderr.trim()) as {
+      error: { code: number; message: string; cause?: { message: string; code?: string } };
+    };
+    expect(payload.error.code).toBe(ExitCode.NETWORK);
+    expect(payload.error.message).toMatch(/unknown cause/);
+    expect(payload.error.cause).toEqual({ message: "fetch failed" });
+  } finally {
+    process.stderr.write = originalWrite;
+    process.exit = originalExit;
+    if (previousOutput === undefined) {
+      delete process.env.DASHSCOPE_OUTPUT;
+    } else {
+      process.env.DASHSCOPE_OUTPUT = previousOutput;
+    }
+  }
+});
@@ -26,15 +26,15 @@ Index: [index.md](index.md)

 #### Flags

-| Flag                     | Type   | Required | Description                                                                              |
-| ------------------------ | ------ | -------- | ---------------------------------------------------------------------------------------- |
-| `--target <server.tool>` | string | yes      | Server code and tool name joined by a dot, e.g. market-cmapi00073529.SmartStockSelection |
-| `--arg <kv>`             | array  | no       | Tool argument (repeatable). Values parsed as JSON if possible, else string.              |
-| `--json <obj>`           | string | no       | Full arguments object as JSON; merged with --arg (arg wins).                             |
-| `--query <text>`         | string | no       | Shortcut for --arg query=<text> (mirrors many DashScope MCP tools).                      |
-| `--url <url>`            | string | no       | Override the MCP endpoint URL (for non-Bailian servers)                                  |
-| `--api-key <key>`        | string | no       | API key                                                                                  |
-| `--base-url <url>`       | string | no       | API base URL                                                                             |
+| Flag                     | Type   | Required | Description                                                                                                 |
+| ------------------------ | ------ | -------- | ----------------------------------------------------------------------------------------------------------- |
+| `--target <server.tool>` | string | yes      | Server code and tool name joined by a dot, e.g. market-cmapi00073529.SmartStockSelection                    |
+| `--arg <kv>`             | array  | no       | Tool argument (repeatable). Values parsed as JSON if possible, else string.                                 |
+| `--json <obj>`           | string | no       | Full arguments object as JSON; merged with --arg (arg wins).                                                |
+| `--query <text>`         | string | no       | Shortcut for --arg query=<text> (mirrors many DashScope MCP tools).                                         |
+| `--url <url>`            | string | no       | Override the MCP endpoint URL (non-Bailian). Tries Streamable HTTP first, then classic SSE on the same URL. |
+| `--api-key <key>`        | string | no       | API key                                                                                                     |
+| `--base-url <url>`       | string | no       | API base URL                                                                                                |

 #### Examples

@@ -97,12 +97,12 @@ bl mcp list --output json

 #### Flags

-| Flag               | Type   | Required | Description                                             |
-| ------------------ | ------ | -------- | ------------------------------------------------------- |
-| `--server <code>`  | string | yes      | Server code from `mcp list` (e.g. market-cmapi00073529) |
-| `--url <url>`      | string | no       | Override the MCP endpoint URL (for non-Bailian servers) |
-| `--api-key <key>`  | string | no       | API key                                                 |
-| `--base-url <url>` | string | no       | API base URL                                            |
+| Flag               | Type   | Required | Description                                                                                                 |
+| ------------------ | ------ | -------- | ----------------------------------------------------------------------------------------------------------- |
+| `--server <code>`  | string | yes      | Server code from `mcp list` (e.g. market-cmapi00073529)                                                     |
+| `--url <url>`      | string | no       | Override the MCP endpoint URL (non-Bailian). Tries Streamable HTTP first, then classic SSE on the same URL. |
+| `--api-key <key>`  | string | no       | API key                                                                                                     |
+| `--base-url <url>` | string | no       | API base URL                                                                                                |

 #### Examples