Treat an access token as expired at its expiry time

src/mcp/client/auth/extensions/identity_assertion.py

@@ -167,7 +167,7 @@ async def _auth_flow(self, request: httpx2.Request) -> AsyncGenerator[httpx2.Req
                 self._expiry = calculate_token_expiry(self._tokens.expires_in) if self._tokens else None
                 self._initialized = True
 
-            if self._tokens and (self._expiry is None or time.time() <= self._expiry):
+            if self._tokens and (self._expiry is None or time.time() < self._expiry):
                 request.headers["Authorization"] = f"Bearer {self._tokens.access_token}"
             response = yield request
 

src/mcp/client/auth/oauth2.py

@@ -188,7 +188,7 @@ def is_token_valid(self) -> bool:
         return bool(
             self.current_tokens
             and self.current_tokens.access_token
-            and (not self.token_expiry_time or time.time() <= self.token_expiry_time)
+            and (not self.token_expiry_time or time.time() < self.token_expiry_time)
         )
 
     def can_refresh_token(self) -> bool:

tests/client/auth/extensions/test_identity_assertion.py

@@ -8,6 +8,7 @@
 import base64
 import json
 import urllib.parse
+from unittest import mock
 
 import httpx2
 import pytest
@@ -324,6 +325,21 @@ async def test_stored_token_is_reused_without_reauthorizing() -> None:
     assert requests[0].headers["Authorization"] == "Bearer cached"
 
 
+@pytest.mark.anyio
+async def test_stored_token_is_not_sent_at_its_expiry_instant() -> None:
+    """SDK-defined: a token is expired at its `expires_in` deadline, like a JWT `exp` (RFC 7519 §4.1.4)."""
+    requests: list[httpx2.Request] = []
+    storage = InMemoryStorage(tokens=OAuthToken(access_token="cached", token_type="Bearer", expires_in=0))
+    auth = make_provider(storage)
+
+    transport = mock_transport(requests, rs_first_status=200)
+    with mock.patch("time.time", return_value=1_800_000_000.0):
+        async with httpx2.AsyncClient(transport=transport, auth=auth) as http:
+            await http.post(f"{RS}/mcp")
+
+    assert "Authorization" not in requests[0].headers
+
+
 @pytest.mark.anyio
 async def test_second_401_re_exchanges_without_refetching_asm() -> None:
     """ASM is discovered once; a later 401 mints a fresh assertion against the cached token endpoint."""

tests/client/test_auth.py

@@ -3667,3 +3667,13 @@ async def echo_callback() -> AuthorizationCodeResult:
         await auth_flow.asend(httpx2.Response(200, request=final_req))
     except StopAsyncIteration:
         pass
+
+
+def test_a_token_is_invalid_at_its_expiry_instant(oauth_provider: OAuthClientProvider):
+    """SDK-defined: the deadline from `expires_in` is exclusive, like a JWT `exp` (RFC 7519 §4.1.4)."""
+    context = oauth_provider.context
+    with mock.patch("time.time", return_value=1_800_000_000.0):
+        token = OAuthToken(access_token="t", token_type="Bearer", expires_in=0)
+        context.current_tokens = token
+        context.update_token_expiry(token)
+        assert not context.is_token_valid()