refactor(mcp): one traversal and one carrier choice-point for upstream listing failures

Both review findings shared one root cause: two exception-tree walkers with drifted semantics.
_extract_upstream_auth_failure walked the incidental __context__ chain before explicit causes, so a
403 raised while handling the causal 401 could shadow it; and the generic _get_tools_from_server arm
classified without extracting the challenge, so a nested 401 at client-build time surfaced without
the WWW-Authenticate the client needs. upstream_auth_challenge and raise_classified_list_failure in
faults/list_outcomes.py are now the single traversal and the single choice-point; both fetch arms
and _extract_upstream_auth_failure (also serving tool calls and the connect-time probe) delegate to
them, with dcr_bridge challenge suppression as a parameter so it holds on every path. The stale
_fetch_tools_with_timeout docstring describing the pre-change 403 absorb is rewritten to the actual
contract: 403 relays with its own status, an upstream-sent challenge relays verbatim per RFC 6750
insufficient_scope, and a challenge is only ever fabricated for a challenge-less 401

litellm/proxy/_experimental/mcp_server/faults/list_outcomes.py

@@ -10,7 +10,7 @@ becomes an outcome, never a second failure.
 
 from __future__ import annotations
 
-from typing import Literal, NamedTuple, TypeAlias
+from typing import Literal, NamedTuple, NoReturn, TypeAlias
 
 import httpx
 from mcp.types import Tool as MCPTool
@@ -87,6 +87,42 @@ def _find_upstream_response(exc: BaseException) -> httpx.Response | None:
     return None
 
 
