From 85d6b9741affc0f9ab153c437b12408ab42bab7b Mon Sep 17 00:00:00 2001 From: Kush Zingade Date: Tue, 25 Aug 2026 02:20:38 +0800 Subject: [PATCH] fix(client): surface non-2xx POST failures to the awaiting caller (#2110) The SSE transport's post_writer caught every exception from the message POST, logged it, and returned. `raise_for_status()` on a 401/403/5xx was therefore swallowed: the request it carried never received a response and its caller waited forever. Fail that request instead, correlated by its own id, so only the caller whose POST was rejected is affected -- a session-wide fan-out would tear down unrelated in-flight requests, and raw `Exception` items on the read stream are also used for non-fatal per-message parse errors. Both transports now also carry the HTTP status on `ErrorData.data`. The JSON-RPC code cannot express it: every non-404 failure maps to INTERNAL_ERROR, which alone cannot tell a caller whether to re-authenticate (401/403) or retry (5xx). Error messages are unchanged. Signed-off-by: Kush Zingade --- src/mcp/client/sse.py | 46 ++++++++++++++++++++++------ src/mcp/client/streamable_http.py | 8 ++++- tests/client/test_streamable_http.py | 25 +++++++++++++++ tests/shared/test_sse.py | 39 +++++++++++++++++++++++ 4 files changed, 108 insertions(+), 10 deletions(-) diff --git a/src/mcp/client/sse.py b/src/mcp/client/sse.py index 31d0f35391..56cf16d55d 100644 --- a/src/mcp/client/sse.py +++ b/src/mcp/client/sse.py @@ -115,21 +115,49 @@ async def sse_reader(task_status: TaskStatus[str] = anyio.TASK_STATUS_IGNORED): finally: await read_stream_writer.aclose() + async def _fail_request(message: types.JSONRPCMessage, exc: httpx2.HTTPError) -> None: + """Resolve the request this failed POST carried, so its caller raises instead of hanging. + + Correlated by request id rather than fanned out across the session: + one rejected POST must not tear down unrelated in-flight requests. + A notification has no waiter, so there is nothing to resolve. + """ + if not isinstance(message, types.JSONRPCRequest): + return + status = exc.response.status_code if isinstance(exc, httpx2.HTTPStatusError) else None + error = types.ErrorData( + code=types.INTERNAL_ERROR, + message=f"Sending the request failed: {exc}", + # The HTTP status is the whole diagnosis for 401/403 vs 5xx, and + # the JSON-RPC code cannot carry it. + data={"http_status": status} if status is not None else None, + ) + reply = SessionMessage(types.JSONRPCError(jsonrpc="2.0", id=message.id, error=error)) + try: + await read_stream_writer.send(reply) + except (anyio.BrokenResourceError, anyio.ClosedResourceError): # pragma: no cover + logger.debug("read stream closed before request %r could be failed", message.id) + async def post_writer(endpoint_url: str): try: async with write_stream_reader, write_stream: async def _send_message(session_message: SessionMessage) -> None: logger.debug(f"Sending client message: {session_message}") - response = await client.post( - endpoint_url, - json=session_message.message.model_dump( - by_alias=True, - mode="json", - exclude_unset=True, - ), - ) - response.raise_for_status() + try: + response = await client.post( + endpoint_url, + json=session_message.message.model_dump( + by_alias=True, + mode="json", + exclude_unset=True, + ), + ) + response.raise_for_status() + except httpx2.HTTPError as exc: + logger.exception("Error sending client message") + await _fail_request(session_message.message, exc) + return logger.debug(f"Client message sent successfully: {response.status_code}") async for session_message in write_stream_reader: diff --git a/src/mcp/client/streamable_http.py b/src/mcp/client/streamable_http.py index 226b0fecf9..269892e703 100644 --- a/src/mcp/client/streamable_http.py +++ b/src/mcp/client/streamable_http.py @@ -367,7 +367,13 @@ async def _handle_post_request(self, ctx: RequestContext) -> None: else: error_data = ErrorData(code=INVALID_REQUEST, message="Session terminated") else: - error_data = ErrorData(code=INTERNAL_ERROR, message="Server returned an error response") + # The status rides on `data`: INTERNAL_ERROR alone can't tell a + # caller whether to re-authenticate (401/403) or retry (5xx). + error_data = ErrorData( + code=INTERNAL_ERROR, + message="Server returned an error response", + data={"http_status": response.status_code}, + ) session_message = SessionMessage(JSONRPCError(jsonrpc="2.0", id=message.id, error=error_data)) await ctx.read_stream_writer.send(session_message) return diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py index d21f520daf..eb43a9798e 100644 --- a/tests/client/test_streamable_http.py +++ b/tests/client/test_streamable_http.py @@ -748,3 +748,28 @@ async def test_resolving_an_abandoned_request_after_the_reader_closed_is_contain _abandoned_request_context(http, send), "evt-7", None, MAX_RECONNECTION_ATTEMPTS ) send.close() + + +@pytest.mark.anyio +@pytest.mark.parametrize("status_code", [401, 403, 500, 502]) +async def test_non_2xx_post_preserves_http_status(status_code: int) -> None: + """The HTTP status survives on `ErrorData.data` so 401/403 stay distinguishable from 5xx. + + The JSON-RPC code cannot carry it: every non-404 failure maps to INTERNAL_ERROR, + which on its own says nothing about whether the caller should re-authenticate. + """ + + def handler(request: httpx2.Request) -> httpx2.Response: + return httpx2.Response(status_code) + + 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=1, method="tools/list", params={}))) + reply = await read.receive() + + assert isinstance(reply, SessionMessage) + assert isinstance(reply.message, JSONRPCError) + assert reply.message.error.data == {"http_status": status_code} diff --git a/tests/shared/test_sse.py b/tests/shared/test_sse.py index c27dd69db3..3d3e262d61 100644 --- a/tests/shared/test_sse.py +++ b/tests/shared/test_sse.py @@ -480,3 +480,42 @@ async def test_sse_session_cleanup_on_disconnect() -> None: headers={"Content-Type": "application/json"}, ) assert response.status_code == 404 + + +def make_post_rejecting_app(status_code: int) -> Starlette: + """SSE transport whose message POST endpoint always fails with `status_code`.""" + sse = SseServerTransport( + "/messages/", security_settings=TransportSecuritySettings(enable_dns_rebinding_protection=False) + ) + server = Server(SERVER_NAME, on_read_resource=_handle_read_resource) + + async def handle_sse(request: Request) -> Response: + async with sse.connect_sse(request.scope, request.receive, request._send) as (read_stream, write_stream): + await server.run(read_stream, write_stream, server.create_initialization_options()) + return Response() + + async def reject_post(request: Request) -> Response: + return Response(status_code=status_code) + + return Starlette( + routes=[ + Route("/sse", endpoint=handle_sse), + Route("/messages/", endpoint=reject_post, methods=["POST"]), + ] + ) + + +@pytest.mark.anyio +@pytest.mark.parametrize("status_code", [401, 403, 500]) +async def test_sse_post_error_reaches_caller(status_code: int) -> None: + """A non-2xx POST surfaces to the awaiting caller instead of hanging it forever.""" + factory = in_process_client_factory(make_post_rejecting_app(status_code)) + + with anyio.fail_after(5): + async with sse_client(f"{BASE_URL}/sse", httpx_client_factory=factory) as streams: + async with ClientSession(*streams) as session: + with pytest.raises(MCPError) as exc_info: + await session.initialize() + + # The HTTP status survives on `data` — 401/403 must stay distinguishable from 5xx. + assert exc_info.value.error.data == {"http_status": status_code}