diff --git a/packages/core/src/mcp/index.ts b/packages/core/src/mcp/index.ts index 213a8b4ec7a..12f983e9f2f 100644 --- a/packages/core/src/mcp/index.ts +++ b/packages/core/src/mcp/index.ts @@ -3,10 +3,11 @@ export * as MCP from "./index" import { Mcp } from "@opencode-ai/schema/mcp" import { McpEvent } from "@opencode-ai/schema/mcp-event" import { Command } from "@opencode-ai/schema/command" -import { Document } from "@opencode-ai/schema/config" +import { Document, Event, type Entry } from "@opencode-ai/schema/config" import { ConfigMCP } from "@opencode-ai/schema/config/mcp" import { createHash } from "node:crypto" -import { Cause, Context, Deferred, Effect, Exit, FiberSet, Layer, Schema, Scope, Stream } from "effect" +import { isDeepStrictEqual } from "node:util" +import { Cause, Context, Deferred, Effect, Exit, FiberSet, Layer, Schema, Scope, Semaphore, Stream } from "effect" import { makeLocationNode } from "@opencode-ai/util/effect/app-node" import { Config } from "../config" import { Credential } from "../credential" @@ -180,26 +181,36 @@ export const layer = (options?: Options) => const fork = yield* FiberSet.makeRuntime() yield* Effect.addFinalizer((exit) => Scope.close(root, exit)) - const documents = (yield* config.entries()).filter((entry): entry is Document => entry.type === "document") - // Global MCP timeout defaults, later config files overriding earlier ones. - const timeout = Object.assign( - {}, - ...documents.flatMap((entry) => (entry.info.mcp?.timeout ? [entry.info.mcp.timeout] : [])), - ) + const loadConfig = (entries: readonly Entry[]) => { + const documents = entries.filter((entry): entry is Document => entry.type === "document") + // Global MCP timeout defaults, later config files overriding earlier ones. + const timeout = Object.assign( + {}, + ...documents.flatMap((entry) => (entry.info.mcp?.timeout ? [entry.info.mcp.timeout] : [])), + ) + const servers = new Map() + for (const entry of documents) { + for (const [name, server] of Object.entries(entry.info.mcp?.servers ?? {})) { + servers.set(ServerName.make(name), { ...server, timeout: { ...timeout, ...server.timeout } }) + } + } + return { timeout, servers } + } + const initial = loadConfig(yield* config.entries()) + const configState = { servers: initial.servers, timeout: initial.timeout } // Later config files win for duplicate server names; per-server timeout overrides globals. const runtime = new Map() // Serializes lifecycle operations per server. Anything taking this lock from a connection // callback must stay forked: lifecycle operations close scopes while holding it, firing onClose. const locks = KeyedMutex.makeUnsafe() + const reloadLock = Semaphore.makeUnsafe(1) const urlElicitations = new Map() - for (const entry of documents) { - for (const [name, server] of Object.entries(entry.info.mcp?.servers ?? {})) { - runtime.set(ServerName.make(name), { - config: { ...server, timeout: { ...timeout, ...server.timeout } }, - status: { status: "pending" }, - startup: Deferred.makeUnsafe(), - }) - } + for (const [name, server] of initial.servers) { + runtime.set(name, { + config: server, + status: { status: "pending" }, + startup: Deferred.makeUnsafe(), + }) } // Register every remote server as an OAuth integration so credentials live in the global store @@ -552,6 +563,68 @@ export const layer = (options?: Options) => yield* bus.publish(Command.Event.Updated, {}).pipe(Effect.ignore) }) + const disposeServer = Effect.fnUntraced(function* (name: ServerName, entry: ServerEntry) { + yield* stopServer(name, entry) + if (entry.integrationID) owned.delete(entry.integrationID) + if (entry.registration) yield* entry.registration.dispose + }) + + const replaceServer = Effect.fnUntraced(function* ( + name: ServerName, + serverConfig: typeof ConfigMCP.Server.Type, + ) { + const previous = runtime.get(name) + if (previous) yield* disposeServer(name, previous) + const entry: ServerEntry = { + config: serverConfig, + status: { status: "pending" }, + startup: Deferred.makeUnsafe(), + } + runtime.set(name, entry) + yield* Effect.gen(function* () { + yield* register(name, entry) + if (serverConfig.disabled) { + entry.status = { status: "disabled" } + yield* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore) + return + } + yield* startServer(name, entry) + }).pipe( + // Settle startup even when registration fails or replacement is interrupted, so readers cannot hang. + Effect.ensuring(Effect.sync(() => Deferred.doneUnsafe(entry.startup, Exit.void))), + ) + }) + + const removeServer = Effect.fnUntraced(function* (name: ServerName) { + const entry = runtime.get(name) + if (!entry) return + yield* disposeServer(name, entry) + // Credentials are keyed by name + URL and intentionally survive removal for a later re-add. + runtime.delete(name) + yield* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore) + }) + + const reloadConfig = Effect.fnUntraced(function* () { + yield* reloadLock.withPermit( + Effect.gen(function* () { + const next = loadConfig(yield* config.entries()) + const names = new Set([...configState.servers.keys(), ...next.servers.keys()]) + for (const name of names) { + const previous = configState.servers.get(name) + const updated = next.servers.get(name) + if (isDeepStrictEqual(previous, updated)) continue + if (!updated) { + yield* removeServer(name).pipe(locks.withLock(name)) + continue + } + yield* replaceServer(name, updated).pipe(locks.withLock(name)) + } + configState.servers = next.servers + configState.timeout = next.timeout + }), + ) + }) + // Disabled servers settle their startup immediately so queries never block on them. for (const [name, entry] of runtime) { if (entry.config.disabled) { @@ -585,6 +658,16 @@ export const layer = (options?: Options) => Effect.ignore, ), ) + yield* bus.subscribe(Event.Updated).pipe( + Stream.runForEach(() => + reloadConfig().pipe( + Effect.catchCause((cause) => Effect.logError("failed to reload MCP config", { cause })), + ), + ), + Effect.forkScoped({ startImmediately: true }), + ) + // Close the gap between the initial snapshot and the live subscription becoming active. + yield* reloadConfig() // Suspend so each await sees current entries; a bare Map iterator is exhausted after one run. const whenAllReady = Effect.suspend(() => @@ -601,33 +684,9 @@ export const layer = (options?: Options) => }), add: Effect.fn("MCP.add")(function* (server, config) { const name = ServerName.make(server) - yield* Effect.gen(function* () { - const previous = runtime.get(name) - if (previous) { - yield* stopServer(name, previous) - if (previous.integrationID) owned.delete(previous.integrationID) - if (previous.registration) yield* previous.registration.dispose - } - const entry: ServerEntry = { - config: { ...config, timeout: { ...timeout, ...config.timeout } }, - status: { status: "pending" }, - startup: Deferred.makeUnsafe(), - } - runtime.set(name, entry) - yield* Effect.gen(function* () { - yield* register(name, entry) - if (config.disabled) { - entry.status = { status: "disabled" } - yield* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore) - return - } - yield* startServer(name, entry) - }).pipe( - // Settle startup even when register fails or add is interrupted, so an entry that made it - // into runtime can never hang readers awaiting its startup. - Effect.ensuring(Effect.sync(() => Deferred.doneUnsafe(entry.startup, Exit.void))), - ) - }).pipe(locks.withLock(name)) + yield* replaceServer(name, { ...config, timeout: { ...configState.timeout, ...config.timeout } }).pipe( + locks.withLock(name), + ) }), connect: Effect.fn("MCP.connect")(function* (server) { const name = ServerName.make(server) @@ -649,14 +708,8 @@ export const layer = (options?: Options) => remove: Effect.fn("MCP.remove")(function* (server) { const name = ServerName.make(server) yield* Effect.gen(function* () { - const target = yield* requireServer(name) - yield* stopServer(name, target.entry) - if (target.entry.integrationID) owned.delete(target.entry.integrationID) - if (target.entry.registration) yield* target.entry.registration.dispose - // Credentials are kept: they are keyed by name + url, so re-adding the same server - // reuses them without forcing re-auth, matching add()'s replacement semantics. - runtime.delete(name) - yield* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore) + yield* requireServer(name) + yield* removeServer(name) }).pipe(locks.withLock(name)) }), tools: Effect.fn("MCP.tools")(function* () { diff --git a/packages/core/test/mcp.test.ts b/packages/core/test/mcp.test.ts index 7989a53734a..66b4eaafaba 100644 --- a/packages/core/test/mcp.test.ts +++ b/packages/core/test/mcp.test.ts @@ -12,14 +12,14 @@ import { ListToolsRequestSchema, ReadResourceRequestSchema, } from "@modelcontextprotocol/sdk/types.js" -import { Document, Info } from "@opencode-ai/schema/config" +import { Document, Event, Info } from "@opencode-ai/schema/config" import { ConfigMCP } from "@opencode-ai/schema/config/mcp" import { Config } from "@opencode-ai/core/config" import { Credential } from "@opencode-ai/core/credential" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" import { LayerNode } from "@opencode-ai/util/effect/layer-node" import { Bus } from "@opencode-ai/core/bus" -import { Event } from "@opencode-ai/schema/event" +import { ID, type Payload } from "@opencode-ai/schema/event" import { Form } from "@opencode-ai/core/form" import { Integration } from "@opencode-ai/core/integration" import { Location } from "@opencode-ai/core/location" @@ -30,7 +30,7 @@ import { AbsolutePath } from "@opencode-ai/core/schema" import { Session } from "@opencode-ai/core/session" import { McpTool } from "@opencode-ai/core/tool/mcp" import { Tool } from "@opencode-ai/core/tool" -import { Deferred, Effect, Exit, Fiber, Layer, Schema, Stream } from "effect" +import { DateTime, Deferred, Effect, Exit, Fiber, Layer, PubSub, Schedule, Schema, Stream } from "effect" import { Image } from "@opencode-ai/core/image" import { testEffect } from "./lib/effect" import { imagePassthrough } from "./lib/image" @@ -67,6 +67,8 @@ function resourceServer( ] as Array<{ uri: string; text: string; mimeType?: string } | { uri: string; blob: string; mimeType?: string }>, resourceLists: 0, templateLists: 0, + toolLists: 0, + initializations: 0, } const protocol = new Server( { name: "mcp-resources", version: "1.0.0" }, @@ -77,15 +79,16 @@ function resourceServer( }, }, ) - protocol.setRequestHandler(ListToolsRequestSchema, () => - Promise.resolve({ + protocol.setRequestHandler(ListToolsRequestSchema, () => { + state.toolLists += 1 + return Promise.resolve({ tools: input.emptyElicitation ? [{ name: "empty-elicitation", inputSchema: { type: "object" as const, properties: {} } }] : input.urlElicitation ? [{ name: "url-elicitation", inputSchema: { type: "object" as const, properties: {} } }] : [], - }), - ) + }) + }) if (input.emptyElicitation) { protocol.setRequestHandler(CallToolRequestSchema, async () => { const result = await protocol.elicitInput({ @@ -133,7 +136,13 @@ function resourceServer( await protocol.connect(transport) const http = Bun.serve({ port: 0, - fetch: (request) => transport.handleRequest(request), + fetch: async (request) => { + const body: unknown = request.method === "POST" ? await request.clone().json() : undefined + if (typeof body === "object" && body !== null && "method" in body && body.method === "initialize") { + state.initializations += 1 + } + return transport.handleRequest(request) + }, }) return { state, @@ -155,6 +164,10 @@ function resourceMcpLayer( server: string | typeof ConfigMCP.Server.Type, onFormCreated?: (form: Form.Info) => Effect.Effect, options?: MCP.Options, + overrides?: { + entries?: Config.Interface["entries"] + subscribe?: Bus.Interface["subscribe"] + }, ) { const directory = AbsolutePath.make(import.meta.dir) const unusedIntegration = () => Effect.die("unused integration service") @@ -162,30 +175,35 @@ function resourceMcpLayer( Layer.provideMerge(Form.layer), Layer.provide( Layer.mergeAll( - Config.testLayer([ - new Document({ - type: "document", - info: new Info({ - mcp: new ConfigMCP.Info({ - servers: { - resources: - typeof server === "string" - ? new ConfigMCP.Remote({ type: "remote", url: server, oauth: false }) - : server, - }, + overrides?.entries + ? Layer.succeed( + Config.Service, + Config.Service.of({ entries: overrides.entries, changes: () => Stream.never }), + ) + : Config.testLayer([ + new Document({ + type: "document", + info: new Info({ + mcp: new ConfigMCP.Info({ + servers: { + resources: + typeof server === "string" + ? new ConfigMCP.Remote({ type: "remote", url: server, oauth: false }) + : server, + }, + }), + }), }), - }), - }), - ]), + ]), Layer.succeed(Location.Service, Location.Service.of(location({ directory }))), Layer.mock(Bus.Service, { - subscribe: () => Stream.never, + subscribe: overrides?.subscribe ?? (() => Stream.never), publish: (definition, data) => { const event = { - id: Event.ID.create(), + id: ID.create(), type: definition.type, data, - } as Event.Payload + } as Payload if (event.type !== Form.Event.Created.type || !onFormCreated) return Effect.succeed(event) return onFormCreated(Schema.decodeUnknownSync(Form.Event.Created.data)(data).form).pipe(Effect.as(event)) }, @@ -785,6 +803,123 @@ test("adds, disconnects, and reconnects MCP servers at runtime", async () => { ) }) +test("reconciles only changed MCP server config", async () => { + await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const server = yield* resourceServer() + const updates = yield* PubSub.unbounded() + const resources = (codemode?: boolean) => + new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false, codemode }) + const added = new ConfigMCP.Local({ type: "local", command: ["unused"], disabled: true }) + const dynamic = new ConfigMCP.Local({ type: "local", command: ["unused"], disabled: true }) + const document = (servers: Record, username?: string) => + new Document({ + type: "document", + info: new Info({ + username, + mcp: new ConfigMCP.Info({ servers }), + }), + }) + let entries = [document({ resources: resources() })] + const publishUpdate = () => + PubSub.publish(updates, { + id: ID.create(), + created: DateTime.makeUnsafe(0), + type: Event.Updated.type, + data: {}, + } satisfies Payload) + + yield* Effect.gen(function* () { + const service = yield* MCP.Service + yield* service.tools() + expect(server.state.toolLists).toBe(1) + expect(server.state.initializations).toBe(1) + + yield* service.add("dynamic", dynamic) + entries = [document({ resources: resources() }, "unrelated")] + yield* publishUpdate() + entries = [document({ resources: resources(), added }, "unrelated")] + yield* publishUpdate() + const appended = yield* service.servers().pipe( + Effect.filterOrFail( + (items) => items.some((item) => item.name === "added"), + () => new Error("MCP config addition was not applied"), + ), + Effect.retry({ times: 100, schedule: Schedule.spaced("10 millis") }), + ) + expect(appended.map((item) => String(item.name)).toSorted()).toEqual(["added", "dynamic", "resources"]) + expect(server.state.toolLists).toBe(1) + expect(server.state.initializations).toBe(1) + + entries = [ + document( + { + resources: resources(false), + added, + }, + "unrelated", + ), + ] + yield* publishUpdate() + yield* Effect.sync(() => server.state.initializations).pipe( + Effect.filterOrFail( + (count) => count === 2, + () => new Error("MCP config change did not reconnect the server"), + ), + Effect.retry({ times: 100, schedule: Schedule.spaced("10 millis") }), + ) + + entries = [document({ added }, "unrelated")] + yield* publishUpdate() + const removed = yield* service.servers().pipe( + Effect.filterOrFail( + (items) => !items.some((item) => item.name === "resources"), + () => new Error("MCP config removal was not applied"), + ), + Effect.retry({ times: 100, schedule: Schedule.spaced("10 millis") }), + ) + expect(removed.map((item) => String(item.name)).toSorted()).toEqual(["added", "dynamic"]) + }).pipe( + Effect.provide( + resourceMcpLayer(resources(), undefined, undefined, { + entries: () => Effect.sync(() => entries), + subscribe: (() => Stream.fromPubSub(updates)) as Bus.Interface["subscribe"], + }), + ), + ) + }), + ), + ) +}) + +test("reconciles MCP config changed during startup", async () => { + const server = new ConfigMCP.Local({ type: "local", command: ["unused"], disabled: true }) + let reads = 0 + const entries = () => + Effect.sync(() => { + reads += 1 + return [ + new Document({ + type: "document", + info: new Info({ + mcp: new ConfigMCP.Info({ servers: reads === 1 ? { initial: server } : { initial: server, added: server } }), + }), + }), + ] + }) + + await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const service = yield* MCP.Service + expect((yield* service.servers()).map((item) => String(item.name))).toEqual(["added", "initial"]) + expect(reads).toBeGreaterThanOrEqual(2) + }).pipe(Effect.provide(resourceMcpLayer(server, undefined, undefined, { entries }))), + ), + ) +}) + test("serializes concurrent MCP lifecycle operations", async () => { await Effect.runPromise( Effect.scoped(