Stop counting an interactive OAuth login against request timeouts

src/mcp/client/auth/oauth2.py

@@ -43,6 +43,7 @@
     validate_metadata_issuer,
 )
 from mcp.shared._httpx_utils import RedirectAwareAuth, redirect_note
+from mcp.shared._request_clock import waiting_on_a_person
 from mcp.shared.auth import (
     AuthorizationCodeResult,
     OAuthClientInformationFull,
@@ -425,10 +426,9 @@ async def _perform_authorization_code_grant(self) -> tuple[str, str]:
                 auth_params["prompt"] = "consent"
 
         authorization_url = f"{auth_endpoint}?{urlencode(auth_params)}"
-        await self.context.redirect_handler(authorization_url)
-
-        # Wait for callback
-        result = await self.context.callback_handler()
+        with waiting_on_a_person():
+            await self.context.redirect_handler(authorization_url)
+            result = await self.context.callback_handler()
 
         if result.state is None or not secrets.compare_digest(result.state, state):
             raise OAuthFlowError(f"State parameter mismatch: {result.state} != {state}")

src/mcp/shared/_request_clock.py

@@ -0,0 +1,81 @@
+"""A request's timeout measures the peer, so its clock stops while the request waits on a person.
+
+The clock travels in a `ContextVar`: the HTTP transports run each outgoing message in its sender's
+context, which is how code far below `send_raw_request` finds the clock of the request it serves.
+
+Nothing in this module is public API: it may change or be removed without notice. It is likely to
+change with the client dispatcher work in https://github.com/modelcontextprotocol/python-sdk/pull/3517.
+"""
+
+import math
+from collections.abc import Iterator
+from contextlib import contextmanager
+from contextvars import ContextVar
+
+import anyio
+
+
+class RequestClock:
+    """One request's timeout, as a budget of seconds that is spent only while the clock runs.
+
+    Not public API: may change or be removed without notice.
+    """
+
+    def __init__(self, timeout: float | None, scope: anyio.CancelScope) -> None:
+        self._budget = math.inf if timeout is None else timeout
+        self._scope = scope
+        self._pauses = 1  # the write, which is off the clock too; `start()` ends it
+
+    def start(self) -> None:
+        self.resume()
+
+    def pause(self) -> None:
+        if not self._pauses:
+            self._budget = self._scope.deadline - anyio.current_time()
+            self._scope.deadline = math.inf
+        self._pauses += 1
+
+    def resume(self) -> None:
+        self._pauses -= 1
+        if not self._pauses:
+            self._scope.deadline = anyio.current_time() + self._budget
+
+
+_clock: ContextVar[RequestClock | None] = ContextVar("request_clock", default=None)
+
+
+@contextmanager
+def request_clock(timeout: float | None) -> Iterator[RequestClock]:
+    """Put a clock, not yet started, in the context of the request sent in this block.
+
+    Raises `TimeoutError` once the clock has run for `timeout` seconds.
+
+    Not public API: may change or be removed without notice.
+    """
+    with anyio.CancelScope() as scope:
+        clock = RequestClock(timeout, scope)
+        token = _clock.set(clock)
+        try:
+            yield clock
+        finally:
+            _clock.reset(token)
+    # Not `fail_after`: it re-reads the deadline here, which a pause may have moved since it expired.
+    if scope.cancelled_caught:
+        raise TimeoutError
+
+
+@contextmanager
+def waiting_on_a_person() -> Iterator[None]:
+    """Stop the clock of the request this code is serving while the block is open; a no-op if there is none.
+
+    Not public API: may change or be removed without notice.
+    """
+    clock = _clock.get()
+    if clock is None:
+        yield
+        return
+    clock.pause()
+    try:
+        yield
+    finally:
+        clock.resume()

src/mcp/shared/jsonrpc_dispatcher.py

@@ -38,6 +38,7 @@
 
 from mcp.shared._compat import resync_tracer
 from mcp.shared._otel import inject_trace_context, otel_span
+from mcp.shared._request_clock import request_clock
 from mcp.shared._stream_protocols import ReadStream, WriteStream
 from mcp.shared.dispatcher import (
     CallOptions,
@@ -404,12 +405,13 @@ async def send_raw_request(
                 # never started; past this point a cancelled write counts as issued.
                 await anyio.lowlevel.checkpoint_if_cancelled()
                 request_write_started = True
-                try:
-                    await self._write(msg, plan.metadata)
-                except (anyio.BrokenResourceError, anyio.ClosedResourceError):
-                    # Transport tore down before run() noticed EOF; surface the documented contract.
-                    raise MCPError(code=CONNECTION_CLOSED, message="Connection closed") from None
-                with anyio.fail_after(opts.get("timeout")):
+                with request_clock(opts.get("timeout")) as clock:
+                    try:
+                        await self._write(msg, plan.metadata)
+                    except (anyio.BrokenResourceError, anyio.ClosedResourceError):
+                        # Transport tore down before run() noticed EOF; surface the documented contract.
+                        raise MCPError(code=CONNECTION_CLOSED, message="Connection closed") from None
+                    clock.start()
                     timeout_armed = True
                     outcome = await receive.receive()
                 if isinstance(outcome, ErrorData) and outcome is not _CLOSED_OUTCOME:
@@ -418,7 +420,7 @@ async def send_raw_request(
                     span.set_status(StatusCode.ERROR, outcome.message)
         except TimeoutError:
             if not timeout_armed:
-                # `fail_after` arms only after the write, so this TimeoutError is the
+                # The clock starts only after the write, so this TimeoutError is the
                 # transport's own bounded send() failing - a transport error, not
                 # `opts["timeout"]` elapsing. Propagate it raw (v1 kept the write
                 # outside the timeout-catching try and did the same).

tests/interaction/_requirements.py

@@ -430,6 +430,15 @@ def __post_init__(self) -> None:
         ),
         added_in="2026-07-28",
     ),
+    "lifecycle:discover:fallback-silence": Requirement(
+        source=f"{SPEC_2026_BASE_URL}/basic/transports/stdio#backward-compatibility",
+        behavior=(
+            "When server/discover goes unanswered for the probe deadline, an auto-negotiating client falls "
+            "back to the legacy initialize handshake."
+        ),
+        added_in="2026-07-28",
+        note="The spec states the timeout rule for stdio only; the SDK applies it on every transport.",
+    ),
     "lifecycle:discover:network-error-raises": Requirement(
         source="sdk",
         behavior=(
@@ -3826,6 +3835,25 @@ def __post_init__(self) -> None:
         transports=("streamable-http",),
         note="OAuth is HTTP-only.",
     ),
+    "client-auth:login-time:not-counted": Requirement(
+        source="issue:#3601",
+        behavior=(
+            "The time OAuthClientProvider spends awaiting redirect_handler and callback_handler does not count "
+            "against the timeout of the request that was challenged; afterwards the timeout resumes with the "
+            "budget that was left. The provider's own HTTP calls do count."
+        ),
+        transports=("streamable-http",),
+        note="OAuth is HTTP-only.",
+    ),
+    "client-auth:login-time:other-requests-keep-counting": Requirement(
+        source="sdk",
+        behavior=(
+            "A login suspends the timeout of the challenged request only: a request queued behind it in the "
+            "provider still times out on its own clock."
+        ),
+        transports=("streamable-http",),
+        note="OAuth is HTTP-only.",
+    ),
     "client-auth:pkce:refuse-if-unsupported": Requirement(
         source=f"{SPEC_BASE_URL}/basic/authorization#authorization-code-protection",
         behavior=(

tests/interaction/auth/_harness.py

@@ -418,6 +418,9 @@ async def connect_with_oauth(
     verify_tokens: bool = True,
     app_shim: Callable[[ASGIApp], ASGIApp] | None = None,
     on_request: Callable[[httpx2.Request], None] | None = None,
+    mode: str = "legacy",
+    read_timeout_seconds: float | None = None,
+    json_response: bool = False,
 ) -> AsyncIterator[tuple[Client, HeadlessOAuth]]:
     """Connect a `Client` to a server's bearer-gated streamable-HTTP app, completing OAuth in process.
 
@@ -455,6 +458,7 @@ async def connect_with_oauth(
     )
 
     app: ASGIApp = server.streamable_http_app(
+        json_response=json_response,
         auth=settings,
         token_verifier=ProviderTokenVerifier(provider) if verify_tokens else None,
         auth_server_provider=provider,
@@ -481,7 +485,11 @@ async def hook(request: httpx2.Request) -> None:
         )
         headless.bind(http_client)
         client = await stack.enter_async_context(
-            # The auth flow tests snapshot the legacy initialize-handshake HTTP shape.
-            Client(streamable_http_client(f"{BASE_URL}/mcp", http_client=http_client), mode="legacy")
+            # The auth flow tests snapshot the legacy initialize-handshake HTTP shape, hence the default mode.
+            Client(
+                streamable_http_client(f"{BASE_URL}/mcp", http_client=http_client),
+                mode=mode,
+                read_timeout_seconds=read_timeout_seconds,
+            )
         )
         yield client, headless