fix(auth): drop a stored credential the server rejected with 401/403

src/google/adk/auth/credential_manager.py

@@ -23,6 +23,7 @@
 from weakref import WeakValueDictionary
 
 from fastapi.openapi.models import OAuth2
+import httpx
 
 from ..agents.callback_context import CallbackContext
 from ..tools.openapi_tool.auth.credential_exchangers.service_account_exchanger import ServiceAccountCredentialExchanger
@@ -44,6 +45,35 @@
 logger = logging.getLogger("google_adk." + __name__)
 
 
+_REJECTED_CREDENTIAL_STATUS_CODES = frozenset({401, 403})
+
+
+def _is_unauthorized_error(error: BaseException) -> bool:
+  """Whether an error says the server rejected the credential (HTTP 401/403).
+
+  Looks through the `__cause__` / `__context__` chain and the members of
+  exception groups, because transports and SDK wrappers routinely re-raise the
+  `httpx.HTTPStatusError` inside another exception.
+  """
+  pending: list[BaseException] = [error]
+  seen: set[int] = set()
+  while pending:
+    current = pending.pop()
+    if id(current) in seen:
+      continue
+    seen.add(id(current))
+    if (
+        isinstance(current, httpx.HTTPStatusError)
+        and current.response.status_code in _REJECTED_CREDENTIAL_STATUS_CODES
+    ):
+      return True
+    pending.extend(getattr(current, "exceptions", ()))
+    for linked in (current.__cause__, current.__context__):
+      if linked is not None:
+        pending.append(linked)
+  return False
+
+
 def _has_unredeemed_auth_code(credential: AuthCredential) -> bool:
   """Whether the credential holds an authorization code that yielded no token.
 
@@ -226,6 +256,32 @@ async def request_credential(self, context: CallbackContext) -> None:
       )
     context.request_credential(self._auth_config)
 
+  async def discard_stored_credential(self, context: CallbackContext) -> None:
+    """Forgets the credential kept in the credential service.
+
+    Call this when the server rejected the credential (HTTP 401/403): the next
+    `get_auth_credential` then finds nothing to reuse, so the caller goes
+    through `request_credential` again instead of replaying the rejected
+    credential for the rest of the session.
+
+    Credentials that are never stored (a statically configured API key or HTTP
+    credential, or a custom auth scheme, which keeps its own state) and
+    invocations without a credential service are left alone.
+    """
+    if isinstance(self._auth_config.auth_scheme, CustomAuthScheme):
+      return
+    if self._is_credential_ready():
+      return
+    if not context._invocation_context.credential_service:
+      return
+    async with self._get_credential_lock(context):
+      await self._save_credential(context, None)
+    logger.info(
+        "Discarded the stored credential for auth scheme %s after the server"
+        " rejected it.",
+        type(self._auth_config.auth_scheme).__name__,
+    )
+
   async def get_auth_credential(
       self, context: CallbackContext
   ) -> Optional[AuthCredential]:
@@ -474,9 +530,12 @@ async def _validate_credential(self) -> None:
     # Additional validation can be added here
 
   async def _save_credential(
-      self, context: CallbackContext, credential: AuthCredential
+      self, context: CallbackContext, credential: Optional[AuthCredential]
   ) -> None:
-    """Save credential to credential service if available."""
+    """Save credential to credential service if available.
+
+    `None` clears the stored credential.
+    """
     credential_service = context._invocation_context.credential_service
     if credential_service:
       auth_config_to_save = self._auth_config.model_copy(deep=True)

src/google/adk/tools/authenticated_function_tool.py

@@ -25,6 +25,7 @@
 
 from ..auth.auth_credential import AuthCredential
 from ..auth.auth_tool import AuthConfig
+from ..auth.credential_manager import _is_unauthorized_error
 from ..auth.credential_manager import CredentialManager
 from .function_tool import FunctionTool
 from .tool_context import ToolContext
@@ -88,9 +89,16 @@ async def run_async(
         await self._credentials_manager.request_credential(tool_context)
         return self._response_for_auth_required or "Pending User Authorization."
 
-    return await self._run_async_impl(
-        args=args, tool_context=tool_context, credential=credential
-    )
+    try:
+      return await self._run_async_impl(
+          args=args, tool_context=tool_context, credential=credential
+      )
+    except Exception as e:
+      # The server rejected the credential: drop it so the next call asks for
+      # a new one instead of replaying this one.
+      if self._credentials_manager and _is_unauthorized_error(e):
+        await self._credentials_manager.discard_stored_credential(tool_context)
+      raise
 
   async def _run_async_impl(
       self,

src/google/adk/tools/base_authenticated_tool.py

@@ -23,6 +23,7 @@
 
 from ..auth.auth_credential import AuthCredential
 from ..auth.auth_tool import AuthConfig
+from ..auth.credential_manager import _is_unauthorized_error
 from ..auth.credential_manager import CredentialManager
 from .base_tool import BaseTool
 from .tool_context import ToolContext
@@ -89,14 +90,33 @@ async def run_async(
         await self._credentials_manager.request_credential(tool_context)
         return self._response_for_auth_required or "Pending User Authorization."
 
-    return await self._run_async_impl(
-        args=args,
-        tool_context=tool_context,
-        # A tool with no credentials manager runs unauthenticated, so this is
-        # None on that path. Widening the abstract signature instead would
-        # invalidate every subclass that declares the narrower type.
-        credential=cast(AuthCredential, credential),
-    )
+    try:
+      return await self._run_async_impl(
+          args=args,
+          tool_context=tool_context,
+          # A tool with no credentials manager runs unauthenticated, so this is
+          # None on that path. Widening the abstract signature instead would
+          # invalidate every subclass that declares the narrower type.
+          credential=cast(AuthCredential, credential),
+      )
+    except Exception as e:
+      if _is_unauthorized_error(e):
+        await self._discard_rejected_credential(tool_context)
+      raise
+
+  async def _discard_rejected_credential(
+      self, tool_context: ToolContext
+  ) -> None:
+    """Drops the stored credential after the server rejected it.
+
+    `run_async` calls this when `_run_async_impl` raises an HTTP 401/403
+    (`httpx.HTTPStatusError`, also when wrapped in another exception). A
+    subclass that reports an unauthorized outcome as a return value instead
+    should call it itself. The next call then goes through
+    `request_credential` rather than reusing the rejected credential.
+    """
+    if self._credentials_manager:
+      await self._credentials_manager.discard_stored_credential(tool_context)
 
   @abstractmethod
   async def _run_async_impl(