From 43b692d2eab3b023f788c39e5fd6e23c162322fe Mon Sep 17 00:00:00 2001 From: macrogui <40064208+macrogui@users.noreply.github.com> Date: Fri, 18 Sep 2026 03:14:03 +0800 Subject: [PATCH] fix(client): contain HTTP transport errors on request POSTs A POST whose HTTP exchange itself fails (server drops the connection before sending a response, connect/read errors, timeouts) escaped _handle_post_request unhandled. The exception propagated out of the per-request task, cancelling the transport's task group: the write stream closed, so every subsequent request on the session failed with 'Connection closed' and transport teardown surfaced the raw error again as an unhandled ExceptionGroup. Resolve only the in-flight request with a synthesized CONNECTION_CLOSED error via the same _resolve_abandoned_request path already used for 202-accepted and non-resumable SSE-drop outcomes, so the session and its transport survive and later requests go out on a fresh connection, matching the TypeScript client's recovery behavior. Fixes #3522 --- src/mcp/client/streamable_http.py | 159 +++++++++++++++------------ tests/client/test_streamable_http.py | 36 ++++++ 2 files changed, 124 insertions(+), 71 deletions(-) diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index 82de50fd05..fa82566115 100644 --- a/src/mcp/client/streamable_http.py +++ b/src/mcp/client/streamable_http.py @@ -352,83 +352,100 @@ async def _handle_post_request(self, ctx: RequestContext) -> None: if ctx.metadata is not None and ctx.metadata.headers is not None: headers.update(ctx.metadata.headers) - async with stream_within_origin( - ctx.client, - "POST", - self.url, - json=message.model_dump(by_alias=True, mode="json", exclude_unset=True), - headers=headers, - ) as response: - if response.status_code == 202: - logger.debug("Received 202 Accepted") - if isinstance(message, JSONRPCRequest): - # A request's response arrives on this POST's body; 202 says - # none will follow. Resolve rather than park the caller forever. - await self._resolve_abandoned_request( - ctx.read_stream_writer, - message.id, - "server answered a request with 202 Accepted", - code=INVALID_REQUEST, - ) - return + try: + async with stream_within_origin( + ctx.client, + "POST", + self.url, + json=message.model_dump(by_alias=True, mode="json", exclude_unset=True), + headers=headers, + ) as response: + if response.status_code == 202: + logger.debug("Received 202 Accepted") + if isinstance(message, JSONRPCRequest): + # A request's response arrives on this POST's body; 202 says + # none will follow. Resolve rather than park the caller forever. + await self._resolve_abandoned_request( + ctx.read_stream_writer, + message.id, + "server answered a request with 202 Accepted", + code=INVALID_REQUEST, + ) + return - if (redirect := _unfollowed_redirect(response)) is not None: - logger.warning(redirect) - if isinstance(message, JSONRPCRequest): - await self._resolve_abandoned_request( - ctx.read_stream_writer, message.id, redirect, code=INVALID_REQUEST - ) - return + if (redirect := _unfollowed_redirect(response)) is not None: + logger.warning(redirect) + if isinstance(message, JSONRPCRequest): + await self._resolve_abandoned_request( + ctx.read_stream_writer, message.id, redirect, code=INVALID_REQUEST + ) + return - if response.status_code >= 400: - if isinstance(message, JSONRPCRequest): - # A spec-correct server may return the JSON-RPC error in the - # body at a non-2xx status (e.g. 400 for INVALID_PARAMS, 404 - # for METHOD_NOT_FOUND). Surface that error rather than the - # status-derived stand-in below. - if response.headers.get("content-type", "").lower().startswith("application/json"): - try: - body = await response.aread() - parsed = jsonrpc_message_adapter.validate_json(body, by_name=False) - if isinstance(parsed, JSONRPCError): - # The server may have set `id: null` (request rejected before its - # id was parsed); use this request's id so correlation works. - reply = JSONRPCError(jsonrpc="2.0", id=message.id, error=parsed.error) - await ctx.read_stream_writer.send(SessionMessage(reply)) - return - except (httpx2.StreamError, ValidationError): - pass - logger.debug("Non-2xx body was not a JSON-RPC error; using fallback") - if response.status_code == 404: - if self.session_id is None: - # No session yet → 404 is the HTTP-level spelling of - # METHOD_NOT_FOUND (gateway / legacy server doesn't know - # this method); "Session terminated" would be a lie here. - error_data = ErrorData(code=METHOD_NOT_FOUND, message="Not Found") + if response.status_code >= 400: + if isinstance(message, JSONRPCRequest): + # A spec-correct server may return the JSON-RPC error in the + # body at a non-2xx status (e.g. 400 for INVALID_PARAMS, 404 + # for METHOD_NOT_FOUND). Surface that error rather than the + # status-derived stand-in below. + if response.headers.get("content-type", "").lower().startswith("application/json"): + try: + body = await response.aread() + parsed = jsonrpc_message_adapter.validate_json(body, by_name=False) + if isinstance(parsed, JSONRPCError): + # The server may have set `id: null` (request rejected before its + # id was parsed); use this request's id so correlation works. + reply = JSONRPCError(jsonrpc="2.0", id=message.id, error=parsed.error) + await ctx.read_stream_writer.send(SessionMessage(reply)) + return + except (httpx2.StreamError, ValidationError): + pass + logger.debug("Non-2xx body was not a JSON-RPC error; using fallback") + if response.status_code == 404: + if self.session_id is None: + # No session yet → 404 is the HTTP-level spelling of + # METHOD_NOT_FOUND (gateway / legacy server doesn't know + # this method); "Session terminated" would be a lie here. + error_data = ErrorData(code=METHOD_NOT_FOUND, message="Not Found") + else: + error_data = ErrorData(code=INVALID_REQUEST, message="Session terminated") else: - error_data = ErrorData(code=INVALID_REQUEST, message="Session terminated") - else: - error_data = ErrorData(code=INTERNAL_ERROR, message="Server returned an error response") - session_message = SessionMessage(JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data)) - await ctx.read_stream_writer.send(session_message) - return + error_data = ErrorData(code=INTERNAL_ERROR, message="Server returned an error response") + session_message = SessionMessage(JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data)) + await ctx.read_stream_writer.send(session_message) + return - if self._is_initialization_request(message): - self._maybe_extract_session_id_from_response(response) + if self._is_initialization_request(message): + self._maybe_extract_session_id_from_response(response) - # Per https://modelcontextprotocol.io/specification/2025-06-18/basic#notifications: - # The server MUST NOT send a response to notifications. + # Per https://modelcontextprotocol.io/specification/2025-06-18/basic#notifications: + # The server MUST NOT send a response to notifications. + if isinstance(message, JSONRPCRequest): + content_type = response.headers.get("content-type", "").lower() + if content_type.startswith("application/json"): + await self._handle_json_response(response, ctx.read_stream_writer, request_id=message.id) + elif content_type.startswith("text/event-stream"): + await self._handle_sse_response(response, ctx) + else: + logger.error(f"Unexpected content type: {content_type}") + error_data = ErrorData(code=INVALID_REQUEST, message=f"Unexpected content type: {content_type}") + error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data)) + await ctx.read_stream_writer.send(error_msg) + + except httpx2.TransportError as exc: + # The HTTP exchange itself failed (connection dropped before a + # response, connect/read errors, timeouts). This must fail only + # the request that was in flight: resolve its waiter with a + # synthesized error so the session and its transport survive, + # mirroring how non-resumable SSE drops are handled above. if isinstance(message, JSONRPCRequest): - content_type = response.headers.get("content-type", "").lower() - if content_type.startswith("application/json"): - await self._handle_json_response(response, ctx.read_stream_writer, request_id=message.id) - elif content_type.startswith("text/event-stream"): - await self._handle_sse_response(response, ctx) - else: - logger.error(f"Unexpected content type: {content_type}") - error_data = ErrorData(code=INVALID_REQUEST, message=f"Unexpected content type: {content_type}") - error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data)) - await ctx.read_stream_writer.send(error_msg) + logger.warning(f"HTTP transport error on POST for request {message.id}: {exc!r}") + await self._resolve_abandoned_request( + ctx.read_stream_writer, + message.id, + f"HTTP transport error: {exc}", + ) + elif isinstance(message, JSONRPCNotification): + logger.warning(f"HTTP transport error on POST for {message.method}: {exc!r}") async def _handle_json_response( self, diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py index c6e62ad94a..ee971ddc18 100644 --- a/tests/client/test_streamable_http.py +++ b/tests/client/test_streamable_http.py @@ -968,3 +968,39 @@ async def test_https_endpoint_redirected_to_plain_http_elsewhere_never_suggests_ The server is likely behind a TLS-terminating proxy whose forwarded headers it does not trust, often combined with a trailing-slash difference. Try https://backend.lan:8000/mcp/ instead, or fix the proxy settings.\ """) + + +@pytest.mark.anyio +async def test_a_post_transport_error_fails_only_that_request_and_keeps_the_session() -> None: + """A POST whose HTTP exchange itself fails (server drops the connection before + responding) must resolve only that request with an error; the session and its + transport stay usable, matching the TypeScript client's recovery behavior.""" + + def handler(request: httpx2.Request) -> httpx2.Response: + body = json.loads(request.content) + if body.get("id") == "drop-1": + raise httpx2.RemoteProtocolError("Server disconnected without sending a response.") + return httpx2.Response( + 200, + headers={"content-type": "application/json"}, + json={"jsonrpc": "2.0", "id": body["id"], "result": {}}, + ) + + with anyio.fail_after(5): + async with ( + httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http, + streamable_http_client("http://test/mcp", http_client=http) as (read, write), + ): + await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id="drop-1", method="tools/call", params={}))) + await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id="after-1", method="ping"))) + first = await read.receive() + second = await read.receive() + + assert isinstance(first, SessionMessage) + assert isinstance(first.message, JSONRPCError) + assert first.message.id == "drop-1" + assert first.message.error.code == CONNECTION_CLOSED + + assert isinstance(second, SessionMessage) + assert isinstance(second.message, JSONRPCResponse) + assert second.message.id == "after-1"