diff --git a/packages/core/src/mcp/index.ts b/packages/core/src/mcp/index.ts index d313c0227f..8d741fb153 100644 --- a/packages/core/src/mcp/index.ts +++ b/packages/core/src/mcp/index.ts @@ -182,7 +182,7 @@ export const layer = (options?: Options) => Layer.effect( const fork = yield* FiberSet.makeRuntime() yield* Effect.addFinalizer((exit) => Scope.close(root, exit)) - const load = Effect.fnUntraced(function* () { + const loadConfig = Effect.fnUntraced(function* () { const documents = (yield* config.entries()).filter( (entry): entry is Config.Document => entry.type === "document", ) @@ -199,8 +199,8 @@ export const layer = (options?: Options) => Layer.effect( } return { timeout, servers } }) - const initial = yield* load() - const configured = { names: new Set(initial.servers.keys()), timeout: initial.timeout } + const initial = yield* loadConfig() + const configState = { names: new Set(initial.servers.keys()), 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 @@ -616,6 +616,22 @@ export const layer = (options?: Options) => Layer.effect( yield* events.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore) }) + const reloadConfig = Effect.fnUntraced(function* () { + const next = yield* loadConfig() + configState.timeout = next.timeout + const names = new Set([...configState.names, ...next.servers.keys()]) + for (const name of names) { + const updated = next.servers.get(name) + if (!updated) { + yield* removeServer(name).pipe(locks.withLock(name)) + continue + } + if (isDeepStrictEqual(runtime.get(name)?.config, updated)) continue + yield* replaceServer(name, updated).pipe(locks.withLock(name)) + } + configState.names = new Set(next.servers.keys()) + }) + // Disabled servers settle their startup immediately so queries never block on them. for (const [name, entry] of runtime) { if (entry.config.disabled) { @@ -653,21 +669,7 @@ export const layer = (options?: Options) => Layer.effect( fork( events.subscribe(Event.Updated).pipe( Stream.runForEach(() => - Effect.gen(function* () { - const next = yield* load() - configured.timeout = next.timeout - const names = new Set([...configured.names, ...next.servers.keys()]) - for (const name of names) { - const updated = next.servers.get(name) - if (updated) { - if (isDeepStrictEqual(runtime.get(name)?.config, updated)) continue - yield* replaceServer(name, updated).pipe(locks.withLock(name)) - continue - } - yield* removeServer(name).pipe(locks.withLock(name)) - } - configured.names = new Set(next.servers.keys()) - }).pipe(Effect.catchCause((cause) => Effect.logError("failed to reload MCP config", { cause }))), + reloadConfig().pipe(Effect.catchCause((cause) => Effect.logError("failed to reload MCP config", { cause }))), ), Effect.ignore, ), @@ -694,7 +696,7 @@ export const layer = (options?: Options) => Layer.effect( }), add: Effect.fn("MCP.add")(function* (server, config) { const name = ServerName.make(server) - yield* replaceServer(name, { ...config, timeout: { ...configured.timeout, ...config.timeout } }).pipe( + yield* replaceServer(name, { ...config, timeout: { ...configState.timeout, ...config.timeout } }).pipe( locks.withLock(name), ) }), diff --git a/packages/core/test/mcp.test.ts b/packages/core/test/mcp.test.ts index 55384e7783..2ee4a6be8f 100644 --- a/packages/core/test/mcp.test.ts +++ b/packages/core/test/mcp.test.ts @@ -150,55 +150,54 @@ function resourceServer( ) } -function resourceMcpLayer( - server: string | typeof ConfigMCP.Server.Type, - onFormCreated?: (form: Form.Info) => Effect.Effect, - options?: MCP.Options, - overrides?: { - entries?: Config.Interface["entries"] - subscribe?: EventV2.Interface["subscribe"] - }, -) { +function resourceMcpLayer(input: { + server: string | typeof ConfigMCP.Server.Type + onFormCreated?: (form: Form.Info) => Effect.Effect + options?: MCP.Options + entries?: Config.Interface["entries"] + subscribe?: EventV2.Interface["subscribe"] +}) { const directory = AbsolutePath.make(import.meta.dir) const unusedIntegration = () => Effect.die("unused integration service") - return MCP.layer(options).pipe( + const entries = + input.entries ?? + (() => + Effect.succeed([ + new Config.Document({ + type: "document", + info: new Config.Info({ + mcp: new ConfigMCP.Info({ + servers: { + resources: + typeof input.server === "string" + ? new ConfigMCP.Remote({ type: "remote", url: input.server, oauth: false }) + : input.server, + }, + }), + }), + }), + ])) + return MCP.layer(input.options).pipe( Layer.provideMerge(Form.layer), Layer.provide( Layer.mergeAll( Layer.succeed( Config.Service, - Config.Service.of({ - entries: - overrides?.entries ?? - (() => - Effect.succeed([ - new Config.Document({ - type: "document", - info: new Config.Info({ - mcp: new ConfigMCP.Info({ - servers: { - resources: - typeof server === "string" - ? new ConfigMCP.Remote({ type: "remote", url: server, oauth: false }) - : server, - }, - }), - }), - }), - ])), - }), + Config.Service.of({ entries }), ), Layer.succeed(Location.Service, Location.Service.of(location({ directory }))), Layer.mock(EventV2.Service, { - subscribe: overrides?.subscribe ?? (() => Stream.never), + subscribe: input.subscribe ?? (() => Stream.never), publish: (definition, data) => { const event = { id: EventV2.ID.create(), type: definition.type, data, } as EventV2.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)) + if (event.type !== Form.Event.Created.type || !input.onFormCreated) return Effect.succeed(event) + return input + .onFormCreated(Schema.decodeUnknownSync(Form.Event.Created.data)(data).form) + .pipe(Effect.as(event)) }, }), Layer.mock(Integration.Service, { @@ -575,7 +574,7 @@ test("accepts empty MCP elicitations without creating forms", async () => { const result = yield* service.callTool({ server: "resources", name: "empty-elicitation" }) expect(yield* forms.list()).toEqual([]) return result - }).pipe(Effect.provide(resourceMcpLayer(server.url))) + }).pipe(Effect.provide(resourceMcpLayer({ server: server.url }))) expect(result.structured).toEqual({ action: "accept", content: {} }) }), @@ -602,7 +601,12 @@ test("acknowledges completed MCP URL elicitations without returning internal con expect(yield* forms.state(form.id)).toEqual({ status: "answered", answer: { elicitation: true } }) return result }).pipe( - Effect.provide(resourceMcpLayer(server.url, (form) => Deferred.succeed(created, form).pipe(Effect.asVoid))), + Effect.provide( + resourceMcpLayer({ + server: server.url, + onFormCreated: (form) => Deferred.succeed(created, form).pipe(Effect.asVoid), + }), + ), ) expect(result.structured).toEqual({ action: "accept" }) @@ -655,7 +659,10 @@ test("loads and reads MCP resources", async () => { expect(server.clientVersion()).toMatchObject({ name: "sdk", version: "1.2.3" }) }).pipe( Effect.provide( - resourceMcpLayer(server.url, undefined, { clientInfo: { name: "sdk", version: "1.2.3" } }), + resourceMcpLayer({ + server: server.url, + options: { clientInfo: { name: "sdk", version: "1.2.3" } }, + }), ), ) }), @@ -718,13 +725,13 @@ test("adds, disconnects, and reconnects MCP servers at runtime", async () => { expect(yield* service.remove("dynamic").pipe(Effect.flip)).toBeInstanceOf(MCP.NotFoundError) }).pipe( Effect.provide( - resourceMcpLayer( - new ConfigMCP.Local({ + resourceMcpLayer({ + server: new ConfigMCP.Local({ type: "local", command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-output-schema.ts")], disabled: true, }), - ), + }), ), ) }), @@ -789,12 +796,11 @@ test("reconciles MCP servers when config updates", async () => { expect(replaced[0]?.status).toEqual({ status: "disabled" }) }).pipe( Effect.provide( - resourceMcpLayer( - new ConfigMCP.Local({ type: "local", command, disabled: true }), - undefined, - undefined, - { entries, subscribe }, - ), + resourceMcpLayer({ + server: new ConfigMCP.Local({ type: "local", command, disabled: true }), + entries, + subscribe, + }), ), ) }), @@ -833,13 +839,13 @@ test("serializes concurrent MCP lifecycle operations", async () => { expect((yield* service.tools()).length).toBeGreaterThan(0) }).pipe( Effect.provide( - resourceMcpLayer( - new ConfigMCP.Local({ + resourceMcpLayer({ + server: new ConfigMCP.Local({ type: "local", command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-output-schema.ts")], disabled: true, }), - ), + }), ), ) }),