mirror of
https://github.com/anomalyco/opencode.git
synced 2026-11-13 01:31:22 +03:00
fix(core): make MCP errors self-describing (#52418)
This commit is contained in:
committed by
GitHub
Unverified
parent
a931e8a9fd
commit
2fb7985bb7
@@ -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<string, unknown>
|
||||
readonly sessionID?: Session.ID
|
||||
}) => Effect.Effect<CallToolResult, Error>
|
||||
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.
|
||||
|
||||
@@ -62,6 +62,19 @@ export class ToolCallError extends Schema.TaggedError<ToolCallError>()("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 }
|
||||
|
||||
@@ -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<void> | undefined
|
||||
let closing: Promise<void> | 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
|
||||
|
||||
Reference in New Issue
Block a user