feat(core): invoke and register mcp server tools
Add MCP.callTool plus a dynamic jsonSchema mode for Tool.make, and register each MCP server's tools as canonical Location-scoped tools via an McpTool producer that reconciles on tools-changed notifications. Content-first tool results preserve image/audio blocks; remove the arbitrary pagination page cap; log mcp connection outcomes. Tool naming keeps v1 parity (server_tool) so existing deny rules apply.
This commit is contained in:
parent
5619595abc
commit
73eb6ee473
6 changed files with 262 additions and 14 deletions
|
|
@ -37,6 +37,7 @@ import { Snapshot } from "./snapshot"
|
|||
import { SystemContextBuiltIns } from "./system-context/builtins"
|
||||
import { SystemContextRegistry } from "./system-context/registry"
|
||||
import { BuiltInTools } from "./tool/builtins"
|
||||
import { McpTool } from "./tool/mcp"
|
||||
import { ReadToolFileSystem } from "./tool/read-filesystem"
|
||||
import { ToolRegistry } from "./tool/registry"
|
||||
import { ToolOutputStore } from "./tool-output-store"
|
||||
|
|
@ -80,6 +81,7 @@ export const locationServices = LayerNode.group([
|
|||
Generate.node,
|
||||
ReadToolFileSystem.node,
|
||||
BuiltInTools.node,
|
||||
McpTool.node,
|
||||
SessionRunnerModel.node,
|
||||
SessionCompaction.node,
|
||||
Snapshot.node,
|
||||
|
|
|
|||
|
|
@ -7,6 +7,7 @@ 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 {
|
||||
CallToolResultSchema,
|
||||
ListRootsRequestSchema,
|
||||
ListToolsResultSchema,
|
||||
ToolListChangedNotificationSchema,
|
||||
|
|
@ -18,7 +19,6 @@ 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.
|
||||
|
|
@ -41,12 +41,27 @@ export interface ToolDefinition {
|
|||
readonly inputSchema: unknown
|
||||
}
|
||||
|
||||
export type CallToolContent =
|
||||
| { readonly type: "text"; readonly text: string }
|
||||
| { readonly type: "media"; readonly data: string; readonly mimeType: string }
|
||||
|
||||
export interface CallToolResult {
|
||||
readonly isError: boolean
|
||||
readonly structured: unknown
|
||||
readonly content: ReadonlyArray<CallToolContent>
|
||||
}
|
||||
|
||||
/** 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<ToolDefinition[], Error>
|
||||
/** Invokes a tool on the server. Interruption aborts the in-flight request. */
|
||||
readonly callTool: (input: {
|
||||
readonly name: string
|
||||
readonly args?: Record<string, unknown>
|
||||
}) => Effect.Effect<CallToolResult, Error>
|
||||
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
|
||||
|
|
@ -129,6 +144,37 @@ export const connect = Effect.fnUntraced(function* (
|
|||
inputSchema: tool.inputSchema,
|
||||
}))
|
||||
}),
|
||||
callTool: (input) =>
|
||||
Effect.tryPromise({
|
||||
try: (signal) =>
|
||||
client.callTool(
|
||||
{ name: input.name, arguments: input.args ?? {} },
|
||||
CallToolResultSchema,
|
||||
// The SDK only sends a progress token when onprogress is present, which enables timeout resets.
|
||||
{ signal, timeout: requestTimeout, resetTimeoutOnProgress: true, onprogress: () => {} },
|
||||
),
|
||||
catch: (error) => (error instanceof Error ? error : new Error(String(error))),
|
||||
}).pipe(
|
||||
Effect.map((result) => ({
|
||||
isError: result.isError === true,
|
||||
structured: result.structuredContent,
|
||||
content: result.content.flatMap((part): CallToolContent[] => {
|
||||
if (part.type === "text") return [{ type: "text", text: part.text }]
|
||||
if (part.type === "image" || part.type === "audio")
|
||||
return [{ type: "media", data: part.data, mimeType: part.mimeType }]
|
||||
if (part.type === "resource_link") return [{ type: "text", text: part.uri }]
|
||||
if (part.type === "resource") {
|
||||
const resource = part.resource
|
||||
if ("text" in resource && typeof resource.text === "string")
|
||||
return [{ type: "text", text: resource.text }]
|
||||
if ("blob" in resource && typeof resource.blob === "string" && typeof resource.mimeType === "string")
|
||||
return [{ type: "media", data: resource.blob, mimeType: resource.mimeType }]
|
||||
return [{ type: "text", text: resource.uri }]
|
||||
}
|
||||
return []
|
||||
}),
|
||||
})),
|
||||
),
|
||||
onClose: (callback) => {
|
||||
client.onclose = callback
|
||||
},
|
||||
|
|
@ -153,15 +199,15 @@ async function paginate<R extends { nextCursor?: string }, T>(
|
|||
const collected: T[] = []
|
||||
const seen = new Set<string>()
|
||||
let cursor: string | undefined
|
||||
for (let page = 0; page < MAX_LIST_PAGES; page++) {
|
||||
while (true) {
|
||||
const result = await list(cursor)
|
||||
collected.push(...items(result))
|
||||
if (result.nextCursor === undefined) return collected
|
||||
// A repeating cursor never terminates; bail instead of hanging the connection forever.
|
||||
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) =>
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import { makeLocationNode } from "../effect/app-node"
|
|||
import { Context, Effect, Layer, Schema } from "effect"
|
||||
import { AgentV2 } from "../agent"
|
||||
import { PermissionV2 } from "../permission"
|
||||
import { McpTool } from "../tool/mcp"
|
||||
import { MCP } from "./index"
|
||||
import { SystemContext } from "../system-context/index"
|
||||
|
||||
|
|
@ -48,7 +49,9 @@ export const layer = Layer.effect(
|
|||
const owned = tools.filter((tool) => tool.server === item.server)
|
||||
return (
|
||||
owned.length === 0 ||
|
||||
owned.some((tool) => PermissionV2.evaluate(tool.name, "*", agent.permissions).effect !== "deny")
|
||||
owned.some(
|
||||
(tool) => PermissionV2.evaluate(McpTool.name(tool.server, tool.name), "*", agent.permissions).effect !== "deny",
|
||||
)
|
||||
)
|
||||
})
|
||||
.map((item) => ({ server: item.server, instructions: item.instructions }))
|
||||
|
|
|
|||
|
|
@ -64,6 +64,20 @@ export class Tool extends Schema.Class<Tool>("MCP.Tool")({
|
|||
inputSchema: Schema.Unknown.pipe(Schema.optional),
|
||||
}) {}
|
||||
|
||||
export const ToolResultContent = Schema.Union([
|
||||
Schema.Struct({ type: Schema.Literal("text"), text: Schema.String }),
|
||||
Schema.Struct({ type: Schema.Literal("media"), data: Schema.String, mimeType: Schema.String }),
|
||||
]).pipe(Schema.toTaggedUnion("type"))
|
||||
export type ToolResultContent = typeof ToolResultContent.Type
|
||||
|
||||
export class ToolResult extends Schema.Class<ToolResult>("MCP.ToolResult")({
|
||||
server: ServerName,
|
||||
tool: Schema.String,
|
||||
isError: Schema.Boolean,
|
||||
structured: Schema.Unknown.pipe(Schema.optional),
|
||||
content: Schema.Array(ToolResultContent),
|
||||
}) {}
|
||||
|
||||
export class PromptArgument extends Schema.Class<PromptArgument>("MCP.PromptArgument")({
|
||||
name: Schema.String,
|
||||
description: Schema.String.pipe(Schema.optional),
|
||||
|
|
@ -135,6 +149,12 @@ export class NotFoundError extends Schema.TaggedErrorClass<NotFoundError>()("MCP
|
|||
server: ServerName,
|
||||
}) {}
|
||||
|
||||
export class ToolCallError extends Schema.TaggedErrorClass<ToolCallError>()("MCP.ToolCallError", {
|
||||
server: ServerName,
|
||||
tool: Schema.String,
|
||||
message: Schema.String,
|
||||
}) {}
|
||||
|
||||
type ServerEntry = {
|
||||
readonly config: typeof ConfigMCP.Server.Type
|
||||
status: Status
|
||||
|
|
@ -149,6 +169,11 @@ type ServerEntry = {
|
|||
export interface Interface {
|
||||
readonly servers: () => Effect.Effect<ServerInfo[]>
|
||||
readonly tools: () => Effect.Effect<Tool[]>
|
||||
readonly callTool: (input: {
|
||||
readonly server: ServerName | string
|
||||
readonly name: string
|
||||
readonly args?: Record<string, unknown>
|
||||
}) => Effect.Effect<ToolResult, NotFoundError | ToolCallError>
|
||||
readonly instructions: () => Effect.Effect<ServerInstructions[]>
|
||||
readonly prompts: () => Effect.Effect<Prompt[]>
|
||||
readonly prompt: (input: {
|
||||
|
|
@ -252,6 +277,7 @@ export const layer = Layer.effect(
|
|||
entry.tools = result.value.defs.map((def) => toTool(name, def))
|
||||
entry.status = { status: "connected" }
|
||||
watch(name, entry, result.value.connection)
|
||||
yield* Effect.logInfo("mcp connected", { server: name, tools: entry.tools.length })
|
||||
return
|
||||
}
|
||||
yield* Scope.close(scope, Exit.void)
|
||||
|
|
@ -261,6 +287,7 @@ export const layer = Layer.effect(
|
|||
error instanceof MCPClient.NeedsAuthError
|
||||
? { status: "needs_auth" }
|
||||
: { status: "failed", error: error instanceof Error ? error.message : String(error) }
|
||||
yield* Effect.logWarning("mcp connect failed", { server: name, status: entry.status })
|
||||
}).pipe(Effect.ensuring(Deferred.succeed(entry.startup, undefined)))
|
||||
|
||||
// Disabled servers settle their startup immediately so queries never block on them.
|
||||
|
|
@ -294,6 +321,26 @@ export const layer = Layer.effect(
|
|||
.flatMap((entry) => entry.tools ?? [])
|
||||
.toSorted((a, b) => a.server.localeCompare(b.server) || a.name.localeCompare(b.name))
|
||||
}),
|
||||
callTool: Effect.fn("MCP.callTool")(function* (input) {
|
||||
const target = yield* requireServer(input.server)
|
||||
yield* Deferred.await(target.entry.startup)
|
||||
if (!target.entry.client)
|
||||
return yield* new ToolCallError({
|
||||
server: target.name,
|
||||
tool: input.name,
|
||||
message: "MCP server is not connected",
|
||||
})
|
||||
const result = yield* target.entry.client
|
||||
.callTool({ name: input.name, args: input.args })
|
||||
.pipe(Effect.mapError((error) => new ToolCallError({ server: target.name, tool: input.name, message: error.message })))
|
||||
return new ToolResult({
|
||||
server: target.name,
|
||||
tool: input.name,
|
||||
isError: result.isError,
|
||||
structured: result.structured,
|
||||
content: result.content,
|
||||
})
|
||||
}),
|
||||
instructions: Effect.fn("MCP.instructions")(function* () {
|
||||
yield* whenAllReady
|
||||
return Array.from(runtime)
|
||||
|
|
|
|||
101
packages/core/src/tool/mcp.ts
Normal file
101
packages/core/src/tool/mcp.ts
Normal file
|
|
@ -0,0 +1,101 @@
|
|||
export * as McpTool from "./mcp"
|
||||
|
||||
import { createHash } from "node:crypto"
|
||||
import { ToolFailure } from "@opencode-ai/llm"
|
||||
import { McpEvent } from "@opencode-ai/schema/mcp-event"
|
||||
import { Effect, Exit, type JsonSchema, Layer, Scope, Semaphore, Stream } from "effect"
|
||||
import { makeLocationNode } from "../effect/app-node"
|
||||
import { EventV2 } from "../event"
|
||||
import { MCP } from "../mcp"
|
||||
import { Tool } from "./tool"
|
||||
import { Tools } from "./tools"
|
||||
import { ToolRegistry } from "./registry"
|
||||
|
||||
const MAX_NAME_LENGTH = 64
|
||||
const HASH_LENGTH = 8
|
||||
|
||||
const sanitize = (value: string) => value.replace(/[^A-Za-z0-9_-]/g, "_")
|
||||
|
||||
// Deterministic short suffix used to keep overlong or colliding names unique and stable across restarts.
|
||||
const hashSuffix = (raw: string) => "_" + createHash("sha1").update(raw).digest("hex").slice(0, HASH_LENGTH)
|
||||
|
||||
const fit = (base: string, raw: string) => base.slice(0, MAX_NAME_LENGTH - HASH_LENGTH - 1) + hashSuffix(raw)
|
||||
|
||||
/**
|
||||
* Registry/permission action name for an MCP tool: V1-compatible `<server>_<tool>` so existing deny
|
||||
* rules keep working. Sanitized to a valid tool name, prefixed when it would not start with a letter,
|
||||
* and hashed down when it would exceed the 64-char limit.
|
||||
*/
|
||||
export const name = (server: string, tool: string) => {
|
||||
const joined = sanitize(server) + "_" + sanitize(tool)
|
||||
const base = /^[A-Za-z]/.test(joined) ? joined : "mcp_" + joined
|
||||
return base.length > MAX_NAME_LENGTH ? fit(base, `${server}\u0000${tool}`) : base
|
||||
}
|
||||
|
||||
const toContent = (part: MCP.ToolResultContent): Tool.Content =>
|
||||
part.type === "text" ? { type: "text", text: part.text } : { type: "file", data: part.data, mime: part.mimeType }
|
||||
|
||||
const errorText = (content: ReadonlyArray<MCP.ToolResultContent>) =>
|
||||
content
|
||||
.flatMap((part) => (part.type === "text" ? [part.text] : []))
|
||||
.join("\n")
|
||||
.trim()
|
||||
|
||||
export const layer = Layer.effectDiscard(
|
||||
Effect.gen(function* () {
|
||||
const mcp = yield* MCP.Service
|
||||
const tools = yield* Tools.Service
|
||||
const events = yield* EventV2.Service
|
||||
const scope = yield* Scope.Scope
|
||||
const lock = Semaphore.makeUnsafe(1)
|
||||
let current: Scope.Closeable | undefined
|
||||
|
||||
const make = (server: MCP.ServerName, tool: MCP.Tool) =>
|
||||
Tool.make({
|
||||
description: tool.description ?? "",
|
||||
jsonSchema: (tool.inputSchema as JsonSchema.JsonSchema | undefined) ?? { type: "object", properties: {} },
|
||||
execute: (input) =>
|
||||
Effect.gen(function* () {
|
||||
const result = yield* mcp.callTool({ server, name: tool.name, args: (input ?? {}) as Record<string, unknown> }).pipe(
|
||||
Effect.catchTags({
|
||||
"MCP.NotFoundError": (error) => new ToolFailure({ message: `MCP server "${error.server}" is not available` }),
|
||||
"MCP.ToolCallError": (error) => new ToolFailure({ message: error.message }),
|
||||
}),
|
||||
)
|
||||
if (result.isError)
|
||||
return yield* new ToolFailure({ message: errorText(result.content) || "MCP tool returned an error" })
|
||||
return { structured: result.structured ?? {}, content: result.content.map(toContent) }
|
||||
}),
|
||||
})
|
||||
|
||||
// Register the current tool set under a fresh child scope, then close the previous one so the
|
||||
// registry never has a gap where MCP tools disappear mid-swap.
|
||||
const reconcile = lock.withPermit(
|
||||
Effect.gen(function* () {
|
||||
const used = new Set<string>()
|
||||
const record: Record<string, Tool.AnyTool> = {}
|
||||
for (const tool of yield* mcp.tools()) {
|
||||
const initial = name(tool.server, tool.name)
|
||||
const key = used.has(initial) ? fit(initial, `${tool.server}\u0000${tool.name}`) : initial
|
||||
used.add(key)
|
||||
record[key] = make(tool.server, tool)
|
||||
}
|
||||
const next = yield* Scope.fork(scope)
|
||||
yield* tools.register(record).pipe(Scope.provide(next), Effect.orDie)
|
||||
if (current) yield* Scope.close(current, Exit.void)
|
||||
current = next
|
||||
}),
|
||||
)
|
||||
|
||||
yield* reconcile.pipe(Effect.forkScoped)
|
||||
yield* events
|
||||
.subscribe(McpEvent.ToolsChanged)
|
||||
.pipe(Stream.runForEach(() => reconcile), Effect.forkScoped({ startImmediately: true }))
|
||||
}),
|
||||
)
|
||||
|
||||
export const node = makeLocationNode({
|
||||
name: "mcp-tools",
|
||||
layer,
|
||||
deps: [ToolRegistry.toolsNode, MCP.node, EventV2.node],
|
||||
})
|
||||
|
|
@ -60,6 +60,23 @@ type Config<
|
|||
}) => ReadonlyArray<Content>
|
||||
}
|
||||
|
||||
export type DynamicOutput = {
|
||||
readonly structured: unknown
|
||||
readonly content: ReadonlyArray<Content>
|
||||
}
|
||||
|
||||
/**
|
||||
* Config for a tool whose input shape is a raw JSON Schema not known at compile
|
||||
* time (MCP servers, plugin manifests). Input is passed through as `unknown`;
|
||||
* `execute` returns the already-projected structured value and model content.
|
||||
*/
|
||||
type DynamicConfig = {
|
||||
readonly description: string
|
||||
readonly jsonSchema: JsonSchema.JsonSchema
|
||||
readonly outputSchema?: JsonSchema.JsonSchema
|
||||
readonly execute: (input: unknown, context: Context) => Effect.Effect<DynamicOutput, ToolFailure>
|
||||
}
|
||||
|
||||
type Runtime = {
|
||||
readonly permission?: string
|
||||
readonly definition: (name: string) => ToolDefinition
|
||||
|
|
@ -72,6 +89,17 @@ export function make<
|
|||
Input extends SchemaType<any>,
|
||||
Output extends SchemaType<any>,
|
||||
Structured extends SchemaType<any> = Output,
|
||||
>(config: Config<Input, Output, Structured>): Definition<Input, Structured>
|
||||
export function make(config: DynamicConfig): AnyTool
|
||||
export function make(config: Config<any, any, any> | DynamicConfig): AnyTool {
|
||||
if ("jsonSchema" in config) return makeDynamic(config)
|
||||
return makeTyped(config)
|
||||
}
|
||||
|
||||
function makeTyped<
|
||||
Input extends SchemaType<any>,
|
||||
Output extends SchemaType<any>,
|
||||
Structured extends SchemaType<any> = Output,
|
||||
>(config: Config<Input, Output, Structured>): Definition<Input, Structured> {
|
||||
const tool = Object.freeze({}) as Definition<Input, Structured>
|
||||
const definitions = new Map<string, ToolDefinition>()
|
||||
|
|
@ -113,16 +141,8 @@ export function make<
|
|||
Effect.map(({ output, structured }) => ({
|
||||
structured,
|
||||
content:
|
||||
config.toModelOutput?.({ input, output }).map((part) =>
|
||||
part.type === "text"
|
||||
? { type: "text" as const, text: part.text }
|
||||
: {
|
||||
type: "file" as const,
|
||||
uri: `data:${part.mime};base64,${part.data}`,
|
||||
mime: part.mime,
|
||||
name: part.name,
|
||||
},
|
||||
) ?? (typeof output === "string" ? [{ type: "text" as const, text: output }] : []),
|
||||
config.toModelOutput?.({ input, output }).map(toModelContent) ??
|
||||
(typeof output === "string" ? [{ type: "text" as const, text: output }] : []),
|
||||
})),
|
||||
),
|
||||
),
|
||||
|
|
@ -131,6 +151,35 @@ export function make<
|
|||
return tool
|
||||
}
|
||||
|
||||
function makeDynamic(config: DynamicConfig): AnyTool {
|
||||
const tool = Object.freeze({}) as AnyTool
|
||||
const definitions = new Map<string, ToolDefinition>()
|
||||
runtimes.set(tool, {
|
||||
definition: (name) => {
|
||||
const cached = definitions.get(name)
|
||||
if (cached) return cached
|
||||
const definition = new ToolDefinition({
|
||||
name,
|
||||
description: config.description,
|
||||
inputSchema: config.jsonSchema,
|
||||
outputSchema: config.outputSchema,
|
||||
})
|
||||
definitions.set(name, definition)
|
||||
return definition
|
||||
},
|
||||
settle: (call, context) =>
|
||||
config
|
||||
.execute(call.input, context)
|
||||
.pipe(Effect.map((output) => ({ structured: output.structured, content: output.content.map(toModelContent) }))),
|
||||
})
|
||||
return tool
|
||||
}
|
||||
|
||||
function toModelContent(part: Content) {
|
||||
if (part.type === "text") return { type: "text" as const, text: part.text }
|
||||
return { type: "file" as const, uri: `data:${part.mime};base64,${part.data}`, mime: part.mime, name: part.name }
|
||||
}
|
||||
|
||||
export const validateName = (name: string) =>
|
||||
/^[A-Za-z][A-Za-z0-9_-]{0,63}$/.test(name)
|
||||
? Effect.void
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue