diff --git a/packages/core/src/mcp/client.ts b/packages/core/src/mcp/client.ts index 0df53d160d..e0064a28d5 100644 --- a/packages/core/src/mcp/client.ts +++ b/packages/core/src/mcp/client.ts @@ -22,11 +22,12 @@ export class ConnectError extends Schema.TaggedErrorClass()("MCP.C message: Schema.String, }) {} -/** - * Connects an MCP client and registers a scoped finalizer that closes it. The - * returned client is owned by the calling scope; closing that scope tears down - * the transport (and, for local servers, the spawned process). - */ +/** Handle over a connected MCP server that keeps the SDK `Client` out of the rest of core. */ +export interface Connection { + readonly onClose: (callback: () => void) => void +} + +/** Connects an MCP server; closing the calling scope tears down the transport and any spawned process. */ export const connect = Effect.fnUntraced(function* ( server: string, config: typeof ConfigMCP.Server.Type, @@ -71,7 +72,11 @@ export const connect = Effect.fnUntraced(function* ( }).pipe(Effect.exit) if (Exit.isSuccess(exit)) { yield* Effect.addFinalizer(() => Effect.promise(() => client.close()).pipe(Effect.ignore)) - return client + return { + onClose: (callback) => { + client.onclose = callback + }, + } satisfies Connection } yield* Effect.promise(() => transport.close()).pipe(Effect.ignore) diff --git a/packages/core/src/mcp/index.ts b/packages/core/src/mcp/index.ts index 49bcedb0e4..02bfd8268a 100644 --- a/packages/core/src/mcp/index.ts +++ b/packages/core/src/mcp/index.ts @@ -1,11 +1,15 @@ export * as MCP from "./index" -import { Context, Effect, Layer, Schema } from "effect" +import { McpEvent } from "@opencode-ai/schema/mcp-event" +import { Cause, Context, Deferred, Effect, Exit, FiberSet, Layer, Schema, Scope } from "effect" import { makeLocationNode } from "../effect/app-node" import { Config } from "../config" import { ConfigMCP } from "../config/mcp" +import { EventV2 } from "../event" import { Integration } from "../integration" import { IntegrationConnection } from "../integration/connection" +import { Location } from "../location" +import { MCPClient } from "./client" export const ServerName = Schema.String.pipe(Schema.brand("MCP.ServerName")) export type ServerName = typeof ServerName.Type @@ -54,6 +58,13 @@ export class ServerInstructions extends Schema.Class("MCP.Se tools: Schema.Array(Schema.String), }) {} +export class Tool extends Schema.Class("MCP.Tool")({ + server: ServerName, + name: Schema.String, + description: Schema.String.pipe(Schema.optional), + inputSchema: Schema.Unknown.pipe(Schema.optional), +}) {} + export class PromptArgument extends Schema.Class("MCP.PromptArgument")({ name: Schema.String, description: Schema.String.pipe(Schema.optional), @@ -127,26 +138,25 @@ export class NotFoundError extends Schema.TaggedErrorClass()("MCP type ServerEntry = { readonly config: typeof ConfigMCP.Server.Type - readonly status: Status + status: Status + readonly startup: Deferred.Deferred + scope?: Scope.Closeable + client?: MCPClient.Connection readonly integrationID?: Integration.ID readonly connection?: IntegrationConnection.Info } export interface Interface { readonly servers: () => Effect.Effect - readonly add: (server: ServerName | string, config: typeof ConfigMCP.Server.Type) => Effect.Effect - readonly connect: (server: ServerName | string) => Effect.Effect - readonly disconnect: (server: ServerName | string) => Effect.Effect + readonly tools: () => Effect.Effect readonly instructions: () => Effect.Effect - readonly prompts: (input?: { readonly server?: ServerName | string }) => Effect.Effect + readonly prompts: () => Effect.Effect readonly prompt: (input: { readonly server: ServerName | string readonly name: string readonly args?: Record }) => Effect.Effect - readonly resourceCatalog: (input?: { - readonly server?: ServerName | string - }) => Effect.Effect + readonly resourceCatalog: () => Effect.Effect readonly readResource: (input: { readonly server: ServerName | string readonly uri: string @@ -159,6 +169,12 @@ export const layer = Layer.effect( Service, Effect.gen(function* () { const config = yield* Config.Service + const location = yield* Location.Service + const events = yield* EventV2.Service + const root = yield* Scope.make() + const fork = yield* FiberSet.makeRuntime() + yield* Effect.addFinalizer((exit) => Scope.close(root, exit)) + const documents = (yield* config.entries()).filter((entry): entry is Config.Document => entry.type === "document") // Global MCP timeout defaults, later config files overriding earlier ones. const timeout = Object.assign( @@ -171,7 +187,8 @@ export const layer = Layer.effect( for (const [name, server] of Object.entries(entry.info.mcp?.servers ?? {})) { runtime.set(ServerName.make(name), { config: { ...server, timeout: { ...timeout, ...server.timeout } }, - status: server.disabled ? { status: "disabled" } : { status: "disconnected" }, + status: { status: "disconnected" }, + startup: Deferred.makeUnsafe(), }) } } @@ -183,62 +200,97 @@ export const layer = Layer.effect( return { name, entry } }) - const visibleStatus = (entry: ServerEntry): Status => (entry.config.disabled ? { status: "disabled" } : entry.status) - const info = (name: ServerName, entry: ServerEntry) => new ServerInfo({ name, config: entry.config, - status: visibleStatus(entry), + status: entry.status, integrationID: entry.integrationID, connection: entry.connection, }) + const watch = (name: ServerName, entry: ServerEntry, connection: MCPClient.Connection) => { + connection.onClose(() => { + entry.client = undefined + entry.status = { status: "failed", error: "Connection closed" } + fork(events.publish(McpEvent.ToolsChanged, { server: name }).pipe(Effect.ignore)) + }) + } + + const startServer = (name: ServerName, entry: ServerEntry) => + Effect.gen(function* () { + const scope = yield* Scope.fork(root) + entry.scope = scope + const result = yield* MCPClient.connect(name, entry.config, location.directory).pipe( + Scope.provide(scope), + Effect.exit, + ) + if (Exit.isSuccess(result)) { + entry.client = result.value + entry.status = { status: "connected" } + watch(name, entry, result.value) + return + } + yield* Scope.close(scope, Exit.void) + entry.scope = undefined + const error = Cause.squash(result.cause) + entry.status = + error instanceof MCPClient.NeedsAuthError + ? { status: "needs_auth" } + : { status: "failed", error: error instanceof Error ? error.message : String(error) } + }).pipe(Effect.ensuring(Deferred.succeed(entry.startup, undefined))) + + // Disabled servers settle their startup immediately so queries never block on them. + for (const [name, entry] of runtime) { + if (entry.config.disabled) { + entry.status = { status: "disabled" } + Deferred.doneUnsafe(entry.startup, Exit.void) + continue + } + fork(startServer(name, entry)) + } + + const whenAllReady = Effect.forEach(runtime.values(), (entry) => Deferred.await(entry.startup), { + concurrency: "unbounded", + discard: true, + }) + const gate = Effect.fnUntraced(function* (server: ServerName | string) { + const target = yield* requireServer(server) + yield* Deferred.await(target.entry.startup) + }) + return Service.of({ servers: Effect.fn("MCP.servers")(function* () { return Array.from(runtime, ([name, entry]) => info(name, entry)).toSorted((a, b) => a.name.localeCompare(b.name), ) }), - add: Effect.fn("MCP.add")(function* (server, config) { - const name = ServerName.make(server) - const status: Status = config.disabled ? { status: "disabled" } : { status: "disconnected" } - const entry = { config, status } - runtime.set(name, entry) - return info(name, entry) - }), - connect: Effect.fn("MCP.connect")(function* (server) { - const current = yield* requireServer(server) - return info(current.name, current.entry) - }), - disconnect: Effect.fn("MCP.disconnect")(function* (server) { - const current = yield* requireServer(server) - const status: Status = current.entry.config.disabled ? { status: "disabled" } : { status: "disconnected" } - const entry = { ...current.entry, status } - runtime.set(current.name, entry) - return info(current.name, entry) - }), - instructions: Effect.fn("MCP.instructions")(function* () { + tools: Effect.fn("MCP.tools")(function* () { + yield* whenAllReady return [] }), - prompts: Effect.fn("MCP.prompts")(function* (input) { - if (input?.server !== undefined) yield* requireServer(input.server) + instructions: Effect.fn("MCP.instructions")(function* () { + yield* whenAllReady + return [] + }), + prompts: Effect.fn("MCP.prompts")(function* () { + yield* whenAllReady return [] }), prompt: Effect.fn("MCP.prompt")(function* (input) { - yield* requireServer(input.server) + yield* gate(input.server) return undefined }), - resourceCatalog: Effect.fn("MCP.resourceCatalog")(function* (input) { - if (input?.server !== undefined) yield* requireServer(input.server) + resourceCatalog: Effect.fn("MCP.resourceCatalog")(function* () { + yield* whenAllReady return new ResourceCatalog({ resources: [], templates: [] }) }), readResource: Effect.fn("MCP.readResource")(function* (input) { - yield* requireServer(input.server) + yield* gate(input.server) return undefined }), }) }), ) -export const node = makeLocationNode({ service: Service, layer, deps: [Config.node] }) +export const node = makeLocationNode({ service: Service, layer, deps: [Config.node, Location.node, EventV2.node] })