From 2fb7985bb7ea428ebe0e75d27d6c6abd9724bd00 Mon Sep 17 00:00:00 2001 From: Aiden Cline <63023139+rekram1-node@users.noreply.github.com> Date: Thu, 1 Oct 2026 01:46:55 -0500 Subject: [PATCH] fix(core): make MCP errors self-describing (#52418) --- packages/core/src/mcp/client.ts | 40 +++++++++++++++++++++++---- packages/core/src/mcp/index.ts | 26 ++++++++++++++--- packages/core/src/mcp/stdio.ts | 49 +++++++++++++++++++++++---------- 3 files changed, 91 insertions(+), 24 deletions(-) diff --git a/packages/core/src/mcp/client.ts b/packages/core/src/mcp/client.ts index a1d3826287b..bd294bbe171 100644 --- a/packages/core/src/mcp/client.ts +++ b/packages/core/src/mcp/client.ts @@ -4,6 +4,8 @@ import path from "node:path" import { pathToFileURL } from "node:url" import { Client, + SdkError, + SdkErrorCode, SdkHttpError, StreamableHTTPClientTransport, UnauthorizedError, @@ -35,6 +37,18 @@ const DEFAULT_EXECUTION_TIMEOUT = 12 * 60 * 60 * 1_000 // 12 hours const TERMINATE_TIMEOUT = 1_000 const toError = (error: unknown) => (error instanceof Error ? error : new Error(String(error))) +// HTTP statuses and network error codes live on error properties and are lost once flattened to a message. +const describe = (error: unknown) => { + if (!(error instanceof Error)) return String(error) + const detail = + error instanceof SdkHttpError + ? `HTTP ${error.status}` + : !(error instanceof SdkError) && "code" in error && typeof error.code === "string" + ? error.code + : undefined + return detail && !error.message.includes(detail) ? `${error.message} (${detail})` : error.message +} + export type { GetPromptResult, Prompt, ReadResourceResult, Resource, Tool } export type ResourceTemplate = ResourceTemplateType @@ -108,7 +122,7 @@ export interface Connection { readonly args?: Record readonly sessionID?: Session.ID }) => Effect.Effect - readonly onClose: (callback: () => void) => void + readonly onClose: (callback: (reason: string) => void) => void readonly onSessionExpired: (callback: () => void) => void readonly onToolsChanged: (callback: () => void) => void readonly onPromptsChanged: (callback: () => void) => void @@ -138,7 +152,15 @@ export const connect = Effect.fnUntraced(function* ( onChanged: () => changed[key](), }) + // The SDK fails pending requests with a bare "Connection closed"; the transport error before it says why. + let lastError: string | undefined + const explain = (error: unknown) => + error instanceof SdkError && error.code === SdkErrorCode.ConnectionClosed && lastError + ? `${describe(error)}: ${lastError}` + : describe(error) + const initialize = Effect.fnUntraced(function* (transport: Transport) { + const runFork = Effect.runForkWith(yield* Effect.context()) const client = new Client(clientInfo, { capabilities: { ...(elicitation ? { elicitation: { form: { applyDefaults: true }, url: {} } } : {}), @@ -153,6 +175,12 @@ export const connect = Effect.fnUntraced(function* ( resources: listChanged("resources"), }, }) + // Background work such as the standalone SSE stream reports failures only here; aborts come from close. + client.onerror = (error) => { + if (error.name === "AbortError") return + lastError = describe(error) + runFork(Effect.logWarning("mcp transport error", { server, error: lastError })) + } client.setRequestHandler("roots/list", () => ({ roots: [{ uri: pathToFileURL(directory).href }] })) if (elicitation) { client.setRequestHandler("elicitation/create", (request, ctx) => @@ -179,13 +207,13 @@ export const connect = Effect.fnUntraced(function* ( reported: false, } const failure = (error: unknown) => { - if (!(error instanceof SdkHttpError) || session.transport?.sessionId === undefined) return toError(error) + if (!(error instanceof SdkHttpError) || session.transport?.sessionId === undefined) return new Error(explain(error)) const expired = error.status === 404 || (error.status === 400 && typeof error.data.text === "string" && error.data.text.includes("Bad Request: Server not initialized")) - if (!expired) return toError(error) + if (!expired) return new Error(explain(error)) if (!session.reported) { session.reported = true session.expired?.() @@ -242,6 +270,8 @@ export const connect = Effect.fnUntraced(function* ( // Close only aborts streams; the legacy session lives on until the server expires it unless // terminated explicitly. Terminate first: close aborts the signal the DELETE shares. const transport = session.transport + // Termination failures are logged below; onerror would report them a second time. + client.onerror = undefined if (transport?.sessionId !== undefined && !session.reported) yield* Effect.tryPromise({ try: () => transport.terminateSession(), catch: toError }).pipe( Effect.timeoutOrElse({ @@ -301,7 +331,7 @@ export const connect = Effect.fnUntraced(function* ( ), ).pipe(Effect.map(toCallToolResult)), onClose: (callback) => { - client.onclose = callback + client.onclose = () => callback(lastError ? `Connection closed: ${lastError}` : "Connection closed") }, onSessionExpired: (callback) => { session.expired = callback @@ -325,7 +355,7 @@ export const connect = Effect.fnUntraced(function* ( server, message: `${error.message}; the server supports ${error.supported.join(", ")}. Set "protocol" for this server to one of those or to "legacy".`, }) - return yield* new ConnectError({ server, message: error instanceof Error ? error.message : String(error) }) + return yield* new ConnectError({ server, message: explain(error) }) }) // Absent config is legacy: the SDK sends the plain initialize handshake with no discover probe. diff --git a/packages/core/src/mcp/index.ts b/packages/core/src/mcp/index.ts index f7e69bc0054..0c021a4c95e 100644 --- a/packages/core/src/mcp/index.ts +++ b/packages/core/src/mcp/index.ts @@ -62,6 +62,19 @@ export class ToolCallError extends Schema.TaggedError()("MCP.Tool message: Schema.String, }) {} +const unavailable = (server: ServerName, status: Status) => { + switch (status.status) { + case "failed": + return `MCP server "${server}" is not connected: ${status.error}. Reconnect it from /mcps.` + case "needs_auth": + return `MCP server "${server}" needs authentication: ${status.error}. Sign in from /mcps.` + case "disabled": + return `MCP server "${server}" is disabled.` + default: + return `MCP server "${server}" is not connected.` + } +} + type ServerEntry = { readonly config: Mcp.ServerConfig status: Status @@ -388,10 +401,10 @@ export const layer = (options?: Options) => const watch = (name: ServerName, entry: ServerEntry, connection: McpClient.Connection) => { const live = whenLive(name, entry, connection) - connection.onClose(() => + connection.onClose((reason) => live( Effect.gen(function* () { - entry.status = { status: "failed", error: "Connection closed" } + entry.status = { status: "failed", error: reason } yield* stopServer(name, entry) yield* bus.publish(McpEvent.StatusChanged, { server: name }) }), @@ -660,13 +673,18 @@ export const layer = (options?: Options) => return yield* new ToolCallError({ server: target.name, tool: input.name, - message: "MCP server is not connected", + message: unavailable(target.name, target.entry.status), }) const result = yield* recovering(target.name, target.entry, target.entry.client, (connection) => connection.callTool({ name: input.name, args: input.args, sessionID: input.sessionID }), ).pipe( Effect.mapError( - (error) => new ToolCallError({ server: target.name, tool: input.name, message: error.message }), + (error) => + new ToolCallError({ + server: target.name, + tool: input.name, + message: `MCP tool "${input.name}" on server "${target.name}" failed: ${error.message}`, + }), ), ) return { ...result, server: target.name, tool: input.name } diff --git a/packages/core/src/mcp/stdio.ts b/packages/core/src/mcp/stdio.ts index 0f07e9129e9..79faf9536e7 100644 --- a/packages/core/src/mcp/stdio.ts +++ b/packages/core/src/mcp/stdio.ts @@ -1,7 +1,7 @@ export * as McpStdio from "./stdio.js" import { ReadBuffer, serializeMessage, type JSONRPCMessage, type Transport } from "@modelcontextprotocol/client" -import { Cause, Duration, Effect, Queue, Scope, Stream } from "effect" +import { Cause, Duration, Effect, Fiber, Option, Queue, Scope, Stream } from "effect" import { ChildProcess } from "effect/unstable/process" import type { ChildProcessHandle } from "effect/unstable/process/ChildProcessSpawner" import { Environment } from "../environment/index.js" @@ -13,6 +13,9 @@ const CLOSE_GRACE = Duration.seconds(2) const FORCE_KILL_AFTER = Duration.seconds(2) const OUTGOING_CAPACITY = 64 const MAX_FRAME_BYTES = 16 * 1024 * 1024 +// An unexpected exit is reported with its exit code and the end of stderr, which usually names the cause. +const EXIT_WAIT = Duration.seconds(1) +const STDERR_TAIL = 1_000 export interface Options { /** Server name; only used to attribute logs. */ @@ -50,6 +53,7 @@ export const make = Effect.fnUntraced(function* (options: Options) { let startup: Promise | undefined let closing: Promise | undefined let trailingBytes = 0 + let stderr = "" const stop = Effect.fnUntraced(function* (handle: ChildProcessHandle) { // Exit completion can precede descendant cleanup after the capture deadline. @@ -139,6 +143,20 @@ export const make = Effect.fnUntraced(function* (options: Options) { const startOutput = (handle: ChildProcessHandle) => Effect.gen(function* () { + // Drain stderr into the debug log so chatty servers cannot stall on a full pipe. + const drain = yield* Effect.forkScoped( + handle.stderr.pipe( + Stream.decodeText(), + Stream.tap((output) => Effect.sync(() => (stderr = (stderr + output).slice(-STDERR_TAIL)))), + Stream.runForEach((output) => + output.trim() === "" + ? Effect.void + : Effect.logDebug("mcp server stderr", { server: options.server, output }), + ), + Effect.ignore, + ), + ) + yield* Effect.forkScoped( Stream.runForEach(handle.stdout, deliver).pipe( Effect.tapCause((cause) => @@ -152,25 +170,26 @@ export const make = Effect.fnUntraced(function* (options: Options) { Effect.ensuring( Effect.gen(function* () { const unexpected = state.phase !== "closed" - if (unexpected) yield* Effect.promise(close) + if (unexpected) { + // stdout can end before the final stderr output has been read. + const code = yield* Fiber.await(drain).pipe( + Effect.andThen(handle.exitCode), + Effect.timeoutOption(EXIT_WAIT), + Effect.orElseSucceed(() => Option.none()), + ) + const exited = Option.match(code, { + onNone: () => "MCP server process exited", + onSome: (code) => `MCP server process exited with code ${code}`, + }) + const output = stderr.trim() + transport.onerror?.(new Error(output ? `${exited}: ${output}` : exited)) + yield* Effect.promise(close) + } transport.onclose?.() }), ), ), ) - - // Drain stderr into the debug log so chatty servers cannot stall on a full pipe. - yield* Effect.forkScoped( - handle.stderr.pipe( - Stream.decodeText(), - Stream.runForEach((output) => - output.trim() === "" - ? Effect.void - : Effect.logDebug("mcp server stderr", { server: options.server, output }), - ), - Effect.ignore, - ), - ) }) return transport