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.
This commit is contained in:
parent
d4e00c061d
commit
5619595abc
2 changed files with 110 additions and 4 deletions
|
|
@ -6,12 +6,25 @@ import { Client, type ClientOptions } from "@modelcontextprotocol/sdk/client/ind
|
||||||
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js"
|
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js"
|
||||||
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"
|
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"
|
||||||
import { UnauthorizedError } from "@modelcontextprotocol/sdk/client/auth.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 { Cause, Effect, Exit, Schema } from "effect"
|
||||||
import { ConfigMCP } from "../config/mcp"
|
import { ConfigMCP } from "../config/mcp"
|
||||||
import { InstallationVersion } from "../installation/version"
|
import { InstallationVersion } from "../installation/version"
|
||||||
|
|
||||||
const DEFAULT_STARTUP_TIMEOUT = 30_000
|
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<NeedsAuthError>()("MCP.NeedsAuthError", {
|
export class NeedsAuthError extends Schema.TaggedErrorClass<NeedsAuthError>()("MCP.NeedsAuthError", {
|
||||||
server: Schema.String,
|
server: Schema.String,
|
||||||
|
|
@ -22,11 +35,21 @@ export class ConnectError extends Schema.TaggedErrorClass<ConnectError>()("MCP.C
|
||||||
message: Schema.String,
|
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. */
|
/** Handle over a connected MCP server that keeps the SDK `Client` out of the rest of core. */
|
||||||
export interface Connection {
|
export interface Connection {
|
||||||
/** Server-supplied usage instructions from the initialize result, if any. */
|
/** Server-supplied usage instructions from the initialize result, if any. */
|
||||||
readonly instructions: string | undefined
|
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<ToolDefinition[], Error>
|
||||||
readonly onClose: (callback: () => void) => void
|
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. */
|
/** 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)
|
}).pipe(Effect.exit)
|
||||||
if (Exit.isSuccess(exit)) {
|
if (Exit.isSuccess(exit)) {
|
||||||
yield* Effect.addFinalizer(() => Effect.promise(() => client.close()).pipe(Effect.ignore))
|
yield* Effect.addFinalizer(() => Effect.promise(() => client.close()).pipe(Effect.ignore))
|
||||||
|
const requestTimeout = config.timeout?.request ?? DEFAULT_REQUEST_TIMEOUT
|
||||||
return {
|
return {
|
||||||
instructions: client.getInstructions()?.trim() || undefined,
|
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) => {
|
onClose: (callback) => {
|
||||||
client.onclose = callback
|
client.onclose = callback
|
||||||
},
|
},
|
||||||
|
onToolsChanged: (callback) => {
|
||||||
|
if (!client.getServerCapabilities()?.tools?.listChanged) return
|
||||||
|
client.setNotificationHandler(ToolListChangedNotificationSchema, async () => callback())
|
||||||
|
},
|
||||||
} satisfies Connection
|
} satisfies Connection
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -88,3 +145,26 @@ export const connect = Effect.fnUntraced(function* (
|
||||||
return yield* new NeedsAuthError({ server })
|
return yield* new NeedsAuthError({ server })
|
||||||
return yield* new ConnectError({ server, message: error instanceof Error ? error.message : String(error) })
|
return yield* new ConnectError({ server, message: error instanceof Error ? error.message : String(error) })
|
||||||
})
|
})
|
||||||
|
|
||||||
|
async function paginate<R extends { nextCursor?: string }, T>(
|
||||||
|
list: (cursor: string | undefined) => Promise<R>,
|
||||||
|
items: (result: R) => T[],
|
||||||
|
) {
|
||||||
|
const collected: T[] = []
|
||||||
|
const seen = new Set<string>()
|
||||||
|
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,
|
||||||
|
)
|
||||||
|
|
|
||||||
|
|
@ -141,6 +141,7 @@ type ServerEntry = {
|
||||||
readonly startup: Deferred.Deferred<void>
|
readonly startup: Deferred.Deferred<void>
|
||||||
scope?: Scope.Closeable
|
scope?: Scope.Closeable
|
||||||
client?: MCPClient.Connection
|
client?: MCPClient.Connection
|
||||||
|
tools?: ReadonlyArray<Tool>
|
||||||
readonly integrationID?: Integration.ID
|
readonly integrationID?: Integration.ID
|
||||||
readonly connection?: IntegrationConnection.Info
|
readonly connection?: IntegrationConnection.Info
|
||||||
}
|
}
|
||||||
|
|
@ -208,26 +209,49 @@ export const layer = Layer.effect(
|
||||||
connection: entry.connection,
|
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) => {
|
const watch = (name: ServerName, entry: ServerEntry, connection: MCPClient.Connection) => {
|
||||||
connection.onClose(() => {
|
connection.onClose(() => {
|
||||||
entry.client = undefined
|
entry.client = undefined
|
||||||
|
entry.tools = undefined
|
||||||
entry.status = { status: "failed", error: "Connection closed" }
|
entry.status = { status: "failed", error: "Connection closed" }
|
||||||
fork(events.publish(McpEvent.ToolsChanged, { server: name }).pipe(Effect.ignore))
|
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) =>
|
const startServer = (name: ServerName, entry: ServerEntry) =>
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
const scope = yield* Scope.fork(root)
|
const scope = yield* Scope.fork(root)
|
||||||
entry.scope = scope
|
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(
|
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),
|
Scope.provide(scope),
|
||||||
Effect.exit,
|
Effect.exit,
|
||||||
)
|
)
|
||||||
if (Exit.isSuccess(result)) {
|
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" }
|
entry.status = { status: "connected" }
|
||||||
watch(name, entry, result.value)
|
watch(name, entry, result.value.connection)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
yield* Scope.close(scope, Exit.void)
|
yield* Scope.close(scope, Exit.void)
|
||||||
|
|
@ -266,7 +290,9 @@ export const layer = Layer.effect(
|
||||||
}),
|
}),
|
||||||
tools: Effect.fn("MCP.tools")(function* () {
|
tools: Effect.fn("MCP.tools")(function* () {
|
||||||
yield* whenAllReady
|
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* () {
|
instructions: Effect.fn("MCP.instructions")(function* () {
|
||||||
yield* whenAllReady
|
yield* whenAllReady
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue