From 6a6d20515a3ded18e9be8a81ef8fdf92c9d22ca6 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Tue, 5 May 2026 16:07:11 -0700 Subject: [PATCH] Clean up MCP Effect error handling --- packages/plugins/mcp/src/api/handlers.ts | 144 +++++++++---- .../plugins/mcp/src/react/EditMcpSource.tsx | 35 ++-- packages/plugins/mcp/src/sdk/discover.ts | 24 +-- packages/plugins/mcp/src/sdk/invoke.ts | 34 ++-- packages/plugins/mcp/src/sdk/manifest.ts | 32 +-- .../src/sdk/per-user-auth-isolation.test.ts | 10 +- packages/plugins/mcp/src/sdk/plugin.ts | 192 +++++++++++------- 7 files changed, 289 insertions(+), 182 deletions(-) diff --git a/packages/plugins/mcp/src/api/handlers.ts b/packages/plugins/mcp/src/api/handlers.ts index 6fd0ecdca..4769514c1 100644 --- a/packages/plugins/mcp/src/api/handlers.ts +++ b/packages/plugins/mcp/src/api/handlers.ts @@ -3,12 +3,15 @@ import { Context, Effect } from "effect"; import { addGroup, capture } from "@executor-js/api"; import type { + McpConnectionAccessFailure, + McpExtensionFailure, McpPluginExtension, McpProbeEndpointInput, McpSourceConfig, McpUpdateSourceInput, } from "../sdk/plugin"; import type { SecretBackedValue } from "../sdk/types"; +import { McpConnectionError } from "../sdk/errors"; import { McpStoredSourceSchema } from "../sdk/stored-source"; import { McpGroup } from "./group"; @@ -29,6 +32,53 @@ export class McpExtensionService extends Context.Service( + effect: Effect.Effect, +): Effect.Effect< + A, + Exclude | McpConnectionError, + R +> => + effect.pipe( + Effect.catchTags({ + ConnectionNotFoundError: (err) => + Effect.fail( + new McpConnectionError({ + transport: "remote", + message: `OAuth connection "${err.connectionId}" was not found`, + }), + ), + ConnectionProviderNotRegisteredError: (err) => + Effect.fail( + new McpConnectionError({ + transport: "remote", + message: `OAuth provider "${err.provider}" is not registered`, + }), + ), + ConnectionRefreshNotSupportedError: (err) => + Effect.fail( + new McpConnectionError({ + transport: "remote", + message: `OAuth provider "${err.provider}" does not support refresh`, + }), + ), + ConnectionReauthRequiredError: (err) => + Effect.fail( + new McpConnectionError({ + transport: "remote", + message: `OAuth connection "${err.connectionId}" requires reauthentication`, + }), + ), + ConnectionRefreshError: (err) => + Effect.fail( + new McpConnectionError({ + transport: "remote", + message: `Failed to refresh OAuth connection "${err.connectionId}"`, + }), + ), + }), + ); + // --------------------------------------------------------------------------- // Convert API payload → McpSourceConfig // --------------------------------------------------------------------------- @@ -100,67 +150,79 @@ export const McpHandlers = HttpApiBuilder.group(ExecutorApiWithMcp, "mcp", (hand handlers .handle("probeEndpoint", ({ payload }) => capture( - Effect.gen(function* () { - const ext = yield* McpExtensionService; - return yield* ext.probeEndpoint(payload as McpProbeEndpointInput); - }), + connectionAccessToMcpError( + Effect.gen(function* () { + const ext = yield* McpExtensionService; + return yield* ext.probeEndpoint(payload as McpProbeEndpointInput); + }), + ), ), ) .handle("addSource", ({ params: path, payload }) => capture( - Effect.gen(function* () { - const ext = yield* McpExtensionService; - return yield* ext.addSource( - toSourceConfig(payload as Parameters[0], path.scopeId), - ); - }), + connectionAccessToMcpError( + Effect.gen(function* () { + const ext = yield* McpExtensionService; + return yield* ext.addSource( + toSourceConfig(payload as Parameters[0], path.scopeId), + ); + }), + ), ), ) .handle("removeSource", ({ params: path, payload }) => capture( - Effect.gen(function* () { - const ext = yield* McpExtensionService; - yield* ext.removeSource(payload.namespace, path.scopeId); - return { removed: true }; - }), + connectionAccessToMcpError( + Effect.gen(function* () { + const ext = yield* McpExtensionService; + yield* ext.removeSource(payload.namespace, path.scopeId); + return { removed: true }; + }), + ), ), ) .handle("refreshSource", ({ params: path, payload }) => capture( - Effect.gen(function* () { - const ext = yield* McpExtensionService; - return yield* ext.refreshSource(payload.namespace, path.scopeId); - }), + connectionAccessToMcpError( + Effect.gen(function* () { + const ext = yield* McpExtensionService; + return yield* ext.refreshSource(payload.namespace, path.scopeId); + }), + ), ), ) .handle("getSource", ({ params: path }) => capture( - Effect.gen(function* () { - const ext = yield* McpExtensionService; - const source = yield* ext.getSource(path.namespace, path.scopeId); - return source - ? new McpStoredSourceSchema({ - namespace: source.namespace, - name: source.name, - config: source.config, - }) - : null; - }), + connectionAccessToMcpError( + Effect.gen(function* () { + const ext = yield* McpExtensionService; + const source = yield* ext.getSource(path.namespace, path.scopeId); + return source + ? new McpStoredSourceSchema({ + namespace: source.namespace, + name: source.name, + config: source.config, + }) + : null; + }), + ), ), ) .handle("updateSource", ({ params: path, payload }) => capture( - Effect.gen(function* () { - const ext = yield* McpExtensionService; - yield* ext.updateSource(path.namespace, path.scopeId, { - name: payload.name, - endpoint: payload.endpoint, - headers: payload.headers, - queryParams: payload.queryParams, - auth: payload.auth as McpUpdateSourceInput["auth"], - }); - return { updated: true }; - }), + connectionAccessToMcpError( + Effect.gen(function* () { + const ext = yield* McpExtensionService; + yield* ext.updateSource(path.namespace, path.scopeId, { + name: payload.name, + endpoint: payload.endpoint, + headers: payload.headers, + queryParams: payload.queryParams, + auth: payload.auth as McpUpdateSourceInput["auth"], + }); + return { updated: true }; + }), + ), ), ), ); diff --git a/packages/plugins/mcp/src/react/EditMcpSource.tsx b/packages/plugins/mcp/src/react/EditMcpSource.tsx index 4ab5e6d07..5a9de87f0 100644 --- a/packages/plugins/mcp/src/react/EditMcpSource.tsx +++ b/packages/plugins/mcp/src/react/EditMcpSource.tsx @@ -1,5 +1,6 @@ import { useState } from "react"; import { useAtomValue, useAtomSet } from "@effect/atom-react"; +import { Exit } from "effect"; import * as AsyncResult from "effect/unstable/reactivity/AsyncResult"; import { mcpSourceAtom, updateMcpSource } from "./atoms"; import { useScope } from "@executor-js/react/api/scope-context"; @@ -35,7 +36,7 @@ function RemoteEditForm(props: { onSave: () => void; }) { const scopeId = useScope(); - const doUpdate = useAtomSet(updateMcpSource, { mode: "promise" }); + const doUpdate = useAtomSet(updateMcpSource, { mode: "promiseExit" }); const secretList = useSecretPickerSecrets(); const identity = useSourceIdentity({ @@ -64,24 +65,24 @@ function RemoteEditForm(props: { setSaving(true); setError(null); const { headers, queryParams } = serializeHttpCredentials(credentials); - try { - await doUpdate({ - params: { scopeId, namespace: props.sourceId }, - payload: { - name: identity.name.trim() || undefined, - endpoint: endpoint.trim() || undefined, - headers, - queryParams, - }, - reactivityKeys: sourceWriteKeys, - }); - setDirty(false); - props.onSave(); - } catch (e) { - setError(e instanceof Error ? e.message : "Failed to update source"); - } finally { + const exit = await doUpdate({ + params: { scopeId, namespace: props.sourceId }, + payload: { + name: identity.name.trim() || undefined, + endpoint: endpoint.trim() || undefined, + headers, + queryParams, + }, + reactivityKeys: sourceWriteKeys, + }); + if (Exit.isFailure(exit)) { + setError("Failed to update source"); setSaving(false); + return; } + setDirty(false); + props.onSave(); + setSaving(false); }; return ( diff --git a/packages/plugins/mcp/src/sdk/discover.ts b/packages/plugins/mcp/src/sdk/discover.ts index 58f6d7278..ed6c711f3 100644 --- a/packages/plugins/mcp/src/sdk/discover.ts +++ b/packages/plugins/mcp/src/sdk/discover.ts @@ -27,10 +27,10 @@ export const discoverTools = ( // Acquire connection const connection = yield* connector.pipe( Effect.mapError( - (err) => + () => new McpToolDiscoveryError({ stage: "connect", - message: `Failed connecting to MCP server: ${err.message}`, + message: "Failed connecting to MCP server", }), ), ); @@ -38,23 +38,19 @@ export const discoverTools = ( // List tools const listResult = yield* Effect.tryPromise({ try: () => connection.client.listTools(), - catch: (cause) => + catch: () => new McpToolDiscoveryError({ stage: "list_tools", - message: `Failed listing MCP tools: ${ - cause instanceof Error ? cause.message : String(cause) - }`, + message: "Failed listing MCP tools", }), }); if (!isListToolsResult(listResult)) { - yield* Effect.promise(() => connection.close().catch(() => {})); - return yield* Effect.fail( - new McpToolDiscoveryError({ - stage: "list_tools", - message: "MCP listTools response did not match the expected schema", - }), - ); + yield* Effect.ignore(Effect.tryPromise(() => connection.close())); + return yield* new McpToolDiscoveryError({ + stage: "list_tools", + message: "MCP listTools response did not match the expected schema", + }); } const manifest = extractManifestFromListToolsResult(listResult, { @@ -62,7 +58,7 @@ export const discoverTools = ( }); // Close the connection after discovery - yield* Effect.promise(() => connection.close().catch(() => {})); + yield* Effect.ignore(Effect.tryPromise(() => connection.close())); return manifest; }); diff --git a/packages/plugins/mcp/src/sdk/invoke.ts b/packages/plugins/mcp/src/sdk/invoke.ts index 6cd7e9dbd..4cce2a7c9 100644 --- a/packages/plugins/mcp/src/sdk/invoke.ts +++ b/packages/plugins/mcp/src/sdk/invoke.ts @@ -10,13 +10,14 @@ // 4. Retrying once on connection failure (invalidate + reconnect). // --------------------------------------------------------------------------- -import { Cause, Effect, Exit, Schema, ScopedCache } from "effect"; +import { Cause, Effect, Exit, Predicate, Schema, ScopedCache } from "effect"; import { ElicitRequestSchema } from "@modelcontextprotocol/sdk/types.js"; import { FormElicitation, UrlElicitation, + type ElicitationDeclinedError, type Elicit, type ElicitationRequest, } from "@executor-js/sdk/core"; @@ -108,14 +109,13 @@ const installElicitationHandler = ( } const failure = exit.cause.reasons.find(Cause.isFailReason); if (failure) { - const err = failure.error as { - readonly _tag?: string; - readonly action?: "decline" | "cancel"; - }; - if (err._tag === "ElicitationDeclinedError") { - return { action: err.action ?? "decline" }; + const err = failure.error; + if (Predicate.isTagged(err, "ElicitationDeclinedError")) { + const action = (err as ElicitationDeclinedError).action; + return { action }; } } + // oxlint-disable-next-line executor/no-try-catch-or-throw -- boundary: MCP SDK request handler must throw JSON-RPC failures throw Cause.squash(exit.cause); }, ); @@ -135,12 +135,10 @@ const useConnection = ( installElicitationHandler(connection.client, elicit); return yield* Effect.tryPromise({ try: () => connection.client.callTool({ name: toolName, arguments: args }), - catch: (cause) => + catch: () => new McpInvocationError({ toolName, - message: `MCP tool call failed for ${toolName}: ${ - cause instanceof Error ? cause.message : String(cause) - }`, + message: `MCP tool call failed for ${toolName}`, }), }).pipe( Effect.withSpan("plugin.mcp.client.call_tool", { @@ -153,7 +151,7 @@ const useConnection = ( // Public API // --------------------------------------------------------------------------- -export interface InvokeMcpToolInput { +export interface InvokeMcpToolInput { readonly toolId: string; readonly toolName: string; readonly args: unknown; @@ -162,22 +160,22 @@ export interface InvokeMcpToolInput { * connection cache key so per-user OAuth/secret resolution doesn't * collapse multiple users onto one shared connection. */ readonly invokerScope: string; - readonly resolveConnector: () => Effect.Effect; + readonly resolveConnector: () => Effect.Effect; readonly connectionCache: ScopedCache.ScopedCache< string, McpConnection, - McpConnectionError + E >; readonly pendingConnectors: Map< string, - Effect.Effect + Effect.Effect >; readonly elicit: Elicit; } -export const invokeMcpTool = ( - input: InvokeMcpToolInput, -): Effect.Effect => { +export const invokeMcpTool = ( + input: InvokeMcpToolInput, +): Effect.Effect => { const transport: string = input.sourceData.transport === "stdio" ? "stdio" diff --git a/packages/plugins/mcp/src/sdk/manifest.ts b/packages/plugins/mcp/src/sdk/manifest.ts index 6b92f1b8a..aee559f2c 100644 --- a/packages/plugins/mcp/src/sdk/manifest.ts +++ b/packages/plugins/mcp/src/sdk/manifest.ts @@ -1,4 +1,4 @@ -import { Schema } from "effect"; +import { Option, Schema } from "effect"; import { McpToolAnnotations } from "./types"; @@ -51,7 +51,7 @@ const decodeListToolsResult = Schema.decodeUnknownOption(ListToolsResult); const decodeServerInfo = Schema.decodeUnknownOption(ServerInfo); export const isListToolsResult = (value: unknown): boolean => - decodeListToolsResult(value)._tag === "Some"; + Option.isSome(decodeListToolsResult(value)); // --------------------------------------------------------------------------- // Tool ID sanitization @@ -86,14 +86,21 @@ export const extractManifestFromListToolsResult = ( ): McpToolManifest => { const seen = new Map(); - const listed = decodeListToolsResult(listToolsResult).pipe((opt) => - opt._tag === "Some" ? opt.value.tools : [], + const listed = decodeListToolsResult(listToolsResult).pipe( + Option.match({ + onNone: () => [], + onSome: (result) => result.tools, + }), ); - const server = decodeServerInfo(metadata?.serverInfo).pipe((opt): McpServerMetadata | null => - opt._tag === "Some" - ? { name: opt.value.name ?? null, version: opt.value.version ?? null } - : null, + const server = decodeServerInfo(metadata?.serverInfo).pipe( + Option.match({ + onNone: (): McpServerMetadata | null => null, + onSome: (info): McpServerMetadata => ({ + name: info.name ?? null, + version: info.version ?? null, + }), + }), ); const tools = listed.flatMap((tool): McpToolManifestEntry[] => { @@ -125,13 +132,8 @@ const slugify = (value: string): string => .replace(/[^a-z0-9]+/g, "_") .replace(/^_+|_+$/g, ""); -const hostnameOf = (url: string): string | null => { - try { - return new URL(url).hostname; - } catch { - return null; - } -}; +const hostnameOf = (url: string): string | null => + URL.canParse(url) ? new URL(url).hostname : null; const basenameOf = (path: string): string => path.trim().split(/[\\/]/).pop() ?? path.trim(); diff --git a/packages/plugins/mcp/src/sdk/per-user-auth-isolation.test.ts b/packages/plugins/mcp/src/sdk/per-user-auth-isolation.test.ts index cf655a4d1..ea6410167 100644 --- a/packages/plugins/mcp/src/sdk/per-user-auth-isolation.test.ts +++ b/packages/plugins/mcp/src/sdk/per-user-auth-isolation.test.ts @@ -19,7 +19,7 @@ import * as http from "node:http"; import { describe, expect, it } from "@effect/vitest"; -import { Cause, Effect, Exit } from "effect"; +import { Cause, Effect, Exit, Predicate } from "effect"; import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js"; import { z } from "zod"; @@ -325,8 +325,8 @@ describe("per-user MCP auth isolation", () => { cause?: { _tag?: string }; } | undefined; - expect(outer?._tag).toBe("ToolInvocationError"); - expect(outer?.cause?._tag).toBe("McpConnectionError"); + expect(Predicate.isTagged(outer, "ToolInvocationError")).toBe(true); + expect(Predicate.isTagged(outer?.cause, "ConnectionNotFoundError")).toBe(true); // CRITICAL: no outbound MCP request was made on user B's behalf // carrying user A's bearer token. Auth resolution must have @@ -431,8 +431,8 @@ describe("per-user MCP auth isolation", () => { cause?: { _tag?: string }; } | undefined; - expect(outer?._tag).toBe("ToolInvocationError"); - expect(outer?.cause?._tag).toBe("McpConnectionError"); + expect(Predicate.isTagged(outer, "ToolInvocationError")).toBe(true); + expect(Predicate.isTagged(outer?.cause, "McpConnectionError")).toBe(true); const afterUserB = server.recorded().slice(recordedBeforeUserB); for (const req of afterUserB) { diff --git a/packages/plugins/mcp/src/sdk/plugin.ts b/packages/plugins/mcp/src/sdk/plugin.ts index 6b227dd19..39b1421f9 100644 --- a/packages/plugins/mcp/src/sdk/plugin.ts +++ b/packages/plugins/mcp/src/sdk/plugin.ts @@ -1,4 +1,4 @@ -import { Duration, Effect, Exit, Result, Scope, ScopedCache } from "effect"; +import { Duration, Effect, Exit, Option, Result, Scope, ScopedCache } from "effect"; import type { OAuthClientProvider } from "@modelcontextprotocol/sdk/client/auth.js"; @@ -10,6 +10,11 @@ import { SourceDetectionResult, Usage, definePlugin, + type ConnectionNotFoundError, + type ConnectionProviderNotRegisteredError, + type ConnectionReauthRequiredError, + type ConnectionRefreshError, + type ConnectionRefreshNotSupportedError, resolveSecretBackedMap as resolveSharedSecretBackedMap, type PluginCtx, type StorageFailure, @@ -182,42 +187,43 @@ const makeOAuthProvider = (accessToken: string): OAuthClientProvider => ({ tokens: () => ({ access_token: accessToken, token_type: "Bearer" }), saveTokens: () => undefined, redirectToAuthorization: async () => { + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: MCP SDK OAuth provider callback must reject unsupported reauth throw new Error("MCP OAuth re-authorization required"); }, saveCodeVerifier: () => undefined, codeVerifier: () => { + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: MCP SDK OAuth provider callback must throw when no PKCE verifier exists throw new Error("No active PKCE verifier"); }, saveDiscoveryState: () => undefined, discoveryState: () => undefined, }); -const remoteConnectionError = (message: string) => - new McpConnectionError({ transport: "remote", message }); - -const mcpDiscoveryError = (message: string) => - new McpToolDiscoveryError({ stage: "list_tools", message }); - const resolveSecretBackedMap = ( values: Record | undefined, ctx: PluginCtx, ): Effect.Effect | undefined, McpConnectionError | StorageFailure> => resolveSharedSecretBackedMap({ values, - getSecret: ctx.secrets.get, + getSecret: (secretId) => + ctx.secrets.get(secretId).pipe( + Effect.catchTag( + "SecretOwnedByConnectionError", + () => + Effect.fail( + new McpConnectionError({ + transport: "remote", + message: `Failed to resolve secret "${secretId}"`, + }), + ), + ), + ), onMissing: (_name, value) => - remoteConnectionError(`Failed to resolve secret "${value.secretId}"`), - onError: (err, _name, value) => - "_tag" in err && err._tag === "SecretOwnedByConnectionError" - ? remoteConnectionError(`Failed to resolve secret "${value.secretId}"`) - : err, - }).pipe( - Effect.mapError((err) => - "_tag" in err && err._tag === "SecretOwnedByConnectionError" - ? remoteConnectionError("Failed to resolve secret") - : err, - ), - ); + new McpConnectionError({ + transport: "remote", + message: `Failed to resolve secret "${value.secretId}"`, + }), + }); const plainStringMap = ( values: Record | undefined, @@ -233,11 +239,23 @@ const plainStringMap = ( // Shared connector resolution — reads secrets, builds stdio/remote input // --------------------------------------------------------------------------- +export type McpConnectionAccessFailure = + | ConnectionNotFoundError + | ConnectionProviderNotRegisteredError + | ConnectionRefreshNotSupportedError + | ConnectionReauthRequiredError + | ConnectionRefreshError; + +type McpConnectorResolutionFailure = + | McpConnectionError + | McpConnectionAccessFailure + | StorageFailure; + const resolveConnectorInput = ( sd: McpStoredSourceData, ctx: PluginCtx, allowStdio: boolean, -): Effect.Effect => { +): Effect.Effect => { if (sd.transport === "stdio") { if (!allowStdio) { return Effect.fail( @@ -268,16 +286,22 @@ const resolveConnectorInput = ( const val = yield* ctx.secrets .get(auth.secretId) .pipe( - Effect.mapError((err) => - "_tag" in err && err._tag === "SecretOwnedByConnectionError" - ? remoteConnectionError(`Failed to resolve secret "${auth.secretId}"`) - : err, + Effect.catchTag( + "SecretOwnedByConnectionError", + () => + Effect.fail( + new McpConnectionError({ + transport: "remote", + message: `Failed to resolve secret "${auth.secretId}"`, + }), + ), ), ); if (val === null) { - return yield* Effect.fail( - remoteConnectionError(`Failed to resolve secret "${auth.secretId}"`), - ); + return yield* new McpConnectionError({ + transport: "remote", + message: `Failed to resolve secret "${auth.secretId}"`, + }); } headers[auth.headerName] = auth.prefix ? `${auth.prefix}${val}` : val; } else if (auth.kind === "oauth2") { @@ -286,17 +310,7 @@ const resolveConnectorInput = ( // The canonical `"oauth2"` ConnectionProvider registered by // core owns the refresh lifecycle; we just wrap the current // token for the SDK's transport. - const accessToken = yield* ctx.connections - .accessToken(auth.connectionId) - .pipe( - Effect.mapError((err) => - remoteConnectionError( - `Failed to resolve OAuth connection "${auth.connectionId}": ${ - "message" in err ? (err as { message: string }).message : String(err) - }`, - ), - ), - ); + const accessToken = yield* ctx.connections.accessToken(auth.connectionId); authProvider = makeOAuthProvider(accessToken); } @@ -318,15 +332,25 @@ const resolveConnectorInput = ( // --------------------------------------------------------------------------- interface McpRuntime { - readonly connectionCache: ScopedCache.ScopedCache; - readonly pendingConnectors: Map>; + readonly connectionCache: ScopedCache.ScopedCache< + string, + McpConnection, + McpConnectorResolutionFailure + >; + readonly pendingConnectors: Map< + string, + Effect.Effect + >; readonly cacheScope: Scope.Closeable; } const makeRuntime = (): Effect.Effect => Effect.gen(function* () { const cacheScope = yield* Scope.make(); - const pendingConnectors = new Map>(); + const pendingConnectors = new Map< + string, + Effect.Effect + >(); const connectionCache = yield* ScopedCache.make({ lookup: (key: string) => Effect.acquireRelease( @@ -342,7 +366,17 @@ const makeRuntime = (): Effect.Effect => } return connector; }), - (connection) => Effect.promise(() => connection.close().catch(() => {})), + (connection) => + Effect.ignore( + Effect.tryPromise({ + try: () => connection.close(), + catch: () => + new McpConnectionError({ + transport: "auto", + message: "Failed to close MCP connection", + }), + }), + ), ), capacity: 64, timeToLive: Duration.minutes(5), @@ -455,7 +489,10 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { const endpoint = typeof input === "string" ? input : input.endpoint; const trimmed = endpoint.trim(); if (!trimmed) { - return yield* Effect.fail(remoteConnectionError("Endpoint URL is required")); + return yield* new McpConnectionError({ + transport: "remote", + message: "Endpoint URL is required", + }); } const name = yield* Effect.try({ @@ -510,13 +547,13 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { queryParams: probeQueryParams, }); if (shape.kind !== "mcp") { - return yield* Effect.fail( - remoteConnectionError( + return yield* new McpConnectionError({ + transport: "remote", + message: shape.kind === "not-mcp" ? `Endpoint does not look like an MCP server: ${shape.reason}` : `Could not reach endpoint: ${shape.reason}`, - ), - ); + }); } const probeResult = yield* ctx.oauth @@ -542,9 +579,10 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { } satisfies McpProbeResult; } - return yield* Effect.fail( - remoteConnectionError("MCP server requires authentication but OAuth discovery failed"), - ); + return yield* new McpConnectionError({ + transport: "remote", + message: "MCP server requires authentication but OAuth discovery failed", + }); }).pipe( Effect.withSpan("mcp.plugin.probe_endpoint", { attributes: { "mcp.endpoint": typeof input === "string" ? input : input.endpoint }, @@ -585,12 +623,15 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { // the caller at the end. const discovery: Result.Result< McpToolManifest, - McpToolDiscoveryError | McpConnectionError | StorageFailure + McpToolDiscoveryError | McpConnectorResolutionFailure > = Result.isSuccess(resolved) ? yield* discoverTools(createMcpConnector(resolved.success)).pipe( - Effect.mapError((err) => - mcpDiscoveryError(`MCP discovery failed: ${err.message}`), + Effect.mapError(() => + new McpToolDiscoveryError({ + stage: "list_tools", + message: "MCP discovery failed", + }), ), Effect.result, Effect.withSpan("mcp.plugin.discover_tools", { @@ -706,9 +747,10 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { }), ); if (!sd) { - return yield* Effect.fail( - remoteConnectionError(`No stored config for MCP source "${namespace}"`), - ); + return yield* new McpConnectionError({ + transport: "remote", + message: `No stored config for MCP source "${namespace}"`, + }); } const ci = yield* resolveConnectorInput(sd, ctx, allowStdio).pipe( @@ -720,7 +762,13 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { }), ); const manifest = yield* discoverTools(createMcpConnector(ci)).pipe( - Effect.mapError((err) => mcpDiscoveryError(`MCP refresh failed: ${err.message}`)), + Effect.mapError( + () => + new McpToolDiscoveryError({ + stage: "list_tools", + message: "MCP refresh failed", + }), + ), Effect.withSpan("mcp.plugin.discover_tools", { attributes: { "mcp.source.namespace": namespace }, }), @@ -836,7 +884,10 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { }), ); if (!entry) { - return yield* Effect.fail(new Error(`No MCP binding found for tool "${toolRow.id}"`)); + return yield* new McpConnectionError({ + transport: "auto", + message: `No MCP binding found for tool "${toolRow.id}"`, + }); } const sd = yield* ctx.storage.getSourceConfig(entry.namespace, toolScope).pipe( @@ -845,9 +896,10 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { }), ); if (!sd) { - return yield* Effect.fail( - new Error(`No MCP source config for namespace "${entry.namespace}"`), - ); + return yield* new McpConnectionError({ + transport: "auto", + message: `No MCP source config for namespace "${entry.namespace}"`, + }); } return yield* invokeMcpTool({ @@ -859,14 +911,6 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { resolveConnector: () => resolveConnectorInput(sd, ctx, allowStdio).pipe( Effect.flatMap((ci) => createMcpConnector(ci)), - Effect.mapError((err) => - err instanceof McpConnectionError - ? err - : new McpConnectionError({ - transport: "auto", - message: err instanceof Error ? err.message : String(err), - }), - ), Effect.withSpan("mcp.plugin.resolve_connector", { attributes: { "mcp.source.namespace": entry.namespace, @@ -896,7 +940,7 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { try: () => new URL(trimmed), catch: (cause) => cause, }).pipe(Effect.option); - if (parsed._tag === "None") return null; + if (Option.isNone(parsed)) return null; const name = parsed.value.hostname || "mcp"; const namespace = deriveMcpNamespace({ endpoint: trimmed }); @@ -1102,7 +1146,11 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { * composition. `UniqueViolationError` passes through — plugins can * `Effect.catchTag` it if they want a friendlier user-facing error. */ -export type McpExtensionFailure = McpConnectionError | McpToolDiscoveryError | StorageFailure; +export type McpExtensionFailure = + | McpConnectionError + | McpToolDiscoveryError + | McpConnectionAccessFailure + | StorageFailure; export interface McpPluginExtension { readonly probeEndpoint: (