feat(code): auto-detect MCP OAuth from 401 challenge (#4364)

MCP servers that require OAuth are now auto-detected from their 401
challenge, so `dcode mcp login <server>` works without manually setting
`auth: oauth`.

---

Previously a remote MCP server only triggered OAuth if the config
explicitly set `auth: oauth`; otherwise an auth-requiring server failed
with an opaque `error`. This detects the spec-compliant OAuth challenge
(HTTP 401 + `WWW-Authenticate`, RFC 9728) during tool discovery and
marks the server `unauthenticated` so the user is prompted to `/mcp
login`. Stored tokens now attach the OAuth provider regardless of the
`auth` field, unless the server config already provides a static
`Authorization` header, and `login` works for any http/sse server. We
deliberately do not infer OAuth from the URL.

Made by [Open
SWE](https://openswe.vercel.app/agents/9e33d490-9aba-d37d-d465-b198856cd367)

---------

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>

libs/code/deepagents_code/app.py

@@ -14169,11 +14169,14 @@ class DeepAgentsApp(App):
             )
             return
 
-        if selection.server_config.get("auth") != "oauth":
+        from deepagents_code.mcp_tools import _resolve_server_type
+
+        transport = _resolve_server_type(selection.server_config)
+        if transport not in {"http", "sse"}:
             await self._mount_message(
                 ErrorMessage(
-                    f"MCP server {server_name!r} does not use OAuth; "
-                    "nothing to log into.",
+                    f"MCP server {server_name!r} uses {transport!r} transport; "
+                    "OAuth login is only valid for http/sse.",
                 ),
             )
             return

libs/code/deepagents_code/mcp_auth.py

@@ -1605,6 +1605,84 @@ def find_reauth_required(exc: BaseException) -> MCPReauthRequiredError | None:
     return None
 
 
+_BEARER_SCHEME_RE = re.compile(r"(?:^|,)\s*bearer\b", re.IGNORECASE)
+"""Match a `Bearer` auth scheme at the start of a challenge or after a comma.
+
+A `WWW-Authenticate` line may list several schemes (RFC 7235); anchoring to
+the start or a preceding comma finds `Bearer` even when it isn't listed first.
+"""
+
+_RESOURCE_METADATA_RE = re.compile(
+    r'(?:^|[\s,])resource_metadata\s*=\s*"?([^",\s]+)',
+    re.IGNORECASE,
+)
+"""Capture the RFC 9728 `resource_metadata` URL from a Bearer challenge."""
+
+
+def _oauth_resource_challenge(headers: httpx.Headers) -> str | None:
+    """Return the RFC 9728 `resource_metadata` URL from a Bearer challenge.
+
+    A single `WWW-Authenticate` header line may carry several comma-separated
+    challenges (RFC 7235), and a response may repeat the header. Scan every
+    value for a `Bearer` scheme — anywhere in the line, not only first — that
+    advertises a `resource_metadata` parameter.
+
+    Args:
+        headers: Response headers to inspect.
+
+    Returns:
+        The `resource_metadata` URL when a Bearer challenge carries one,
+            else `None`.
+    """
+    for value in headers.get_list("www-authenticate"):
+        if _BEARER_SCHEME_RE.search(value) is None:
+            continue
+        match = _RESOURCE_METADATA_RE.search(value)
+        if match is not None:
+            return match.group(1)
+    return None
+
+
+def find_oauth_challenge(exc: BaseException) -> str | None:
+    """Return the `resource_metadata` URL of a 401 OAuth challenge in `exc`.
+
+    Per the MCP authorization spec (RFC 9728), a server requiring OAuth
+    answers an unauthenticated request with HTTP 401 plus a Bearer
+    `WWW-Authenticate` challenge pointing at its protected-resource metadata.
+    The MCP client surfaces that as an `httpx.HTTPStatusError`. Walks
+    `exceptions` (for `ExceptionGroup`), then `__cause__`/`__context__`,
+    tracking visited nodes to terminate on cyclic chains.
+
+    Args:
+        exc: Root exception to inspect.
+
+    Returns:
+        The `resource_metadata` URL when a 401 response carrying a Bearer
+            challenge is found, else `None`.
+    """
+    visited: set[int] = set()
+    stack: list[BaseException] = [exc]
+    while stack:
+        current = stack.pop()
+        if id(current) in visited:
+            continue
+        visited.add(id(current))
+        if isinstance(current, httpx.HTTPStatusError):
+            response = current.response
+            if (
+                response is not None and response.status_code == 401  # noqa: PLR2004  # HTTP Unauthorized
+            ):
+                challenge = _oauth_resource_challenge(response.headers)
+                if challenge is not None:
+                    return challenge
+        if isinstance(current, BaseExceptionGroup):
+            stack.extend(current.exceptions)
+        cause = current.__cause__ or current.__context__
+        if cause is not None:
+            stack.append(cause)
+    return None
+
+
 async def _drive_handshake(connections: dict) -> None:
     """Open a one-shot MCP session for `connections` to trigger OAuth handshake."""
     from langchain_mcp_adapters.client import MultiServerMCPClient
@@ -1630,7 +1708,7 @@ async def login(
             during the flow.
 
     Raises:
-        ValueError: If `server_config` isn't an OAuth http/sse server.
+        ValueError: If `server_config` isn't an http/sse server.
         RuntimeError: If header env-var interpolation fails, the device
             flow fails or times out, or the OAuth handshake aborts.
     """  # noqa: DOC502 - `RuntimeError` surfaces via `resolve_headers` / `_run_device_flow`
@@ -1639,15 +1717,12 @@ async def login(
         StreamableHttpConnection,
     )
 
-    if server_config.get("auth") != "oauth":
-        msg = (
-            f"Server '{server_name}' does not use OAuth "
-            '(set "auth": "oauth" in mcpServers).'
-        )
-        raise ValueError(msg)
-
     from deepagents_code.mcp_tools import _resolve_server_type
 
+    # OAuth login is discovery-based (RFC 9728), so it works for any remote
+    # http/sse server — whether the config opted in with `auth: oauth` or the
+    # server was auto-detected as needing auth via a 401 challenge. Only the
+    # transport needs gating; stdio servers can't speak OAuth.
     transport = _resolve_server_type(server_config)
     if transport not in {"http", "sse"}:
         msg = (

libs/code/deepagents_code/mcp_tools.py

@@ -1477,26 +1477,40 @@ async def _load_tools_from_config(
                         server_name=server_name,
                     )
 
-                if server_config.get("auth") == "oauth":
-                    from deepagents_code.mcp_auth import (
-                        FileTokenStorage,
-                        build_oauth_provider,
-                    )
+                from deepagents_code.mcp_auth import (
+                    FileTokenStorage,
+                    build_oauth_provider,
+                )
 
-                    storage = FileTokenStorage(
+                explicit_oauth = server_config.get("auth") == "oauth"
+                header_names = {
+                    name.lower() for name in (server_config.get("headers") or {})
+                }
+                has_authorization_header = "authorization" in header_names
+                storage = FileTokenStorage(
+                    server_name,
+                    server_url=server_config["url"],
+                )
+                stored_tokens = await storage.get_tokens()
+
+                if explicit_oauth and stored_tokens is None:
+                    # Config opted into OAuth but no tokens are stored yet —
+                    # require an upfront login before connecting.
+                    auth_msg = f"MCP server {server_name!r} needs re-authentication."
+                    logger.warning(
+                        "MCP server '%s' skipped: not authenticated.",
                         server_name,
-                        server_url=server_config["url"],
                     )
-                    if await storage.get_tokens() is None:
-                        auth_msg = (
-                            f"MCP server {server_name!r} needs re-authentication."
-                        )
-                        logger.warning(
-                            "MCP server '%s' skipped: not authenticated.",
-                            server_name,
-                        )
-                        skipped[server_name] = ("unauthenticated", auth_msg)
-                        continue
+                    skipped[server_name] = ("unauthenticated", auth_msg)
+                    continue
+
+                if explicit_oauth or (
+                    stored_tokens is not None and not has_authorization_header
+                ):
+                    # Attach the provider when the user opted in, or when a
+                    # prior login (possibly triggered by 401 auto-detection)
+                    # already stored tokens for this server. Static
+                    # Authorization headers take precedence over stored OAuth.
                     conn["auth"] = build_oauth_provider(
                         server_name=server_name,
                         server_url=server_config["url"],
@@ -1550,10 +1564,32 @@ async def _load_tools_from_config(
         except (asyncio.CancelledError, KeyboardInterrupt, SystemExit):
             raise
         except Exception as exc:
-            from deepagents_code.mcp_auth import find_reauth_required
+            from deepagents_code.mcp_auth import (
+                find_oauth_challenge,
+                find_reauth_required,
+            )
 
-            reauth = find_reauth_required(exc)
             status: MCPServerStatus
+            try:
+                reauth = find_reauth_required(exc)
+                challenge_url = (
+                    find_oauth_challenge(exc)
+                    if transport in _SUPPORTED_REMOTE_TYPES
+                    else None
+                )
+            except Exception:
+                # Classifying the failure is best-effort. If a classifier
+                # itself raises, degrade this one server to a plain error
+                # rather than letting the exception abort tool loading for
+                # every remaining server.
+                reauth = None
+                challenge_url = None
+                logger.debug(
+                    "MCP server '%s': failed to classify discovery error",
+                    server_name,
+                    exc_info=True,
+                )
+
             if reauth is not None:
                 # Tokens existed (we checked above) but the OAuth provider
                 # fell back to interactive reauth — the refresh attempt
@@ -1575,6 +1611,27 @@ async def _load_tools_from_config(
                     server_name,
                     exc_info=True,
                 )
+            elif challenge_url is not None:
+                # A remote server answered with a 401 OAuth challenge
+                # (RFC 9728) that wasn't already handled as a token refresh —
+                # typically a server not opted into OAuth in config. Surface it
+                # as unauthenticated so the user can log in, rather than as an
+                # opaque connection error.
+                status = "unauthenticated"
+                error = (
+                    f"MCP server {server_name!r} requires authentication; "
+                    f"run `dcode mcp login {server_name}`."
+                )
+                logger.warning(
+                    "MCP server '%s' skipped: %s",
+                    server_name,
+                    error,
+                )
+                logger.debug(
+                    "MCP server '%s' skipped: 401 OAuth challenge detected",
+                    server_name,
+                    exc_info=True,
+                )
             else:
                 status = "error"
                 error = str(exc)

libs/code/tests/unit_tests/test_app.py

@@ -22,6 +22,7 @@ if TYPE_CHECKING:
 
     from langchain_core.messages import HumanMessage
 
+    from deepagents_code.mcp_auth import McpServerSpec
     from deepagents_code.notifications import PendingNotification
     from deepagents_code.sessions import ThreadInfo
 
@@ -17688,6 +17689,111 @@ class TestMCPLoginCommand:
             assert sentinel not in rendered
             assert "_FakeMcpError" in rendered or "FakeMcpError" in rendered
 
+    async def test_mcp_login_worker_allows_auto_detected_remote_oauth(
+        self,
+    ) -> None:
+        """Remote servers can log in after OAuth is auto-detected."""
+        from pathlib import Path as _Path
+
+        from deepagents_code.mcp_login_service import (
+            ConfigResolution,
+            ServerSelection,
+        )
+
+        app = DeepAgentsApp(agent=MagicMock())
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app._mcp_preload_kwargs = {
+                "mcp_config_path": None,
+                "no_mcp": False,
+                "trust_project_mcp": None,
+            }
+
+            config: McpServerSpec = {"type": "http", "url": "https://example"}
+            resolution = ConfigResolution(
+                config={"mcpServers": {"notion": config}},
+                used_paths=(_Path("/tmp/mcp.json"),),
+            )
+            selection = ServerSelection(
+                server_name="notion",
+                server_config=config,
+            )
+
+            with (
+                patch(
+                    "deepagents_code.mcp_login_service.resolve_mcp_config",
+                    return_value=resolution,
+                ),
+                patch(
+                    "deepagents_code.mcp_login_service.select_server",
+                    return_value=selection,
+                ),
+                patch("deepagents_code.mcp_auth.login", new=AsyncMock()) as login,
+                patch.object(app, "_prompt_mcp_reconnect", new=AsyncMock()) as prompt,
+            ):
+                await app._run_mcp_login_worker("notion")
+                await pilot.pause()
+
+            login.assert_awaited_once()
+            awaited = login.await_args
+            assert awaited is not None
+            assert awaited.kwargs["server_config"] == config
+            prompt.assert_awaited_once_with("notion")
+            assert not any(
+                "does not use OAuth" in str(w._content) for w in app.query(ErrorMessage)
+            )
+
+    async def test_mcp_login_worker_rejects_stdio_transport(self) -> None:
+        """A stdio server can't speak OAuth; the worker reports the transport."""
+        from pathlib import Path as _Path
+
+        from deepagents_code.mcp_login_service import (
+            ConfigResolution,
+            ServerSelection,
+        )
+
+        app = DeepAgentsApp(agent=MagicMock())
+        async with app.run_test() as pilot:
+            await pilot.pause()
+            app._mcp_preload_kwargs = {
+                "mcp_config_path": None,
+                "no_mcp": False,
+                "trust_project_mcp": None,
+            }
+
+            config: McpServerSpec = {"command": "some-server"}
+            resolution = ConfigResolution(
+                config={"mcpServers": {"local": config}},
+                used_paths=(_Path("/tmp/mcp.json"),),
+            )
+            selection = ServerSelection(
+                server_name="local",
+                server_config=config,
+            )
+
+            with (
+                patch(
+                    "deepagents_code.mcp_login_service.resolve_mcp_config",
+                    return_value=resolution,
+                ),
+                patch(
+                    "deepagents_code.mcp_login_service.select_server",
+                    return_value=selection,
+                ),
+                patch("deepagents_code.mcp_auth.login", new=AsyncMock()) as login,
+                patch.object(app, "_prompt_mcp_reconnect", new=AsyncMock()) as prompt,
+            ):
+                await app._run_mcp_login_worker("local")
+                await pilot.pause()
+
+            login.assert_not_awaited()
+            prompt.assert_not_awaited()
+            assert any(
+                "stdio" in str(w._content)
+                and "only valid for http/sse" in str(w._content)
+                for w in app.query(ErrorMessage)
+            )
+
     async def test_mcp_login_success_invokes_reconnect_prompt(self) -> None:
         """A successful login routes through `_prompt_mcp_reconnect`.
 

libs/code/tests/unit_tests/test_mcp_auth.py

@@ -11,6 +11,7 @@ from pathlib import Path
 from typing import Any, Literal, cast
 from unittest.mock import patch
 
+import httpx
 import pytest
 from mcp.client.auth import TokenStorage
 from mcp.shared.auth import OAuthToken
@@ -18,11 +19,27 @@ from mcp.shared.auth import OAuthToken
 from deepagents_code.mcp_auth import (
     FileTokenStorage,
     MCPReauthRequiredError,
+    find_oauth_challenge,
     find_reauth_required,
     format_login_failure,
     resolve_headers,
 )
 
+_RESOURCE_METADATA_URL = "https://mcp.example.com/.well-known/oauth-protected-resource"
+_BEARER_CHALLENGE = f'Bearer resource_metadata="{_RESOURCE_METADATA_URL}"'
+"""A minimal RFC 9728 Bearer challenge pointing at the resource metadata."""
+
+
+def _http_status_error(
+    status_code: int,
+    *,
+    headers: dict[str, str] | list[tuple[str, str]] | None = None,
+) -> httpx.HTTPStatusError:
+    """Build an `httpx.HTTPStatusError` with a canned response."""
+    request = httpx.Request("GET", "https://mcp.example.com/")
+    response = httpx.Response(status_code, headers=headers or {}, request=request)
+    return httpx.HTTPStatusError("boom", request=request, response=response)
+
 
 @pytest.fixture
 def fake_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
@@ -870,6 +887,150 @@ class TestFindReauthRequired:
         assert find_reauth_required(a) is None
 
 
+class TestFindOauthChallenge:
+    """Tests for detecting a 401 OAuth challenge in an exception tree."""
+
+    def test_direct_401_with_challenge(self) -> None:
+        """A 401 carrying an RFC 9728 Bearer challenge yields its URL."""
+        exc = _http_status_error(
+            401,
+            headers={"WWW-Authenticate": _BEARER_CHALLENGE},
+        )
+        assert find_oauth_challenge(exc) == _RESOURCE_METADATA_URL
+
+    def test_401_header_match_is_case_insensitive(self) -> None:
+        """The scheme and parameter matching ignore casing."""
+        exc = _http_status_error(
+            401,
+            headers={
+                "www-authenticate": (
+                    f'bearer resource_METADATA="{_RESOURCE_METADATA_URL}"'
+                )
+            },
+        )
+        assert find_oauth_challenge(exc) == _RESOURCE_METADATA_URL
+
+    def test_401_multiparam_bearer_challenge(self) -> None:
+        """`resource_metadata` is found after other Bearer auth-params."""
+        exc = _http_status_error(
+            401,
+            headers={
+                "WWW-Authenticate": (
+                    'Bearer error="invalid_token", '
+                    'error_description="The access token expired", '
+                    f'resource_metadata="{_RESOURCE_METADATA_URL}"'
+                )
+            },
+        )
+        assert find_oauth_challenge(exc) == _RESOURCE_METADATA_URL
+
+    def test_401_bearer_not_first_in_multischeme_line(self) -> None:
+        """A Bearer challenge behind another scheme on one line is detected."""
+        exc = _http_status_error(
+            401,
+            headers={"WWW-Authenticate": f'Basic realm="mcp", {_BEARER_CHALLENGE}'},
+        )
+        assert find_oauth_challenge(exc) == _RESOURCE_METADATA_URL
+
+    def test_401_bearer_across_repeated_headers(self) -> None:
+        """A Bearer challenge on a second `WWW-Authenticate` line is detected."""
+        exc = _http_status_error(
+            401,
+            headers=[
+                ("WWW-Authenticate", 'Basic realm="mcp"'),
+                (
+                    "WWW-Authenticate",
+                    _BEARER_CHALLENGE,
+                ),
+            ],
+        )
+        assert find_oauth_challenge(exc) == _RESOURCE_METADATA_URL
+
+    def test_401_without_challenge_header_ignored(self) -> None:
+        """A 401 lacking `WWW-Authenticate` is not an OAuth challenge."""
+        exc = _http_status_error(401)
+        assert find_oauth_challenge(exc) is None
+
+    def test_401_basic_challenge_ignored(self) -> None:
+        """A non-OAuth auth challenge is not treated as an MCP login prompt."""
+        exc = _http_status_error(
+            401,
+            headers={"WWW-Authenticate": 'Basic realm="mcp"'},
+        )
+        assert find_oauth_challenge(exc) is None
+
+    def test_401_bearer_without_resource_metadata_ignored(self) -> None:
+        """A Bearer challenge with params but no `resource_metadata` is ignored."""
+        exc = _http_status_error(
+            401,
+            headers={"WWW-Authenticate": 'Bearer realm="mcp"'},
+        )
+        assert find_oauth_challenge(exc) is None
+
+    def test_401_resource_metadata_substring_not_matched(self) -> None:
+        """`resource_metadata` embedded in another token is not a match."""
+        exc = _http_status_error(
+            401,
+            headers={"WWW-Authenticate": 'Bearer error="x_resource_metadata_y"'},
+        )
+        assert find_oauth_challenge(exc) is None
+
+    def test_non_401_status_ignored(self) -> None:
+        """Other status codes never count as a challenge."""
+        exc = _http_status_error(
+            403,
+            headers={"WWW-Authenticate": _BEARER_CHALLENGE},
+        )
+        assert find_oauth_challenge(exc) is None
+
+    def test_found_inside_exception_group(self) -> None:
+        """Nested exception groups are searched recursively."""
+        exc = ExceptionGroup(
+            "outer",
+            [
+                RuntimeError("x"),
+                _http_status_error(
+                    401,
+                    headers={"WWW-Authenticate": (_BEARER_CHALLENGE)},
+                ),
+            ],
+        )
+        assert find_oauth_challenge(exc) == _RESOURCE_METADATA_URL
+
+    def test_found_via_cause_chain(self) -> None:
+        """`raise X from HTTPStatusError(...)` is unwrapped."""
+        challenge = _http_status_error(
+            401,
+            headers={"WWW-Authenticate": _BEARER_CHALLENGE},
+        )
+        wrapped = RuntimeError("wrapped")
+        wrapped.__cause__ = challenge
+        assert find_oauth_challenge(wrapped) == _RESOURCE_METADATA_URL
+
+    def test_found_via_context_chain(self) -> None:
+        """Implicit chaining (`__context__`) is unwrapped, not only `__cause__`."""
+        challenge = _http_status_error(
+            401,
+            headers={"WWW-Authenticate": _BEARER_CHALLENGE},
+        )
+        wrapped = RuntimeError("wrapped")
+        wrapped.__context__ = challenge
+        assert find_oauth_challenge(wrapped) == _RESOURCE_METADATA_URL
+
+    def test_returns_none_when_absent(self) -> None:
+        """Trees without a 401 challenge yield `None`."""
+        exc = ExceptionGroup("outer", [RuntimeError("x"), ValueError("y")])
+        assert find_oauth_challenge(exc) is None
+
+    def test_handles_cyclic_chain(self) -> None:
+        """Self-referencing `__context__` cycles terminate without recursion."""
+        a = RuntimeError("a")
+        b = RuntimeError("b")
+        a.__context__ = b
+        b.__context__ = a
+        assert find_oauth_challenge(a) is None
+
+
 class TestFormatLoginFailure:
     """Tests for the token-safe summary helper used in app + CLI logs."""
 
@@ -2092,18 +2253,34 @@ class TestLogin:
         assert tokens is not None
         assert tokens.access_token == "new"
 
-    async def test_login_rejects_non_oauth_server(self) -> None:
-        """Only `auth: oauth` servers support the login command."""
+    async def test_login_allows_http_server_without_explicit_oauth(self) -> None:
+        """Auto-detected servers (no `auth: oauth`) can still run OAuth login."""
         from deepagents_code.mcp_auth import login
         from deepagents_code.mcp_oauth_ui import CliOAuthInteraction
 
-        with pytest.raises(ValueError, match="does not use OAuth"):
+        async def _fake_handshake(connections: dict) -> None:
+            server_name, connection = next(iter(connections.items()))
+            storage = FileTokenStorage(server_name, server_url=connection["url"])
+            await storage.set_tokens(
+                OAuthToken(access_token="new", token_type="Bearer")
+            )
+            await storage.set_client_info(_make_client_info())
+
+        with patch("deepagents_code.mcp_auth._drive_handshake", _fake_handshake):
             await login(
-                server_name="srv",
-                server_config={"transport": "http", "url": "https://example.com"},
+                server_name="notion",
+                server_config={
+                    "transport": "http",
+                    "url": "https://mcp.notion.com/mcp",
+                },
                 ui=CliOAuthInteraction(),
             )
 
+        storage = FileTokenStorage("notion", server_url="https://mcp.notion.com/mcp")
+        tokens = await storage.get_tokens()
+        assert tokens is not None
+        assert tokens.access_token == "new"
+
     async def test_login_rejects_stdio_server(self) -> None:
         """OAuth login is limited to HTTP/SSE transports."""
         from deepagents_code.mcp_auth import login

libs/code/tests/unit_tests/test_mcp_tools.py

@@ -10,6 +10,7 @@ from pathlib import Path
 from typing import TYPE_CHECKING, Any, cast
 from unittest.mock import AsyncMock, MagicMock, patch
 
+import httpx
 import pytest
 
 if TYPE_CHECKING:
@@ -1377,6 +1378,244 @@ class TestLoadToolsFromConfigOAuth:
         assert "Exception Group Traceback" not in caplog.text
         await manager.cleanup()
 
+    async def test_stored_tokens_attach_provider_without_explicit_oauth(
+        self,
+    ) -> None:
+        """Stored tokens attach a provider even when `auth: oauth` is absent."""
+        from mcp.client.auth import OAuthClientProvider
+        from mcp.shared.auth import OAuthToken
+
+        storage = FileTokenStorage("notion", server_url="https://mcp.notion.com/mcp")
+        await storage.set_tokens(OAuthToken(access_token="at", token_type="Bearer"))
+
+        recorded: list[dict[str, Any]] = []
+        session = AsyncMock()
+        session.initialize = AsyncMock()
+        session.list_tools = AsyncMock(return_value=_make_tool_page([]))
+
+        @asynccontextmanager
+        async def _fake(
+            connection: dict[str, Any],
+            *,
+            _mcp_callbacks: object | None = None,
+        ) -> AsyncIterator[AsyncMock]:
+            await asyncio.sleep(0)
+            recorded.append(connection)
+            yield session
+
+        with patch("langchain_mcp_adapters.sessions.create_session", _fake):
+            config = {
+                "mcpServers": {
+                    "notion": {
+                        "transport": "http",
+                        "url": "https://mcp.notion.com/mcp",
+                    }
+                }
+            }
+            tools, manager, _ = await _load_tools_from_config(config)
+
+        assert tools == []
+        assert isinstance(manager, MCPSessionManager)
+        assert isinstance(recorded[0].get("auth"), OAuthClientProvider)
+        await manager.cleanup()
+
+    async def test_authorization_header_skips_stored_oauth_without_explicit_oauth(
+        self,
+        monkeypatch: pytest.MonkeyPatch,
+    ) -> None:
+        """Static `Authorization` headers take precedence over stored OAuth."""
+        from mcp.shared.auth import OAuthToken
+
+        monkeypatch.setenv("DA_TOKEN", "tok-123")
+        storage = FileTokenStorage("notion", server_url="https://mcp.notion.com/mcp")
+        await storage.set_tokens(OAuthToken(access_token="at", token_type="Bearer"))
+
+        recorded: list[dict[str, Any]] = []
+        session = AsyncMock()
+        session.initialize = AsyncMock()
+        session.list_tools = AsyncMock(return_value=_make_tool_page([]))
+
+        @asynccontextmanager
+        async def _fake(
+            connection: dict[str, Any],
+            *,
+            _mcp_callbacks: object | None = None,
+        ) -> AsyncIterator[AsyncMock]:
+            await asyncio.sleep(0)
+            recorded.append(connection)
+            yield session
+
+        with patch("langchain_mcp_adapters.sessions.create_session", _fake):
+            config = {
+                "mcpServers": {
+                    "notion": {
+                        "transport": "http",
+                        "url": "https://mcp.notion.com/mcp",
+                        "headers": {"Authorization": "Bearer ${DA_TOKEN}"},
+                    }
+                }
+            }
+            tools, manager, _ = await _load_tools_from_config(config)
+
+        assert tools == []
+        assert isinstance(manager, MCPSessionManager)
+        assert recorded[0]["headers"] == {"Authorization": "Bearer tok-123"}
+        assert "auth" not in recorded[0]
+        await manager.cleanup()
+
+    async def test_discovery_401_challenge_marks_unauthenticated(self) -> None:
+        """A 401 OAuth challenge during discovery is surfaced as unauthenticated."""
+        request = httpx.Request("GET", "https://mcp.notion.com/mcp")
+        response = httpx.Response(
+            401,
+            headers={
+                "WWW-Authenticate": (
+                    'Bearer resource_metadata="https://mcp.notion.com/.well-known/'
+                    'oauth-protected-resource"'
+                )
+            },
+            request=request,
+        )
+        challenge = httpx.HTTPStatusError("boom", request=request, response=response)
+
+        @asynccontextmanager
+        async def _fake(
+            _connection: dict[str, Any],
+            *,
+            _mcp_callbacks: object | None = None,
+        ) -> AsyncIterator[None]:
+            await asyncio.sleep(0)
+            raise challenge
+            yield
+
+        with patch("langchain_mcp_adapters.sessions.create_session", _fake):
+            config = {
+                "mcpServers": {
+                    "notion": {
+                        "transport": "http",
+                        "url": "https://mcp.notion.com/mcp",
+                    }
+                }
+            }
+            tools, manager, server_infos = await _load_tools_from_config(config)
+
+        assert tools == []
+        assert isinstance(manager, MCPSessionManager)
+        assert server_infos[0].status == "unauthenticated"
+        assert "mcp login notion" in (server_infos[0].error or "")
+        await manager.cleanup()
+
+    async def test_discovery_401_without_challenge_stays_error(self) -> None:
+        """A 401 lacking `WWW-Authenticate` is not treated as an OAuth challenge."""
+        request = httpx.Request("GET", "https://mcp.notion.com/mcp")
+        response = httpx.Response(401, request=request)
+        error = httpx.HTTPStatusError("boom", request=request, response=response)
+
+        @asynccontextmanager
+        async def _fake(
+            _connection: dict[str, Any],
+            *,
+            _mcp_callbacks: object | None = None,
+        ) -> AsyncIterator[None]:
+            await asyncio.sleep(0)
+            raise error
+            yield
+
+        with patch("langchain_mcp_adapters.sessions.create_session", _fake):
+            config = {
+                "mcpServers": {
+                    "notion": {
+                        "transport": "http",
+                        "url": "https://mcp.notion.com/mcp",
+                    }
+                }
+            }
+            tools, manager, server_infos = await _load_tools_from_config(config)
+
+        assert tools == []
+        assert isinstance(manager, MCPSessionManager)
+        assert server_infos[0].status == "error"
+        await manager.cleanup()
+
+    async def test_discovery_401_basic_challenge_stays_error(self) -> None:
+        """A non-OAuth auth challenge is not treated as an MCP login prompt."""
+        request = httpx.Request("GET", "https://mcp.notion.com/mcp")
+        response = httpx.Response(
+            401,
+            headers={"WWW-Authenticate": 'Basic realm="mcp"'},
+            request=request,
+        )
+        error = httpx.HTTPStatusError("boom", request=request, response=response)
+
+        @asynccontextmanager
+        async def _fake(
+            _connection: dict[str, Any],
+            *,
+            _mcp_callbacks: object | None = None,
+        ) -> AsyncIterator[None]:
+            await asyncio.sleep(0)
+            raise error
+            yield
+
+        with patch("langchain_mcp_adapters.sessions.create_session", _fake):
+            config = {
+                "mcpServers": {
+                    "notion": {
+                        "transport": "http",
+                        "url": "https://mcp.notion.com/mcp",
+                    }
+                }
+            }
+            tools, manager, server_infos = await _load_tools_from_config(config)
+
+        assert tools == []
+        assert isinstance(manager, MCPSessionManager)
+        assert server_infos[0].status == "error"
+        await manager.cleanup()
+
+    async def test_discovery_401_challenge_marks_unauthenticated_sse(self) -> None:
+        """The 401 challenge classification also applies to SSE transports."""
+        request = httpx.Request("GET", "https://mcp.notion.com/sse")
+        response = httpx.Response(
+            401,
+            headers={
+                "WWW-Authenticate": (
+                    'Bearer resource_metadata="https://mcp.notion.com/.well-known/'
+                    'oauth-protected-resource"'
+                )
+            },
+            request=request,
+        )
+        challenge = httpx.HTTPStatusError("boom", request=request, response=response)
+
+        @asynccontextmanager
+        async def _fake(
+            _connection: dict[str, Any],
+            *,
+            _mcp_callbacks: object | None = None,
+        ) -> AsyncIterator[None]:
+            await asyncio.sleep(0)
+            raise challenge
+            yield
+
+        with patch("langchain_mcp_adapters.sessions.create_session", _fake):
+            config = {
+                "mcpServers": {
+                    "notion": {
+                        "transport": "sse",
+                        "url": "https://mcp.notion.com/sse",
+                    }
+                }
+            }
+            tools, manager, server_infos = await _load_tools_from_config(config)
+
+        assert tools == []
+        assert isinstance(manager, MCPSessionManager)
+        assert server_infos[0].transport == "sse"
+        assert server_infos[0].status == "unauthenticated"
+        assert "mcp login notion" in (server_infos[0].error or "")
+        await manager.cleanup()
+
 
 class TestResolveAndLoadMcpTools:
     """Test the unified resolve-and-load entrypoint."""