diff --git a/packages/core/src/location-services.ts b/packages/core/src/location-services.ts index 3447204bd0..93efe2ea17 100644 --- a/packages/core/src/location-services.ts +++ b/packages/core/src/location-services.ts @@ -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, diff --git a/packages/core/src/mcp/client.ts b/packages/core/src/mcp/client.ts index 68f55104dd..4eaf7d04e2 100644 --- a/packages/core/src/mcp/client.ts +++ b/packages/core/src/mcp/client.ts @@ -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 +} + /** 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 + /** Invokes a tool on the server. Interruption aborts the in-flight request. */ + readonly callTool: (input: { + readonly name: string + readonly args?: Record + }) => 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 @@ -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( const collected: T[] = [] const seen = new Set() 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) => diff --git a/packages/core/src/mcp/guidance.ts b/packages/core/src/mcp/guidance.ts index 1950cde9ed..7aa73f63df 100644 --- a/packages/core/src/mcp/guidance.ts +++ b/packages/core/src/mcp/guidance.ts @@ -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 })) diff --git a/packages/core/src/mcp/index.ts b/packages/core/src/mcp/index.ts index 8edbcd3476..3774e31659 100644 --- a/packages/core/src/mcp/index.ts +++ b/packages/core/src/mcp/index.ts @@ -64,6 +64,20 @@ export class Tool extends Schema.Class("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("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("MCP.PromptArgument")({ name: Schema.String, description: Schema.String.pipe(Schema.optional), @@ -135,6 +149,12 @@ export class NotFoundError extends Schema.TaggedErrorClass()("MCP server: ServerName, }) {} +export class ToolCallError extends Schema.TaggedErrorClass()("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 readonly tools: () => Effect.Effect + readonly callTool: (input: { + readonly server: ServerName | string + readonly name: string + readonly args?: Record + }) => Effect.Effect readonly instructions: () => Effect.Effect readonly prompts: () => Effect.Effect 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) diff --git a/packages/core/src/tool/mcp.ts b/packages/core/src/tool/mcp.ts new file mode 100644 index 0000000000..571b747bab --- /dev/null +++ b/packages/core/src/tool/mcp.ts @@ -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 `_` 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) => + 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 }).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() + const record: Record = {} + 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], +}) diff --git a/packages/core/src/tool/tool.ts b/packages/core/src/tool/tool.ts index 1d9a82e952..3c394f13c8 100644 --- a/packages/core/src/tool/tool.ts +++ b/packages/core/src/tool/tool.ts @@ -60,6 +60,23 @@ type Config< }) => ReadonlyArray } +export type DynamicOutput = { + readonly structured: unknown + readonly content: ReadonlyArray +} + +/** + * 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 +} + type Runtime = { readonly permission?: string readonly definition: (name: string) => ToolDefinition @@ -72,6 +89,17 @@ export function make< Input extends SchemaType, Output extends SchemaType, Structured extends SchemaType = Output, +>(config: Config): Definition +export function make(config: DynamicConfig): AnyTool +export function make(config: Config | DynamicConfig): AnyTool { + if ("jsonSchema" in config) return makeDynamic(config) + return makeTyped(config) +} + +function makeTyped< + Input extends SchemaType, + Output extends SchemaType, + Structured extends SchemaType = Output, >(config: Config): Definition { const tool = Object.freeze({}) as Definition const definitions = new Map() @@ -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() + 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