From d8b3d6d345d960892921cf924f1bf1631ff61585 Mon Sep 17 00:00:00 2001
From: Tin Chi Lo <tin@berri.ai>
Date: Wed, 24 Jun 2026 10:54:29 -0700
Subject: [PATCH 01/11] feat(mcp): let CredError.of_unauthorized carry a 401
 challenge

The unauthorized case becomes a structured Unauthorized (detail + optional WWW-Authenticate
header + optional structured body) instead of a bare string, and raise_public emits the header
and body when present. This lets a mode reproduce a rich 401 challenge (e.g. BYOK's
provisioning prompt) through the generic resolver edge. of_unauthorized's new params are
keyword-only and default to None, so existing callers and the summary string are unchanged.
---
 .../outbound_credentials/adapter.py           | 11 ++++++-
 .../mcp_server/outbound_credentials/types.py  | 32 ++++++++++++++++---
 .../outbound_credentials/test_adapter.py      | 23 +++++++++++++
 3 files changed, 60 insertions(+), 6 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/outbound_credentials/adapter.py b/litellm/proxy/_experimental/mcp_server/outbound_credentials/adapter.py
index 39db2314aeeb..91876df762fa 100644
--- a/litellm/proxy/_experimental/mcp_server/outbound_credentials/adapter.py
+++ b/litellm/proxy/_experimental/mcp_server/outbound_credentials/adapter.py
@@ -126,7 +126,16 @@ def raise_public(error: CredError) -> NoReturn:
     """Map a resolver CredError onto the proxy's public HTTP contract. The one edge that raises."""
     match error.tag:
         case "unauthorized":
-            raise HTTPException(status_code=401, detail=error.summary)
+            challenge = error.unauthorized
+            raise HTTPException(
+                status_code=401,
+                detail=challenge.body if challenge.body is not None else error.summary,
+                headers=(
+                    {"WWW-Authenticate": challenge.www_authenticate}
+                    if challenge.www_authenticate
+                    else None
+                ),
+            )
         case "misconfigured":
             raise HTTPException(status_code=500, detail=error.summary)
         case "upstream_unavailable":
diff --git a/litellm/proxy/_experimental/mcp_server/outbound_credentials/types.py b/litellm/proxy/_experimental/mcp_server/outbound_credentials/types.py
index 2088dc77252e..f0e81272ffd3 100644
--- a/litellm/proxy/_experimental/mcp_server/outbound_credentials/types.py
+++ b/litellm/proxy/_experimental/mcp_server/outbound_credentials/types.py
@@ -26,7 +26,7 @@
 from __future__ import annotations
 
 from enum import Enum
-from typing import Annotated, Literal
+from typing import Annotated, Literal, Mapping, Optional
 
 from expression import case, tag, tagged_union
 from pydantic import BaseModel, ConfigDict, Field, SecretStr
@@ -59,6 +59,19 @@ class AuthSpecKind(str, Enum):
     aws_sigv4 = "aws_sigv4"  # AWS SigV4 per-request signing (e.g. Bedrock AgentCore)
 
 
