diff --git a/packages/cli/src/acp/service.ts b/packages/cli/src/acp/service.ts index c681db0ad85..8ae800a4d4d 100644 --- a/packages/cli/src/acp/service.ts +++ b/packages/cli/src/acp/service.ts @@ -98,7 +98,7 @@ export function make(input: { readonly connection: ACPConnection.Connection }): Interface { const sessions = new Map() - const registeredMcp = new Map>() + const registeredMcp = new Map>() const active = new Map }>() const capabilities = { writeTextFile: false, childSessionUpdates: false } @@ -482,34 +482,22 @@ async function messages(client: OpenCodeClient, sessionID: string) { async function registerMcpServers( client: OpenCodeClient, - registered: Map>, + registered: Map>, session: Attached, servers: readonly McpServer[], ) { - const current = registered.get(session.id) ?? new Map() + const current = registered.get(session.id) ?? new Set() registered.set(session.id, current) await Promise.all( - servers.flatMap((server) => { - const config = mcpConfig(server) - const key = stableStringify(config) - if (current.get(server.name) === key) return [] - current.set(server.name, key) - return [ - client.session.mcp.add({ sessionID: session.id, server: server.name, config }).catch((error) => { - if (current.get(server.name) === key) current.delete(server.name) - throw error - }), - ] + servers.map(async (server) => { + await client.session.mcp.add({ sessionID: session.id, server: server.name, config: mcpConfig(server) }) + current.add(server.name) }), ) } -async function releaseMcpServers( - client: OpenCodeClient, - registered: Map>, - sessionID: string, -) { - const servers = Array.from(registered.get(sessionID)?.keys() ?? []) +async function releaseMcpServers(client: OpenCodeClient, registered: Map>, sessionID: string) { + const servers = Array.from(registered.get(sessionID) ?? []) registered.delete(sessionID) await Promise.all( servers.map((server) => @@ -537,15 +525,6 @@ function mcpConfig(server: McpServer) { } } -function stableStringify(value: unknown): string { - if (Array.isArray(value)) return `[${value.map(stableStringify).join(",")}]` - if (!value || typeof value !== "object") return JSON.stringify(value) - return `{${Object.entries(value) - .toSorted(([a], [b]) => a.localeCompare(b)) - .map(([key, item]) => `${JSON.stringify(key)}:${stableStringify(item)}`) - .join(",")}}` -} - async function sendUsageUpdate( client: OpenCodeClient, connection: ACPConnection.Connection, diff --git a/packages/cli/test/acp/service-directory.test.ts b/packages/cli/test/acp/service-directory.test.ts index aa57353d180..fc0e4c70ac8 100644 --- a/packages/cli/test/acp/service-directory.test.ts +++ b/packages/cli/test/acp/service-directory.test.ts @@ -317,7 +317,7 @@ describe("acp service directory behavior", () => { expect(invalidConfig).toMatchObject({ _tag: "ACPInvalidConfigOptionError" }) }) - test("registers MCP configs per session and deduplicates identical re-registrations", async () => { + test("registers MCP configs for the session on every attach", async () => { const local: McpServer = { name: "tools", command: "bun", @@ -349,13 +349,15 @@ describe("acp service directory behavior", () => { }, }) - await fixture.service.newSession({ cwd: "/workspace", mcpServers: [local, local, remote] }) + await fixture.service.newSession({ cwd: "/workspace", mcpServers: [local, remote] }) await fixture.service.resumeSession({ cwd: "/workspace", sessionId: "ses_1", mcpServers: [local, remote] }) await fixture.service.resumeSession({ cwd: "/workspace", sessionId: "ses_1", mcpServers: [changed] }) await fixture.service.newSession({ cwd: "/workspace", mcpServers: [local] }) const adds = fixture.requests.filter((request) => request.method === "PUT" && mcp.test(request.path)) expect(adds.map((request) => request.path.match(mcp)?.slice(1))).toEqual([ + ["ses_1", "tools"], + ["ses_1", "docs"], ["ses_1", "tools"], ["ses_1", "docs"], ["ses_1", "tools"], @@ -369,6 +371,13 @@ describe("acp service directory behavior", () => { environment: { TOKEN: "x" }, }, }, + { + config: { + type: "local", + command: ["bun", "server.ts"], + environment: { TOKEN: "x" }, + }, + }, { config: { type: "local", @@ -392,7 +401,7 @@ describe("acp service directory behavior", () => { oauth: false, }, }) - expect(adds.map((request) => request.query)).toEqual([{}, {}, {}, {}]) + expect(adds.map((request) => request.query)).toEqual([{}, {}, {}, {}, {}, {}]) }) }) diff --git a/packages/core/src/instance.ts b/packages/core/src/instance.ts index 0699bc2dfb2..6cd8065c0f4 100644 --- a/packages/core/src/instance.ts +++ b/packages/core/src/instance.ts @@ -22,6 +22,7 @@ import { LocationLifecycle } from "./location-lifecycle.js" import { FileAccess } from "./file-access.js" import { ModelResolver } from "./model-resolver.js" import { Mcp } from "./mcp/index.js" +import { McpSession } from "./mcp/session.js" import { Permission } from "./permission.js" import { Plugin } from "./plugin.js" import { PluginHooks } from "./plugin/hooks.js" @@ -88,6 +89,7 @@ const nodes = [ FileMutation.node, Formatter.node, Mcp.node, + McpSession.node, Permission.node, Tool.node, ToolOutput.node, diff --git a/packages/core/src/mcp/elicitation.ts b/packages/core/src/mcp/elicitation.ts new file mode 100644 index 00000000000..94130d893b1 --- /dev/null +++ b/packages/core/src/mcp/elicitation.ts @@ -0,0 +1,145 @@ +export * as McpElicitation from "./elicitation.js" + +import { Effect } from "effect" +import { waitForAbort } from "@opencode/util/process" +import { Form } from "../form.js" +import type { McpClient } from "./client.js" + +const URL_FIELD_KEY = "elicitation" + +/** Answers one connection's elicitation requests with forms owned by `sessionID`. */ +export const handler = (forms: Form.Interface, sessionID: string): McpClient.ElicitationHandler => { + // Legacy era only: pending URL-mode elicitation forms, settled by notifications/elicitation/complete. + const pending = new Map() + return { + create: (request) => + Effect.gen(function* () { + if (request.params.mode === "url") { + const formID = Form.ID.create() + // Legacy only: 2026-07-28 has no elicitationId and no completion notification, so the form + // settles when the user confirms and the SDK retries the tool call itself. + const elicitationID: string | undefined = request.params.elicitationId + if (elicitationID !== undefined) pending.set(elicitationID, formID) + return yield* forms + .ask({ + id: formID, + sessionID, + title: `${request.server} is requesting input`, + metadata: { + kind: "mcp-elicitation", + server: request.server, + ...(elicitationID === undefined ? {} : { elicitationID }), + message: request.params.message, + }, + fields: [{ key: URL_FIELD_KEY, type: "external", url: request.params.url }], + }) + .pipe( + Effect.raceFirst(waitForAbort(request.signal)), + Effect.ensuring(Effect.sync(() => elicitationID !== undefined && pending.delete(elicitationID))), + Effect.map( + (state): McpClient.ElicitationResult => ({ + action: state.status === "answered" ? "accept" : "cancel", + }), + ), + ) + } + const params = request.params + const [field, ...fields] = Object.entries(params.requestedSchema.properties).map(([key, property]) => + toField(key, property, params.requestedSchema.required?.includes(key) === true), + ) + if (!field) return { action: "accept", content: {} } + return yield* forms + .ask({ + sessionID, + title: `${request.server} is requesting input`, + metadata: { kind: "mcp-elicitation", server: request.server, message: params.message }, + fields: [field, ...fields], + }) + .pipe( + Effect.raceFirst(waitForAbort(request.signal)), + Effect.map((state): McpClient.ElicitationResult => { + if (state.status !== "answered") return { action: "cancel" } + return { + action: "accept", + content: Object.fromEntries( + Object.entries(state.answer).map( + ([key, value]): [string, NonNullable[string]] => + typeof value === "object" ? [key, Array.from(value)] : [key, value], + ), + ), + } + }), + ) + }), + complete: (request) => + Effect.gen(function* () { + const formID = pending.get(request.elicitationID) + if (!formID) return + yield* forms.reply({ id: formID, answer: { [URL_FIELD_KEY]: true } }).pipe(Effect.ignore) + }), + } +} + +// Schema `optional` strips undefined-valued properties on encode, so fields can assign +// optional properties directly instead of conditionally spreading them. +function toField(key: string, property: ElicitationProperty, required: boolean): Form.Field { + // Some servers emit machine titles like "string with format email"; prefer description/key over those. + const machineTitle = /^(boolean|string|number|integer|array|object)(\s+with\b.*|\s+in\b.*)?$/i + const title = + property.title && !machineTitle.test(property.title.trim()) ? property.title : (property.description ?? key) + const base = { + key, + title, + description: property.description === title ? undefined : property.description, + required: required || undefined, + } + switch (property.type) { + case "boolean": + return { ...base, type: "boolean", default: property.default } + case "number": + case "integer": + return { + ...base, + type: property.type, + minimum: property.minimum, + maximum: property.maximum, + default: property.default, + } + case "array": + return { + ...base, + type: "multiselect", + options: + "anyOf" in property.items + ? property.items.anyOf.map((option) => ({ value: option.const, label: option.title })) + : property.items.enum.map((value) => ({ value, label: value })), + custom: false, + minItems: property.minItems, + maxItems: property.maxItems, + default: property.default, + } + case "string": { + const options = + "oneOf" in property + ? property.oneOf.map((option) => ({ value: option.const, label: option.title })) + : "enum" in property + ? property.enum.map((value, index) => ({ + value, + label: ("enumNames" in property ? property.enumNames?.[index] : undefined) ?? value, + })) + : undefined + return { + ...base, + type: "string", + format: "format" in property ? property.format : undefined, + minLength: "minLength" in property ? property.minLength : undefined, + maxLength: "maxLength" in property ? property.maxLength : undefined, + default: property.default, + options, + custom: options ? false : undefined, + } + } + } +} + +type ElicitationProperty = McpClient.ElicitationFormParams["requestedSchema"]["properties"][string] diff --git a/packages/core/src/mcp/index.ts b/packages/core/src/mcp/index.ts index a383cfb3ab4..b48083a14b0 100644 --- a/packages/core/src/mcp/index.ts +++ b/packages/core/src/mcp/index.ts @@ -4,7 +4,6 @@ import { Mcp } from "@opencode/schema/mcp" import { McpEvent } from "@opencode/schema/mcp-event" import { ephemeral } from "@opencode/schema/event" import type { Session } from "@opencode/schema/session" -import { SessionEvent } from "@opencode/schema/session-event" import { createHash } from "node:crypto" import { isDeepStrictEqual } from "node:util" import { Cause, Context, Effect, Exit, FiberSet, Latch, Layer, Schema, Scope, Semaphore, Stream, Types } from "effect" @@ -16,10 +15,9 @@ import { Form } from "../form.js" import { Integration } from "../integration.js" import { KeyedMutex } from "../effect/keyed-mutex.js" import { Location } from "../location.js" -import { SessionStore } from "../session/store.js" -import { waitForAbort } from "@opencode/util/process" import { State } from "../state.js" import type { McpClient } from "./client.js" +import { McpElicitation } from "./elicitation.js" export const ServerName = Schema.String.pipe(Schema.brand("MCP.ServerName")) export type ServerName = typeof ServerName.Type @@ -27,6 +25,8 @@ export const PromptsChanged = ephemeral({ type: "mcp.prompts.changed", schema: { export const Status = Mcp.Status export type Status = Mcp.Status +export const ServerConfig = Mcp.ServerConfig +export type ServerConfig = Mcp.ServerConfig export type ServerInfo = Mcp.Server export interface ServerInstructions { @@ -35,12 +35,7 @@ export interface ServerInstructions { } /** SDK tool definition tagged with the server that owns it. */ -export type Tool = McpClient.Tool & { - readonly server: ServerName - readonly codemode?: boolean - /** Owner of the Session-scoped server that provides this tool; absent for Location servers. */ - readonly sessionID?: Session.ID -} +export type Tool = McpClient.Tool & { readonly server: ServerName; readonly codemode?: boolean } export type ToolResultContent = McpClient.CallToolContent export type ToolResult = McpClient.CallToolResult & { readonly server: ServerName; readonly tool: string } export type Prompt = McpClient.Prompt & { readonly server: ServerName } @@ -69,9 +64,12 @@ export class ToolCallError extends Schema.TaggedError()("MCP.Tool message: Schema.String, }) {} -type ServerEntry = { +export type Owner = { readonly type: "location" } | { readonly type: "session"; readonly id: Session.ID } +export type Lock = (effect: Effect.Effect) => Effect.Effect + +export type Entry = { + readonly owner: O readonly config: Mcp.ServerConfig - readonly sessionID?: Session.ID status: Status readonly startup: Latch.Latch scope?: Scope.Closeable @@ -83,11 +81,13 @@ type ServerEntry = { registration?: State.Registration } -// MCP elicitations are Location-scoped, not Session-scoped: the server cannot attribute them to a -// persisted session row, so their forms are owned by this opaque sentinel session identifier. +const LOCATION = { type: "location" } as const +type LocationEntry = Entry + +// Location servers cannot attribute elicitations to a persisted session row, so their forms are owned +// by this opaque sentinel session identifier. Session-scoped servers ask on behalf of their owner. const GLOBAL_ELICITATION_SESSION_ID = "global" -const URL_ELICITATION_FIELD_KEY = "elicitation" -// Connections remain Location-scoped, but shared remote endpoints should not receive concurrent startup bursts. +// Shared remote endpoints should not receive concurrent startup bursts, whichever registry owns them. const endpointLoads = KeyedMutex.makeUnsafe() type Data = { @@ -103,43 +103,33 @@ export type Editor = { remove: (server: ServerName | string) => void } -const cloneConfig = (config: Mcp.ServerConfig) => structuredClone(config) as Types.DeepMutable +export const cloneConfig = (config: Mcp.ServerConfig) => structuredClone(config) as Types.DeepMutable -/** - * Methods taking a `sessionID` resolve the servers visible to that Session: servers registered for the - * Session or one of its ancestors, nearest owner first, shadowing Location servers of the same name. - * Without a `sessionID` they address Location servers only. - */ export interface Interface extends State.Transformable { - readonly servers: (sessionID?: Session.ID) => Effect.Effect - /** Session-scoped servers are process-local and released when removed or when their Session is deleted or moved. */ - readonly add: (server: ServerName | string, config: Mcp.ServerConfig, sessionID?: Session.ID) => Effect.Effect + readonly servers: () => Effect.Effect + readonly add: (server: ServerName | string, config: Mcp.ServerConfig) => Effect.Effect readonly connect: (server: ServerName | string) => Effect.Effect readonly disconnect: (server: ServerName | string) => Effect.Effect - readonly remove: (server: ServerName | string, sessionID?: Session.ID) => Effect.Effect - readonly tools: (sessionID?: Session.ID) => Effect.Effect + readonly remove: (server: ServerName | string) => Effect.Effect + readonly tools: () => Effect.Effect readonly callTool: (input: { readonly server: ServerName | string readonly name: string readonly args?: Record readonly sessionID?: Session.ID }) => Effect.Effect - readonly instructions: (sessionID?: Session.ID) => Effect.Effect + readonly instructions: () => Effect.Effect readonly prompts: () => Effect.Effect readonly prompt: (input: { readonly server: ServerName | string readonly name: string readonly args?: Record }) => Effect.Effect - readonly resourceCatalog: (sessionID?: Session.ID) => Effect.Effect - readonly resources: (input: { - readonly server: ServerName | string - readonly sessionID?: Session.ID - }) => Effect.Effect + readonly resourceCatalog: () => Effect.Effect + readonly resources: (input: { readonly server: ServerName | string }) => Effect.Effect readonly readResource: (input: { readonly server: ServerName | string readonly uri: string - readonly sessionID?: Session.ID }) => Effect.Effect } @@ -155,34 +145,324 @@ export const Options = Schema.Struct({ }) export type Options = typeof Options.Type +/** + * Connection lifecycle shared by the Location registry and the Session-scoped registry. Each registry + * builds its own runtime and supplies the per-server lock; only Location entries publish MCP events. + */ +export const makeRuntime = Effect.fnUntraced(function* (input: { + readonly options?: Options + readonly lock: (name: ServerName, owner: O) => Lock +}) { + const location = yield* Location.Service + const environment = yield* Environment.Service + const bus = yield* Bus.Service + const forms = yield* Form.Service + const credentials = yield* Credential.Service + const root = yield* Effect.scope + const fork = yield* FiberSet.makeRuntime() + // Lifecycle operations close scopes while holding the server lock, firing onClose, so anything + // taking it from a connection callback must stay forked. + const lock = (name: ServerName, entry: Entry) => input.lock(name, entry.owner) + const notify = ( + entry: Entry, + definition: + | typeof McpEvent.StatusChanged + | typeof McpEvent.ToolsChanged + | typeof McpEvent.ResourcesChanged + | typeof PromptsChanged, + name: ServerName, + ) => (entry.owner.type === "location" ? bus.publish(definition, { server: name }).pipe(Effect.asVoid) : Effect.void) + + const connectProvider = Effect.fnUntraced(function* (entry: Entry) { + if (entry.config.type !== "remote" || !entry.integrationID) return undefined + const { McpOAuth } = yield* Effect.promise(() => import("./oauth.js")) + return yield* McpOAuth.connectProvider({ config: entry.config, integrationID: entry.integrationID }).pipe( + Effect.provideService(Credential.Service, credentials), + ) + }) + + const toTool = (server: ServerName, entry: Entry, tool: McpClient.Tool): Tool => ({ + ...tool, + server, + ...(entry.config.codemode === undefined ? {} : { codemode: entry.config.codemode }), + }) + + const refreshTools = (name: ServerName, entry: Entry, connection: McpClient.Connection) => + connection.tools().pipe( + Effect.map((tools) => { + entry.tools = tools.map((tool) => toTool(name, entry, tool)) + }), + ) + + const refreshPrompts = (name: ServerName, entry: Entry, connection: McpClient.Connection) => + connection.prompts().pipe( + Effect.orElseSucceed(() => []), + Effect.map((prompts) => { + entry.prompts = prompts.map((prompt): Prompt => ({ ...prompt, server: name })) + }), + Effect.andThen(notify(entry, PromptsChanged, name)), + ) + + // Runs a connection callback under the server lock, dropping it if the connection is no longer + // the entry's live client, so late SDK callbacks cannot commit obsolete state. + const whenLive = + (name: ServerName, entry: Entry, connection: McpClient.Connection) => + (effect: Effect.Effect) => + fork( + Effect.suspend(() => (entry.client === connection ? effect : Effect.void)).pipe( + lock(name, entry), + Effect.ignore, + ), + ) + + // Re-establishes a server whose HTTP session the server dropped, unless another path already + // replaced the connection while this one waited for the lock. + const recover = (name: ServerName, entry: Entry, connection: McpClient.Connection) => + Effect.gen(function* () { + if (entry.client !== connection) return + yield* Effect.logInfo("mcp session expired, reconnecting", { server: name }) + yield* stopServer(name, entry) + yield* startServer(name, entry) + }).pipe(lock(name, entry)) + + // Runs a request against the live connection and, if that request observed a session expiry, + // reconnects and runs it once more against the replacement. Any other failure passes through. + const recovering = ( + name: ServerName, + entry: Entry, + connection: McpClient.Connection, + run: (connection: McpClient.Connection) => Effect.Effect, + ) => + run(connection).pipe( + Effect.catchIf( + // The client module is loaded lazily, so match the tagged error by tag rather than class. + (error) => "_tag" in error && error._tag === "MCP.SessionExpiredError", + (error) => + recover(name, entry, connection).pipe( + Effect.flatMap(() => (entry.client ? run(entry.client) : Effect.fail(error))), + ), + ), + ) + + const loadCatalog = (name: ServerName, entry: Entry, connection: McpClient.Connection) => + recovering(name, entry, connection, (connection) => + Effect.all( + { + resources: connection.resources(), + // Some servers declare resources without implementing template listing. + templates: connection.resourceTemplates().pipe(Effect.orElseSucceed(() => [])), + }, + { concurrency: "unbounded" }, + ), + ).pipe( + Effect.map((catalog) => + ResourceCatalog.make({ + resources: catalog.resources.map((resource) => + Resource.make({ + server: name, + name: resource.name, + uri: resource.uri, + description: resource.description, + mimeType: resource.mimeType, + }), + ), + templates: catalog.templates.map((template) => + ResourceTemplate.make({ + server: name, + name: template.name, + uriTemplate: template.uriTemplate, + description: template.description, + mimeType: template.mimeType, + }), + ), + }), + ), + ) + + const watch = (name: ServerName, entry: Entry, connection: McpClient.Connection) => { + const live = whenLive(name, entry, connection) + connection.onClose(() => + live( + Effect.gen(function* () { + entry.status = { status: "failed", error: "Connection closed" } + yield* stopServer(name, entry) + yield* notify(entry, McpEvent.StatusChanged, name) + }), + ), + ) + // Background refreshes (list-changed) can observe the expiry too; they do not retry, so + // reconnect here for them. Foreground calls reconnect through `recovering`. + connection.onSessionExpired(() => fork(recover(name, entry, connection))) + connection.onToolsChanged(() => + live(refreshTools(name, entry, connection).pipe(Effect.andThen(notify(entry, McpEvent.ToolsChanged, name)))), + ) + connection.onPromptsChanged(() => live(refreshPrompts(name, entry, connection))) + connection.onResourcesChanged(() => live(notify(entry, McpEvent.ResourcesChanged, name))) + } + + const startServer = (name: ServerName, entry: Entry) => + Effect.gen(function* () { + // Announce the handshake so connect() and credential reconnects don't show a stale + // disabled/failed status for the duration of the connection attempt. + entry.status = { status: "pending" } + yield* notify(entry, McpEvent.StatusChanged, name) + const scope = yield* Scope.fork(root) + entry.scope = scope + const authProvider = yield* connectProvider(entry) + const { McpClient } = yield* Effect.promise(() => import("./client.js")) + // List tools as part of connect so a failure here marks the server failed rather than + // leaving it connected with a silently empty tool list and no path to recover. + const load = McpClient.connect( + name, + entry.config, + location.directory, + authProvider, + McpElicitation.handler(forms, entry.owner.type === "session" ? entry.owner.id : GLOBAL_ELICITATION_SESSION_ID), + input.options?.clientInfo, + ).pipe( + Effect.flatMap((connection) => connection.tools().pipe(Effect.map((tools) => ({ connection, tools })))), + // A stdio server is spawned on this location's execution plane, not the host's. + Effect.provideService(Environment.Service, environment), + Scope.provide(scope), + ) + const result = yield* ( + entry.config.type === "remote" ? endpointLoads.withLock(entry.config.url)(load) : load + ).pipe(Effect.exit) + if (Exit.isSuccess(result)) { + entry.client = result.value.connection + entry.tools = result.value.tools.map((tool) => toTool(name, entry, tool)) + entry.prompts = [] + entry.status = { status: "connected" } + watch(name, entry, result.value.connection) + yield* Effect.logInfo("mcp connected", { server: name, tools: entry.tools.length }) + // The tool registry reads on this event; a late-connecting server has no other way to appear. + yield* notify(entry, McpEvent.ToolsChanged, name) + yield* notify(entry, McpEvent.ResourcesChanged, name) + yield* notify(entry, McpEvent.StatusChanged, name) + whenLive(name, entry, result.value.connection)(refreshPrompts(name, entry, result.value.connection)) + 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", error: error.message } + : { status: "failed", error: error instanceof Error ? error.message : String(error) } + yield* Effect.logWarning("mcp connect failed", { server: name, status: entry.status }) + yield* notify(entry, McpEvent.StatusChanged, name) + }).pipe( + Effect.ensuring(entry.startup.open), + Effect.annotateLogs({ server: name, directory: location.directory, connectionID: crypto.randomUUID() }), + ) + + const stopServer = Effect.fnUntraced(function* (name: ServerName, entry: Entry) { + const scope = entry.scope + if (!scope) return + entry.scope = undefined + entry.client = undefined + entry.tools = undefined + entry.prompts = undefined + yield* Scope.close(scope, Exit.void) + yield* notify(entry, McpEvent.ToolsChanged, name) + yield* notify(entry, McpEvent.ResourcesChanged, name) + yield* notify(entry, PromptsChanged, name) + }) + + // Brings a new entry online, or marks it disabled, and settles its startup latch even when + // `prepare` fails or the caller is interrupted, so readers cannot hang. + const open = (name: ServerName, entry: Entry, prepare: Effect.Effect = Effect.void) => + Effect.gen(function* () { + yield* prepare + if (entry.config.disabled) { + entry.status = { status: "disabled" } + yield* notify(entry, McpEvent.StatusChanged, name) + return + } + yield* startServer(name, entry) + }).pipe(Effect.ensuring(entry.startup.open)) + + const settled = (entry: Entry) => entry.startup.await.pipe(Effect.map(() => entry.client)) + + return { + fork, + start: startServer, + stop: stopServer, + open, + callTool: Effect.fnUntraced(function* ( + name: ServerName, + entry: Entry, + call: { readonly name: string; readonly args?: Record; readonly sessionID?: Session.ID }, + ) { + const client = yield* settled(entry) + if (!client) + return yield* new ToolCallError({ server: name, tool: call.name, message: "MCP server is not connected" }) + const result = yield* recovering(name, entry, client, (connection) => + connection.callTool({ name: call.name, args: call.args, sessionID: call.sessionID }), + ).pipe(Effect.mapError((error) => new ToolCallError({ server: name, tool: call.name, message: error.message }))) + return { ...result, server: name, tool: call.name } + }), + prompt: Effect.fnUntraced(function* ( + name: ServerName, + entry: Entry, + prompt: { readonly name: string; readonly args?: Record }, + ) { + const client = yield* settled(entry) + if (!client) return undefined + const result = yield* recovering(name, entry, client, (connection) => + connection.prompt({ name: prompt.name, args: prompt.args }), + ).pipe(Effect.orElseSucceed(() => undefined)) + if (!result) return undefined + return { ...result, server: name, name: prompt.name } + }), + // Reports what is connected now without waiting for servers still starting. + catalog: (name: ServerName, entry: Entry) => + entry.client + ? loadCatalog(name, entry, entry.client).pipe(Effect.orElseSucceed(() => emptyCatalog)) + : Effect.succeed(emptyCatalog), + resources: Effect.fnUntraced(function* (name: ServerName, entry: Entry) { + const client = yield* settled(entry) + if (!client) return emptyCatalog + return mergeCatalogs([yield* loadCatalog(name, entry, client)]) + }), + readResource: Effect.fnUntraced(function* (name: ServerName, entry: Entry, uri: string) { + const client = yield* settled(entry) + if (!client) return undefined + const result = yield* recovering(name, entry, client, (connection) => connection.readResource({ uri })) + if (!result) return undefined + return ResourceContent.make({ + server: name, + uri, + contents: result.contents.map((part) => + "text" in part + ? { type: "text", uri: part.uri, text: part.text, mimeType: part.mimeType } + : { type: "blob", uri: part.uri, blob: part.blob, mimeType: part.mimeType }, + ), + }) + }), + } +}) + +const emptyCatalog = ResourceCatalog.make({ resources: [], templates: [] }) + export const layer = (options?: Options) => Layer.effect( Service, Effect.gen(function* () { - const location = yield* Location.Service - const environment = yield* Environment.Service const bus = yield* Bus.Service - const forms = yield* Form.Service const integration = yield* Integration.Service const credentials = yield* Credential.Service - const store = yield* SessionStore.Service const root = yield* Effect.scope - const fork = yield* FiberSet.makeRuntime() - const entries = new Map() - const sessionEntries = 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 slot = (name: ServerName, sessionID: Session.ID | undefined) => - sessionID === undefined ? name : sessionID + "\u0000" + name - // Legacy era only: pending URL-mode elicitation forms, settled by notifications/elicitation/complete. - const urlElicitations = new Map() + const entries = new Map() + const locks = KeyedMutex.makeUnsafe() + const runtime = yield* makeRuntime({ options, lock: (name) => locks.withLock(name) }) + const fork = runtime.fork // Register every remote server as an OAuth integration so credentials live in the global store // rather than in committed config. Servers that connect anonymously simply never use the method. const owned = new Set() - const register = Effect.fnUntraced(function* (name: ServerName, entry: ServerEntry) { + const register = Effect.fnUntraced(function* (name: ServerName, entry: LocationEntry) { if (entry.config.type !== "remote" || entry.config.oauth === false) return const remote = entry.config // Key identity on name + url, not url alone: two configs for the same url under different names are @@ -222,305 +502,15 @@ export const layer = (options?: Options) => }) .pipe(Scope.provide(scope)) }) - const sessionServers = Effect.fnUntraced(function* (sessionID: Session.ID | undefined) { - const scoped = new Map() - if (sessionID === undefined || sessionEntries.size === 0) return scoped - let current: Session.ID | undefined = sessionID - while (current) { - for (const [name, entry] of sessionEntries.get(current) ?? []) if (!scoped.has(name)) scoped.set(name, entry) - current = (yield* store.get(current))?.parentID - } - return scoped - }) - - const visibleServers = Effect.fnUntraced(function* (sessionID: Session.ID | undefined) { - const scoped = yield* sessionServers(sessionID) - return [...Array.from(entries).filter(([name]) => !scoped.has(name)), ...scoped] - }) - - const requireServer = Effect.fnUntraced(function* (server: ServerName | string, sessionID?: Session.ID) { + const requireServer = Effect.fnUntraced(function* (server: ServerName | string) { const name = ServerName.make(server) - const entry = (yield* sessionServers(sessionID)).get(name) ?? entries.get(name) + const entry = entries.get(name) if (!entry) return yield* new NotFoundError({ server: name }) return { name, entry } }) - const connectProvider = Effect.fnUntraced(function* (entry: ServerEntry) { - if (entry.config.type !== "remote" || !entry.integrationID) return undefined - const { McpOAuth } = yield* Effect.promise(() => import("./oauth.js")) - return yield* McpOAuth.connectProvider({ config: entry.config, integrationID: entry.integrationID }).pipe( - Effect.provideService(Credential.Service, credentials), - ) - }) - - const elicitation = { - create: (input: { - readonly server: string - readonly params: McpClient.ElicitationParams - readonly signal: AbortSignal - }) => - Effect.gen(function* () { - if (input.params.mode === "url") { - const formID = Form.ID.create() - // Legacy only: 2026-07-28 has no elicitationId and no completion notification, so the form - // settles when the user confirms and the SDK retries the tool call itself. - const elicitationID: string | undefined = input.params.elicitationId - const key = elicitationID === undefined ? undefined : input.server + "\u0000" + elicitationID - if (key) urlElicitations.set(key, formID) - return yield* forms - .ask({ - id: formID, - sessionID: GLOBAL_ELICITATION_SESSION_ID, - title: `${input.server} is requesting input`, - metadata: { - kind: "mcp-elicitation", - server: input.server, - ...(elicitationID === undefined ? {} : { elicitationID }), - message: input.params.message, - }, - fields: [{ key: URL_ELICITATION_FIELD_KEY, type: "external", url: input.params.url }], - }) - .pipe( - Effect.raceFirst(waitForAbort(input.signal)), - Effect.ensuring(Effect.sync(() => key && urlElicitations.delete(key))), - Effect.map( - (state): McpClient.ElicitationResult => ({ - action: state.status === "answered" ? "accept" : "cancel", - }), - ), - ) - } - const params = input.params - const [field, ...fields] = Object.entries(params.requestedSchema.properties).map(([key, property]) => - toElicitationField(key, property, params.requestedSchema.required?.includes(key) === true), - ) - if (!field) return { action: "accept", content: {} } - return yield* forms - .ask({ - sessionID: GLOBAL_ELICITATION_SESSION_ID, - title: `${input.server} is requesting input`, - metadata: { kind: "mcp-elicitation", server: input.server, message: params.message }, - fields: [field, ...fields], - }) - .pipe( - Effect.raceFirst(waitForAbort(input.signal)), - Effect.map((state): McpClient.ElicitationResult => { - if (state.status !== "answered") return { action: "cancel" } - return { - action: "accept", - content: Object.fromEntries( - Object.entries(state.answer).map( - ([key, value]): [string, NonNullable[string]] => - typeof value === "object" ? [key, Array.from(value)] : [key, value], - ), - ), - } - }), - ) - }), - complete: (input: { readonly server: string; readonly elicitationID: string }) => - Effect.gen(function* () { - const formID = urlElicitations.get(input.server + "\u0000" + input.elicitationID) - if (!formID) return - yield* forms.reply({ id: formID, answer: { [URL_ELICITATION_FIELD_KEY]: true } }).pipe(Effect.ignore) - }), - } satisfies McpClient.ElicitationHandler - - const toTool = (server: ServerName, entry: ServerEntry, tool: McpClient.Tool): Tool => ({ - ...tool, - server, - ...(entry.config.codemode === undefined ? {} : { codemode: entry.config.codemode }), - ...(entry.sessionID === undefined ? {} : { sessionID: entry.sessionID }), - }) - - const refreshTools = (name: ServerName, entry: ServerEntry, connection: McpClient.Connection) => - connection.tools().pipe( - Effect.map((tools) => { - entry.tools = tools.map((tool) => toTool(name, entry, tool)) - }), - ) - - const refreshPrompts = (name: ServerName, entry: ServerEntry, connection: McpClient.Connection) => - connection.prompts().pipe( - Effect.orElseSucceed(() => []), - Effect.map((prompts) => { - entry.prompts = prompts.map((prompt): Prompt => ({ ...prompt, server: name })) - }), - Effect.andThen(bus.publish(PromptsChanged, { server: name })), - ) - - // Runs a connection callback under the server lock, dropping it if the connection is no longer - // the entry's live client, so late SDK callbacks cannot commit obsolete state. - const whenLive = - (name: ServerName, entry: ServerEntry, connection: McpClient.Connection) => - (effect: Effect.Effect) => - fork( - Effect.suspend(() => (entry.client === connection ? effect : Effect.void)).pipe( - locks.withLock(slot(name, entry.sessionID)), - Effect.ignore, - ), - ) - - // Re-establishes a server whose HTTP session the server dropped, unless another path already - // replaced the connection while this one waited for the lock. - const recover = (name: ServerName, entry: ServerEntry, connection: McpClient.Connection) => - Effect.gen(function* () { - if (entry.client !== connection) return - yield* Effect.logInfo("mcp session expired, reconnecting", { server: name }) - yield* stopServer(name, entry) - yield* startServer(name, entry) - }).pipe(locks.withLock(slot(name, entry.sessionID))) - - // Runs a request against the live connection and, if that request observed a session expiry, - // reconnects and runs it once more against the replacement. Any other failure passes through. - const recovering = ( - name: ServerName, - entry: ServerEntry, - connection: McpClient.Connection, - run: (connection: McpClient.Connection) => Effect.Effect, - ) => - run(connection).pipe( - Effect.catchIf( - // The client module is loaded lazily, so match the tagged error by tag rather than class. - (error) => "_tag" in error && error._tag === "MCP.SessionExpiredError", - (error) => - recover(name, entry, connection).pipe( - Effect.flatMap(() => (entry.client ? run(entry.client) : Effect.fail(error))), - ), - ), - ) - - const loadCatalog = (name: ServerName, entry: ServerEntry, connection: McpClient.Connection) => - recovering(name, entry, connection, (connection) => - Effect.all( - { - resources: connection.resources(), - // Some servers declare resources without implementing template listing. - templates: connection.resourceTemplates().pipe(Effect.orElseSucceed(() => [])), - }, - { concurrency: "unbounded" }, - ), - ).pipe( - Effect.map((catalog) => - ResourceCatalog.make({ - resources: catalog.resources.map((resource) => - Resource.make({ - server: name, - name: resource.name, - uri: resource.uri, - description: resource.description, - mimeType: resource.mimeType, - }), - ), - templates: catalog.templates.map((template) => - ResourceTemplate.make({ - server: name, - name: template.name, - uriTemplate: template.uriTemplate, - description: template.description, - mimeType: template.mimeType, - }), - ), - }), - ), - ) - - const watch = (name: ServerName, entry: ServerEntry, connection: McpClient.Connection) => { - const live = whenLive(name, entry, connection) - connection.onClose(() => - live( - Effect.gen(function* () { - entry.status = { status: "failed", error: "Connection closed" } - yield* stopServer(name, entry) - yield* bus.publish(McpEvent.StatusChanged, { server: name }) - }), - ), - ) - // Background refreshes (list-changed) can observe the expiry too; they do not retry, so - // reconnect here for them. Foreground calls reconnect through `recovering`. - connection.onSessionExpired(() => fork(recover(name, entry, connection))) - connection.onToolsChanged(() => - live( - refreshTools(name, entry, connection).pipe( - Effect.andThen(bus.publish(McpEvent.ToolsChanged, { server: name })), - ), - ), - ) - connection.onPromptsChanged(() => live(refreshPrompts(name, entry, connection))) - connection.onResourcesChanged(() => live(bus.publish(McpEvent.ResourcesChanged, { server: name }))) - } - - const startServer = (name: ServerName, entry: ServerEntry) => - Effect.gen(function* () { - // Announce the handshake so connect() and credential reconnects don't show a stale - // disabled/failed status for the duration of the connection attempt. - entry.status = { status: "pending" } - yield* bus.publish(McpEvent.StatusChanged, { server: name }) - const scope = yield* Scope.fork(root) - entry.scope = scope - const authProvider = yield* connectProvider(entry) - const { McpClient } = yield* Effect.promise(() => import("./client.js")) - // List tools as part of connect so a failure here marks the server failed rather than - // leaving it connected with a silently empty tool list and no path to recover. - const load = McpClient.connect( - name, - entry.config, - location.directory, - authProvider, - elicitation, - options?.clientInfo, - ).pipe( - Effect.flatMap((connection) => connection.tools().pipe(Effect.map((tools) => ({ connection, tools })))), - // A stdio server is spawned on this location's execution plane, not the host's. - Effect.provideService(Environment.Service, environment), - Scope.provide(scope), - ) - const result = yield* ( - entry.config.type === "remote" ? endpointLoads.withLock(entry.config.url)(load) : load - ).pipe(Effect.exit) - if (Exit.isSuccess(result)) { - entry.client = result.value.connection - entry.tools = result.value.tools.map((tool) => toTool(name, entry, tool)) - entry.prompts = [] - entry.status = { status: "connected" } - watch(name, entry, result.value.connection) - yield* Effect.logInfo("mcp connected", { server: name, tools: entry.tools.length }) - // The tool registry reads on this event; a late-connecting server has no other way to appear. - yield* bus.publish(McpEvent.ToolsChanged, { server: name }) - yield* bus.publish(McpEvent.ResourcesChanged, { server: name }) - yield* bus.publish(McpEvent.StatusChanged, { server: name }) - whenLive(name, entry, result.value.connection)(refreshPrompts(name, entry, result.value.connection)) - 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", error: error.message } - : { status: "failed", error: error instanceof Error ? error.message : String(error) } - yield* Effect.logWarning("mcp connect failed", { server: name, status: entry.status }) - yield* bus.publish(McpEvent.StatusChanged, { server: name }) - }).pipe( - Effect.ensuring(entry.startup.open), - Effect.annotateLogs({ server: name, directory: location.directory, connectionID: crypto.randomUUID() }), - ) - - const stopServer = Effect.fnUntraced(function* (name: ServerName, entry: ServerEntry) { - const scope = entry.scope - if (!scope) return - entry.scope = undefined - entry.client = undefined - entry.tools = undefined - entry.prompts = undefined - yield* Scope.close(scope, Exit.void) - yield* bus.publish(McpEvent.ToolsChanged, { server: name }) - yield* bus.publish(McpEvent.ResourcesChanged, { server: name }) - yield* bus.publish(PromptsChanged, { server: name }) - }) - - const disposeServer = Effect.fnUntraced(function* (name: ServerName, entry: ServerEntry) { - yield* stopServer(name, entry) + const disposeServer = Effect.fnUntraced(function* (name: ServerName, entry: LocationEntry) { + yield* runtime.stop(name, entry) if (entry.integrationID) owned.delete(entry.integrationID) if (entry.registration) yield* entry.registration.dispose }) @@ -528,24 +518,14 @@ export const layer = (options?: Options) => const replaceServer = Effect.fnUntraced(function* (name: ServerName, serverConfig: Mcp.ServerConfig) { const previous = entries.get(name) if (previous) yield* disposeServer(name, previous) - const entry: ServerEntry = { + const entry: LocationEntry = { + owner: LOCATION, config: serverConfig, status: { status: "pending" }, startup: Latch.makeUnsafe(), } entries.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 }) - return - } - yield* startServer(name, entry) - }).pipe( - // Settle startup even when registration fails or replacement is interrupted, so readers cannot hang. - Effect.ensuring(entry.startup.open), - ) + yield* runtime.open(name, entry, register(name, entry)) }) const removeServer = Effect.fnUntraced(function* (name: ServerName) { @@ -557,48 +537,6 @@ export const layer = (options?: Options) => yield* bus.publish(McpEvent.StatusChanged, { server: name }) }) - // Session-scoped servers skip OAuth integration registration: integrations are keyed by name and - // URL, so a Session registration would share and then dispose another owner's credential wiring. - const addSessionServer = (sessionID: Session.ID, name: ServerName, serverConfig: Mcp.ServerConfig) => - Effect.gen(function* () { - const config = cloneConfig(serverConfig) - const servers = sessionEntries.get(sessionID) ?? new Map() - const previous = servers.get(name) - if (previous && isDeepStrictEqual(previous.config, config)) return - if (previous) yield* stopServer(name, previous) - const entry: ServerEntry = { config, sessionID, status: { status: "pending" }, startup: Latch.makeUnsafe() } - sessionEntries.set(sessionID, servers.set(name, entry)) - yield* ( - config.disabled - ? Effect.sync(() => { - entry.status = { status: "disabled" } - }) - : startServer(name, entry) - ).pipe(Effect.ensuring(entry.startup.open)) - }).pipe(locks.withLock(slot(name, sessionID))) - - const removeSessionServer = (sessionID: Session.ID, name: ServerName) => - Effect.gen(function* () { - const servers = sessionEntries.get(sessionID) - const entry = servers?.get(name) - if (!servers || !entry) return yield* new NotFoundError({ server: name }) - yield* stopServer(name, entry) - servers.delete(name) - if (servers.size === 0) sessionEntries.delete(sessionID) - }).pipe(locks.withLock(slot(name, sessionID))) - - fork( - bus.subscribe([SessionEvent.Deleted, SessionEvent.Moved]).pipe( - Stream.runForEach((event) => - Effect.forEach( - Array.from(sessionEntries.get(event.data.sessionID)?.keys() ?? []), - (name) => removeSessionServer(event.data.sessionID, name).pipe(Effect.ignore), - { discard: true }, - ), - ), - ), - ) - let applied: Map | undefined const overrides = new Map() const reconcileLock = Semaphore.makeUnsafe(1) @@ -607,6 +545,7 @@ export const layer = (options?: Options) => if (!applied && entries.size === 0) { for (const [name, server] of servers) { entries.set(name, { + owner: LOCATION, config: server, status: { status: "pending" }, startup: Latch.makeUnsafe(), @@ -623,7 +562,7 @@ export const layer = (options?: Options) => yield* bus.publish(McpEvent.StatusChanged, { server: name }) continue } - fork(startServer(name, entry).pipe(locks.withLock(name))) + fork(runtime.start(name, entry).pipe(locks.withLock(name))) } return } @@ -654,8 +593,8 @@ export const layer = (options?: Options) => const entry = entries.get(name) if (!entry || entry.integrationID !== integrationID) return if (entry.status.status === "disabled") return - yield* stopServer(name, entry) - yield* startServer(name, entry) + yield* runtime.stop(name, entry) + yield* runtime.start(name, entry) }).pipe(locks.withLock(name)) }) fork( @@ -695,14 +634,13 @@ export const layer = (options?: Options) => return Service.of({ transform: state.transform, reload: state.reload, - servers: Effect.fn("MCP.servers")(function* (sessionID) { - return (yield* visibleServers(sessionID)) + servers: Effect.fn("MCP.servers")(function* () { + return Array.from(entries) .toSorted(([a], [b]) => a.localeCompare(b)) .map(([name, entry]): ServerInfo => ({ name, status: entry.status, integrationID: entry.integrationID })) }), - add: Effect.fn("MCP.add")(function* (server, config, sessionID) { + add: Effect.fn("MCP.add")(function* (server, config) { const name = ServerName.make(server) - if (sessionID !== undefined) return yield* addSessionServer(sessionID, name, config) overrides.set(name, config) yield* state.reload() }), @@ -710,52 +648,37 @@ export const layer = (options?: Options) => const name = ServerName.make(server) yield* Effect.gen(function* () { const target = yield* requireServer(name) - yield* stopServer(name, target.entry) - yield* startServer(name, target.entry) + yield* runtime.stop(name, target.entry) + yield* runtime.start(name, target.entry) }).pipe(locks.withLock(name)) }), disconnect: Effect.fn("MCP.disconnect")(function* (server) { const name = ServerName.make(server) yield* Effect.gen(function* () { const target = yield* requireServer(name) - yield* stopServer(name, target.entry) + yield* runtime.stop(name, target.entry) target.entry.status = { status: "disabled" } yield* bus.publish(McpEvent.StatusChanged, { server: name }) }).pipe(locks.withLock(name)) }), - remove: Effect.fn("MCP.remove")(function* (server, sessionID) { + remove: Effect.fn("MCP.remove")(function* (server) { const name = ServerName.make(server) - if (sessionID !== undefined) return yield* removeSessionServer(sessionID, name) yield* requireServer(name) overrides.set(name, false) yield* state.reload() }), // Reads report what is connected now; servers still starting contribute once they publish a change. - tools: Effect.fn("MCP.tools")(function* (sessionID) { - return (yield* visibleServers(sessionID)) - .flatMap(([, entry]) => entry.tools ?? []) + tools: Effect.fn("MCP.tools")(function* () { + return Array.from(entries.values()) + .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, input.sessionID) - yield* target.entry.startup.await - if (!target.entry.client) - return yield* new ToolCallError({ - server: target.name, - tool: input.name, - message: "MCP server is not connected", - }) - const result = yield* recovering(target.name, target.entry, target.entry.client, (connection) => - connection.callTool({ name: input.name, args: input.args, sessionID: input.sessionID }), - ).pipe( - Effect.mapError( - (error) => new ToolCallError({ server: target.name, tool: input.name, message: error.message }), - ), - ) - return { ...result, server: target.name, tool: input.name } + const target = yield* requireServer(input.server) + return yield* runtime.callTool(target.name, target.entry, input) }), - instructions: Effect.fn("MCP.instructions")(function* (sessionID) { - return (yield* visibleServers(sessionID)) + instructions: Effect.fn("MCP.instructions")(function* () { + return Array.from(entries) .flatMap(([server, entry]) => { const instructions = entry.client?.instructions if (!instructions) return [] @@ -770,55 +693,28 @@ export const layer = (options?: Options) => }), prompt: Effect.fn("MCP.prompt")(function* (input) { const target = yield* requireServer(input.server) - yield* target.entry.startup.await - if (!target.entry.client) return undefined - const result = yield* recovering(target.name, target.entry, target.entry.client, (connection) => - connection.prompt({ name: input.name, args: input.args }), - ).pipe(Effect.orElseSucceed(() => undefined)) - if (!result) return undefined - return { ...result, server: target.name, name: input.name } + return yield* runtime.prompt(target.name, target.entry, input) }), - resourceCatalog: Effect.fn("MCP.resourceCatalog")(function* (sessionID) { - const empty = ResourceCatalog.make({ resources: [], templates: [] }) - const catalogs = yield* Effect.forEach( - yield* visibleServers(sessionID), - ([name, entry]) => - entry.client - ? loadCatalog(name, entry, entry.client).pipe(Effect.orElseSucceed(() => empty)) - : Effect.succeed(empty), - { concurrency: "unbounded" }, + resourceCatalog: Effect.fn("MCP.resourceCatalog")(function* () { + return mergeCatalogs( + yield* Effect.forEach(entries, ([name, entry]) => runtime.catalog(name, entry), { + concurrency: "unbounded", + }), ) - return mergeCatalogs(catalogs) }), resources: Effect.fn("MCP.resources")(function* (input) { - const target = yield* requireServer(input.server, input.sessionID) - yield* target.entry.startup.await - if (!target.entry.client) return ResourceCatalog.make({ resources: [], templates: [] }) - return mergeCatalogs([yield* loadCatalog(target.name, target.entry, target.entry.client)]) + const target = yield* requireServer(input.server) + return yield* runtime.resources(target.name, target.entry) }), readResource: Effect.fn("MCP.readResource")(function* (input) { - const target = yield* requireServer(input.server, input.sessionID) - yield* target.entry.startup.await - if (!target.entry.client) return undefined - const result = yield* recovering(target.name, target.entry, target.entry.client, (connection) => - connection.readResource({ uri: input.uri }), - ) - if (!result) return undefined - return ResourceContent.make({ - server: target.name, - uri: input.uri, - contents: result.contents.map((part) => - "text" in part - ? { type: "text", uri: part.uri, text: part.text, mimeType: part.mimeType } - : { type: "blob", uri: part.uri, blob: part.blob, mimeType: part.mimeType }, - ), - }) + const target = yield* requireServer(input.server) + return yield* runtime.readResource(target.name, target.entry, input.uri) }), }) }), ) -function mergeCatalogs(catalogs: ReadonlyArray) { +export function mergeCatalogs(catalogs: ReadonlyArray) { return ResourceCatalog.make({ resources: catalogs .flatMap((catalog) => catalog.resources) @@ -840,72 +736,8 @@ export function configured(options?: Options) { return makeLocationNode({ service: Service, layer: layer(options), - deps: [Location.node, Environment.node, Bus.node, Form.node, Integration.node, Credential.node, SessionStore.node], + deps: [Location.node, Environment.node, Bus.node, Form.node, Integration.node, Credential.node], }) } export const node = configured() - -// Schema `optional` strips undefined-valued properties on encode, so fields can assign -// optional properties directly instead of conditionally spreading them. -function toElicitationField(key: string, property: ElicitationProperty, required: boolean): Form.Field { - // Some servers emit machine titles like "string with format email"; prefer description/key over those. - const machineTitle = /^(boolean|string|number|integer|array|object)(\s+with\b.*|\s+in\b.*)?$/i - const title = - property.title && !machineTitle.test(property.title.trim()) ? property.title : (property.description ?? key) - const base = { - key, - title, - description: property.description === title ? undefined : property.description, - required: required || undefined, - } - switch (property.type) { - case "boolean": - return { ...base, type: "boolean", default: property.default } - case "number": - case "integer": - return { - ...base, - type: property.type, - minimum: property.minimum, - maximum: property.maximum, - default: property.default, - } - case "array": - return { - ...base, - type: "multiselect", - options: - "anyOf" in property.items - ? property.items.anyOf.map((option) => ({ value: option.const, label: option.title })) - : property.items.enum.map((value) => ({ value, label: value })), - custom: false, - minItems: property.minItems, - maxItems: property.maxItems, - default: property.default, - } - case "string": { - const options = - "oneOf" in property - ? property.oneOf.map((option) => ({ value: option.const, label: option.title })) - : "enum" in property - ? property.enum.map((value, index) => ({ - value, - label: ("enumNames" in property ? property.enumNames?.[index] : undefined) ?? value, - })) - : undefined - return { - ...base, - type: "string", - format: "format" in property ? property.format : undefined, - minLength: "minLength" in property ? property.minLength : undefined, - maxLength: "maxLength" in property ? property.maxLength : undefined, - default: property.default, - options, - custom: options ? false : undefined, - } - } - } -} - -type ElicitationProperty = McpClient.ElicitationFormParams["requestedSchema"]["properties"][string] diff --git a/packages/core/src/mcp/instructions.ts b/packages/core/src/mcp/instructions.ts index 0c89ba88d5a..f326341d161 100644 --- a/packages/core/src/mcp/instructions.ts +++ b/packages/core/src/mcp/instructions.ts @@ -1,11 +1,10 @@ export * as McpInstructions from "./instructions.js" -import type { Session } from "@opencode/schema/session" import { makeLocationNode } from "@opencode/util/effect/app-node" import { Context, Effect, Layer, Schema } from "effect" import { Permission } from "../permission.js" import { McpTool } from "../tool/mcp.js" -import { Mcp } from "./index.js" +import type { McpSession } from "./session.js" import { Instructions } from "../instructions/index.js" const Summary = Schema.Struct({ @@ -56,57 +55,53 @@ const update = (previous: ReadonlyArray, current: ReadonlyArray Effect.Effect + readonly load: ( + permissions: Permission.Ruleset, + view: Pick, + ) => Effect.Effect } export class Service extends Context.Service()("@opencode/McpInstructions") {} -export const layer = Layer.effect( +export const layer = Layer.succeed( Service, - Effect.gen(function* () { - const mcp = yield* Mcp.Service - - return Service.of({ - load: Effect.fn("McpInstructions.load")(function* (permissions, sessionID) { - const source = (value: ReadonlyArray | Instructions.Removed) => - Instructions.make>({ - key: Instructions.Key.make("core/mcp-guidance"), - codec: Schema.toCodecJson(Schema.Array(Summary)), - read: Effect.succeed(value), - render: { - initial: render, - changed: update, - removed: () => "MCP server instructions are no longer available.", - }, - }) - const [instructions, tools] = yield* Effect.all([mcp.instructions(sessionID), mcp.tools(sessionID)], { - concurrency: "unbounded", + Service.of({ + load: Effect.fn("McpInstructions.load")(function* (permissions, view) { + const source = (value: ReadonlyArray | Instructions.Removed) => + Instructions.make>({ + key: Instructions.Key.make("core/mcp-guidance"), + codec: Schema.toCodecJson(Schema.Array(Summary)), + read: Effect.succeed(value), + render: { + initial: render, + changed: update, + removed: () => "MCP server instructions are no longer available.", + }, }) - const canExecute = Permission.evaluate("execute", "*", permissions).effect !== "deny" - // Instructions are useful only when this Session can reach at least one server tool. - const visible = instructions - .flatMap((item) => { - const owned = tools.filter((tool) => tool.server === item.server) - const codemode = owned[0]?.codemode !== false - if (codemode && !canExecute) return [] - if ( - !owned.some( - (tool) => - Permission.evaluate(McpTool.name(tool.server, tool.name), "*", permissions).effect !== "deny", - ) + const tools = [...view.tools, ...view.owned.map((owned) => owned.tool)] + const canExecute = Permission.evaluate("execute", "*", permissions).effect !== "deny" + // Instructions are useful only when this Session can reach at least one server tool. + const visible = view.instructions + .flatMap((item) => { + const owned = tools.filter((tool) => tool.server === item.server) + const codemode = owned[0]?.codemode !== false + if (codemode && !canExecute) return [] + if ( + !owned.some( + (tool) => Permission.evaluate(McpTool.name(tool.server, tool.name), "*", permissions).effect !== "deny", ) - return [] - return [ - codemode - ? { server: item.server, instructions: item.instructions } - : { server: item.server, instructions: item.instructions, codemode: false as const }, - ] - }) - .toSorted((a, b) => a.server.localeCompare(b.server)) - return source(visible.length === 0 ? Instructions.removed : visible) - }), - }) + ) + return [] + return [ + codemode + ? { server: item.server, instructions: item.instructions } + : { server: item.server, instructions: item.instructions, codemode: false as const }, + ] + }) + .toSorted((a, b) => a.server.localeCompare(b.server)) + return source(visible.length === 0 ? Instructions.removed : visible) + }), }), ) -export const node = makeLocationNode({ service: Service, layer, deps: [Mcp.node] }) +export const node = makeLocationNode({ service: Service, layer, deps: [] }) diff --git a/packages/core/src/mcp/session.ts b/packages/core/src/mcp/session.ts new file mode 100644 index 00000000000..ed02a3e69bf --- /dev/null +++ b/packages/core/src/mcp/session.ts @@ -0,0 +1,229 @@ +export * as McpSession from "./session.js" + +import type { Session } from "@opencode/schema/session" +import { SessionEvent } from "@opencode/schema/session-event" +import { isDeepStrictEqual } from "node:util" +import { Context, Effect, Latch, Layer, Stream } from "effect" +import { makeLocationNode } from "@opencode/util/effect/app-node" +import { Bus } from "../bus.js" +import { Credential } from "../credential.js" +import { KeyedMutex } from "../effect/keyed-mutex.js" +import { Environment } from "../environment/index.js" +import { Form } from "../form.js" +import { Location } from "../location.js" +import { SessionStore } from "../session/store.js" +import { Mcp } from "./index.js" + +type Owner = Extract +type Entry = Mcp.Entry +type Registry = { readonly servers: Map; readonly locks: KeyedMutex.KeyedMutex } + +export interface OwnedTool { + readonly tool: Mcp.Tool + readonly call: (input: { + readonly args?: Record + readonly sessionID: Session.ID + }) => Effect.Effect +} + +/** + * MCP capabilities visible to one Session: Location servers plus the servers registered for the Session + * or one of its ancestors, nearest owner first. A visible Session-scoped server shadows the Location + * server of the same name. + */ +export interface View { + readonly shadowed: ReadonlySet + readonly servers: ReadonlyArray + /** Tools of the visible Location servers. */ + readonly tools: ReadonlyArray + readonly owned: ReadonlyArray + readonly instructions: ReadonlyArray + readonly resourceCatalog: Effect.Effect + readonly resources: (server: string) => Effect.Effect + readonly readResource: (input: { + readonly server: string + readonly uri: string + }) => Effect.Effect +} + +/** Process-local MCP servers owned by a Session, released on removal or when the Session is deleted or moved away. */ +export interface Interface { + /** Re-adding an identical config is a no-op; a different config replaces only this Session's server. */ + readonly add: (sessionID: Session.ID, server: string, config: Mcp.ServerConfig) => Effect.Effect + readonly remove: (sessionID: Session.ID, server: string) => Effect.Effect + readonly view: (sessionID: Session.ID) => Effect.Effect +} + +export class Service extends Context.Service()("@opencode/McpSession") {} + +export const layer = (options?: Mcp.Options) => + Layer.effect( + Service, + Effect.gen(function* () { + const mcp = yield* Mcp.Service + const store = yield* SessionStore.Service + const bus = yield* Bus.Service + const location = yield* Location.Service + const registries = new Map() + + const withRegistry = ( + sessionID: Session.ID, + name: Mcp.ServerName, + effect: (servers: Map) => Effect.Effect, + ) => + Effect.suspend(() => { + const registry = registries.get(sessionID) ?? { servers: new Map(), locks: KeyedMutex.makeUnsafe() } + registries.set(sessionID, registry) + return registry.locks + .withLock(name)(effect(registry.servers)) + .pipe( + // Forget the Session once its last server is gone and no operation still waits on its locks. + Effect.ensuring( + Effect.gen(function* () { + if (registry.servers.size > 0 || (yield* registry.locks.size) > 0) return + if (registries.get(sessionID) === registry) registries.delete(sessionID) + }), + ), + ) + }) + + const runtime = yield* Mcp.makeRuntime({ + options, + lock: (name, owner) => (effect) => withRegistry(owner.id, name, () => effect), + }) + + const remove = (sessionID: Session.ID, server: string) => { + const name = Mcp.ServerName.make(server) + return withRegistry(sessionID, name, (servers) => + Effect.gen(function* () { + const entry = servers.get(name) + if (!entry) return yield* new Mcp.NotFoundError({ server: name }) + yield* runtime.stop(name, entry) + servers.delete(name) + }), + ) + } + + const visible = Effect.fnUntraced(function* (sessionID: Session.ID) { + const scoped = new Map() + if (registries.size === 0) return scoped + let current: Session.ID | undefined = sessionID + while (current) { + for (const [name, entry] of registries.get(current)?.servers ?? []) + if (!scoped.has(name)) scoped.set(name, entry) + current = (yield* store.get(current))?.parentID + } + return scoped + }) + + yield* bus.subscribe([SessionEvent.Deleted, SessionEvent.Moved]).pipe( + Stream.filter( + (event) => + event.type === "session.deleted" || + event.data.location.directory !== location.directory || + event.data.location.workspaceID !== location.workspaceID, + ), + Stream.runForEach((event) => + Effect.forEach( + Array.from(registries.get(event.data.sessionID)?.servers.keys() ?? []), + (name) => remove(event.data.sessionID, name).pipe(Effect.ignore), + { concurrency: "unbounded", discard: true }, + ), + ), + Effect.forkScoped({ startImmediately: true }), + ) + + return Service.of({ + add: Effect.fn("McpSession.add")(function* (sessionID, server, serverConfig) { + const name = Mcp.ServerName.make(server) + const config = Mcp.cloneConfig(serverConfig) + yield* withRegistry(sessionID, name, (servers) => + Effect.gen(function* () { + const previous = servers.get(name) + if (previous && isDeepStrictEqual(previous.config, config)) return + if (previous) yield* runtime.stop(name, previous) + const entry: Entry = { + owner: { type: "session", id: sessionID }, + config, + status: { status: "pending" }, + startup: Latch.makeUnsafe(), + } + servers.set(name, entry) + // Session-scoped servers are not registered as OAuth integrations: integration IDs derive from + // name and URL, so owners sharing one would share and prematurely dispose its registration. + yield* runtime.open(name, entry) + }), + ) + }), + remove: Effect.fn("McpSession.remove")(remove), + view: Effect.fn("McpSession.view")(function* (sessionID) { + const scoped = yield* visible(sessionID) + const local = yield* Effect.all({ + servers: mcp.servers(), + tools: mcp.tools(), + instructions: mcp.instructions(), + }) + const shadowed = new Set(scoped.keys()) + const owned = Array.from(scoped) + return { + shadowed, + servers: [ + ...local.servers.filter((server) => !shadowed.has(server.name)), + ...owned.map(([name, entry]): Mcp.ServerInfo => ({ name, status: entry.status })), + ].toSorted((a, b) => a.name.localeCompare(b.name)), + tools: local.tools.filter((tool) => !shadowed.has(tool.server)), + owned: owned + .flatMap(([name, entry]) => + (entry.tools ?? []).map( + (tool): OwnedTool => ({ + tool, + call: (input) => runtime.callTool(name, entry, { ...input, name: tool.name }), + }), + ), + ) + .toSorted((a, b) => a.tool.server.localeCompare(b.tool.server) || a.tool.name.localeCompare(b.tool.name)), + instructions: [ + ...local.instructions.filter((item) => !shadowed.has(item.server)), + ...owned.flatMap(([server, entry]) => + entry.client?.instructions ? [{ server, instructions: entry.client.instructions }] : [], + ), + ].toSorted((a, b) => a.server.localeCompare(b.server)), + resourceCatalog: Effect.all( + [mcp.resourceCatalog(), ...owned.map(([name, entry]) => runtime.catalog(name, entry))], + { concurrency: "unbounded" }, + ).pipe( + Effect.map(([catalog, ...catalogs]) => + Mcp.mergeCatalogs([ + { + resources: catalog.resources.filter((resource) => !shadowed.has(resource.server)), + templates: catalog.templates.filter((template) => !shadowed.has(template.server)), + }, + ...catalogs, + ]), + ), + ), + resources: (server) => { + const name = Mcp.ServerName.make(server) + const entry = scoped.get(name) + return entry ? runtime.resources(name, entry) : mcp.resources({ server }) + }, + readResource: (input) => { + const name = Mcp.ServerName.make(input.server) + const entry = scoped.get(name) + return entry ? runtime.readResource(name, entry, input.uri) : mcp.readResource(input) + }, + } + }), + }) + }), + ) + +export function configured(options?: Mcp.Options) { + return makeLocationNode({ + service: Service, + layer: layer(options), + deps: [Mcp.node, SessionStore.node, Location.node, Environment.node, Bus.node, Form.node, Credential.node], + }) +} + +export const node = configured() diff --git a/packages/core/src/plugin/internal.ts b/packages/core/src/plugin/internal.ts index 36fa55c94c0..ee95cbc7d3c 100644 --- a/packages/core/src/plugin/internal.ts +++ b/packages/core/src/plugin/internal.ts @@ -53,6 +53,7 @@ import { Location } from "../location.js" import { ManagedPolicy } from "../managed-policy.js" import { ModelsDev } from "../models-dev.js" import { Mcp } from "../mcp/index.js" +import { McpSession } from "../mcp/session.js" import { Npm } from "@opencode/util/npm" import { Permission } from "../permission.js" import { Reference } from "../reference.js" @@ -132,6 +133,7 @@ const services = [ ManagedPolicy.Service, ModelsDev.Service, Mcp.Service, + McpSession.Service, Npm.Service, Permission.Service, Form.Service, @@ -185,6 +187,7 @@ export const requirements = LayerNode.group([ ManagedPolicy.node, ModelsDev.node, Mcp.node, + McpSession.node, Npm.node, Permission.node, Form.node, diff --git a/packages/core/src/session/context.ts b/packages/core/src/session/context.ts index 9ebd84f9359..d1484f432a4 100644 --- a/packages/core/src/session/context.ts +++ b/packages/core/src/session/context.ts @@ -12,6 +12,7 @@ import { Instructions } from "../instructions/index.js" import { InstructionBuiltIns } from "../instructions/builtins.js" import { Location } from "../location.js" import { McpInstructions } from "../mcp/instructions.js" +import { McpSession } from "../mcp/session.js" import { McpTool } from "../tool/mcp.js" import { ReferenceInstructions } from "../reference/instructions.js" import { SkillInstructions } from "../skill/instructions.js" @@ -82,6 +83,7 @@ const layer = Layer.effect( const entries = yield* InstructionEntry.Service const location = yield* Location.Service const mcpInstructions = yield* McpInstructions.Service + const mcpSessions = yield* McpSession.Service const mcpTools = yield* McpTool.Service const models = yield* SessionRunnerModel.Service const request = yield* SessionModelRequest.Service @@ -129,14 +131,15 @@ const layer = Layer.effect( if (!agent.info) return yield* new AgentNotFoundError({ sessionID: session.id, agent: session.agent ?? agent.id }) // Session permissions narrow discovery the same way they narrow the tool snapshot. const permissions = Permission.merge(agent.info.permissions, session.permissions ?? []) + const mcp = yield* mcpSessions.view(sessionID) const loaded = yield* Effect.all( { - tools: mcpTools.overlay(sessionID).pipe(Effect.flatMap((overlay) => registry.snapshot(permissions, overlay))), + tools: mcpTools.overlay(mcp).pipe(Effect.flatMap((overlay) => registry.snapshot(permissions, overlay))), builtins: builtins.load(), discovery: discovery.load(), skills: skillInstructions.load(permissions), references: referenceInstructions.load(), - mcp: mcpInstructions.load(permissions, sessionID), + mcp: mcpInstructions.load(permissions, mcp), entries: entries.load(sessionID), }, { concurrency: "unbounded" }, @@ -197,6 +200,7 @@ export const node = makeLocationNode({ InstructionEntry.node, Location.node, McpInstructions.node, + McpSession.node, McpTool.node, ReferenceInstructions.node, SessionRunnerModel.node, diff --git a/packages/core/src/tool/AGENTS.md b/packages/core/src/tool/AGENTS.md index ff5f9512dfe..f9c74322046 100644 --- a/packages/core/src/tool/AGENTS.md +++ b/packages/core/src/tool/AGENTS.md @@ -43,7 +43,7 @@ The service uses shared `State` to replay synchronous transforms in registration MCP owns one stable tool transform that reads its latest discovered tools. Tool-list changes update that source and reload the tool state instead of re-registering at the end of the transform order. MCP refresh therefore preserves the precedence of later plugin overrides. -Session-scoped MCP servers never enter the registry. `McpTool.overlay(sessionID)` returns their tools plus the effective names of the Location MCP tools they shadow, and `Tool.Service.snapshot(permissions, overlay)` layers them over the registry for that snapshot only. Registry transforms, including plugin overrides, do not apply to overlay tools. +Session-scoped MCP servers never enter the registry. `McpTool.overlay(view)` turns a Session's `McpSession.View` into its owned tools plus the effective names of the Location MCP tools whose servers the view shadows, and `Tool.Service.snapshot(permissions, overlay)` layers them over the registry for that snapshot only. Registry transforms, including plugin overrides, do not apply to overlay tools. Type safety ends at registration. The registry validates model input and declared output at runtime and should not carry producer schema generics through storage or execution. diff --git a/packages/core/src/tool/mcp.ts b/packages/core/src/tool/mcp.ts index 1168bac6937..1be6877e04d 100644 --- a/packages/core/src/tool/mcp.ts +++ b/packages/core/src/tool/mcp.ts @@ -3,6 +3,7 @@ export * as McpTool from "./mcp.js" import { ToolFailure } from "@opencode/ai" import { McpEvent } from "@opencode/schema/mcp-event" import type { Session } from "@opencode/schema/session" +import type { McpSession } from "../mcp/session.js" import { Context, Effect, Fiber, type JsonSchema, Layer, PubSub, Semaphore, Stream } from "effect" import { makeLocationNode } from "@opencode/util/effect/app-node" import { Bus } from "../bus.js" @@ -20,8 +21,8 @@ export const name = (server: string, tool: string) => `${namespace(server)}_${to export interface Interface { /** Wait for the initial MCP tool registration to settle. */ readonly flush: Effect.Effect - /** Session-scoped server tools visible to the Session, hiding the Location server tools they shadow. */ - readonly overlay: (sessionID: Session.ID) => Effect.Effect + /** Session-owned server tools from a Session's MCP view, hiding the Location server tools they shadow. */ + readonly overlay: (view: Pick) => Effect.Effect } export class Service extends Context.Service()("@opencode/McpTool") {} @@ -36,9 +37,16 @@ export const layer = Layer.effect( const lock = Semaphore.makeUnsafe(1) let discovered: Mcp.Tool[] = [] - // Keyed by the MCP tool object so unchanged tools keep their identity across snapshots. + // Keyed by the MCP tool object so unchanged tools keep their identity across snapshots. A tool object + // belongs to exactly one server entry, so the first call bound to it stays valid. const infos = new WeakMap() - const info = (tool: Mcp.Tool) => { + const info = ( + tool: Mcp.Tool, + call: (input: { + readonly args: Record + readonly sessionID: Session.ID + }) => Effect.Effect, + ) => { const cached = infos.get(tool) if (cached) return cached const created: Tool.Info = { @@ -62,20 +70,16 @@ export const layer = Layer.effect( id: context.id, }, }) - const result = yield* mcp - .callTool({ - server: tool.server, - name: tool.name, - args: (input ?? {}) as Record, - sessionID: context.sessionID, - }) - .pipe( - Effect.catchTags({ - "MCP.NotFoundError": (error) => - new ToolFailure({ message: `MCP server "${error.server}" is not available` }), - "MCP.ToolCallError": (error) => new ToolFailure({ message: error.message }), - }), - ) + const result = yield* call({ + args: (input ?? {}) as Record, + sessionID: context.sessionID, + }).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: @@ -127,7 +131,8 @@ export const layer = Layer.effect( Effect.gen(function* () { discovered = yield* mcp.tools() yield* tools.transform((editor) => { - for (const tool of discovered) editor.add(info(tool)) + for (const tool of discovered) + editor.add(info(tool, (input) => mcp.callTool({ ...input, server: tool.server, name: tool.name }))) }) }), ) @@ -154,16 +159,16 @@ export const layer = Layer.effect( ) return Service.of({ flush: Effect.asVoid(Fiber.await(initial)), - overlay: Effect.fn("McpTool.overlay")(function* (sessionID) { - const visible = yield* mcp.tools(sessionID) - const owned = visible.filter((tool) => tool.sessionID !== undefined) - const shown = new Set( - visible.filter((tool) => tool.sessionID === undefined).map((tool) => name(tool.server, tool.name)), - ) - const hidden = new Set(discovered.map((tool) => name(tool.server, tool.name)).filter((id) => !shown.has(id))) - if (owned.length === 0 && hidden.size === 0) return undefined - return { tools: owned.map(info), hidden } - }), + overlay: (view) => + lock.withPermit( + Effect.sync(() => { + const hidden = new Set( + discovered.filter((tool) => view.shadowed.has(tool.server)).map((tool) => name(tool.server, tool.name)), + ) + if (view.owned.length === 0 && hidden.size === 0) return undefined + return { tools: view.owned.map((owned) => info(owned.tool, owned.call)), hidden } + }), + ), }) }), ) diff --git a/packages/core/src/tool/plugin/mcp-resource.ts b/packages/core/src/tool/plugin/mcp-resource.ts index 29bd7271f7d..cda4965cb81 100644 --- a/packages/core/src/tool/plugin/mcp-resource.ts +++ b/packages/core/src/tool/plugin/mcp-resource.ts @@ -4,12 +4,13 @@ import { ToolFailure } from "@opencode/ai" import type { Context } from "@opencode/plugin/effect/plugin" import { Effect, Schema } from "effect" import { Mcp } from "../../mcp/index.js" +import { McpSession } from "../../mcp/session.js" import { Permission } from "../../permission.js" export const Plugin = { id: "opencode.tools.mcp-resources", effect: Effect.fn("McpResourceTools.Plugin")(function* (ctx: Context) { - const mcp = yield* Mcp.Service + const mcp = yield* McpSession.Service const permission = yield* Permission.Service yield* ctx.tool @@ -30,11 +31,9 @@ export const Plugin = { }), execute: (input, context) => Effect.gen(function* () { + const view = yield* mcp.view(context.sessionID) // Listing every server checks each server name so per-server rules still apply. - const servers = - input.server === undefined - ? (yield* mcp.servers(context.sessionID)).map((server) => server.name) - : [input.server] + const servers = input.server === undefined ? view.servers.map((server) => server.name) : [input.server] yield* permission.assert({ action: "opencode_list_mcp_resources", resources: servers, @@ -44,8 +43,8 @@ export const Plugin = { agent: context.agent, source: { type: "tool", messageID: context.messageID, id: context.id }, }) - if (input.server === undefined) return { output: yield* mcp.resourceCatalog(context.sessionID) } - return { output: yield* mcp.resources({ server: input.server, sessionID: context.sessionID }) } + if (input.server === undefined) return { output: yield* view.resourceCatalog } + return { output: yield* view.resources(input.server) } }).pipe(Effect.mapError((error) => new ToolFailure({ message: error.message, error }))), }) editor.add({ @@ -73,7 +72,8 @@ export const Plugin = { agent: context.agent, source: { type: "tool", messageID: context.messageID, id: context.id }, }) - const resource = yield* mcp.readResource({ ...input, sessionID: context.sessionID }) + const view = yield* mcp.view(context.sessionID) + const resource = yield* view.readResource(input) if (!resource) return yield* new ToolFailure({ message: `MCP server "${input.server}" is not connected or does not expose resources`, diff --git a/packages/core/test/mcp-instructions.test.ts b/packages/core/test/mcp-instructions.test.ts index 906a6d78ca8..8801e6eeabe 100644 --- a/packages/core/test/mcp-instructions.test.ts +++ b/packages/core/test/mcp-instructions.test.ts @@ -1,6 +1,5 @@ import { describe, expect } from "bun:test" -import { Effect, Layer } from "effect" -import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder" +import { Effect } from "effect" import { Mcp } from "@opencode/core/mcp/index" import { McpInstructions } from "@opencode/core/mcp/instructions" import { McpTool } from "@opencode/core/tool/mcp" @@ -14,25 +13,28 @@ const schema = { type: "object" as const } const tool = (server: string, name = "search") => ({ server: Mcp.ServerName.make(server), name, inputSchema: schema }) satisfies Mcp.Tool -const layer = (catalog: () => Mcp.ServerInstructions[], tools: () => Mcp.Tool[]) => - AppNodeBuilder.build(McpInstructions.node, [ - Mcp.node.replace( - Layer.mock(Mcp.Service, { - instructions: () => Effect.succeed(catalog()), - tools: () => Effect.succeed(tools()), - }), - ), - ]) +const view = (instructions: Mcp.ServerInstructions[], tools: Mcp.Tool[]) => ({ instructions, tools, owned: [] }) describe("McpInstructions", () => { it.effect("renders instructions for servers with at least one permitted tool", () => Effect.gen(function* () { const service = yield* McpInstructions.Service const generation = yield* service - .load([ - { action: McpTool.name("alpha", "restricted"), resource: "*", effect: "deny" }, - { action: McpTool.name("hidden", "search"), resource: "*", effect: "deny" }, - ]) + .load( + [ + { action: McpTool.name("alpha", "restricted"), resource: "*", effect: "deny" }, + { action: McpTool.name("hidden", "search"), resource: "*", effect: "deny" }, + ], + view( + [ + instructions("beta", "Beta instructions"), + instructions("unused", "No tools"), + instructions("hidden", "Denied tool"), + instructions("alpha", "Alpha line one\nAlpha line two"), + ], + [tool("alpha"), tool("alpha", "restricted"), tool("beta"), tool("hidden")], + ), + ) .pipe(Effect.flatMap(readInitial)) expect(generation.text).toBe( @@ -50,44 +52,31 @@ describe("McpInstructions", () => { "", ].join("\n"), ) - }).pipe( - Effect.provide( - layer( - () => [ - instructions("beta", "Beta instructions"), - instructions("unused", "No tools"), - instructions("hidden", "Denied tool"), - instructions("alpha", "Alpha line one\nAlpha line two"), - ], - () => [tool("alpha"), tool("alpha", "restricted"), tool("beta"), tool("hidden")], - ), - ), - ), + }).pipe(Effect.provide(McpInstructions.layer)), ) it.effect("omits instructions when the agent cannot use execute", () => Effect.gen(function* () { const service = yield* McpInstructions.Service const generation = yield* service - .load([{ action: "execute", resource: "*", effect: "deny" }]) + .load( + [{ action: "execute", resource: "*", effect: "deny" }], + view([instructions("alpha", "Alpha instructions")], [tool("alpha")]), + ) .pipe(Effect.flatMap(readInitial)) expect(generation.text).toBe("") - }).pipe( - Effect.provide( - layer( - () => [instructions("alpha", "Alpha instructions")], - () => [tool("alpha")], - ), - ), - ), + }).pipe(Effect.provide(McpInstructions.layer)), ) it.effect("keeps MCP instructions when Code Mode is disabled and execute is denied", () => Effect.gen(function* () { const service = yield* McpInstructions.Service const generation = yield* service - .load([{ action: "execute", resource: "*", effect: "deny" }]) + .load( + [{ action: "execute", resource: "*", effect: "deny" }], + view([instructions("alpha", "Alpha instructions")], [{ ...tool("alpha"), codemode: false }]), + ) .pipe(Effect.flatMap(readInitial)) expect(generation.text).toBe( @@ -99,31 +88,19 @@ describe("McpInstructions", () => { "", ].join("\n"), ) - }).pipe( - Effect.provide( - layer( - () => [instructions("alpha", "Alpha instructions")], - () => [ - { - server: Mcp.ServerName.make("alpha"), - name: "search", - inputSchema: schema, - codemode: false, - } satisfies Mcp.Tool, - ], - ), - ), - ), + }).pipe(Effect.provide(McpInstructions.layer)), ) - it.effect("restates guidance when Code Mode is disabled for a server", () => { - let tools: Mcp.Tool[] = [tool("alpha")] - return Effect.gen(function* () { + it.effect("restates guidance when Code Mode is disabled for a server", () => + Effect.gen(function* () { const service = yield* McpInstructions.Service - const initialized = yield* service.load([]).pipe(Effect.flatMap(readInitial)) + const catalog = [instructions("alpha", "Alpha instructions")] + const initialized = yield* service.load([], view(catalog, [tool("alpha")])).pipe(Effect.flatMap(readInitial)) - tools = [{ ...tool("alpha"), codemode: false }] - const changed = yield* readUpdate(yield* service.load([]), initialized) + const changed = yield* readUpdate( + yield* service.load([], view(catalog, [{ ...tool("alpha"), codemode: false }])), + initialized, + ) expect(changed.text).toBe( [ "The available MCP server instructions have changed. This list supersedes the previous one.", @@ -134,25 +111,20 @@ describe("McpInstructions", () => { "", ].join("\n"), ) - }).pipe( - Effect.provide( - layer( - () => [instructions("alpha", "Alpha instructions")], - () => tools, - ), - ), - ) - }) + }).pipe(Effect.provide(McpInstructions.layer)), + ) - it.effect("renders additions, changes, and removal", () => { - let catalog = [instructions("alpha", "Alpha instructions")] - const tools = [tool("alpha"), tool("beta")] - return Effect.gen(function* () { + it.effect("renders additions, changes, and removal", () => + Effect.gen(function* () { const service = yield* McpInstructions.Service - const initialized = yield* service.load([]).pipe(Effect.flatMap(readInitial)) + const tools = [tool("alpha"), tool("beta")] + const load = (catalog: Mcp.ServerInstructions[]) => service.load([], view(catalog, tools)) + const initialized = yield* load([instructions("alpha", "Alpha instructions")]).pipe(Effect.flatMap(readInitial)) - catalog = [instructions("alpha", "Alpha instructions"), instructions("beta", "Beta instructions")] - const added = yield* readUpdate(yield* service.load([]), initialized) + const added = yield* readUpdate( + yield* load([instructions("alpha", "Alpha instructions"), instructions("beta", "Beta instructions")]), + initialized, + ) expect(added.text).toBe( [ "New MCP server instructions are available in addition to those previously listed:", @@ -163,8 +135,10 @@ describe("McpInstructions", () => { ].join("\n"), ) - catalog = [instructions("alpha", "Updated alpha"), instructions("beta", "Beta instructions")] - const changed = yield* readUpdate(yield* service.load([]), added) + const changed = yield* readUpdate( + yield* load([instructions("alpha", "Updated alpha"), instructions("beta", "Beta instructions")]), + added, + ) expect(changed.text).toBe( [ "The available MCP server instructions have changed. This list supersedes the previous one.", @@ -181,21 +155,12 @@ describe("McpInstructions", () => { ].join("\n"), ) - catalog = [instructions("beta", "Beta instructions")] - const removed = yield* readUpdate(yield* service.load([]), changed) + const removed = yield* readUpdate(yield* load([instructions("beta", "Beta instructions")]), changed) expect(removed.text).toBe("Instructions for the following MCP servers are no longer available: alpha.") - catalog = [] - expect((yield* readUpdate(yield* service.load([]), removed)).text).toBe( + expect((yield* readUpdate(yield* load([]), removed)).text).toBe( "MCP server instructions are no longer available.", ) - }).pipe( - Effect.provide( - layer( - () => catalog, - () => tools, - ), - ), - ) - }) + }).pipe(Effect.provide(McpInstructions.layer)), + ) }) diff --git a/packages/core/test/mcp-session.test.ts b/packages/core/test/mcp-session.test.ts index d011229dad0..fd57c60e0b3 100644 --- a/packages/core/test/mcp-session.test.ts +++ b/packages/core/test/mcp-session.test.ts @@ -1,7 +1,9 @@ import path from "node:path" +import { mkdir, rm } from "node:fs/promises" import { describe, expect } from "bun:test" -import { Context, Effect, Layer, Schedule } from "effect" +import { Context, Effect, Layer, Schedule, Stream } from "effect" import { ConfigMCP } from "@opencode/schema/config/mcp" +import { McpEvent } from "@opencode/schema/mcp-event" import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder" import { LayerNode } from "@opencode/util/effect/layer-node" import { Database } from "@opencode/core/database/database" @@ -10,6 +12,7 @@ import { Instance } from "@opencode/core/instance/service" import { Location } from "@opencode/core/location" import { LocationServiceMap } from "@opencode/core/location-services" import { Mcp } from "@opencode/core/mcp/index" +import { McpSession } from "@opencode/core/mcp/session" import { Project } from "@opencode/core/project" import { AbsolutePath } from "@opencode/core/schema" import { Session } from "@opencode/core/session" @@ -56,43 +59,53 @@ const it = testEffect( ), ) -const fixture = Effect.gen(function* () { - const temporary = yield* tmpdirScoped() - const location = Location.Ref.make({ directory: AbsolutePath.make(temporary.path) }) - const locations = yield* LocationServiceMap.Service - const context = yield* Effect.acquireRelease(locations.contextEffect(location), () => locations.invalidate(location)) - const server = (identity: string) => - new ConfigMCP.Local({ - type: "local", - command: [ - process.execPath, - path.join(import.meta.dir, "fixture/mcp-identity.ts"), - identity, - path.join(temporary.path, `${identity}.pid`), - ], - }) - const pid = (identity: string) => - Effect.promise(() => Bun.file(path.join(temporary.path, `${identity}.pid`)).text()).pipe(Effect.map(Number)) - return { - location, - server, - pid, - mcp: Context.get(context, Mcp.Service), - mcpTools: Context.get(context, McpTool.Service), - registry: Context.get(context, Tool.Service), - } -}) +const fixture = (subdirectory?: string) => + Effect.gen(function* () { + const temporary = yield* tmpdirScoped() + const directory = subdirectory ? path.join(temporary.path, subdirectory) : temporary.path + if (subdirectory) yield* Effect.promise(() => mkdir(directory)) + const location = Location.Ref.make({ directory: AbsolutePath.make(directory) }) + const locations = yield* LocationServiceMap.Service + const context = yield* Effect.acquireRelease(locations.contextEffect(location), () => + locations.invalidate(location), + ) + const pidFile = (identity: string) => path.join(temporary.path, `${identity}.pid`) + return { + root: AbsolutePath.make(temporary.path), + directory, + location, + server: (identity: string) => + new ConfigMCP.Local({ + type: "local", + command: [ + process.execPath, + path.join(import.meta.dir, "fixture/mcp-identity.ts"), + identity, + pidFile(identity), + ], + }), + pid: (identity: string) => Effect.promise(() => Bun.file(pidFile(identity)).text()).pipe(Effect.map(Number)), + mcp: Context.get(context, Mcp.Service), + sessions: Context.get(context, McpSession.Service), + mcpTools: Context.get(context, McpTool.Service), + registry: Context.get(context, Tool.Service), + } + }) -const whoami = (mcp: Mcp.Interface, sessionID?: Session.ID) => - mcp - .callTool({ server: "ctx", name: "whoami", sessionID }) - .pipe(Effect.map((result) => result.content.flatMap((part) => (part.type === "text" ? [part.text] : [])).join(""))) +const text = (result: Mcp.ToolResult) => + result.content.flatMap((part) => (part.type === "text" ? [part.text] : [])).join("") // The harness still reads the user's global config, so assertions only look at the fixture's server. -const toolNames = (tools: ReadonlyArray) => +const names = (tools: ReadonlyArray) => tools.filter((tool) => tool.server === "ctx").map((tool) => `${tool.server}.${tool.name}`) -const serverNames = (servers: ReadonlyArray) => - servers.map((server) => server.name).filter((name) => name === "ctx") +const owned = (view: McpSession.View) => names(view.owned.map((item) => item.tool)) + +const whoami = (view: McpSession.View, sessionID: Session.ID) => + Effect.gen(function* () { + const tool = view.owned.find((item) => item.tool.server === "ctx" && item.tool.name === "whoami") + if (!tool) return yield* Effect.die(new Error("whoami is not owned by this view")) + return text(yield* tool.call({ sessionID })) + }) const running = (pid: number) => { try { @@ -111,37 +124,44 @@ const exited = (pid: number) => describe("Session-scoped MCP servers", () => { it.live("are visible only to their Session tree and shadow Location servers there", () => Effect.gen(function* () { - const test = yield* fixture + const test = yield* fixture() const sessions = yield* Session.Service const root = yield* sessions.create({ location: test.location }) const child = yield* sessions.create({ parentID: root.id }) const other = yield* sessions.create({ location: test.location }) const bystander = yield* sessions.create({ location: test.location }) + const broken = yield* sessions.create({ location: test.location }) - yield* test.mcp.add("ctx", test.server("root"), root.id) + yield* test.sessions.add(root.id, "ctx", test.server("root")) - expect(toolNames(yield* test.mcp.tools(root.id))).toEqual(["ctx.only_root", "ctx.whoami"]) - expect(toolNames(yield* test.mcp.tools(child.id))).toEqual(["ctx.only_root", "ctx.whoami"]) - expect(toolNames(yield* test.mcp.tools(other.id))).toEqual([]) - expect(toolNames(yield* test.mcp.tools())).toEqual([]) - expect(serverNames(yield* test.mcp.servers())).toEqual([]) - expect(serverNames(yield* test.mcp.servers(child.id))).toEqual(["ctx"]) - expect(yield* whoami(test.mcp, child.id)).toBe("root") - expect(yield* whoami(test.mcp, other.id).pipe(Effect.flip)).toBeInstanceOf(Mcp.NotFoundError) + expect(owned(yield* test.sessions.view(root.id))).toEqual(["ctx.only_root", "ctx.whoami"]) + expect(owned(yield* test.sessions.view(child.id))).toEqual(["ctx.only_root", "ctx.whoami"]) + expect(owned(yield* test.sessions.view(other.id))).toEqual([]) + expect(names(yield* test.mcp.tools())).toEqual([]) + expect((yield* test.mcp.servers()).some((server) => server.name === "ctx")).toBe(false) + expect((yield* test.sessions.view(child.id)).servers.some((server) => server.name === "ctx")).toBe(true) + expect(yield* whoami(yield* test.sessions.view(child.id), child.id)).toBe("root") - yield* test.mcp.add("ctx", test.server("other"), other.id) + yield* test.sessions.add(other.id, "ctx", test.server("other")) + yield* test.sessions.add(broken.id, "ctx", new ConfigMCP.Local({ type: "local", command: ["false"] })) yield* test.mcp.add("ctx", test.server("location")) - expect(yield* whoami(test.mcp, root.id)).toBe("root") - expect(yield* whoami(test.mcp, other.id)).toBe("other") - expect(yield* whoami(test.mcp, bystander.id)).toBe("location") - expect(yield* whoami(test.mcp)).toBe("location") - expect(toolNames(yield* test.mcp.tools(root.id))).toEqual(["ctx.only_root", "ctx.whoami"]) - expect(toolNames(yield* test.mcp.tools())).toEqual(["ctx.only_location", "ctx.whoami"]) + expect(yield* whoami(yield* test.sessions.view(root.id), root.id)).toBe("root") + expect(yield* whoami(yield* test.sessions.view(other.id), other.id)).toBe("other") + expect(text(yield* test.mcp.callTool({ server: "ctx", name: "whoami", sessionID: bystander.id }))).toBe( + "location", + ) + const rootView = yield* test.sessions.view(root.id) + expect(rootView.shadowed.has("ctx")).toBe(true) + expect(names(rootView.tools)).toEqual([]) + const bystanderView = yield* test.sessions.view(bystander.id) + expect(bystanderView.shadowed.size).toBe(0) + expect(names(bystanderView.tools)).toEqual(["ctx.only_location", "ctx.whoami"]) yield* waitForCodeModeTool(test.registry, "ctx.only_location") const listed = (sessionID: Session.ID) => - test.mcpTools.overlay(sessionID).pipe( + test.sessions.view(sessionID).pipe( + Effect.flatMap(test.mcpTools.overlay), Effect.flatMap((overlay) => test.registry.snapshot(undefined, overlay)), Effect.map((snapshot) => (snapshot.codeModeCatalog ? codeModeListings(snapshot.codeModeCatalog) : []) @@ -151,40 +171,73 @@ describe("Session-scoped MCP servers", () => { ) expect(yield* listed(child.id)).toEqual(["ctx.only_root", "ctx.whoami"]) expect(yield* listed(bystander.id)).toEqual(["ctx.only_location", "ctx.whoami"]) + expect(yield* listed(broken.id)).toEqual([]) }), ) - it.live("releases servers on replacement, removal, and Session deletion", () => + it.live("releases servers on replacement, removal, and Session deletion without Location events", () => Effect.gen(function* () { - const test = yield* fixture + const test = yield* fixture() const sessions = yield* Session.Service + const bus = yield* Bus.Service + const events: string[] = [] + yield* bus + .subscribe([McpEvent.StatusChanged, McpEvent.ToolsChanged, McpEvent.ResourcesChanged, Mcp.PromptsChanged]) + .pipe( + Stream.runForEach((event) => Effect.sync(() => events.push(`${event.type}:${event.data.server}`))), + Effect.forkScoped({ startImmediately: true }), + ) const first = yield* sessions.create({ location: test.location }) const second = yield* sessions.create({ location: test.location }) - yield* test.mcp.add("ctx", test.server("first"), first.id) - yield* test.mcp.add("ctx", test.server("second"), second.id) + yield* test.sessions.add(first.id, "ctx", test.server("first")) + yield* test.sessions.add(second.id, "ctx", test.server("second")) const firstPid = yield* test.pid("first") const secondPid = yield* test.pid("second") - yield* test.mcp.add("ctx", test.server("first"), first.id) + yield* test.sessions.add(first.id, "ctx", test.server("first")) expect(yield* test.pid("first")).toBe(firstPid) expect(running(firstPid)).toBe(true) - yield* test.mcp.add("ctx", test.server("replaced"), first.id) + yield* test.sessions.add(first.id, "ctx", test.server("replaced")) yield* exited(firstPid) - expect(yield* whoami(test.mcp, first.id)).toBe("replaced") - expect(yield* whoami(test.mcp, second.id)).toBe("second") + expect(yield* whoami(yield* test.sessions.view(first.id), first.id)).toBe("replaced") + expect(yield* whoami(yield* test.sessions.view(second.id), second.id)).toBe("second") expect(running(secondPid)).toBe(true) const replacedPid = yield* test.pid("replaced") - yield* test.mcp.remove("ctx", first.id) + yield* test.sessions.remove(first.id, "ctx") yield* exited(replacedPid) - expect(toolNames(yield* test.mcp.tools(first.id))).toEqual([]) - expect(yield* test.mcp.remove("ctx", first.id).pipe(Effect.flip)).toBeInstanceOf(Mcp.NotFoundError) + expect(owned(yield* test.sessions.view(first.id))).toEqual([]) + expect(yield* test.sessions.remove(first.id, "ctx").pipe(Effect.flip)).toBeInstanceOf(Mcp.NotFoundError) yield* sessions.remove(second.id) yield* exited(secondPid) - expect(toolNames(yield* test.mcp.tools(second.id))).toEqual([]) + expect(owned(yield* test.sessions.view(second.id))).toEqual([]) + + yield* test.mcp.add("probe", new ConfigMCP.Local({ type: "local", command: ["unused"], disabled: true })) + yield* Effect.suspend(() => + events.includes(`${McpEvent.StatusChanged.type}:probe`) ? Effect.void : Effect.fail("probe event missing"), + ).pipe(Effect.retry({ times: 100, schedule: Schedule.spaced("10 millis") })) + expect(events.filter((event) => event.endsWith(":ctx"))).toEqual([]) + }), + ) + + it.live("releases servers in the former Location when their Session moves", () => + Effect.gen(function* () { + const test = yield* fixture("source") + const sessions = yield* Session.Service + const session = yield* sessions.create({ location: test.location }) + + yield* test.sessions.add(session.id, "ctx", test.server("moved")) + const pid = yield* test.pid("moved") + // A missing source directory applies the move immediately instead of at a step boundary. + yield* Effect.promise(() => rm(test.directory, { recursive: true })) + yield* sessions.move({ sessionID: session.id, directory: test.root }) + + expect((yield* sessions.get(session.id)).location.directory).toBe(test.root) + yield* exited(pid) + expect(owned(yield* test.sessions.view(session.id))).toEqual([]) }), ) }) diff --git a/packages/core/test/mcp.test.ts b/packages/core/test/mcp.test.ts index 18419b02415..600912a70a6 100644 --- a/packages/core/test/mcp.test.ts +++ b/packages/core/test/mcp.test.ts @@ -25,6 +25,7 @@ import { Environment } from "@opencode/core/environment/index" import { EnvironmentUnavailable } from "@opencode/core/environment/unavailable" import { Location } from "@opencode/core/location" import { Mcp } from "@opencode/core/mcp/index" +import { McpSession } from "@opencode/core/mcp/session" import { McpClient } from "@opencode/core/mcp/client" import { McpStdio } from "@opencode/core/mcp/stdio" import { Permission } from "@opencode/core/permission" @@ -284,6 +285,7 @@ function resourceMcpLayer( yield* ConfigMcpPlugin.register(bus.subscribe()) }), ).pipe( + Layer.provideMerge(McpSession.layer(options)), Layer.provideMerge(Mcp.layer(options)), Layer.provideMerge(Form.layer), Layer.provide( @@ -349,7 +351,7 @@ function resourceMcpLayer( }, }), Layer.mock(Credential.Service, {}), - Layer.mock(SessionStore.Service, {}), + AppNodeBuilder.build(SessionStore.node), overrides?.environment ?? hostEnvironmentLayer, ), ), @@ -1991,17 +1993,8 @@ testEffect(Layer.empty).live("keeps MCP config snapshots stable during an in-fli const shutdownIt = testEffect( AppNodeBuilder.build( - LayerNode.group([ - Bus.node, - Integration.node, - Credential.node, - Form.node, - Environment.node, - Location.node, - SessionStore.node, - ]), + LayerNode.group([Bus.node, Integration.node, Credential.node, Form.node, Environment.node, Location.node]), [ - SessionStore.node.replace(Layer.mock(SessionStore.Service, {})), Location.node.replace( Layer.succeed( Location.Service, diff --git a/packages/server/src/handlers/mcp.ts b/packages/server/src/handlers/mcp.ts index 47933234585..d96f56b11b4 100644 --- a/packages/server/src/handlers/mcp.ts +++ b/packages/server/src/handlers/mcp.ts @@ -5,7 +5,7 @@ import { HttpApiBuilder, HttpApiSchema } from "effect/unstable/httpapi" import { Api } from "../api" import { response } from "../location" -const notFound = (effect: Effect.Effect) => +export const notFound = (effect: Effect.Effect) => effect.pipe(Effect.mapError((error) => new McpServerNotFoundError({ server: error.server, message: error.message }))) export const McpHandler = HttpApiBuilder.group(Api, "server.mcp", (handlers) => diff --git a/packages/server/src/handlers/session.ts b/packages/server/src/handlers/session.ts index f5f355450c1..b2e6e409162 100644 --- a/packages/server/src/handlers/session.ts +++ b/packages/server/src/handlers/session.ts @@ -4,7 +4,7 @@ import { SessionTitle } from "@opencode/core/session/title" import { SessionTransfer } from "@opencode/core/session/transfer" import { InstructionEntry } from "@opencode/core/session/instruction-entry" import { Form } from "@opencode/core/form" -import { Mcp } from "@opencode/core/mcp/index" +import { McpSession } from "@opencode/core/mcp/session" import { DateTime, Effect, Stream } from "effect" import { HttpApiBuilder, HttpApiSchema } from "effect/unstable/httpapi" import { Api } from "../api" @@ -18,7 +18,6 @@ import { FormNotFoundError, InvalidRequestError, InvalidCursorError, - McpServerNotFoundError, MessageNotFoundError, ServiceUnavailableError, SessionBusyError, @@ -26,6 +25,7 @@ import { } from "@opencode/protocol/errors" import { AbsolutePath } from "@opencode/core/schema" import { failedMessageDecode, failedSnapshot, missingMessage, missingSession } from "./session-error" +import { notFound } from "./mcp" const DefaultSessionsLimit = 50 @@ -588,22 +588,16 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl .handle( "session.mcp.add", Effect.fn(function* (ctx) { - const mcp = yield* Mcp.Service - yield* mcp.add(ctx.params.server, ctx.payload.config, ctx.params.sessionID) + const mcp = yield* McpSession.Service + yield* mcp.add(ctx.params.sessionID, ctx.params.server, ctx.payload.config) return HttpApiSchema.NoContent.make() }), ) .handle( "session.mcp.remove", Effect.fn(function* (ctx) { - const mcp = yield* Mcp.Service - yield* mcp - .remove(ctx.params.server, ctx.params.sessionID) - .pipe( - Effect.mapError( - (error) => new McpServerNotFoundError({ server: error.server, message: error.message }), - ), - ) + const mcp = yield* McpSession.Service + yield* notFound(mcp.remove(ctx.params.sessionID, ctx.params.server)) return HttpApiSchema.NoContent.make() }), ) diff --git a/packages/server/src/routes.ts b/packages/server/src/routes.ts index c4f746f1d82..355bbd25d76 100644 --- a/packages/server/src/routes.ts +++ b/packages/server/src/routes.ts @@ -21,6 +21,7 @@ import { SessionTransfer } from "@opencode/core/session/transfer" import { ShellSelect } from "@opencode/core/shell/select" import { Job } from "@opencode/core/job" import { Mcp } from "@opencode/core/mcp/index" +import { McpSession } from "@opencode/core/mcp/session" import { Global } from "@opencode/util/global" import { InstructionDiscovery } from "@opencode/core/instruction-discovery" import { LocationServiceMap } from "@opencode/core/location-service-map" @@ -112,6 +113,7 @@ function makeRoutes( overrides: LayerNode.Replacements, instances?: InstanceNode, ) { + const clientInfo = { name: options.app?.name ?? "opencode", version: options.app?.version ?? "unknown" } const standard: LayerNode.Replacements = [ Database.node.replace(Database.configured(options.database)), PersistentPty.node.replace(PersistentPty.configured(options.pty)), @@ -130,14 +132,8 @@ function makeRoutes( ), InstructionDiscovery.node.replace(InstructionDiscovery.configured({ project: options.config?.project })), ShellSelect.node.replace(ShellSelect.configured({ gitbash: options.windows?.gitbash })), - Mcp.node.replace( - Mcp.configured({ - clientInfo: { - name: options.app?.name ?? "opencode", - version: options.app?.version ?? "unknown", - }, - }), - ), + Mcp.node.replace(Mcp.configured({ clientInfo })), + McpSession.node.replace(McpSession.configured({ clientInfo })), ] const build = (overrides: LayerNode.Replacements) => { const replacements: LayerNode.Replacements = [