From d53d47bf1c8de443179baa13597c9206df36e582 Mon Sep 17 00:00:00 2001 From: Himanshu Gupta Date: Tue, 14 Jul 2026 01:19:43 +0200 Subject: [PATCH 1/2] fix(client): surface non-2xx HTTP responses to the waiting caller MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A 4xx/5xx response to a message POST hit a bare `response.raise_for_status()` inside the POST's background task. Nothing caught the exception and nothing resolved the request, so the caller sat until its read timeout fired: a 401 was indistinguishable from a slow server, and the escaping error tore down the transport's task group with it. Non-2xx statuses are now answered rather than raised, generalising the handling 404 already had: a `JSONRPCError` stamped with the original request's id, sent as a `SessionMessage` over the read stream, so it resolves through the normal response-correlation path. A spec-compliant JSON-RPC error in the body is preferred over the status-derived stand-in. The connection stays usable. This matches the behaviour already shipped on the 2.0 line, with one deliberate difference: 404 is answered ahead of the body check. A terminated session 404s *with* a JSON-RPC error body, so preferring it would replace this line's `Session terminated` / code 32600 — which callers match on for session expiry — with the server's own wording. Callers still cannot tell a terminal 401 from a retryable 503; both remain `-32603 Server returned an error response`, as on 2.0. Carrying the status would be a follow-up on the 2.0 line. Github-Issue: #3091 Reported-by: afterrburn --- src/mcp/client/streamable_http.py | 60 ++++++--- tests/shared/test_streamable_http.py | 190 +++++++++++++++++++++++++++ 2 files changed, 234 insertions(+), 16 deletions(-) diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index ed28fcc275..12385da42a 100644 --- a/src/mcp/client/streamable_http.py +++ b/src/mcp/client/streamable_http.py @@ -20,6 +20,7 @@ from anyio.abc import TaskGroup from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream from httpx_sse import EventSource, ServerSentEvent, aconnect_sse +from pydantic import ValidationError from typing_extensions import deprecated from mcp.shared._httpx_utils import ( @@ -28,6 +29,7 @@ ) from mcp.shared.message import ClientMessageMetadata, SessionMessage from mcp.types import ( + INTERNAL_ERROR, ErrorData, InitializeResult, JSONRPCError, @@ -347,15 +349,14 @@ async def _handle_post_request(self, ctx: RequestContext) -> None: logger.debug("Received 202 Accepted") return - if response.status_code == 404: # pragma: no branch + if response.status_code >= 400: + # Raising here would only throw into this POST's background task, where nothing + # catches it: the waiting caller would never be told, and would sit until its + # read timeout fired. Answer it instead, on the id it is waiting for. if isinstance(message.root, JSONRPCRequest): - await self._send_session_terminated_error( # pragma: no cover - ctx.read_stream_writer, # pragma: no cover - message.root.id, # pragma: no cover - ) # pragma: no cover - return # pragma: no cover + await self._send_http_status_error(response, ctx.read_stream_writer, message.root.id) + return - response.raise_for_status() if is_initialization: self._maybe_extract_session_id_from_response(response) @@ -507,19 +508,46 @@ async def _handle_unexpected_content_type( logger.error(error_msg) # pragma: no cover await read_stream_writer.send(ValueError(error_msg)) # pragma: no cover - async def _send_session_terminated_error( + async def _send_http_status_error( self, + response: httpx.Response, read_stream_writer: StreamWriter, request_id: RequestId, ) -> None: - """Send a session terminated error response.""" - jsonrpc_error = JSONRPCError( - jsonrpc="2.0", - id=request_id, - error=ErrorData(code=32600, message="Session terminated"), - ) - session_message = SessionMessage(JSONRPCMessage(jsonrpc_error)) - await read_stream_writer.send(session_message) + """Surface a non-2xx response to the caller that is waiting on `request_id`. + + The error is routed as a `JSONRPCError` stamped with the original request's id, because + that id is what the session correlates a reply against; a bare exception carries none and + so can never resolve the caller. + """ + if response.status_code == 404: + # Answered ahead of the body, unlike every other status: a terminated session 404s + # *with* a JSON-RPC error body, and preferring it would rewrite the message and code + # this release line already reports. Callers match on both, so 404 stays verbatim. + error_data = ErrorData(code=32600, message="Session terminated") + await read_stream_writer.send( + SessionMessage(JSONRPCMessage(JSONRPCError(jsonrpc="2.0", id=request_id, error=error_data))) + ) + return + + # A spec-correct server may put the JSON-RPC error in the body of a non-2xx (e.g. 400 for + # INVALID_PARAMS). Prefer it: the server's account of why it refused beats our stand-in. + if response.headers.get(CONTENT_TYPE, "").lower().startswith(JSON): + try: + body = await response.aread() + parsed = JSONRPCMessage.model_validate_json(body) + if isinstance(parsed.root, JSONRPCError): + # Re-stamped with this request's id rather than trusting the body's: a server + # echoing someone else's id would otherwise resolve the wrong caller. + reply = JSONRPCError(jsonrpc="2.0", id=request_id, error=parsed.root.error) + await read_stream_writer.send(SessionMessage(JSONRPCMessage(reply))) + return + except (httpx.StreamError, ValidationError): + logger.debug("Non-2xx body was not a JSON-RPC error; using the status-derived fallback") + + error_data = ErrorData(code=INTERNAL_ERROR, message="Server returned an error response") + jsonrpc_error = JSONRPCError(jsonrpc="2.0", id=request_id, error=error_data) + await read_stream_writer.send(SessionMessage(JSONRPCMessage(jsonrpc_error))) async def post_writer( self, diff --git a/tests/shared/test_streamable_http.py b/tests/shared/test_streamable_http.py index 631c96e86c..f230616ef5 100644 --- a/tests/shared/test_streamable_http.py +++ b/tests/shared/test_streamable_http.py @@ -2353,3 +2353,193 @@ async def test_streamablehttp_client_deprecation_warning(basic_server: None, bas await session.initialize() tools = await session.list_tools() assert len(tools.tools) > 0 + + +@pytest.mark.anyio +@pytest.mark.parametrize("status", [401, 403, 500, 503]) +async def test_non_2xx_status_reaches_the_waiting_caller(status: int) -> None: + """A 4xx/5xx to a request is answered as a JSON-RPC error rather than raised into the void. + + Regression test for #3091: the status used to escape `raise_for_status()` inside the POST's + background task, where nothing caught it and nothing resolved the caller, so the request hung + until its read timeout — a 401 was indistinguishable from a slow server. The `fail_after` below + is what pins that: without the fix this test hangs rather than failing an assertion. + """ + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(status) + + with anyio.fail_after(5): + async with ( + httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client, + streamable_http_client("http://test/mcp", http_client=http_client) as (read, write, _), + read, + write, + ): + request = types.JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/list", params={}) + await write.send(SessionMessage(types.JSONRPCMessage(request))) + reply = await read.receive() + + assert isinstance(reply, SessionMessage) + assert isinstance(reply.message.root, types.JSONRPCError) + assert reply.message.root.id == 1 + assert reply.message.root.error == types.ErrorData( + code=types.INTERNAL_ERROR, message="Server returned an error response" + ) + + +@pytest.mark.anyio +async def test_404_still_reports_session_terminated() -> None: + """The 404 spelling is unchanged by #3091: same message, same (positive) code callers match on.""" + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(404) + + with anyio.fail_after(5): + async with ( + httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client, + streamable_http_client("http://test/mcp", http_client=http_client) as (read, write, _), + read, + write, + ): + request = types.JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/list", params={}) + await write.send(SessionMessage(types.JSONRPCMessage(request))) + reply = await read.receive() + + assert isinstance(reply, SessionMessage) + assert isinstance(reply.message.root, types.JSONRPCError) + assert reply.message.root.error == types.ErrorData(code=32600, message="Session terminated") + + +@pytest.mark.anyio +async def test_non_2xx_prefers_the_servers_own_jsonrpc_error_body() -> None: + """When the server explains itself in the body, the caller gets that, not our stand-in. + + The reply is re-stamped with the request's id rather than trusting the body's, so a server that + echoes a different id cannot resolve the wrong caller (or, echoing none, no caller at all). + """ + + def handler(request: httpx.Request) -> httpx.Response: + error = {"code": types.INVALID_PARAMS, "message": "Unknown tool: nope"} + return httpx.Response(400, json={"jsonrpc": "2.0", "id": 99, "error": error}) + + with anyio.fail_after(5): + async with ( + httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client, + streamable_http_client("http://test/mcp", http_client=http_client) as (read, write, _), + read, + write, + ): + request = types.JSONRPCRequest(jsonrpc="2.0", id=7, method="tools/call", params={}) + await write.send(SessionMessage(types.JSONRPCMessage(request))) + reply = await read.receive() + + assert isinstance(reply, SessionMessage) + assert isinstance(reply.message.root, types.JSONRPCError) + assert reply.message.root.id == 7 + assert reply.message.root.error == types.ErrorData(code=types.INVALID_PARAMS, message="Unknown tool: nope") + + +@pytest.mark.anyio +async def test_a_failed_request_leaves_the_session_usable() -> None: + """The connection survives a 500: the error resolves one call, it does not tear down the transport. + + In v1 the escaping `HTTPStatusError` cancelled the transport's task group, so a single 500 took + the whole session with it. + """ + + def handler(request: httpx.Request) -> httpx.Response: + request_id = json.loads(request.content)["id"] + if request_id == 1: + return httpx.Response(500) + return httpx.Response(200, json={"jsonrpc": "2.0", "id": request_id, "result": {}}) + + with anyio.fail_after(5): + async with ( + httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client, + streamable_http_client("http://test/mcp", http_client=http_client) as (read, write, _), + read, + write, + ): + failing = types.JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/list", params={}) + await write.send(SessionMessage(types.JSONRPCMessage(failing))) + first = await read.receive() + + following = types.JSONRPCRequest(jsonrpc="2.0", id=2, method="tools/list", params={}) + await write.send(SessionMessage(types.JSONRPCMessage(following))) + second = await read.receive() + + assert isinstance(first, SessionMessage) + assert isinstance(first.message.root, types.JSONRPCError) + assert isinstance(second, SessionMessage) + assert isinstance(second.message.root, types.JSONRPCResponse) + + +@pytest.mark.anyio +async def test_non_2xx_to_a_notification_resolves_nobody_and_keeps_the_session_alive() -> None: + """A notification has no id and no waiter, so a 4xx to one cannot be surfaced — only survived. + + The following request still gets its answer: the rejected notification must not take the + transport down with it. + """ + + def handler(request: httpx.Request) -> httpx.Response: + body = json.loads(request.content) + if "id" not in body: + return httpx.Response(500) + return httpx.Response(200, json={"jsonrpc": "2.0", "id": body["id"], "result": {}}) + + with anyio.fail_after(5): + async with ( + httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client, + streamable_http_client("http://test/mcp", http_client=http_client) as (read, write, _), + read, + write, + ): + notification = types.JSONRPCNotification(jsonrpc="2.0", method="notifications/progress") + await write.send(SessionMessage(types.JSONRPCMessage(notification))) + + request = types.JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/list", params={}) + await write.send(SessionMessage(types.JSONRPCMessage(request))) + reply = await read.receive() + + assert isinstance(reply, SessionMessage) + assert isinstance(reply.message.root, types.JSONRPCResponse) + + +@pytest.mark.anyio +@pytest.mark.parametrize( + ("body", "case"), + [ + ({"jsonrpc": "2.0", "id": 1, "result": {}}, "valid JSON-RPC, but not an error"), + ({"detail": "gateway rejected the request"}, "JSON, but not JSON-RPC at all"), + ], +) +async def test_a_json_body_that_is_not_a_jsonrpc_error_falls_back_to_the_stand_in( + body: dict[str, Any], case: str +) -> None: + """`Content-Type: application/json` is not a promise of a JSON-RPC error body. + + Proxies and gateways send JSON that is nothing of the sort. Neither shape may be handed to the + caller as their answer, and neither may be allowed to escape: both fall back to the stand-in. + """ + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(500, json=body) + + with anyio.fail_after(5): + async with ( + httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client, + streamable_http_client("http://test/mcp", http_client=http_client) as (read, write, _), + read, + write, + ): + request = types.JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/list", params={}) + await write.send(SessionMessage(types.JSONRPCMessage(request))) + reply = await read.receive() + + assert isinstance(reply, SessionMessage) + assert isinstance(reply.message.root, types.JSONRPCError) + assert reply.message.root.error == types.ErrorData( + code=types.INTERNAL_ERROR, message="Server returned an error response" + ) From a674d2ac2dbecb2aeff0837ed445fd0f2f99d865 Mon Sep 17 00:00:00 2001 From: Himanshu Gupta Date: Tue, 14 Jul 2026 01:44:18 +0200 Subject: [PATCH 2/2] fix(client): catch read failures on the non-2xx body MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `httpx.StreamError` derives from `RuntimeError`, not `httpx.HTTPError`, so it does not cover a body that stalls or truncates while being read: `aread()` raises `ReadTimeout`/`RemoteProtocolError` there. Those escaped into the POST's background task and stranded the caller — the bug this path exists to prevent. Github-Issue: #3091 --- src/mcp/client/streamable_http.py | 8 ++++-- tests/shared/test_streamable_http.py | 38 +++++++++++++++++++++++++++- 2 files changed, 43 insertions(+), 3 deletions(-) diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index 12385da42a..acda1caa4f 100644 --- a/src/mcp/client/streamable_http.py +++ b/src/mcp/client/streamable_http.py @@ -542,8 +542,12 @@ async def _send_http_status_error( reply = JSONRPCError(jsonrpc="2.0", id=request_id, error=parsed.root.error) await read_stream_writer.send(SessionMessage(JSONRPCMessage(reply))) return - except (httpx.StreamError, ValidationError): - logger.debug("Non-2xx body was not a JSON-RPC error; using the status-derived fallback") + # `httpx.HTTPError` covers reading the body as much as parsing it: a stalled or + # truncated error body raises `ReadTimeout`/`RemoteProtocolError` here, and letting + # that escape would strand the caller exactly as the bug this method exists to fix. + # `StreamError` is not one of its subclasses (it derives from `RuntimeError`). + except (httpx.HTTPError, httpx.StreamError, ValidationError): + logger.debug("Could not read a JSON-RPC error from the non-2xx body; using the fallback") error_data = ErrorData(code=INTERNAL_ERROR, message="Server returned an error response") jsonrpc_error = JSONRPCError(jsonrpc="2.0", id=request_id, error=error_data) diff --git a/tests/shared/test_streamable_http.py b/tests/shared/test_streamable_http.py index f230616ef5..b04b152ad2 100644 --- a/tests/shared/test_streamable_http.py +++ b/tests/shared/test_streamable_http.py @@ -8,7 +8,7 @@ import multiprocessing import socket import time -from collections.abc import Generator +from collections.abc import AsyncIterator, Generator from datetime import timedelta from typing import Any from unittest.mock import MagicMock @@ -2543,3 +2543,39 @@ def handler(request: httpx.Request) -> httpx.Response: assert reply.message.root.error == types.ErrorData( code=types.INTERNAL_ERROR, message="Server returned an error response" ) + + +@pytest.mark.anyio +async def test_a_non_2xx_body_that_fails_to_read_still_resolves_the_caller() -> None: + """A stalled or truncated error body must not strand the caller. + + `aread()` on the non-2xx body raises `ReadTimeout`/`RemoteProtocolError` when the connection + dies mid-body. Those derive from `httpx.HTTPError`, not `httpx.StreamError`, so letting the + body-parsing `except` miss them would put the escaping exception right back into the POST's + background task — the very bug this path exists to fix. + """ + + class _DyingStream(httpx.AsyncByteStream): + async def __aiter__(self) -> AsyncIterator[bytes]: + raise httpx.ReadTimeout("connection died mid-body") + yield b"" # pragma: no cover + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(500, headers={"content-type": "application/json"}, stream=_DyingStream()) + + with anyio.fail_after(5): + async with ( + httpx.AsyncClient(transport=httpx.MockTransport(handler)) as http_client, + streamable_http_client("http://test/mcp", http_client=http_client) as (read, write, _), + read, + write, + ): + request = types.JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/list", params={}) + await write.send(SessionMessage(types.JSONRPCMessage(request))) + reply = await read.receive() + + assert isinstance(reply, SessionMessage) + assert isinstance(reply.message.root, types.JSONRPCError) + assert reply.message.root.error == types.ErrorData( + code=types.INTERNAL_ERROR, message="Server returned an error response" + )