fix(client): map HTTP 401 on streamable HTTP to Unauthorized JSON-RPC error

Operation-specific auth denials now surface as a distinguishable
INVALID_REQUEST / "Unauthorized" JSON-RPC error with data.http_status
instead of collapsing into the generic "Server returned an error response"
fallback.

Fixes #1295

Author: mturac

src/mcp/client/streamable_http.py

@@ -366,6 +366,15 @@ async def _handle_post_request(self, ctx: RequestContext) -> None:
                             error_data = ErrorData(code=METHOD_NOT_FOUND, message="Not Found")
                         else:
                             error_data = ErrorData(code=INVALID_REQUEST, message="Session terminated")
+                    elif response.status_code == 401:
+                        # Operation-specific auth denials must stay distinguishable so
+                        # agents can handle them (issue #1295) instead of collapsing into
+                        # an opaque "Server returned an error response".
+                        error_data = ErrorData(
+                            code=INVALID_REQUEST,
+                            message="Unauthorized",
+                            data={"http_status": 401},
+                        )
                     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))

tests/client/test_notification_response.py

@@ -151,6 +151,23 @@ async def test_http_error_status_sends_jsonrpc_error() -> None:
                     await session.list_tools()
 
 
+async def test_http_401_surfaces_unauthorized_to_session() -> None:
+    """Bare HTTP 401 after initialize must surface as Unauthorized (issue #1295).
+
+    Agents need a distinguishable auth denial for operation-specific 401s, not the
+    generic transport fallback string used for other 4xx/5xx statuses.
+    """
+    async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=_create_http_error_app(401))) 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:  # pragma: no branch
+                await session.initialize()
+
+                with pytest.raises(MCPError, match="Unauthorized") as exc:  # pragma: no branch
+                    await session.list_tools()
+                assert exc.value.error.code == types.INVALID_REQUEST
+                assert exc.value.error.data == {"http_status": 401}
+
+
 async def test_http_error_on_notification_does_not_hang() -> None:
     """Verify HTTP errors on notifications are silently ignored.
 

tests/client/test_streamable_http.py

@@ -132,6 +132,33 @@ def handler(request: httpx2.Request) -> httpx2.Response:
     assert reply.message.error.code == METHOD_NOT_FOUND
 
 
+@pytest.mark.anyio
+async def test_bare_401_maps_to_unauthorized_jsonrpc_error() -> None:
+    """Bare HTTP 401 must reach the caller as a correlated, distinguishable JSON-RPC error.
+
+    Authorization failures can be operation-specific (issue #1295). Collapsing them into the
+    generic "Server returned an error response" fallback prevents agents from handling the
+    denial without tearing down the whole session.
+    """
+
+    def handler(request: httpx2.Request) -> httpx2.Response:
+        return httpx2.Response(401)
+
+    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/call", params={})))
+            reply = await read.receive()
+    assert isinstance(reply, SessionMessage)
+    assert isinstance(reply.message, JSONRPCError)
+    assert reply.message.id == 1
+    assert reply.message.error.code == INVALID_REQUEST
+    assert reply.message.error.message == "Unauthorized"
+    assert reply.message.error.data == {"http_status": 401}
+
+
 @pytest.mark.anyio
 async def test_initialize_post_clears_cached_pv_header_and_unstamped_posts_read_it() -> None:
     """``initialize`` discards the cached protocol-version header; every other POST reads it.