From 5619595abcd9cbed9bb8c5949b2d9946b4cfb5df Mon Sep 17 00:00:00 2001 From: Aiden Cline Date: Mon, 29 Jun 2026 13:00:51 -0500 Subject: [PATCH] feat(core): list mcp server tools with change tracking Add tools() and onToolsChanged() to the MCPClient.Connection abstraction: list is capability-gated, paginated with cursor-dedup, and tolerates unresolvable outputSchemas via a per-page fallback. Tool listing is folded into connect so a listing failure marks the server failed rather than leaving it connected with a silently empty list. Cache MCP.Tool values per server, refresh on tools/list_changed notifications, and publish McpEvent.ToolsChanged. MCP.tools() aggregates the cached tools across all connected servers. --- packages/core/src/mcp/client.ts | 82 ++++++++++++++++++++++++++++++++- packages/core/src/mcp/index.ts | 32 +++++++++++-- 2 files changed, 110 insertions(+), 4 deletions(-) diff --git a/packages/core/src/mcp/client.ts b/packages/core/src/mcp/client.ts index a9ffeb89b7..68f55104dd 100644 --- a/packages/core/src/mcp/client.ts +++ b/packages/core/src/mcp/client.ts @@ -6,12 +6,25 @@ import { Client, type ClientOptions } from "@modelcontextprotocol/sdk/client/ind import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js" import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js" import { UnauthorizedError } from "@modelcontextprotocol/sdk/client/auth.js" -import { ListRootsRequestSchema } from "@modelcontextprotocol/sdk/types.js" +import { + ListRootsRequestSchema, + ListToolsResultSchema, + ToolListChangedNotificationSchema, + ToolSchema, +} from "@modelcontextprotocol/sdk/types.js" import { Cause, Effect, Exit, Schema } from "effect" import { ConfigMCP } from "../config/mcp" import { InstallationVersion } from "../installation/version" const DEFAULT_STARTUP_TIMEOUT = 30_000 +const DEFAULT_REQUEST_TIMEOUT = 30_000 +const MAX_LIST_PAGES = 1_000 + +// Some servers advertise tool outputSchemas the SDK's strict validator can't resolve; this drops +// only that field so a single bad schema doesn't blank out the whole tool list. +const TolerantListToolsResult = ListToolsResultSchema.extend({ + tools: ToolSchema.omit({ outputSchema: true }).array(), +}) export class NeedsAuthError extends Schema.TaggedErrorClass()("MCP.NeedsAuthError", { server: Schema.String, @@ -22,11 +35,21 @@ export class ConnectError extends Schema.TaggedErrorClass()("MCP.C message: Schema.String, }) {} +export interface ToolDefinition { + readonly name: string + readonly description: string | undefined + readonly inputSchema: unknown +} + /** Handle over a connected MCP server that keeps the SDK `Client` out of the rest of core. */ export interface Connection { /** Server-supplied usage instructions from the initialize result, if any. */ readonly instructions: string | undefined + /** Lists the server's tools; returns [] when the server doesn't advertise tool support, fails on a transport error. */ + readonly tools: () => Effect.Effect readonly onClose: (callback: () => void) => void + /** Registers a callback fired when the server announces its tool list changed; no-op if unsupported. */ + readonly onToolsChanged: (callback: () => void) => void } /** Connects an MCP server; closing the calling scope tears down the transport and any spawned process. */ @@ -74,11 +97,45 @@ export const connect = Effect.fnUntraced(function* ( }).pipe(Effect.exit) if (Exit.isSuccess(exit)) { yield* Effect.addFinalizer(() => Effect.promise(() => client.close()).pipe(Effect.ignore)) + const requestTimeout = config.timeout?.request ?? DEFAULT_REQUEST_TIMEOUT return { instructions: client.getInstructions()?.trim() || undefined, + tools: () => + Effect.gen(function* () { + if (!client.getServerCapabilities()?.tools) return [] + const tools = yield* Effect.tryPromise({ + try: () => + paginate( + async (cursor) => { + const params = cursor === undefined ? undefined : { cursor } + try { + return await client.listTools(params, { timeout: requestTimeout }) + } catch (error) { + if (!(error instanceof Error) || !isOutputSchemaError(error)) throw error + return client.request({ method: "tools/list", params }, TolerantListToolsResult, { + timeout: requestTimeout, + }) + } + }, + (result) => result.tools, + ), + catch: (error) => (error instanceof Error ? error : new Error(String(error))), + }).pipe( + Effect.tapError((error) => Effect.logWarning("failed to list MCP tools", { server, error: error.message })), + ) + return tools.map((tool) => ({ + name: tool.name, + description: tool.description, + inputSchema: tool.inputSchema, + })) + }), onClose: (callback) => { client.onclose = callback }, + onToolsChanged: (callback) => { + if (!client.getServerCapabilities()?.tools?.listChanged) return + client.setNotificationHandler(ToolListChangedNotificationSchema, async () => callback()) + }, } satisfies Connection } @@ -88,3 +145,26 @@ export const connect = Effect.fnUntraced(function* ( return yield* new NeedsAuthError({ server }) return yield* new ConnectError({ server, message: error instanceof Error ? error.message : String(error) }) }) + +async function paginate( + list: (cursor: string | undefined) => Promise, + items: (result: R) => T[], +) { + const collected: T[] = [] + const seen = new Set() + let cursor: string | undefined + for (let page = 0; page < MAX_LIST_PAGES; page++) { + const result = await list(cursor) + collected.push(...items(result)) + if (result.nextCursor === undefined) return collected + if (seen.has(result.nextCursor)) throw new Error(`MCP list returned duplicate cursor: ${result.nextCursor}`) + seen.add(result.nextCursor) + cursor = result.nextCursor + } + throw new Error(`MCP list exceeded ${MAX_LIST_PAGES} pages`) +} + +const isOutputSchemaError = (error: Error) => + /can't resolve reference|resolves to more than one schema|outputSchema|schema.*reference|reference.*schema/i.test( + error.message, + ) diff --git a/packages/core/src/mcp/index.ts b/packages/core/src/mcp/index.ts index c3f27fcfcc..8edbcd3476 100644 --- a/packages/core/src/mcp/index.ts +++ b/packages/core/src/mcp/index.ts @@ -141,6 +141,7 @@ type ServerEntry = { readonly startup: Deferred.Deferred scope?: Scope.Closeable client?: MCPClient.Connection + tools?: ReadonlyArray readonly integrationID?: Integration.ID readonly connection?: IntegrationConnection.Info } @@ -208,26 +209,49 @@ export const layer = Layer.effect( connection: entry.connection, }) + const toTool = (server: ServerName, def: MCPClient.ToolDefinition) => + new Tool({ server, name: def.name, description: def.description, inputSchema: def.inputSchema }) + + const refreshTools = (name: ServerName, entry: ServerEntry, connection: MCPClient.Connection) => + connection.tools().pipe( + Effect.map((defs) => { + entry.tools = defs.map((def) => toTool(name, def)) + }), + ) + const watch = (name: ServerName, entry: ServerEntry, connection: MCPClient.Connection) => { connection.onClose(() => { entry.client = undefined + entry.tools = undefined entry.status = { status: "failed", error: "Connection closed" } fork(events.publish(McpEvent.ToolsChanged, { server: name }).pipe(Effect.ignore)) }) + connection.onToolsChanged(() => { + fork( + refreshTools(name, entry, connection).pipe( + Effect.andThen(events.publish(McpEvent.ToolsChanged, { server: name })), + Effect.ignore, + ), + ) + }) } const startServer = (name: ServerName, entry: ServerEntry) => Effect.gen(function* () { const scope = yield* Scope.fork(root) entry.scope = scope + // List tools as part of connect so a failure here marks the server failed rather than + // leaving it connected with a silently empty tool list and no path to recover. const result = yield* MCPClient.connect(name, entry.config, location.directory).pipe( + Effect.flatMap((connection) => connection.tools().pipe(Effect.map((defs) => ({ connection, defs })))), Scope.provide(scope), Effect.exit, ) if (Exit.isSuccess(result)) { - entry.client = result.value + entry.client = result.value.connection + entry.tools = result.value.defs.map((def) => toTool(name, def)) entry.status = { status: "connected" } - watch(name, entry, result.value) + watch(name, entry, result.value.connection) return } yield* Scope.close(scope, Exit.void) @@ -266,7 +290,9 @@ export const layer = Layer.effect( }), tools: Effect.fn("MCP.tools")(function* () { yield* whenAllReady - return [] + return Array.from(runtime.values()) + .flatMap((entry) => entry.tools ?? []) + .toSorted((a, b) => a.server.localeCompare(b.server) || a.name.localeCompare(b.name)) }), instructions: Effect.fn("MCP.instructions")(function* () { yield* whenAllReady