+class Unauthorized(BaseModel):
+    """A 401 plus the optional challenge a client needs to recover.
+
+    ``detail`` is the human message; ``www_authenticate`` and ``body`` carry a scheme-specific
+    challenge (e.g. BYOK's provisioning prompt) so the edge can reproduce it verbatim.
+    """
+
+    model_config = ConfigDict(frozen=True)
+    detail: str
+    www_authenticate: Optional[str] = None
+    body: Optional[Mapping[str, str]] = None
+
+
 @tagged_union(frozen=True)
 class CredError:
     """Why a credential could not be produced. Fail-closed: an arm yields this or an `httpx.Auth`.
@@ -76,7 +89,7 @@ class CredError:
         "not_implemented",
     ] = tag()
 
-    unauthorized: str = (
+    unauthorized: Unauthorized = (
         case()
     )  # no usable credential for this (subject, server) -> 401 challenge
     misconfigured: str = (
@@ -96,8 +109,17 @@ class CredError:
     )  # the declared mode's resolver arm is not built yet -> 501 (not operator error)
 
     @staticmethod
-    def of_unauthorized(detail: str) -> CredError:
-        return CredError(unauthorized=detail)
+    def of_unauthorized(
+        detail: str,
+        *,
+        www_authenticate: Optional[str] = None,
+        body: Optional[Mapping[str, str]] = None,
+    ) -> CredError:
+        return CredError(
+            unauthorized=Unauthorized(
+                detail=detail, www_authenticate=www_authenticate, body=body
+            )
+        )
 
     @staticmethod
     def of_misconfigured(detail: str) -> CredError:
@@ -125,7 +147,7 @@ def summary(self) -> str:
         # only while that stays true (a `case _` would defeat reportMatchNotExhaustive).
         match self.tag:
             case "unauthorized":
-                return f"unauthorized: {self.unauthorized}"
+                return f"unauthorized: {self.unauthorized.detail}"
             case "misconfigured":
                 return f"misconfigured: {self.misconfigured}"
             case "upstream_unavailable":
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_adapter.py b/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_adapter.py
index 5481d60a22a6..594e9dcc9697 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_adapter.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_adapter.py
@@ -139,3 +139,26 @@ def test_raise_public_maps_each_error_to_its_status(error, status):
     with pytest.raises(HTTPException) as exc_info:
         raise_public(error)
     assert exc_info.value.status_code == status
+
+
+def test_raise_public_emits_unauthorized_challenge():
+    body = {"error": "byok_auth_required", "server_id": "s1"}
+    error = CredError.of_unauthorized(
+        "needs key", www_authenticate='Bearer resource_metadata="/x"', body=body
+    )
+    with pytest.raises(HTTPException) as exc_info:
+        raise_public(error)
+    exc = exc_info.value
+    assert exc.status_code == 401
+    assert exc.detail == body
+    assert exc.headers is not None
+    assert exc.headers["WWW-Authenticate"] == 'Bearer resource_metadata="/x"'
+
+
+def test_raise_public_plain_unauthorized_has_no_challenge():
+    with pytest.raises(HTTPException) as exc_info:
+        raise_public(CredError.of_unauthorized("nope"))
+    exc = exc_info.value
+    assert exc.status_code == 401
+    assert exc.detail == "unauthorized: nope"
+    assert exc.headers is None

From a42fb2fd1158bf598fe39d523e419ebb0c3cc2a3 Mon Sep 17 00:00:00 2001
From: Tin Chi Lo <tin@berri.ai>
Date: Wed, 24 Jun 2026 11:21:55 -0700
Subject: [PATCH 02/11] fix(mcp): make Unauthorized a frozen dataclass to keep
 the type budget flat

CredError's unauthorized payload was a pydantic BaseModel, whose base resolves as unknown in
this repo's basedpyright (every model in the file trips reportUntypedBaseClass plus an unknown
model_config), so the tagged-union case read as unknown and the public edge's challenge access
added reportUnknownMemberType errors over the per-rule ceiling. A frozen dataclass is fully
typed here, so error.unauthorized resolves directly with no cast or accessor and the per-rule
basedpyright counts match base.
---
 .../_experimental/mcp_server/outbound_credentials/types.py   | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/outbound_credentials/types.py b/litellm/proxy/_experimental/mcp_server/outbound_credentials/types.py
index f0e81272ffd3..26ad3f3fb28a 100644
--- a/litellm/proxy/_experimental/mcp_server/outbound_credentials/types.py
+++ b/litellm/proxy/_experimental/mcp_server/outbound_credentials/types.py
@@ -25,6 +25,7 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass
 from enum import Enum
 from typing import Annotated, Literal, Mapping, Optional
 
@@ -59,14 +60,14 @@ class AuthSpecKind(str, Enum):
     aws_sigv4 = "aws_sigv4"  # AWS SigV4 per-request signing (e.g. Bedrock AgentCore)
 
 
-class Unauthorized(BaseModel):
+@dataclass(frozen=True, slots=True)
+class Unauthorized:
     """A 401 plus the optional challenge a client needs to recover.
 
     ``detail`` is the human message; ``www_authenticate`` and ``body`` carry a scheme-specific
     challenge (e.g. BYOK's provisioning prompt) so the edge can reproduce it verbatim.
     """
 
-    model_config = ConfigDict(frozen=True)
     detail: str
     www_authenticate: Optional[str] = None
     body: Optional[Mapping[str, str]] = None

From 1d417a8caed6b37adbf7509df12a7350a291db24 Mon Sep 17 00:00:00 2001
From: Tin Chi Lo <tin@berri.ai>
Date: Wed, 24 Jun 2026 20:25:05 -0700
Subject: [PATCH 03/11] feat(mcp): OAuth token store seam + expiry-aware cache
 for authorization_code

Lay the foundation for the authorization_code resolver arm: OAuthToken (access_token,
expires_at, refresh_token), the OAuthTokenStore Protocol seam, TokenStoreUnavailable for
outages, and CachedOAuthTokenStore, an expiry-aware cache that serves a token only while
unexpired, caches the "not authorized" None for a default TTL, and propagates a store outage
without caching it. Mirrors the BYOK store/cache pattern, adapted for tokens. Refresh and
distributed single-flight are deferred to the hardening step.
---
 .../outbound_credentials/oauth_token_store.py | 101 ++++++++++++++
 .../test_oauth_token_store.py                 | 123 ++++++++++++++++++
 2 files changed, 224 insertions(+)
 create mode 100644 litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
 create mode 100644 tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py

diff --git a/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py b/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
new file mode 100644
index 000000000000..7cdeccaebc4b
--- /dev/null
+++ b/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
@@ -0,0 +1,101 @@
+"""Per-user OAuth token store for the ``authorization_code`` mode.
+
+The resolver reads a user's token through the injected ``OAuthTokenStore`` seam;
+``CachedOAuthTokenStore`` is an expiry-aware cache in front of it. ``TokenStoreUnavailable``
+signals an unreachable backing store, so an outage is never cached or read as "not authorized".
+
+Refresh (using ``refresh_token`` once the access token has expired) and distributed single-flight
+are the later hardening; this cache only avoids serving a token past its own expiry.
+"""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass
+from typing import Callable, Dict, Optional, Protocol, Tuple
+
+
+@dataclass(frozen=True, slots=True)
+class OAuthToken:
+    """A user's OAuth credential: the bearer value, when it expires, and how to refresh it.
+
+    ``expires_at`` is epoch seconds (``None`` means no known expiry). ``refresh_token`` is kept for
+    the later refresh step; it is never minted into a header directly.
+    """
+
+    access_token: str
+    expires_at: Optional[float] = None
+    refresh_token: Optional[str] = None
+
+
+class TokenStoreUnavailable(Exception):
+    """Raised by ``fetch`` when the backing token store is unreachable (e.g. the DB is down).
+
+    Distinct from returning ``None`` for "the user has not authorized this server": a read-through
+    cache skips caching the failure, and the resolver maps it to its fail-closed status rather than
+    treating an outage as a definite absence.
+    """
+
+
+class OAuthTokenStore(Protocol):
+    """Per-user OAuth token lookup for the ``authorization_code`` mode.
+
+    Returns the user's token for an upstream, or ``None`` when they have not completed the OAuth
+    flow (the arm turns that into a 401 challenge). The ``(user_id, server_id)`` pair fully scopes
+    the lookup, so an implementation must never return one subject's token to another. Raises
+    ``TokenStoreUnavailable`` when the backing store is unreachable, so an outage is never cached or
+    read as a definite absence.
+    """
+
+    async def fetch(self, user_id: str, server_id: str) -> Optional[OAuthToken]: ...
+
+
+class CachedOAuthTokenStore:
+    """Expiry-aware cache over an ``OAuthTokenStore``.
+
+    A cached token is served only while it is unexpired (minus ``expiry_skew_seconds``); past that
+    the inner store is read again. Tokens with no known expiry, and the ``None`` "not authorized"
+    result, are held for ``default_ttl_seconds`` so the store is not hit on every call. The clock is
+    injected (wall-clock, since ``expires_at`` is epoch) so expiry is deterministic in tests, and a
+    store outage (``TokenStoreUnavailable``) propagates without being cached.
+    """
+
+    def __init__(
+        self,
+        inner: OAuthTokenStore,
+        *,
+        default_ttl_seconds: float,
+        expiry_skew_seconds: float = 30.0,
+        max_size: int = 4096,
+        clock: Callable[[], float] = time.time,
+    ) -> None:
+        self._inner = inner
+        self._default_ttl_seconds = default_ttl_seconds
+        self._expiry_skew_seconds = expiry_skew_seconds
+        self._max_size = max_size
+        self._clock = clock
+        self._cache: Dict[Tuple[str, str], Tuple[Optional[OAuthToken], float]] = {}
+
+    def _valid_until(self, token: Optional[OAuthToken]) -> float:
+        if token is not None and token.expires_at is not None:
+            return token.expires_at - self._expiry_skew_seconds
+        return self._clock() + self._default_ttl_seconds
+
+    async def fetch(self, user_id: str, server_id: str) -> Optional[OAuthToken]:
+        key = (user_id, server_id)
+        hit = self._cache.get(key)
+        if hit is not None:
+            token, valid_until = hit
+            if self._clock() < valid_until:
+                return token
+
+        token = await self._inner.fetch(user_id, server_id)
+        if len(self._cache) >= self._max_size:
+            self._cache.clear()
+        self._cache[key] = (token, self._valid_until(token))
+        return token
+
+    def invalidate(self, user_id: str, server_id: str) -> None:
+        """Drop a cached entry after the user (re)authorizes or revokes, so a stale token or a
+        stale "not authorized" None cannot mask the change."""
+        self._cache.pop((user_id, server_id), None)
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py b/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
new file mode 100644
index 000000000000..5c5017966cc5
--- /dev/null
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
@@ -0,0 +1,123 @@
+"""Tests for the v2 OAuth token cache (CachedOAuthTokenStore)."""
+
+from typing import Dict, List, Optional, Tuple
+
+import pytest
+
+from litellm.proxy._experimental.mcp_server.outbound_credentials.oauth_token_store import (
+    CachedOAuthTokenStore,
+    OAuthToken,
+    TokenStoreUnavailable,
+)
+
+
+class _FakeStore:
+    """An OAuthTokenStore that records calls and returns canned tokens."""
+
+    def __init__(self, values: Dict[Tuple[str, str], Optional[OAuthToken]]) -> None:
+        self._values = values
+        self.calls: List[Tuple[str, str]] = []
+
+    async def fetch(self, user_id: str, server_id: str) -> Optional[OAuthToken]:
+        self.calls.append((user_id, server_id))
+        return self._values.get((user_id, server_id))
+
+
+class _Clock:
+    def __init__(self, t: float = 1000.0) -> None:
+        self.t = t
+
+    def __call__(self) -> float:
+        return self.t
+
+
+async def test_serves_token_until_its_expiry():
+    token = OAuthToken(access_token="at", expires_at=1100.0)
+    inner = _FakeStore({("u", "s"): token})
+    clock = _Clock(1000.0)
+    store = CachedOAuthTokenStore(
+        inner, default_ttl_seconds=60, expiry_skew_seconds=30, clock=clock
+    )
+
+    assert await store.fetch("u", "s") is token
+    clock.t = 1060.0  # still before expiry - skew (1100 - 30 = 1070)
+    assert await store.fetch("u", "s") is token
+    assert inner.calls == [("u", "s")]  # served from cache, store hit once
+
+
+async def test_refetches_once_token_has_expired():
+    token = OAuthToken(access_token="at", expires_at=1100.0)
+    inner = _FakeStore({("u", "s"): token})
+    clock = _Clock(1000.0)
+    store = CachedOAuthTokenStore(
+        inner, default_ttl_seconds=60, expiry_skew_seconds=30, clock=clock
+    )
+
+    await store.fetch("u", "s")
+    clock.t = 1080.0  # past expiry - skew (1070)
+    await store.fetch("u", "s")
+    assert len(inner.calls) == 2  # re-read after the cached token expired
+
+
+async def test_caches_not_authorized_none_for_default_ttl():
+    inner = _FakeStore({})  # user has not authorized
+    clock = _Clock(1000.0)
+    store = CachedOAuthTokenStore(inner, default_ttl_seconds=60, clock=clock)
+
+    assert await store.fetch("u", "s") is None
+    clock.t = 1059.0
+    assert await store.fetch("u", "s") is None
+    assert inner.calls == [("u", "s")]  # None cached for the TTL window
+
+
+async def test_default_ttl_applies_to_tokens_without_expiry():
+    token = OAuthToken(access_token="at", expires_at=None)
+    inner = _FakeStore({("u", "s"): token})
+    clock = _Clock(1000.0)
+    store = CachedOAuthTokenStore(inner, default_ttl_seconds=60, clock=clock)
+
+    await store.fetch("u", "s")
+    clock.t = 1061.0
+    await store.fetch("u", "s")
+    assert len(inner.calls) == 2  # no-expiry token re-read after the default TTL
+
+
+async def test_invalidate_forces_refetch():
+    inner = _FakeStore({})
+    store = CachedOAuthTokenStore(inner, default_ttl_seconds=60, clock=_Clock())
+
+    assert await store.fetch("u", "s") is None
+    inner._values[("u", "s")] = OAuthToken(access_token="fresh")
+    store.invalidate("u", "s")
+    result = await store.fetch("u", "s")
+    assert result is not None and result.access_token == "fresh"
+
+
+async def test_store_unavailable_is_not_cached():
+    class _FailingStore:
+        def __init__(self) -> None:
+            self.calls = 0
+
+        async def fetch(self, user_id: str, server_id: str) -> Optional[OAuthToken]:
+            self.calls += 1
+            raise TokenStoreUnavailable("down")
+
+    inner = _FailingStore()
+    store = CachedOAuthTokenStore(inner, default_ttl_seconds=60, clock=_Clock())
+
+    for _ in range(2):
+        with pytest.raises(TokenStoreUnavailable):
+            await store.fetch("u", "s")
+    assert inner.calls == 2  # outage re-attempted, not cached
+
+
+async def test_isolates_by_subject():
+    a = OAuthToken(access_token="a")
+    b = OAuthToken(access_token="b")
+    inner = _FakeStore({("u1", "s"): a, ("u2", "s"): b})
+    store = CachedOAuthTokenStore(inner, default_ttl_seconds=60, clock=_Clock())
+
+    first = await store.fetch("u1", "s")
+    second = await store.fetch("u2", "s")
+    assert first is not None and first.access_token == "a"
+    assert second is not None and second.access_token == "b"

From fd2a3003b587f50705b6b190067f4b03d07dedac Mon Sep 17 00:00:00 2001
From: Tin Chi Lo <tin@berri.ai>
Date: Wed, 24 Jun 2026 21:23:16 -0700
Subject: [PATCH 04/11] feat(mcp): proactive token refresh with self-cleaning
 single-flight

Add TokenRefresher (a mode-supplied seam: mint a fresh token from an expired one and persist it)
and RefreshingTokenStore: when the stored token is near expiry, the first caller refreshes while
concurrent callers await the same in-flight task and share its result, so the IdP is not
stampeded. The task self-cleans (a done-callback drops its entry), so the map is bounded by
in-flight refreshes rather than by distinct users/servers, and is detached from the caller so a
cancelled caller does not abort the refresh. An expired token the refresher cannot renew surfaces
as None so the arm challenges, never a stale bearer; it composes under CachedOAuthTokenStore.
OAuthToken's repr masks the access/refresh tokens so a stray log cannot leak them. Cross-replica
single-flight (Redis) and reactive-401 refresh are the later distributed hardening.
---
 .../outbound_credentials/oauth_token_store.py |  91 ++++++++++++++-
 .../test_oauth_token_store.py                 | 107 +++++++++++++++++-
 2 files changed, 193 insertions(+), 5 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py b/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
index 7cdeccaebc4b..f1e4b17da291 100644
--- a/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
+++ b/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
@@ -4,29 +4,40 @@
 ``CachedOAuthTokenStore`` is an expiry-aware cache in front of it. ``TokenStoreUnavailable``
 signals an unreachable backing store, so an outage is never cached or read as "not authorized".
 
-Refresh (using ``refresh_token`` once the access token has expired) and distributed single-flight
-are the later hardening; this cache only avoids serving a token past its own expiry.
+``RefreshingTokenStore`` mints a fresh token through an injected ``TokenRefresher`` when the stored
+one is near expiry, under in-process per-(user, server) single-flight so concurrent callers share
+one refresh. Distributed (cross-replica) single-flight and reactive-401 refresh are the later
+hardening. The mode plugs in its own source and refresher; the cache, store seam, and refresh
+machinery are shared across the oauth2 modes (authorization_code / client_credentials /
+token_exchange).
 """
 
 from __future__ import annotations
 
+import asyncio
 import time
 from dataclasses import dataclass
 from typing import Callable, Dict, Optional, Protocol, Tuple
 
 
-@dataclass(frozen=True, slots=True)
+@dataclass(frozen=True, slots=True, repr=False)
 class OAuthToken:
     """A user's OAuth credential: the bearer value, when it expires, and how to refresh it.
 
     ``expires_at`` is epoch seconds (``None`` means no known expiry). ``refresh_token`` is kept for
-    the later refresh step; it is never minted into a header directly.
+    the later refresh step; it is never minted into a header directly. ``repr`` masks both secrets
+    so a stray log line cannot leak them (the values are still plain ``str`` for the header path,
+    since ``SecretStr`` resolves as unknown under this repo's basedpyright).
     """
 
     access_token: str
     expires_at: Optional[float] = None
     refresh_token: Optional[str] = None
 
+    def __repr__(self) -> str:
+        has_refresh = self.refresh_token is not None
+        return f"OAuthToken(access_token=***, expires_at={self.expires_at!r}, has_refresh_token={has_refresh})"
+
 
 class TokenStoreUnavailable(Exception):
     """Raised by ``fetch`` when the backing token store is unreachable (e.g. the DB is down).
@@ -50,6 +61,18 @@ class OAuthTokenStore(Protocol):
     async def fetch(self, user_id: str, server_id: str) -> Optional[OAuthToken]: ...
 
 
+class TokenRefresher(Protocol):
+    """Mints a fresh token from an expired one and persists it, returning the new token.
+
+    The action is mode-specific: the ``authorization_code`` refresh_token grant, the
+    ``client_credentials`` grant, or an RFC 8693 re-exchange. Returns ``None`` when it cannot
+    refresh (e.g. no ``refresh_token``), which the caller turns into a 401 challenge. It must
+    persist the new token so later requests (and the surrounding cache) read it without refreshing.
+    """
+
+    async def refresh(self, token: OAuthToken) -> Optional[OAuthToken]: ...
+
+
 class CachedOAuthTokenStore:
     """Expiry-aware cache over an ``OAuthTokenStore``.
 
@@ -99,3 +122,63 @@ def invalidate(self, user_id: str, server_id: str) -> None:
         """Drop a cached entry after the user (re)authorizes or revokes, so a stale token or a
         stale "not authorized" None cannot mask the change."""
         self._cache.pop((user_id, server_id), None)
+
+
+class RefreshingTokenStore:
+    """An ``OAuthTokenStore`` that proactively refreshes a near-expiry token.
+
+    Reads from an inner store; if the token is within ``expiry_skew_seconds`` of expiry, it mints a
+    fresh one via the injected ``TokenRefresher`` under per-(user, server) single-flight: the first
+    caller refreshes while concurrent callers await the same in-flight future and share its result,
+    instead of stampeding the IdP. The refresher persists the new token so later requests (and the
+    surrounding cache) read it without refreshing again. An expired token the refresher cannot renew
+    (``None``) is surfaced as ``None`` so the arm challenges, never a stale bearer.
+
+    Single-flight here is in-process (one event loop). Cross-replica single-flight (Redis SET NX)
+    and reactive-401 refresh are the later distributed hardening. Composes under
+    ``CachedOAuthTokenStore`` so the refreshed token is cached until its own expiry.
+    """
+
+    def __init__(
+        self,
+        inner: OAuthTokenStore,
+        refresher: TokenRefresher,
+        *,
+        expiry_skew_seconds: float = 30.0,
+        clock: Callable[[], float] = time.time,
+    ) -> None:
+        self._inner = inner
+        self._refresher = refresher
+        self._expiry_skew_seconds = expiry_skew_seconds
+        self._clock = clock
+        # In-flight refreshes, one future per (user, server). Entries exist only while a refresh
+        # is running (removed in `finally`), so the map is bounded by concurrency, not by the
+        # number of distinct users/servers ever seen.
+        self._inflight: Dict[Tuple[str, str], asyncio.Future[Optional[OAuthToken]]] = {}
+
+    def _is_expired(self, token: OAuthToken) -> bool:
+        return (
+            token.expires_at is not None
+            and self._clock() >= token.expires_at - self._expiry_skew_seconds
+        )
+
+    async def fetch(self, user_id: str, server_id: str) -> Optional[OAuthToken]:
+        token = await self._inner.fetch(user_id, server_id)
+        if token is None or not self._is_expired(token):
+            return token
+        return await self._refresh_single_flight(user_id, server_id, token)
+
+    async def _refresh_single_flight(
+        self, user_id: str, server_id: str, token: OAuthToken
+    ) -> Optional[OAuthToken]:
+        key = (user_id, server_id)
+        task = self._inflight.get(key)
+        if task is None:
+            # First caller starts the refresh; concurrent callers await the same task and share its
+            # result (or exception). The done-callback removes the entry, so the map self-cleans and
+            # is bounded by in-flight refreshes, not by the number of distinct users/servers. The
+            # task is detached from the caller, so a cancelled caller does not abort the refresh.
+            task = asyncio.ensure_future(self._refresher.refresh(token))
+            self._inflight[key] = task
+            task.add_done_callback(lambda _t, k=key: self._inflight.pop(k, None))
+        return await task
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py b/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
index 5c5017966cc5..92f8a8fa1fa5 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
@@ -1,5 +1,6 @@
-"""Tests for the v2 OAuth token cache (CachedOAuthTokenStore)."""
+"""Tests for the v2 OAuth token cache and refresh (CachedOAuthTokenStore, RefreshingTokenStore)."""
 
+import asyncio
 from typing import Dict, List, Optional, Tuple
 
 import pytest
@@ -7,6 +8,7 @@
 from litellm.proxy._experimental.mcp_server.outbound_credentials.oauth_token_store import (
     CachedOAuthTokenStore,
     OAuthToken,
+    RefreshingTokenStore,
     TokenStoreUnavailable,
 )
 
@@ -121,3 +123,106 @@ async def test_isolates_by_subject():
     second = await store.fetch("u2", "s")
     assert first is not None and first.access_token == "a"
     assert second is not None and second.access_token == "b"
+
+
+class _RefreshablePair:
+    """A store + refresher pair that simulates persistence: refresh() updates what fetch returns,
+    and yields once so concurrent callers actually contend on the single-flight lock."""
+
+    def __init__(self, initial: Optional[OAuthToken]) -> None:
+        self._current = initial
+        self.fetch_calls = 0
+        self.refresh_calls = 0
+
+    async def fetch(self, user_id: str, server_id: str) -> Optional[OAuthToken]:
+        self.fetch_calls += 1
+        return self._current
+
+    async def refresh(self, token: OAuthToken) -> Optional[OAuthToken]:
+        self.refresh_calls += 1
+        await asyncio.sleep(
+            0
+        )  # yield so other concurrent callers reach the lock and wait
+        self._current = OAuthToken(access_token="refreshed", expires_at=9999.0)
+        return self._current
+
+
+async def test_refreshing_passes_through_a_fresh_token():
+    pair = _RefreshablePair(OAuthToken(access_token="ok", expires_at=9999.0))
+    store = RefreshingTokenStore(
+        pair, pair, expiry_skew_seconds=30, clock=_Clock(1000.0)
+    )
+
+    token = await store.fetch("u", "s")
+    assert token is not None and token.access_token == "ok"
+    assert pair.refresh_calls == 0  # not near expiry -> no refresh
+
+
+async def test_refreshing_mints_a_fresh_token_when_expired():
+    pair = _RefreshablePair(OAuthToken(access_token="old", expires_at=900.0))
+    store = RefreshingTokenStore(
+        pair, pair, expiry_skew_seconds=30, clock=_Clock(1000.0)
+    )
+
+    token = await store.fetch("u", "s")
+    assert token is not None and token.access_token == "refreshed"
+    assert pair.refresh_calls == 1
+
+
+async def test_refreshing_returns_none_when_it_cannot_refresh():
+    class _NoRefresh:
+        async def fetch(self, user_id: str, server_id: str) -> Optional[OAuthToken]:
+            return OAuthToken(access_token="old", expires_at=900.0)
+
+        async def refresh(self, token: OAuthToken) -> Optional[OAuthToken]:
+            return None  # e.g. no refresh_token
+
+    src = _NoRefresh()
+    store = RefreshingTokenStore(src, src, expiry_skew_seconds=30, clock=_Clock(1000.0))
+    # expired and unrefreshable -> None (the arm challenges), never a stale bearer
+    assert await store.fetch("u", "s") is None
+
+
+async def test_refreshing_is_single_flight_under_concurrency():
+    pair = _RefreshablePair(OAuthToken(access_token="old", expires_at=900.0))
+    store = RefreshingTokenStore(
+        pair, pair, expiry_skew_seconds=30, clock=_Clock(1000.0)
+    )
+
+    results = await asyncio.gather(*[store.fetch("u", "s") for _ in range(5)])
+    assert pair.refresh_calls == 1  # one refresh shared across 5 concurrent callers
+    assert all(r is not None and r.access_token == "refreshed" for r in results)
+
+
+async def test_refresh_failure_is_shared_by_joiners_not_re_run():
+    class _FailingRefresher:
+        def __init__(self) -> None:
+            self.calls = 0
+
+        async def fetch(self, user_id: str, server_id: str) -> Optional[OAuthToken]:
+            return OAuthToken(access_token="old", expires_at=900.0)
+
+        async def refresh(self, token: OAuthToken) -> Optional[OAuthToken]:
+            self.calls += 1
+            await asyncio.sleep(0)  # let the concurrent callers join the same task
+            raise RuntimeError("refresh boom")
+
+    src = _FailingRefresher()
+    store = RefreshingTokenStore(src, src, expiry_skew_seconds=30, clock=_Clock(1000.0))
+
+    results = await asyncio.gather(
+        *[store.fetch("u", "s") for _ in range(3)], return_exceptions=True
+    )
+    assert src.calls == 1  # single-flight: one attempt, the failure is shared
+    assert all(isinstance(r, RuntimeError) for r in results)
+
+
+def test_oauth_token_repr_masks_the_secrets():
+    token = OAuthToken(
+        access_token="super-secret", expires_at=123.0, refresh_token="rt-secret"
+    )
+    rendered = repr(token)
+    assert "super-secret" not in rendered
+    assert "rt-secret" not in rendered
+    assert "access_token=***" in rendered
+    assert "has_refresh_token=True" in rendered

From 701f79604d5140613ba95ed1f578a4a9bfacb9ff Mon Sep 17 00:00:00 2001
From: Tin Chi Lo <tin@berri.ai>
Date: Wed, 24 Jun 2026 22:56:34 -0700
Subject: [PATCH 05/11] style(mcp): modern type annotations (dict/tuple/X |
 None) + sorted imports in the token modules

---
 .../outbound_credentials/oauth_token_store.py | 23 ++++++++++---------
 .../mcp_server/outbound_credentials/types.py  | 11 +++++----
 2 files changed, 18 insertions(+), 16 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py b/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
index f1e4b17da291..e69772ca22a9 100644
--- a/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
+++ b/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
@@ -16,8 +16,9 @@
 
 import asyncio
 import time
+from collections.abc import Callable
 from dataclasses import dataclass
-from typing import Callable, Dict, Optional, Protocol, Tuple
+from typing import Protocol
 
 
 @dataclass(frozen=True, slots=True, repr=False)
@@ -31,8 +32,8 @@ class OAuthToken:
     """
 
     access_token: str
-    expires_at: Optional[float] = None
-    refresh_token: Optional[str] = None
+    expires_at: float | None = None
+    refresh_token: str | None = None
 
     def __repr__(self) -> str:
         has_refresh = self.refresh_token is not None
@@ -58,7 +59,7 @@ class OAuthTokenStore(Protocol):
     read as a definite absence.
     """
 
-    async def fetch(self, user_id: str, server_id: str) -> Optional[OAuthToken]: ...
+    async def fetch(self, user_id: str, server_id: str) -> OAuthToken | None: ...
 
 
 class TokenRefresher(Protocol):
@@ -70,7 +71,7 @@ class TokenRefresher(Protocol):
     persist the new token so later requests (and the surrounding cache) read it without refreshing.
     """
 
-    async def refresh(self, token: OAuthToken) -> Optional[OAuthToken]: ...
+    async def refresh(self, token: OAuthToken) -> OAuthToken | None: ...
 
 
 class CachedOAuthTokenStore:
@@ -97,14 +98,14 @@ def __init__(
         self._expiry_skew_seconds = expiry_skew_seconds
         self._max_size = max_size
         self._clock = clock
-        self._cache: Dict[Tuple[str, str], Tuple[Optional[OAuthToken], float]] = {}
+        self._cache: dict[tuple[str, str], tuple[OAuthToken | None, float]] = {}
 
-    def _valid_until(self, token: Optional[OAuthToken]) -> float:
+    def _valid_until(self, token: OAuthToken | None) -> float:
         if token is not None and token.expires_at is not None:
             return token.expires_at - self._expiry_skew_seconds
         return self._clock() + self._default_ttl_seconds
 
-    async def fetch(self, user_id: str, server_id: str) -> Optional[OAuthToken]:
+    async def fetch(self, user_id: str, server_id: str) -> OAuthToken | None:
         key = (user_id, server_id)
         hit = self._cache.get(key)
         if hit is not None:
@@ -154,7 +155,7 @@ def __init__(
         # In-flight refreshes, one future per (user, server). Entries exist only while a refresh
         # is running (removed in `finally`), so the map is bounded by concurrency, not by the
         # number of distinct users/servers ever seen.
-        self._inflight: Dict[Tuple[str, str], asyncio.Future[Optional[OAuthToken]]] = {}
+        self._inflight: dict[tuple[str, str], asyncio.Future[OAuthToken | None]] = {}
 
     def _is_expired(self, token: OAuthToken) -> bool:
         return (
@@ -162,7 +163,7 @@ def _is_expired(self, token: OAuthToken) -> bool:
             and self._clock() >= token.expires_at - self._expiry_skew_seconds
         )
 
-    async def fetch(self, user_id: str, server_id: str) -> Optional[OAuthToken]:
+    async def fetch(self, user_id: str, server_id: str) -> OAuthToken | None:
         token = await self._inner.fetch(user_id, server_id)
         if token is None or not self._is_expired(token):
             return token
@@ -170,7 +171,7 @@ async def fetch(self, user_id: str, server_id: str) -> Optional[OAuthToken]:
 
     async def _refresh_single_flight(
         self, user_id: str, server_id: str, token: OAuthToken
-    ) -> Optional[OAuthToken]:
+    ) -> OAuthToken | None:
         key = (user_id, server_id)
         task = self._inflight.get(key)
         if task is None:
diff --git a/litellm/proxy/_experimental/mcp_server/outbound_credentials/types.py b/litellm/proxy/_experimental/mcp_server/outbound_credentials/types.py
index 26ad3f3fb28a..8e589d3a24bb 100644
--- a/litellm/proxy/_experimental/mcp_server/outbound_credentials/types.py
+++ b/litellm/proxy/_experimental/mcp_server/outbound_credentials/types.py
@@ -25,9 +25,10 @@
 
 from __future__ import annotations
 
+from collections.abc import Mapping
 from dataclasses import dataclass
 from enum import Enum
-from typing import Annotated, Literal, Mapping, Optional
+from typing import Annotated, Literal
 
 from expression import case, tag, tagged_union
 from pydantic import BaseModel, ConfigDict, Field, SecretStr
@@ -69,8 +70,8 @@ class Unauthorized:
     """
 
     detail: str
-    www_authenticate: Optional[str] = None
-    body: Optional[Mapping[str, str]] = None
+    www_authenticate: str | None = None
+    body: Mapping[str, str] | None = None
 
 
 @tagged_union(frozen=True)
@@ -113,8 +114,8 @@ class CredError:
     def of_unauthorized(
         detail: str,
         *,
-        www_authenticate: Optional[str] = None,
-        body: Optional[Mapping[str, str]] = None,
+        www_authenticate: str | None = None,
+        body: Mapping[str, str] | None = None,
     ) -> CredError:
         return CredError(
             unauthorized=Unauthorized(

From 6ed9ecfaa7e59c59978f61e25bc42ba0a145238e Mon Sep 17 00:00:00 2001
From: Tin Chi Lo <tin@berri.ai>
Date: Wed, 24 Jun 2026 23:03:32 -0700
Subject: [PATCH 06/11] refactor(mcp): FIFO cache eviction, fix stale
 single-flight comment + refresh_token docstring

---
 .../outbound_credentials/oauth_token_store.py | 22 ++++++++++-------
 .../test_oauth_token_store.py                 | 24 +++++++++++++++++++
 2 files changed, 37 insertions(+), 9 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py b/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
index e69772ca22a9..2e9044a3db64 100644
--- a/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
+++ b/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
@@ -25,10 +25,12 @@
 class OAuthToken:
     """A user's OAuth credential: the bearer value, when it expires, and how to refresh it.
 
-    ``expires_at`` is epoch seconds (``None`` means no known expiry). ``refresh_token`` is kept for
-    the later refresh step; it is never minted into a header directly. ``repr`` masks both secrets
-    so a stray log line cannot leak them (the values are still plain ``str`` for the header path,
-    since ``SecretStr`` resolves as unknown under this repo's basedpyright).
+    ``expires_at`` is epoch seconds (``None`` means no known expiry). ``refresh_token`` is what a
+    ``TokenRefresher`` uses to mint a new access token when this one nears expiry (the refresh
+    mechanism, ``RefreshingTokenStore``, is in this module; the concrete per-mode refresher lands
+    with each mode); it is never minted into a header directly. ``repr`` masks both secrets so a
+    stray log line cannot leak them (the values are still plain ``str`` for the header path, since
+    ``SecretStr`` resolves as unknown under this repo's basedpyright).
     """
 
     access_token: str
@@ -114,8 +116,10 @@ async def fetch(self, user_id: str, server_id: str) -> OAuthToken | None:
                 return token
 
         token = await self._inner.fetch(user_id, server_id)
-        if len(self._cache) >= self._max_size:
-            self._cache.clear()
+        if key not in self._cache and len(self._cache) >= self._max_size:
+            # Evict the oldest entry (insertion order) to make room, rather than clearing the
+            # whole cache and forcing every key to re-read the store at once.
+            self._cache.pop(next(iter(self._cache)), None)
         self._cache[key] = (token, self._valid_until(token))
         return token
 
@@ -152,9 +156,9 @@ def __init__(
         self._refresher = refresher
         self._expiry_skew_seconds = expiry_skew_seconds
         self._clock = clock
-        # In-flight refreshes, one future per (user, server). Entries exist only while a refresh
-        # is running (removed in `finally`), so the map is bounded by concurrency, not by the
-        # number of distinct users/servers ever seen.
+        # In-flight refreshes, one task per (user, server). Each entry is removed by the task's
+        # done-callback, so the map is bounded by concurrent refreshes, not by the number of
+        # distinct users/servers ever seen.
         self._inflight: dict[tuple[str, str], asyncio.Future[OAuthToken | None]] = {}
 
     def _is_expired(self, token: OAuthToken) -> bool:
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py b/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
index 92f8a8fa1fa5..e8baae782b6c 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
@@ -125,6 +125,30 @@ async def test_isolates_by_subject():
     assert second is not None and second.access_token == "b"
 
 
+async def test_bounded_cache_evicts_oldest_not_everything():
+    inner = _FakeStore(
+        {
+            ("u1", "s"): OAuthToken(access_token="k1"),
+            ("u2", "s"): OAuthToken(access_token="k2"),
+            ("u3", "s"): OAuthToken(access_token="k3"),
+        }
+    )
+    store = CachedOAuthTokenStore(
+        inner, default_ttl_seconds=60, max_size=2, clock=_Clock()
+    )
+
+    await store.fetch("u1", "s")
+    await store.fetch("u2", "s")
+    await store.fetch("u3", "s")  # at capacity -> evict the oldest (u1), keep u2
+    await store.fetch("u2", "s")  # still cached
+    await store.fetch("u1", "s")  # was evicted -> re-read
+
+    assert (
+        inner.calls.count(("u2", "s")) == 1
+    )  # only the oldest was evicted, not everything
+    assert inner.calls.count(("u1", "s")) == 2
+
+
 class _RefreshablePair:
     """A store + refresher pair that simulates persistence: refresh() updates what fetch returns,
     and yields once so concurrent callers actually contend on the single-flight lock."""

From 92b2b8c26e022c61437d8e29fd360abd7b2d0518 Mon Sep 17 00:00:00 2001
From: Tin Chi Lo <tin@berri.ai>
Date: Thu, 25 Jun 2026 10:00:15 -0700
Subject: [PATCH 07/11] refactor(mcp): cache positive tokens only, matching v1
 (no negative caching)

CachedOAuthTokenStore no longer caches the "not authorized" None result; every miss re-reads the
inner store. v1's per-user token cache never caches misses, so a token written by the OAuth flow
is visible on the next request without an invalidation hook, and uniformly across replicas since
the in-process cache holds no stale None to clear. invalidate() now only covers rotation or
revocation of a cached token. Negative caching (with distributed invalidation) can return later
if a slow DB-backed v2-native source makes per-miss reads expensive.
---
 .../outbound_credentials/oauth_token_store.py | 27 ++++++++------
 .../test_oauth_token_store.py                 | 35 ++++++++++++++-----
 2 files changed, 44 insertions(+), 18 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py b/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
index 2e9044a3db64..8a379cc06ec1 100644
--- a/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
+++ b/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
@@ -77,13 +77,15 @@ async def refresh(self, token: OAuthToken) -> OAuthToken | None: ...
 
 
 class CachedOAuthTokenStore:
-    """Expiry-aware cache over an ``OAuthTokenStore``.
-
-    A cached token is served only while it is unexpired (minus ``expiry_skew_seconds``); past that
-    the inner store is read again. Tokens with no known expiry, and the ``None`` "not authorized"
-    result, are held for ``default_ttl_seconds`` so the store is not hit on every call. The clock is
-    injected (wall-clock, since ``expires_at`` is epoch) so expiry is deterministic in tests, and a
-    store outage (``TokenStoreUnavailable``) propagates without being cached.
+    """Expiry-aware cache over an ``OAuthTokenStore``. Caches positive tokens only.
+
+    A cached token is served only while it is unexpired (minus ``expiry_skew_seconds``), or for
+    ``default_ttl_seconds`` if it carries no expiry; past that the inner store is read again. A
+    "not authorized" (``None``) result is never cached: every miss re-reads the inner store, so a
+    token written after the OAuth flow is visible immediately on every replica, matching v1 (which
+    never caches misses). The clock is injected (wall-clock, since ``expires_at`` is epoch) so
+    expiry is deterministic in tests, and a store outage (``TokenStoreUnavailable``) propagates
+    without being cached.
     """
 
     def __init__(
@@ -100,10 +102,10 @@ def __init__(
         self._expiry_skew_seconds = expiry_skew_seconds
         self._max_size = max_size
         self._clock = clock
-        self._cache: dict[tuple[str, str], tuple[OAuthToken | None, float]] = {}
+        self._cache: dict[tuple[str, str], tuple[OAuthToken, float]] = {}
 
-    def _valid_until(self, token: OAuthToken | None) -> float:
-        if token is not None and token.expires_at is not None:
+    def _valid_until(self, token: OAuthToken) -> float:
+        if token.expires_at is not None:
             return token.expires_at - self._expiry_skew_seconds
         return self._clock() + self._default_ttl_seconds
 
@@ -116,6 +118,11 @@ async def fetch(self, user_id: str, server_id: str) -> OAuthToken | None:
                 return token
 
         token = await self._inner.fetch(user_id, server_id)
+        if token is None:
+            # Never cache "not authorized": drop any stale entry and re-read on the next call, so
+            # a token stored after the OAuth flow is seen immediately rather than after a TTL.
+            self._cache.pop(key, None)
+            return token
         if key not in self._cache and len(self._cache) >= self._max_size:
             # Evict the oldest entry (insertion order) to make room, rather than clearing the
             # whole cache and forcing every key to re-read the store at once.
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py b/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
index e8baae782b6c..4fd4c27ac671 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
@@ -61,7 +61,7 @@ async def test_refetches_once_token_has_expired():
     assert len(inner.calls) == 2  # re-read after the cached token expired
 
 
-async def test_caches_not_authorized_none_for_default_ttl():
+async def test_does_not_cache_the_not_authorized_miss():
     inner = _FakeStore({})  # user has not authorized
     clock = _Clock(1000.0)
     store = CachedOAuthTokenStore(inner, default_ttl_seconds=60, clock=clock)
@@ -69,7 +69,22 @@ async def test_caches_not_authorized_none_for_default_ttl():
     assert await store.fetch("u", "s") is None
     clock.t = 1059.0
     assert await store.fetch("u", "s") is None
-    assert inner.calls == [("u", "s")]  # None cached for the TTL window
+    assert inner.calls == [
+        ("u", "s"),
+        ("u", "s"),
+    ]  # misses re-read the store, never cached
+
+
+async def test_token_stored_after_a_miss_is_visible_immediately():
+    # No invalidation needed: a miss is never cached, so a token written after the OAuth flow is
+    # served on the very next call (matching v1, where misses always re-read the source).
+    inner = _FakeStore({})
+    store = CachedOAuthTokenStore(inner, default_ttl_seconds=60, clock=_Clock())
+
+    assert await store.fetch("u", "s") is None
+    inner._values[("u", "s")] = OAuthToken(access_token="fresh")
+    result = await store.fetch("u", "s")
+    assert result is not None and result.access_token == "fresh"
 
 
 async def test_default_ttl_applies_to_tokens_without_expiry():
@@ -84,15 +99,19 @@ async def test_default_ttl_applies_to_tokens_without_expiry():
     assert len(inner.calls) == 2  # no-expiry token re-read after the default TTL
 
 
-async def test_invalidate_forces_refetch():
-    inner = _FakeStore({})
+async def test_invalidate_drops_a_cached_token():
+    # invalidate covers rotation/revocation of a *cached* token (the miss path needs no invalidate).
+    inner = _FakeStore({("u", "s"): OAuthToken(access_token="t1")})
     store = CachedOAuthTokenStore(inner, default_ttl_seconds=60, clock=_Clock())
 
-    assert await store.fetch("u", "s") is None
-    inner._values[("u", "s")] = OAuthToken(access_token="fresh")
+    first = await store.fetch("u", "s")
+    assert first is not None and first.access_token == "t1"  # cached
+    inner._values[("u", "s")] = OAuthToken(access_token="t2")  # rotated
     store.invalidate("u", "s")
-    result = await store.fetch("u", "s")
-    assert result is not None and result.access_token == "fresh"
+    second = await store.fetch("u", "s")
+    assert (
+        second is not None and second.access_token == "t2"
+    )  # re-read after invalidate
 
 
 async def test_store_unavailable_is_not_cached():

From 54414ffe2986773aea7ce56d7108456cb30661b3 Mon Sep 17 00:00:00 2001
From: Tin Chi Lo <tin@berri.ai>
Date: Thu, 25 Jun 2026 17:09:49 -0700
Subject: [PATCH 08/11] fix(mcp): default OAuth expiry skew to 60s, the
 industry standard

The proactive token-refresh / cache-expiry buffer defaulted to 30s, which is
an outlier among OAuth clients. Spring Security uses 60s as both its JWT
clock-skew tolerance and its refresh buffer, and 60s sits inside RFC 7519's
"a few minutes" leeway while preserving nearly all of a typical token's life;
30s was untested, so pin the default with two boundary-probe regression tests.
---
 .../outbound_credentials/oauth_token_store.py |  4 +--
 .../test_oauth_token_store.py                 | 36 +++++++++++++++++++
 2 files changed, 38 insertions(+), 2 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py b/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
index 8a379cc06ec1..656ffff5c47c 100644
--- a/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
+++ b/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
@@ -93,7 +93,7 @@ def __init__(
         inner: OAuthTokenStore,
         *,
         default_ttl_seconds: float,
-        expiry_skew_seconds: float = 30.0,
+        expiry_skew_seconds: float = 60.0,
         max_size: int = 4096,
         clock: Callable[[], float] = time.time,
     ) -> None:
@@ -156,7 +156,7 @@ def __init__(
         inner: OAuthTokenStore,
         refresher: TokenRefresher,
         *,
-        expiry_skew_seconds: float = 30.0,
+        expiry_skew_seconds: float = 60.0,
         clock: Callable[[], float] = time.time,
     ) -> None:
         self._inner = inner
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py b/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
index 4fd4c27ac671..c54a027f3407 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
@@ -260,6 +260,42 @@ async def refresh(self, token: OAuthToken) -> Optional[OAuthToken]:
     assert all(isinstance(r, RuntimeError) for r in results)
 
 
+async def test_cache_default_skew_is_60_seconds():
+    # The default expiry skew is the industry-standard 60s (Spring Security's clock-skew and
+    # refresh-buffer default; within RFC 7519's "a few minutes" leeway), so a cached token stops
+    # being served 60s before its real expiry. The boundary sits at expires_at - 60 = 1040.
+    token = OAuthToken(access_token="at", expires_at=1100.0)
+    inner = _FakeStore({("u", "s"): token})
+    clock = _Clock(1000.0)
+    store = CachedOAuthTokenStore(inner, default_ttl_seconds=600, clock=clock)
+
+    await store.fetch("u", "s")
+    clock.t = 1039.0  # just inside expires_at - 60 -> still served from cache
+    await store.fetch("u", "s")
+    assert len(inner.calls) == 1
+    clock.t = 1041.0  # just past expires_at - 60 -> re-read (a 30s skew would still cache here)
+    await store.fetch("u", "s")
+    assert len(inner.calls) == 2
+
+
+async def test_refreshing_default_skew_is_60_seconds():
+    # Same 60s default for the proactive refresh threshold: a token within 60s of expiry refreshes,
+    # one further out does not. At clock 1000 the boundary expires_at - 60 = 1000 lands on expires_at
+    # 1060 (refresh) vs 1061 (no refresh), pinning the default to exactly 60 (a 30s skew would not
+    # refresh either case).
+    not_near = _RefreshablePair(OAuthToken(access_token="ok", expires_at=1061.0))
+    not_near_store = RefreshingTokenStore(not_near, not_near, clock=_Clock(1000.0))
+    not_near_token = await not_near_store.fetch("u", "s")
+    assert not_near_token is not None and not_near_token.access_token == "ok"
+    assert not_near.refresh_calls == 0
+
+    near = _RefreshablePair(OAuthToken(access_token="old", expires_at=1060.0))
+    near_store = RefreshingTokenStore(near, near, clock=_Clock(1000.0))
+    near_token = await near_store.fetch("u", "s")
+    assert near_token is not None and near_token.access_token == "refreshed"
+    assert near.refresh_calls == 1
+
+
 def test_oauth_token_repr_masks_the_secrets():
     token = OAuthToken(
         access_token="super-secret", expires_at=123.0, refresh_token="rt-secret"

From eaf7932d950d20cc9d2c757fd06a481372a40763 Mon Sep 17 00:00:00 2001
From: Tin Chi Lo <tin@berri.ai>
Date: Thu, 25 Jun 2026 21:43:33 -0700
Subject: [PATCH 09/11] refactor(mcp): thread user_id/server_id through the
 TokenRefresher seam

The refresh seam took only the OAuthToken, but a refresher needs the server's
config (token endpoint, client credentials, scopes) to run the grant and the
(user_id, server_id) key to persist the minted token, neither of which is
derivable from the token. Widen TokenRefresher.refresh to (user_id, server_id,
token) and pass them through from RefreshingTokenStore so each stacked mode PR
plugs into the final seam rather than forcing a later signature change across
the stack.
---
 .../outbound_credentials/oauth_token_store.py   | 12 ++++++++++--
 .../test_oauth_token_store.py                   | 17 ++++++++++++++---
 2 files changed, 24 insertions(+), 5 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py b/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
index 656ffff5c47c..e909c99685f8 100644
--- a/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
+++ b/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
@@ -71,9 +71,15 @@ class TokenRefresher(Protocol):
     ``client_credentials`` grant, or an RFC 8693 re-exchange. Returns ``None`` when it cannot
     refresh (e.g. no ``refresh_token``), which the caller turns into a 401 challenge. It must
     persist the new token so later requests (and the surrounding cache) read it without refreshing.
+
+    ``server_id`` selects the upstream's config (token endpoint, client credentials, scopes) the
+    grant runs against; ``(user_id, server_id)`` is the key the new token is persisted under. They
+    are not derivable from ``token``, so the seam threads them alongside it.
     """
 
-    async def refresh(self, token: OAuthToken) -> OAuthToken | None: ...
+    async def refresh(
+        self, user_id: str, server_id: str, token: OAuthToken
+    ) -> OAuthToken | None: ...
 
 
 class CachedOAuthTokenStore:
@@ -190,7 +196,9 @@ async def _refresh_single_flight(
             # result (or exception). The done-callback removes the entry, so the map self-cleans and
             # is bounded by in-flight refreshes, not by the number of distinct users/servers. The
             # task is detached from the caller, so a cancelled caller does not abort the refresh.
-            task = asyncio.ensure_future(self._refresher.refresh(token))
+            task = asyncio.ensure_future(
+                self._refresher.refresh(user_id, server_id, token)
+            )
             self._inflight[key] = task
             task.add_done_callback(lambda _t, k=key: self._inflight.pop(k, None))
         return await task
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py b/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
index c54a027f3407..eb7c31e8dc62 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
@@ -176,13 +176,17 @@ def __init__(self, initial: Optional[OAuthToken]) -> None:
         self._current = initial
         self.fetch_calls = 0
         self.refresh_calls = 0
+        self.refresh_args: List[Tuple[str, str]] = []
 
     async def fetch(self, user_id: str, server_id: str) -> Optional[OAuthToken]:
         self.fetch_calls += 1
         return self._current
 
-    async def refresh(self, token: OAuthToken) -> Optional[OAuthToken]:
+    async def refresh(
+        self, user_id: str, server_id: str, token: OAuthToken
+    ) -> Optional[OAuthToken]:
         self.refresh_calls += 1
+        self.refresh_args.append((user_id, server_id))
         await asyncio.sleep(
             0
         )  # yield so other concurrent callers reach the lock and wait
@@ -210,6 +214,9 @@ async def test_refreshing_mints_a_fresh_token_when_expired():
     token = await store.fetch("u", "s")
     assert token is not None and token.access_token == "refreshed"
     assert pair.refresh_calls == 1
+    assert pair.refresh_args == [
+        ("u", "s")
+    ]  # the seam threads the grant/persist key through
 
 
 async def test_refreshing_returns_none_when_it_cannot_refresh():
@@ -217,7 +224,9 @@ class _NoRefresh:
         async def fetch(self, user_id: str, server_id: str) -> Optional[OAuthToken]:
             return OAuthToken(access_token="old", expires_at=900.0)
 
-        async def refresh(self, token: OAuthToken) -> Optional[OAuthToken]:
+        async def refresh(
+            self, user_id: str, server_id: str, token: OAuthToken
+        ) -> Optional[OAuthToken]:
             return None  # e.g. no refresh_token
 
     src = _NoRefresh()
@@ -245,7 +254,9 @@ def __init__(self) -> None:
         async def fetch(self, user_id: str, server_id: str) -> Optional[OAuthToken]:
             return OAuthToken(access_token="old", expires_at=900.0)
 
-        async def refresh(self, token: OAuthToken) -> Optional[OAuthToken]:
+        async def refresh(
+            self, user_id: str, server_id: str, token: OAuthToken
+        ) -> Optional[OAuthToken]:
             self.calls += 1
             await asyncio.sleep(0)  # let the concurrent callers join the same task
             raise RuntimeError("refresh boom")

From 83df471b058c18a064d02f7d071d8ffe85ca4c9c Mon Sep 17 00:00:00 2001
From: Tin Chi Lo <tin@berri.ai>
Date: Thu, 25 Jun 2026 22:17:10 -0700
Subject: [PATCH 10/11] feat(mcp): inject cache-backend and refresh-coordinator
 seams (cross-replica token caching)

Make CachedOAuthTokenStore's storage and RefreshingTokenStore's single-flight injectable so a
cross-replica deployment can back them with Redis without touching the resolver. The defaults preserve
today's behavior exactly: InMemoryTokenCacheBackend (the bounded per-process dict) and
InProcessRefreshCoordinator (the asyncio single-flight). A distributed deployment injects a shared
DualCache-backed backend and a SET NX PX coordinator. invalidate() is now async (the backend may be).
The cache stores via the backend with a TTL derived from the token's expiry; the coordinator threads a
reread callback for the cross-replica case (losers re-read the persisted token) that the in-process
default ignores.
---
 .../outbound_credentials/oauth_token_store.py | 181 +++++++++++++-----
 .../test_oauth_token_store.py                 |  80 +++++++-
 2 files changed, 210 insertions(+), 51 deletions(-)

diff --git a/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py b/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
index e909c99685f8..e37aafd69133 100644
--- a/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
+++ b/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
@@ -16,7 +16,7 @@
 
 import asyncio
 import time
-from collections.abc import Callable
+from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 from typing import Protocol
 
@@ -82,6 +82,57 @@ async def refresh(
     ) -> OAuthToken | None: ...
 
 
+class TokenCacheBackend(Protocol):
+    """Storage behind ``CachedOAuthTokenStore``: hold a token under ``(user_id, server_id)`` for
+    ``ttl_seconds``, then forget it. The default ``InMemoryTokenCacheBackend`` is per-process; a
+    cross-replica deployment injects a shared (Redis) backend so every worker reads one refresh,
+    matching v1. ``get`` returns ``None`` once the entry's TTL has elapsed.
+    """
+
+    async def get(self, user_id: str, server_id: str) -> OAuthToken | None: ...
+
+    async def set(
+        self, user_id: str, server_id: str, token: OAuthToken, ttl_seconds: float
+    ) -> None: ...
+
+    async def delete(self, user_id: str, server_id: str) -> None: ...
+
+
+class InMemoryTokenCacheBackend:
+    """Per-process token cache: a bounded dict with wall-clock TTLs (the default backend)."""
+
+    def __init__(
+        self, *, max_size: int = 4096, clock: Callable[[], float] = time.time
+    ) -> None:
+        self._max_size = max_size
+        self._clock = clock
+        self._cache: dict[tuple[str, str], tuple[OAuthToken, float]] = {}
+
+    async def get(self, user_id: str, server_id: str) -> OAuthToken | None:
+        key = (user_id, server_id)
+        hit = self._cache.get(key)
+        if hit is None:
+            return None
+        token, valid_until = hit
+        if self._clock() < valid_until:
+            return token
+        self._cache.pop(key, None)
+        return None
+
+    async def set(
+        self, user_id: str, server_id: str, token: OAuthToken, ttl_seconds: float
+    ) -> None:
+        key = (user_id, server_id)
+        if key not in self._cache and len(self._cache) >= self._max_size:
+            # Evict the oldest entry (insertion order), rather than clearing the whole cache and
+            # forcing every key to re-read the store at once.
+            self._cache.pop(next(iter(self._cache)), None)
+        self._cache[key] = (token, self._clock() + ttl_seconds)
+
+    async def delete(self, user_id: str, server_id: str) -> None:
+        self._cache.pop((user_id, server_id), None)
+
+
 class CachedOAuthTokenStore:
     """Expiry-aware cache over an ``OAuthTokenStore``. Caches positive tokens only.
 
@@ -101,60 +152,102 @@ def __init__(
         default_ttl_seconds: float,
         expiry_skew_seconds: float = 60.0,
         max_size: int = 4096,
+        backend: TokenCacheBackend | None = None,
         clock: Callable[[], float] = time.time,
     ) -> None:
         self._inner = inner
         self._default_ttl_seconds = default_ttl_seconds
         self._expiry_skew_seconds = expiry_skew_seconds
-        self._max_size = max_size
         self._clock = clock
-        self._cache: dict[tuple[str, str], tuple[OAuthToken, float]] = {}
+        self._backend: TokenCacheBackend = backend or InMemoryTokenCacheBackend(
+            max_size=max_size, clock=clock
+        )
 
-    def _valid_until(self, token: OAuthToken) -> float:
+    def _ttl(self, token: OAuthToken) -> float:
         if token.expires_at is not None:
-            return token.expires_at - self._expiry_skew_seconds
-        return self._clock() + self._default_ttl_seconds
+            return max(
+                0.0, token.expires_at - self._expiry_skew_seconds - self._clock()
+            )
+        return self._default_ttl_seconds
 
     async def fetch(self, user_id: str, server_id: str) -> OAuthToken | None:
-        key = (user_id, server_id)
-        hit = self._cache.get(key)
+        hit = await self._backend.get(user_id, server_id)
         if hit is not None:
-            token, valid_until = hit
-            if self._clock() < valid_until:
-                return token
+            return hit
 
         token = await self._inner.fetch(user_id, server_id)
         if token is None:
             # Never cache "not authorized": drop any stale entry and re-read on the next call, so
             # a token stored after the OAuth flow is seen immediately rather than after a TTL.
-            self._cache.pop(key, None)
+            await self._backend.delete(user_id, server_id)
             return token
-        if key not in self._cache and len(self._cache) >= self._max_size:
-            # Evict the oldest entry (insertion order) to make room, rather than clearing the
-            # whole cache and forcing every key to re-read the store at once.
-            self._cache.pop(next(iter(self._cache)), None)
-        self._cache[key] = (token, self._valid_until(token))
+        await self._backend.set(user_id, server_id, token, self._ttl(token))
         return token
 
-    def invalidate(self, user_id: str, server_id: str) -> None:
+    async def invalidate(self, user_id: str, server_id: str) -> None:
         """Drop a cached entry after the user (re)authorizes or revokes, so a stale token or a
         stale "not authorized" None cannot mask the change."""
-        self._cache.pop((user_id, server_id), None)
+        await self._backend.delete(user_id, server_id)
+
+
+class RefreshCoordinator(Protocol):
+    """Ensures one refresh runs per ``(user_id, server_id)`` at a time. Concurrent callers either
+    share the winner's result (the default ``InProcessRefreshCoordinator``) or, in a cross-replica
+    coordinator, wait for the holder and ``reread`` the token it persisted - so the IdP sees one
+    refresh per key across all workers, not one per worker.
+    """
+
+    async def run(
+        self,
+        user_id: str,
+        server_id: str,
+        refresh: Callable[[], Awaitable[OAuthToken | None]],
+        reread: Callable[[], Awaitable[OAuthToken | None]],
+    ) -> OAuthToken | None: ...
+
+
+class InProcessRefreshCoordinator:
+    """Single-flight within one event loop (the default): the first caller per key refreshes while
+    concurrent callers await the same in-flight task and share its result. ``reread`` is unused here -
+    the shared task already yields the new token - and exists for the cross-replica coordinator, where
+    losers re-read the persisted token instead of sharing an in-process future.
+    """
+
+    def __init__(self) -> None:
+        # In-flight refreshes, one task per (user, server); each entry is removed by the task's
+        # done-callback, so the map is bounded by concurrent refreshes, not by distinct keys seen.
+        self._inflight: dict[tuple[str, str], asyncio.Future[OAuthToken | None]] = {}
+
+    async def run(
+        self,
+        user_id: str,
+        server_id: str,
+        refresh: Callable[[], Awaitable[OAuthToken | None]],
+        reread: Callable[[], Awaitable[OAuthToken | None]],
+    ) -> OAuthToken | None:
+        key = (user_id, server_id)
+        task = self._inflight.get(key)
+        if task is None:
+            # The task is detached from the caller, so a cancelled caller does not abort the refresh.
+            task = asyncio.ensure_future(refresh())
+            self._inflight[key] = task
+            task.add_done_callback(lambda _t, k=key: self._inflight.pop(k, None))
+        return await task
 
 
 class RefreshingTokenStore:
     """An ``OAuthTokenStore`` that proactively refreshes a near-expiry token.
 
     Reads from an inner store; if the token is within ``expiry_skew_seconds`` of expiry, it mints a
-    fresh one via the injected ``TokenRefresher`` under per-(user, server) single-flight: the first
-    caller refreshes while concurrent callers await the same in-flight future and share its result,
-    instead of stampeding the IdP. The refresher persists the new token so later requests (and the
-    surrounding cache) read it without refreshing again. An expired token the refresher cannot renew
-    (``None``) is surfaced as ``None`` so the arm challenges, never a stale bearer.
-
-    Single-flight here is in-process (one event loop). Cross-replica single-flight (Redis SET NX)
-    and reactive-401 refresh are the later distributed hardening. Composes under
-    ``CachedOAuthTokenStore`` so the refreshed token is cached until its own expiry.
+    fresh one via the injected ``TokenRefresher``, serialized per ``(user, server)`` by the injected
+    ``RefreshCoordinator`` so callers don't stampede the IdP. The refresher persists the new token so
+    later requests (and the surrounding cache) read it without refreshing again. An expired token the
+    refresher cannot renew (``None``) is surfaced as ``None`` so the arm challenges, never a stale
+    bearer.
+
+    The default coordinator is in-process; a cross-replica deployment injects a distributed one (Redis
+    SET NX). Reactive-401 refresh is later hardening (it lives in the egress transport, which sees the
+    upstream's 401). Composes under ``CachedOAuthTokenStore`` so the refreshed token is cached.
     """
 
     def __init__(
@@ -163,16 +256,16 @@ def __init__(
         refresher: TokenRefresher,
         *,
         expiry_skew_seconds: float = 60.0,
+        coordinator: RefreshCoordinator | None = None,
         clock: Callable[[], float] = time.time,
     ) -> None:
         self._inner = inner
         self._refresher = refresher
         self._expiry_skew_seconds = expiry_skew_seconds
         self._clock = clock
-        # In-flight refreshes, one task per (user, server). Each entry is removed by the task's
-        # done-callback, so the map is bounded by concurrent refreshes, not by the number of
-        # distinct users/servers ever seen.
-        self._inflight: dict[tuple[str, str], asyncio.Future[OAuthToken | None]] = {}
+        self._coordinator: RefreshCoordinator = (
+            coordinator or InProcessRefreshCoordinator()
+        )
 
     def _is_expired(self, token: OAuthToken) -> bool:
         return (
@@ -184,21 +277,9 @@ async def fetch(self, user_id: str, server_id: str) -> OAuthToken | None:
         token = await self._inner.fetch(user_id, server_id)
         if token is None or not self._is_expired(token):
             return token
-        return await self._refresh_single_flight(user_id, server_id, token)
-
-    async def _refresh_single_flight(
-        self, user_id: str, server_id: str, token: OAuthToken
-    ) -> OAuthToken | None:
-        key = (user_id, server_id)
-        task = self._inflight.get(key)
-        if task is None:
-            # First caller starts the refresh; concurrent callers await the same task and share its
-            # result (or exception). The done-callback removes the entry, so the map self-cleans and
-            # is bounded by in-flight refreshes, not by the number of distinct users/servers. The
-            # task is detached from the caller, so a cancelled caller does not abort the refresh.
-            task = asyncio.ensure_future(
-                self._refresher.refresh(user_id, server_id, token)
-            )
-            self._inflight[key] = task
-            task.add_done_callback(lambda _t, k=key: self._inflight.pop(k, None))
-        return await task
+        return await self._coordinator.run(
+            user_id,
+            server_id,
+            refresh=lambda: self._refresher.refresh(user_id, server_id, token),
+            reread=lambda: self._inner.fetch(user_id, server_id),
+        )
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py b/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
index eb7c31e8dc62..e341f703829a 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
@@ -107,7 +107,7 @@ async def test_invalidate_drops_a_cached_token():
     first = await store.fetch("u", "s")
     assert first is not None and first.access_token == "t1"  # cached
     inner._values[("u", "s")] = OAuthToken(access_token="t2")  # rotated
-    store.invalidate("u", "s")
+    await store.invalidate("u", "s")
     second = await store.fetch("u", "s")
     assert (
         second is not None and second.access_token == "t2"
@@ -316,3 +316,81 @@ def test_oauth_token_repr_masks_the_secrets():
     assert "rt-secret" not in rendered
     assert "access_token=***" in rendered
     assert "has_refresh_token=True" in rendered
+
+
+class _RecordingBackend:
+    """A TokenCacheBackend that records calls, proving CachedOAuthTokenStore delegates storage."""
+
+    def __init__(self) -> None:
+        self.sets: List[Tuple[str, str, OAuthToken, float]] = []
+        self.deletes: List[Tuple[str, str]] = []
+        self._store: Dict[Tuple[str, str], OAuthToken] = {}
+
+    async def get(self, user_id: str, server_id: str) -> Optional[OAuthToken]:
+        return self._store.get((user_id, server_id))
+
+    async def set(
+        self, user_id: str, server_id: str, token: OAuthToken, ttl_seconds: float
+    ) -> None:
+        self.sets.append((user_id, server_id, token, ttl_seconds))
+        self._store[(user_id, server_id)] = token
+
+    async def delete(self, user_id: str, server_id: str) -> None:
+        self.deletes.append((user_id, server_id))
+        self._store.pop((user_id, server_id), None)
+
+
+async def test_cache_delegates_storage_to_an_injected_backend():
+    backend = _RecordingBackend()
+    inner = _FakeStore({("u", "s"): OAuthToken(access_token="at", expires_at=1100.0)})
+    store = CachedOAuthTokenStore(
+        inner,
+        default_ttl_seconds=60,
+        expiry_skew_seconds=30,
+        backend=backend,
+        clock=_Clock(1000.0),
+    )
+
+    token = await store.fetch("u", "s")
+    assert token is not None and token.access_token == "at"
+    # written through to the injected backend, TTL = expires_at - skew - now = 1100 - 30 - 1000
+    assert backend.sets == [("u", "s", token, 70.0)]
+    again = await store.fetch("u", "s")  # served by the backend, not the inner store
+    assert again is not None
+    assert inner.calls == [("u", "s")]
+
+
+async def test_cache_miss_deletes_from_the_injected_backend():
+    backend = _RecordingBackend()
+    store = CachedOAuthTokenStore(
+        _FakeStore({}), default_ttl_seconds=60, backend=backend, clock=_Clock()
+    )
+    assert await store.fetch("u", "s") is None
+    assert backend.deletes == [("u", "s")]  # a miss is never cached
+
+
+class _RecordingCoordinator:
+    """A RefreshCoordinator that records the call and runs the refresh, proving delegation."""
+
+    def __init__(self) -> None:
+        self.calls = 0
+
+    async def run(self, user_id, server_id, refresh, reread):
+        self.calls += 1
+        return await refresh()
+
+
+async def test_refreshing_delegates_single_flight_to_an_injected_coordinator():
+    pair = _RefreshablePair(OAuthToken(access_token="old", expires_at=900.0))
+    coordinator = _RecordingCoordinator()
+    store = RefreshingTokenStore(
+        pair,
+        pair,
+        expiry_skew_seconds=30,
+        coordinator=coordinator,
+        clock=_Clock(1000.0),
+    )
+
+    token = await store.fetch("u", "s")
+    assert token is not None and token.access_token == "refreshed"
+    assert coordinator.calls == 1  # the injected coordinator drove the refresh

From 66a33c9368bcef9da5a09fdda70cb824f816d98d Mon Sep 17 00:00:00 2001
From: Cursor Agent <cursoragent@cursor.com>
Date: Sat, 27 Jun 2026 00:30:37 +0000
Subject: [PATCH 11/11] fix: reread oauth token before refresh

---
 .../outbound_credentials/oauth_token_store.py |  9 +++-
 .../test_oauth_token_store.py                 | 50 +++++++++++++++++++
 2 files changed, 58 insertions(+), 1 deletion(-)

diff --git a/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py b/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
index e37aafd69133..1473d15da5e5 100644
--- a/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
+++ b/litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
@@ -277,9 +277,16 @@ async def fetch(self, user_id: str, server_id: str) -> OAuthToken | None:
         token = await self._inner.fetch(user_id, server_id)
         if token is None or not self._is_expired(token):
             return token
+
+        async def refresh_latest_token() -> OAuthToken | None:
+            latest_token = await self._inner.fetch(user_id, server_id)
+            if latest_token is None or not self._is_expired(latest_token):
+                return latest_token
+            return await self._refresher.refresh(user_id, server_id, latest_token)
+
         return await self._coordinator.run(
             user_id,
             server_id,
-            refresh=lambda: self._refresher.refresh(user_id, server_id, token),
+            refresh=refresh_latest_token,
             reread=lambda: self._inner.fetch(user_id, server_id),
         )
diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py b/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
index e341f703829a..63cb407de195 100644
--- a/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_oauth_token_store.py
@@ -246,6 +246,56 @@ async def test_refreshing_is_single_flight_under_concurrency():
     assert all(r is not None and r.access_token == "refreshed" for r in results)
 
 
+class _StaleReadRacePair:
+    def __init__(self) -> None:
+        self._old_token = OAuthToken(access_token="old", expires_at=900.0)
+        self._current = self._old_token
+        self.fetch_calls = 0
+        self.refresh_calls = 0
+        self.refresh_started = asyncio.Event()
+        self.finish_refresh = asyncio.Event()
+        self.stale_read_started = asyncio.Event()
+        self.finish_stale_read = asyncio.Event()
+
+    async def fetch(self, user_id: str, server_id: str) -> Optional[OAuthToken]:
+        self.fetch_calls += 1
+        if self.fetch_calls == 3:
+            self.stale_read_started.set()
+            await self.finish_stale_read.wait()
+            return self._old_token
+        return self._current
+
+    async def refresh(
+        self, user_id: str, server_id: str, token: OAuthToken
+    ) -> Optional[OAuthToken]:
+        self.refresh_calls += 1
+        self.refresh_started.set()
+        await self.finish_refresh.wait()
+        self._current = OAuthToken(access_token="refreshed", expires_at=9999.0)
+        return self._current
+
+
+async def test_stale_read_after_refresh_rereads_before_starting_new_refresh():
+    pair = _StaleReadRacePair()
+    store = RefreshingTokenStore(
+        pair, pair, expiry_skew_seconds=30, clock=_Clock(1000.0)
+    )
+
+    first = asyncio.create_task(store.fetch("u", "s"))
+    await pair.refresh_started.wait()
+    second = asyncio.create_task(store.fetch("u", "s"))
+    await pair.stale_read_started.wait()
+
+    pair.finish_refresh.set()
+    first_result = await first
+    pair.finish_stale_read.set()
+    second_result = await second
+
+    assert first_result is not None and first_result.access_token == "refreshed"
+    assert second_result is not None and second_result.access_token == "refreshed"
+    assert pair.refresh_calls == 1
+
+
 async def test_refresh_failure_is_shared_by_joiners_not_re_run():
     class _FailingRefresher:
         def __init__(self) -> None: