From 945773a9fe73675b39edb814275541b76f711ec6 Mon Sep 17 00:00:00 2001 From: Felix Weinberger <3823880+felixweinberger@users.noreply.github.com> Date: Fri, 23 Oct 2026 15:25:57 +0000 Subject: [PATCH] fix(client): do not wait for the end of an unread POST reply body **Before:** after a successful POST whose reply body the client does not read (a 202, a 2xx reply to a message with no requests, a 2xx reply with an unexpected content type, any OK reply on the SSE transport), `send()` awaited `response.text()`, which settles only when the body ends. A server or proxy that sends the headers and keeps the response open stalled `send()`, and with it `Client.connect()`. **After:** the body is cancelled without awaiting. A fully received reply keeps its connection for reuse; a partly received one has it closed. **History of these lines:** #1214 added `await response.body?.cancel()`. #1320 replaced it with `await response.text()`, because `cancel()` could hang. Both forms wait, and both waits can hang: `cancel()` on a response that a custom fetch has cloned does not settle until the other branch is consumed. Not awaiting avoids both; a test covers the cloned case. A custom fetch may return a body that is not a WHATWG stream (node-fetch returns a Node stream, which has no `cancel()`). Such a body is still read with `response.text()`, also without awaiting. The four sites share one internal helper, `releaseBody()`, which never throws or rejects. The non-2xx drains, the GET stream and the DELETE are left as they are. --- .changeset/release-post-reply-body.md | 5 + packages/client/src/client/sse.ts | 3 +- packages/client/src/client/streamableHttp.ts | 7 +- .../core-internal/src/shared/transport.ts | 6 + .../test/client/heldOpenPostReply.test.ts | 195 ++++++++++++++++++ 5 files changed, 212 insertions(+), 4 deletions(-) create mode 100644 .changeset/release-post-reply-body.md create mode 100644 test/integration/test/client/heldOpenPostReply.test.ts diff --git a/.changeset/release-post-reply-body.md b/.changeset/release-post-reply-body.md new file mode 100644 index 0000000000..29073d9d07 --- /dev/null +++ b/.changeset/release-post-reply-body.md @@ -0,0 +1,5 @@ +--- +'@modelcontextprotocol/client': patch +--- + +`StreamableHTTPClientTransport.send()` and `SSEClientTransport.send()` no longer wait for the end of a successful POST reply whose body they do not read (a `202`, a reply to a message with no requests, a reply with an unexpected content type). A server or proxy that sends the headers and keeps the response open no longer stalls `send()` or `Client.connect()`. The unread body is now cancelled without waiting, instead of being read to its end; when part of it has not arrived yet, the connection is closed instead of reused. A body with no `cancel()`, as a custom `fetch` may return, is still read, but no longer waited for. diff --git a/packages/client/src/client/sse.ts b/packages/client/src/client/sse.ts index 55c346ac85..8cadfcd5bf 100644 --- a/packages/client/src/client/sse.ts +++ b/packages/client/src/client/sse.ts @@ -5,6 +5,7 @@ import { fetchLeavingRedirects, fetchWithinOrigin, JSONRPCMessageSchema, + releaseBody, SdkError, SdkErrorCode, SdkHttpError, @@ -491,7 +492,7 @@ export class SSEClientTransport implements Transport { } // Release connection - POST responses don't have content we need - await response.text?.().catch(() => {}); + releaseBody(response); } catch (error) { this.onerror?.(error as Error); throw error; diff --git a/packages/client/src/client/streamableHttp.ts b/packages/client/src/client/streamableHttp.ts index 091674e101..9f49b4ecb5 100644 --- a/packages/client/src/client/streamableHttp.ts +++ b/packages/client/src/client/streamableHttp.ts @@ -16,6 +16,7 @@ import { mcpNameSource, mediaTypeEssence, PROTOCOL_VERSION_META_KEY, + releaseBody, SdkError, SdkErrorCode, SdkHttpError, @@ -1164,7 +1165,7 @@ export class StreamableHTTPClientTransport implements Transport { // If the response is 202 Accepted, there's no body to process if (response.status === 202) { - await response.text?.().catch(() => {}); + releaseBody(response); // if the accepted notification is initialized, we start the SSE stream // if it's supported by the server if (isInitializedNotification(message)) { @@ -1208,14 +1209,14 @@ export class StreamableHTTPClientTransport implements Transport { this.onmessage?.(msg); } } else { - await response.text?.().catch(() => {}); + releaseBody(response); throw new SdkError(SdkErrorCode.ClientHttpUnexpectedContent, `Unexpected content type: ${contentType}`, { contentType }); } } else { // No requests in message but got 200 OK - still need to release connection - await response.text?.().catch(() => {}); + releaseBody(response); } } catch (error) { // Intentional per-request abort BEFORE response headers (the diff --git a/packages/core-internal/src/shared/transport.ts b/packages/core-internal/src/shared/transport.ts index 1679995081..89194fd444 100644 --- a/packages/core-internal/src/shared/transport.ts +++ b/packages/core-internal/src/shared/transport.ts @@ -111,6 +111,12 @@ export function unfollowedRedirect(url: string | URL, response: Response): strin return `Redirect to ${target.href} not followed; use that URL as the endpoint if it is the intended server`; } +/** Releases the unread body of `response` without waiting for it; reads it to its end when it has no `cancel()`. Never throws. */ +export function releaseBody(response: Response): void { + const release = async () => (typeof response.body?.cancel === 'function' ? response.body.cancel() : response.text?.()); + release().catch(() => {}); +} + /** * Options for sending a JSON-RPC message. */ diff --git a/test/integration/test/client/heldOpenPostReply.test.ts b/test/integration/test/client/heldOpenPostReply.test.ts new file mode 100644 index 0000000000..fac63c36d6 --- /dev/null +++ b/test/integration/test/client/heldOpenPostReply.test.ts @@ -0,0 +1,195 @@ +import type { IncomingMessage, Server, ServerResponse } from 'node:http'; +import { createServer } from 'node:http'; +import type { AddressInfo } from 'node:net'; +import { Readable } from 'node:stream'; + +import { Client, SSEClientTransport, StreamableHTTPClientTransport } from '@modelcontextprotocol/client'; +import { afterEach, describe, expect, it } from 'vitest'; + +const initializeResult = (id: unknown) => ({ + jsonrpc: '2.0', + id, + result: { protocolVersion: '2025-06-18', capabilities: {}, serverInfo: { name: 'held-open', version: '1.0.0' } } +}); + +const readJson = async (req: IncomingMessage) => { + let raw = ''; + for await (const chunk of req) raw += chunk; + return JSON.parse(raw); +}; + +// Sends the headers of a bodiless reply and never ends the response. +const holdOpen = (res: ServerResponse, status = 202, headers?: Record) => { + res.writeHead(status, headers); + res.flushHeaders(); +}; + +// A fetch whose POST replies carry `body` in place of their own. +const fetchWithBody = + (body: unknown, replies: Response[] = []): typeof fetch => + async (input, init) => { + const response = await fetch(input, init); + if (init?.method !== 'POST') return response; + replies.push(response); + return Object.defineProperty(response, 'body', { value: body }); + }; + +describe('POST reply that is never ended', () => { + let server: Server; + + const listen = async (handler: (req: IncomingMessage, res: ServerResponse) => void) => { + server = createServer(handler); + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)); + return new URL(`http://127.0.0.1:${(server.address() as AddressInfo).port}/mcp`); + }; + + afterEach(async () => { + server.closeAllConnections(); + await new Promise(resolve => server.close(resolve)); + }); + + const streamableHttpServer = () => + listen(async (req, res) => { + if (req.method !== 'POST') { + res.writeHead(405).end(); + return; + } + const message = await readJson(req); + if (message.method === 'initialize') { + res.writeHead(200, { 'content-type': 'application/json' }).end(JSON.stringify(initializeResult(message.id))); + return; + } + holdOpen(res); + }); + + const sseServer = () => { + let stream: ServerResponse | undefined; + return listen(async (req, res) => { + if (req.method === 'GET') { + res.writeHead(200, { 'content-type': 'text/event-stream' }); + res.write('event: endpoint\ndata: /messages\n\n'); + stream = res; + return; + } + const message = await readJson(req); + holdOpen(res); + if (message.method === 'initialize') { + stream?.write(`event: message\ndata: ${JSON.stringify(initializeResult(message.id))}\n\n`); + } + }); + }; + + describe.each([ + ['the global fetch', fetch], + ['a fetch whose reply body is a Node stream, which has no cancel()', fetchWithBody(Readable.from([]))], + [ + 'a fetch whose reply body has a cancel() that throws', + fetchWithBody({ + cancel() { + throw new Error('cancel failed'); + } + }) + ], + ['a fetch whose reply body has a cancel() that returns undefined', fetchWithBody({ cancel() {} })] + ])('with %s', (_name, fetch) => { + it('does not stall send() on the streamable HTTP transport', async () => { + const transport = new StreamableHTTPClientTransport(await streamableHttpServer(), { fetch }); + await transport.start(); + + await transport.send({ jsonrpc: '2.0', method: 'notifications/initialized' }); + + await transport.close(); + }); + + it('does not stall Client.connect() over streamable HTTP', async () => { + const client = new Client({ name: 'test', version: '1.0.0' }); + + await client.connect(new StreamableHTTPClientTransport(await streamableHttpServer(), { fetch })); + + expect(client.getServerVersion()?.name).toBe('held-open'); + await client.close(); + }); + + it('does not stall send() of a notification on a 200 reply', async () => { + const url = await listen((_req, res) => holdOpen(res, 200)); + const transport = new StreamableHTTPClientTransport(url, { fetch }); + await transport.start(); + + await transport.send({ jsonrpc: '2.0', method: 'notifications/initialized' }); + + await transport.close(); + }); + + it('does not stall the rejection of a reply with an unexpected content type', async () => { + const url = await listen((_req, res) => holdOpen(res, 200, { 'content-type': 'text/plain' })); + const transport = new StreamableHTTPClientTransport(url, { fetch }); + await transport.start(); + + await expect(transport.send({ jsonrpc: '2.0', id: 1, method: 'ping' })).rejects.toThrow('Unexpected content type: text/plain'); + + await transport.close(); + }); + + it('does not stall send() on the SSE transport', async () => { + const transport = new SSEClientTransport(await sseServer(), { fetch }); + await transport.start(); + + await transport.send({ jsonrpc: '2.0', method: 'notifications/initialized' }); + + await transport.close(); + }); + + it('does not stall Client.connect() over SSE', async () => { + const client = new Client({ name: 'test', version: '1.0.0' }); + + await client.connect(new SSEClientTransport(await sseServer(), { fetch })); + + expect(client.getServerVersion()?.name).toBe('held-open'); + await client.close(); + }); + }); + + it('does not stall send() when a custom fetch has cloned the response', async () => { + const url = await listen((req, res) => { + req.resume(); + res.writeHead(202).end(); + }); + const fetchWithClone: typeof fetch = async (input, init) => { + const response = await fetch(input, init); + response.clone(); + return response; + }; + const transport = new StreamableHTTPClientTransport(url, { fetch: fetchWithClone }); + await transport.start(); + + await transport.send({ jsonrpc: '2.0', method: 'notifications/initialized' }); + + await transport.close(); + }); + + it('closes the connection of the reply', async () => { + let closed: Promise | undefined; + const url = await listen((_req, res) => { + closed = new Promise(resolve => res.on('close', resolve)); + holdOpen(res); + }); + const transport = new StreamableHTTPClientTransport(url); + await transport.start(); + + await transport.send({ jsonrpc: '2.0', method: 'notifications/initialized' }); + + await closed; + await transport.close(); + }); + + it('reads a reply whose body has no cancel()', async () => { + const replies: Response[] = []; + const transport = new StreamableHTTPClientTransport(await streamableHttpServer(), { fetch: fetchWithBody({}, replies) }); + await transport.start(); + + await transport.send({ jsonrpc: '2.0', method: 'notifications/initialized' }); + + expect(replies[0]?.bodyUsed).toBe(true); + await transport.close(); + }); +});