diff --git a/packages/opencode/src/mcp/index.ts b/packages/opencode/src/mcp/index.ts index 9a5a9e9993..c2ac9b9535 100644 --- a/packages/opencode/src/mcp/index.ts +++ b/packages/opencode/src/mcp/index.ts @@ -10,7 +10,6 @@ import { CallToolResultSchema, ListToolsResultSchema, ToolSchema, - type Tool as MCPToolDef, ToolListChangedNotificationSchema, } from "@modelcontextprotocol/sdk/types.js" import { Config } from "@/config/config" @@ -71,6 +70,7 @@ export class NotFoundError extends Schema.TaggedErrorClass()("MCP }) {} type MCPClient = Client +type MCPToolDef = Awaited>["tools"][number] const StatusConnected = Schema.Struct({ status: Schema.Literal("connected") }).annotate({ identifier: "MCPStatusConnected", @@ -114,7 +114,7 @@ function isMcpConfigured(entry: McpEntry): entry is ConfigMCPV1.Info { const sanitize = (s: string) => s.replace(/[^a-zA-Z0-9_-]/g, "_") const MAX_LIST_PAGES = 1_000 -function remoteURL(key: string, value: string) { +function remoteURL(value: string) { if (URL.canParse(value)) return new URL(value) } @@ -124,6 +124,10 @@ function isOutputSchemaValidationError(error: Error) { ) } +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value) +} + async function paginate( list: (cursor?: string) => Promise, items: (result: R) => T[], @@ -168,12 +172,12 @@ function convertMcpTool(mcpTool: MCPToolDef, client: MCPClient, timeout?: number const inputSchema = mcpTool.inputSchema // Spread first, then override type to ensure it's always "object" - const schema: JSONSchema7 = { - ...(inputSchema as JSONSchema7), + const schema = { + ...inputSchema, type: "object", - properties: (inputSchema.properties ?? {}) as JSONSchema7["properties"], + properties: inputSchema.properties ?? {}, additionalProperties: false, - } + } satisfies JSONSchema7 return dynamicTool({ description: mcpTool.description ?? "", @@ -182,7 +186,7 @@ function convertMcpTool(mcpTool: MCPToolDef, client: MCPClient, timeout?: number return client.callTool( { name: mcpTool.name, - arguments: (args || {}) as Record, + arguments: isRecord(args) ? args : {}, }, CallToolResultSchema, { @@ -196,25 +200,15 @@ function convertMcpTool(mcpTool: MCPToolDef, client: MCPClient, timeout?: number } function defs(client: MCPClient, timeout?: number) { - return listTools(client, timeout ?? DEFAULT_TIMEOUT).pipe( - Effect.catch((err) => { - return Effect.succeed(undefined) - }), - ) + return listTools(client, timeout ?? DEFAULT_TIMEOUT).pipe(Effect.catch(() => Effect.void)) } function fetchFromClient( clientName: string, client: Client, listFn: (c: Client) => Promise, - label: string, ) { - return Effect.tryPromise({ - try: () => listFn(client), - catch: (e: any) => { - return e - }, - }).pipe( + return Effect.tryPromise(() => listFn(client)).pipe( Effect.map((items) => { const out: Record = {} const sanitizedClient = sanitize(clientName) @@ -227,11 +221,11 @@ function fetchFromClient( ) } -interface CreateResult { - mcpClient?: MCPClient - status: Status - defs?: MCPToolDef[] -} +type UnavailableStatus = Exclude +type ConnectResult = { type: "connected"; client: MCPClient } | { type: "unavailable"; status: UnavailableStatus } +type CreateResult = + | { type: "connected"; client: MCPClient; defs: MCPToolDef[] } + | { type: "unavailable"; status: UnavailableStatus } interface AuthResult { authorizationUrl: string @@ -308,20 +302,20 @@ export const layer = Layer.effect( (t, exit) => (Exit.isFailure(exit) ? Effect.tryPromise(() => t.close()).pipe(Effect.ignore) : Effect.void), ) - const DISABLED_RESULT: CreateResult = { status: { status: "disabled" } } + const DISABLED_RESULT = { + type: "unavailable", + status: { status: "disabled" }, + } satisfies CreateResult - const connectRemote = Effect.fn("MCP.connectRemote")(function* ( - key: string, - mcp: ConfigMCPV1.Info & { type: "remote" }, - ) { + const connectRemote = Effect.fn("MCP.connectRemote")(function* (key: string, mcp: ConfigMCPV1.Remote) { const oauthDisabled = mcp.oauth === false const oauthConfig = typeof mcp.oauth === "object" ? mcp.oauth : undefined - const url = remoteURL(key, mcp.url) + const url = remoteURL(mcp.url) if (!url) { return { - client: undefined as MCPClient | undefined, - status: { status: "failed" as const, error: `Invalid MCP URL for "${key}"` }, - } + type: "unavailable", + status: { status: "failed", error: `Invalid MCP URL for "${key}"` }, + } satisfies ConnectResult } let authProvider: McpOAuthProvider | undefined @@ -361,7 +355,7 @@ export const layer = Layer.effect( ] const connectTimeout = mcp.timeout ?? DEFAULT_TIMEOUT - let lastStatus: Status | undefined + let lastStatus: UnavailableStatus | undefined for (const { name, transport } of transports) { const result = yield* connectTransport(transport, connectTimeout).pipe( @@ -374,7 +368,7 @@ export const layer = Layer.effect( if (isAuthError) { if (lastError.message.includes("registration") || lastError.message.includes("client_id")) { lastStatus = { - status: "needs_client_registration" as const, + status: "needs_client_registration", error: "Server does not support dynamic client registration. Please provide clientId in config.", } return events @@ -387,7 +381,7 @@ export const layer = Layer.effect( .pipe(Effect.ignore, Effect.as(undefined)) } else { pendingOAuthTransports.set(key, transport) - lastStatus = { status: "needs_auth" as const } + lastStatus = { status: "needs_auth" } return events .publish(TuiEvent.ToastShow, { title: "MCP Authentication Required", @@ -399,27 +393,22 @@ export const layer = Layer.effect( } } - lastStatus = { status: "failed" as const, error: lastError.message } - return Effect.succeed(undefined) + lastStatus = { status: "failed", error: lastError.message } + return Effect.void }), ) - if (result) { - return { client: result.client as MCPClient | undefined, status: { status: "connected" } as Status } - } + if (result) return { type: "connected", client: result.client } satisfies ConnectResult // If this was an auth error, stop trying other transports if (lastStatus?.status === "needs_auth" || lastStatus?.status === "needs_client_registration") break } return { - client: undefined as MCPClient | undefined, - status: (lastStatus ?? { status: "failed", error: "Unknown error" }) as Status, - } + type: "unavailable", + status: lastStatus ?? { status: "failed", error: "Unknown error" }, + } satisfies ConnectResult }) - const connectLocal = Effect.fn("MCP.connectLocal")(function* ( - key: string, - mcp: ConfigMCPV1.Info & { type: "local" }, - ) { + const connectLocal = Effect.fn("MCP.connectLocal")(function* (mcp: ConfigMCPV1.Local) { const [cmd, ...args] = mcp.command const cwd = yield* InstanceState.directory const transport = new StdioClientTransport({ @@ -436,13 +425,13 @@ export const layer = Layer.effect( const connectTimeout = mcp.timeout ?? DEFAULT_TIMEOUT return yield* connectTransport(transport, connectTimeout).pipe( - Effect.map((client): { client: MCPClient | undefined; status: Status } => ({ - client, - status: { status: "connected" }, - })), - Effect.catch((error): Effect.Effect<{ client: MCPClient | undefined; status: Status }> => { + Effect.map((client) => ({ type: "connected", client }) satisfies ConnectResult), + Effect.catch((error) => { const msg = error instanceof Error ? error.message : String(error) - return Effect.succeed({ client: undefined, status: { status: "failed", error: msg } }) + return Effect.succeed({ + type: "unavailable", + status: { status: "failed", error: msg }, + } satisfies ConnectResult) }), ) }) @@ -452,35 +441,33 @@ export const layer = Layer.effect( return DISABLED_RESULT } - const { client: mcpClient, status } = - mcp.type === "remote" - ? yield* connectRemote(key, mcp as ConfigMCPV1.Info & { type: "remote" }) - : yield* connectLocal(key, mcp as ConfigMCPV1.Info & { type: "local" }) + const result = mcp.type === "remote" ? yield* connectRemote(key, mcp) : yield* connectLocal(mcp) - if (!mcpClient) { - if (status.status !== "connected" && status.status !== "disabled") { - yield* Effect.logWarning("server unavailable", { key, type: mcp.type, status: status.status }) - } - return { status } satisfies CreateResult + if (result.type === "unavailable") { + yield* Effect.logWarning("server unavailable", { key, type: mcp.type, status: result.status.status }) + return result satisfies CreateResult } - const listed = mcpClient.getServerCapabilities()?.tools ? yield* defs(mcpClient, mcp.timeout) : [] + const listed = result.client.getServerCapabilities()?.tools ? yield* defs(result.client, mcp.timeout) : [] if (!listed) { - yield* Effect.tryPromise(() => mcpClient.close()).pipe(Effect.ignore) - return { status: { status: "failed", error: "Failed to get tools" } } satisfies CreateResult + yield* Effect.tryPromise(() => result.client.close()).pipe(Effect.ignore) + return { + type: "unavailable", + status: { status: "failed", error: "Failed to get tools" }, + } satisfies CreateResult } - return { mcpClient, status, defs: listed } satisfies CreateResult + return { type: "connected", client: result.client, defs: listed } satisfies CreateResult }) const cfgSvc = yield* Config.Service const descendants = Effect.fnUntraced( function* (pid: number) { - if (process.platform === "win32") return [] as number[] + if (process.platform === "win32") return Array() const pids: number[] = [] const queue = [pid] - while (queue.length > 0) { - const current = queue.shift()! + for (let index = 0; index < queue.length; index++) { + const current = queue[index] const handle = yield* spawner.spawn(ChildProcess.make("pgrep", ["-P", String(current)], { stdin: "ignore" })) const text = yield* Stream.mkString(Stream.decodeText(handle.stdout)) yield* handle.exitCode @@ -495,7 +482,7 @@ export const layer = Layer.effect( return pids }, Effect.scoped, - Effect.catch(() => Effect.succeed([] as number[])), + Effect.catch(() => Effect.succeed(Array())), ) function watch(s: State, name: string, client: MCPClient, bridge: EffectBridge.Shape, timeout?: number) { @@ -540,12 +527,14 @@ export const layer = Layer.effect( const result = yield* create(key, mcp).pipe(Effect.catch(() => Effect.void)) if (!result) return - s.status[key] = result.status - if (result.mcpClient) { - s.clients[key] = result.mcpClient - s.defs[key] = result.defs! - watch(s, key, result.mcpClient, bridge, mcp.timeout) + if (result.type === "unavailable") { + s.status[key] = result.status + return } + s.status[key] = { status: "connected" } + s.clients[key] = result.client + s.defs[key] = result.defs + watch(s, key, result.client, bridge, mcp.timeout) }), { concurrency: "unbounded" }, ) @@ -628,14 +617,14 @@ export const layer = Layer.effect( const s = yield* InstanceState.get(state) const result = yield* create(name, mcp) - s.status[name] = result.status - if (!result.mcpClient) { + if (result.type === "unavailable") { + s.status[name] = result.status yield* closeClient(s, name) delete s.clients[name] return result.status } - return yield* storeClient(s, name, result.mcpClient, result.defs!, mcp.timeout) + return yield* storeClient(s, name, result.client, result.defs, mcp.timeout) }) const add = Effect.fn("MCP.add")(function* (name: string, mcp: ConfigMCPV1.Info) { @@ -666,92 +655,59 @@ export const layer = Layer.effect( const config = cfg.mcp ?? {} const defaultTimeout = cfg.experimental?.mcp_timeout - const connectedClients = Object.entries(s.clients).filter( - ([clientName]) => s.status[clientName]?.status === "connected", - ) + for (const [clientName, client] of Object.entries(s.clients)) { + if (s.status[clientName]?.status !== "connected") continue + const mcpConfig = config[clientName] + const entry = mcpConfig && isMcpConfigured(mcpConfig) ? mcpConfig : s.config[clientName] + const listed = s.defs[clientName] + if (!listed) continue - yield* Effect.forEach( - connectedClients, - ([clientName, client]) => - Effect.gen(function* () { - const mcpConfig = config[clientName] - const entry = mcpConfig && isMcpConfigured(mcpConfig) ? mcpConfig : s.config[clientName] - - const listed = s.defs[clientName] - if (!listed) { - return - } - - const timeout = entry?.timeout ?? defaultTimeout - for (const mcpTool of listed) { - result[sanitize(clientName) + "_" + sanitize(mcpTool.name)] = convertMcpTool(mcpTool, client, timeout) - } - }), - { concurrency: "unbounded" }, - ) + const timeout = entry?.timeout ?? defaultTimeout + for (const mcpTool of listed) { + result[sanitize(clientName) + "_" + sanitize(mcpTool.name)] = convertMcpTool(mcpTool, client, timeout) + } + } return result }) - function collectFromConnected( - s: State, - listFn: (c: Client) => Promise, - label: string, - ) { + function collectFromConnected(s: State, listFn: (c: Client) => Promise) { return Effect.forEach( Object.entries(s.clients).filter(([name]) => s.status[name]?.status === "connected"), ([clientName, client]) => - fetchFromClient(clientName, client, listFn, label).pipe(Effect.map((items) => Object.entries(items ?? {}))), + fetchFromClient(clientName, client, listFn).pipe(Effect.map((items) => Object.entries(items ?? {}))), { concurrency: "unbounded" }, ).pipe(Effect.map((results) => Object.fromEntries(results.flat()))) } const prompts = Effect.fn("MCP.prompts")(function* () { const s = yield* InstanceState.get(state) - return yield* collectFromConnected( - s, - (c) => - c.getServerCapabilities()?.prompts - ? paginate( - (cursor) => c.listPrompts(cursor === undefined ? undefined : { cursor }), - (result) => result.prompts, - ) - : Promise.resolve([]), - "prompts", + return yield* collectFromConnected(s, (c) => + c.getServerCapabilities()?.prompts + ? paginate( + (cursor) => c.listPrompts(cursor === undefined ? undefined : { cursor }), + (result) => result.prompts, + ) + : Promise.resolve([]), ) }) const resources = Effect.fn("MCP.resources")(function* () { const s = yield* InstanceState.get(state) - return yield* collectFromConnected( - s, - (c) => - c.getServerCapabilities()?.resources - ? paginate( - (cursor) => c.listResources(cursor === undefined ? undefined : { cursor }), - (result) => result.resources, - ) - : Promise.resolve([]), - "resources", + return yield* collectFromConnected(s, (c) => + c.getServerCapabilities()?.resources + ? paginate( + (cursor) => c.listResources(cursor === undefined ? undefined : { cursor }), + (result) => result.resources, + ) + : Promise.resolve([]), ) }) - const withClient = Effect.fnUntraced(function* ( - clientName: string, - fn: (client: MCPClient) => Promise, - label: string, - meta?: Record, - ) { + const withClient = Effect.fnUntraced(function* (clientName: string, fn: (client: MCPClient) => Promise) { const s = yield* InstanceState.get(state) const client = s.clients[clientName] - if (!client) { - return undefined - } - return yield* Effect.tryPromise({ - try: () => fn(client), - catch: (e: any) => { - return e - }, - }).pipe(Effect.orElseSucceed(() => undefined)) + if (!client) return undefined + return yield* Effect.tryPromise(() => fn(client)).pipe(Effect.orElseSucceed(() => undefined)) }) const getPrompt = Effect.fn("MCP.getPrompt")(function* ( @@ -759,15 +715,11 @@ export const layer = Layer.effect( name: string, args?: Record, ) { - return yield* withClient(clientName, (client) => client.getPrompt({ name, arguments: args }), "getPrompt", { - promptName: name, - }) + return yield* withClient(clientName, (client) => client.getPrompt({ name, arguments: args })) }) const readResource = Effect.fn("MCP.readResource")(function* (clientName: string, resourceUri: string) { - return yield* withClient(clientName, (client) => client.readResource({ uri: resourceUri }), "readResource", { - resourceUri, - }) + return yield* withClient(clientName, (client) => client.readResource({ uri: resourceUri })) }) const getMcpConfig = Effect.fnUntraced(function* (mcpName: string) { @@ -790,7 +742,7 @@ export const layer = Layer.effect( const mcpConfig = yield* requireMcpConfig(mcpName) if (mcpConfig.type !== "remote") throw new Error(`MCP server ${mcpName} is not a remote server`) if (mcpConfig.oauth === false) throw new Error(`MCP server ${mcpName} has OAuth explicitly disabled`) - const url = remoteURL(mcpName, mcpConfig.url) + const url = remoteURL(mcpConfig.url) if (!url) throw new Error(`Invalid MCP URL for "${mcpName}"`) // OAuth config is optional - if not provided, we'll use auto-discovery @@ -862,7 +814,7 @@ export const layer = Layer.effect( : undefined if (!client || !listed) { yield* Effect.tryPromise(() => client?.close() ?? Promise.resolve()).pipe(Effect.ignore) - return { status: "failed", error: "Failed to get tools" } as Status + return { status: "failed", error: "Failed to get tools" } satisfies Status } const s = yield* InstanceState.get(state) @@ -910,14 +862,14 @@ export const layer = Layer.effect( if (!transport) throw new Error(`No pending OAuth flow for MCP server: ${mcpName}`) const result = yield* Effect.tryPromise({ - try: () => transport.finishAuth(authorizationCode).then(() => true as const), + try: () => transport.finishAuth(authorizationCode).then((): true => true), catch: (error) => { return error }, }).pipe(Effect.option) if (Option.isNone(result)) { - return { status: "failed", error: "OAuth completion failed" } as Status + return { status: "failed", error: "OAuth completion failed" } satisfies Status } yield* auth.clearCodeVerifier(mcpName) @@ -946,9 +898,9 @@ export const layer = Layer.effect( const getAuthStatus = Effect.fn("MCP.getAuthStatus")(function* (mcpName: string) { const entry = yield* auth.get(mcpName) - if (!entry?.tokens) return "not_authenticated" as AuthStatus + if (!entry?.tokens) return "not_authenticated" const expired = yield* auth.isTokenExpired(mcpName) - return (expired ? "expired" : "authenticated") as AuthStatus + return expired ? "expired" : "authenticated" }) return Service.of({