Handle invalid stdio JSON-RPC requests without leaving clients waiting

.changeset/smart-zebras-carry.md

@@ -0,0 +1,6 @@
+---
+'@modelcontextprotocol/core': patch
+'@modelcontextprotocol/server': patch
+---
+
+Treat schema-invalid stdio JSON-RPC payloads as invalid requests, exposing the dedicated core error type and letting stdio servers reply with a JSON-RPC `Invalid Request` error before continuing.

packages/core/src/shared/stdio.ts

@@ -1,6 +1,17 @@
 import type { JSONRPCMessage } from '../types/index.js';
 import { JSONRPCMessageSchema } from '../types/index.js';
 
+export class InvalidJSONRPCMessageError extends Error {
+    constructor(
+        message: string,
+        readonly rawMessage: unknown,
+        options?: ErrorOptions
+    ) {
+        super(message, options);
+        this.name = 'InvalidJSONRPCMessageError';
+    }
+}
+
 /**
  * Buffers a continuous stdio stream into discrete JSON-RPC messages.
  */
@@ -42,7 +53,16 @@ export class ReadBuffer {
 }
 
 export function deserializeMessage(line: string): JSONRPCMessage {
-    return JSONRPCMessageSchema.parse(JSON.parse(line));
+    const rawMessage = JSON.parse(line);
+    const parseResult = JSONRPCMessageSchema.safeParse(rawMessage);
+
+    if (parseResult.success) {
+        return parseResult.data;
+    }
+
+    throw new InvalidJSONRPCMessageError('Invalid JSON-RPC message', rawMessage, {
+        cause: parseResult.error
+    });
 }
 
 export function serializeMessage(message: JSONRPCMessage): string {

packages/core/test/shared/stdio.test.ts

@@ -1,4 +1,4 @@
-import { ReadBuffer } from '../../src/shared/stdio.js';
+import { InvalidJSONRPCMessageError, ReadBuffer } from '../../src/shared/stdio.js';
 import type { JSONRPCMessage } from '../../src/types/index.js';
 
 const testMessage: JSONRPCMessage = {
@@ -110,6 +110,13 @@ describe('non-JSON line filtering', () => {
         const readBuffer = new ReadBuffer();
         readBuffer.append(Buffer.from('{"not": "a jsonrpc message"}\n'));
 
-        expect(() => readBuffer.readMessage()).toThrow();
+        expect(() => readBuffer.readMessage()).toThrow(InvalidJSONRPCMessageError);
+    });
+
+    test('should reject request ids larger than Number.MAX_SAFE_INTEGER', () => {
+        const readBuffer = new ReadBuffer();
+        readBuffer.append(Buffer.from('{"jsonrpc":"2.0","id":9007199254740992,"method":"ping"}\n'));
+
+        expect(() => readBuffer.readMessage()).toThrow(InvalidJSONRPCMessageError);
     });
 });

packages/server/src/server/stdio.ts

@@ -1,7 +1,7 @@
 import type { Readable, Writable } from 'node:stream';
 
-import type { JSONRPCMessage, Transport } from '@modelcontextprotocol/core';
-import { ReadBuffer, serializeMessage } from '@modelcontextprotocol/core';
+import type { JSONRPCErrorResponse, JSONRPCMessage, Transport } from '@modelcontextprotocol/core';
+import { InvalidJSONRPCMessageError, ProtocolErrorCode, ReadBuffer, serializeMessage } from '@modelcontextprotocol/core';
 import { process } from '@modelcontextprotocol/server/_shims';
 
 /**
@@ -72,6 +72,19 @@ export class StdioServerTransport implements Transport {
                 this.onmessage?.(message);
             } catch (error) {
                 this.onerror?.(error as Error);
+                if (error instanceof InvalidJSONRPCMessageError) {
+                    const invalidRequestResponse: JSONRPCErrorResponse = {
+                        jsonrpc: '2.0',
+                        error: {
+                            code: ProtocolErrorCode.InvalidRequest,
+                            message: 'Invalid Request'
+                        }
+                    };
+
+                    this.send(invalidRequestResponse).catch(sendError => {
+                        this.onerror?.(sendError);
+                    });
+                }
             }
         }
     }

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

@@ -1,7 +1,7 @@
 import { Readable, Writable } from 'node:stream';
 
 import type { JSONRPCMessage } from '@modelcontextprotocol/core';
-import { ReadBuffer, serializeMessage } from '@modelcontextprotocol/core';
+import { InvalidJSONRPCMessageError, ProtocolErrorCode, ReadBuffer, serializeMessage } from '@modelcontextprotocol/core';
 
 import { StdioServerTransport } from '../../src/server/stdio.js';
 
@@ -95,8 +95,7 @@ test('should read multiple messages', async () => {
         };
     });
 
-    input.push(serializeMessage(messages[0]!));
-    input.push(serializeMessage(messages[1]!));
+    input.push(Buffer.from(serializeMessage(messages[0]!) + serializeMessage(messages[1]!)));
 
     await server.start();
     await finished;
@@ -179,3 +178,41 @@ test('should fire onerror before onclose on stdout error', async () => {
 
     expect(events).toEqual(['error', 'close']);
 });
+
+test('should send an invalid request error for schema-invalid JSON-RPC and continue reading', async () => {
+    const server = new StdioServerTransport(input, output);
+    const receivedErrors: Error[] = [];
+    const validMessage: JSONRPCMessage = {
+        jsonrpc: '2.0',
+        id: 1,
+        method: 'ping'
+    };
+
+    server.onerror = error => {
+        receivedErrors.push(error);
+    };
+
+    const validMessageReceived = new Promise<void>(resolve => {
+        server.onmessage = message => {
+            if (JSON.stringify(message) === JSON.stringify(validMessage)) {
+                resolve();
+            }
+        };
+    });
+
+    await server.start();
+    input.push(Buffer.from('{"jsonrpc":"2.0","id":9007199254740992,"method":"ping"}\n' + serializeMessage(validMessage)));
+
+    await validMessageReceived;
+
+    expect(receivedErrors).toHaveLength(1);
+    expect(receivedErrors[0]).toBeInstanceOf(InvalidJSONRPCMessageError);
+    expect(outputBuffer.readMessage()).toEqual({
+        jsonrpc: '2.0',
+        error: {
+            code: ProtocolErrorCode.InvalidRequest,
+            message: 'Invalid Request'
+        }
+    });
+    expect(outputBuffer.readMessage()).toBeNull();
+});