diff --git a/packages/commands/src/commands/mcp/activate-hint.ts b/packages/commands/src/commands/mcp/activate-hint.ts index ff7e11c..5fe4060 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 3d6ba0b..9252e9d 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 fc38f42..0b1871f 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 823a6b8..223c7fb 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 e24558a..b72ad86 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 31bd04a..a26958f 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 0000000..1b8f155 --- /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 ef2d5d8..9801719 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 0000000..f594d73 --- /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; + } +});