From 313966d7a974557c1d904b18c5ca18fcb561e4a2 Mon Sep 17 00:00:00 2001 From: clh02467605 Date: Wed, 12 Aug 2026 15:44:51 +0800 Subject: [PATCH 1/6] feat(mcp): add SSE support with fallback mechanism for MCP connections - Add McpSseClient implementation for classic HTTP+SSE MCP protocol - Implement connectBailianMcpWithFallback with Streamable HTTP to SSE fallback - Add isStreamableHttpUnsupported helper to detect 405 streamableHttp errors - Update activate-hint logic to handle WebSearch 405 streamableHttp cases - Replace direct MCP client usage with connection manager in call/tools commands - Add proper client cleanup with close() calls in finally blocks - Export new MCP connection utilities and types from core client module - Add comprehensive tests for SSE client and fallback behavior --- .../src/commands/mcp/activate-hint.ts | 18 +- packages/commands/src/commands/mcp/call.ts | 15 +- packages/commands/src/commands/mcp/tools.ts | 15 +- .../commands/tests/mcp-activate-hint.test.ts | 30 ++ packages/core/src/client/client.ts | 27 +- packages/core/src/client/index.ts | 15 +- packages/core/src/client/mcp-sse.ts | 346 ++++++++++++++++++ packages/core/src/client/mcp.ts | 67 ++++ packages/core/tests/mcp.test.ts | 261 +++++++++++++ 9 files changed, 777 insertions(+), 17 deletions(-) create mode 100644 packages/core/src/client/mcp-sse.ts create mode 100644 packages/core/tests/mcp.test.ts diff --git a/packages/commands/src/commands/mcp/activate-hint.ts b/packages/commands/src/commands/mcp/activate-hint.ts index ff7e11c0..5fe4060d 100644 --- a/packages/commands/src/commands/mcp/activate-hint.ts +++ b/packages/commands/src/commands/mcp/activate-hint.ts @@ -1,4 +1,4 @@ -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). */ @@ -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; } diff --git a/packages/commands/src/commands/mcp/call.ts b/packages/commands/src/commands/mcp/call.ts index 3d6ba0b0..9252e9d9 100644 --- a/packages/commands/src/commands/mcp/call.ts +++ b/packages/commands/src/commands/mcp/call.ts @@ -114,14 +114,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 +130,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 +147,8 @@ export default defineCommand({ rethrowWithMcpActivateHint(error, serverCode); } throw error; + } finally { + client?.close?.(); } }, }); diff --git a/packages/commands/src/commands/mcp/tools.ts b/packages/commands/src/commands/mcp/tools.ts index fc38f428..0b1871fe 100644 --- a/packages/commands/src/commands/mcp/tools.ts +++ b/packages/commands/src/commands/mcp/tools.ts @@ -28,24 +28,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?.(); } }, }); diff --git a/packages/commands/tests/mcp-activate-hint.test.ts b/packages/commands/tests/mcp-activate-hint.test.ts index 823a6b83..223c7fbf 100644 --- a/packages/commands/tests/mcp-activate-hint.test.ts +++ b/packages/commands/tests/mcp-activate-hint.test.ts @@ -38,6 +38,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( diff --git a/packages/core/src/client/client.ts b/packages/core/src/client/client.ts index e24558a8..b72ad860 100644 --- a/packages/core/src/client/client.ts +++ b/packages/core/src/client/client.ts @@ -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+streamableHttp (except WebSearch). + * `urlOverride` maps to `--url` and uses Streamable only (no fallback). + */ + 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(api: string, data: Record): Promise { if (!this.deps.consoleCred) { throw new BailianError("This command needs a console access token.", ExitCode.AUTH); diff --git a/packages/core/src/client/index.ts b/packages/core/src/client/index.ts index 31bd04a7..a26958fa 100644 --- a/packages/core/src/client/index.ts +++ b/packages/core/src/client/index.ts @@ -57,7 +57,18 @@ 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, + connectBailianMcpWithFallback, +} from "./mcp.ts"; export type { ServerSentEvent } from "./stream.ts"; export { parseSSE } from "./stream.ts"; diff --git a/packages/core/src/client/mcp-sse.ts b/packages/core/src/client/mcp-sse.ts new file mode 100644 index 00000000..1b8f1557 --- /dev/null +++ b/packages/core/src/client/mcp-sse.ts @@ -0,0 +1,346 @@ +/** + * 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; +}; + +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(); + private endpointReady: Promise; + private resolveEndpoint: (() => void) | undefined; + private rejectEndpoint: ((reason: unknown) => void) | undefined; + private closed = false; + + constructor(deps: HttpDeps, sseUrl: string, authToken?: string) { + this.deps = deps; + this.sseUrl = sseUrl; + this.authToken = authToken; + this.endpointReady = new Promise((resolve, reject) => { + this.resolveEndpoint = resolve; + this.rejectEndpoint = reject; + }); + } + + /** Open the SSE session and run initialize / notifications/initialized. */ + async initialize(): Promise { + 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 { + const result = (await this.rpc("tools/list")) as { tools: McpTool[] }; + return result.tools || []; + } + + async callTool(name: string, args: Record): Promise { + 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(); + for (const [, waiter] of this.pending) { + waiter.reject(new BailianError("MCP SSE session closed.", ExitCode.GENERAL)); + } + this.pending.clear(); + } + + private async openSse(): Promise { + if (this.abortController) return; + + // Keep the GET open until close(); timeouts apply only to endpoint wait / per-RPC. + this.abortController = new AbortController(); + + const headers: Record = { + 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}`); + } + + const response = await fetch(this.sseUrl, { + method: "GET", + headers, + signal: this.abortController.signal, + }); + + if (this.deps.settings.verbose) { + console.error(`< ${response.status} ${response.statusText}`); + } + + if (!response.ok) { + let errMsg = `MCP request failed: ${response.status} ${response.statusText}`; + try { + const errBody = await response.text(); + if (errBody) errMsg += ` - ${errBody.slice(0, 500)}`; + } catch { + /* ignore */ + } + const error = new BailianError(errMsg, ExitCode.GENERAL); + this.rejectEndpoint?.(error); + throw error; + } + + 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); + for (const [, waiter] of this.pending) { + waiter.reject(reason); + } + this.pending.clear(); + }); + + const timeoutMs = this.deps.settings.timeout * 1000; + 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 { + for await (const event of parseSSE(response)) { + if (this.closed) break; + + if (event.event === "endpoint" || (!event.event && !this.messageUrl)) { + 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; + } + + 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 (!waiter) continue; + this.pending.delete(payload.id); + waiter.resolve(payload); + } + } + + if (!this.messageUrl) { + const error = new BailianError( + "MCP SSE stream ended before endpoint event.", + ExitCode.GENERAL, + ); + this.rejectEndpoint?.(error); + throw error; + } + } + + private async rpc(method: string, params?: Record): Promise { + const id = this.nextId++; + const body = { + jsonrpc: "2.0" as const, + id, + method, + ...(params ? { params } : {}), + }; + + const timeoutMs = this.deps.settings.timeout * 1000; + const responsePromise = new Promise((resolve, reject) => { + this.pending.set(id, { resolve, reject }); + }); + const responseTimeout = cancellableTimeoutReject( + timeoutMs, + `MCP SSE timed out waiting for response to ${method}.`, + ); + + try { + await this.postMessage(body); + 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(id); + throw error; + } finally { + responseTimeout.cancel(); + } + } + + private async notify(method: string, params?: Record): Promise { + const body = { + jsonrpc: "2.0" as const, + method, + ...(params ? { params } : {}), + }; + await this.postMessage(body); + } + + private async postMessage(body: unknown): Promise { + if (!this.messageUrl) { + throw new BailianError("MCP SSE message endpoint is not ready.", ExitCode.GENERAL); + } + + const headers: Record = { + "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; + const res = await fetch(this.messageUrl, { + method: "POST", + headers, + body: JSON.stringify(body), + signal: AbortSignal.timeout(timeoutMs), + }); + + if (this.deps.settings.verbose) { + console.error(`< ${res.status} ${res.statusText}`); + } + + if (!res.ok) { + let errMsg = `MCP request failed: ${res.status} ${res.statusText}`; + try { + const errBody = await res.text(); + if (errBody) errMsg += ` - ${errBody.slice(0, 500)}`; + } catch { + /* ignore */ + } + throw new BailianError(errMsg, ExitCode.GENERAL); + } + } +} + +/** 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; cancel: () => void } { + let timer: ReturnType | undefined; + const promise = new Promise((_, 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; + } + }, + }; +} diff --git a/packages/core/src/client/mcp.ts b/packages/core/src/client/mcp.ts index ef2d5d85..98017192 100644 --- a/packages/core/src/client/mcp.ts +++ b/packages/core/src/client/mcp.ts @@ -15,6 +15,7 @@ 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"; // ---- JSON-RPC 2.0 Types ---- @@ -61,6 +62,72 @@ export function bailianMcpPath(serverCode: string): string { return `/api/v1/mcps/${serverCode}/mcp`; } +/** Classic SSE path: `/api/v1/mcps//sse`. */ +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). */ +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); +} + +export type McpConnectedClient = { + initialize(): Promise; + listTools(): Promise; + callTool(name: string, args: Record): Promise; + 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: Streamable only, no SSE fallback. */ + urlOverride?: string; +}; + +/** + * Connect via Streamable HTTP first; on 405+streamableHttp (except WebSearch), fall back to SSE. + * 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 client = new McpClient(deps, urlOverride, authToken); + await client.initialize(); + return { client, url: urlOverride }; + } + + 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 { diff --git a/packages/core/tests/mcp.test.ts b/packages/core/tests/mcp.test.ts new file mode 100644 index 00000000..f594d734 --- /dev/null +++ b/packages/core/tests/mcp.test.ts @@ -0,0 +1,261 @@ +import { expect, test } from "vite-plus/test"; +import type { Identity, Settings } from "../src/index.ts"; +import { + BailianError, + bailianMcpPath, + bailianMcpSsePath, + connectBailianMcpWithFallback, + isStreamableHttpUnsupported, +} from "../src/index.ts"; +import { McpSseClient, resolveSameOriginMessageUrl } from "../src/client/mcp-sse.ts"; + +function testDeps(): { 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, + }, + }; +} + +function jsonRpcResult(id: number, 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(false); + expect(isStreamableHttpUnsupported(new Error("405 streamableHttp"))).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; + } + + // 405 streamableHttp → SSE + let sseController: ReadableStreamDefaultController | 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("current mcp not support streamableHttp", { + status: 405, + statusText: "Method Not Allowed", + }); + } + + if (url.endsWith("/sse") && (init?.method ?? "GET") === "GET") { + const stream = new ReadableStream({ + 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 / 非目标错误不降级", 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); + + urls.length = 0; + 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", + urlOverride: "https://custom.example/mcp", + }), + ).rejects.toBeInstanceOf(BailianError); + expect(urls).toEqual(["https://custom.example/mcp"]); + } 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.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({ + 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; + } +}); From 798ce596f6fdcc82e3dd0d792b7c21f4bc56d6d2 Mon Sep 17 00:00:00 2001 From: clh02467605 Date: Thu, 13 Aug 2026 15:37:32 +0800 Subject: [PATCH 2/6] fix(mcp): harden SSE fallback for Bailian and --url overrides --- packages/commands/src/commands/mcp/call.ts | 3 +- packages/commands/src/commands/mcp/tools.ts | 3 +- packages/core/src/client/client.ts | 4 +- packages/core/src/client/index.ts | 1 + packages/core/src/client/mcp-sse.ts | 62 +++++- packages/core/src/client/mcp.ts | 89 +++++++- packages/core/tests/mcp.test.ts | 218 ++++++++++++++++++-- skills/bailian-cli/reference/mcp.md | 30 +-- 8 files changed, 352 insertions(+), 58 deletions(-) diff --git a/packages/commands/src/commands/mcp/call.ts b/packages/commands/src/commands/mcp/call.ts index 9252e9d9..0c9d0f02 100644 --- a/packages/commands/src/commands/mcp/call.ts +++ b/packages/commands/src/commands/mcp/call.ts @@ -36,7 +36,8 @@ const CALL_FLAGS = { url: { type: "string", valueHint: "", - 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; diff --git a/packages/commands/src/commands/mcp/tools.ts b/packages/commands/src/commands/mcp/tools.ts index 0b1871fe..f60e76f7 100644 --- a/packages/commands/src/commands/mcp/tools.ts +++ b/packages/commands/src/commands/mcp/tools.ts @@ -16,7 +16,8 @@ export default defineCommand({ url: { type: "string", valueHint: "", - 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: [ diff --git a/packages/core/src/client/client.ts b/packages/core/src/client/client.ts index b72ad860..4dfafa38 100644 --- a/packages/core/src/client/client.ts +++ b/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, diff --git a/packages/core/src/client/index.ts b/packages/core/src/client/index.ts index a26958fa..5884d6ab 100644 --- a/packages/core/src/client/index.ts +++ b/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"; diff --git a/packages/core/src/client/mcp-sse.ts b/packages/core/src/client/mcp-sse.ts index 1b8f1557..193a8351 100644 --- a/packages/core/src/client/mcp-sse.ts +++ b/packages/core/src/client/mcp-sse.ts @@ -24,6 +24,11 @@ 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; @@ -31,11 +36,13 @@ export class McpSseClient { private deps: HttpDeps; private authToken: string | undefined; private abortController: AbortController | undefined; - private pending = new Map(); + private pending = new Map(); private endpointReady: Promise; 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 { 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,6 +197,7 @@ export class McpSseClient { continue; } + // 缺省 event 类型在 SSE 中等同 message。 if (event.event === "message" || event.event === undefined) { let payload: JsonRpcResponse; try { @@ -185,14 +205,17 @@ export class McpSseClient { } 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): Promise { + 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,8 +246,10 @@ export class McpSseClient { const timeoutMs = this.deps.settings.timeout * 1000; const responsePromise = new Promise((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}.`, @@ -223,6 +257,9 @@ export class McpSseClient { 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 { + 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); } diff --git a/packages/core/src/client/mcp.ts b/packages/core/src/client/mcp.ts index 98017192..88591c76 100644 --- a/packages/core/src/client/mcp.ts +++ b/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 { + 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 { + 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 { const headers: Record = { "Content-Type": "application/json", diff --git a/packages/core/tests/mcp.test.ts b/packages/core/tests/mcp.test.ts index f594d734..af1e0678 100644 --- a/packages/core/tests/mcp.test.ts +++ b/packages/core/tests/mcp.test.ts @@ -6,10 +6,12 @@ import { bailianMcpSsePath, connectBailianMcpWithFallback, isStreamableHttpUnsupported, + isUrlOverrideSseFallbackCandidate, + McpClient, } from "../src/index.ts"; import { McpSseClient, resolveSameOriginMessageUrl } from "../src/client/mcp-sse.ts"; -function testDeps(): { identity: Identity; settings: Settings } { +function testDeps(overrides?: Partial): { identity: Identity; settings: Settings } { return { identity: { binName: "bl", @@ -25,11 +27,12 @@ function testDeps(): { identity: Identity; settings: Settings } { quiet: true, dryRun: false, telemetry: true, + ...overrides, }, }; } -function jsonRpcResult(id: number, result: unknown): string { +function jsonRpcResult(id: number | string, result: unknown): string { return `event:message\ndata:${JSON.stringify({ jsonrpc: "2.0", id, result })}\n\n`; } @@ -50,10 +53,23 @@ test("bailianMcp 路径与 isStreamableHttpUnsupported", () => { ), ), ).toBe(true); + // 裸 405 也应触发降级，不依赖英文文案 expect( isStreamableHttpUnsupported(new BailianError("MCP request failed: 405 Method Not Allowed")), - ).toBe(false); + ).toBe(true); + expect(isStreamableHttpUnsupported(new BailianError("MCP request failed: 404 Not Found"))).toBe( + false, + ); expect(isStreamableHttpUnsupported(new Error("405 streamableHttp"))).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); }); test("resolveSameOriginMessageUrl：同源通过、跨域拒绝", () => { @@ -102,7 +118,7 @@ test("connectBailianMcpWithFallback：成功走 Streamable；405 降级 SSE", as globalThis.fetch = originalFetch; } - // 405 streamableHttp → SSE + // 裸 405（无 streamableHttp 文案）→ SSE let sseController: ReadableStreamDefaultController | undefined; const encoder = new TextEncoder(); const urls: string[] = []; @@ -112,7 +128,7 @@ test("connectBailianMcpWithFallback：成功走 Streamable；405 降级 SSE", as urls.push(`${init?.method ?? "GET"} ${url}`); if (url.endsWith("/mcp")) { - return new Response("current mcp not support streamableHttp", { + return new Response("Method Not Allowed", { status: 405, statusText: "Method Not Allowed", }); @@ -164,7 +180,7 @@ test("connectBailianMcpWithFallback：成功走 Streamable；405 降级 SSE", as } }); -test("connectBailianMcpWithFallback：WebSearch / urlOverride / 非目标错误不降级", async () => { +test("connectBailianMcpWithFallback：WebSearch 不降级；urlOverride 同 URL 降级 SSE；404 不降级 Bailian 路径", async () => { const originalFetch = globalThis.fetch; const urls: string[] = []; @@ -187,19 +203,66 @@ test("connectBailianMcpWithFallback：WebSearch / urlOverride / 非目标错误 }), ).rejects.toBeInstanceOf(BailianError); expect(urls.some((url) => url.includes("/sse"))).toBe(false); + } finally { + globalThis.fetch = originalFetch; + } - urls.length = 0; - 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", - urlOverride: "https://custom.example/mcp", - }), - ).rejects.toBeInstanceOf(BailianError); - expect(urls).toEqual(["https://custom.example/mcp"]); + // urlOverride：POST 405 后应对同一 URL 发 GET SSE + urls.length = 0; + let sseController: ReadableStreamDefaultController | 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({ + 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; } @@ -222,6 +285,125 @@ test("connectBailianMcpWithFallback：WebSearch / urlOverride / 非目标错误 } }); +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")) { + // 发完 endpoint 后立刻关流 + const stream = new ReadableStream({ + 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 | 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({ + start(controller) { + sseController = controller; + // 无名事件不应被当成 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) { + // 以 string id 回传 + 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; diff --git a/skills/bailian-cli/reference/mcp.md b/skills/bailian-cli/reference/mcp.md index c7b8df69..13f92465 100644 --- a/skills/bailian-cli/reference/mcp.md +++ b/skills/bailian-cli/reference/mcp.md @@ -25,15 +25,15 @@ Index: [index.md](index.md) #### Flags -| Flag | Type | Required | Description | -| ------------------------ | ------ | -------- | ---------------------------------------------------------------------------------------- | -| `--target ` | string | yes | Server code and tool name joined by a dot, e.g. market-cmapi00073529.SmartStockSelection | -| `--arg ` | array | no | Tool argument (repeatable). Values parsed as JSON if possible, else string. | -| `--json ` | string | no | Full arguments object as JSON; merged with --arg (arg wins). | -| `--query ` | string | no | Shortcut for --arg query= (mirrors many DashScope MCP tools). | -| `--url ` | string | no | Override the MCP endpoint URL (for non-Bailian servers) | -| `--api-key ` | string | no | API key | -| `--base-url ` | string | no | API base URL | +| Flag | Type | Required | Description | +| ------------------------ | ------ | -------- | ----------------------------------------------------------------------------------------------------------- | +| `--target ` | string | yes | Server code and tool name joined by a dot, e.g. market-cmapi00073529.SmartStockSelection | +| `--arg ` | array | no | Tool argument (repeatable). Values parsed as JSON if possible, else string. | +| `--json ` | string | no | Full arguments object as JSON; merged with --arg (arg wins). | +| `--query ` | string | no | Shortcut for --arg query= (mirrors many DashScope MCP tools). | +| `--url ` | string | no | Override the MCP endpoint URL (non-Bailian). Tries Streamable HTTP first, then classic SSE on the same URL. | +| `--api-key ` | string | no | API key | +| `--base-url ` | string | no | API base URL | #### Examples @@ -94,12 +94,12 @@ bl mcp list --output json #### Flags -| Flag | Type | Required | Description | -| ------------------ | ------ | -------- | ------------------------------------------------------- | -| `--server ` | string | yes | Server code from `mcp list` (e.g. market-cmapi00073529) | -| `--url ` | string | no | Override the MCP endpoint URL (for non-Bailian servers) | -| `--api-key ` | string | no | API key | -| `--base-url ` | string | no | API base URL | +| Flag | Type | Required | Description | +| ------------------ | ------ | -------- | ----------------------------------------------------------------------------------------------------------- | +| `--server ` | string | yes | Server code from `mcp list` (e.g. market-cmapi00073529) | +| `--url ` | string | no | Override the MCP endpoint URL (non-Bailian). Tries Streamable HTTP first, then classic SSE on the same URL. | +| `--api-key ` | string | no | API key | +| `--base-url ` | string | no | API base URL | #### Examples From 4dcec7d075fa675fd60fe6bbc89728837b5f72ea Mon Sep 17 00:00:00 2001 From: clh02467605 Date: Thu, 13 Aug 2026 18:32:54 +0800 Subject: [PATCH 3/6] fix(mcp): fix SSE header timeout, 405 fallback matching, and parseSSE chunking --- packages/core/src/client/mcp-sse.ts | 58 ++++++++++++++------- packages/core/src/client/mcp.ts | 12 ++--- packages/core/src/client/stream.ts | 4 +- packages/core/tests/mcp.test.ts | 78 ++++++++++++++++++++++++++--- packages/core/tests/stream.test.ts | 54 ++++++++++++++++++++ 5 files changed, 174 insertions(+), 32 deletions(-) create mode 100644 packages/core/tests/stream.test.ts diff --git a/packages/core/src/client/mcp-sse.ts b/packages/core/src/client/mcp-sse.ts index 193a8351..fd62f63a 100644 --- a/packages/core/src/client/mcp-sse.ts +++ b/packages/core/src/client/mcp-sse.ts @@ -24,7 +24,7 @@ type PendingResolver = { reject: (reason: unknown) => void; }; -/** 用字符串键匹配 JSON-RPC id（兼容 number / string 回传）。 */ +/** Match JSON-RPC ids with string keys (number or string echo from server). */ function pendingKey(id: number | string): string { return String(id); } @@ -41,7 +41,7 @@ export class McpSseClient { private resolveEndpoint: (() => void) | undefined; private rejectEndpoint: ((reason: unknown) => void) | undefined; private closed = false; - /** SSE GET 已结束（非主动 close）时置位，后续 RPC 立即失败。 */ + /** Set when the SSE GET ends without an intentional close(); later RPCs fail fast. */ private streamEnded = false; constructor(deps: HttpDeps, sseUrl: string, authToken?: string) { @@ -114,8 +114,15 @@ export class McpSseClient { private async openSse(): Promise { if (this.abortController) return; - // Keep the GET open until close(); timeouts apply only to endpoint wait / per-RPC. + // use shared abortController：header wait use timer abort；after getting header, clearTimeout, + // the long-lived stream is only ended by close()/session abort (compatible with Node 18, no AbortSignal.any). 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 = { Accept: "text/event-stream", @@ -130,11 +137,28 @@ export class McpSseClient { console.error(`> GET ${this.sseUrl}`); } - const response = await fetch(this.sseUrl, { - method: "GET", - headers, - signal: this.abortController.signal, - }); + let response: Response; + try { + response = await fetch(this.sseUrl, { + method: "GET", + headers, + signal: this.abortController.signal, + }); + } catch (error) { + clearTimeout(headerTimer); + 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); + } + throw new BailianError( + `MCP SSE request failed: ${error instanceof Error ? error.message : String(error)}`, + ExitCode.NETWORK, + ); + } + // 已收到响应头：取消 header 等待，后续仅由 abortController 结束流。 + clearTimeout(headerTimer); if (this.deps.settings.verbose) { console.error(`< ${response.status} ${response.statusText}`); @@ -148,9 +172,8 @@ export class McpSseClient { } catch { /* ignore */ } - const error = new BailianError(errMsg, ExitCode.GENERAL); - this.rejectEndpoint?.(error); - throw error; + // Throw only — do not rejectEndpoint; this path never awaits endpointReady. + throw new BailianError(errMsg, ExitCode.GENERAL); } void this.consumeSse(response).catch((error) => { @@ -163,13 +186,12 @@ export class McpSseClient { ExitCode.GENERAL, ); this.rejectEndpoint?.(reason); - // consumeSse 在正常结束路径已 markStreamEnded；此处覆盖解析/读取异常。 + // consumeSse already markStreamEnded on a clean end; cover parse/read failures here. if (!this.streamEnded) { this.markStreamEnded(reason); } }); - const timeoutMs = this.deps.settings.timeout * 1000; const endpointTimeout = cancellableTimeoutReject( timeoutMs, "MCP SSE timed out waiting for endpoint event.", @@ -185,7 +207,7 @@ export class McpSseClient { for await (const event of parseSSE(response)) { if (this.closed) break; - // 规范要求首事件为 event: endpoint；不接受无名事件以免误把 JSON 当 URL。 + // 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; @@ -197,7 +219,7 @@ export class McpSseClient { continue; } - // 缺省 event 类型在 SSE 中等同 message。 + // Omitted SSE event type defaults to "message". if (event.event === "message" || event.event === undefined) { let payload: JsonRpcResponse; try { @@ -225,8 +247,8 @@ export class McpSseClient { throw error; } - // 已拿到 endpoint 后流仍结束：标记会话死亡并唤醒 pending；不再 throw， - // 避免 void consumeSse().catch 之外再冒出未处理 rejection。 + // Stream ended after endpoint: mark session dead and wake pending; do not throw, + // so void consumeSse().catch does not surface an extra unhandled rejection. this.markStreamEnded(new BailianError("MCP SSE stream ended unexpectedly.", ExitCode.GENERAL)); } @@ -248,7 +270,7 @@ export class McpSseClient { const responsePromise = new Promise((resolve, reject) => { this.pending.set(key, { resolve, reject }); }); - // 流可能在 Promise.race 之前结束并 reject pending，先挂上 catch 避免 unhandledRejection。 + // Stream may end and reject pending before Promise.race; attach catch to avoid unhandledRejection. void responsePromise.catch(() => undefined); const responseTimeout = cancellableTimeoutReject( timeoutMs, diff --git a/packages/core/src/client/mcp.ts b/packages/core/src/client/mcp.ts index 88591c76..00228e02 100644 --- a/packages/core/src/client/mcp.ts +++ b/packages/core/src/client/mcp.ts @@ -70,20 +70,20 @@ export function bailianMcpSsePath(serverCode: string): string { /** * True when Streamable HTTP is unsupported and classic SSE fallback should be tried. - * 以 HTTP 405 为准，不依赖服务端英文文案（避免文案变更导致降级失效）。 - * Bailian 的 404（未开通）不在此列，避免误降级。 + * Match HTTP wrapper text `MCP request failed: 405` only — not JSON-RPC `MCP error (405)`. + * Bailian HTTP 404 (not activated) is intentionally excluded. */ export function isStreamableHttpUnsupported(error: unknown): boolean { if (!(error instanceof BailianError)) return false; - return /405\b/i.test(error.message); + return /MCP request failed:\s*405\b/i.test(error.message); } /** - * `--url` 覆盖时的 SSE 降级条件（官方 backwards-compat：同 URL 上 405/404 后尝试 GET SSE）。 + * 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 /405\b/i.test(error.message) || /404\b/i.test(error.message); + return /MCP request failed:\s*(405|404)\b/i.test(error.message); } export type McpConnectedClient = { @@ -242,7 +242,7 @@ export class McpClient { } /** - * 按 Content-Type 读取 JSON-RPC 响应：支持 application/json 与 text/event-stream。 + * Read a JSON-RPC response by Content-Type: application/json or text/event-stream. */ private async readJsonRpcResponse( response: Response, diff --git a/packages/core/src/client/stream.ts b/packages/core/src/client/stream.ts index 6fe6ac29..bca5e09a 100644 --- a/packages/core/src/client/stream.ts +++ b/packages/core/src/client/stream.ts @@ -20,6 +20,8 @@ export async function* parseSSE(response: Response): AsyncGenerator = {}; + while (true) { const { done, value } = await reader.read(); if (done) break; @@ -32,8 +34,6 @@ export async function* parseSSE(response: Response): AsyncGenerator = {}; - for (const line of lines) { if (line === "") { if (event.data !== undefined) { diff --git a/packages/core/tests/mcp.test.ts b/packages/core/tests/mcp.test.ts index af1e0678..d3b41252 100644 --- a/packages/core/tests/mcp.test.ts +++ b/packages/core/tests/mcp.test.ts @@ -53,7 +53,6 @@ test("bailianMcp 路径与 isStreamableHttpUnsupported", () => { ), ), ).toBe(true); - // 裸 405 也应触发降级，不依赖英文文案 expect( isStreamableHttpUnsupported(new BailianError("MCP request failed: 405 Method Not Allowed")), ).toBe(true); @@ -61,6 +60,10 @@ test("bailianMcp 路径与 isStreamableHttpUnsupported", () => { 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, + ); expect( isUrlOverrideSseFallbackCandidate(new BailianError("MCP request failed: 404 Not Found")), @@ -70,6 +73,9 @@ test("bailianMcp 路径与 isStreamableHttpUnsupported", () => { new BailianError("MCP request failed: 405 Method Not Allowed"), ), ).toBe(true); + expect(isUrlOverrideSseFallbackCandidate(new BailianError("MCP error (404): not found"))).toBe( + false, + ); }); test("resolveSameOriginMessageUrl：同源通过、跨域拒绝", () => { @@ -118,7 +124,7 @@ test("connectBailianMcpWithFallback：成功走 Streamable；405 降级 SSE", as globalThis.fetch = originalFetch; } - // 裸 405（无 streamableHttp 文案）→ SSE + // Bare HTTP 405 (no streamableHttp body text) → SSE let sseController: ReadableStreamDefaultController | undefined; const encoder = new TextEncoder(); const urls: string[] = []; @@ -207,7 +213,7 @@ test("connectBailianMcpWithFallback：WebSearch 不降级；urlOverride 同 URL globalThis.fetch = originalFetch; } - // urlOverride：POST 405 后应对同一 URL 发 GET SSE + // urlOverride: after POST 405, fall back with GET SSE on the same URL urls.length = 0; let sseController: ReadableStreamDefaultController | undefined; const encoder = new TextEncoder(); @@ -292,7 +298,7 @@ test("McpSseClient：流结束后立刻失败 pending（不干等到 timeout）" globalThis.fetch = async (input, init) => { const url = requestUrl(input); if ((init?.method ?? "GET") === "GET" || url.endsWith("/sse")) { - // 发完 endpoint 后立刻关流 + // Close the stream immediately after the endpoint event const stream = new ReadableStream({ start(controller) { controller.enqueue( @@ -335,7 +341,7 @@ test("McpSseClient：string JSON-RPC id 可匹配；仅认 event:endpoint", asyn const stream = new ReadableStream({ start(controller) { sseController = controller; - // 无名事件不应被当成 endpoint + // Untyped events must not be treated as endpoint controller.enqueue( encoder.encode(`data:${JSON.stringify({ jsonrpc: "2.0", id: 99, result: {} })}\n\n`), ); @@ -354,7 +360,7 @@ test("McpSseClient：string JSON-RPC id 可匹配；仅认 event:endpoint", asyn const body = typeof init?.body === "string" ? JSON.parse(init.body) : {}; queueMicrotask(() => { if (body.id != null && sseController) { - // 以 string id 回传 + // Echo id as a string sseController.enqueue(encoder.encode(jsonRpcResult(String(body.id), {}))); } }); @@ -441,3 +447,63 @@ test("McpSseClient.close 可中止挂起 GET", async () => { 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; + } +}); diff --git a/packages/core/tests/stream.test.ts b/packages/core/tests/stream.test.ts new file mode 100644 index 00000000..12fe3f8a --- /dev/null +++ b/packages/core/tests/stream.test.ts @@ -0,0 +1,54 @@ +import { expect, test } from "vite-plus/test"; +import { parseSSE } from "../src/client/stream.ts"; + +async function collectEvents( + chunks: string[], +): Promise> { + const encoder = new TextEncoder(); + const stream = new ReadableStream({ + 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" }, + ]); +}); From 3ea2931152bdd1d46296556f2dc4eaa7c3063f52 Mon Sep 17 00:00:00 2001 From: clh02467605 Date: Fri, 14 Aug 2026 11:11:47 +0800 Subject: [PATCH 4/6] fix(mcp): harden SSE parsing, abort, and fallback matching --- .../src/commands/mcp/activate-hint.ts | 2 +- .../commands/tests/mcp-activate-hint.test.ts | 6 + packages/core/src/client/mcp-sse.ts | 78 ++++++-- packages/core/src/client/mcp.ts | 7 +- packages/core/src/client/stream.ts | 139 +++++++++----- packages/core/tests/mcp.test.ts | 177 ++++++++++++++++++ packages/core/tests/stream.test.ts | 25 +++ 7 files changed, 370 insertions(+), 64 deletions(-) diff --git a/packages/commands/src/commands/mcp/activate-hint.ts b/packages/commands/src/commands/mcp/activate-hint.ts index 5fe4060d..0be2e562 100644 --- a/packages/commands/src/commands/mcp/activate-hint.ts +++ b/packages/commands/src/commands/mcp/activate-hint.ts @@ -5,7 +5,7 @@ import { mcpMarketplaceDetailPage } from "bailian-cli-runtime"; 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); } diff --git a/packages/commands/tests/mcp-activate-hint.test.ts b/packages/commands/tests/mcp-activate-hint.test.ts index 223c7fbf..6a1602e6 100644 --- a/packages/commands/tests/mcp-activate-hint.test.ts +++ b/packages/commands/tests/mcp-activate-hint.test.ts @@ -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 广场深链", () => { diff --git a/packages/core/src/client/mcp-sse.ts b/packages/core/src/client/mcp-sse.ts index fd62f63a..97188b5e 100644 --- a/packages/core/src/client/mcp-sse.ts +++ b/packages/core/src/client/mcp-sse.ts @@ -114,8 +114,7 @@ export class McpSseClient { private async openSse(): Promise { if (this.abortController) return; - // use shared abortController：header wait use timer abort；after getting header, clearTimeout, - // the long-lived stream is only ended by close()/session abort (compatible with Node 18, no AbortSignal.any). + // 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; @@ -146,6 +145,8 @@ export class McpSseClient { }); } 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); } @@ -155,27 +156,43 @@ export class McpSseClient { throw new BailianError( `MCP SSE request failed: ${error instanceof Error ? error.message : String(error)}`, ExitCode.NETWORK, + undefined, + { cause: error }, ); } - // 已收到响应头：取消 header 等待，后续仅由 abortController 结束流。 - clearTimeout(headerTimer); 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 { - /* ignore */ + } 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 }); } - // Throw only — do not rejectEndpoint; this path never awaits endpointReady. + 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 = @@ -247,8 +264,7 @@ export class McpSseClient { throw error; } - // Stream ended after endpoint: mark session dead and wake pending; do not throw, - // so void consumeSse().catch does not surface an extra unhandled rejection. + // After endpoint: mark dead and wake pending; don't throw (avoid unhandledRejection). this.markStreamEnded(new BailianError("MCP SSE stream ended unexpectedly.", ExitCode.GENERAL)); } @@ -332,12 +348,24 @@ export class McpSseClient { } const timeoutMs = this.deps.settings.timeout * 1000; - const res = await fetch(this.messageUrl, { - method: "POST", - headers, - body: JSON.stringify(body), - signal: AbortSignal.timeout(timeoutMs), - }); + // Combine per-RPC timeout with session abort so close() cancels in-flight POSTs. + const requestSignal = createLinkedAbortSignal(timeoutMs, this.abortController?.signal); + let res: Response; + 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; + } finally { + requestSignal.cleanup(); + } if (this.deps.settings.verbose) { console.error(`< ${res.status} ${res.statusText}`); @@ -406,3 +434,23 @@ function cancellableTimeoutReject( }, }; } + +/** Timeout + optional parent abort without AbortSignal.any (Node 18). */ +function createLinkedAbortSignal( + timeoutMs: number, + parentSignal?: AbortSignal, +): { signal: AbortSignal; cleanup: () => void } { + const controller = new AbortController(); + const timeout = setTimeout(() => 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 }; +} diff --git a/packages/core/src/client/mcp.ts b/packages/core/src/client/mcp.ts index 00228e02..c3ffe9e4 100644 --- a/packages/core/src/client/mcp.ts +++ b/packages/core/src/client/mcp.ts @@ -70,12 +70,11 @@ export function bailianMcpSsePath(serverCode: string): string { /** * True when Streamable HTTP is unsupported and classic SSE fallback should be tried. - * Match HTTP wrapper text `MCP request failed: 405` only — not JSON-RPC `MCP error (405)`. - * Bailian HTTP 404 (not activated) is intentionally excluded. + * 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); + return /^MCP request failed:\s*405\b/i.test(error.message); } /** @@ -83,7 +82,7 @@ export function isStreamableHttpUnsupported(error: unknown): boolean { */ export function isUrlOverrideSseFallbackCandidate(error: unknown): boolean { if (!(error instanceof BailianError)) return false; - return /MCP request failed:\s*(405|404)\b/i.test(error.message); + return /^MCP request failed:\s*(405|404)\b/i.test(error.message); } export type McpConnectedClient = { diff --git a/packages/core/src/client/stream.ts b/packages/core/src/client/stream.ts index bca5e09a..05e70ca3 100644 --- a/packages/core/src/client/stream.ts +++ b/packages/core/src/client/stream.ts @@ -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, + maxBuffer: number, +): { event: Partial; 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 = { ...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 { const reader = response.body?.getReader(); if (!reader) return; @@ -14,69 +79,55 @@ export async function* parseSSE(response: Response): AsyncGenerator = {}; 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() || ""; + 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(); diff --git a/packages/core/tests/mcp.test.ts b/packages/core/tests/mcp.test.ts index d3b41252..55e10344 100644 --- a/packages/core/tests/mcp.test.ts +++ b/packages/core/tests/mcp.test.ts @@ -64,6 +64,12 @@ test("bailianMcp 路径与 isStreamableHttpUnsupported", () => { 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")), @@ -76,6 +82,11 @@ test("bailianMcp 路径与 isStreamableHttpUnsupported", () => { 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：同源通过、跨域拒绝", () => { @@ -507,3 +518,169 @@ test("McpSseClient：非 2xx 不产生 unhandledRejection", async () => { 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((_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 失败保留 cause", 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"); + await expect(client.initialize()).rejects.toMatchObject({ + message: expect.stringMatching(/MCP SSE request failed:\s*fetch failed/i), + exitCode: 6, + cause: fetchFailed, + }); + client.close(); + } finally { + globalThis.fetch = originalFetch; + } +}); + +test("McpSseClient：fetch 失败后同实例可重新 openSse", async () => { + const originalFetch = globalThis.fetch; + let attempt = 0; + let sseController: ReadableStreamDefaultController | 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({ + 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({ + 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; + } +}); diff --git a/packages/core/tests/stream.test.ts b/packages/core/tests/stream.test.ts index 12fe3f8a..2bf44798 100644 --- a/packages/core/tests/stream.test.ts +++ b/packages/core/tests/stream.test.ts @@ -52,3 +52,28 @@ test("parseSSE：跨 chunk 保留 id，且多事件连续正确", async () => { { 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" }]); +}); From d5d9fcb50f23d845de30f083a4c886949ca418f0 Mon Sep 17 00:00:00 2001 From: clh02467605 Date: Fri, 14 Aug 2026 14:41:37 +0800 Subject: [PATCH 5/6] fix: fixed sse error --- packages/core/src/client/mcp-sse.ts | 84 +++++++++++++++++------------ packages/core/tests/mcp.test.ts | 62 ++++++++++++++++++--- 2 files changed, 107 insertions(+), 39 deletions(-) diff --git a/packages/core/src/client/mcp-sse.ts b/packages/core/src/client/mcp-sse.ts index 97188b5e..a156557a 100644 --- a/packages/core/src/client/mcp-sse.ts +++ b/packages/core/src/client/mcp-sse.ts @@ -153,12 +153,8 @@ export class McpSseClient { if (headerTimedOut) { throw new BailianError("MCP SSE timed out waiting for response headers.", ExitCode.TIMEOUT); } - throw new BailianError( - `MCP SSE request failed: ${error instanceof Error ? error.message : String(error)}`, - ExitCode.NETWORK, - undefined, - { cause: error }, - ); + // Rethrow fetch failures so runtime can surface errno (e.g. ENOTFOUND) in JSON/text. + throw error; } if (this.deps.settings.verbose) { @@ -352,34 +348,46 @@ export class McpSseClient { const requestSignal = createLinkedAbortSignal(timeoutMs, this.abortController?.signal); let res: Response; 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); + 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; } - throw error; - } finally { - requestSignal.cleanup(); - } - if (this.deps.settings.verbose) { - console.error(`< ${res.status} ${res.statusText}`); - } + if (this.deps.settings.verbose) { + console.error(`< ${res.status} ${res.statusText}`); + } - if (!res.ok) { - let errMsg = `MCP request failed: ${res.status} ${res.statusText}`; - try { - const errBody = await res.text(); - if (errBody) errMsg += ` - ${errBody.slice(0, 500)}`; - } catch { - /* ignore */ + 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); } - throw new BailianError(errMsg, ExitCode.GENERAL); + } finally { + requestSignal.cleanup(); } } } @@ -439,9 +447,13 @@ function cancellableTimeoutReject( function createLinkedAbortSignal( timeoutMs: number, parentSignal?: AbortSignal, -): { signal: AbortSignal; cleanup: () => void } { +): { signal: AbortSignal; cleanup: () => void; timedOut: boolean } { const controller = new AbortController(); - const timeout = setTimeout(() => controller.abort(), timeoutMs); + const state = { timedOut: false }; + const timeout = setTimeout(() => { + state.timedOut = true; + controller.abort(); + }, timeoutMs); const abortFromParent = () => controller.abort(parentSignal?.reason); const cleanup = () => { clearTimeout(timeout); @@ -452,5 +464,11 @@ function createLinkedAbortSignal( else parentSignal?.addEventListener("abort", abortFromParent, { once: true }); controller.signal.addEventListener("abort", cleanup, { once: true }); - return { signal: controller.signal, cleanup }; + return { + signal: controller.signal, + cleanup, + get timedOut() { + return state.timedOut; + }, + }; } diff --git a/packages/core/tests/mcp.test.ts b/packages/core/tests/mcp.test.ts index 55e10344..357f4192 100644 --- a/packages/core/tests/mcp.test.ts +++ b/packages/core/tests/mcp.test.ts @@ -563,7 +563,7 @@ test("McpSseClient：非 2xx 读 body 仍受 --timeout 约束", async () => { } }); -test("McpSseClient：fetch 失败保留 cause", async () => { +test("McpSseClient：fetch 失败抛出原始 TypeError（保留 ENOTFOUND）", async () => { const originalFetch = globalThis.fetch; const root = Object.assign(new Error("getaddrinfo ENOTFOUND example.test"), { code: "ENOTFOUND", @@ -576,11 +576,9 @@ test("McpSseClient：fetch 失败保留 cause", async () => { try { const client = new McpSseClient(testDeps(), "https://example.test/sse", "sk-test"); - await expect(client.initialize()).rejects.toMatchObject({ - message: expect.stringMatching(/MCP SSE request failed:\s*fetch failed/i), - exitCode: 6, - cause: fetchFailed, - }); + 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; @@ -684,3 +682,55 @@ test("McpSseClient：close 可中止进行中的 POST", async () => { 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({ + 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({ + 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; + } +}); From 98ba3279fa8abedfa615f2e4b5682dd52d96f598 Mon Sep 17 00:00:00 2001 From: clh02467605 Date: Fri, 14 Aug 2026 15:27:26 +0800 Subject: [PATCH 6/6] fix(runtime): expose errno in fetch-failed JSON cause.code --- packages/runtime/src/error-handler.ts | 3 +- packages/runtime/tests/error-handler.test.ts | 85 +++++++++++++ skills/bailian-cli/reference/index.md | 120 +++++++++---------- skills/bailian-cli/reference/usage.md | 25 ++-- 4 files changed, 160 insertions(+), 73 deletions(-) create mode 100644 packages/runtime/tests/error-handler.test.ts diff --git a/packages/runtime/src/error-handler.ts b/packages/runtime/src/error-handler.ts index 11494ef0..b3d89204 100644 --- a/packages/runtime/src/error-handler.ts +++ b/packages/runtime/src/error-handler.ts @@ -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 }, ); } diff --git a/packages/runtime/tests/error-handler.test.ts b/packages/runtime/tests/error-handler.test.ts new file mode 100644 index 00000000..a1286576 --- /dev/null +++ b/packages/runtime/tests/error-handler.test.ts @@ -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; + } + } +}); diff --git a/skills/bailian-cli/reference/index.md b/skills/bailian-cli/reference/index.md index 2fa87ae9..e9df5cca 100644 --- a/skills/bailian-cli/reference/index.md +++ b/skills/bailian-cli/reference/index.md @@ -9,66 +9,66 @@ Use this index for the skill-scoped quick index and global flags. ## Quick index -| Command | Description | Detail | -| ------------------------------- | ---------------------------------------------------------------------------------------------- | ------------------------------ | -| `bl advisor recommend` | Recommend the best models for your use case (intent analysis → candidate recall → LLM ranking) | [advisor.md](advisor.md) | -| `bl app call` | Call a Bailian application (agent or workflow) | [app.md](app.md) | -| `bl app list` | List Bailian applications | [app.md](app.md) | -| `bl auth generate-access-token` | Generate a CLI access token using OpenAPI AK/SK | [auth.md](auth.md) | -| `bl auth login` | Authenticate with API key, console browser login, or OpenAPI AK/SK (credentials can coexist) | [auth.md](auth.md) | -| `bl auth logout` | Clear stored credentials; full logout also clears the model Base URL | [auth.md](auth.md) | -| `bl auth status` | Show current authentication state | [auth.md](auth.md) | -| `bl config agent` | Configure a coding agent to use DashScope API | [config.md](config.md) | -| `bl config list` | List config profiles and show the active profile | [config.md](config.md) | -| `bl config set` | Set a config value | [config.md](config.md) | -| `bl config show` | Display current configuration | [config.md](config.md) | -| `bl config ui` | Open a local web UI to manage config profiles | [config.md](config.md) | -| `bl config use` | Set the active config profile | [config.md](config.md) | -| `bl console call` | Call a Bailian console API via the CLI gateway | [console.md](console.md) | -| `bl file upload` | Upload a local file to DashScope temporary storage (48h) | [file.md](file.md) | -| `bl knowledge chat` | Chat with a Bailian knowledge base (RAG Q&A with streaming) | [knowledge.md](knowledge.md) | -| `bl knowledge retrieve` | Retrieve from a Bailian knowledge base (deprecated, use `search` instead) | [knowledge.md](knowledge.md) | -| `bl knowledge search` | Search a Bailian knowledge base (RAG semantic retrieval) | [knowledge.md](knowledge.md) | -| `bl mcp call` | Call a tool on an MCP server (tools/call) | [mcp.md](mcp.md) | -| `bl mcp list` | List MCP servers activated under your Bailian account | [mcp.md](mcp.md) | -| `bl mcp tools` | List tools exposed by an MCP server (tools/list) | [mcp.md](mcp.md) | -| `bl memory add` | Add memory from messages or custom content | [memory.md](memory.md) | -| `bl memory delete` | Delete a memory node | [memory.md](memory.md) | -| `bl memory list` | List memory nodes for a user | [memory.md](memory.md) | -| `bl memory profile create` | Create a user profile schema for memory profiling | [memory.md](memory.md) | -| `bl memory profile get` | Get user profile by schema ID and user ID | [memory.md](memory.md) | -| `bl memory search` | Search memory nodes by query or messages | [memory.md](memory.md) | -| `bl memory update` | Update a memory node content | [memory.md](memory.md) | -| `bl model list` | Browse model families or show detailed model info in the Bailian model marketplace | [model.md](model.md) | -| `bl pipeline run` | Run a pipeline workflow definition | [pipeline.md](pipeline.md) | -| `bl pipeline validate` | Validate a pipeline definition without executing | [pipeline.md](pipeline.md) | -| `bl plugin install` | Install or upgrade an allowlisted Command Pack | [plugin.md](plugin.md) | -| `bl plugin link` | Link an allowlisted local Command Pack for development | [plugin.md](plugin.md) | -| `bl plugin list` | List installed Command Packs and their load status | [plugin.md](plugin.md) | -| `bl plugin remove` | Remove an installed Command Pack | [plugin.md](plugin.md) | -| `bl quota check` | Check current usage against rate limits | [quota.md](quota.md) | -| `bl quota history` | View quota change history | [quota.md](quota.md) | -| `bl quota list` | View model RPM/TPM rate limits | [quota.md](quota.md) | -| `bl quota request` | Request a temporary quota increase | [quota.md](quota.md) | -| `bl search web` | Search the web using DashScope MCP WebSearch service | [search.md](search.md) | -| `bl skill add` | Install skills from the Bailian skill registry into local agents | [skill.md](skill.md) | -| `bl skill init` | Install all bailian-\* skills (one-shot bootstrap for new environments) | [skill.md](skill.md) | -| `bl skill list` | List registry skills and diff against local installs | [skill.md](skill.md) | -| `bl skill remove` | Remove locally installed skills (registry is untouched) | [skill.md](skill.md) | -| `bl skill update` | Update installed skills to the latest registry versions | [skill.md](skill.md) | -| `bl text chat` | Send a chat completion (OpenAI compatible, DashScope) | [text.md](text.md) | -| `bl token-plan add-member` | Add a member to a Token Plan organization | [token-plan.md](token-plan.md) | -| `bl token-plan assign-seats` | Batch assign Token Plan seats to members | [token-plan.md](token-plan.md) | -| `bl token-plan create-key` | Create a Token Plan API key for a seat | [token-plan.md](token-plan.md) | -| `bl token-plan list-seats` | List Token Plan subscription seat details | [token-plan.md](token-plan.md) | -| `bl update` | Update the CLI to the latest or a specified version | [update.md](update.md) | -| `bl usage free` | Query free-tier quota for models (all models if --model is omitted) | [usage.md](usage.md) | -| `bl usage freetier` | Enable or disable auto-stop for free-tier models. Enables by default; use --off to disable | [usage.md](usage.md) | -| `bl usage stats` | Query model usage statistics | [usage.md](usage.md) | -| `bl usage summary` | Show a unified usage summary: free-tier quota and recent usage overview | [usage.md](usage.md) | -| `bl usage token-plan` | Show Token Plan quota usage | [usage.md](usage.md) | -| `bl workspace init` | Initialize Bailian workspace and activate postpaid services | [workspace.md](workspace.md) | -| `bl workspace list` | List all workspaces | [workspace.md](workspace.md) | +| Command | Authentication | Description | Detail | +| ------------------------------- | -------------- | ---------------------------------------------------------------------------------------------- | ------------------------------ | +| `bl advisor recommend` | API Key | Recommend the best models for your use case (intent analysis → candidate recall → LLM ranking) | [advisor.md](advisor.md) | +| `bl app call` | API Key | Call a Bailian application (agent or workflow) | [app.md](app.md) | +| `bl app list` | Console | List Bailian applications | [app.md](app.md) | +| `bl auth generate-access-token` | No Auth | Generate a CLI access token using OpenAPI AK/SK | [auth.md](auth.md) | +| `bl auth login` | No Auth | Authenticate with API key, console browser login, or OpenAPI AK/SK (credentials can coexist) | [auth.md](auth.md) | +| `bl auth logout` | No Auth | Clear stored credentials; full logout also clears the model Base URL | [auth.md](auth.md) | +| `bl auth status` | No Auth | Show current authentication state | [auth.md](auth.md) | +| `bl config agent` | No Auth | Configure a coding agent to use DashScope API | [config.md](config.md) | +| `bl config list` | No Auth | List config profiles and show the active profile | [config.md](config.md) | +| `bl config set` | No Auth | Set a config value | [config.md](config.md) | +| `bl config show` | No Auth | Display current configuration | [config.md](config.md) | +| `bl config ui` | No Auth | Open a local web UI to manage config profiles | [config.md](config.md) | +| `bl config use` | No Auth | Set the active config profile | [config.md](config.md) | +| `bl console call` | Console | Call a Bailian console API via the CLI gateway | [console.md](console.md) | +| `bl file upload` | API Key | Upload a local file to DashScope temporary storage (48h) | [file.md](file.md) | +| `bl knowledge chat` | API Key | Chat with a Bailian knowledge base (RAG Q&A with streaming) | [knowledge.md](knowledge.md) | +| `bl knowledge retrieve` | API Key | Retrieve from a Bailian knowledge base (deprecated, use `search` instead) | [knowledge.md](knowledge.md) | +| `bl knowledge search` | API Key | Search a Bailian knowledge base (RAG semantic retrieval) | [knowledge.md](knowledge.md) | +| `bl mcp call` | API Key | Call a tool on an MCP server (tools/call) | [mcp.md](mcp.md) | +| `bl mcp list` | Console | List MCP servers activated under your Bailian account | [mcp.md](mcp.md) | +| `bl mcp tools` | API Key | List tools exposed by an MCP server (tools/list) | [mcp.md](mcp.md) | +| `bl memory add` | API Key | Add memory from messages or custom content | [memory.md](memory.md) | +| `bl memory delete` | API Key | Delete a memory node | [memory.md](memory.md) | +| `bl memory list` | API Key | List memory nodes for a user | [memory.md](memory.md) | +| `bl memory profile create` | API Key | Create a user profile schema for memory profiling | [memory.md](memory.md) | +| `bl memory profile get` | API Key | Get user profile by schema ID and user ID | [memory.md](memory.md) | +| `bl memory search` | API Key | Search memory nodes by query or messages | [memory.md](memory.md) | +| `bl memory update` | API Key | Update a memory node content | [memory.md](memory.md) | +| `bl model list` | Console | Browse model families or show detailed model info in the Bailian model marketplace | [model.md](model.md) | +| `bl pipeline run` | No Auth | Run a pipeline workflow definition | [pipeline.md](pipeline.md) | +| `bl pipeline validate` | No Auth | Validate a pipeline definition without executing | [pipeline.md](pipeline.md) | +| `bl plugin install` | No Auth | Install or upgrade an allowlisted Command Pack | [plugin.md](plugin.md) | +| `bl plugin link` | No Auth | Link an allowlisted local Command Pack for development | [plugin.md](plugin.md) | +| `bl plugin list` | No Auth | List installed Command Packs and their load status | [plugin.md](plugin.md) | +| `bl plugin remove` | No Auth | Remove an installed Command Pack | [plugin.md](plugin.md) | +| `bl quota check` | Console | Check current usage against rate limits | [quota.md](quota.md) | +| `bl quota history` | Console | View quota change history | [quota.md](quota.md) | +| `bl quota list` | Console | View model RPM/TPM rate limits | [quota.md](quota.md) | +| `bl quota request` | Console | Request a temporary quota increase | [quota.md](quota.md) | +| `bl search web` | API Key | Search the web using DashScope MCP WebSearch service | [search.md](search.md) | +| `bl skill add` | No Auth | Install skills from the Bailian skill registry into local agents | [skill.md](skill.md) | +| `bl skill init` | No Auth | Install all bailian-\* skills (one-shot bootstrap for new environments) | [skill.md](skill.md) | +| `bl skill list` | No Auth | List registry skills and diff against local installs | [skill.md](skill.md) | +| `bl skill remove` | No Auth | Remove locally installed skills (registry is untouched) | [skill.md](skill.md) | +| `bl skill update` | No Auth | Update installed skills to the latest registry versions | [skill.md](skill.md) | +| `bl text chat` | API Key | Send a chat completion (OpenAI compatible, DashScope) | [text.md](text.md) | +| `bl token-plan add-member` | AK/SK | Add a member to a Token Plan organization | [token-plan.md](token-plan.md) | +| `bl token-plan assign-seats` | AK/SK | Batch assign Token Plan seats to members | [token-plan.md](token-plan.md) | +| `bl token-plan create-key` | AK/SK | Create a Token Plan API key for a seat | [token-plan.md](token-plan.md) | +| `bl token-plan list-seats` | AK/SK | List Token Plan subscription seat details | [token-plan.md](token-plan.md) | +| `bl update` | No Auth | Update the CLI to the latest or a specified version | [update.md](update.md) | +| `bl usage free` | Console | Query free-tier quota for models (all models if --model is omitted) | [usage.md](usage.md) | +| `bl usage freetier` | Console | Enable or disable auto-stop for free-tier models. Enables by default; use --off to disable | [usage.md](usage.md) | +| `bl usage stats` | Console | Query model usage statistics | [usage.md](usage.md) | +| `bl usage summary` | Console | Show a unified usage summary: free-tier quota and recent usage overview | [usage.md](usage.md) | +| `bl usage token-plan` | Console | Show Token Plan quota usage | [usage.md](usage.md) | +| `bl workspace init` | No Auth | Initialize Bailian workspace and activate postpaid services | [workspace.md](workspace.md) | +| `bl workspace list` | Console | List all workspaces | [workspace.md](workspace.md) | ## By group diff --git a/skills/bailian-cli/reference/usage.md b/skills/bailian-cli/reference/usage.md index 1fed6019..80bf57c1 100644 --- a/skills/bailian-cli/reference/usage.md +++ b/skills/bailian-cli/reference/usage.md @@ -7,13 +7,13 @@ Index: [index.md](index.md) ## Commands in this group -| Command | Description | -| --------------------- | ------------------------------------------------------------------------------------------ | -| `bl usage free` | Query free-tier quota for models (all models if --model is omitted) | -| `bl usage freetier` | Enable or disable auto-stop for free-tier models. Enables by default; use --off to disable | -| `bl usage stats` | Query model usage statistics | -| `bl usage summary` | Show a unified usage summary: free-tier quota and recent usage overview | -| `bl usage token-plan` | Show Token Plan quota usage | +| Command | Authentication | Description | +| --------------------- | -------------- | ------------------------------------------------------------------------------------------ | +| `bl usage free` | Console | Query free-tier quota for models (all models if --model is omitted) | +| `bl usage freetier` | Console | Enable or disable auto-stop for free-tier models. Enables by default; use --off to disable | +| `bl usage stats` | Console | Query model usage statistics | +| `bl usage summary` | Console | Show a unified usage summary: free-tier quota and recent usage overview | +| `bl usage token-plan` | Console | Show Token Plan quota usage | ## Command details @@ -207,11 +207,12 @@ bl usage summary --output json ### `bl usage token-plan` -| Field | Value | -| --------------- | ----------------------------- | -| **Name** | `usage token-plan` | -| **Description** | Show Token Plan quota usage | -| **Usage** | `bl usage token-plan [flags]` | +| Field | Value | +| ------------------ | ----------------------------- | +| **Name** | `usage token-plan` | +| **Description** | Show Token Plan quota usage | +| **Authentication** | Console | +| **Usage** | `bl usage token-plan [flags]` | #### Flags