diff --git a/packages/core/src/mcp/client.ts b/packages/core/src/mcp/client.ts index 49a9e0fc210..b5cba91ff8a 100644 --- a/packages/core/src/mcp/client.ts +++ b/packages/core/src/mcp/client.ts @@ -1,11 +1,10 @@ export * as MCPClient from "./client.js" import path from "node:path" -import { execFile } from "node:child_process" import { pathToFileURL } from "node:url" import { Client } from "@modelcontextprotocol/sdk/client/index.js" -import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js" import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js" +import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js" import { UnauthorizedError, type OAuthClientProvider } from "@modelcontextprotocol/sdk/client/auth.js" import { CallToolResultSchema, @@ -30,13 +29,12 @@ import { } from "@modelcontextprotocol/sdk/types.js" import { Cause, Effect, Exit, Schema } from "effect" import { ConfigMCP } from "@opencode-ai/schema/config/mcp" +import { MCPStdio } from "./stdio.js" const DEFAULT_STARTUP_TIMEOUT = 30_000 const DEFAULT_CATALOG_TIMEOUT = 30_000 const DEFAULT_EXECUTION_TIMEOUT = 12 * 60 * 60 * 1_000 // 12 hours -type Transport = StdioClientTransport | StreamableHTTPClientTransport - // Some servers advertise tool outputSchemas the SDK's strict validator can't resolve; this drops // only that field so a single bad schema doesn't blank out the whole tool list. const TolerantListToolsResult = ListToolsResultSchema.extend({ @@ -176,7 +174,12 @@ export interface Connection { readonly onResourcesChanged: (callback: () => void) => void } -/** Connects an MCP server; closing the calling scope tears down the transport and any spawned process. */ +/** + * Connects an MCP server; closing the calling scope tears down the transport and any spawned process. + * + * A stdio server is spawned through the location's `Environment`, so it runs on the same execution + * plane as the location's shell commands rather than always on the host. + */ export const connect = Effect.fnUntraced(function* ( server: string, config: typeof ConfigMCP.Server.Type, @@ -190,13 +193,12 @@ export const connect = Effect.fnUntraced(function* ( const transport: Transport = yield* Effect.gen(function* () { if (config.type === "local") { const [command, ...args] = config.command - return new StdioClientTransport({ + return yield* MCPStdio.make({ + server, command, args, cwd: config.cwd ? path.resolve(directory, config.cwd) : directory, - stderr: "pipe", - env: { - ...(process.env as Record), + environment: { ...(command === "opencode" ? { BUN_BE_BUN: "1" } : {}), ...config.environment, }, @@ -233,9 +235,9 @@ export const connect = Effect.fnUntraced(function* ( catch: (error) => error, }).pipe(Effect.exit) if (Exit.isSuccess(exit)) { - yield* Effect.addFinalizer(() => - cleanupStdioDescendants(transport).pipe(Effect.andThen(Effect.promise(() => client.close())), Effect.ignore), - ) + // Closing the client closes the transport, which ends stdin and then kills through the spawner + // handle if the server does not exit cleanly. The process scope remains a final backstop. + yield* Effect.addFinalizer(() => Effect.promise(() => client.close()).pipe(Effect.ignore)) const catalogTimeout = config.timeout?.catalog ?? DEFAULT_CATALOG_TIMEOUT const executionTimeout = config.timeout?.execution ?? DEFAULT_EXECUTION_TIMEOUT return { @@ -434,58 +436,12 @@ export const connect = Effect.fnUntraced(function* ( } satisfies Connection } - yield* cleanupStdioDescendants(transport).pipe(Effect.andThen(Effect.promise(() => transport.close())), Effect.ignore) + yield* Effect.promise(() => transport.close()).pipe(Effect.ignore) const error = Cause.squash(exit.cause) if (error instanceof UnauthorizedError) return yield* new NeedsAuthError({ server }) return yield* new ConnectError({ server, message: error instanceof Error ? error.message : String(error) }) }) -// SDK close stops the MCP process, but not child processes it spawned. -const cleanupStdioDescendants = (transport: Transport) => - Effect.gen(function* () { - if (!(transport instanceof StdioClientTransport)) return - const pid = transport.pid - if (typeof pid !== "number") return - yield* Effect.forEach( - yield* descendantPids(pid), - (pid) => - Effect.try({ - try: () => process.kill(pid, "SIGTERM"), - catch: () => undefined, - }).pipe(Effect.ignore), - { discard: true }, - ) - }) - -const descendantPids = Effect.fnUntraced(function* (root: number) { - if (process.platform === "win32") return [] - const result: number[] = [] - const queue = [root] - for (let index = 0; index < queue.length; index++) { - const parent = queue[index] - if (parent === undefined) return result - const children = (yield* childPids(parent)).filter((pid) => !result.includes(pid)) - result.push(...children) - queue.push(...children) - } - return result -}) - -const childPids = (pid: number) => - Effect.promise( - () => - new Promise((resolve) => { - execFile("pgrep", ["-P", String(pid)], { encoding: "utf8" }, (_error, stdout) => { - resolve( - stdout - .split("\n") - .map((line) => Number.parseInt(line, 10)) - .filter((pid) => Number.isInteger(pid)), - ) - }) - }), - ) - async function paginate( list: (cursor: string | undefined) => Promise, items: (result: R) => T[], diff --git a/packages/core/src/mcp/index.ts b/packages/core/src/mcp/index.ts index c1020cbff41..4685bc31eb5 100644 --- a/packages/core/src/mcp/index.ts +++ b/packages/core/src/mcp/index.ts @@ -12,6 +12,7 @@ import { makeLocationNode } from "@opencode-ai/util/effect/app-node" import { Config } from "../config.js" import { Credential } from "../credential.js" import { Bus } from "../bus.js" +import { Environment } from "../environment/index.js" import { Form } from "../form.js" import { Integration } from "../integration.js" import { KeyedMutex } from "../effect/keyed-mutex.js" @@ -173,13 +174,13 @@ export const layer = (options?: Options) => Effect.gen(function* () { const config = yield* Config.Service const location = yield* Location.Service + const environment = yield* Environment.Service const bus = yield* Bus.Service const forms = yield* Form.Service const integration = yield* Integration.Service const credentials = yield* Credential.Service - const root = yield* Scope.make() + const root = yield* Effect.scope const fork = yield* FiberSet.makeRuntime() - yield* Effect.addFinalizer((exit) => Scope.close(root, exit)) const loadConfig = (entries: readonly Entry[]) => { const documents = entries.filter((entry): entry is Document => entry.type === "document") @@ -459,13 +460,8 @@ export const layer = (options?: Options) => connection.onClose(() => live( Effect.gen(function* () { - entry.client = undefined - entry.tools = undefined - entry.prompts = undefined entry.status = { status: "failed", error: "Connection closed" } - yield* bus.publish(McpEvent.ToolsChanged, { server: name }).pipe(Effect.ignore) - yield* bus.publish(McpEvent.ResourcesChanged, { server: name }).pipe(Effect.ignore) - yield* bus.publish(Command.Event.Updated, {}).pipe(Effect.ignore) + yield* stopServer(name, entry) yield* bus.publish(McpEvent.StatusChanged, { server: name }).pipe(Effect.ignore) }), ), @@ -520,6 +516,8 @@ export const layer = (options?: Options) => options?.clientInfo, ).pipe( Effect.flatMap((connection) => connection.tools().pipe(Effect.map((tools) => ({ connection, tools })))), + // A stdio server is spawned on this location's execution plane, not the host's. + Effect.provideService(Environment.Service, environment), Scope.provide(scope), Effect.exit, ) @@ -828,7 +826,7 @@ export function configured(options?: Options) { return makeLocationNode({ service: Service, layer: layer(options), - deps: [Config.node, Location.node, Bus.node, Form.node, Integration.node, Credential.node], + deps: [Config.node, Location.node, Environment.node, Bus.node, Form.node, Integration.node, Credential.node], }) } diff --git a/packages/core/src/mcp/stdio.ts b/packages/core/src/mcp/stdio.ts new file mode 100644 index 00000000000..2da5c71a1d0 --- /dev/null +++ b/packages/core/src/mcp/stdio.ts @@ -0,0 +1,181 @@ +export * as MCPStdio from "./stdio.js" + +import { ReadBuffer, serializeMessage } from "@modelcontextprotocol/sdk/shared/stdio.js" +import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js" +import type { JSONRPCMessage } from "@modelcontextprotocol/sdk/types.js" +import { Cause, Duration, Effect, Queue, Scope, Stream } from "effect" +import { ChildProcess } from "effect/unstable/process" +import type { ChildProcessHandle } from "effect/unstable/process/ChildProcessSpawner" +import { Environment } from "../environment/index.js" + +/** Mirrors StdioClientTransport: wait this long for a graceful exit after stdin closes. */ +const CLOSE_GRACE = Duration.seconds(2) + +/** Mirrors StdioClientTransport: escalate SIGTERM to SIGKILL after this long. */ +const FORCE_KILL_AFTER = Duration.seconds(2) +const OUTGOING_CAPACITY = 64 +const MAX_FRAME_BYTES = 16 * 1024 * 1024 + +export interface Options { + /** Server name; only used to attribute logs. */ + readonly server: string + readonly command: string + readonly args: ReadonlyArray + readonly cwd: string + /** + * Environment declared by the server config, and nothing else. + * + * The host environment is merged in by the spawner via `extendEnv`, which keeps the merge on the + * side that actually runs the process: the local driver extends with the host's `process.env` + * (what the MCP SDK's transport did), while a workspace driver extends with the sandbox's own + * environment. Host variables therefore never cross the seam into a remote workspace. + */ + readonly environment: Record +} + +/** + * MCP stdio transport that spawns its server through the location's `Environment` instead of the + * SDK's host-bound `StdioClientTransport`, so a workspace-backed location runs its MCP servers + * wherever the rest of its execution happens. + * + * The process is acquired in the calling scope: closing the scope kills it (the spawner kills the + * whole process group, so descendants go too) regardless of whether the transport was closed. + */ +export const make = Effect.fnUntraced(function* (options: Options) { + const environment = yield* Environment.Service + const scope = yield* Effect.scope + // Outgoing frames are queued rather than written to `handle.stdin` directly: the sink closes the + // stream it is run with, and stdin must stay open across the whole session. + const outgoing = yield* Queue.bounded(OUTGOING_CAPACITY) + const buffer = new ReadBuffer() + const state: { phase: "ready" | "starting" | "open" | "closed"; handle?: ChildProcessHandle } = { phase: "ready" } + let startup: Promise | undefined + let closing: Promise | undefined + let trailingBytes = 0 + + const stop = (handle: ChildProcessHandle) => + Effect.gen(function* () { + const exit = yield* Effect.timeoutOption(handle.exitCode, CLOSE_GRACE) + if (exit._tag === "Some") return + const terminated = yield* Effect.timeoutOption(handle.kill({ killSignal: "SIGTERM" }), FORCE_KILL_AFTER) + if (terminated._tag === "None") yield* handle.kill({ killSignal: "SIGKILL" }) + }).pipe(Effect.ignore) + + const close = () => + (closing ??= Effect.runPromise( + Effect.gen(function* () { + state.phase = "closed" + Queue.endUnsafe(outgoing) + if (startup) yield* Effect.promise(() => startup!.catch(() => undefined)) + const handle = state.handle + if (!handle) return + state.handle = undefined + yield* stop(handle) + }).pipe(Effect.ensuring(Queue.shutdown(outgoing)), Effect.ensuring(Effect.sync(() => buffer.clear()))), + )) + + const transport: Transport = { + start: () => { + if (state.phase !== "ready") return Promise.reject(new Error("Stdio transport already started")) + state.phase = "starting" + startup = Effect.runPromise( + Effect.gen(function* () { + const handle = yield* environment.spawner.spawn( + ChildProcess.make(options.command, [...options.args], { + cwd: options.cwd, + env: options.environment, + extendEnv: true, + stdin: { stream: Stream.encodeText(Stream.fromQueue(outgoing)), endOnDone: true }, + stdout: "pipe", + stderr: "pipe", + forceKillAfter: FORCE_KILL_AFTER, + }), + ) + state.handle = handle + if (state.phase === "closed") { + state.handle = undefined + return yield* stop(handle) + } + state.phase = "open" + yield* startOutput(handle) + }).pipe(Scope.provide(scope)), + ) + return startup + }, + send: (message: JSONRPCMessage) => + state.phase !== "open" + ? Promise.reject(new Error("Not connected")) + : Effect.runPromise( + Queue.offer(outgoing, serializeMessage(message)).pipe( + Effect.flatMap((offered) => (offered ? Effect.void : Effect.fail(new Error("Not connected")))), + ), + ), + close, + } + + const deliver = (chunk: Uint8Array) => + Effect.gen(function* () { + for (const byte of chunk) { + trailingBytes = byte === 10 ? 0 : trailingBytes + 1 + if (trailingBytes > MAX_FRAME_BYTES) return yield* Effect.fail(new Error("MCP stdio frame exceeded 16 MiB")) + } + buffer.append(Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength)) + while (true) { + // `undefined` means the frame failed to parse: the buffer has already advanced past it, so + // keep draining. `null` means the buffer holds no complete frame yet. + const message = yield* Effect.try({ + try: () => buffer.readMessage(), + catch: (cause) => (cause instanceof Error ? cause : new Error(String(cause))), + }).pipe( + Effect.catch((error) => + Effect.sync(() => { + transport.onerror?.(error) + return undefined + }), + ), + ) + if (message === undefined) continue + if (message === null) return + transport.onmessage?.(message) + } + }) + + const startOutput = (handle: ChildProcessHandle) => + Effect.gen(function* () { + yield* Effect.forkScoped( + Stream.runForEach(handle.stdout, deliver).pipe( + Effect.tapCause((cause) => + Effect.sync(() => { + const error = Cause.squash(cause) + transport.onerror?.(error instanceof Error ? error : new Error(String(error))) + }), + ), + Effect.ignore, + // stdout ending means the server is gone; the SDK transport reports that the same way. + Effect.ensuring( + Effect.gen(function* () { + const unexpected = state.phase !== "closed" + if (unexpected) yield* Effect.promise(close) + transport.onclose?.() + }), + ), + ), + ) + + // StdioClientTransport pipes stderr into a stream nobody reads. Drain chunks into the debug + // log so chatty servers cannot stall and newline-free output is not buffered without bound. + yield* Effect.forkScoped( + handle.stderr.pipe( + Stream.decodeText(), + Stream.runForEach((output) => + output.trim() === "" + ? Effect.void + : Effect.logDebug("mcp server stderr", { server: options.server, output }), + ), + Effect.ignore, + ), + ) + }) + + return transport +}) diff --git a/packages/core/test/fixture/environment.ts b/packages/core/test/fixture/environment.ts index 67c4f858acc..d9a341a6265 100644 --- a/packages/core/test/fixture/environment.ts +++ b/packages/core/test/fixture/environment.ts @@ -1,7 +1,42 @@ import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" import { Environment } from "@opencode-ai/core/environment/index" import { Location } from "@opencode-ai/core/location" +import { CrossSpawnSpawner } from "@opencode-ai/util/cross-spawn-spawner" +import { LayerNode } from "@opencode-ai/util/effect/layer-node" import { Effect, Layer } from "effect" +import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process" + +/** + * The host environment, without the workspace machinery: what a location with no `workspaceID` + * resolves to. + */ +export const hostEnvironmentLayer = Layer.effect( + Environment.Service, + Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner + const driver = Environment.makeLocalDriver(spawner) + return Environment.Service.of({ files: Environment.makeFiles(driver), spawner: driver.spawner }) + }), +).pipe(Layer.provide(LayerNode.compile(CrossSpawnSpawner.node))) + +/** + * The host environment with its spawner wrapped so a test can assert on every command that crosses + * the seam. Spawning still really happens, so the process under test behaves normally. + */ +export const recordingEnvironmentLayer = (spawns: Array) => + Layer.effect( + Environment.Service, + Effect.gen(function* () { + const environment = yield* Environment.Service + return Environment.Service.of({ + ...environment, + spawner: ChildProcessSpawner.make((command) => { + spawns.push(command) + return environment.spawner.spawn(command) + }), + }) + }), + ).pipe(Layer.provide(hostEnvironmentLayer)) export type EnvironmentFilesTransform = (files: Environment.Files) => Partial diff --git a/packages/core/test/mcp.test.ts b/packages/core/test/mcp.test.ts index 1f2cf2d2d2b..9c2e49c628b 100644 --- a/packages/core/test/mcp.test.ts +++ b/packages/core/test/mcp.test.ts @@ -22,19 +22,24 @@ import { Bus } from "@opencode-ai/core/bus" import { ID, type Payload } from "@opencode-ai/schema/event" import { Form } from "@opencode-ai/core/form" import { Integration } from "@opencode-ai/core/integration" +import { Environment } from "@opencode-ai/core/environment/index" import { Location } from "@opencode-ai/core/location" import { MCP } from "@opencode-ai/core/mcp/index" import { MCPClient } from "@opencode-ai/core/mcp/client" +import { MCPStdio } from "@opencode-ai/core/mcp/stdio" import { Permission } from "@opencode-ai/core/permission" import { AbsolutePath } from "@opencode-ai/core/schema" import { Session } from "@opencode-ai/core/session" import { McpTool } from "@opencode-ai/core/tool/mcp" import { Tool } from "@opencode-ai/core/tool" -import { DateTime, Deferred, Effect, Exit, Fiber, Layer, PubSub, Schedule, Schema, Stream } from "effect" +import { DateTime, Deferred, Effect, Exit, Fiber, Layer, PubSub, Schedule, Schema, Sink, Stream } from "effect" +import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process" +import { ExitCode, makeHandle, ProcessId } from "effect/unstable/process/ChildProcessSpawner" import { Image } from "@opencode-ai/core/image" import { testEffect } from "./lib/effect" import { imagePassthrough } from "./lib/image" import { location } from "./fixture/location" +import { hostEnvironmentLayer, recordingEnvironmentLayer } from "./fixture/environment" import { executeTool, toolDefinitions, toolIdentity, waitForCodeModeTool, waitForTool } from "./lib/tool" let assertion: Deferred.Deferred | undefined @@ -167,6 +172,7 @@ function resourceMcpLayer( overrides?: { entries?: Config.Interface["entries"] subscribe?: Bus.Interface["subscribe"] + environment?: Layer.Layer }, ) { const directory = AbsolutePath.make(import.meta.dir) @@ -229,11 +235,15 @@ function resourceMcpLayer( }, }), Layer.mock(Credential.Service, {}), + overrides?.environment ?? hostEnvironmentLayer, ), ), ) } +const connect = (server: string, config: typeof ConfigMCP.Server.Type, directory: string) => + MCPClient.connect(server, config, directory).pipe(Effect.provide(hostEnvironmentLayer)) + const mcp = Layer.mock(MCP.Service, { tools: () => Effect.succeed([ @@ -407,7 +417,7 @@ test("retains output schemas across paginated MCP discovery", async () => { const tools = await Effect.runPromise( Effect.scoped( Effect.gen(function* () { - const connection = yield* MCPClient.connect( + const connection = yield* connect( "pagination", new ConfigMCP.Local({ type: "local", @@ -440,11 +450,139 @@ test("retains output schemas across paginated MCP discovery", async () => { ]) }) +test("spawns local MCP servers through the location environment", async () => { + const spawns: Array = [] + const cwd = path.join(import.meta.dir, "fixture") + const config = new ConfigMCP.Local({ + type: "local", + command: [process.execPath, path.join(import.meta.dir, "fixture/mcp-output-schema.ts")], + cwd: "fixture", + environment: { MCP_LOCATION_TEST: "configured" }, + }) + + await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const connection = yield* MCPClient.connect("environment", config, import.meta.dir) + yield* connection.tools() + }), + ).pipe(Effect.provide(recordingEnvironmentLayer(spawns))), + ) + + expect(spawns).toHaveLength(1) + const command = spawns[0] + if (!command || !ChildProcess.isStandardCommand(command)) throw new Error("Expected a standard process command") + expect(command.command).toBe(process.execPath) + expect(command.options.cwd).toBe(cwd) + expect(command.options.extendEnv).toBe(true) + expect(command.options.env).toEqual({ MCP_LOCATION_TEST: "configured" }) +}) + +test("rejects sends before the stdio transport is started", async () => { + await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const transport = yield* MCPStdio.make({ + server: "not-started", + command: process.execPath, + args: [path.join(import.meta.dir, "fixture/mcp-output-schema.ts")], + cwd: import.meta.dir, + environment: {}, + }) + yield* Effect.tryPromise({ + try: () => transport.send({ jsonrpc: "2.0", method: "notifications/initialized" }), + catch: (cause) => (cause instanceof Error ? cause : new Error(String(cause))), + }).pipe( + Effect.flip, + Effect.tap((error) => Effect.sync(() => expect(error.message).toBe("Not connected"))), + ) + }).pipe(Effect.provide(hostEnvironmentLayer)), + ), + ) +}) + +test("joins concurrent stdio transport closes", async () => { + await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const transport = yield* MCPStdio.make({ + server: "concurrent-close", + command: "unused", + args: [], + cwd: import.meta.dir, + environment: {}, + }) + const first = transport.close() + expect(transport.close()).toBe(first) + yield* Effect.promise(() => first) + }).pipe(Effect.provide(hostEnvironmentLayer)), + ), + ) +}) + +test("closes a stdio process that finishes spawning after close", async () => { + const spawning = Deferred.makeUnsafe() + const release = Deferred.makeUnsafe() + const exited = Deferred.makeUnsafe() + const signals: Array = [] + const driver = Environment.makeMemoryDriver() + const environment = Layer.succeed( + Environment.Service, + Environment.Service.of({ + files: Environment.makeFiles(driver), + spawner: ChildProcessSpawner.make(() => + Effect.gen(function* () { + yield* Deferred.succeed(spawning, undefined) + yield* Deferred.await(release) + return makeHandle({ + pid: ProcessId(1), + exitCode: Deferred.await(exited), + isRunning: Deferred.isDone(exited).pipe(Effect.map((done) => !done)), + kill: (options) => + Effect.gen(function* () { + signals.push(options?.killSignal ?? "SIGTERM") + yield* Deferred.succeed(exited, ExitCode(143)) + }), + stdin: Sink.drain, + stdout: Stream.never, + stderr: Stream.empty, + all: Stream.never, + getInputFd: () => Sink.drain, + getOutputFd: () => Stream.empty, + unref: Effect.succeed(Effect.void), + }) + }), + ), + }), + ) + + await Effect.runPromise( + Effect.scoped( + Effect.gen(function* () { + const transport = yield* MCPStdio.make({ + server: "close-during-spawn", + command: "unused", + args: [], + cwd: import.meta.dir, + environment: {}, + }) + const start = transport.start() + yield* Deferred.await(spawning) + const close = transport.close() + yield* Deferred.succeed(release, undefined) + yield* Effect.promise(() => Promise.all([start, close])) + }).pipe(Effect.provide(environment)), + ), + ) + + expect(signals).toEqual(["SIGTERM"]) +}) + test("applies the configured MCP catalog timeout", async () => { const result = Effect.runPromise( Effect.scoped( Effect.gen(function* () { - const connection = yield* MCPClient.connect( + const connection = yield* connect( "catalog-timeout", new ConfigMCP.Local({ type: "local", @@ -466,7 +604,7 @@ test("applies the configured MCP execution timeout", async () => { const result = Effect.runPromise( Effect.scoped( Effect.gen(function* () { - const connection = yield* MCPClient.connect( + const connection = yield* connect( "execution-timeout", new ConfigMCP.Local({ type: "local", @@ -487,7 +625,7 @@ test("applies the configured MCP execution timeout to prompts", async () => { const result = Effect.runPromise( Effect.scoped( Effect.gen(function* () { - const connection = yield* MCPClient.connect( + const connection = yield* connect( "prompt-timeout", new ConfigMCP.Local({ type: "local", @@ -508,7 +646,7 @@ test("applies configured MCP timeouts to resource operations", async () => { const catalog = Effect.runPromise( Effect.scoped( Effect.gen(function* () { - const connection = yield* MCPClient.connect( + const connection = yield* connect( "resource-catalog-timeout", new ConfigMCP.Local({ type: "local", @@ -527,7 +665,7 @@ test("applies configured MCP timeouts to resource operations", async () => { const read = Effect.runPromise( Effect.scoped( Effect.gen(function* () { - const connection = yield* MCPClient.connect( + const connection = yield* connect( "resource-read-timeout", new ConfigMCP.Local({ type: "local", @@ -562,7 +700,7 @@ test("lists, reads, and reports MCP resource changes", async () => { }, "templates-2": { items: [{ name: "Issue", uriTemplate: "issue://{id}", description: "Issue" }] }, } - const connection = yield* MCPClient.connect( + const connection = yield* connect( "resources", new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false }), import.meta.dir, @@ -633,7 +771,7 @@ test("skips MCP resource requests when the capability is absent", async () => { Effect.scoped( Effect.gen(function* () { const server = yield* resourceServer({ resources: false }) - const connection = yield* MCPClient.connect( + const connection = yield* connect( "resources", new ConfigMCP.Remote({ type: "remote", url: server.url, oauth: false }), import.meta.dir,