+def upstream_auth_challenge(exc: BaseException) -> tuple[int, str | None] | None:
+    """The upstream 401/403 and its ``WWW-Authenticate`` challenge, both read from the SAME response
+    the deliberate-order traversal selects, so the status that picks the carrier channel and the
+    challenge that rides with it can never come from two different responses in the tree."""
+    response = _find_upstream_response(exc)
+    if response is None or response.status_code not in (401, 403):
+        return None
+    try:
+        challenge = response.headers.get("www-authenticate")
+    except Exception:
+        challenge = None
+    return response.status_code, challenge
+
+
+def raise_classified_list_failure(
+    exc: BaseException,
+    server_name: str,
+    suppress_challenge: bool = False,
+) -> NoReturn:
+    """The one place a failed server fetch chooses its carrier: an upstream 401/403 travels as
+    ``MCPUpstreamAuthError`` with the upstream's own challenge preserved (a challenge is only ever
+    fabricated at the HTTP edge, and only for a 401), everything else as ``MCPServerListError`` with
+    a classified fault. Every fetch site delegates here so the two channels cannot drift apart per
+    call site. ``suppress_challenge`` is for dcr_bridge servers, whose upstream challenge points
+    clients at the wrong protected-resource metadata and must never relay."""
+    auth = upstream_auth_challenge(exc)
+    if auth is not None:
+        status_code, challenge = auth
+        raise MCPUpstreamAuthError(
+            status_code=status_code,
+            www_authenticate=None if suppress_challenge else challenge,
+            server_name=server_name,
+        ) from exc
+    raise MCPServerListError(classify_list_exception(exc), server_name) from exc
+
+
 def classify_list_exception(exc: BaseException) -> ServerListFault:
     """Classify a per-server listing failure into exactly one outcome. Total: an exception this
     function cannot recognize is the gateway's own fault (``internal``), never a re-raise."""

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -56,7 +56,8 @@ from litellm.proxy._experimental.mcp_server.exceptions import (
 )
 from litellm.proxy._experimental.mcp_server.faults.list_outcomes import (
     ServerListFault,
-    classify_list_exception,
+    raise_classified_list_failure,
+    upstream_auth_challenge,
 )
 from litellm.proxy._experimental.mcp_server.elicitation_handler import (
     MCP_ELICITATION_AVAILABLE,
@@ -470,49 +471,14 @@ def _caller_authorization_fans_out(
 def _extract_upstream_auth_failure(
     exc: BaseException,
 ) -> Optional[tuple[int, Optional[str]]]:
-    """Walk the exception tree looking for an HTTP 401/403 response from the
-    upstream MCP server.
+    """The upstream 401/403 and its ``WWW-Authenticate`` header from the exception tree, or ``None``.
 
-    The MCP SDK wraps transport errors in anyio ``ExceptionGroup`` objects and
-    may chain through ``__cause__`` / ``__context__``. We inspect all of those
-    layers for an ``httpx.Response``-bearing exception (typically
-    ``httpx.HTTPStatusError``) and extract the status code and any upstream
-    ``WWW-Authenticate`` header.
-
-    Returns ``(status_code, www_authenticate)`` on match, else ``None``.
-    """
-    seen: set[int] = set()
-    stack: list[BaseException] = [exc]
-    while stack:
-        current = stack.pop()
-        if id(current) in seen:
-            continue
-        seen.add(id(current))
-
-        response = getattr(current, "response", None)
-        if response is not None:
-            status_code = getattr(response, "status_code", None)
-            if isinstance(status_code, int) and status_code in (401, 403):
-                www_authenticate: Optional[str] = None
-                headers = getattr(response, "headers", None)
-                if headers is not None:
-                    try:
-                        www_authenticate = headers.get("www-authenticate")
-                    except Exception:
-                        www_authenticate = None
-                return status_code, www_authenticate
-
-        # anyio / PEP 654 ExceptionGroup
-        sub_exceptions = getattr(current, "exceptions", None)
-        if sub_exceptions:
-            stack.extend(sub_exceptions)
-
-        if current.__cause__ is not None:
-            stack.append(current.__cause__)
-        if current.__context__ is not None and current.__context__ is not current.__cause__:
-            stack.append(current.__context__)
-
-    return None
+    Delegates to the shared traversal in ``faults.list_outcomes`` so every consumer (tool listing,
+    tool calls, the connect-time probe) selects the same response with the same deliberate order:
+    explicit ``raise ... from`` causes first, ExceptionGroup members in raise order, the incidental
+    ``__context__`` chain last. A response raised while handling the real failure can therefore never
+    shadow the causal one."""
+    return upstream_auth_challenge(exc)
 
 
 def _warn_on_server_name_fields(
@@ -2779,7 +2745,7 @@ class MCPServerManager:
             raise
         except Exception as e:
             verbose_logger.warning(f"Failed to get tools from server {server.name}: {str(e)}")
-            raise MCPServerListError(classify_list_exception(e), server.name) from e
+            raise_classified_list_failure(e, server.name, suppress_challenge=server.is_dcr_bridge)
 
     async def get_prompts_from_server(
         self,
@@ -3365,25 +3331,24 @@ class MCPServerManager:
         Uses anyio.fail_after() instead of asyncio.wait_for() to avoid conflicts
         with the MCP SDK's anyio TaskGroup. See GitHub issue #20715 for details.
 
-        An upstream HTTP 401 is converted into :class:`MCPUpstreamAuthError`
-        instead of being swallowed to an empty tool list, regardless of the
-        server's auth_type. Callers route it by surface: the single-server HTTP
-        routes turn it into a 401 + ``WWW-Authenticate`` challenge so standards-
-        compliant MCP clients trigger the upstream OAuth flow, while the
-        multi-server ``/mcp`` aggregator absorbs it to an empty list so one
-        unauthenticated server doesn't fail the whole listing. Only a 401
-        (missing/invalid credential) drives the re-auth challenge; a 403
-        (authenticated but forbidden, e.g. insufficient scope) is not a re-auth
-        signal and, like other non-auth errors, returns an empty list.
+        Failures never return an empty tool list. An upstream 401 or 403 raises
+        :class:`MCPUpstreamAuthError` carrying the upstream's own
+        ``WWW-Authenticate`` challenge when one was sent (a challenge is only
+        ever fabricated at the HTTP edge, and only for a 401: a 403 means the
+        caller is authenticated but not allowed, so prompting re-auth would be
+        wrong, while an upstream-sent 403 challenge is the RFC 6750
+        insufficient_scope step-up and relays verbatim). Every other failure
+        raises :class:`MCPServerListError` with a classified fault. Each
+        boundary then applies its own policy: single-server routes relay the
+        truthful status, the multi-server aggregator absorbs the failure into
+        that server's listing outcome.
 
         Args:
             client: MCP client instance
             server_name: Name of the server for logging
 
         Returns:
-            List of tools from the server. Failures never return an empty list: an upstream 401/403
-            raises MCPUpstreamAuthError and everything else raises MCPServerListError carrying a
-            classified fault, so each boundary applies its own absorb-or-relay policy.
+            List of tools from the server
         """
         try:
             with anyio.fail_after(MCP_TOOL_LISTING_TIMEOUT):
@@ -3400,17 +3365,8 @@ class MCPServerManager:
             verbose_logger.warning(f"Connection error while listing tools from {server_name}: {str(e)}")
             raise MCPServerListError(ServerListFault(tag="unreachable"), server_name) from e
         except Exception as e:
-            auth_info = _extract_upstream_auth_failure(e)
-            if auth_info is not None and auth_info[0] in (401, 403):
-                status_code, www_authenticate = auth_info
-                verbose_logger.info(f"Upstream auth failure from MCP server {server_name}: HTTP {status_code}")
-                raise MCPUpstreamAuthError(
-                    status_code=status_code,
-                    www_authenticate=www_authenticate,
-                    server_name=server_name,
-                ) from e
             verbose_logger.warning(f"Error listing tools from {server_name}: {str(e)}")
-            raise MCPServerListError(classify_list_exception(e), server_name) from e
+            raise_classified_list_failure(e, server_name)
 
     _SHORT_PREFIX_MAX_REHASH_ATTEMPTS = 1024
 

tests/test_litellm/proxy/_experimental/mcp_server/faults/test_list_outcomes.py

@@ -119,3 +119,58 @@ async def test_cancelled_fetch_is_a_classified_fault_not_a_healthy_empty_server(
         await manager._fetch_tools_with_timeout(client, "cancelled_srv")
 
     assert exc_info.value.fault.tag == "internal"
+
+
+def test_auth_challenge_and_status_come_from_the_causal_response():
+    """An incidental 403 raised while handling the causal 401 (context chain) must not shadow it:
+    the carrier channel and the challenge both derive from the response on the explicit causal
+    chain, so the caller is challenged to authenticate rather than told it is forbidden."""
+    from litellm.proxy._experimental.mcp_server.faults.list_outcomes import upstream_auth_challenge
+
+    causal = httpx.HTTPStatusError(
+        "auth",
+        request=httpx.Request("POST", "https://mcp.example.com/mcp"),
+        response=httpx.Response(
+            401,
+            headers={"www-authenticate": 'Bearer resource_metadata="https://mcp.example.com/.well-known"'},
+            request=httpx.Request("POST", "https://mcp.example.com/mcp"),
+        ),
+    )
+    incidental = httpx.HTTPStatusError(
+        "hook",
+        request=httpx.Request("POST", "https://hook.example.com/log"),
+        response=httpx.Response(403, request=httpx.Request("POST", "https://hook.example.com/log")),
+    )
+    wrapper = RuntimeError("fetch failed")
+    wrapper.__cause__ = causal
+    wrapper.__context__ = incidental
+
+    result = upstream_auth_challenge(wrapper)
+    assert result is not None
+    status_code, challenge = result
+    assert status_code == 401
+    assert challenge == 'Bearer resource_metadata="https://mcp.example.com/.well-known"'
+
+
+def test_raise_classified_list_failure_routes_auth_to_upstream_auth_error():
+    """The single choice-point sends 401/403 through MCPUpstreamAuthError with the upstream's own
+    challenge and everything else through MCPServerListError, so fetch sites cannot drift."""
+    from litellm.proxy._experimental.mcp_server.faults.list_outcomes import raise_classified_list_failure
+
+    auth_exc = httpx.HTTPStatusError(
+        "auth",
+        request=httpx.Request("POST", "https://mcp.example.com/mcp"),
+        response=httpx.Response(
+            401,
+            headers={"www-authenticate": "Bearer realm=x"},
+            request=httpx.Request("POST", "https://mcp.example.com/mcp"),
+        ),
+    )
+    with pytest.raises(MCPUpstreamAuthError) as auth_info:
+        raise_classified_list_failure(auth_exc, "srv")
+    assert auth_info.value.status_code == 401
+    assert auth_info.value.www_authenticate == "Bearer realm=x"
+
+    with pytest.raises(MCPServerListError) as fault_info:
+        raise_classified_list_failure(RuntimeError("boom"), "srv")
+    assert fault_info.value.fault.tag == "internal"

tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py

@@ -6979,6 +6979,78 @@ class TestMCPToolsListAuthSurfacing:
         assert exc_info.value.fault == ServerListFault(tag="internal", status_code=500)
         assert exc_info.value.server_name == "stdio-srv"
 
+    @pytest.mark.asyncio
+    async def test_get_tools_from_server_generic_arm_extracts_nested_auth_challenge(self):
+        """A 401 buried in the exception tree at client-build time must travel the same channel as
+        one raised during the fetch: MCPUpstreamAuthError with the upstream's own challenge. Before
+        the shared choice-point it classified into a challenge-less fault, so single-server routes
+        answered 401 without the WWW-Authenticate the client needs to start the OAuth flow."""
+        import httpx
+
+        from litellm.proxy._experimental.mcp_server.exceptions import (
+            MCPUpstreamAuthError,
+        )
+
+        manager = MCPServerManager()
+        server = MCPServer(server_id="nested-srv", name="nested-srv", transport=MCPTransport.http)
+        causal = httpx.HTTPStatusError(
+            "auth",
+            request=httpx.Request("POST", "https://mcp.example.com/mcp"),
+            response=httpx.Response(
+                401,
+                headers={"www-authenticate": "Bearer realm=upstream"},
+                request=httpx.Request("POST", "https://mcp.example.com/mcp"),
+            ),
+        )
+        wrapper = RuntimeError("client build failed")
+        wrapper.__cause__ = causal
+        manager._create_mcp_client = AsyncMock(side_effect=wrapper)
+
+        with pytest.raises(MCPUpstreamAuthError) as exc_info:
+            await manager._get_tools_from_server(server)
+
+        assert exc_info.value.status_code == 401
+        assert exc_info.value.www_authenticate == "Bearer realm=upstream"
+
+    @pytest.mark.asyncio
+    async def test_get_tools_from_server_generic_arm_strips_challenge_for_dcr_bridge(self):
+        """The dcr_bridge challenge suppression must hold on the generic arm too, not only when the
+        fetch itself raised MCPUpstreamAuthError: a bridge client following the upstream challenge
+        would fail the RFC 9728 resource match against the gateway URL it dialed."""
+        import httpx
+
+        from litellm.proxy._experimental.mcp_server.exceptions import (
+            MCPUpstreamAuthError,
+        )
+        from litellm.types.mcp import MCPAuth
+
+        manager = MCPServerManager()
+        bridge_server = MCPServer(
+            server_id="bridge-nested",
+            name="bridge-nested",
+            transport=MCPTransport.http,
+            auth_type=MCPAuth.true_passthrough,
+            dcr_bridge=True,
+        )
+        causal = httpx.HTTPStatusError(
+            "auth",
+            request=httpx.Request("POST", "https://upstream.example/mcp"),
+            response=httpx.Response(
+                401,
+                headers={"www-authenticate": 'Bearer resource_metadata="https://upstream.example/.wk"'},
+                request=httpx.Request("POST", "https://upstream.example/mcp"),
+            ),
+        )
+        wrapper = RuntimeError("client build failed")
+        wrapper.__cause__ = causal
+        manager._create_mcp_client = AsyncMock(side_effect=wrapper)
+
+        with pytest.raises(MCPUpstreamAuthError) as exc_info:
+            await manager._get_tools_from_server(bridge_server)
+
+        assert exc_info.value.status_code == 401
+        assert exc_info.value.www_authenticate is None
+
     @pytest.mark.asyncio
     async def test_get_tools_from_server_suppresses_upstream_challenge_for_dcr_bridge(self):
         """A dcr_bridge server must never relay the upstream's own WWW-Authenticate: it points