diff --git a/packages/codework/test/__artifacts__/cli.mcp.json b/packages/codework/test/__artifacts__/cli.mcp.json new file mode 100644 index 00000000..4110475a --- /dev/null +++ b/packages/codework/test/__artifacts__/cli.mcp.json @@ -0,0 +1,19 @@ +{ + "requests": 2, + "declared": { + "name": "mcp__fixture__echo", + "description": "Echo text back, with the directory the server runs in.", + "parameters": { + "type": "object", + "properties": { + "text": { + "type": "string" + } + }, + "required": [ + "text" + ] + } + }, + "answer": "echo: from the cli (pid , cwd )" +} diff --git a/packages/codework/test/cli.mcp.test.ts b/packages/codework/test/cli.mcp.test.ts new file mode 100644 index 00000000..3df66f9f --- /dev/null +++ b/packages/codework/test/cli.mcp.test.ts @@ -0,0 +1,157 @@ +/* @effect-diagnostics nodeBuiltinImport:off -- this suite spawns the CLI and serves a fake provider. */ +/* @effect-diagnostics globalTimers:off -- the subprocess deadline runs outside any Effect. */ +import { spawn } from "node:child_process"; +import { mkdirSync, mkdtempSync, realpathSync, rmSync, writeFileSync } from "node:fs"; +import { createServer, type IncomingMessage } from "node:http"; +import type { AddressInfo } from "node:net"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { expect, it, onTestFinished } from "vite-plus/test"; + +const cli = fileURLToPath(new URL("../src/index.ts", import.meta.url)); +const models = fileURLToPath(new URL("../../../models.gen.json", import.meta.url)); +const fixture = fileURLToPath(new URL("./fixtures/mcp-echo.mjs", import.meta.url)); + +interface ChatRequest { + readonly tools?: ReadonlyArray<{ readonly function: { readonly name: string; readonly parameters: unknown } }>; + readonly messages: ReadonlyArray<{ readonly role: string; readonly content?: unknown }>; +} + +const read = (request: IncomingMessage) => + new Promise((resolve, reject) => { + const chunks: Buffer[] = []; + request.on("data", (chunk: Buffer) => chunks.push(chunk)); + request.on("end", () => resolve(Buffer.concat(chunks).toString("utf8"))); + request.on("error", reject); + }); + +const sse = (chunks: ReadonlyArray) => + `${chunks.map((chunk) => `data: ${JSON.stringify(chunk)}\n\n`).join("")}data: [DONE]\n\n`; +const chunk = (choice: object) => ({ + id: "chatcmpl-mcp", + object: "chat.completion.chunk", + created: 1, + model: "mcp-model", + choices: [{ index: 0, ...choice }], +}); + +/** The first request calls the MCP tool; the one carrying its result repeats that result as the answer. */ +const provider = async () => { + const requests: ChatRequest[] = []; + const server = createServer(async (request, response) => { + const body = JSON.parse(await read(request)) as ChatRequest; + requests.push(body); + response.writeHead(200, { "content-type": "text/event-stream" }); + const result = body.messages.find((message) => message.role === "tool"); + if (result !== undefined) { + const text = typeof result.content === "string" ? result.content : JSON.stringify(result.content); + response.end( + sse([ + chunk({ delta: { role: "assistant", content: `Tool said: ${text}` } }), + chunk({ delta: {}, finish_reason: "stop" }), + ]), + ); + return; + } + response.end( + sse([ + chunk({ + delta: { + role: "assistant", + tool_calls: [ + { + index: 0, + id: "call_mcp", + type: "function", + function: { name: "mcp__fixture__echo", arguments: '{"text":"from the cli"}' }, + }, + ], + }, + }), + chunk({ delta: {}, finish_reason: "tool_calls" }), + ]), + ); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + onTestFinished(() => new Promise((resolve) => server.close(() => resolve()))); + return { url: `http://127.0.0.1:${(server.address() as AddressInfo).port}`, requests }; +}; + +const runCli = (args: ReadonlyArray, options: { readonly cwd: string; readonly env: NodeJS.ProcessEnv }) => + new Promise<{ readonly status: number | null; readonly stdout: string; readonly stderr: string }>( + (resolve, reject) => { + const child = spawn(process.execPath, ["--conditions=development", cli, ...args], options); + let stdout = ""; + let stderr = ""; + child.stdout.on("data", (data: Buffer) => (stdout += data.toString("utf8"))); + child.stderr.on("data", (data: Buffer) => (stderr += data.toString("utf8"))); + const deadline = setTimeout(() => child.kill("SIGKILL"), 45_000); + child.on("error", reject); + child.on("close", (status) => { + clearTimeout(deadline); + resolve({ status, stdout, stderr }); + }); + }, + ); + +it("codework run calls a tool of a stdio MCP server from mcp.json, and stops the server on exit", async () => { + const home = mkdtempSync(join(tmpdir(), "codework-cli-mcp-")); + onTestFinished(() => rmSync(home, { recursive: true, force: true })); + const project = join(home, "project"); + const settings = join(home, "settings"); + mkdirSync(project); + mkdirSync(settings); + const fake = await provider(); + writeFileSync( + join(settings, "settings.jsonc"), + JSON.stringify({ + model: { provider: "mcp", id: "mcp-model" }, + models: [ + { + id: "mcp-model", + name: "mcp-model", + provider: { id: "mcp", name: "Mcp", source: "config", env: [] }, + baseUrl: `${fake.url}/v1`, + reasoning: false, + input: ["text"], + cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, + contextWindow: 32768, + maxTokens: 8192, + protocol: "openai-compatible", + }, + ], + }), + ); + // `mcp.json` sits next to the user settings the run reads. + writeFileSync( + join(settings, "mcp.json"), + JSON.stringify({ mcpServers: { fixture: { command: process.execPath, args: [fixture] } } }), + ); + + const result = await runCli( + ["--home", home, "--database", ":memory:", "--user-config-dir", settings, "run", "Echo something."], + { cwd: project, env: { ...process.env, HOME: home, CODEWORK_MODELS_FILE: models } }, + ); + expect(result.status, result.stderr).toBe(0); + + const answer = /echo: from the cli \(pid (\d+), cwd ([^)]+)\)/.exec(result.stdout); + expect(answer, result.stdout).not.toBeNull(); + const [, pid, cwd] = answer!; + // The server ran on this machine, in the session's directory, and is gone once the CLI exits. + expect(cwd).toBe(realpathSync(project)); + expect(() => process.kill(Number(pid), 0)).toThrow(); + + const declared = fake.requests[0]?.tools?.find((tool) => tool.function.name === "mcp__fixture__echo"); + await expect( + `${JSON.stringify( + { + requests: fake.requests.length, + declared: declared?.function, + answer: answer![0].replace(pid!, "").replace(cwd!, ""), + }, + null, + "\t", + )}\n`, + ).toMatchFileSnapshot("./__artifacts__/cli.mcp.json"); +}, 60_000); diff --git a/packages/codework/test/fixtures/mcp-echo.mjs b/packages/codework/test/fixtures/mcp-echo.mjs new file mode 100644 index 00000000..4bf5158d --- /dev/null +++ b/packages/codework/test/fixtures/mcp-echo.mjs @@ -0,0 +1,42 @@ +// A minimal stdio MCP server, written against the wire protocol rather than an SDK: one JSON-RPC +// message per line in, one per line out. It offers one tool, `echo`. +import { createInterface } from "node:readline"; + +const send = (message) => process.stdout.write(`${JSON.stringify({ jsonrpc: "2.0", ...message })}\n`); + +const answer = (method, params) => { + switch (method) { + case "initialize": + return { + protocolVersion: params.protocolVersion, + capabilities: { tools: {} }, + serverInfo: { name: "cli-fixture", version: "1.0.0" }, + }; + case "tools/list": + return { + tools: [ + { + name: "echo", + description: "Echo text back, with the directory the server runs in.", + inputSchema: { type: "object", properties: { text: { type: "string" } }, required: ["text"] }, + }, + ], + }; + case "tools/call": + return { + content: [ + { type: "text", text: `echo: ${params.arguments.text} (pid ${process.pid}, cwd ${process.cwd()})` }, + ], + }; + default: + return {}; + } +}; + +for await (const line of createInterface({ input: process.stdin })) { + if (!line.trim()) continue; + const message = JSON.parse(line); + // Notifications carry no id and get no answer. + if (message.id === undefined) continue; + send({ id: message.id, result: answer(message.method, message.params ?? {}) }); +} diff --git a/packages/harness/src/effect/harness.ts b/packages/harness/src/effect/harness.ts index f8566fa3..3c4f022f 100644 --- a/packages/harness/src/effect/harness.ts +++ b/packages/harness/src/effect/harness.ts @@ -1,4 +1,4 @@ -import { Effect, Layer, Option, Ref } from "effect"; +import { type Duration, Effect, Layer, Option, Ref } from "effect"; import { Context } from "../context/context.ts"; import { Control } from "../control.ts"; import { Database } from "../db/db.ts"; @@ -64,6 +64,8 @@ export interface Options { * Otherwise it is checked once, at boot. */ readonly watchModels?: boolean; + /** How long an MCP server no running session holds stays open. Default: 10 minutes. */ + readonly mcpIdleTimeout?: Duration.Input; } export const layer = (options: Options = {}) => @@ -256,7 +258,11 @@ export const layer = (options: Options = {}) => return Control.layer.pipe( Layer.provideMerge(RunnerExecute.layer.pipe(Layer.provide(loop))), Layer.provideMerge(State.layer({}, pool, references, followStore, rebuild, viewOf)), - Layer.provideMerge(Mcp.layer({ hostCwd, home: paths.home }).pipe(Layer.provide(Process.host))), + Layer.provideMerge( + Mcp.layer({ hostCwd, home: paths.home, idleTimeout: options.mcpIdleTimeout }).pipe( + Layer.provide(Process.host), + ), + ), Layer.provideMerge(Settings.layer(settingsOptions)), Layer.provideMerge(SessionRuntime.layer), Layer.provideMerge(sandboxes), diff --git a/packages/harness/src/mcp/config.ts b/packages/harness/src/mcp/config.ts index 80db70c7..b402f15f 100644 --- a/packages/harness/src/mcp/config.ts +++ b/packages/harness/src/mcp/config.ts @@ -24,6 +24,7 @@ const EXPOSURES: ReadonlyArray = ["codemode", "deferred", "direct", "hid /** Older exposure names, read as their current one. */ const EXPOSURE_ALIASES: Readonly> = { "codemode-deferred": "codemode" }; const alias = (value: unknown) => (Predicate.isString(value) ? (EXPOSURE_ALIASES[value] ?? value) : value); +const SCOPES: ReadonlyArray = ["session", "space", "app"]; const OVERRIDE_KEYS: ReadonlyArray = ["enabled", "exposure", "toolExposure"]; /** Names that differ only in `-` and `_` would give their tools the same names. */ @@ -48,7 +49,7 @@ export const validate = (name: string, raw: unknown): ServerConfig | string => { value.toolExposure = Object.fromEntries( Object.entries(raw.toolExposure).map(([tool, one]) => [tool, alias(one)]), ); - const { type, exposure, enabled, timeout, toolExposure, description } = value; + const { type, exposure, enabled, timeout, toolExposure, description, scope } = value; const exposures = EXPOSURES.map((one) => `"${one}"`).join(", "); if (exposure !== undefined && !EXPOSURES.includes(exposure as string)) return `server "${name}": exposure must be one of ${exposures}`; @@ -60,6 +61,8 @@ export const validate = (name: string, raw: unknown): ServerConfig | string => { } } if (enabled !== undefined && !Predicate.isBoolean(enabled)) return `server "${name}": enabled must be a boolean`; + if (scope !== undefined && !SCOPES.includes(scope as string)) + return `server "${name}": scope must be one of ${SCOPES.map((one) => `"${one}"`).join(", ")}`; if (description !== undefined && !Predicate.isString(description)) return `server "${name}": description must be a string`; if (timeout !== undefined && !(Predicate.isNumber(timeout) && timeout > 0)) diff --git a/packages/harness/src/mcp/content.ts b/packages/harness/src/mcp/content.ts index d223e002..bda217d1 100644 --- a/packages/harness/src/mcp/content.ts +++ b/packages/harness/src/mcp/content.ts @@ -3,6 +3,7 @@ import type { CallToolResult, ContentBlock } from "@modelcontextprotocol/client" import { Effect } from "effect"; import { Image } from "../tool/image/index.ts"; import { formatSize, truncateMiddle } from "../tool/truncate.ts"; +import type { OutputError } from "./error.ts"; export type Content = Message.TextContent | Message.ImageContent; @@ -13,7 +14,7 @@ export const MAX_BYTES = 20 * 1024; export const READ_RESOURCE_TOOL = "read_mcp_resource"; /** Keeps data the model is not shown in full in a host file; `extension` includes the dot. */ -export type Save = (data: string | Uint8Array, extension: string) => Effect.Effect; +export type Save = (data: string | Uint8Array, extension: string) => Effect.Effect; export interface Options { readonly save: Save; @@ -97,8 +98,8 @@ const fromBlock = Effect.fnUntraced(function* ( onSuccess: (path): ReadonlyArray => [ { type: "text", text: `[Binary resource ${uri} (${kind}) saved to ${path}]` }, ], - onFailure: (reason): ReadonlyArray => [ - { type: "text", text: `[Binary resource ${uri} (${kind}) could not be saved: ${reason}]` }, + onFailure: (error): ReadonlyArray => [ + { type: "text", text: `[Binary resource ${uri} (${kind}) could not be saved: ${error.message}]` }, ], }), ); @@ -117,7 +118,7 @@ export const limit = Effect.fnUntraced(function* (content: ReadonlyArray `[Full output: ${path} (read it with offset/limit)]`, - onFailure: (reason) => `[Could not save the full output: ${reason}]`, + onFailure: (error) => `[Could not save the full output: ${error.message}]`, }), ); const tokens = Math.ceil(cut.totalBytes / 4); diff --git a/packages/harness/src/mcp/error.ts b/packages/harness/src/mcp/error.ts index 12f3a49e..52d58cf4 100644 --- a/packages/harness/src/mcp/error.ts +++ b/packages/harness/src/mcp/error.ts @@ -21,15 +21,6 @@ export class SessionExpiredError extends Schema.TaggedError } } -/** The thread runs no MCP server of this name. */ -export class UnknownServerError extends Schema.TaggedError()("Mcp.UnknownServerError", { - server: Schema.String, -}) { - override get message(): string { - return `unknown MCP server: ${this.server}`; - } -} - /** A request on a connected server failed. */ export class RequestError extends Schema.TaggedError()("Mcp.RequestError", { server: Schema.String, @@ -37,6 +28,46 @@ export class RequestError extends Schema.TaggedError()("Mcp.Reques message: Schema.String, }) {} +/** + * The server cannot take the request: disabled, shut down, not connected, waiting for sign-in, or no + * longer offering the tool. `message` is what the model reads. + */ +export class UnavailableError extends Schema.TaggedError()("Mcp.UnavailableError", { + server: Schema.String, + message: Schema.String, +}) {} + +/** The thread has no MCP server of this name, or a turn replaced it meanwhile. */ +export class UnknownServerError extends Schema.TaggedError()("Mcp.UnknownServerError", { + server: Schema.String, + message: Schema.String, +}) {} + +/** A tool call or resource request reached the server and failed. `message` is what the model reads. */ +export class ToolCallError extends Schema.TaggedError()("Mcp.ToolCallError", { + server: Schema.String, + tool: Schema.String, + message: Schema.String, +}) {} + +/** A resource tool was called with arguments it cannot use, or for a server without resources. */ +export class ResourceError extends Schema.TaggedError()("Mcp.ResourceError", { + message: Schema.String, +}) {} + +/** A server's `env` or `headers` value could not be resolved. */ +export class ValueError extends Schema.TaggedError()("Mcp.ValueError", { + message: Schema.String, +}) {} + +/** Output the model is not shown in full could not be saved to a file. */ +export class OutputError extends Schema.TaggedError()("Mcp.OutputError", { + message: Schema.String, +}) {} + +/** What an MCP-backed tool can fail with; each says in `message` what happened. */ +export type Failure = UnavailableError | ToolCallError | ResourceError; + /** A failure inside the stdio transport, reported to the SDK through `onerror` or a rejected promise. */ export class StdioError extends Schema.TaggedError()("Mcp.StdioError", { message: Schema.String, diff --git a/packages/harness/src/mcp/mcp.ts b/packages/harness/src/mcp/mcp.ts index 6d88f71b..182b4852 100644 --- a/packages/harness/src/mcp/mcp.ts +++ b/packages/harness/src/mcp/mcp.ts @@ -1,11 +1,20 @@ /** - * The MCP servers each thread runs, kept connected across its turns. + * The MCP servers threads run, kept connected across their turns. + * + * A server's `scope` decides who shares a running instance of it: the threads of one session at the + * same place, every session in a space, or the whole app. An instance is keyed by that scope and by + * everything its connection is made from, so threads that resolve different settings never share + * one. * * Every turn hands over the thread's servers as its `mcp` plugins left them ({@link - * Interface.reconcile}); servers whose connection settings changed restart, removed ones close, new - * ones connect in the background. The tools a turn registers are the ones the servers offered at - * that moment, each with its exposure, frozen with the rest of the turn; the connection behind a - * tool is looked up when it is called: + * Interface.reconcile}); a thread moves to a new instance when a server's connection settings + * change, stops using one it no longer names, and new instances connect in the background. Exposure + * stays the thread's own: two threads can share an instance and show the model different tools. + * + * A running drain holds a {@link Interface.lease} on its thread's instances. An instance no thread + * uses any more closes at once; one no drain holds closes after an idle timeout. The tools a turn + * registers are the ones the servers offered at that moment, each with its exposure, frozen with the + * rest of the turn; the connection behind a tool is looked up when it is called: * * - A connection that drops fails the calls running on it, and the next call reconnects. * - Reconcile never restarts a failed server; only a config change or {@link Interface.connect} @@ -15,7 +24,8 @@ * - A tool call is never retried after the server may have run it. * - A changed tool list takes effect at the next turn; a call to a withdrawn tool says so. * - * Every state change is published to the owning thread as an ephemeral `mcp.server.status` event. + * Every state change of an instance is published, as an ephemeral `mcp.server.status` event, to + * each thread using it. * * Each connection lives in its own scope under the service's, so closing it, or shutting the * service down, stops the server process and ends the HTTP session. @@ -39,7 +49,14 @@ import { Semaphore, } from "effect"; import type { ChildProcessSpawner } from "effect/process"; -import { NeedsAuthError, UnknownServerError } from "./error.ts"; +import { + ConnectError, + type McpError, + NeedsAuthError, + ToolCallError, + UnavailableError, + UnknownServerError, +} from "./error.ts"; import { Event } from "../event/event.ts"; import { EventList } from "../event/list.ts"; import type { ID as SessionId } from "../session/schema.ts"; @@ -57,6 +74,8 @@ import { expandTilde } from "../util/home.ts"; /** A turn waits this long for the servers it started whose tools it declares or has loaded. */ const STARTUP_WAIT = Duration.seconds(10); +/** An instance no running drain holds closes after this long. */ +const IDLE_TIMEOUT = Duration.minutes(10); const DEFAULT_TIMEOUT_SECONDS = 60; export type State = EventList.McpServerState["state"]; @@ -74,12 +93,14 @@ export interface Input { readonly threadId: ThreadId; readonly servers: ReadonlyArray; /** - * Where stdio servers start (a server's own relative `cwd` resolves against it), and the root - * sent to servers: the session's directory when its sandbox is the host's own filesystem. Absent - * otherwise, since that directory does not exist here; servers then start in the session's host - * directory and are sent no roots. + * Where a `session` server starts (its own relative `cwd` resolves against it), and the root sent + * to it: the thread's directory when its sandbox is the host's own filesystem. Absent otherwise, + * since that directory does not exist here; servers then start in the session's host directory + * and are sent no roots. */ readonly directory?: string | undefined; + /** The thread's space; `root`, its folder, only when the space is on the host's own filesystem. */ + readonly space: { readonly id: string; readonly root?: string | undefined }; readonly hostDir?: string | undefined; /** Configuration problems from `mcp.json`, reported once per thread. */ readonly errors?: ReadonlyArray; @@ -90,12 +111,20 @@ export interface Input { export interface Interface { /** Settle a thread's servers for its next turn, and return every tool they offer, with its exposure. */ readonly reconcile: (input: Input) => Effect.Effect>; + /** Keep the thread's instances open for as long as the calling scope, a drain, runs. */ + readonly lease: (thread: { + readonly sessionId: SessionId; + readonly threadId: ThreadId; + }) => Effect.Effect; readonly status: (threadId: ThreadId) => Effect.Effect>; - /** Close the server's connection or attempt and connect it afresh, returning its state after that attempt. */ + /** + * Close the instance's connection or attempt and connect it afresh, for every thread sharing it, + * returning its state after that attempt. + */ readonly connect: (threadId: ThreadId, server: string) => Effect.Effect; - /** Close the server and keep it disconnected until {@link connect} or a config change. */ + /** Close the instance and keep it disconnected, for every thread sharing it, until {@link connect} or a config change. */ readonly disconnect: (threadId: ThreadId, server: string) => Effect.Effect; - /** Forget the threads of a deleted session and close their servers. */ + /** Forget the threads of a deleted session, and close the instances no other thread uses. */ readonly remove: (sessionId: SessionId) => Effect.Effect; } @@ -110,11 +139,12 @@ interface Live { refreshes: number; } +/** One running instance of a server, shared by the threads whose servers resolve to its key. */ interface Server { readonly name: string; - readonly thread: Thread; - config: ServerConfig; - /** Everything a connection is made from; a change restarts the server. */ + /** The config it was started with. Exposure is read from each thread's own config instead. */ + readonly config: ServerConfig; + /** The scope and everything a connection is made from. */ readonly key: string; readonly cwd: string; readonly roots: ReadonlyArray | undefined; @@ -125,7 +155,7 @@ interface Server { resources: boolean; live: Live | undefined; /** The connect attempt in progress. Callers share it; closing the server interrupts it. */ - attempt: Fiber.Fiber | undefined; + attempt: Fiber.Fiber | undefined; /** Settles when the first background attempt finishes. */ ready: Deferred.Deferred | undefined; closed: boolean; @@ -136,37 +166,81 @@ interface Server { readonly lock: Semaphore.Semaphore; /** Runs `connect` and `disconnect` one at a time. */ readonly control: Semaphore.Semaphore; + /** Threads whose servers resolve to this instance. */ + readonly users: Set; + /** Drains holding it open. */ + leases: number; + /** Bumped whenever the lease count changes, so an idle timer can tell it went stale. */ + generation: number; +} + +/** One of a thread's servers: its own config, and the instance it resolves to unless disabled. */ +interface Entry { + readonly config: ServerConfig; + readonly server: Server | undefined; } interface Thread { + readonly id: ThreadId; readonly sessionId: SessionId; - readonly threadId: ThreadId; - readonly servers: Map; + readonly servers: Map; readonly names: ReturnType; readonly lock: Semaphore.Semaphore; readonly reported: Set; + /** Whether the thread was told its `codemode` tools cannot be reached. */ + warnedUnreachable: boolean; + /** Running drains of this thread. */ + leases: number; } -/** The parts of a config that decide the connection; exposure and description do not. */ -const connectionKey = (config: ServerConfig, cwd: string, roots: ReadonlyArray | undefined) => { - const { exposure: _exposure, toolExposure: _tools, description: _description, enabled, ...rest } = config; - return JSON.stringify({ rest, enabled: enabled !== false, cwd, roots }); +/** JSON with object keys sorted, so the same settings written in another order give the same key. */ +const canonical = (value: unknown): string => + Array.isArray(value) + ? `[${value.map(canonical).join(",")}]` + : value !== null && typeof value === "object" + ? `{${Object.keys(value) + .sort() + .map((key) => `${JSON.stringify(key)}:${canonical((value as Record)[key])}`) + .join(",")}}` + : JSON.stringify(value); + +/** + * The scope, the server's name and the parts of its config that decide the connection; exposure and + * description do not. The name is part of it so that two entries are always two instances, and a + * thread's servers map one to one onto the instances it uses. + */ +const instanceKey = ( + scope: string, + name: string, + config: ServerConfig, + cwd: string, + roots: ReadonlyArray | undefined, +) => { + const { exposure: _exposure, toolExposure: _tools, description: _description, enabled: _enabled, ...rest } = config; + return canonical({ scope, name, rest, cwd, roots }); }; +const exposureOf = (config: ServerConfig) => config.exposure ?? "direct"; + const hasDirectTools = (config: ServerConfig) => - (config.exposure ?? "direct") === "direct" || + exposureOf(config) === "direct" || Object.values(config.toolExposure ?? {}).some((exposure) => exposure === "direct"); /** Whether a loaded tool name could belong to the server; its tools are not known before it connects. */ const loadsFrom = (server: string, loaded: ReadonlyArray) => loaded.some((name) => name.startsWith(`${namespace(server)}__`)); -const statusOf = (server: Server): Status => ({ - server: server.name, - state: server.state, - ...(server.error === undefined ? {} : { error: server.error }), - tools: server.tools.map((tool) => ({ name: tool.name, exposure: McpTool.exposure(server.config, tool.name) })), -}); +/** A thread's server as that thread sees it: the instance's state, with the thread's own exposure. */ +const statusOf = (name: string, entry: Entry): Status => { + const server = entry.server; + if (server === undefined) return { server: name, state: "disabled", tools: [] }; + return { + server: name, + state: server.closed ? "disconnected" : server.state, + ...(server.error === undefined ? {} : { error: server.error }), + tools: server.tools.map((tool) => ({ name: tool.name, exposure: McpTool.exposure(entry.config, tool.name) })), + }; +}; const eventStatus = (server: Server): EventList.McpServerState => { const { name, state } = server; @@ -184,7 +258,12 @@ const eventStatus = (server: Server): EventList.McpServerState => { } }; -export const layer = (options: { readonly hostCwd: string; readonly home: string }) => +export const layer = (options: { + readonly hostCwd: string; + readonly home: string; + /** How long an instance no drain holds stays open. Default: 10 minutes. */ + readonly idleTimeout?: Duration.Input | undefined; +}) => Layer.effect( Service, Effect.gen(function* () { @@ -194,10 +273,12 @@ export const layer = (options: { readonly hostCwd: string; readonly home: string // the rest. Made before the fibers below, so connect attempts are interrupted before it closes. const servers = yield* Scope.fork(root, "parallel"); const fork = yield* FiberSet.makeRuntime(); - const forkAttempt = yield* FiberSet.makeRuntime(); + const forkAttempt = yield* FiberSet.makeRuntime(); const context = yield* Effect.context(); const log = yield* McpLog.make(hostPath.join(options.home, "mcp.log")); const threads = new Map(); + const instances = new Map(); + const idleTimeout = options.idleTimeout ?? IDLE_TIMEOUT; // Shared remote endpoints should not see a burst of concurrent startups. const endpoints = new Map(); const endpointLock = (url: string) => { @@ -206,7 +287,42 @@ export const layer = (options: { readonly hostCwd: string; readonly home: string return lock; }; - /** Publish the server's status to its thread, unless it is the one published last. */ + /** Publish a status to each of these threads. */ + const publishTo = (users: Iterable, status: EventList.McpServerState) => + Effect.flatMap(DateTime.now, (timestamp) => + Effect.forEach( + [...users].flatMap((threadId) => { + const thread = threads.get(threadId); + return thread === undefined ? [] : [thread]; + }), + (thread) => + events.publish(EventList.McpServerStatus, { + timestamp, + sessionId: thread.sessionId, + threadId: thread.id, + status, + }), + { discard: true }, + ), + ); + + /** Tell a thread that joined a running instance its status, unless the instance published since. */ + const publishJoined = (thread: Thread, server: Server, seen: string | undefined) => + Effect.flatMap(DateTime.now, (timestamp) => + // Checked with nothing in between it and the publish, so no newer status can slip ahead. + seen === undefined || server.published !== seen + ? Effect.void + : Effect.asVoid( + events.publish(EventList.McpServerStatus, { + timestamp, + sessionId: thread.sessionId, + threadId: thread.id, + status: eventStatus(server), + }), + ), + ); + + /** Publish the instance's status to every thread using it, unless it is the one published last. */ const publish = (server: Server) => Effect.suspend(() => { const status = eventStatus(server); @@ -217,16 +333,10 @@ export const layer = (options: { readonly hostCwd: string; readonly home: string ].join("\0"); if (key === server.published) return Effect.void; server.published = key; - const { sessionId, threadId } = server.thread; - return DateTime.now.pipe( - Effect.flatMap((timestamp) => - events.publish(EventList.McpServerStatus, { timestamp, sessionId, threadId, status }), - ), - Effect.asVoid, - ); + return publishTo(server.users, status); }); - /** The one place a server's state changes. */ + /** The one place an instance's state changes. */ const setState = (server: Server, state: State, error?: string) => Effect.suspend(() => { server.state = state; @@ -312,7 +422,7 @@ export const layer = (options: { readonly hostCwd: string; readonly home: string }; /** One connect attempt, run apart from any caller so that closing the server can interrupt it. */ - const attempt = (server: Server): Effect.Effect => + const attempt = (server: Server): Effect.Effect => Effect.gen(function* () { const scope = yield* Scope.fork(servers, "sequential"); const config = server.config; @@ -339,15 +449,15 @@ export const layer = (options: { readonly hostCwd: string; readonly home: string const error = Cause.squash(exit.cause); const message = typeof error === "string" ? error : error instanceof Error ? error.message : String(error); - if (server.closed || server.stopped) return yield* Effect.fail(message); + if (server.closed || server.stopped) return yield* new ConnectError({ server: server.name, message }); yield* setState(server, Schema.is(NeedsAuthError)(error) ? "authenticate" : "failed", message); yield* Effect.logWarning("mcp connect failed", { server: server.name, error: message }); - return yield* Effect.fail(message); + return yield* new ConnectError({ server: server.name, message }); } // Closed or disconnected while connecting: the connection must not outlive that. if (server.closed || server.stopped) { yield* Scope.close(scope, Exit.void); - return yield* Effect.fail(halted(server)); + return yield* new ConnectError({ server: server.name, message: halted(server) }); } const live: Live = { connection: exit.value.connection, @@ -377,10 +487,11 @@ export const layer = (options: { readonly hostCwd: string; readonly home: string `MCP server "${server.name}" is ${server.closed ? "shut down" : server.stopped ? "disconnected" : "reconnecting"}`; /** The server's current connection, connecting first when it has none. */ - const open = (server: Server): Effect.Effect => + const open = (server: Server): Effect.Effect => Effect.gen(function* () { + const unavailable = () => new UnavailableError({ server: server.name, message: halted(server) }); const next = yield* Effect.gen(function* () { - if (server.closed || server.stopped) return yield* Effect.fail(halted(server)); + if (server.closed || server.stopped) return yield* unavailable(); const current = server.live; // A drop is known here before its `onClose` handler has run. const reason = current?.connection.dropped(); @@ -403,11 +514,11 @@ export const layer = (options: { readonly hostCwd: string; readonly home: string if (Exit.isSuccess(exit)) return exit.value; // Interrupted by `connect`, `disconnect` or closing; the server's state now says which. return yield* Cause.hasInterruptsOnly(exit.cause) - ? Effect.fail(halted(server)) + ? Effect.fail(unavailable()) : Effect.failCause(exit.cause); }); - /** Interrupt a connect in progress, and retire the connection the server has. */ + /** Interrupt a connect in progress, and retire the connection the instance has. */ const stop = (server: Server) => Effect.gen(function* () { const pending = server.attempt; @@ -415,11 +526,13 @@ export const layer = (options: { readonly hostCwd: string; readonly home: string if (server.live !== undefined) yield* retire(server, server.live); }).pipe(server.lock.withPermits(1)); - /** Stop the server for good: it was removed, its connection settings changed, or its session was deleted. */ + /** Stop the instance for good: nobody uses it any more, or it idled out. */ const close = (server: Server) => Effect.suspend(() => { + if (instances.get(server.key) === server) instances.delete(server.key); server.closed = true; - return stop(server); + // Threads still using an instance that idled out learn it is gone, as `status` already says. + return Effect.andThen(setState(server, "disconnected"), stop(server)); }); /** @@ -428,80 +541,89 @@ export const layer = (options: { readonly hostCwd: string; readonly home: string * runs once more on a new one; nothing else is retried here. */ const request = - /** `what` prefixes a failure, naming what failed; without it the server's message is the failure. */ - (thread: Thread, name: string, what: string | undefined) => - ( - run: (connection: McpClient.Connection) => Effect.Effect, - /** Why the server, as it stands, no longer offers what is asked; checked before every attempt. */ - unavailable: (server: Server) => string | undefined = () => undefined, - ): Effect.Effect => - Effect.gen(function* () { - for (let attempt = 1; ; attempt++) { - const server = thread.servers.get(name); - if (server === undefined || server.config.enabled === false) - return yield* Effect.fail(`MCP server "${name}" is disabled.`); - if (server.closed) return yield* Effect.fail(`MCP server "${name}" is shut down.`); - const reason = unavailable(server); - if (reason !== undefined) return yield* Effect.fail(reason); - const live = yield* open(server).pipe( - Effect.mapError((error) => `MCP server "${name}" is not connected: ${error}`), - ); - live.inflight += 1; - const exit = yield* run(live.connection).pipe( - Effect.exit, - Effect.ensuring( - Effect.suspend(() => { - live.inflight -= 1; - return release(live); - }), - ), - ); - if (Exit.isSuccess(exit)) return exit.value; - const error = Cause.findErrorOption(exit.cause); - if (error._tag === "None") return yield* Effect.failCause(exit.cause as Cause.Cause); - const failure = error.value; - if (failure._tag === "Mcp.SessionExpiredError" && attempt === 1) { - yield* retire(server, live).pipe(server.lock.withPermits(1)); - continue; - } - if (failure._tag === "Mcp.NeedsAuthError") { - yield* Effect.gen(function* () { - yield* setState(server, "authenticate", failure.message); - yield* retire(server, live); - }).pipe(server.lock.withPermits(1)); - return yield* Effect.fail(`MCP server "${name}" needs sign-in: ${failure.message}`); - } - return yield* Effect.fail( - what === undefined ? failure.message : `${what} failed: ${failure.message}`, - ); + (thread: Thread, name: string, failed: (message: string) => ToolCallError) => + ( + run: (connection: McpClient.Connection) => Effect.Effect, + /** Why the server, as it stands, no longer offers what is asked; checked before every attempt. */ + unavailable: (entry: Entry, server: Server) => string | undefined = () => undefined, + ): Effect.Effect => + Effect.gen(function* () { + const refuse = (message: string) => new UnavailableError({ server: name, message }); + for (let attempt = 1; ; attempt++) { + const entry = thread.servers.get(name); + const server = entry?.server; + if (entry === undefined || server === undefined) + return yield* refuse(`MCP server "${name}" is disabled.`); + if (server.closed) return yield* refuse(`MCP server "${name}" is shut down.`); + const reason = unavailable(entry, server); + if (reason !== undefined) return yield* refuse(reason); + const live = yield* open(server).pipe( + Effect.mapError((error) => refuse(`MCP server "${name}" is not connected: ${error.message}`)), + ); + live.inflight += 1; + const exit = yield* run(live.connection).pipe( + Effect.exit, + Effect.ensuring( + Effect.suspend(() => { + live.inflight -= 1; + return release(live); + }), + ), + ); + if (Exit.isSuccess(exit)) return exit.value; + const error = Cause.findErrorOption(exit.cause); + if (error._tag === "None") return yield* Effect.failCause(exit.cause as Cause.Cause); + const failure = error.value; + if (failure._tag === "Mcp.SessionExpiredError" && attempt === 1) { + yield* retire(server, live).pipe(server.lock.withPermits(1)); + continue; + } + if (failure._tag === "Mcp.NeedsAuthError") { + yield* Effect.gen(function* () { + yield* setState(server, "authenticate", failure.message); + yield* retire(server, live); + }).pipe(server.lock.withPermits(1)); + return yield* refuse(`MCP server "${name}" needs sign-in: ${failure.message}`); } - }); + return yield* failed(failure.message); + } + }); /** One call of one of the server's tools. A tool call is never retried after the server may have run it. */ const call = (thread: Thread, name: string, tool: string) => (run: McpTool.Run) => request( thread, name, - `MCP tool "${name}/${tool}"`, + (message) => + new ToolCallError({ server: name, tool, message: `MCP tool "${name}/${tool}" failed: ${message}` }), )( (connection) => connection.callTool({ name: tool, args: run.args, onProgress: run.onProgress }), - (server) => - McpTool.exposure(server.config, tool) === "hidden" || + (entry, server) => + McpTool.exposure(entry.config, tool) === "hidden" || (server.live !== undefined && !server.tools.some((offered) => offered.name === tool)) ? `MCP tool "${name}/${tool}" is no longer available.` : undefined, ); - /** Servers the resource tools reach: enabled, connectable, offering resources, and not hidden. */ - const withResources = (server: Server) => - server.config.enabled !== false && - !server.stopped && - server.resources && - (server.config.exposure ?? "direct") !== "hidden"; + /** Servers the resource tools reach: enabled, connectable, offering resources, and not hidden for this thread. */ + const withResources = (entry: Entry) => + entry.server !== undefined && + !entry.server.stopped && + entry.server.resources && + exposureOf(entry.config) !== "hidden"; const resourceServer = (thread: Thread, name: string): McpResources.Server => { - // Listings name the server beside each failure already. - const reach = request(thread, name, undefined); + const reach = request( + thread, + name, + (message) => + new ToolCallError({ + server: name, + tool: "resources", + // Listings name the server beside each failure already. + message, + }), + ); return { name, resources: (cursor) => reach((connection) => connection.resources(cursor)), @@ -513,146 +635,239 @@ export const layer = (options: { readonly hostCwd: string; readonly home: string }; /** Connect in the background; `ready` settles when the attempt does. */ - const start = (server: Server) => - Effect.suspend(() => { - if (server.config.enabled === false) return setState(server, "disabled"); - const ready = Deferred.makeUnsafe(); - server.ready = ready; - fork(open(server).pipe(Effect.ignore, Effect.ensuring(Deferred.succeed(ready, undefined)))); - return Effect.void; - }); + const start = (server: Server) => { + const ready = Deferred.makeUnsafe(); + server.ready = ready; + fork(open(server).pipe(Effect.ignore, Effect.ensuring(Deferred.succeed(ready, undefined)))); + }; + + /** Close the instance once it has gone the idle timeout with no drain holding it. */ + const arm = (server: Server) => { + server.generation += 1; + if (server.leases > 0 || server.closed) return; + const generation = server.generation; + fork( + Effect.sleep(idleTimeout).pipe( + Effect.andThen( + Effect.suspend(() => + server.leases === 0 && server.generation === generation && !server.closed + ? close(server).pipe( + Effect.andThen(Effect.logInfo("mcp server idle, closed", { server: server.name })), + ) + : Effect.void, + ), + ), + ), + ); + }; - const threadOf = (input: Input) => { - let thread = threads.get(input.threadId); + /** One more or one less drain holding the instance; the last one out starts its idle timer. */ + const hold = (server: Server, delta: number) => { + server.leases += delta; + arm(server); + }; + + /** The instances a thread uses, each once. */ + const used = (thread: Thread) => + new Set([...thread.servers.values()].flatMap((entry) => (entry.server ? [entry.server] : []))); + + /** The thread stops using these instances, and its drains stop holding them; returns the ones nobody uses now. */ + const leave = (thread: Thread, servers: Iterable) => { + const unused: Server[] = []; + for (const server of servers) { + server.users.delete(thread.id); + hold(server, -thread.leases); + if (server.users.size === 0) unused.push(server); + } + return unused; + }; + + const threadOf = (threadId: ThreadId, sessionId: SessionId) => { + let thread = threads.get(threadId); if (thread === undefined) { thread = { - sessionId: input.sessionId, - threadId: input.threadId, + id: threadId, + sessionId, + leases: 0, servers: new Map(), names: McpTool.names(), lock: Semaphore.makeUnsafe(1), reported: new Set(), + warnedUnreachable: false, }; - threads.set(input.threadId, thread); + threads.set(threadId, thread); } return thread; }; - const serverOf = (threadId: ThreadId, name: string) => { - const server = threads.get(threadId)?.servers.get(name); - return server === undefined - ? Effect.fail(new UnknownServerError({ server: name })) - : Effect.succeed(server); - }; - const reconcile = Effect.fn("Mcp.reconcile")(function* (input: Input) { - const thread = threadOf(input); + const thread = threadOf(input.threadId, input.sessionId); for (const error of input.errors ?? []) { if (thread.reported.has(error)) continue; thread.reported.add(error); yield* Effect.logWarning(`mcp config: ${error}`); } - const cwd = input.directory ?? input.hostDir ?? options.hostCwd; - const roots = input.directory === undefined ? undefined : [input.directory]; - const wanted = new Map(input.servers); - const started: Server[] = []; - yield* Effect.gen(function* () { - for (const [name, server] of thread.servers) { - const config = wanted.get(name); - // Exposure and description apply to the next turn without reconnecting. - if (config !== undefined && connectionKey(config, cwd, roots) === server.key) { - server.config = config; - continue; + /** Where a server of this scope starts, the roots it is sent, and the key of its instances. */ + const placement = (scope: ServerConfig["scope"]) => { + switch (scope ?? "session") { + case "app": + return { key: "app", cwd: options.hostCwd, roots: undefined }; + case "space": { + const root = input.space.root; + return { + key: `space:${input.space.id}`, + cwd: root ?? input.hostDir ?? options.hostCwd, + roots: root === undefined ? undefined : [root], + }; } - thread.servers.delete(name); - // Stopping can take seconds; the prompt does not wait for it. - fork(close(server)); + default: + // Threads of a session at the same place share one; `cwd` and `roots` are in the key. + return { + key: `session:${input.sessionId}`, + cwd: input.directory ?? input.hostDir ?? options.hostCwd, + roots: input.directory === undefined ? undefined : [input.directory], + }; } - for (const [name, config] of wanted) { - if (thread.servers.has(name)) continue; - const server: Server = { - name, - thread, - config, - key: connectionKey(config, cwd, roots), - cwd, - roots, - state: "connecting", - error: undefined, - tools: [], - resources: false, - live: undefined, - attempt: undefined, - ready: undefined, - closed: false, - stopped: false, - published: undefined, - lock: Semaphore.makeUnsafe(1), - control: Semaphore.makeUnsafe(1), - }; - thread.servers.set(name, server); - yield* start(server); - started.push(server); + }; + const started: Array = []; + const created: Server[] = []; + /** Running instances the thread joins, with the status published last when it joined. */ + const joined: Array = []; + const disabled: string[] = []; + // All bookkeeping happens in one step, so an interrupted turn cannot leave an instance + // counting a thread or a drain it no longer has; closing what nobody uses runs on its own. + const closing = yield* Effect.sync(() => { + const before = new Map(thread.servers); + const previous = used(thread); + thread.servers.clear(); + for (const [name, config] of input.servers) { + if (config.enabled === false) { + thread.servers.set(name, { config, server: undefined }); + const was = before.get(name); + if (was === undefined || was.server !== undefined) disabled.push(name); + continue; + } + const where = placement(config.scope); + const key = instanceKey(where.key, name, config, where.cwd, where.roots); + let server = instances.get(key); + if (server === undefined || server.closed) { + server = { + name, + config, + key, + cwd: where.cwd, + roots: where.roots, + state: "connecting", + error: undefined, + tools: [], + resources: false, + live: undefined, + attempt: undefined, + ready: undefined, + closed: false, + stopped: false, + published: undefined, + lock: Semaphore.makeUnsafe(1), + control: Semaphore.makeUnsafe(1), + users: new Set(), + leases: 0, + generation: 0, + }; + instances.set(key, server); + created.push(server); + } else if (!server.users.has(thread.id)) joined.push([server, server.published]); + const entry: Entry = { config, server }; + thread.servers.set(name, entry); + if (before.get(name)?.server !== server) started.push([name, entry, server]); + if (!server.users.has(thread.id)) { + server.users.add(thread.id); + hold(server, thread.leases); + } } - // A thread with no servers and nothing reported has nothing worth keeping. - if (thread.servers.size === 0 && thread.reported.size === 0 && threads.get(thread.threadId) === thread) - threads.delete(thread.threadId); + // Instances this thread no longer uses: a removed server, or changed settings. + const current = used(thread); + return leave( + thread, + [...previous].filter((server) => !current.has(server)), + ); }).pipe(thread.lock.withPermits(1)); - - // Wait for servers this turn started (all of them at first, then new or restarted ones) so - // their tools are offered in its first request, but never indefinitely: one hanging server + // A thread joining a running instance learns where it stands. Skipped when the instance has + // published since it joined: that newer status already reached the thread, and this one would be stale. + for (const [server, seen] of joined) yield* publishJoined(thread, server, seen); + for (const server of closing) fork(close(server)); + // A new instance publishes from its first state on, to every thread using it by then. + for (const server of created) start(server); + for (const name of disabled) yield* publishTo([thread.id], { server: name, state: "disabled" }); + + // Wait for servers this turn started using (all of them at first, then new or restarted ones) + // so their tools are offered in its first request, but never indefinitely: one hanging server // must not hold up the prompt. const loaded = input.loaded ?? []; - const pending = started.filter( - (server) => - server.config.enabled !== false && (hasDirectTools(server.config) || loadsFrom(server.name, loaded)), - ); + const pending = started.filter(([name, entry]) => hasDirectTools(entry.config) || loadsFrom(name, loaded)); if (pending.length > 0) { - const finished = yield* Effect.forEach(pending, (server) => - server.ready === undefined ? Effect.void : Deferred.await(server.ready), + const finished = yield* Effect.forEach( + pending, + ([, , server]) => (server.ready === undefined ? Effect.void : Deferred.await(server.ready)), + { concurrency: "unbounded", discard: true }, ).pipe(Effect.timeoutOption(STARTUP_WAIT)); if (finished._tag === "None") yield* Effect.logInfo("mcp servers are still connecting; their tools appear once connected"); } + // Until codemode exists, nothing reaches these tools; say so once. + const unreachable = [...thread.servers.values()].some( + (entry) => + entry.server !== undefined && + [exposureOf(entry.config), ...Object.values(entry.config.toolExposure ?? {})].some( + (exposure) => exposure === "codemode", + ), + ); + if (unreachable && !thread.warnedUnreachable) { + thread.warnedUnreachable = true; + yield* Effect.logWarning( + "MCP tools exposed to codemode are only reachable from the codemode tool, which is not active; they cannot be called.", + ); + } // Every tool is registered with its exposure; the registry decides which ones are offered. const tools: RegisteredTool[] = []; const output = { save: McpOutput.save }; - const content = (server: Server) => ({ ...output, readable: withResources(server) }); - for (const server of thread.servers.values()) { - if (server.config.enabled === false || server.stopped || server.tools.length === 0) continue; + // Resource tools follow their servers' exposure: declared when one of them is `direct`. + const resourceTools = [...thread.servers.values()].some( + (entry) => withResources(entry) && exposureOf(entry.config) === "direct", + ); + for (const [name, entry] of thread.servers) { + const server = entry.server; + if (server === undefined || server.stopped || server.tools.length === 0) continue; const names = thread.names( - server.name, + name, server.tools.map((tool) => tool.name), ); for (const tool of server.tools) { tools.push( McpTool.make({ - server: server.name, + server: name, tool, name: names.get(tool.name)!, - exposure: McpTool.exposure(server.config, tool.name), - call: call(thread, server.name, tool.name), - content: content(server), + exposure: McpTool.exposure(entry.config, tool.name), + call: call(thread, name, tool.name), + // A resource link points to `read_mcp_resource` only when the model has it. + content: { ...output, readable: resourceTools && withResources(entry) }, }), ); } } - if ([...thread.servers.values()].some(withResources)) + if (resourceTools) tools.push( ...McpResources.make({ servers: () => - [...thread.servers.values()] - .filter(withResources) - .map((server) => resourceServer(thread, server.name)), + [...thread.servers] + .filter(([, entry]) => withResources(entry)) + .map(([name]) => resourceServer(thread, name)), // A server still connecting may offer resources once it has. ready: () => Effect.forEach( [...thread.servers.values()], - (server) => - server.config.enabled === false || server.ready === undefined - ? Effect.void - : Deferred.await(server.ready), + ({ server }) => (server?.ready === undefined ? Effect.void : Deferred.await(server.ready)), { discard: true }, ), content: output, @@ -661,18 +876,52 @@ export const layer = (options: { readonly hostCwd: string; readonly home: string return tools; }); + const lease = (input: { readonly sessionId: SessionId; readonly threadId: ThreadId }) => + Effect.acquireRelease( + Effect.sync(() => { + const thread = threadOf(input.threadId, input.sessionId); + thread.leases += 1; + for (const server of used(thread)) hold(server, 1); + return thread; + }), + (thread) => + Effect.sync(() => { + thread.leases -= 1; + for (const server of used(thread)) hold(server, -1); + // Every thread drains, MCP or not; one with no servers and nothing reported has nothing worth keeping. + if ( + thread.leases === 0 && + thread.servers.size === 0 && + thread.reported.size === 0 && + threads.get(thread.id) === thread + ) + threads.delete(thread.id); + }), + ).pipe(Effect.asVoid); + /** - * Run `operation` on the thread's server, one operation per server at a time. A turn that - * replaced or removed the server meanwhile makes it unknown, so a stale status is never returned. + * Run `operation` on the instance behind the thread's server, one operation per instance at a + * time. A turn that moved the thread to another instance meanwhile makes the server unknown, + * so a stale status is never returned. */ const control = (threadId: ThreadId, name: string, operation: (server: Server) => Effect.Effect) => Effect.gen(function* () { - const server = yield* serverOf(threadId, name); + const unknown = new UnknownServerError({ server: name, message: `unknown MCP server "${name}"` }); + const entry = threads.get(threadId)?.servers.get(name); + if (entry === undefined) return yield* unknown; + const server = entry.server; + if (server === undefined) return statusOf(name, entry); return yield* Effect.gen(function* () { - if (server.config.enabled !== false) yield* operation(server); - if (threads.get(threadId)?.servers.get(name) !== server) - return yield* new UnknownServerError({ server: name }); - return statusOf(server); + // Only an idle timeout closes an instance a thread still uses; the next turn opens a new one. + if (server.closed) + return yield* new UnknownServerError({ + server: name, + message: `MCP server "${name}" was closed while idle; the thread's next message starts it again`, + }); + yield* operation(server); + const current = threads.get(threadId)?.servers.get(name); + if (current?.server !== server) return yield* unknown; + return statusOf(name, current); }).pipe(server.control.withPermits(1)); }); @@ -701,20 +950,24 @@ export const layer = (options: { readonly hostCwd: string; readonly home: string const remove = Effect.fn("Mcp.remove")(function* (sessionId: SessionId) { for (const thread of threads.values()) { if (thread.sessionId !== sessionId) continue; - const servers = yield* Effect.sync(() => { - if (threads.get(thread.threadId) === thread) threads.delete(thread.threadId); - const servers = [...thread.servers.values()]; + const unused = yield* Effect.sync(() => { + if (threads.get(thread.id) === thread) threads.delete(thread.id); + const unused = leave(thread, used(thread)); thread.servers.clear(); - return servers; + return unused; }).pipe(thread.lock.withPermits(1)); // Stopping can take seconds; deleting the session does not wait for it. - for (const server of servers) fork(close(server)); + for (const server of unused) fork(close(server)); } }); return Service.of({ reconcile, - status: (threadId) => Effect.sync(() => [...(threads.get(threadId)?.servers.values() ?? [])].map(statusOf)), + lease, + status: (threadId) => + Effect.sync(() => + [...(threads.get(threadId)?.servers ?? [])].map(([name, entry]) => statusOf(name, entry)), + ), connect, disconnect, remove, diff --git a/packages/harness/src/mcp/output.ts b/packages/harness/src/mcp/output.ts index 3db5f59c..8091cf8f 100644 --- a/packages/harness/src/mcp/output.ts +++ b/packages/harness/src/mcp/output.ts @@ -7,6 +7,7 @@ import { tmpdir } from "node:os"; import { Effect } from "effect"; import { Hex } from "effect/encoding"; import { crypto, fileSystem, hostPath } from "../host.ts"; +import { OutputError } from "./error.ts"; /** Output can carry private data, so only the user may read the files. */ const MODE = 0o600; @@ -17,7 +18,9 @@ export const save = Effect.fn("Mcp.saveOutput")(function* (data: string | Uint8A const path = hostPath.join(tmpdir(), `codework-mcp-${Hex.encode(random)}${extension}`); const bytes = typeof data === "string" ? new TextEncoder().encode(data) : data; // `wx` never follows a link someone else placed at the path. - yield* fileSystem.writeFile(path, bytes, { flag: "wx", mode: MODE }).pipe(Effect.mapError((error) => error.message)); + yield* fileSystem + .writeFile(path, bytes, { flag: "wx", mode: MODE }) + .pipe(Effect.mapError((error) => new OutputError({ message: error.message }))); return path; }); diff --git a/packages/harness/src/mcp/resources.ts b/packages/harness/src/mcp/resources.ts index 03ab8e42..6ca99886 100644 --- a/packages/harness/src/mcp/resources.ts +++ b/packages/harness/src/mcp/resources.ts @@ -21,6 +21,7 @@ import type { import { Effect, Predicate } from "effect"; import type { RegisteredTool } from "../tool/tool.ts"; import { McpContent } from "./content.ts"; +import { type McpError, ResourceError } from "./error.ts"; import { McpTool } from "./tool.ts"; export const LIST_RESOURCES_TOOL = "list_mcp_resources"; @@ -30,12 +31,12 @@ export const READ_RESOURCE_TOOL = McpContent.READ_RESOURCE_TOOL; /** A server that offers resources, reached through its current connection. */ export interface Server { readonly name: string; - readonly resources: (cursor?: string) => Effect.Effect; - readonly templates: (cursor?: string) => Effect.Effect; + readonly resources: (cursor?: string) => Effect.Effect; + readonly templates: (cursor?: string) => Effect.Effect; /** Every page, with the SDK's own pagination limits. */ - readonly allResources: () => Effect.Effect, string>; - readonly allTemplates: () => Effect.Effect, string>; - readonly read: (uri: string) => Effect.Effect; + readonly allResources: () => Effect.Effect, McpError.Failure>; + readonly allTemplates: () => Effect.Effect, McpError.Failure>; + readonly read: (uri: string) => Effect.Effect; } const isMcpApp = (item: { @@ -74,10 +75,10 @@ const READ_PARAMETERS = { }; /** A string argument; absent when missing. */ -const raw = (args: Record, key: string): Effect.Effect => { +const raw = (args: Record, key: string): Effect.Effect => { const value = args[key]; if (value === undefined || value === null) return Effect.undefined; - if (!Predicate.isString(value)) return Effect.fail(`${key} must be a string`); + if (!Predicate.isString(value)) return Effect.fail(new ResourceError({ message: `${key} must be a string` })); return Effect.succeed(value); }; @@ -106,7 +107,9 @@ export const make = (input: { if (server !== undefined) return Effect.succeed(server); const available = servers.map((candidate) => candidate.name).join(", "); return Effect.fail( - `MCP server "${name}" has no resources${available ? `. Servers with resources: ${available}` : ""}`, + new ResourceError({ + message: `MCP server "${name}" has no resources${available ? `. Servers with resources: ${available}` : ""}`, + }), ); }; @@ -118,8 +121,8 @@ export const make = (input: { /** One page of one server, or every page of every server. */ const list = ( key: "resources" | "resourceTemplates", - page: (server: Server, cursor?: string) => Effect.Effect, - all: (server: Server) => Effect.Effect, string>, + page: (server: Server, cursor?: string) => Effect.Effect, + all: (server: Server) => Effect.Effect, McpError.Failure>, ) => Effect.fnUntraced(function* (args: Record) { const name = yield* argument(args, "server"); @@ -135,7 +138,8 @@ export const make = (input: { ...(result.nextCursor === undefined ? {} : { nextCursor: result.nextCursor }), }); } - if (cursor !== undefined) return yield* Effect.fail("cursor can only be used when a server is specified"); + if (cursor !== undefined) + return yield* new ResourceError({ message: "cursor can only be used when a server is specified" }); const servers = [...input.servers()].sort((a, b) => a.name.localeCompare(b.name)); const results = yield* Effect.forEach(servers, (server) => Effect.result(all(server)), { concurrency: "unbounded", @@ -146,7 +150,7 @@ export const make = (input: { const server = servers[index]!.name; if (result._tag === "Success") items.push(...result.success.filter(visible).map((item) => listed(server, item))); - else errors.push({ server, error: result.failure }); + else errors.push({ server, error: result.failure.message }); }); return yield* json({ [key]: items, ...(errors.length > 0 ? { errors } : {}) }); }); @@ -154,8 +158,8 @@ export const make = (input: { const read = Effect.fnUntraced(function* (args: Record) { const name = yield* argument(args, "server"); const uri = yield* argument(args, "uri"); - if (name === undefined) return yield* Effect.fail("server must be provided"); - if (uri === undefined) return yield* Effect.fail("uri must be provided"); + if (name === undefined) return yield* new ResourceError({ message: "server must be provided" }); + if (uri === undefined) return yield* new ResourceError({ message: "uri must be provided" }); yield* input.ready(); const server = yield* find(name); const result = yield* server.read(uri); diff --git a/packages/harness/src/mcp/tool.ts b/packages/harness/src/mcp/tool.ts index b421ec82..49020df0 100644 --- a/packages/harness/src/mcp/tool.ts +++ b/packages/harness/src/mcp/tool.ts @@ -11,6 +11,7 @@ import { ToolProgress } from "../tool/progress.ts"; import * as Tool from "../tool/tool.ts"; import { Hash } from "../util/hash.ts"; import { McpContent } from "./content.ts"; +import type { McpError } from "./error.ts"; import type { Progress } from "./client.ts"; /** Provider tool names are limited to 64 characters of `[A-Za-z0-9_-]`. */ @@ -129,7 +130,7 @@ export const define = (input: { readonly server: string; readonly tool: string; readonly exposure: Exposure; - readonly run: (run: Run) => Effect.Effect; + readonly run: (run: Run) => Effect.Effect; }) => { const outcome = (result: Result): Outcome => ({ content: result.content, @@ -162,7 +163,7 @@ export const define = (input: { report(progress.report({ content: [{ type: "text", text }] })); }, }) - .pipe(Effect.mapError((message) => outcome({ content: [{ type: "text", text: message }] }))); + .pipe(Effect.mapError((error) => outcome({ content: [{ type: "text", text: error.message }] }))); return result.isError ? yield* Effect.fail(outcome(result)) : outcome(result); }), }), @@ -185,7 +186,7 @@ export const make = (input: { readonly tool: McpTool; readonly name: string; readonly exposure: Exposure; - readonly call: (run: Run) => Effect.Effect; + readonly call: (run: Run) => Effect.Effect; readonly content: McpContent.Options; }) => { const { server, tool } = input; diff --git a/packages/harness/src/mcp/value.ts b/packages/harness/src/mcp/value.ts index 5dbf85ea..b870bfed 100644 --- a/packages/harness/src/mcp/value.ts +++ b/packages/harness/src/mcp/value.ts @@ -14,6 +14,7 @@ import { Config, Duration, Effect, Fiber, Option, Stream } from "effect"; import { ChildProcess, ChildProcessSpawner } from "effect/process"; +import { ValueError } from "./error.ts"; const COMMAND_TIMEOUT = Duration.seconds(10); const NAME = /^[A-Za-z_][A-Za-z0-9_]*$/; @@ -111,7 +112,8 @@ const command = Effect.fnUntraced(function* (line: string, what: string) { commands.set(line, running); } const output = yield* Fiber.join(running); - if (output === undefined) return yield* Effect.fail(`Failed to resolve ${what} from shell command: ${line}`); + if (output === undefined) + return yield* new ValueError({ message: `Failed to resolve ${what} from shell command: ${line}` }); return output; }); @@ -126,9 +128,9 @@ const variables = Effect.fnUntraced(function* (value: string, what: string) { } const missing = names.filter((name) => !env.has(name)); if (missing.length > 0) - return yield* Effect.fail( - `Failed to resolve ${what} from environment variable${missing.length === 1 ? "" : "s"}: ${missing.join(", ")}`, - ); + return yield* new ValueError({ + message: `Failed to resolve ${what} from environment variable${missing.length === 1 ? "" : "s"}: ${missing.join(", ")}`, + }); return parts.map((part) => (part.type === "literal" ? part.value : env.get(part.name))).join(""); }); @@ -138,7 +140,8 @@ const variables = Effect.fnUntraced(function* (value: string, what: string) { */ export const resolve = Effect.fnUntraced(function* (value: string, kind: keyof typeof UNSAFE, what: string) { const resolved = yield* value.startsWith("!") ? command(value.slice(1), what) : variables(value, what); - if (UNSAFE[kind].pattern.test(resolved)) return yield* Effect.fail(`${what} contains ${UNSAFE[kind].what}`); + if (UNSAFE[kind].pattern.test(resolved)) + return yield* new ValueError({ message: `${what} contains ${UNSAFE[kind].what}` }); return resolved; }); diff --git a/packages/harness/src/runner/execute.ts b/packages/harness/src/runner/execute.ts index f8e72d41..07c3712c 100644 --- a/packages/harness/src/runner/execute.ts +++ b/packages/harness/src/runner/execute.ts @@ -18,6 +18,7 @@ import { Runner } from "./run.ts"; import { SessionFailure } from "../session/failure.ts"; import { SessionSchema } from "../session/schema.ts"; import { Session } from "../session/session.ts"; +import { Mcp } from "../mcp/mcp.ts"; import { ThreadSchema } from "../thread/schema.ts"; import { Thread } from "../thread/thread.ts"; @@ -35,6 +36,7 @@ export const layer = Layer.effect( // in the callback would leave the tag in it. const runner = yield* Runner.Service; const events = yield* Event.Service; + const mcp = yield* Mcp.Service; // A lifecycle publish that fails is reported, never thrown back into the // coordinator: losing the observation must not change how the drain ends. const reportLifecycle = (threadId: ThreadId, publish: Effect.Effect) => @@ -169,6 +171,8 @@ export const layer = Layer.effect( ), ); + // The thread's MCP servers stay open while its drain runs, and idle out after. + yield* mcp.lease({ sessionId, threadId }); return yield* runner .run({ threadId, force, recovered }) .pipe(Effect.provideContext(Context.merge(locationContext, mountContext))); diff --git a/packages/harness/src/state/state.ts b/packages/harness/src/state/state.ts index 8542e157..7ad622e1 100644 --- a/packages/harness/src/state/state.ts +++ b/packages/harness/src/state/state.ts @@ -505,6 +505,10 @@ export const layer = ( servers, // MCP servers run on the host, so only a host sandbox's directory exists for them. directory: sandbox.kind === "local" ? location.directory : undefined, + space: { + id: location.space.id, + ...(sandbox.kind === "local" ? { root: location.space.location } : {}), + }, hostDir, errors: loadedSettings.mcp.errors, loaded: agent.modelTools?.add, diff --git a/packages/harness/test/__artifacts__/mcp.session.json b/packages/harness/test/__artifacts__/mcp.session.json index 6a6df06d..9d1c248c 100644 --- a/packages/harness/test/__artifacts__/mcp.session.json +++ b/packages/harness/test/__artifacts__/mcp.session.json @@ -49,6 +49,10 @@ [ "badheader", true + ], + [ + "later", + true ] ], [ @@ -67,6 +71,10 @@ [ "badheader", true + ], + [ + "later", + true ] ], [ @@ -85,6 +93,10 @@ [ "badheader", true + ], + [ + "later", + true ] ], [ @@ -103,6 +115,10 @@ [ "badheader", true + ], + [ + "later", + true ] ], [ @@ -121,6 +137,10 @@ [ "badheader", true + ], + [ + "later", + true ] ], [ @@ -139,6 +159,10 @@ [ "badheader", true + ], + [ + "later", + true ] ], [ @@ -157,6 +181,10 @@ [ "badheader", true + ], + [ + "later", + true ] ], [ @@ -175,6 +203,10 @@ [ "badheader", true + ], + [ + "later", + true ] ] ], @@ -278,7 +310,7 @@ "status": "completed", "isError": false, "content": [ - "{\"resources\":[{\"server\":\"e2e\",\"name\":\"notes\",\"uri\":\"fixture://notes.txt\",\"mimeType\":\"text/plain\"},{\"server\":\"e2e\",\"name\":\"data\",\"uri\":\"fixture://data.bin\",\"mimeType\":\"application/octet-stream\"}]}" + "{\"resources\":[{\"server\":\"e2e\",\"name\":\"notes\",\"uri\":\"fixture://notes.txt\",\"mimeType\":\"text/plain\"},{\"server\":\"e2e\",\"name\":\"data\",\"uri\":\"fixture://data.bin\",\"mimeType\":\"application/octet-stream\"},{\"server\":\"later\",\"name\":\"notes\",\"uri\":\"fixture://notes.txt\",\"mimeType\":\"text/plain\"},{\"server\":\"later\",\"name\":\"data\",\"uri\":\"fixture://data.bin\",\"mimeType\":\"application/octet-stream\"}]}" ] }, "resources-page": { @@ -306,7 +338,7 @@ "status": "completed", "isError": false, "content": [ - "{\"resourceTemplates\":[{\"server\":\"e2e\",\"name\":\"daily notes\",\"uriTemplate\":\"fixture://notes/{day}\"}]}" + "{\"resourceTemplates\":[{\"server\":\"e2e\",\"name\":\"daily notes\",\"uriTemplate\":\"fixture://notes/{day}\"},{\"server\":\"later\",\"name\":\"daily notes\",\"uriTemplate\":\"fixture://notes/{day}\"}]}" ] }, "read-text": { @@ -327,7 +359,7 @@ "status": "error", "isError": true, "content": [ - "MCP server \"nope\" has no resources. Servers with resources: e2e" + "MCP server \"nope\" has no resources. Servers with resources: e2e, later" ] } }, @@ -404,6 +436,56 @@ "state": "failed", "error": "MCP server \"badheader\" header \"Authorization\" contains a line break or NUL character", "tools": [] + }, + { + "server": "later", + "state": "connected", + "tools": [ + { + "name": "echo", + "exposure": "codemode" + }, + { + "name": "args", + "exposure": "codemode" + }, + { + "name": "fail", + "exposure": "codemode" + }, + { + "name": "pid", + "exposure": "codemode" + }, + { + "name": "crash", + "exposure": "codemode" + }, + { + "name": "my-tool.v2", + "exposure": "codemode" + }, + { + "name": "secret", + "exposure": "codemode" + }, + { + "name": "bye", + "exposure": "codemode" + }, + { + "name": "cwd", + "exposure": "codemode" + }, + { + "name": "link", + "exposure": "codemode" + }, + { + "name": "big", + "exposure": "codemode" + } + ] } ], "configErrors": [ diff --git a/packages/harness/test/__artifacts__/mcp.session.lifecycle.json b/packages/harness/test/__artifacts__/mcp.session.lifecycle.json index bf559d8b..c4ff0f87 100644 --- a/packages/harness/test/__artifacts__/mcp.session.lifecycle.json +++ b/packages/harness/test/__artifacts__/mcp.session.lifecycle.json @@ -124,7 +124,6 @@ "read_mcp_resource" ], [ - "mcp__remote__echo", "mcp__e2e__echo", "mcp__e2e__args", "mcp__e2e__fail", @@ -135,6 +134,7 @@ "mcp__e2e__cwd", "mcp__e2e__link", "mcp__e2e__big", + "mcp__remote__echo", "list_mcp_resources", "list_mcp_resource_templates", "read_mcp_resource" @@ -157,16 +157,6 @@ "POST tools/call" ], "status": [ - { - "server": "remote", - "state": "connected", - "tools": [ - { - "name": "echo", - "exposure": "direct" - } - ] - }, { "server": "e2e", "state": "connected", @@ -216,6 +206,16 @@ "exposure": "direct" } ] + }, + { + "server": "remote", + "state": "connected", + "tools": [ + { + "name": "echo", + "exposure": "direct" + } + ] } ] } diff --git a/packages/harness/test/__artifacts__/mcp.session.scopes.json b/packages/harness/test/__artifacts__/mcp.session.scopes.json new file mode 100644 index 00000000..2cf2fdf2 --- /dev/null +++ b/packages/harness/test/__artifacts__/mcp.session.scopes.json @@ -0,0 +1,630 @@ +{ + "processes": { + "a": { + "mine": "", + "twin": "", + "ours": "", + "all": "" + }, + "b": { + "mine": "", + "twin": "", + "ours": "", + "all": "" + }, + "c": { + "mine": "", + "twin": "", + "ours": "", + "all": "" + }, + "again": { + "mine": "", + "twin": "", + "ours": "", + "all": "" + } + }, + "idle": [ + { + "server": "mine", + "state": "disconnected", + "tools": [ + { + "name": "echo", + "exposure": "direct" + }, + { + "name": "args", + "exposure": "direct" + }, + { + "name": "fail", + "exposure": "direct" + }, + { + "name": "pid", + "exposure": "direct" + }, + { + "name": "crash", + "exposure": "direct" + }, + { + "name": "my-tool.v2", + "exposure": "direct" + }, + { + "name": "secret", + "exposure": "direct" + }, + { + "name": "bye", + "exposure": "direct" + }, + { + "name": "cwd", + "exposure": "direct" + }, + { + "name": "link", + "exposure": "direct" + }, + { + "name": "big", + "exposure": "direct" + } + ] + }, + { + "server": "twin", + "state": "disconnected", + "tools": [ + { + "name": "echo", + "exposure": "direct" + }, + { + "name": "args", + "exposure": "direct" + }, + { + "name": "fail", + "exposure": "direct" + }, + { + "name": "pid", + "exposure": "direct" + }, + { + "name": "crash", + "exposure": "direct" + }, + { + "name": "my-tool.v2", + "exposure": "direct" + }, + { + "name": "secret", + "exposure": "direct" + }, + { + "name": "bye", + "exposure": "direct" + }, + { + "name": "cwd", + "exposure": "direct" + }, + { + "name": "link", + "exposure": "direct" + }, + { + "name": "big", + "exposure": "direct" + } + ] + }, + { + "server": "ours", + "state": "disconnected", + "tools": [ + { + "name": "echo", + "exposure": "direct" + }, + { + "name": "args", + "exposure": "direct" + }, + { + "name": "fail", + "exposure": "direct" + }, + { + "name": "pid", + "exposure": "direct" + }, + { + "name": "crash", + "exposure": "direct" + }, + { + "name": "my-tool.v2", + "exposure": "direct" + }, + { + "name": "secret", + "exposure": "direct" + }, + { + "name": "bye", + "exposure": "direct" + }, + { + "name": "cwd", + "exposure": "direct" + }, + { + "name": "link", + "exposure": "direct" + }, + { + "name": "big", + "exposure": "direct" + } + ] + }, + { + "server": "all", + "state": "disconnected", + "tools": [ + { + "name": "echo", + "exposure": "direct" + }, + { + "name": "args", + "exposure": "direct" + }, + { + "name": "fail", + "exposure": "direct" + }, + { + "name": "pid", + "exposure": "direct" + }, + { + "name": "crash", + "exposure": "direct" + }, + { + "name": "my-tool.v2", + "exposure": "direct" + }, + { + "name": "secret", + "exposure": "direct" + }, + { + "name": "bye", + "exposure": "direct" + }, + { + "name": "cwd", + "exposure": "direct" + }, + { + "name": "link", + "exposure": "direct" + }, + { + "name": "big", + "exposure": "direct" + } + ] + }, + { + "server": "mine", + "state": "disconnected", + "tools": [ + { + "name": "echo", + "exposure": "direct" + }, + { + "name": "args", + "exposure": "direct" + }, + { + "name": "fail", + "exposure": "direct" + }, + { + "name": "pid", + "exposure": "direct" + }, + { + "name": "crash", + "exposure": "direct" + }, + { + "name": "my-tool.v2", + "exposure": "direct" + }, + { + "name": "secret", + "exposure": "direct" + }, + { + "name": "bye", + "exposure": "direct" + }, + { + "name": "cwd", + "exposure": "direct" + }, + { + "name": "link", + "exposure": "direct" + }, + { + "name": "big", + "exposure": "direct" + } + ] + }, + { + "server": "twin", + "state": "disconnected", + "tools": [ + { + "name": "echo", + "exposure": "direct" + }, + { + "name": "args", + "exposure": "direct" + }, + { + "name": "fail", + "exposure": "direct" + }, + { + "name": "pid", + "exposure": "direct" + }, + { + "name": "crash", + "exposure": "direct" + }, + { + "name": "my-tool.v2", + "exposure": "direct" + }, + { + "name": "secret", + "exposure": "direct" + }, + { + "name": "bye", + "exposure": "direct" + }, + { + "name": "cwd", + "exposure": "direct" + }, + { + "name": "link", + "exposure": "direct" + }, + { + "name": "big", + "exposure": "direct" + } + ] + }, + { + "server": "ours", + "state": "disconnected", + "tools": [ + { + "name": "echo", + "exposure": "direct" + }, + { + "name": "args", + "exposure": "direct" + }, + { + "name": "fail", + "exposure": "direct" + }, + { + "name": "pid", + "exposure": "direct" + }, + { + "name": "crash", + "exposure": "direct" + }, + { + "name": "my-tool.v2", + "exposure": "direct" + }, + { + "name": "secret", + "exposure": "direct" + }, + { + "name": "bye", + "exposure": "direct" + }, + { + "name": "cwd", + "exposure": "direct" + }, + { + "name": "link", + "exposure": "direct" + }, + { + "name": "big", + "exposure": "direct" + } + ] + }, + { + "server": "all", + "state": "disconnected", + "tools": [ + { + "name": "echo", + "exposure": "direct" + }, + { + "name": "args", + "exposure": "direct" + }, + { + "name": "fail", + "exposure": "direct" + }, + { + "name": "pid", + "exposure": "direct" + }, + { + "name": "crash", + "exposure": "direct" + }, + { + "name": "my-tool.v2", + "exposure": "direct" + }, + { + "name": "secret", + "exposure": "direct" + }, + { + "name": "bye", + "exposure": "direct" + }, + { + "name": "cwd", + "exposure": "direct" + }, + { + "name": "link", + "exposure": "direct" + }, + { + "name": "big", + "exposure": "direct" + } + ] + }, + { + "server": "mine", + "state": "disconnected", + "tools": [ + { + "name": "echo", + "exposure": "direct" + }, + { + "name": "args", + "exposure": "direct" + }, + { + "name": "fail", + "exposure": "direct" + }, + { + "name": "pid", + "exposure": "direct" + }, + { + "name": "crash", + "exposure": "direct" + }, + { + "name": "my-tool.v2", + "exposure": "direct" + }, + { + "name": "secret", + "exposure": "direct" + }, + { + "name": "bye", + "exposure": "direct" + }, + { + "name": "cwd", + "exposure": "direct" + }, + { + "name": "link", + "exposure": "direct" + }, + { + "name": "big", + "exposure": "direct" + } + ] + }, + { + "server": "twin", + "state": "disconnected", + "tools": [ + { + "name": "echo", + "exposure": "direct" + }, + { + "name": "args", + "exposure": "direct" + }, + { + "name": "fail", + "exposure": "direct" + }, + { + "name": "pid", + "exposure": "direct" + }, + { + "name": "crash", + "exposure": "direct" + }, + { + "name": "my-tool.v2", + "exposure": "direct" + }, + { + "name": "secret", + "exposure": "direct" + }, + { + "name": "bye", + "exposure": "direct" + }, + { + "name": "cwd", + "exposure": "direct" + }, + { + "name": "link", + "exposure": "direct" + }, + { + "name": "big", + "exposure": "direct" + } + ] + }, + { + "server": "ours", + "state": "disconnected", + "tools": [ + { + "name": "echo", + "exposure": "direct" + }, + { + "name": "args", + "exposure": "direct" + }, + { + "name": "fail", + "exposure": "direct" + }, + { + "name": "pid", + "exposure": "direct" + }, + { + "name": "crash", + "exposure": "direct" + }, + { + "name": "my-tool.v2", + "exposure": "direct" + }, + { + "name": "secret", + "exposure": "direct" + }, + { + "name": "bye", + "exposure": "direct" + }, + { + "name": "cwd", + "exposure": "direct" + }, + { + "name": "link", + "exposure": "direct" + }, + { + "name": "big", + "exposure": "direct" + } + ] + }, + { + "server": "all", + "state": "disconnected", + "tools": [ + { + "name": "echo", + "exposure": "direct" + }, + { + "name": "args", + "exposure": "direct" + }, + { + "name": "fail", + "exposure": "direct" + }, + { + "name": "pid", + "exposure": "direct" + }, + { + "name": "crash", + "exposure": "direct" + }, + { + "name": "my-tool.v2", + "exposure": "direct" + }, + { + "name": "secret", + "exposure": "direct" + }, + { + "name": "bye", + "exposure": "direct" + }, + { + "name": "cwd", + "exposure": "direct" + }, + { + "name": "link", + "exposure": "direct" + }, + { + "name": "big", + "exposure": "direct" + } + ] + } + ] +} diff --git a/packages/harness/test/__artifacts__/mcp.thread.idle.json b/packages/harness/test/__artifacts__/mcp.thread.idle.json new file mode 100644 index 00000000..331f576b --- /dev/null +++ b/packages/harness/test/__artifacts__/mcp.thread.idle.json @@ -0,0 +1,39 @@ +{ + "afterTurn": "connected", + "idled": "disconnected", + "events": { + "main": { + "idle": [ + { + "server": "idle", + "state": "connecting" + }, + { + "server": "idle", + "state": "connected", + "tools": [ + "echo", + "args", + "fail", + "pid", + "crash", + "my-tool.v2", + "secret", + "bye", + "cwd", + "link", + "big" + ] + }, + { + "server": "idle", + "state": "disconnected" + } + ] + } + }, + "connect": { + "tag": "Mcp.UnknownServerError", + "message": "MCP server \"idle\" was closed while idle; the thread's next message starts it again" + } +} diff --git a/packages/harness/test/__artifacts__/mcp.thread.json b/packages/harness/test/__artifacts__/mcp.thread.json index 4ad70663..d3be73e6 100644 --- a/packages/harness/test/__artifacts__/mcp.thread.json +++ b/packages/harness/test/__artifacts__/mcp.thread.json @@ -474,6 +474,15 @@ "search": "Loaded 1 tool. They are available from your next call:\n- mcp__e2e__args: The arguments as received.", "deferred-loaded": "{\"word\":\"abc\"}", "fork-pid": "" + }, + "sibling": { + "main-pid": "", + "codemode": "Tool \"mcp__e2e__fail\" is not available in this thread.", + "hidden": "Tool \"mcp__e2e__secret\" is not available in this thread.", + "deferred-unloaded": "Tool \"mcp__e2e__args\" is not available in this thread.", + "search": "Loaded 1 tool. They are available from your next call:\n- mcp__e2e__args: The arguments as received.", + "deferred-loaded": "{\"word\":\"abc\"}", + "sibling-pid": "" } }, "changes": [ @@ -558,9 +567,74 @@ "startsAfterRaces": 5, "unknown": { "tag": "Mcp.UnknownServerError", - "message": "unknown MCP server: nope" + "message": "unknown MCP server \"nope\"" }, "events": { + "fork": { + "e2e": [ + { + "server": "e2e", + "state": "connecting" + }, + { + "server": "e2e", + "state": "connected", + "tools": [ + "echo", + "args", + "fail", + "pid", + "crash", + "my-tool.v2", + "secret", + "bye", + "cwd", + "link", + "big" + ] + } + ], + "extra": [ + { + "server": "extra", + "state": "connecting" + }, + { + "server": "extra", + "state": "connected", + "tools": [ + "echo", + "args", + "fail", + "pid", + "crash", + "my-tool.v2", + "secret", + "bye", + "cwd", + "link", + "big" + ] + } + ], + "gated": [ + { + "server": "gated", + "state": "connecting" + }, + { + "server": "gated", + "state": "failed", + "error": "connection closed: mcp server process exited with code 1: gate closed" + } + ], + "off": [ + { + "server": "off", + "state": "disabled" + } + ] + }, "main": { "e2e": [ { @@ -721,8 +795,51 @@ } ] }, - "fork": { + "sibling": { "e2e": [ + { + "server": "e2e", + "state": "connected", + "tools": [ + "echo", + "args", + "fail", + "pid", + "crash", + "my-tool.v2", + "secret", + "bye", + "cwd", + "link", + "big" + ] + }, + { + "server": "e2e", + "state": "disconnected", + "error": "connection closed: mcp server process exited with code 0" + }, + { + "server": "e2e", + "state": "connecting" + }, + { + "server": "e2e", + "state": "connected", + "tools": [ + "echo", + "args", + "fail", + "pid", + "crash", + "my-tool.v2", + "secret", + "bye", + "cwd", + "link", + "big" + ] + }, { "server": "e2e", "state": "connecting" @@ -746,29 +863,67 @@ } ], "gated": [ + { + "server": "gated", + "state": "failed", + "error": "connection closed: mcp server process exited with code 1: gate closed" + }, { "server": "gated", "state": "connecting" }, { "server": "gated", - "state": "failed", - "error": "connection closed: mcp server process exited with code 1: gate closed" - } - ], - "off": [ + "state": "connected", + "tools": [ + "echo", + "args", + "fail", + "pid", + "crash", + "my-tool.v2", + "secret", + "bye", + "cwd", + "link", + "big" + ] + }, { - "server": "off", - "state": "disabled" - } - ], - "extra": [ + "server": "gated", + "state": "disconnected" + }, { - "server": "extra", + "server": "gated", "state": "connecting" }, { - "server": "extra", + "server": "gated", + "state": "connected", + "tools": [ + "echo", + "args", + "fail", + "pid", + "crash", + "my-tool.v2", + "secret", + "bye", + "cwd", + "link", + "big" + ] + }, + { + "server": "gated", + "state": "disconnected" + }, + { + "server": "gated", + "state": "connecting" + }, + { + "server": "gated", "state": "connected", "tools": [ "echo", @@ -784,6 +939,12 @@ "big" ] } + ], + "off": [ + { + "server": "off", + "state": "disabled" + } ] } }, diff --git a/packages/harness/test/mcp.session.test.ts b/packages/harness/test/mcp.session.test.ts index 3a0d6ce6..fc0c9a79 100644 --- a/packages/harness/test/mcp.session.test.ts +++ b/packages/harness/test/mcp.session.test.ts @@ -77,6 +77,8 @@ it("mcp: config layers, plugin bucket, direct tools, a crash mid-turn and the ne broken: { url: "ftp://example.com/mcp" }, signin: { url: "https://example.com/mcp", oauth: { clientId: "e2e" } }, badheader: { url: "http://127.0.0.1:9/mcp", headers: { Authorization: "${MCP_E2E_BAD_HEADER}" } }, + // Codemode and deferred exposures are valid; until codemode exists these tools are simply not declared. + later: stdio({ exposure: "codemode-deferred" }), }, }); // The project turns a user server off without restating it, and cannot add an override with no base. @@ -197,6 +199,8 @@ it("mcp: config layers, plugin bucket, direct tools, a crash mid-turn and the ne const firstTools = requests[0]?.context.tools ?? []; expect(firstTools.map((tool) => tool.name)).not.toContain("mcp__e2e__secret"); + expect(firstTools.some((tool) => tool.name.startsWith("mcp__later__"))).toBe(false); + expect(result.errors.some((error) => error.includes("later"))).toBe(false); // The tool set is frozen per turn and stable across them, so every request declares the same tools. for (const request of requests) expect(request.context.tools).toEqual(firstTools); // The server's JSON Schema reaches the provider as written. @@ -390,3 +394,125 @@ it("mcp: a server exiting while idle, reconfiguration, and an expired HTTP sessi "./__artifacts__/mcp.session.lifecycle.json", ); }, 60_000); + +it("mcp: scopes decide who shares a server, and idle servers close", async () => { + await using dir = await tmpdir(); + const home = join(dir.path, "home"); + const shared = join(dir.path, "shared"); + const other = join(dir.path, "other"); + await Promise.all([fs.mkdir(shared, { recursive: true }), fs.mkdir(other, { recursive: true })]); + process.env.MCP_E2E_GREETING = "hello"; + await write(join(home, "mcp.json"), { + // `twin` has exactly `mine`'s settings: still its own server, with its own instance. + mcpServers: { mine: stdio(), twin: stdio(), ours: stdio({ scope: "space" }), all: stdio({ scope: "app" }) }, + }); + + const calls = (prefix: string) => + ["mine", "twin", "ours", "all"].map((server) => pendingCall(`mcp__${server}__pid`, {}, `${prefix}-${server}`)); + const requests: LLM.Input[] = []; + const llm = scripted([calls("a"), [], calls("b"), [], calls("c"), [], calls("again"), []], requests); + + const result = await Effect.gen(function* () { + const model = { provider: "openai", id: "gpt-5.6-luna" }; + // A and B work in one space, C in another. + const a = yield* Session.create({ directory: shared, model }); + const b = yield* Session.create({ directory: shared, model }); + const c = yield* Session.create({ directory: other, model }); + // A lease on A's servers spans the three runs, so a slow runner cannot idle out the shared + // instances between them; the short timeout only matters once it is released. + yield* Effect.gen(function* () { + yield* Mcp.Service.use((mcp) => mcp.lease({ sessionId: a.id, threadId: a.main.id })); + yield* a.main.run("a"); + yield* b.main.run("b"); + yield* c.main.run("c"); + }).pipe(Effect.scoped); + // No drain holds any server now, so every instance closes once the idle timeout passes. + const status = Mcp.Service.use((mcp) => + Effect.forEach([a, b, c], (session) => mcp.status(session.main.id)).pipe(Effect.map((all) => all.flat())), + ); + for (let tries = 0; tries < 250 && (yield* status).some((one) => one.state !== "disconnected"); tries++) + yield* Effect.sleep("40 millis"); + const idle = yield* status; + // A closed instance is started again by the next turn that needs it. + yield* a.main.run("again"); + const messages = yield* Effect.forEach([a, b, c], (session) => + messageEntries(session.main).pipe(Effect.flatMap((line) => Effect.forEach(line, ContextCodec.decodeMessage))), + ); + return { messages: messages.flat(), idle }; + }).pipe( + Effect.provide( + Harness.layer({ + home, + hostCwd: dir.path, + database: ":memory:", + plugins: [defaultPromptPlugin], + llm, + // Short, to watch idle instances close. + mcpIdleTimeout: "1500 millis", + }), + ), + Effect.scoped, + Effect.runPromise, + ); + + const pids = new Map( + result.messages + .flatMap((message) => (message.role === "system" ? [] : message.parts)) + .flatMap((part) => + part.type === "toolCall" && "result" in part + ? [[part.callID, (part.result.content[0] as { readonly text: string }).text] as const] + : [], + ), + ); + const pid = (session: string, server: string) => pids.get(`${session}-${server}`); + for (const id of pids.values()) expect(id).toMatch(/^\d+$/); + // `session`: each its own. `space`: A and B share, C has its own. `app`: one for all. + expect(new Set([pid("a", "mine"), pid("b", "mine"), pid("c", "mine")]).size).toBe(3); + expect(pid("a", "twin")).not.toBe(pid("a", "mine")); + expect(pid("a", "ours")).toBe(pid("b", "ours")); + expect(pid("c", "ours")).not.toBe(pid("a", "ours")); + expect(new Set([pid("a", "all"), pid("b", "all"), pid("c", "all")]).size).toBe(1); + + // Every instance closed, and its process with it. + expect(result.idle.every((one) => one.state === "disconnected")).toBe(true); + const alive = (value: string) => { + try { + process.kill(Number(value), 0); + return true; + } catch { + return false; + } + }; + const before = ["a", "b", "c"].flatMap((session) => + ["mine", "twin", "ours", "all"].map((server) => pid(session, server)), + ); + expect(before.filter((value) => value !== undefined && alive(value))).toEqual([]); + for (const server of ["mine", "twin", "ours", "all"]) expect(before).not.toContain(pid("again", server)); + + // Distinct processes per scope, in the order the sessions first saw them. + const order = [ + ...new Set( + ["a", "b", "c", "again"].flatMap((session) => + ["mine", "twin", "ours", "all"].map((server) => pid(session, server)), + ), + ), + ]; + const label = (value: string | undefined) => ``; + await expect( + `${JSON.stringify( + { + processes: Object.fromEntries( + ["a", "b", "c", "again"].map((session) => [ + session, + Object.fromEntries( + ["mine", "twin", "ours", "all"].map((server) => [server, label(pid(session, server))]), + ), + ]), + ), + idle: result.idle, + }, + null, + "\t", + )}\n`, + ).toMatchFileSnapshot("./__artifacts__/mcp.session.scopes.json"); +}, 60_000); diff --git a/packages/harness/test/mcp.thread.test.ts b/packages/harness/test/mcp.thread.test.ts index c6fcdade..1b7ba84e 100644 --- a/packages/harness/test/mcp.thread.test.ts +++ b/packages/harness/test/mcp.thread.test.ts @@ -70,7 +70,11 @@ const scripted = () => { return { open, turn, requests }; }; -/** Every `mcp.server.status` event, by thread label and server, in order. */ +/** Threads and servers connect concurrently; their keys are sorted so the artifact does not depend on it. */ +const byName = (record: Record) => + Object.fromEntries(Object.entries(record).sort(([a], [b]) => a.localeCompare(b))); + +/** Every `mcp.server.status` event, by thread label and server, in order; `read` takes a copy. */ const statusEvents = (label: (threadId: string) => string) => Effect.gen(function* () { const seen: Record>> = {}; @@ -83,7 +87,10 @@ const statusEvents = (label: (threadId: string) => string) => (byServer[status.server] ??= []).push(status); }), ); - return seen; + return { + read: () => + byName(Object.fromEntries(Object.entries(seen).map(([thread, servers]) => [thread, byName(servers)]))), + }; }); /** The text of each settled tool call of a thread, by call id. */ @@ -131,10 +138,9 @@ it("mcp: state per thread, no retry of a failed server, connect and disconnect, await using dir = await tmpdir(); const home = join(dir.path, "home"); const work = join(dir.path, "work"); - const elsewhere = join(work, "sub"); + // A worktree of the same project: a thread can move there, and it is another place. + const elsewhere = join(dir.path, "elsewhere"); await fs.mkdir(work, { recursive: true }); - await fs.mkdir(elsewhere, { recursive: true }); - // One project, so a thread can move to another directory of it. await git(work, "init", "-q", "-b", "main"); await git( work, @@ -148,6 +154,7 @@ it("mcp: state per thread, no retry of a failed server, connect and disconnect, "-m", "init", ); + await git(work, "worktree", "add", "-q", "--detach", elsewhere); const gate = join(dir.path, "gate"); const starts = join(dir.path, "starts"); await fs.writeFile(gate, "closed"); @@ -168,6 +175,7 @@ it("mcp: state per thread, no retry of a failed server, connect and disconnect, // Only the forked thread runs `extra`: MCP config may differ per thread. let forkId: string | undefined; + let siblingId: string | undefined; const perThread = define({ id: "test.mcp.per-thread", kind: "mcp", @@ -208,7 +216,8 @@ it("mcp: state per thread, no retry of a failed server, connect and disconnect, const result = await Effect.gen(function* () { const session = yield* Session.create({ directory: work, model: { provider: "openai", id: "gpt-5.6-luna" } }); const main = session.main; - const label = (threadId: string) => (threadId === main.id ? "main" : threadId === forkId ? "fork" : threadId); + const label = (threadId: string) => + threadId === main.id ? "main" : threadId === forkId ? "fork" : threadId === siblingId ? "sibling" : threadId; const seen = yield* statusEvents(label); const mcp = yield* Mcp.Service; const status = (threadId: string) => @@ -251,9 +260,15 @@ it("mcp: state per thread, no retry of a failed server, connect and disconnect, forkId = forked.thread.id; const fork = forked.thread; yield* Session.relink({ sessionId: session.id, threadId: fork.id, directory: elsewhere }); + // A sibling stays where main is, so it shares main's instances. + const sibling = (yield* main.fork({ entryId: line.at(-1)!.entry.id })).thread; + siblingId = sibling.id; + model.turn(sibling.id, [pendingCall("mcp__e2e__pid", {}, "sibling-pid")]); model.turn(main.id, [pendingCall("mcp__e2e__pid", {}, "main-pid-2")]); model.turn(fork.id, [pendingCall("mcp__e2e__pid", {}, "fork-pid")]); - yield* Effect.all([main.run("two"), fork.run("two, forked")], { concurrency: "unbounded" }); + yield* Effect.all([main.run("two"), fork.run("two, forked"), sibling.run("two, sibling")], { + concurrency: "unbounded", + }); const mainStatus = yield* status(main.id); const forkStatus = yield* status(fork.id); const forkRegistered = registered[fork.id]; @@ -322,7 +337,7 @@ it("mcp: state per thread, no retry of a failed server, connect and disconnect, declaredAfterDisconnect, stillDisconnected, startsAfterDisconnect, - calls: { main: yield* results(main), fork: yield* results(fork) }, + calls: { main: yield* results(main), fork: yield* results(fork), sibling: yield* results(sibling) }, changes: changes(model.requests.at(-1)?.context.messages ?? []), loaded: (yield* Document.Service.use((documents) => documents.agent(main.id))).modelTools, restarted, @@ -330,7 +345,7 @@ it("mcp: state per thread, no retry of a failed server, connect and disconnect, disconnectThenConnect, startsAfterRaces, unknown: { tag: unknown._tag, message: unknown.message }, - events: seen, + events: seen.read(), }; // Deleting the session closes its servers and reports nothing more to its threads. @@ -362,8 +377,9 @@ it("mcp: state per thread, no retry of a failed server, connect and disconnect, expect(result.firstDeclared).not.toContain(name); expect(result.calls.main["main-pid"]).toMatch(/^\d+$/); expect(result.calls.fork["fork-pid"]).toMatch(/^\d+$/); - // Each thread runs its own process. + // A thread placed elsewhere runs its own process. expect(result.calls.fork["fork-pid"]).not.toBe(result.calls.main["main-pid-2"]); + expect(result.calls.sibling["sibling-pid"]).toBe(result.calls.main["main-pid-2"]); expect(result.calls.main["main-pid-2"]).toBe(result.calls.main["main-pid"]); expect(result.mainStatus.extra).toBeUndefined(); expect(result.forkStatus.extra?.state).toBe("connected"); @@ -401,7 +417,9 @@ it("mcp: state per thread, no retry of a failed server, connect and disconnect, pids.forEach((pid, index) => { serialized = serialized.replaceAll(`"${pid}"`, `""`); }); - const artifact = JSON.parse(serialized.replaceAll(forkId ?? "\0", "")); + const artifact = JSON.parse( + serialized.replaceAll(forkId ?? "\0", "").replaceAll(siblingId ?? "\0", ""), + ); await expect(`${JSON.stringify(artifact, null, "\t")}\n`).toMatchFileSnapshot("./__artifacts__/mcp.thread.json"); }, 60_000); @@ -471,3 +489,50 @@ it("mcp: a loaded deferred tool's server is awaited up front, and its name stays "./__artifacts__/mcp.thread.loaded.json", ); }, 60_000); + +it("mcp: an instance that idles out tells the threads still using it", async () => { + await using dir = await tmpdir(); + const home = join(dir.path, "home"); + const work = join(dir.path, "work"); + await fs.mkdir(work, { recursive: true }); + await write(join(home, "mcp.json"), { mcpServers: { idle: stdio() } }); + + const model = scripted(); + const result = await Effect.gen(function* () { + const session = yield* Session.create({ directory: work, model: { provider: "openai", id: "gpt-5.6-luna" } }); + const main = session.main; + const seen = yield* statusEvents((threadId) => (threadId === main.id ? "main" : threadId)); + const mcp = yield* Mcp.Service; + const stateOf = mcp.status(main.id).pipe(Effect.map((all) => all[0]?.state)); + // The turn's drain ends with it; nothing holds the instance after that. + yield* main.run("one"); + const afterTurn = yield* stateOf; + const idled = yield* waitFor(stateOf, (state) => state === "disconnected"); + const events = yield* waitFor( + Effect.sync(() => seen.read()), + (read) => read.main?.idle?.at(-1)?.state === "disconnected", + ); + // Connect does not revive it; the thread's next turn does. + const connect = yield* Effect.flip(mcp.connect(main.id, "idle")); + return { afterTurn, idled, events, connect: { tag: connect._tag, message: connect.message } }; + }).pipe( + Effect.provide( + Harness.layer({ + home, + hostCwd: dir.path, + database: ":memory:", + plugins: [defaultPromptPlugin], + llm: model.open, + mcpIdleTimeout: "500 millis", + }), + ), + Effect.scoped, + Effect.runPromise, + ); + + expect(result.afterTurn).toBe("connected"); + expect(result.idled).toBe("disconnected"); + expect(result.events.main?.idle?.at(-1)?.state).toBe("disconnected"); + expect(result.connect.message).toContain("was closed while idle"); + await expect(`${JSON.stringify(result, null, "\t")}\n`).toMatchFileSnapshot("./__artifacts__/mcp.thread.idle.json"); +}, 60_000); diff --git a/packages/plugin/src/plugin/mcp.ts b/packages/plugin/src/plugin/mcp.ts index 19c45d31..6735c409 100644 --- a/packages/plugin/src/plugin/mcp.ts +++ b/packages/plugin/src/plugin/mcp.ts @@ -20,6 +20,15 @@ */ export type Exposure = "direct" | "codemode" | "deferred" | "hidden"; +/** + * Who shares one running instance of a server. + * + * - `session`: threads of one session at the same place share one. + * - `space`: sessions in the same space (sandbox and folder) share one. + * - `app`: every session shares one. + */ +export type Scope = "session" | "space" | "app"; + interface Base { /** Default: `direct`. */ readonly exposure?: Exposure; @@ -31,6 +40,8 @@ interface Base { readonly toolExposure?: Readonly>; /** `false` keeps the entry without connecting. Default: `true`. */ readonly enabled?: boolean; + /** Default: `session`. Share a server only when it is safe to share. */ + readonly scope?: Scope; /** Per-request timeout in seconds. Progress notifications from the server reset it. Default: 60. */ readonly timeout?: number; /** What the server offers, in a sentence. */ diff --git a/sites/www/content/docs/customize/mcp.mdx b/sites/www/content/docs/customize/mcp.mdx index ffc65dee..4df767c0 100644 --- a/sites/www/content/docs/customize/mcp.mdx +++ b/sites/www/content/docs/customize/mcp.mdx @@ -40,6 +40,7 @@ User servers go in `~/.codework/mcp.json`, project servers in `.codework/mcp.jso | `command`, `args`, `env`, `cwd` | stdio | One executable and its arguments, not a shell line. `cwd` is relative to where the server starts. | | `url`, `headers` | HTTP | A streamable HTTP endpoint. SSE is not supported. | | `enabled`, `timeout`, `exposure`, `toolExposure` | both | Off switch, per-request timeout in seconds (default 60), and which tools the model sees. | +| `scope` | both | Who shares one running server: `session` (default), `space` or `app`. | `env` and `headers` values can use `${NAME}` or `$NAME` for an environment variable (`$$` for a literal `$`), or `!command` to use a command's output. A command runs once, and its output is reused until CodeWork restarts. @@ -55,19 +56,43 @@ A project entry with the same name replaces a user entry. An entry with only `en { "mcpServers": { "internal-tools": { "enabled": false } } } ``` +## Sharing a Server + +By default each session runs its own copy of a server. `scope` shares one instead: + +| Scope | Shared by | A stdio server starts in | +| --------- | ----------------------------------------- | ------------------------ | +| `session` | Threads of one session in the same folder | The thread's directory | +| `space` | Sessions in the same folder and sandbox | The folder | +| `app` | Every session | Where CodeWork started | + +When the session's sandbox isn't the local one, `session` and `space` servers start in the session's folder on your machine instead. `app` servers are sent no roots, so a path like `.` in their arguments means where CodeWork started. + +Share a server only when it keeps no per-session state. Sessions whose settings for a server differ never share it. + ## Which Tools the Model Sees -`exposure` is `direct` (the default) or `hidden`. `toolExposure` sets single tools, by name or with `*` patterns: +`exposure` is `direct` (the default), `hidden`, `codemode` or `deferred`: + +| Exposure | The model | +| ---------- | ------------------------------------------------------------------------------------------ | +| `direct` | Sees the tools like any other tool. | +| `deferred` | Finds them with `toolsearch` (turn on `codework.tool.toolsearch`), then sees them. | +| `hidden` | Never sees them. | +| `codemode` | Can't reach them yet: they need codemode, which is coming. CodeWork warns once per thread. | + +`toolExposure` sets single tools, by name or with `*` patterns: ```json { "exposure": "hidden", "toolExposure": { "search_*": "direct" } } ``` -Servers with resources also add `list_mcp_resources`, `list_mcp_resource_templates` and `read_mcp_resource`. +Servers with resources also add `list_mcp_resources`, `list_mcp_resource_templates` and `read_mcp_resource`, when at least one of them is `direct`. ## Connections -- Servers connect in the background. A message waits up to 10 seconds for servers it starts whose tools the model sees: all of them at first, then new or changed ones. -- If a server stops, the next call starts it again. If it fails to connect, your next message tries again. +- Servers connect in the background. A message waits up to 10 seconds for servers it starts whose tools the model sees or has found: all of them at first, then new or changed ones. +- If a server stops, the next call starts it again. A server that fails to connect stays failed until you change its settings. - Edits to `mcp.json` apply from your next message. Changing only `exposure` or `toolExposure` keeps the server running. +- A server no running session needs closes after 10 minutes, and starts again when one does. - Server log messages go to `~/.codework/mcp.log`.