fix: handle HTTP 4xx/5xx gracefully in StreamableHTTP transport

Replace raise_for_status() calls with graceful status code checks
in GET SSE paths, and include HTTP status code in POST error
messages for better client-side error handling.

Changes:
- POST error message now includes HTTP status code (e.g. 'Server
  returned HTTP 401') instead of generic 'Server returned an error
  response'
- GET SSE listener: check status code and retry instead of crashing
- Resumption GET: return JSONRPC error with status code to client
- Reconnection GET: return JSONRPC error with status code to client

Fixes #1295

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: Varun Sharma

src/mcp/client/streamable_http.py

@@ -194,7 +194,10 @@ async def handle_get_stream(self, client: httpx.AsyncClient, read_stream_writer:
                     headers[LAST_EVENT_ID] = last_event_id
 
                 async with aconnect_sse(client, "GET", self.url, headers=headers) as event_source:
-                    event_source.response.raise_for_status()
+                    if event_source.response.status_code >= 400:
+                        logger.warning(f"GET SSE returned HTTP {event_source.response.status_code}")
+                        attempt += 1
+                        continue
                     logger.debug("GET SSE connection established")
 
                     async for sse in event_source.aiter_sse():
@@ -237,7 +240,16 @@ async def _handle_resumption_request(self, ctx: RequestContext) -> None:
             original_request_id = ctx.session_message.message.id
 
         async with aconnect_sse(ctx.client, "GET", self.url, headers=headers) as event_source:
-            event_source.response.raise_for_status()
+            if event_source.response.status_code >= 400:
+                logger.warning(f"Resumption GET returned HTTP {event_source.response.status_code}")
+                if original_request_id is not None:
+                    error_data = ErrorData(
+                        code=INTERNAL_ERROR,
+                        message=f"Server returned HTTP {event_source.response.status_code}",
+                    )
+                    error_msg = SessionMessage(JSONRPCError(jsonrpc="2.0", id=original_request_id, error=error_data))
+                    await ctx.read_stream_writer.send(error_msg)
+                return
             logger.debug("Resumption GET SSE connection established")
 
             async for sse in event_source.aiter_sse():  # pragma: no branch
@@ -276,7 +288,10 @@ async def _handle_post_request(self, ctx: RequestContext) -> None:
 
             if response.status_code >= 400:
                 if isinstance(message, JSONRPCRequest):
-                    error_data = ErrorData(code=INTERNAL_ERROR, message="Server returned an error response")
+                    error_data = ErrorData(
+                        code=INTERNAL_ERROR,
+                        message=f"Server returned HTTP {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
@@ -398,7 +413,18 @@ async def _handle_reconnection(
 
         try:
             async with aconnect_sse(ctx.client, "GET", self.url, headers=headers) as event_source:
-                event_source.response.raise_for_status()
+                if event_source.response.status_code >= 400:
+                    logger.warning(f"Reconnection GET returned HTTP {event_source.response.status_code}")
+                    if original_request_id is not None:
+                        error_data = ErrorData(
+                            code=INTERNAL_ERROR,
+                            message=f"Server returned HTTP {event_source.response.status_code}",
+                        )
+                        error_msg = SessionMessage(
+                            JSONRPCError(jsonrpc="2.0", id=original_request_id, error=error_data)
+                        )
+                        await ctx.read_stream_writer.send(error_msg)
+                    return
                 logger.info("Reconnected to SSE stream")
 
                 # Track for potential further reconnection

tests/client/test_notification_response.py

@@ -148,7 +148,7 @@ async def test_http_error_status_sends_jsonrpc_error() -> None:
             async with ClientSession(read_stream, write_stream) as session:  # pragma: no branch
                 await session.initialize()
 
-                with pytest.raises(MCPError, match="Server returned an error response"):  # pragma: no branch
+                with pytest.raises(MCPError, match="Server returned HTTP 500"):  # pragma: no branch
                     await session.list_tools()
 
 

tests/client/test_streamable_http.py

@@ -0,0 +1,102 @@
+"""Tests for StreamableHTTP client transport HTTP error handling.
+
+Verifies that HTTP 4xx/5xx responses are handled gracefully
+instead of crashing the program.
+"""
+
+import json
+
+import httpx
+import pytest
+from starlette.applications import Starlette
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response
+from starlette.routing import Route
+
+from mcp import ClientSession, types
+from mcp.client.streamable_http import streamable_http_client
+from mcp.shared.session import RequestResponder
+
+pytestmark = pytest.mark.anyio
+
+INIT_RESPONSE = {
+    "serverInfo": {"name": "test-http-error-server", "version": "1.0.0"},
+    "protocolVersion": "2024-11-05",
+    "capabilities": {},
+}
+
+
+def _create_401_server_app() -> Starlette:
+    """Create a server that returns 401 for non-init requests."""
+
+    async def handle_mcp_request(request: Request) -> Response:
+        body = await request.body()
+        data = json.loads(body)
+
+        if data.get("method") == "initialize":
+            return JSONResponse({"jsonrpc": "2.0", "id": data["id"], "result": INIT_RESPONSE})
+
+        if "id" not in data:
+            return Response(status_code=202)
+
+        return Response(status_code=401)
+
+    return Starlette(debug=True, routes=[Route("/mcp", handle_mcp_request, methods=["POST"])])
+
+
+async def test_http_401_returns_jsonrpc_error() -> None:
+    """Test that a 401 response returns a JSONRPC error instead of crashing.
+
+    Regression test for https://github.com/modelcontextprotocol/python-sdk/issues/1295
+    """
+    returned_exception = None
+
+    async def message_handler(
+        message: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception,
+    ) -> None:
+        nonlocal returned_exception
+        if isinstance(message, Exception):  # pragma: no cover
+            returned_exception = message
+
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=_create_401_server_app())) as client:
+        async with streamable_http_client("http://localhost/mcp", http_client=client) as (read_stream, write_stream):
+            async with ClientSession(read_stream, write_stream, message_handler=message_handler) as session:
+                await session.initialize()
+
+                # list_tools should get a JSONRPC error with HTTP status, not crash
+                with pytest.raises(Exception) as exc_info:
+                    await session.list_tools()
+                assert "401" in str(exc_info.value)
+
+    if returned_exception:  # pragma: no cover
+        pytest.fail(f"Unexpected exception: {returned_exception}")
+
+
+def _create_503_server_app() -> Starlette:
+    """Create a server that returns 503 for non-init requests."""
+
+    async def handle_mcp_request(request: Request) -> Response:
+        body = await request.body()
+        data = json.loads(body)
+
+        if data.get("method") == "initialize":
+            return JSONResponse({"jsonrpc": "2.0", "id": data["id"], "result": INIT_RESPONSE})
+
+        if "id" not in data:
+            return Response(status_code=202)
+
+        return Response(status_code=503)
+
+    return Starlette(debug=True, routes=[Route("/mcp", handle_mcp_request, methods=["POST"])])
+
+
+async def test_http_503_returns_jsonrpc_error() -> None:
+    """Test that a 503 response returns a JSONRPC error instead of crashing."""
+    async with httpx.AsyncClient(transport=httpx.ASGITransport(app=_create_503_server_app())) as client:
+        async with streamable_http_client("http://localhost/mcp", http_client=client) as (read_stream, write_stream):
+            async with ClientSession(read_stream, write_stream) as session:
+                await session.initialize()
+
+                with pytest.raises(Exception) as exc_info:
+                    await session.list_tools()
+                assert "503" in str(exc_info.value)