fix(opencode): paginate MCP catalogs (#31442)
This commit is contained in:
@@ -112,6 +112,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) {
|
||||
if (URL.canParse(value)) return new URL(value)
|
||||
@@ -123,32 +124,43 @@ function isOutputSchemaValidationError(error: Error) {
|
||||
)
|
||||
}
|
||||
|
||||
function listTools(key: string, client: MCPClient, timeout: number) {
|
||||
return Effect.tryPromise({
|
||||
try: () => client.listTools(undefined, { timeout }),
|
||||
catch: (err) => (err instanceof Error ? err : new Error(String(err))),
|
||||
}).pipe(
|
||||
Effect.map((result) => result.tools),
|
||||
Effect.catch((error) => {
|
||||
if (!isOutputSchemaValidationError(error)) return Effect.fail(error)
|
||||
async function paginate<T, R extends { nextCursor?: string }>(
|
||||
list: (cursor?: string) => Promise<R>,
|
||||
items: (result: R) => T[],
|
||||
) {
|
||||
const result: T[] = []
|
||||
const cursors = new Set<string>()
|
||||
let cursor: string | undefined
|
||||
|
||||
return Effect.tryPromise({
|
||||
try: () =>
|
||||
client.request({ method: "tools/list" }, TolerantListToolsResultSchema, {
|
||||
timeout,
|
||||
}),
|
||||
catch: (err) => (err instanceof Error ? err : new Error(String(err))),
|
||||
}).pipe(
|
||||
Effect.map((result) =>
|
||||
result.tools.map((tool) => ({
|
||||
name: tool.name,
|
||||
description: tool.description,
|
||||
inputSchema: tool.inputSchema,
|
||||
})),
|
||||
),
|
||||
)
|
||||
}),
|
||||
)
|
||||
for (let page = 0; page < MAX_LIST_PAGES; page++) {
|
||||
const page = await list(cursor)
|
||||
result.push(...items(page))
|
||||
if (page.nextCursor === undefined) return result
|
||||
if (cursors.has(page.nextCursor)) throw new Error(`MCP list returned duplicate cursor: ${page.nextCursor}`)
|
||||
cursors.add(page.nextCursor)
|
||||
cursor = page.nextCursor
|
||||
}
|
||||
|
||||
throw new Error(`MCP list exceeded ${MAX_LIST_PAGES} pages`)
|
||||
}
|
||||
|
||||
function listTools(client: MCPClient, timeout: number) {
|
||||
return Effect.tryPromise({
|
||||
try: () =>
|
||||
paginate(
|
||||
async (cursor) => {
|
||||
const params = cursor === undefined ? undefined : { cursor }
|
||||
try {
|
||||
return await client.listTools(params, { timeout })
|
||||
} catch (error) {
|
||||
if (!(error instanceof Error) || !isOutputSchemaValidationError(error)) throw error
|
||||
return client.request({ method: "tools/list", params }, TolerantListToolsResultSchema, { timeout })
|
||||
}
|
||||
},
|
||||
(result) => result.tools,
|
||||
),
|
||||
catch: (err) => (err instanceof Error ? err : new Error(String(err))),
|
||||
})
|
||||
}
|
||||
|
||||
// Convert MCP tool definition to AI SDK Tool type
|
||||
@@ -182,8 +194,8 @@ function convertMcpTool(mcpTool: MCPToolDef, client: MCPClient, timeout?: number
|
||||
})
|
||||
}
|
||||
|
||||
function defs(key: string, client: MCPClient, timeout?: number) {
|
||||
return listTools(key, client, timeout ?? DEFAULT_TIMEOUT).pipe(
|
||||
function defs(client: MCPClient, timeout?: number) {
|
||||
return listTools(client, timeout ?? DEFAULT_TIMEOUT).pipe(
|
||||
Effect.catch((err) => {
|
||||
return Effect.succeed(undefined)
|
||||
}),
|
||||
@@ -448,7 +460,7 @@ export const layer = Layer.effect(
|
||||
return { status } satisfies CreateResult
|
||||
}
|
||||
|
||||
const listed = mcpClient.getServerCapabilities()?.tools ? yield* defs(key, mcpClient, mcp.timeout) : []
|
||||
const listed = mcpClient.getServerCapabilities()?.tools ? yield* defs(mcpClient, mcp.timeout) : []
|
||||
if (!listed) {
|
||||
yield* Effect.tryPromise(() => mcpClient.close()).pipe(Effect.ignore)
|
||||
return { status: { status: "failed", error: "Failed to get tools" } } satisfies CreateResult
|
||||
@@ -487,7 +499,7 @@ export const layer = Layer.effect(
|
||||
client.setNotificationHandler(ToolListChangedNotificationSchema, async () => {
|
||||
if (s.clients[name] !== client || s.status[name]?.status !== "connected") return
|
||||
|
||||
const listed = await bridge.promise(defs(name, client, timeout))
|
||||
const listed = await bridge.promise(defs(client, timeout))
|
||||
if (!listed) return
|
||||
if (s.clients[name] !== client || s.status[name]?.status !== "connected") return
|
||||
|
||||
@@ -693,7 +705,13 @@ export const layer = Layer.effect(
|
||||
const s = yield* InstanceState.get(state)
|
||||
return yield* collectFromConnected(
|
||||
s,
|
||||
(c) => (c.getServerCapabilities()?.prompts ? c.listPrompts().then((r) => r.prompts) : Promise.resolve([])),
|
||||
(c) =>
|
||||
c.getServerCapabilities()?.prompts
|
||||
? paginate(
|
||||
(cursor) => c.listPrompts(cursor === undefined ? undefined : { cursor }),
|
||||
(result) => result.prompts,
|
||||
)
|
||||
: Promise.resolve([]),
|
||||
"prompts",
|
||||
)
|
||||
})
|
||||
@@ -703,7 +721,12 @@ export const layer = Layer.effect(
|
||||
return yield* collectFromConnected(
|
||||
s,
|
||||
(c) =>
|
||||
c.getServerCapabilities()?.resources ? c.listResources().then((r) => r.resources) : Promise.resolve([]),
|
||||
c.getServerCapabilities()?.resources
|
||||
? paginate(
|
||||
(cursor) => c.listResources(cursor === undefined ? undefined : { cursor }),
|
||||
(result) => result.resources,
|
||||
)
|
||||
: Promise.resolve([]),
|
||||
"resources",
|
||||
)
|
||||
})
|
||||
@@ -830,7 +853,7 @@ export const layer = Layer.effect(
|
||||
|
||||
const listed = client
|
||||
? client.getServerCapabilities()?.tools
|
||||
? yield* defs(mcpName, client, mcpConfig.timeout)
|
||||
? yield* defs(client, mcpConfig.timeout)
|
||||
: []
|
||||
: undefined
|
||||
if (!client || !listed) {
|
||||
|
||||
Reference in New Issue
Block a user