Merge branch 'main' into fix/transport-onerror-callbacks

Author: felixweinberger

.changeset/fix-stdio-epipe-crash.md

@@ -0,0 +1,5 @@
+---
+'@modelcontextprotocol/server': patch
+---
+
+Handle stdout errors (e.g. EPIPE) in `StdioServerTransport` gracefully instead of crashing. When the client disconnects abruptly, the transport now catches the stdout error, surfaces it via `onerror`, and closes.

packages/server/src/server/stdio.ts

@@ -19,6 +19,7 @@ import { process } from '@modelcontextprotocol/server/_shims';
 export class StdioServerTransport implements Transport {
     private _readBuffer: ReadBuffer = new ReadBuffer();
     private _started = false;
+    private _closed = false;
 
     constructor(
         private _stdin: Readable = process.stdin,
@@ -37,6 +38,12 @@ export class StdioServerTransport implements Transport {
     _onerror = (error: Error) => {
         this.onerror?.(error);
     };
+    _onstdouterror = (error: Error) => {
+        this.onerror?.(error);
+        this.close().catch(() => {
+            // Ignore errors during close — we're already in an error path
+        });
+    };
 
     /**
      * Starts listening for messages on `stdin`.
@@ -51,6 +58,7 @@ export class StdioServerTransport implements Transport {
         this._started = true;
         this._stdin.on('data', this._ondata);
         this._stdin.on('error', this._onerror);
+        this._stdout.on('error', this._onstdouterror);
     }
 
     private processReadBuffer() {
@@ -69,9 +77,15 @@ export class StdioServerTransport implements Transport {
     }
 
     async close(): Promise<void> {
+        if (this._closed) {
+            return;
+        }
+        this._closed = true;
+
         // Remove our event listeners first
         this._stdin.off('data', this._ondata);
         this._stdin.off('error', this._onerror);
+        this._stdout.off('error', this._onstdouterror);
 
         // Check if we were the only data listener
         const remainingDataListeners = this._stdin.listenerCount('data');
@@ -87,12 +101,37 @@ export class StdioServerTransport implements Transport {
     }
 
     send(message: JSONRPCMessage): Promise<void> {
-        return new Promise(resolve => {
+        if (this._closed) {
+            return Promise.reject(new Error('StdioServerTransport is closed'));
+        }
+        return new Promise((resolve, reject) => {
             const json = serializeMessage(message);
+
+            let settled = false;
+            const onError = (error: Error) => {
+                if (settled) return;
+                settled = true;
+                this._stdout.off('error', onError);
+                this._stdout.off('drain', onDrain);
+                reject(error);
+            };
+            const onDrain = () => {
+                if (settled) return;
+                settled = true;
+                this._stdout.off('error', onError);
+                this._stdout.off('drain', onDrain);
+                resolve();
+            };
+
+            this._stdout.once('error', onError);
+
             if (this._stdout.write(json)) {
+                if (settled) return;
+                settled = true;
+                this._stdout.off('error', onError);
                 resolve();
-            } else {
-                this._stdout.once('drain', resolve);
+            } else if (!settled) {
+                this._stdout.once('drain', onDrain);
             }
         });
     }

packages/server/test/server/stdio.test.ts

@@ -102,3 +102,80 @@ test('should read multiple messages', async () => {
     await finished;
     expect(readMessages).toEqual(messages);
 });
+
+test('should close and fire onerror when stdout errors', async () => {
+    const server = new StdioServerTransport(input, output);
+
+    let receivedError: Error | undefined;
+    server.onerror = err => {
+        receivedError = err;
+    };
+    let closeCount = 0;
+    server.onclose = () => {
+        closeCount++;
+    };
+
+    await server.start();
+    output.emit('error', new Error('EPIPE'));
+
+    expect(receivedError?.message).toBe('EPIPE');
+    expect(closeCount).toBe(1);
+});
+
+test('should not fire onclose twice when close() is called after stdout error', async () => {
+    const server = new StdioServerTransport(input, output);
+    server.onerror = () => {};
+
+    let closeCount = 0;
+    server.onclose = () => {
+        closeCount++;
+    };
+
+    await server.start();
+    output.emit('error', new Error('EPIPE'));
+    await server.close();
+
+    expect(closeCount).toBe(1);
+});
+
+test('should reject send() when stdout errors before drain', async () => {
+    let completeWrite: ((error?: Error | null) => void) | undefined;
+    const slowOutput = new Writable({
+        highWaterMark: 0,
+        write(_chunk, _encoding, callback) {
+            completeWrite = callback;
+        }
+    });
+
+    const server = new StdioServerTransport(input, slowOutput);
+    server.onerror = () => {};
+    await server.start();
+
+    const sendPromise = server.send({ jsonrpc: '2.0', id: 1, method: 'ping' });
+    completeWrite!(new Error('write EPIPE'));
+
+    await expect(sendPromise).rejects.toThrow('write EPIPE');
+    expect(slowOutput.listenerCount('drain')).toBe(0);
+    expect(slowOutput.listenerCount('error')).toBe(0);
+});
+
+test('should reject send() after transport is closed', async () => {
+    const server = new StdioServerTransport(input, output);
+    await server.start();
+    await server.close();
+
+    await expect(server.send({ jsonrpc: '2.0', id: 1, method: 'ping' })).rejects.toThrow('closed');
+});
+
+test('should fire onerror before onclose on stdout error', async () => {
+    const server = new StdioServerTransport(input, output);
+
+    const events: string[] = [];
+    server.onerror = () => events.push('error');
+    server.onclose = () => events.push('close');
+
+    await server.start();
+    output.emit('error', new Error('EPIPE'));
+
+    expect(events).toEqual(['error', 'close']);
+});