diff --git a/packages/mcp/CHANGELOG.md b/packages/mcp/CHANGELOG.md index a0e3f2cc8..a971d0b59 100644 --- a/packages/mcp/CHANGELOG.md +++ b/packages/mcp/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Added + +- Added `PipeTransport`, MCP's stdio transport over a program started by the caller through an `McpPipe` (`write`, `closeStdin`, `close`, `exited`), for example `@earendil-works/pi-env`'s `openProcess()` on another machine. `StdioTransport` and `PipeTransport` share the newline-delimited framing and the stderr tail. + ## [1.5.0] - 2026-11-02 ### Changed diff --git a/packages/mcp/README.md b/packages/mcp/README.md index ebb1b2973..bce56542d 100644 --- a/packages/mcp/README.md +++ b/packages/mcp/README.md @@ -2,7 +2,7 @@ A small, standalone Model Context Protocol client. It does not depend on the official MCP SDK or other pi packages. -The package provides a transport-neutral client core, stdio and Streamable HTTP transports, and an in-memory testing transport. +The package provides a transport-neutral client core, stdio and Streamable HTTP transports, a stdio transport over a pipe to a program started elsewhere, and an in-memory testing transport. ## Usage @@ -27,6 +27,22 @@ await client.close(); For a remote server, use `new StreamableHttpTransport({ url, headers })`. Fetch can be injected for proxying or custom networking. +### Stdio servers started elsewhere + +`PipeTransport` speaks MCP's stdio framing over a program that something else starts, for example on another machine. `open` starts the program, connects its stdout and stderr to the given handlers, and returns an `McpPipe` (`write`, `closeStdin`, `close` to kill, and an `exited` promise). `@earendil-works/pi-env`'s `openProcess()` returns one, without an adapter: + +```typescript +import { openProcess } from "@earendil-works/pi-env"; +import { PipeTransport } from "@earendil-works/pi-mcp"; + +const transport = new PipeTransport({ + open: (handlers) => openProcess(connection, { cwd, argv, ...handlers }), +}); +await client.connect(transport); +``` + +Closing the transport closes the program's stdin and kills it if it has not exited after `closeTimeoutMs` (default 2000). The transport closes when the program ends. + ### Tools for an LLM `toLlmContent(result)` converts a `CallToolResult` to text and image content for a model, in the shape of `@earendil-works/pi-ai`'s `TextContent` and `ImageContent`. Text and images pass through, embedded text and image resources are unwrapped, and audio, resource links, and binary resources become short text placeholders. A result without content blocks but with `structuredContent` becomes its JSON. @@ -123,6 +139,7 @@ An MCP transport owns framing and I/O. It delivers individual JSON-RPC messages - request cancellation - Streamable HTTP sessions, the server-to-client GET stream with reconnection, and resumption of dropped response streams with `Last-Event-ID` - stdio shutdown per the spec (close stdin, then SIGTERM, then SIGKILL), applied to the server's whole process group +- stdio over a caller-provided pipe (`PipeTransport`) - server `ping` and `roots/list` requests - logging and tool-list-change notifications through the generic notification API - OAuth protected-resource and authorization-server discovery diff --git a/packages/mcp/src/index.ts b/packages/mcp/src/index.ts index 3d0f8fc03..e11132e8a 100644 --- a/packages/mcp/src/index.ts +++ b/packages/mcp/src/index.ts @@ -55,6 +55,12 @@ export { type ToolAnnotations, type ToolExecution, } from "./protocol/types.ts"; +export { + type McpPipe, + type McpPipeHandlers, + PipeTransport, + type PipeTransportOptions, +} from "./transports/pipe.ts"; export { StdioTransport, type StdioTransportOptions } from "./transports/stdio.ts"; export { McpAuthRequiredError, diff --git a/packages/mcp/src/transports/lines.ts b/packages/mcp/src/transports/lines.ts new file mode 100644 index 000000000..8d53d9c27 --- /dev/null +++ b/packages/mcp/src/transports/lines.ts @@ -0,0 +1,77 @@ +import { type JsonRpcMessage, parseJsonRpcMessage } from "../protocol/jsonrpc.ts"; +import { DEFAULT_MAX_MESSAGE_BYTES } from "./transport.ts"; + +const DEFAULT_MAX_STDERR_BYTES = 64 * 1024; + +export interface JsonRpcLineReaderOptions { + maxMessageBytes?: number; + maxStderrBytes?: number; + /** Every stderr chunk, decoded as UTF-8. */ + onStderr?: (chunk: string) => void; + onMessage(message: JsonRpcMessage): void; + onError(error: unknown): void; +} + +/** + * Reads a program's output as MCP's stdio framing: newline-delimited JSON-RPC messages on stdout, and a bounded + * tail of stderr for diagnostics. + */ +export class JsonRpcLineReader { + private readonly options: JsonRpcLineReaderOptions; + private stdoutBuffer = Buffer.alloc(0); + private stderrBuffer = Buffer.alloc(0); + + constructor(options: JsonRpcLineReaderOptions) { + this.options = options; + } + + /** The last `maxStderrBytes` of stderr. */ + get stderr(): string { + return this.stderrBuffer.toString("utf8"); + } + + /** Takes the next stdout bytes and reports every complete message, or the error parsing it. */ + readStdout(chunk: Uint8Array): void { + this.stdoutBuffer = Buffer.concat([this.stdoutBuffer, chunk]); + const maxMessageBytes = this.options.maxMessageBytes ?? DEFAULT_MAX_MESSAGE_BYTES; + while (true) { + const newline = this.stdoutBuffer.indexOf(0x0a); + if (newline < 0) { + if (this.stdoutBuffer.length > maxMessageBytes) { + this.stdoutBuffer = Buffer.alloc(0); + this.options.onError(new Error(`MCP stdio message exceeds ${maxMessageBytes} bytes`)); + } + return; + } + const line = this.stdoutBuffer.subarray(0, newline); + this.stdoutBuffer = this.stdoutBuffer.subarray(newline + 1); + if (line.length > maxMessageBytes) { + this.options.onError(new Error(`MCP stdio message exceeds ${maxMessageBytes} bytes`)); + continue; + } + const text = line.toString("utf8").replace(/\r$/, ""); + if (!text.trim()) continue; + try { + this.options.onMessage(parseJsonRpcMessage(JSON.parse(text))); + } catch (error) { + this.options.onError(error); + } + } + } + + /** Takes the next stderr bytes into the tail and passes them to `onStderr`. */ + readStderr(chunk: Uint8Array): void { + const maxStderrBytes = this.options.maxStderrBytes ?? DEFAULT_MAX_STDERR_BYTES; + this.stderrBuffer = Buffer.concat([this.stderrBuffer, chunk]); + if (this.stderrBuffer.length > maxStderrBytes) this.stderrBuffer = this.stderrBuffer.subarray(-maxStderrBytes); + this.options.onStderr?.(Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength).toString("utf8")); + } + + /** Ends stdout: a message without its newline is reported as an error and dropped. */ + end(): void { + if (this.stdoutBuffer.toString("utf8").trim()) { + this.options.onError(new Error("MCP stdio server closed with an incomplete JSON-RPC message")); + } + this.stdoutBuffer = Buffer.alloc(0); + } +} diff --git a/packages/mcp/src/transports/pipe.ts b/packages/mcp/src/transports/pipe.ts new file mode 100644 index 000000000..005fa475a --- /dev/null +++ b/packages/mcp/src/transports/pipe.ts @@ -0,0 +1,129 @@ +import { type JsonRpcMessage, McpConnectionClosedError } from "../protocol/jsonrpc.ts"; +import { JsonRpcLineReader } from "./lines.ts"; +import { type McpTransport, TransportEvents } from "./transport.ts"; + +const DEFAULT_CLOSE_TIMEOUT_MS = 2_000; + +/** A running program's stdio, as `@earendil-works/pi-env`'s `RemoteProcess` provides it. */ +export interface McpPipe { + write(data: Uint8Array | string): void; + closeStdin(): void; + /** Kill the program (and what it started). The pipe owns cleaning up what the program started. */ + close(): void; + /** Settles when the program has ended, whatever the reason; it must also settle after `close()`. */ + readonly exited: Promise; +} + +/** Handlers `PipeTransportOptions.open` connects to the program's output. */ +export interface McpPipeHandlers { + onStdout(bytes: Uint8Array): void; + onStderr(bytes: Uint8Array): void; +} + +export interface PipeTransportOptions { + /** Start the program; `onStdout`/`onStderr` receive its output from then on. */ + open(handlers: McpPipeHandlers): Promise; + maxMessageBytes?: number; + maxStderrBytes?: number; + onStderr?: (chunk: string) => void; + /** After closing stdin, how long the program may take to exit before `close()` kills it. Default 2000. */ + closeTimeoutMs?: number; +} + +/** + * MCP's stdio transport over a program that something else runs, for example on another machine through pi-env. + * The transport closes when the program ends; a rejected `exited` that `close()` did not cause is reported as an + * error first. + */ +export class PipeTransport extends TransportEvents implements McpTransport { + readonly options: Readonly; + private readonly lines: JsonRpcLineReader; + private pipe: McpPipe | undefined; + private started = false; + private closed = false; + private ended = false; + + constructor(options: PipeTransportOptions) { + super(); + this.options = Object.freeze({ ...options }); + this.lines = new JsonRpcLineReader({ + maxMessageBytes: this.options.maxMessageBytes, + maxStderrBytes: this.options.maxStderrBytes, + onStderr: this.options.onStderr, + onMessage: (message) => this.emitMessage(message), + onError: (error) => this.emitError(error), + }); + } + + /** The last `maxStderrBytes` (default 64 KiB) of the program's stderr. */ + get stderr(): string { + return this.lines.stderr; + } + + async start(): Promise { + if (this.started) throw new Error("MCP pipe transport already started"); + if (this.closed) throw new McpConnectionClosedError(); + this.started = true; + const pipe = await this.options.open({ + onStdout: (bytes) => this.lines.readStdout(bytes), + onStderr: (bytes) => this.lines.readStderr(bytes), + }); + if (this.closed) { + // close() ran while the program was starting and found nothing to shut down. Nobody else awaits + // `exited`, and a pipe may reject it after a kill. + pipe.exited.catch(() => {}); + pipe.close(); + throw new McpConnectionClosedError(); + } + this.pipe = pipe; + pipe.exited.then( + () => this.exit(), + (error: unknown) => this.fail(error), + ); + } + + async send(message: JsonRpcMessage): Promise { + if (!this.pipe || this.closed || this.ended) throw new McpConnectionClosedError(); + this.pipe.write(`${JSON.stringify(message)}\n`); + } + + async close(): Promise { + if (this.closed) return; + this.closed = true; + const pipe = this.pipe; + if (!pipe) { + this.emitClose(); + return; + } + if (this.ended) return; + // Shutdown per the spec: close stdin and let the program exit, then kill it. + pipe.closeStdin(); + let timer: ReturnType | undefined; + const exitedInTime = await Promise.race([ + pipe.exited.then( + () => true, + () => true, + ), + new Promise((resolve) => { + timer = setTimeout(() => resolve(false), this.options.closeTimeoutMs ?? DEFAULT_CLOSE_TIMEOUT_MS); + }), + ]); + clearTimeout(timer); + if (exitedInTime) return; + pipe.close(); + await pipe.exited.catch(() => {}); + } + + /** The program ended: report an unfinished message, then close. */ + private exit(): void { + this.ended = true; + this.lines.end(); + this.emitClose(); + } + + /** `exited` rejected: an error unless `close()` caused it, then the same as an exit. */ + private fail(error: unknown): void { + if (!this.closed) this.emitError(error); + this.exit(); + } +} diff --git a/packages/mcp/src/transports/stdio.ts b/packages/mcp/src/transports/stdio.ts index 132c6765e..f26c118bb 100644 --- a/packages/mcp/src/transports/stdio.ts +++ b/packages/mcp/src/transports/stdio.ts @@ -1,10 +1,10 @@ import { type ChildProcess, spawn } from "node:child_process"; import process from "node:process"; import crossSpawn from "cross-spawn"; -import { type JsonRpcMessage, McpConnectionClosedError, parseJsonRpcMessage } from "../protocol/jsonrpc.ts"; -import { DEFAULT_MAX_MESSAGE_BYTES, type McpTransport, TransportEvents } from "./transport.ts"; +import { type JsonRpcMessage, McpConnectionClosedError } from "../protocol/jsonrpc.ts"; +import { JsonRpcLineReader } from "./lines.ts"; +import { type McpTransport, TransportEvents } from "./transport.ts"; -const DEFAULT_MAX_STDERR_BYTES = 64 * 1024; const DEFAULT_CLOSE_TIMEOUT_MS = 2_000; /** How long a server gets to exit on its own after stdin closes, before it is sent SIGTERM. */ const STDIN_CLOSE_GRACE_MS = 500; @@ -69,14 +69,20 @@ export interface StdioTransportOptions { export class StdioTransport extends TransportEvents implements McpTransport { readonly options: Readonly; private child: ChildProcess | undefined; - private stdoutBuffer = Buffer.alloc(0); - private stderrBuffer = Buffer.alloc(0); + private readonly lines: JsonRpcLineReader; private started = false; private closed = false; constructor(options: StdioTransportOptions) { super(); this.options = Object.freeze({ ...options, args: options.args ? [...options.args] : undefined }); + this.lines = new JsonRpcLineReader({ + maxMessageBytes: this.options.maxMessageBytes, + maxStderrBytes: this.options.maxStderrBytes, + onStderr: this.options.onStderr, + onMessage: (message) => this.emitMessage(message), + onError: (error) => this.emitError(error), + }); } get pid(): number | undefined { @@ -84,7 +90,7 @@ export class StdioTransport extends TransportEvents implements McpTransport { } get stderr(): string { - return this.stderrBuffer.toString("utf8"); + return this.lines.stderr; } async start(): Promise { @@ -108,19 +114,16 @@ export class StdioTransport extends TransportEvents implements McpTransport { // Not on "exit": children of the server can outlive it while still holding its pipes. child.once("close", () => liveProcessGroups.delete(pid)); } - child.stdout?.on("data", (chunk: Buffer | string) => this.handleStdout(chunk)); + child.stdout?.on("data", (chunk: Buffer) => this.lines.readStdout(chunk)); child.stdout?.on("error", (error) => this.emitError(error)); child.stdin?.on("error", (error) => { if (!this.closed) this.emitError(error); }); - child.stderr?.on("data", (chunk: Buffer | string) => this.handleStderr(chunk)); + child.stderr?.on("data", (chunk: Buffer) => this.lines.readStderr(chunk)); child.stderr?.on("error", (error) => this.emitError(error)); child.on("close", () => { this.child = undefined; - if (this.stdoutBuffer.toString("utf8").trim()) { - this.emitError(new Error("MCP stdio server closed with an incomplete JSON-RPC message")); - } - this.stdoutBuffer = Buffer.alloc(0); + this.lines.end(); this.emitClose(); }); @@ -180,40 +183,4 @@ export class StdioTransport extends TransportEvents implements McpTransport { child.stdin?.end(); }); } - - private handleStdout(chunk: Buffer | string): void { - this.stdoutBuffer = Buffer.concat([this.stdoutBuffer, Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)]); - const maxMessageBytes = this.options.maxMessageBytes ?? DEFAULT_MAX_MESSAGE_BYTES; - while (true) { - const newline = this.stdoutBuffer.indexOf(0x0a); - if (newline < 0) { - if (this.stdoutBuffer.length > maxMessageBytes) { - this.stdoutBuffer = Buffer.alloc(0); - this.emitError(new Error(`MCP stdio message exceeds ${maxMessageBytes} bytes`)); - } - return; - } - const line = this.stdoutBuffer.subarray(0, newline); - this.stdoutBuffer = this.stdoutBuffer.subarray(newline + 1); - if (line.length > maxMessageBytes) { - this.emitError(new Error(`MCP stdio message exceeds ${maxMessageBytes} bytes`)); - continue; - } - const text = line.toString("utf8").replace(/\r$/, ""); - if (!text.trim()) continue; - try { - this.emitMessage(parseJsonRpcMessage(JSON.parse(text))); - } catch (error) { - this.emitError(error); - } - } - } - - private handleStderr(chunk: Buffer | string): void { - const buffer = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); - const maxStderrBytes = this.options.maxStderrBytes ?? DEFAULT_MAX_STDERR_BYTES; - this.stderrBuffer = Buffer.concat([this.stderrBuffer, buffer]); - if (this.stderrBuffer.length > maxStderrBytes) this.stderrBuffer = this.stderrBuffer.subarray(-maxStderrBytes); - this.options.onStderr?.(buffer.toString("utf8")); - } } diff --git a/packages/mcp/test/pipe.test.ts b/packages/mcp/test/pipe.test.ts new file mode 100644 index 000000000..95e0b4417 --- /dev/null +++ b/packages/mcp/test/pipe.test.ts @@ -0,0 +1,265 @@ +import { spawn } from "node:child_process"; +import { fileURLToPath } from "node:url"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + type JsonRpcMessage, + McpClient, + McpConnectionClosedError, + type McpPipe, + type McpPipeHandlers, + PipeTransport, + type PipeTransportOptions, +} from "../src/index.ts"; + +const fixture = fileURLToPath(new URL("./fixtures/stdio-server.mjs", import.meta.url)); + +interface FakeProgram { + transport: PipeTransport; + /** Calls of `closeStdin` and `close`, in order. */ + calls: string[]; + written: string[]; + messages: JsonRpcMessage[]; + errors: Error[]; + closed: () => boolean; + stdout(text: string): void; + stderr(text: string): void; + exit(code?: number): void; + fail(error: Error): void; +} + +/** A `PipeTransport` over an in-memory program that exits only when told to, or when killed. */ +async function startFake(options: Omit = {}): Promise { + const exited = Promise.withResolvers(); + const calls: string[] = []; + const written: string[] = []; + let handlers: McpPipeHandlers | undefined; + const pipe: McpPipe = { + write: (data) => written.push(String(data)), + closeStdin: () => calls.push("closeStdin"), + close: () => { + calls.push("close"); + exited.reject(new Error("aborted")); + }, + exited: exited.promise, + }; + const transport = new PipeTransport({ + ...options, + open: async (openHandlers) => { + handlers = openHandlers; + return pipe; + }, + }); + const messages: JsonRpcMessage[] = []; + const errors: Error[] = []; + let closed = false; + transport.onMessage((message) => messages.push(message)); + transport.onError((error) => errors.push(error)); + transport.onClose(() => { + closed = true; + }); + await transport.start(); + const encoder = new TextEncoder(); + return { + transport, + calls, + written, + messages, + errors, + closed: () => closed, + stdout: (text) => handlers?.onStdout(encoder.encode(text)), + stderr: (text) => handlers?.onStderr(encoder.encode(text)), + exit: (code = 0) => exited.resolve(code), + fail: (error) => exited.reject(error), + }; +} + +/** Let the transport's reactions to `exited` run. */ +const settle = () => new Promise((resolve) => setTimeout(resolve, 0)); + +const ping = (id: number) => ({ jsonrpc: "2.0", id, method: "ping" }) as const; + +afterEach(() => { + vi.useRealTimers(); +}); + +describe("PipeTransport framing", () => { + it("joins messages split across chunks and splits several messages in one chunk", async () => { + const program = await startFake(); + const [first, second, third] = [ping(1), ping(2), ping(3)].map((message) => JSON.stringify(message)); + program.stdout(first.slice(0, 10)); + program.stdout(`${first.slice(10)}\n${second}`); + program.stdout(`\n${third}\n`); + expect(program.messages).toEqual([ping(1), ping(2), ping(3)]); + expect(program.errors).toEqual([]); + }); + + it("strips a trailing carriage return and skips blank lines", async () => { + const program = await startFake(); + program.stdout(`\n\r\n \n${JSON.stringify(ping(1))}\r\n\n`); + expect(program.messages).toEqual([ping(1)]); + expect(program.errors).toEqual([]); + }); + + it("reports invalid messages and keeps reading", async () => { + const program = await startFake(); + program.stdout(`not json\n{"jsonrpc":"1.0"}\n${JSON.stringify(ping(1))}\n`); + expect(program.messages).toEqual([ping(1)]); + expect(program.errors).toHaveLength(2); + }); + + it("drops messages over maxMessageBytes, complete or not", async () => { + const program = await startFake({ maxMessageBytes: 48 }); + program.stdout(`${JSON.stringify({ jsonrpc: "2.0", method: "x".repeat(48) })}\n${JSON.stringify(ping(1))}\n`); + program.stdout("y".repeat(49)); + program.stdout(`\n${JSON.stringify(ping(2))}\n`); + expect(program.messages).toEqual([ping(1), ping(2)]); + expect(program.errors.map((error) => error.message)).toEqual([ + "MCP stdio message exceeds 48 bytes", + "MCP stdio message exceeds 48 bytes", + ]); + }); + + it("keeps a bounded stderr tail and passes every chunk to onStderr", async () => { + const chunks: string[] = []; + const program = await startFake({ maxStderrBytes: 8, onStderr: (chunk) => chunks.push(chunk) }); + program.stderr("starting\n"); + program.stderr("ready\n"); + expect(chunks).toEqual(["starting\n", "ready\n"]); + expect(program.transport.stderr).toBe("g\nready\n"); + }); +}); + +describe("PipeTransport lifecycle", () => { + it("writes each message as one line", async () => { + const program = await startFake(); + await program.transport.send(ping(1)); + expect(program.written).toEqual([`${JSON.stringify(ping(1))}\n`]); + }); + + it("reports an incomplete message when the program exits, then closes", async () => { + const program = await startFake(); + program.stdout('{"jsonrpc":"2.0"'); + program.exit(1); + await settle(); + expect(program.errors.map((error) => error.message)).toEqual([ + "MCP stdio server closed with an incomplete JSON-RPC message", + ]); + expect(program.closed()).toBe(true); + }); + + it("reports a failed exit it did not cause, then closes", async () => { + const program = await startFake(); + program.fail(new Error("connection lost")); + await settle(); + expect(program.errors.map((error) => error.message)).toEqual(["connection lost"]); + expect(program.closed()).toBe(true); + }); + + it("rejects send after close and after the program ended", async () => { + const closedByUs = await startFake(); + closedByUs.exit(); + await closedByUs.transport.close(); + await expect(closedByUs.transport.send(ping(1))).rejects.toBeInstanceOf(McpConnectionClosedError); + + const ended = await startFake(); + ended.exit(); + await settle(); + await expect(ended.transport.send(ping(1))).rejects.toBeInstanceOf(McpConnectionClosedError); + expect(ended.written).toEqual([]); + }); + + it("closes stdin and does not kill a program that exits in time", async () => { + const program = await startFake(); + const closing = program.transport.close(); + expect(program.calls).toEqual(["closeStdin"]); + await expect(program.transport.send(ping(1))).rejects.toBeInstanceOf(McpConnectionClosedError); + expect(program.written).toEqual([]); + program.exit(); + await closing; + expect(program.calls).toEqual(["closeStdin"]); + expect(program.closed()).toBe(true); + expect(program.errors).toEqual([]); + }); + + it("shuts down once when closed twice", async () => { + const program = await startFake(); + const first = program.transport.close(); + const second = program.transport.close(); + program.exit(); + await Promise.all([first, second]); + await program.transport.close(); + expect(program.calls).toEqual(["closeStdin"]); + expect(program.closed()).toBe(true); + }); + + it("kills a program only after closeTimeoutMs", async () => { + vi.useFakeTimers(); + const program = await startFake({ closeTimeoutMs: 1_000 }); + const closing = program.transport.close(); + await vi.advanceTimersByTimeAsync(999); + expect(program.calls).toEqual(["closeStdin"]); + await vi.advanceTimersByTimeAsync(1); + await closing; + expect(program.calls).toEqual(["closeStdin", "close"]); + expect(program.closed()).toBe(true); + // The rejection of `exited` that killing caused is not an error. + expect(program.errors).toEqual([]); + }); + + it("rejects start when open rejects", async () => { + const transport = new PipeTransport({ open: () => Promise.reject(new Error("not_supported")) }); + await expect(transport.start()).rejects.toThrow("not_supported"); + }); + + it("kills a program that finished starting after close", async () => { + const opened = Promise.withResolvers(); + const transport = new PipeTransport({ open: () => opened.promise }); + const starting = transport.start(); + await transport.close(); + const exited = Promise.withResolvers(); + // Like pi-env's RemoteProcess: killing rejects `exited`, which must not become an unhandled rejection. + const close = vi.fn(() => exited.reject(new Error("aborted"))); + opened.resolve({ write: () => {}, closeStdin: () => {}, close, exited: exited.promise }); + await expect(starting).rejects.toBeInstanceOf(McpConnectionClosedError); + expect(close).toHaveBeenCalledOnce(); + await settle(); + }); +}); + +/** Runs a program with `node:child_process`, standing in for pi-env's `openProcess`. */ +function spawnPipe(argv: readonly string[], handlers: McpPipeHandlers): Promise { + const [command, ...args] = argv; + const child = spawn(command, args, { stdio: "pipe" }); + child.stdout.on("data", handlers.onStdout); + child.stderr.on("data", handlers.onStderr); + // Writes racing the program's exit fail with EPIPE; `exited` reports the end. + child.stdin.on("error", () => {}); + const exited = new Promise((resolve) => child.once("close", resolve)); + return new Promise((resolve, reject) => { + child.once("error", reject); + child.once("spawn", () => + resolve({ + write: (data) => child.stdin.write(data), + closeStdin: () => child.stdin.end(), + close: () => child.kill("SIGKILL"), + exited, + }), + ); + }); +} + +describe("PipeTransport with a real program", () => { + it("runs initialize, tools/list and tools/call", async () => { + const transport = new PipeTransport({ open: (handlers) => spawnPipe([process.execPath, fixture], handlers) }); + const client = new McpClient({ name: "pipe-test", version: "1.0.0" }); + const result = await client.connect(transport); + expect(result.serverInfo).toEqual({ name: "stdio-fixture", version: "1.0.0" }); + expect(await client.listTools()).toEqual([{ name: "echo", inputSchema: { type: "object" } }]); + expect(await client.callTool("echo", { text: "hello" })).toEqual({ + content: [{ type: "text", text: "hello" }], + }); + expect(transport.stderr).toContain("stdio fixture ready"); + await client.close(); + expect(client.connectionState).toBe("closed"); + }); +});