fix(client): map bare HTTP 401/403 to distinguishable JSON-RPC errors

src/mcp/client/streamable_http.py

@@ -366,6 +366,21 @@ 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=INTERNAL_ERROR,
+                            message="Unauthorized",
+                            data={"http_status": 401},
+                        )
+                    elif response.status_code == 403:
+                        error_data = ErrorData(
+                            code=INTERNAL_ERROR,
+                            message="Forbidden",
+                            data={"http_status": 403},
+                        )
                     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.INTERNAL_ERROR
+                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

@@ -19,6 +19,7 @@
     CLIENT_CAPABILITIES_META_KEY,
     CLIENT_INFO_META_KEY,
     CONNECTION_CLOSED,
+    INTERNAL_ERROR,
     INVALID_REQUEST,
     METHOD_NOT_FOUND,
     PROTOCOL_VERSION_META_KEY,
@@ -132,6 +133,40 @@ def handler(request: httpx2.Request) -> httpx2.Response:
     assert reply.message.error.code == METHOD_NOT_FOUND
 
 
+@pytest.mark.anyio
+@pytest.mark.parametrize(
+    ("status", "message"),
+    [
+        (401, "Unauthorized"),
+        (403, "Forbidden"),
+    ],
+)
+async def test_bare_auth_http_error_maps_to_distinguishable_jsonrpc_error(status: int, message: str) -> None:
+    """Bare HTTP 401/403 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(status)
+
+    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 == INTERNAL_ERROR
+    assert reply.message.error.message == message
+    assert reply.message.error.data == {"http_status": status}
+
+
 @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.