fix(server): settle the POST for a cancelled request

.changeset/cancelled-post-settles.md

@@ -0,0 +1,5 @@
+---
+'@modelcontextprotocol/server': patch
+---
+
+Settle a cancelled request's POST and release its stream mapping, instead of leaving the response pending forever.

packages/server/src/server/streamableHttp.ts

@@ -7,12 +7,21 @@
  * For Node.js Express/HTTP compatibility, use {@linkcode @modelcontextprotocol/node!NodeStreamableHTTPServerTransport | NodeStreamableHTTPServerTransport} which wraps this transport.
  */
 
-import type { AuthInfo, JSONRPCMessage, JSONRPCRequest, MessageExtraInfo, RequestId, Transport } from '@modelcontextprotocol/core-internal';
+import type {
+    AuthInfo,
+    CancelledNotificationParams,
+    JSONRPCMessage,
+    JSONRPCRequest,
+    MessageExtraInfo,
+    RequestId,
+    Transport
+} from '@modelcontextprotocol/core-internal';
 import {
     DEFAULT_NEGOTIATED_PROTOCOL_VERSION,
     isInitializeRequest,
     isJsonContentType,
     isJSONRPCErrorResponse,
+    isJSONRPCNotification,
     isJSONRPCRequest,
     isJSONRPCResultResponse,
     JSONRPCMessageSchema,
@@ -910,7 +919,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
             if (!hasRequests) {
                 // if it only contains notifications or responses, return 202
                 for (const message of messages) {
-                    this.onmessage?.(message, { authInfo: options?.authInfo, request });
+                    this._deliverMessage(message, { authInfo: options?.authInfo, request });
                 }
                 return new Response(null, { status: 202 });
             }
@@ -944,7 +953,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                     }
 
                     for (const message of messages) {
-                        this.onmessage?.(message, { authInfo: options?.authInfo, request });
+                        this._deliverMessage(message, { authInfo: options?.authInfo, request });
                     }
                 });
             }
@@ -1024,7 +1033,7 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
                     };
                 }
 
-                this.onmessage?.(message, { authInfo: options?.authInfo, request, closeSSEStream, closeStandaloneSSEStream });
+                this._deliverMessage(message, { authInfo: options?.authInfo, request, closeSSEStream, closeStandaloneSSEStream });
             }
             // The server SHOULD NOT close the SSE stream before sending all JSON-RPC responses
             // This will be handled by the send() method when responses are ready
@@ -1040,6 +1049,73 @@ export class WebStandardStreamableHTTPServerTransport implements Transport {
         }
     }
 
+    /**
+     * Dispatches a message, then retires the mapped stream of a cancelled
+     * request. A cancellation replaces that request's response, so nothing
+     * else will settle its POST or close its SSE stream.
+     */
+    private _deliverMessage(message: JSONRPCMessage, extra: MessageExtraInfo): void {
+        this.onmessage?.(message, extra);
+
+        if (isJSONRPCNotification(message) && message.method === 'notifications/cancelled') {
+            this._settleCancelledRequest((message.params as CancelledNotificationParams | undefined)?.requestId);
+        }
+    }
+
+    /**
+     * Drops a cancelled id from the request bookkeeping. When no other
+     * in-flight request needs the same stream, the stream is completed with an
+     * empty 202 in JSON mode, or closed in SSE mode. Batched siblings that are
+     * still pending keep the stream open.
+     */
+    private _settleCancelledRequest(requestId: RequestId | undefined): void {
+        if (requestId === undefined) {
+            return;
+        }
+
+        const streamId = this._requestToStreamMapping.get(requestId);
+        if (streamId === undefined) {
+            return;
+        }
+
+        this._requestToStreamMapping.delete(requestId);
+        this._requestResponseMap.delete(requestId);
+
+        const relatedIds = [...this._requestToStreamMapping.entries()].filter(([, sid]) => sid === streamId).map(([id]) => id);
+        if (!relatedIds.every(id => this._requestResponseMap.has(id))) {
+            return;
+        }
+
+        const stream = this._streamMapping.get(streamId);
+        if (stream !== undefined) {
+            if (this._enableJsonResponse && stream.resolveJson !== undefined) {
+                const headers: Record<string, string> = {};
+                if (this.sessionId !== undefined) {
+                    headers['mcp-session-id'] = this.sessionId;
+                }
+
+                if (relatedIds.length === 0) {
+                    stream.resolveJson(new Response(null, { status: 202, headers }));
+                } else {
+                    headers['Content-Type'] = 'application/json';
+                    const responses = relatedIds.map(id => this._requestResponseMap.get(id)!);
+                    stream.resolveJson(Response.json(responses.length === 1 ? responses[0] : responses, { status: 200, headers }));
+                }
+            }
+
+            stream.cleanup();
+        }
+
+        // Retire answered siblings even when there is no live stream: with an
+        // eventStore their responses were stored for replay, and without one
+        // nothing is left that could deliver them — keeping the ids would only
+        // leak.
+        for (const id of relatedIds) {
+            this._requestResponseMap.delete(id);
+            this._requestToStreamMapping.delete(id);
+        }
+    }
+
     /**
      * Handles `DELETE` requests to terminate sessions
      */

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

