From ca0c4237a2ccff24f7ee0a5abde5bf7476cab888 Mon Sep 17 00:00:00 2001 From: Tyler Gibbs Date: Fri, 24 Jul 2026 01:00:08 -0500 Subject: [PATCH] answer unsupported discovery before mcp initialization --- .changeset/mcp-discover-fallback.md | 5 + apps/cli/src/main.ts | 32 +++- .../hosts/mcp/src/stdio-integration.test.ts | 162 +++++++++++++++--- 3 files changed, 176 insertions(+), 23 deletions(-) create mode 100644 .changeset/mcp-discover-fallback.md diff --git a/.changeset/mcp-discover-fallback.md b/.changeset/mcp-discover-fallback.md new file mode 100644 index 000000000..fca683f91 --- /dev/null +++ b/.changeset/mcp-discover-fallback.md @@ -0,0 +1,5 @@ +--- +"executor": patch +--- + +keep `executor mcp` connections open when clients probe the unsupported `server/discover` method, allowing them to fall back to the legacy `initialize` handshake. diff --git a/apps/cli/src/main.ts b/apps/cli/src/main.ts index bae18617c..35d5dadb1 100644 --- a/apps/cli/src/main.ts +++ b/apps/cli/src/main.ts @@ -77,7 +77,12 @@ import * as Option from "effect/Option"; import * as Cause from "effect/Cause"; import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js"; import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js"; -import type { JSONRPCMessage } from "@modelcontextprotocol/sdk/types.js"; +import { + ErrorCode, + isJSONRPCRequest, + type JSONRPCErrorResponse, + type JSONRPCMessage, +} from "@modelcontextprotocol/sdk/types.js"; import { ExecutorApi, checkForUpdate } from "@executor-js/api"; import { @@ -1348,6 +1353,18 @@ const mcpUrlForActiveLocalServer = ( return url; }; +const preInitializeMcpFallback = (message: JSONRPCMessage): JSONRPCErrorResponse | null => + isJSONRPCRequest(message) && message.method === "server/discover" + ? { + jsonrpc: "2.0", + id: message.id, + error: { + code: ErrorCode.MethodNotFound, + message: "Method not found", + }, + } + : null; + /** * Bridge a stdio MCP client to a local server's HTTP `/mcp` endpoint. `executor * mcp` owns NO database: it forwards JSON-RPC between the client's stdin/stdout @@ -1425,8 +1442,17 @@ const runMcpHttpBridge = async (input: { http.onclose = shutdown; stdio.onerror = (error) => reportError("stdio transport error", error); http.onerror = (error) => reportError("daemon transport error", error); - stdio.onmessage = forwardMessage((message) => http.send(message), "failed to send to daemon"); - http.onmessage = forwardMessage((message) => stdio.send(message), "failed to send to stdio"); + const sendToDaemon = forwardMessage((message) => http.send(message), "failed to send to daemon"); + const sendToStdio = forwardMessage((message) => stdio.send(message), "failed to send to stdio"); + stdio.onmessage = (message) => { + const fallback = http.sessionId ? null : preInitializeMcpFallback(message); + if (fallback) { + sendToStdio(fallback); + return; + } + sendToDaemon(message); + }; + http.onmessage = sendToStdio; try { await http.start(); diff --git a/packages/hosts/mcp/src/stdio-integration.test.ts b/packages/hosts/mcp/src/stdio-integration.test.ts index d66f9893f..066b39679 100644 --- a/packages/hosts/mcp/src/stdio-integration.test.ts +++ b/packages/hosts/mcp/src/stdio-integration.test.ts @@ -1,8 +1,12 @@ import { describe, expect, it } from "@effect/vitest"; -import { Client } from "@modelcontextprotocol/sdk/client/index.js"; import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js"; -import { Effect } from "effect"; -import { mkdtempSync } from "node:fs"; +import { + ErrorCode, + LATEST_PROTOCOL_VERSION, + type JSONRPCMessage, +} from "@modelcontextprotocol/sdk/types.js"; +import { Effect, Option, Schema } from "effect"; +import { mkdtempSync, readFileSync, rmSync } from "node:fs"; import { tmpdir } from "node:os"; import { join, resolve } from "node:path"; @@ -10,41 +14,159 @@ const repoRoot = resolve(import.meta.dirname, "../../../.."); const cliEntry = resolve(repoRoot, "apps/cli/src/main.ts"); const testScope = resolve(repoRoot, "apps/local"); +const decodeServerManifest = Schema.decodeUnknownOption( + Schema.fromJsonString(Schema.Struct({ pid: Schema.optional(Schema.Number) })), +); + +const stopAutoSpawnedDaemon = (dataDir: string): Effect.Effect => + Effect.try({ + try: () => readFileSync(join(dataDir, "server-control", "server.json"), "utf8"), + catch: () => undefined, + }).pipe( + Effect.flatMap((text) => { + const manifest = decodeServerManifest(text); + const pid = Option.isSome(manifest) ? manifest.value.pid : undefined; + return pid + ? Effect.try({ + try: () => process.kill(pid, "SIGTERM"), + catch: () => undefined, + }).pipe(Effect.ignore) + : Effect.void; + }), + Effect.ignore, + ); + +const withTempData = Effect.acquireRelease( + Effect.sync(() => mkdtempSync(join(tmpdir(), "executor-mcp-discover-test-"))), + (dataDir) => + stopAutoSpawnedDaemon(dataDir).pipe( + Effect.ensuring(Effect.sync(() => rmSync(dataDir, { recursive: true, force: true }))), + ), +); + +const messageQueue = (transport: StdioClientTransport) => { + const messages: Array = []; + const waiters: Array<(message: JSONRPCMessage) => void> = []; + + transport.onmessage = (message) => { + const waiter = waiters.shift(); + if (waiter) { + waiter(message); + } else { + messages.push(message); + } + }; + + return { + next: (): Promise => { + const message = messages.shift(); + return message ? Promise.resolve(message) : new Promise((resolve) => waiters.push(resolve)); + }, + }; +}; + describe("MCP stdio integration", () => { it.effect( - "execute tool returns result over stdio transport", + "unsupported discovery keeps the connection open for initialization and tool calls", () => Effect.gen(function* () { - // Fresh temp dir so the test doesn't migrate against the developer's - // real ~/.executor/data.db. - const dataDir = mkdtempSync(join(tmpdir(), "executor-mcp-test-")); - + const dataDir = yield* withTempData; const transport = new StdioClientTransport({ command: "bun", args: ["run", cliEntry, "mcp", "--scope", testScope], env: { ...process.env, EXECUTOR_DATA_DIR: dataDir }, }); - - const client = new Client({ name: "test-client", version: "1.0.0" }, { capabilities: {} }); + const responses = messageQueue(transport); yield* Effect.acquireRelease( - Effect.promise(() => client.connect(transport)), + Effect.promise(() => transport.start()), () => Effect.promise(() => transport.close()), ); - const { tools } = yield* Effect.promise(() => client.listTools()); - expect(tools.map((t) => t.name)).toContain("execute"); + yield* Effect.promise(() => + transport.send({ + jsonrpc: "2.0", + id: 1, + method: "server/discover", + params: { + _meta: { + "io.modelcontextprotocol/protocolVersion": "2026-07-28", + "io.modelcontextprotocol/clientInfo": { + name: "discovery-test-client", + version: "1.0.0", + }, + "io.modelcontextprotocol/clientCapabilities": {}, + }, + }, + }), + ); + + expect(yield* Effect.promise(() => responses.next())).toEqual({ + jsonrpc: "2.0", + id: 1, + error: { + code: ErrorCode.MethodNotFound, + message: "Method not found", + }, + }); - const result = yield* Effect.promise(() => - client.callTool({ - name: "execute", - arguments: { code: "return 2+2" }, + yield* Effect.promise(() => + transport.send({ + jsonrpc: "2.0", + id: 2, + method: "initialize", + params: { + protocolVersion: LATEST_PROTOCOL_VERSION, + capabilities: {}, + clientInfo: { + name: "discovery-test-client", + version: "1.0.0", + }, + }, }), ); - const text = (result.content as Array<{ type: string; text: string }>)[0]?.text; - expect(text).toContain("4"); - expect(result.isError).toBeFalsy(); + const initialize = yield* Effect.promise(() => responses.next()); + expect(initialize).toHaveProperty("result.protocolVersion"); + + yield* Effect.promise(() => + transport.send({ + jsonrpc: "2.0", + method: "notifications/initialized", + }), + ); + yield* Effect.promise(() => + transport.send({ + jsonrpc: "2.0", + id: 3, + method: "tools/list", + params: {}, + }), + ); + + const listed = yield* Effect.promise(() => responses.next()); + expect(listed).toHaveProperty( + "result.tools", + expect.arrayContaining([expect.objectContaining({ name: "execute" })]), + ); + + yield* Effect.promise(() => + transport.send({ + jsonrpc: "2.0", + id: 4, + method: "tools/call", + params: { + name: "execute", + arguments: { code: "return 2+2" }, + }, + }), + ); + + const called = yield* Effect.promise(() => responses.next()); + expect(called).toHaveProperty( + "result.content", + expect.arrayContaining([expect.objectContaining({ text: expect.stringContaining("4") })]), + ); }).pipe(Effect.scoped), { timeout: 30_000 }, );