Add an expiration time to cached MCP auth credentials (#1872)

* Remote MCP servers may expire registered auth clients, when this happens the NAT MCP client will need to re-register client credentials.
* This adds a new `oauth_client_ttl` configuration attribute to `MCPOAuth2ProviderConfig`
* This incorporates changes from PR #1871

## By Submitting this PR I confirm:
- I am familiar with the [Contributing Guidelines](https://github.com/NVIDIA/NeMo-Agent-Toolkit/blob/develop/docs/source/resources/contributing/index.md).
- We require that all contributors "sign-off" on their commits. This certifies that the contribution is your original work, or you have rights to submit it under the same license, or a compatible license.
  - Any contribution which contains commits that are not Signed-Off will not be accepted.
- When the PR is ready for review, new or existing tests cover these changes.
- When the PR is ready for review, the documentation is up to date with these changes.



## Summary by CodeRabbit

* **New Features**
  * Configurable oauth_client_ttl for OAuth2 credential caching (default 270s; 0 disables caching).

* **Bug Fixes**
  * Improved authentication robustness: serialized discovery/registration to avoid races, automatic re-registration on failures, safer handling when credentials or endpoints are missing/expired, and retry logic for registration rejections.

* **Documentation**
  * Documented oauth_client_ttl behavior, defaults, and TTL guidance.

* **Tests**
  * Added tests for credential TTL, cache expiry, and related authentication flows.

Authors:
  - David Gardner (https://github.com/dagardner-nv)
  - Anuradha Karuppiah (https://github.com/AnuradhaKaruppiah)

Approvers:
  - Anuradha Karuppiah (https://github.com/AnuradhaKaruppiah)

URL: #1872

Author: dagardner-nv

docs/source/components/auth/mcp-auth/index.md

@@ -49,6 +49,7 @@ authentication:
 Configuration options:
 - `server_url`: The URL of the MCP server that requires authentication.
 - `redirect_uri`: The redirect URI for the OAuth2 flow. This must match the address where your server is accessible from your browser.
+- `oauth_client_ttl`: Amount of time, in seconds, to cache OAuth client credentials obtained via Dynamic Client Registration. Some MCP servers will invalidate client credentials after a certain period, requiring this value to match the timeout setting of the server minus a small safety buffer (for example, 30 seconds). After this period elapses, the client re-registers with the authorization server and obtains a new `client_id`. Defaults to `270` seconds. Set to `0` to disable caching (re-register on every authentication attempt).
 
 To view all configuration options for the `mcp_oauth2` authentication provider, run the following command:
 ```bash

packages/nvidia_nat_mcp/src/nat/plugins/mcp/auth/auth_provider.py

@@ -13,7 +13,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import asyncio
 import logging
+import time
 from collections.abc import Awaitable
 from collections.abc import Callable
 from urllib.parse import urljoin
@@ -316,7 +318,9 @@ def __init__(self, config: MCPOAuth2ProviderConfig, builder=None):
 
         # Client registration
         self._registrar = DynamicClientRegistration(config)
+        self._credentials_cache_time: float | None = None
         self._cached_credentials: OAuth2Credentials | None = None
+        self._discover_register_lock = asyncio.Lock()
 
         # For the OAuth2 flow
         self._auth_code_provider = None
@@ -331,12 +335,48 @@ def __init__(self, config: MCPOAuth2ProviderConfig, builder=None):
         if self.config.token_storage_object_store:
             # Store object store name, will be resolved later when builder context is available
             self._token_storage_object_store_name = self.config.token_storage_object_store
-            logger.info(f"Configured to use object store '{self._token_storage_object_store_name}' for token storage")
+            logger.info("Configured to use object store '%s' for token storage", self._token_storage_object_store_name)
         else:
             # Default: use in-memory token storage
             from nat.authentication.token_storage import InMemoryTokenStorage
             self._token_storage = InMemoryTokenStorage()
 
+    def _invalidate_cached_registration(self, reason: str) -> None:
+        """Invalidate cached OAuth client registration and auth provider."""
+        previous_client_id = self._cached_credentials.client_id if self._cached_credentials else None
+        self._credentials_cache_time = None
+        self._cached_credentials = None
+        self._auth_code_provider = None
+        logger.info("Invalidated cached OAuth2 registration: reason=%s previous_client_id=%s",
+                    reason,
+                    previous_client_id)
+
+    def _is_cached_credentials_expired(self) -> bool:
+        """Check if cached credentials are expired."""
+        if self._credentials_cache_time is None:
+            return True
+
+        # `0` means "do not reuse across attempts", not "invalidate within the same attempt".
+        if self.config.oauth_client_ttl == 0:
+            return False
+
+        return (time.monotonic() - self._credentials_cache_time) >= self.config.oauth_client_ttl
+
+    def _is_redirect_uri_registration_error(self, error: Exception) -> bool:
+        """Check if error indicates AS rejected redirect URI registration for this client."""
+        msg = str(error).lower()
+        return ("redirect uri" in msg and "not registered for client" in msg)
+
+    async def _discover_and_register_locked(self,
+                                            response: httpx.Response | None = None,
+                                            *,
+                                            force_refresh: bool = False):
+        """Serialize discovery/registration to avoid races across concurrent auth flows."""
+        async with self._discover_register_lock:
+            if force_refresh:
+                self._invalidate_cached_registration(reason="forced-refresh")
+            await self._discover_and_register(response=response)
+
     def _set_custom_auth_callback(self,
                                   auth_callback: Callable[[OAuth2AuthCodeFlowProviderConfig, AuthFlowType],
                                                           Awaitable[AuthenticatedContext]]):
@@ -364,9 +404,19 @@ async def authenticate(self, user_id: str | None = None, **kwargs) -> AuthResult
 
         response = kwargs.get('response')
         if response and response.status_code == 401:
-            await self._discover_and_register(response=response)
+            await self._discover_and_register_locked(response=response)
 
-        return await self._nat_oauth2_authenticate(user_id=user_id)
+        try:
+            return await self._nat_oauth2_authenticate(user_id=user_id)
+        except RuntimeError as e:
+            # Some AS deployments intermittently reject authorize requests with
+            # "redirect URI not registered" for a cached client_id. Force one
+            # re-registration attempt to self-heal before failing the request.
+            if self._is_redirect_uri_registration_error(e):
+                logger.warning("Detected redirect URI registration error; forcing re-registration and retry")
+                await self._discover_and_register_locked(response=response, force_refresh=True)
+                return await self._nat_oauth2_authenticate(user_id=user_id)
+            raise
 
     @property
     def _effective_scopes(self) -> list[str]:
@@ -382,12 +432,12 @@ async def _discover_and_register(self, response: httpx.Response | None = None):
         self._cached_endpoints, endpoints_changed = await self._discoverer.discover(response=response)
         if endpoints_changed:
             logger.info("OAuth2 endpoints: %s", self._cached_endpoints)
-            self._cached_credentials = None  # invalidate credentials tied to old AS
-            self._auth_code_provider = None
+            self._invalidate_cached_registration(reason="endpoints-changed")
         effective_scopes = self._effective_scopes
 
         # Client registration
-        if not self._cached_credentials:
+        if (not self._cached_credentials or self.config.oauth_client_ttl == 0 or self._is_cached_credentials_expired()):
+            self._invalidate_cached_registration(reason="registration-expired")
             if self.config.client_id:
                 # Manual registration mode
                 self._cached_credentials = OAuth2Credentials(
@@ -400,12 +450,19 @@ async def _discover_and_register(self, response: httpx.Response | None = None):
                 self._cached_credentials = await self._registrar.register(self._cached_endpoints, effective_scopes)
                 logger.info("Registered OAuth2 client: %s", self._cached_credentials.client_id)
 
+            self._credentials_cache_time = time.monotonic()
+
     async def _nat_oauth2_authenticate(self, user_id: str | None = None) -> AuthResult:
         """Perform the OAuth2 flow using MCP-specific authentication flow handler."""
         from nat.authentication.oauth2.oauth2_auth_code_flow_provider import OAuth2AuthCodeFlowProvider
 
-        if not self._cached_endpoints or not self._cached_credentials:
+        if (not self._cached_endpoints or not self._cached_credentials or self._is_cached_credentials_expired()):
             # if discovery is yet to to be done return empty auth result
+            logger.warning(
+                "OAuth2 endpoints or credentials not available or expired for user_id=%s. "
+                "Discovery and registration must be performed before authentication. "
+                "Returning empty AuthResult.",
+                user_id)
             return AuthResult(credentials=[], token_expires_at=None, raw={})
 
         endpoints = self._cached_endpoints
@@ -422,8 +479,10 @@ async def _nat_oauth2_authenticate(self, user_id: str | None = None) -> AuthResu
                 logger.info(f"Initialized token storage with object store '{self._token_storage_object_store_name}'")
             except Exception as e:
                 logger.warning(
-                    f"Failed to resolve object store '{self._token_storage_object_store_name}' for token storage: {e}. "
-                    "Falling back to in-memory storage.")
+                    "Failed to resolve object store '%s' for token storage: %s. Falling back to in-memory storage.",
+                    self._token_storage_object_store_name,
+                    e,
+                )
                 from nat.authentication.token_storage import InMemoryTokenStorage
                 self._token_storage = InMemoryTokenStorage()
 

packages/nvidia_nat_mcp/src/nat/plugins/mcp/auth/auth_provider_config.py

@@ -56,6 +56,12 @@ class MCPOAuth2ProviderConfig(AuthProviderBaseConfig, name="mcp_oauth2"):
     default_user_id: str | None = Field(default=None, description="Default user ID for authentication")
     allow_default_user_id_for_tool_calls: bool = Field(default=True, description="Allow default user ID for tool calls")
 
+    # OAuth client credential caching
+    oauth_client_ttl: float = Field(default=270.0,
+                                    ge=0.0,
+                                    description="Amount of time, in seconds, to cache oauth client credentials. "
+                                    "Setting this to 0 disables caching.")
+
     # Token storage configuration
     token_storage_object_store: str | None = Field(
         default=None,

packages/nvidia_nat_mcp/tests/client/test_mcp_auth_provider.py

@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import asyncio
 from unittest.mock import AsyncMock
 from unittest.mock import MagicMock
 from unittest.mock import patch
@@ -752,3 +753,75 @@ async def test_effective_scopes_config_overrides_discovered(self, mock_config):
 
         scopes = provider._effective_scopes
         assert scopes == ['config_scope']  # Config should take precedence
+
+    @pytest.mark.parametrize("oauth_client_ttl", [0.01, 0.0], ids=["0.01", "disabled"])
+    async def test_oauth_client_ttl(self, mock_endpoints, oauth_client_ttl):
+        """Test that expired oauth_client_ttl causes re-registration with a new client_id."""
+        config = MCPOAuth2ProviderConfig(
+            server_url="https://example.com/mcp",  # type: ignore
+            redirect_uri="https://example.com/callback",  # type: ignore
+            enable_dynamic_registration=True,
+            oauth_client_ttl=oauth_client_ttl,
+        )
+        provider = MCPOAuth2Provider(config)
+
+        first_credentials = OAuth2Credentials(client_id="first_client_id", client_secret="secret")
+        second_credentials = OAuth2Credentials(client_id="second_client_id", client_secret="secret")
+
+        with patch.object(provider._discoverer, 'discover') as mock_discover:
+            mock_discover.return_value = (mock_endpoints, False)
+
+            with patch.object(provider._registrar, 'register') as mock_register:
+                mock_register.return_value = first_credentials
+
+                await provider._discover_and_register()
+
+                assert provider._cached_credentials.client_id == "first_client_id"
+                assert provider._auth_code_provider is None  # not built yet
+                first_cache_time = provider._credentials_cache_time
+
+                # Wait for TTL to expire
+                await asyncio.sleep(oauth_client_ttl)
+
+                mock_register.return_value = second_credentials
+                await provider._discover_and_register()
+
+                assert provider._cached_credentials.client_id == "second_client_id"
+                assert provider._credentials_cache_time > first_cache_time
+                assert provider._auth_code_provider is None  # reset on re-registration
+                assert mock_register.call_count == 2
+
+    async def test_oauth_client_ttl_not_expired(self, mock_endpoints):
+        """Test that credentials are not refreshed when oauth_client_ttl has not elapsed."""
+        config = MCPOAuth2ProviderConfig(
+            server_url="https://example.com/mcp",  # type: ignore
+            redirect_uri="https://example.com/callback",  # type: ignore
+            enable_dynamic_registration=True,
+            oauth_client_ttl=100,
+        )
+        provider = MCPOAuth2Provider(config)
+
+        first_credentials = OAuth2Credentials(client_id="first_client_id", client_secret="secret")
+        second_credentials = OAuth2Credentials(client_id="second_client_id", client_secret="secret")
+
+        with patch.object(provider._discoverer, 'discover') as mock_discover:
+            mock_discover.return_value = (mock_endpoints, False)
+
+            with patch.object(provider._registrar, 'register') as mock_register:
+                mock_register.return_value = first_credentials
+
+                await provider._discover_and_register()
+
+                assert provider._cached_credentials.client_id == "first_client_id"
+                first_cache_time = provider._credentials_cache_time
+
+                # Wait well under the TTL
+                await asyncio.sleep(0.01)
+
+                mock_register.return_value = second_credentials
+                await provider._discover_and_register()
+
+                # Credentials should be unchanged — no re-registration occurred
+                assert provider._cached_credentials.client_id == "first_client_id"
+                assert provider._credentials_cache_time == first_cache_time
+                assert mock_register.call_count == 1

packages/nvidia_nat_mcp/tests/client/test_mcp_token_storage.py

@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import time
 from datetime import UTC
 from datetime import datetime
 from datetime import timedelta
@@ -338,6 +339,7 @@ async def test_token_storage_lazy_resolution(self, mock_config, sample_auth_resu
             token_url="https://auth.example.com/token",  # type: ignore
         )
         provider._cached_credentials = OAuth2Credentials(client_id="test", client_secret="secret")
+        provider._credentials_cache_time = time.time()  # A non-none value to indicate credentials are "cached"
 
         # Trigger authentication which should resolve the object store
         with patch('nat.authentication.oauth2.oauth2_auth_code_flow_provider.OAuth2AuthCodeFlowProvider'