@@ -574,6 +574,123 @@ describe('Zod v4', () => {
         });
     });
 
+    describe('HTTPServerTransport - Cancellation', () => {
+        async function createTransport(enableJsonResponse: boolean): Promise<WebStandardStreamableHTTPServerTransport> {
+            const mcpServer = new McpServer({ name: 'test-server', version: '1.0.0' });
+
+            // Resolves only when the request is cancelled, so the POST can never
+            // settle by the handler finishing.
+            mcpServer.registerTool(
+                'hang',
+                { description: 'Waits for its request to be cancelled', inputSchema: z.object({}) },
+                async (_args, ctx) => {
+                    await new Promise<void>(resolve => {
+                        ctx.mcpReq.signal.addEventListener('abort', () => resolve(), { once: true });
+                    });
+                    return { content: [] };
+                }
+            );
+            mcpServer.registerTool('fast', { description: 'Returns immediately' }, async () => ({ content: [] }));
+
+            const transport = new WebStandardStreamableHTTPServerTransport({
+                sessionIdGenerator: () => randomUUID(),
+                enableJsonResponse
+            });
+            await mcpServer.connect(transport);
+            return transport;
+        }
+
+        async function initialize(transport: WebStandardStreamableHTTPServerTransport): Promise<string> {
+            const response = await transport.handleRequest(createRequest('POST', TEST_MESSAGES.initialize));
+            expect(response.status).toBe(200);
+            const sessionId = response.headers.get('mcp-session-id');
+            expect(sessionId).toBeDefined();
+            return sessionId as string;
+        }
+
+        function cancelRequest(sessionId: string, requestId: string): Request {
+            return createRequest('POST', { jsonrpc: '2.0', method: 'notifications/cancelled', params: { requestId } }, { sessionId });
+        }
+
+        it('settles a cancelled JSON-mode POST with an empty 202 and removes its mappings', async () => {
+            const transport = await createTransport(true);
+            try {
+                const sessionId = await initialize(transport);
+
+                const call = transport.handleRequest(
+                    createRequest(
+                        'POST',
+                        { jsonrpc: '2.0', method: 'tools/call', params: { name: 'hang' }, id: 'cancel-me' },
+                        { sessionId }
+                    )
+                );
+                // Let the POST register its stream and reach the handler.
+                await new Promise(resolve => setImmediate(resolve));
+
+                const cancel = await transport.handleRequest(cancelRequest(sessionId, 'cancel-me'));
+                expect(cancel.status).toBe(202);
+
+                const response = await call;
+                expect(response.status).toBe(202);
+                expect(await response.text()).toBe('');
+                expect((transport as any)._streamMapping.size).toBe(0);
+                expect((transport as any)._requestToStreamMapping.size).toBe(0);
+            } finally {
+                await transport.close();
+            }
+        });
+
+        it('closes a cancelled per-request SSE stream and removes its mappings', async () => {
+            const transport = await createTransport(false);
+            try {
+                const sessionId = await initialize(transport);
+
+                const response = await transport.handleRequest(
+                    createRequest(
+                        'POST',
+                        { jsonrpc: '2.0', method: 'tools/call', params: { name: 'hang' }, id: 'cancel-me' },
+                        { sessionId }
+                    )
+                );
+                expect(response.status).toBe(200);
+                const reader = response.body!.getReader();
+                await new Promise(resolve => setImmediate(resolve));
+
+                await transport.handleRequest(cancelRequest(sessionId, 'cancel-me'));
+
+                expect((await reader.read()).done).toBe(true);
+                expect((transport as any)._streamMapping.size).toBe(0);
+                expect((transport as any)._requestToStreamMapping.size).toBe(0);
+            } finally {
+                await transport.close();
+            }
+        });
+
+        it('still delivers a completed batch sibling when the cancelled id is retired', async () => {
+            const transport = await createTransport(true);
+            try {
+                const sessionId = await initialize(transport);
+
+                const batch: JSONRPCMessage[] = [
+                    { jsonrpc: '2.0', method: 'tools/call', params: { name: 'hang' }, id: 'cancel-me' },
+                    { jsonrpc: '2.0', method: 'tools/call', params: { name: 'fast' }, id: 'fast-1' }
+                ];
+                const pending = transport.handleRequest(createRequest('POST', batch, { sessionId }));
+                await new Promise(resolve => setImmediate(resolve));
+
+                await transport.handleRequest(cancelRequest(sessionId, 'cancel-me'));
+
+                const response = await pending;
+                expect(response.status).toBe(200);
+                expect(await response.json()).toMatchObject({ jsonrpc: '2.0', id: 'fast-1' });
+                expect((transport as any)._streamMapping.size).toBe(0);
+                expect((transport as any)._requestToStreamMapping.size).toBe(0);
+            } finally {
+                await transport.close();
+            }
+        });
+    });
+
     describe('HTTPServerTransport - Session Callbacks', () => {
         it('should call onsessioninitialized callback', async () => {
             const onInitialized = vi.fn();
@@ -825,6 +942,71 @@ describe('Zod v4', () => {
             expect((transport as any)._requestToStreamMapping.has('disconnect-1')).toBe(false);
         });
 
+        it('retires an answered batch sibling when the cancelled id has no live stream left', async () => {
+            // Batch [answered, cancelled]: the answered sibling's response is
+            // stored and parked in _requestResponseMap while the cancelled id
+            // is still pending. The handler closed the per-request stream, so
+            // by the time the cancellation settles the batch there is no
+            // _streamMapping entry. The answered id must still be retired —
+            // send() does the same when the stream is missing.
+            mcpServer.registerTool(
+                'closeThenReturn',
+                { description: 'closeSSE then return', inputSchema: z.object({}) },
+                async (_args, ctx): Promise<CallToolResult> => {
+                    ctx.http?.closeSSE?.();
+                    return { content: [{ type: 'text', text: 'answered' }] };
+                }
+            );
+            mcpServer.registerTool(
+                'hangUntilCancelled',
+                { description: 'wait for cancellation', inputSchema: z.object({}) },
+                async (_args, ctx): Promise<CallToolResult> => {
+                    await new Promise<void>(resolve => {
+                        ctx.mcpReq.signal.addEventListener('abort', () => resolve(), { once: true });
+                    });
+                    return { content: [] };
+                }
+            );
+
+            sessionId = await initializeServer();
+            storedEvents.clear();
+
+            const batch: JSONRPCMessage[] = [
+                { jsonrpc: '2.0', method: 'tools/call', params: { name: 'closeThenReturn', arguments: {} }, id: 'answered' },
+                { jsonrpc: '2.0', method: 'tools/call', params: { name: 'hangUntilCancelled', arguments: {} }, id: 'cancel-me' }
+            ];
+            const response = await transport.handleRequest(createRequest('POST', batch, { sessionId }));
+            await response.text().catch(() => {});
+
+            // Wait until the answered sibling is parked and its stream is gone.
+            for (let i = 0; i < 100; i++) {
+                // eslint-disable-next-line @typescript-eslint/no-explicit-any
+                if ((transport as any)._requestResponseMap.has('answered')) break;
+                await new Promise(resolve => setTimeout(resolve, 5));
+            }
+            // eslint-disable-next-line @typescript-eslint/no-explicit-any
+            expect((transport as any)._streamMapping.size).toBe(0);
+            // eslint-disable-next-line @typescript-eslint/no-explicit-any
+            expect((transport as any)._requestResponseMap.has('answered')).toBe(true);
+
+            const cancel = await transport.handleRequest(
+                createRequest(
+                    'POST',
+                    { jsonrpc: '2.0', method: 'notifications/cancelled', params: { requestId: 'cancel-me' } },
+                    { sessionId }
+                )
+            );
+            expect(cancel.status).toBe(202);
+
+            // eslint-disable-next-line @typescript-eslint/no-explicit-any
+            expect((transport as any)._requestResponseMap.size).toBe(0);
+            // eslint-disable-next-line @typescript-eslint/no-explicit-any
+            expect((transport as any)._requestToStreamMapping.size).toBe(0);
+
+            const stored = [...storedEvents.values()].map(e => e.message);
+            expect(stored.some(m => 'id' in m && m.id === 'answered' && 'result' in m)).toBe(true);
+        });
+
         it('should accept Last-Event-ID reconnect after closeSSEStream() and replay stored events', async () => {
             // closeSSEStream() removes the _streamMapping entry, so the
             // replayEvents() conflict check sees no active connection and the