diff --git a/packages/core/src/client/mcp-sse.ts b/packages/core/src/client/mcp-sse.ts index 193a835..fd62f63 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 88591c7..00228e0 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 6fe6ac2..bca5e09 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 af1e067..d3b4125 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 0000000..12fe3f8 --- /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" }, + ]); +});