### Code Review Fixes

**Round 1 - Initial Review:**
- ✅ Added ID-JAG expiry tracking to prevent stale tokens
- ✅ Enhanced audience override with WARNING level logging
- ✅ Documented error handling delegation pattern
- ✅ Added metadata stability documentation
- ✅ Fixed empty scope handling with strip() check
- ✅ Added negative JWT decode tests

**Round 2 - Systematic Review:**
- ✅ Made ID-JAG expiry configurable (DEFAULT_ID_JAG_EXPIRY_SECONDS = 900)
- ✅ Added comprehensive thread safety documentation.
- ✅ Moved time import to module level
- ✅ Extracted magic numbers to constants
- ✅ Enhanced logging with default expiry notifications

**Round 3 - Final Fixes:**
- ✅ Changed validation exceptions to OAuthFlowError for consistency
- ✅ Added edge case test for oauth_metadata=None
- ✅ Fixed critical bug: instance variable vs class constant

### Test Coverage

**Statistics:**
- 43 test cases covering all code paths
- 100% code coverage (166 statements, 44 branches)
- Edge cases: expired tokens, malformed JWT, HTTP errors, empty scope
- Conformance test integration: `auth/cross-app-access-complete-flow`

### Breaking Changes

**None.** All changes are backward compatible with optional parameters.

Author: BinoyOza-okta

.github/actions/conformance/client.py

@@ -414,7 +414,9 @@ async def run_cross_app_access_complete_flow(server_url: str) -> None:
 
 async def _run_auth_session(server_url: str, oauth_auth: OAuthClientProvider) -> None:
     """Common session logic for all OAuth flows."""
-    client = httpx.AsyncClient(auth=oauth_auth, timeout=30.0)
+    # Allow timeout to be configured via environment variable for different test scenarios
+    timeout = float(os.environ.get("MCP_CONFORMANCE_TIMEOUT", "30.0"))
+    client = httpx.AsyncClient(auth=oauth_auth, timeout=timeout)
     async with streamable_http_client(url=server_url, http_client=client) as (read_stream, write_stream):
         async with ClientSession(
             read_stream, write_stream, elicitation_callback=default_elicitation_callback

README.v2.md

@@ -2630,6 +2630,8 @@ async def advanced_manual_flow() -> None:
     async with httpx.AsyncClient() as client:
         # Step 1: Exchange ID token for ID-JAG
         id_jag = await enterprise_auth.exchange_token_for_id_jag(client)
+        # WARNING: Only log tokens in development/testing environments
+        # In production, NEVER log tokens or token fragments as they are sensitive credentials
         print(f"Obtained ID-JAG: {id_jag[:50]}...")
 
         # Step 2: Build JWT bearer grant request
@@ -2646,13 +2648,54 @@ async def advanced_manual_flow() -> None:
             token_type=token_data["token_type"],
             expires_in=token_data.get("expires_in"),
         )
+        # WARNING: In production, do not log token expiry or any token information
         print(f"Access token obtained, expires in: {access_token.expires_in}s")
 
         # Use the access token for API calls
         _ = {"Authorization": f"Bearer {access_token.access_token}"}
         # ... make authenticated requests with headers
 
 
+async def token_refresh_example() -> None:
+    """Example showing how to refresh tokens when they expire.
+
+    When your access token expires, you need to obtain a fresh ID token
+    from your enterprise IdP and use the refresh helper method.
+    """
+    # Initial setup
+    id_token = await get_id_token_from_idp()
+
+    token_exchange_params = TokenExchangeParameters.from_id_token(
+        id_token=id_token,
+        mcp_server_auth_issuer="https://auth.mcp-server.example.com",
+        mcp_server_resource_id="https://mcp-server.example.com",
+    )
+
+    enterprise_auth = EnterpriseAuthOAuthClientProvider(
+        server_url="https://mcp-server.example.com",
+        client_metadata=OAuthClientMetadata(
+            redirect_uris=[AnyUrl("http://localhost:3000/callback")],
+        ),
+        storage=SimpleTokenStorage(),
+        idp_token_endpoint="https://your-idp.com/oauth2/v1/token",
+        token_exchange_params=token_exchange_params,
+    )
+
+    _ = httpx.AsyncClient(auth=enterprise_auth, timeout=30.0)
+
+    # Use the client for MCP operations...
+    # ... time passes and token expires ...
+
+    # When token expires, get a fresh ID token from your IdP
+    new_id_token = await get_id_token_from_idp()
+
+    # Refresh the authentication using the helper method
+    await enterprise_auth.refresh_with_new_id_token(new_id_token)
+
+    # Next API call will automatically use the refreshed tokens
+    # No need to recreate the client or session
+
+
 if __name__ == "__main__":
     asyncio.run(main())
 ```

examples/snippets/clients/enterprise_managed_auth_client.py

@@ -188,6 +188,8 @@ async def advanced_manual_flow() -> None:
     async with httpx.AsyncClient() as client:
         # Step 1: Exchange ID token for ID-JAG
         id_jag = await enterprise_auth.exchange_token_for_id_jag(client)
+        # WARNING: Only log tokens in development/testing environments
+        # In production, NEVER log tokens or token fragments as they are sensitive credentials
         print(f"Obtained ID-JAG: {id_jag[:50]}...")
 
         # Step 2: Build JWT bearer grant request
@@ -204,12 +206,53 @@ async def advanced_manual_flow() -> None:
             token_type=token_data["token_type"],
             expires_in=token_data.get("expires_in"),
         )
+        # WARNING: In production, do not log token expiry or any token information
         print(f"Access token obtained, expires in: {access_token.expires_in}s")
 
         # Use the access token for API calls
         _ = {"Authorization": f"Bearer {access_token.access_token}"}
         # ... make authenticated requests with headers
 
 
+async def token_refresh_example() -> None:
+    """Example showing how to refresh tokens when they expire.
+
+    When your access token expires, you need to obtain a fresh ID token
+    from your enterprise IdP and use the refresh helper method.
+    """
+    # Initial setup
+    id_token = await get_id_token_from_idp()
+
+    token_exchange_params = TokenExchangeParameters.from_id_token(
+        id_token=id_token,
+        mcp_server_auth_issuer="https://auth.mcp-server.example.com",
+        mcp_server_resource_id="https://mcp-server.example.com",
+    )
+
+    enterprise_auth = EnterpriseAuthOAuthClientProvider(
+        server_url="https://mcp-server.example.com",
+        client_metadata=OAuthClientMetadata(
+            redirect_uris=[AnyUrl("http://localhost:3000/callback")],
+        ),
+        storage=SimpleTokenStorage(),
+        idp_token_endpoint="https://your-idp.com/oauth2/v1/token",
+        token_exchange_params=token_exchange_params,
+    )
+
+    _ = httpx.AsyncClient(auth=enterprise_auth, timeout=30.0)
+
+    # Use the client for MCP operations...
+    # ... time passes and token expires ...
+
+    # When token expires, get a fresh ID token from your IdP
+    new_id_token = await get_id_token_from_idp()
+
+    # Refresh the authentication using the helper method
+    await enterprise_auth.refresh_with_new_id_token(new_id_token)
+
+    # Next API call will automatically use the refreshed tokens
+    # No need to recreate the client or session
+
+
 if __name__ == "__main__":
     asyncio.run(main())