fix(server): call onerror callback for all transport errors (#1433)

Co-authored-by: Felix Weinberger <fweinberger@anthropic.com>
Co-authored-by: Felix Weinberger <3823880+felixweinberger@users.noreply.github.com>

packages/server/src/server/streamableHttp.ts

@@ -404,6 +404,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
         // The client MUST include an Accept header, listing text/event-stream as a supported content type.
         const acceptHeader = req.headers.get('accept');
         if (!acceptHeader?.includes('text/event-stream')) {
+            this.onerror?.(new Error('Not Acceptable: Client must accept text/event-stream'));
             return this.createJsonErrorResponse(406, -32_000, 'Not Acceptable: Client must accept text/event-stream');
         }
 
@@ -430,6 +431,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
         // Check if there's already an active standalone SSE stream for this session
         if (this._streamMapping.get(this._standaloneSseStreamId) !== undefined) {
             // Only one GET SSE stream is allowed per session
+            this.onerror?.(new Error('Conflict: Only one SSE stream is allowed per session'));
             return this.createJsonErrorResponse(409, -32_000, 'Conflict: Only one SSE stream is allowed per session');
         }
 
@@ -481,6 +483,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
      */
     private async replayEvents(lastEventId: string): Promise<Response> {
         if (!this._eventStore) {
+            this.onerror?.(new Error('Event store not configured'));
             return this.createJsonErrorResponse(400, -32_000, 'Event store not configured');
         }
 
@@ -491,11 +494,13 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                 streamId = await this._eventStore.getStreamIdForEventId(lastEventId);
 
                 if (!streamId) {
+                    this.onerror?.(new Error('Invalid event ID format'));
                     return this.createJsonErrorResponse(400, -32_000, 'Invalid event ID format');
                 }
 
                 // Check conflict with the SAME streamId we'll use for mapping
                 if (this._streamMapping.get(streamId) !== undefined) {
+                    this.onerror?.(new Error('Conflict: Stream already has an active connection'));
                     return this.createJsonErrorResponse(409, -32_000, 'Conflict: Stream already has an active connection');
                 }
             }
@@ -529,7 +534,6 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                 send: async (eventId: string, message: JSONRPCMessage) => {
                     const success = this.writeSSEEvent(streamController!, encoder, message, eventId);
                     if (!success) {
-                        this.onerror?.(new Error('Failed replay events'));
                         try {
                             streamController!.close();
                         } catch {
@@ -577,7 +581,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
             eventData += `data: ${JSON.stringify(message)}\n\n`;
             controller.enqueue(encoder.encode(eventData));
             return true;
-        } catch {
+        } catch (error) {
+            this.onerror?.(error as Error);
             return false;
         }
     }
@@ -586,6 +591,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
      * Handles unsupported requests (`PUT`, `PATCH`, etc.)
      */
     private handleUnsupportedRequest(): Response {
+        this.onerror?.(new Error('Method not allowed.'));
         return Response.json(
             {
                 jsonrpc: '2.0',
@@ -614,6 +620,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
             const acceptHeader = req.headers.get('accept');
             // The client MUST include an Accept header, listing both application/json and text/event-stream as supported content types.
             if (!acceptHeader?.includes('application/json') || !acceptHeader.includes('text/event-stream')) {
+                this.onerror?.(new Error('Not Acceptable: Client must accept both application/json and text/event-stream'));
                 return this.createJsonErrorResponse(
                     406,
                     -32_000,
@@ -623,6 +630,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
 
             const ct = req.headers.get('content-type');
             if (!ct || !ct.includes('application/json')) {
+                this.onerror?.(new Error('Unsupported Media Type: Content-Type must be application/json'));
                 return this.createJsonErrorResponse(415, -32_000, 'Unsupported Media Type: Content-Type must be application/json');
             }
 
@@ -635,7 +643,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
             if (options?.parsedBody === undefined) {
                 try {
                     rawMessage = await req.json();
-                } catch {
+                } catch (error) {
+                    this.onerror?.(error as Error);
                     return this.createJsonErrorResponse(400, -32_700, 'Parse error: Invalid JSON');
                 }
             } else {
@@ -649,7 +658,8 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                 messages = Array.isArray(rawMessage)
                     ? rawMessage.map(msg => JSONRPCMessageSchema.parse(msg))
                     : [JSONRPCMessageSchema.parse(rawMessage)];
-            } catch {
+            } catch (error) {
+                this.onerror?.(error as Error);
                 return this.createJsonErrorResponse(400, -32_700, 'Parse error: Invalid JSON-RPC message');
             }
 
@@ -660,9 +670,11 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                 // If it's a server with session management and the session ID is already set we should reject the request
                 // to avoid re-initialization.
                 if (this._initialized && this.sessionId !== undefined) {
+                    this.onerror?.(new Error('Invalid Request: Server already initialized'));
                     return this.createJsonErrorResponse(400, -32_600, 'Invalid Request: Server already initialized');
                 }
                 if (messages.length > 1) {
+                    this.onerror?.(new Error('Invalid Request: Only one initialization request is allowed'));
                     return this.createJsonErrorResponse(400, -32_600, 'Invalid Request: Only one initialization request is allowed');
                 }
                 this.sessionId = this.sessionIdGenerator?.();
@@ -842,6 +854,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
         }
         if (!this._initialized) {
             // If the server has not been initialized yet, reject all requests
+            this.onerror?.(new Error('Bad Request: Server not initialized'));
             return this.createJsonErrorResponse(400, -32_000, 'Bad Request: Server not initialized');
         }
 
@@ -849,11 +862,13 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
 
         if (!sessionId) {
             // Non-initialization requests without a session ID should return 400 Bad Request
+            this.onerror?.(new Error('Bad Request: Mcp-Session-Id header is required'));
             return this.createJsonErrorResponse(400, -32_000, 'Bad Request: Mcp-Session-Id header is required');
         }
 
         if (sessionId !== this.sessionId) {
             // Reject requests with invalid session ID with 404 Not Found
+            this.onerror?.(new Error('Session not found'));
             return this.createJsonErrorResponse(404, -32_001, 'Session not found');
         }
 
@@ -877,11 +892,9 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
         const protocolVersion = req.headers.get('mcp-protocol-version');
 
         if (protocolVersion !== null && !this._supportedProtocolVersions.includes(protocolVersion)) {
-            return this.createJsonErrorResponse(
-                400,
-                -32_000,
-                `Bad Request: Unsupported protocol version: ${protocolVersion} (supported versions: ${this._supportedProtocolVersions.join(', ')})`
-            );
+            const error = `Bad Request: Unsupported protocol version: ${protocolVersion} (supported versions: ${this._supportedProtocolVersions.join(', ')})`;
+            this.onerror?.(new Error(error));
+            return this.createJsonErrorResponse(400, -32_000, error);
         }
         return undefined;
     }

packages/server/test/server/streamableHttp.test.ts

@@ -765,4 +765,195 @@ describe('Zod v4', () => {
             await expect(transport.start()).rejects.toThrow('Transport already started');
         });
     });
+
+    describe('HTTPServerTransport - onerror callback', () => {
+        let transport: WebStandardStreamableHTTPServerTransport;
+        let mcpServer: McpServer;
+        let errors: Error[];
+
+        beforeEach(async () => {
+            errors = [];
+            mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' }, { capabilities: {} });
+
+            transport = new WebStandardStreamableHTTPServerTransport({
+                sessionIdGenerator: () => randomUUID()
+            });
+
+            transport.onerror = err => errors.push(err);
+
+            await mcpServer.connect(transport);
+        });
+
+        afterEach(async () => {
+            await transport.close();
+        });
+
+        async function initializeServer(): Promise<string> {
+            const request = createRequest('POST', TEST_MESSAGES.initialize);
+            const response = await transport.handleRequest(request);
+            return response.headers.get('mcp-session-id') as string;
+        }
+
+        it('should call onerror for invalid JSON', async () => {
+            const request = new Request('http://localhost/mcp', {
+                method: 'POST',
+                headers: {
+                    Accept: 'application/json, text/event-stream',
+                    'Content-Type': 'application/json'
+                },
+                body: 'not valid json'
+            });
+
+            const response = await transport.handleRequest(request);
+
+            expect(response.status).toBe(400);
+            expect(errors.length).toBeGreaterThan(0);
+            const error = errors[0];
+            expect(error).toBeDefined();
+            expect(error).toBeInstanceOf(SyntaxError);
+        });
+
+        it('should call onerror for invalid JSON-RPC message', async () => {
+            const request = new Request('http://localhost/mcp', {
+                method: 'POST',
+                headers: {
+                    Accept: 'application/json, text/event-stream',
+                    'Content-Type': 'application/json'
+                },
+                body: JSON.stringify({ not: 'valid jsonrpc' })
+            });
+
+            const response = await transport.handleRequest(request);
+
+            expect(response.status).toBe(400);
+            expect(errors.length).toBeGreaterThan(0);
+            const error = errors[0];
+            expect(error).toBeDefined();
+            expect(error?.name).toBe('ZodError');
+        });
+
+        it('should call onerror for missing Accept header on POST', async () => {
+            const request = createRequest('POST', TEST_MESSAGES.initialize, { accept: 'application/json' });
+
+            const response = await transport.handleRequest(request);
+
+            expect(response.status).toBe(406);
+            expect(errors.length).toBeGreaterThan(0);
+            const error = errors[0];
+            expect(error).toBeDefined();
+            expect(error?.message).toContain('Not Acceptable');
+        });
+
+        it('should call onerror for unsupported Content-Type', async () => {
+            const request = new Request('http://localhost/mcp', {
+                method: 'POST',
+                headers: {
+                    Accept: 'application/json, text/event-stream',
+                    'Content-Type': 'text/plain'
+                },
+                body: JSON.stringify(TEST_MESSAGES.initialize)
+            });
+
+            const response = await transport.handleRequest(request);
+
+            expect(response.status).toBe(415);
+            expect(errors.length).toBeGreaterThan(0);
+            const error = errors[0];
+            expect(error).toBeDefined();
+            expect(error?.message).toContain('Unsupported Media Type');
+        });
+
+        it('should call onerror for server not initialized', async () => {
+            const request = createRequest('POST', TEST_MESSAGES.toolsList);
+
+            const response = await transport.handleRequest(request);
+
+            expect(response.status).toBe(400);
+            expect(errors.length).toBeGreaterThan(0);
+            const error = errors[0];
+            expect(error).toBeDefined();
+            expect(error?.message).toContain('Server not initialized');
+        });
+
+        it('should call onerror for invalid session ID', async () => {
+            await initializeServer();
+
+            const request = createRequest('POST', TEST_MESSAGES.toolsList, { sessionId: 'invalid-session-id' });
+
+            const response = await transport.handleRequest(request);
+
+            expect(response.status).toBe(404);
+            expect(errors.length).toBeGreaterThan(0);
+            const error = errors[0];
+            expect(error).toBeDefined();
+            expect(error?.message).toContain('Session not found');
+        });
+
+        it('should call onerror for re-initialization attempt', async () => {
+            await initializeServer();
+
+            const request = createRequest('POST', TEST_MESSAGES.initialize);
+
+            const response = await transport.handleRequest(request);
+
+            expect(response.status).toBe(400);
+            expect(errors.length).toBeGreaterThan(0);
+            const error = errors[0];
+            expect(error).toBeDefined();
+            expect(error?.message).toContain('Server already initialized');
+        });
+
+        it('should call onerror for GET without Accept header', async () => {
+            const sessionId = await initializeServer();
+
+            const request = createRequest('GET', undefined, { sessionId, accept: 'application/json' });
+
+            const response = await transport.handleRequest(request);
+
+            expect(response.status).toBe(406);
+            expect(errors.length).toBeGreaterThan(0);
+            const error = errors[0];
+            expect(error).toBeDefined();
+            expect(error?.message).toContain('Not Acceptable');
+        });
+
+        it('should call onerror for concurrent SSE streams', async () => {
+            const sessionId = await initializeServer();
+
+            const request1 = createRequest('GET', undefined, { sessionId });
+            await transport.handleRequest(request1);
+
+            const request2 = createRequest('GET', undefined, { sessionId });
+            const response2 = await transport.handleRequest(request2);
+
+            expect(response2.status).toBe(409);
+            expect(errors.length).toBeGreaterThan(0);
+            const error = errors[0];
+            expect(error).toBeDefined();
+            expect(error?.message).toContain('Conflict');
+        });
+
+        it('should call onerror for unsupported protocol version', async () => {
+            const sessionId = await initializeServer();
+
+            const request = new Request('http://localhost/mcp', {
+                method: 'POST',
+                headers: {
+                    'Content-Type': 'application/json',
+                    Accept: 'application/json, text/event-stream',
+                    'mcp-session-id': sessionId,
+                    'mcp-protocol-version': 'unsupported-version'
+                },
+                body: JSON.stringify(TEST_MESSAGES.toolsList)
+            });
+
+            const response = await transport.handleRequest(request);
+
+            expect(response.status).toBe(400);
+            expect(errors.length).toBeGreaterThan(0);
+            const error = errors[0];
+            expect(error).toBeDefined();
+            expect(error?.message).toContain('Unsupported protocol version');
+        });
+    });
 });