chore: fix mypy strict type errors in adk auth

Co-authored-by: George Weale <gweale@google.com>
PiperOrigin-RevId: 945169969

Author: GWeale

src/google/adk/auth/__init__.py

@@ -14,7 +14,6 @@
 
 from __future__ import annotations
 
-import importlib
 from typing import TYPE_CHECKING
 
 from .auth_credential import AuthCredential
@@ -30,7 +29,9 @@
   from .auth_handler import AuthHandler
 
 
-def __getattr__(name: str):
-  if name == 'AuthHandler':
-    return importlib.import_module(f'{__name__}.auth_handler').AuthHandler
-  raise AttributeError(f'module {__name__!r} has no attribute {name!r}')
+def __getattr__(name: str) -> type[AuthHandler]:
+  if name == "AuthHandler":
+    from .auth_handler import AuthHandler
+
+    return AuthHandler
+  raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

src/google/adk/auth/auth_handler.py

@@ -81,7 +81,7 @@ async def parse_and_store_auth_response(self, state: State) -> None:
     state[credential_key] = await self.exchange_auth_token()
 
   def _validate(self) -> None:
-    if not self.auth_scheme:
+    if not self.auth_config.auth_scheme:
       raise ValueError("auth_scheme is empty.")
 
   def get_auth_response(self, state: State) -> AuthCredential:
@@ -236,9 +236,10 @@ def generate_auth_uri(
     )
 
     exchanged_auth_credential = auth_credential.model_copy(deep=True)
-    exchanged_auth_credential.oauth2.auth_uri = uri
-    exchanged_auth_credential.oauth2.state = state
-    if code_verifier:
-      exchanged_auth_credential.oauth2.code_verifier = code_verifier
+    if exchanged_auth_credential.oauth2 is not None:
+      exchanged_auth_credential.oauth2.auth_uri = uri
+      exchanged_auth_credential.oauth2.state = state
+      if code_verifier:
+        exchanged_auth_credential.oauth2.code_verifier = code_verifier
 
     return exchanged_auth_credential

src/google/adk/auth/auth_preprocessor.py

@@ -34,7 +34,7 @@
 # Prefix used by toolset auth credential IDs.
 # Auth requests with this prefix are for toolset authentication (before tool
 # listing) and don't require resuming a function call.
-TOOLSET_AUTH_CREDENTIAL_ID_PREFIX = '_adk_toolset_auth_'
+TOOLSET_AUTH_CREDENTIAL_ID_PREFIX = "_adk_toolset_auth_"
 
 
 async def _store_auth_and_collect_resume_targets(
@@ -132,7 +132,7 @@ async def run_async(
       self, invocation_context: InvocationContext, llm_request: LlmRequest
   ) -> AsyncGenerator[Event, None]:
     agent = invocation_context.agent
-    if not hasattr(agent, 'canonical_tools'):
+    if agent is None or not hasattr(agent, "canonical_tools"):
       return
     events = invocation_context._get_events(current_branch=True)
     if not events:
@@ -147,7 +147,7 @@ async def run_async(
         last_event_with_content = event
         break
 
-    if not last_event_with_content or last_event_with_content.author != 'user':
+    if not last_event_with_content or last_event_with_content.author != "user":
       return
 
     responses = last_event_with_content.get_function_responses()

src/google/adk/auth/auth_provider_registry.py

@@ -26,7 +26,7 @@
 class AuthProviderRegistry:
   """Registry for auth provider instances."""
 
-  def __init__(self):
+  def __init__(self) -> None:
     self._providers: dict[type[AuthScheme], BaseAuthProvider] = {}
 
   def register(

src/google/adk/auth/auth_schemes.py

@@ -68,7 +68,7 @@ class OAuthGrantType(str, Enum):
   PASSWORD = "password"
 
   @staticmethod
-  def from_flow(flow: OAuthFlows) -> "OAuthGrantType":
+  def from_flow(flow: OAuthFlows) -> Optional["OAuthGrantType"]:
     """Converts an OAuthFlows object to a OAuthGrantType."""
     if flow.clientCredentials:
       return OAuthGrantType.CLIENT_CREDENTIALS

src/google/adk/auth/auth_tool.py

@@ -16,6 +16,7 @@
 
 import hashlib
 import json
+from typing import Any
 from typing import Optional
 
 from pydantic import BaseModel
@@ -36,7 +37,8 @@ def _stable_model_digest(model: BaseModel) -> str:
   """
   if getattr(model, "model_extra", None):
     model = model.model_copy(deep=True)
-    model.model_extra.clear()
+    if model.model_extra is not None:
+      model.model_extra.clear()
 
   dumped = model.model_dump(by_alias=True, exclude_none=True, mode="json")
   canonical_json = json.dumps(
@@ -79,12 +81,12 @@ class AuthConfig(BaseModelWithConfig):
   service.
   """
 
-  def __init__(self, **data):
+  def __init__(self, **data: Any) -> None:
     super().__init__(**data)
     if self.credential_key:
       return
     for obj in (self.raw_auth_credential, self.auth_scheme):
-      if not obj or not getattr(obj, "model_extra", None):
+      if not obj or not obj.model_extra:
         continue
       for key in ("credential_key", "credentialKey"):
         value = obj.model_extra.get(key)
@@ -94,7 +96,7 @@ def __init__(self, **data):
     self.credential_key = self.get_credential_key()
 
   @deprecated("This method is deprecated. Use credential_key instead.")
-  def get_credential_key(self):
+  def get_credential_key(self) -> str:
     """Builds a stable key based on auth_scheme and raw_auth_credential.
 
     This is used to save/load credentials to/from a credential service when
@@ -105,7 +107,8 @@ def get_credential_key(self):
 
     if auth_scheme.model_extra:
       auth_scheme = auth_scheme.model_copy(deep=True)
-      auth_scheme.model_extra.clear()
+      if auth_scheme.model_extra is not None:
+        auth_scheme.model_extra.clear()
 
     type_ = auth_scheme.type_
     type_name = type_.name if type_ and hasattr(type_, "name") else str(type_)
@@ -118,7 +121,8 @@ def get_credential_key(self):
     auth_credential = self.raw_auth_credential
     if auth_credential and auth_credential.model_extra:
       auth_credential = auth_credential.model_copy(deep=True)
-      auth_credential.model_extra.clear()
+      if auth_credential.model_extra is not None:
+        auth_credential.model_extra.clear()
     if auth_credential and auth_credential.oauth2:
       auth_credential = auth_credential.model_copy(deep=True)
       if auth_credential.oauth2:

src/google/adk/auth/credential_manager.py

@@ -27,7 +27,6 @@
 from .auth_credential import AuthCredential
 from .auth_credential import AuthCredentialTypes
 from .auth_provider_registry import AuthProviderRegistry
-from .auth_schemes import AuthScheme
 from .auth_schemes import AuthSchemeType
 from .auth_schemes import CustomAuthScheme
 from .auth_schemes import ExtendedOAuth2
@@ -43,7 +42,8 @@
 
 
 def _rehydrate_custom_scheme(
-    scheme: CustomAuthScheme, supported_schemes: Sequence[type[AuthScheme]]
+    scheme: CustomAuthScheme,
+    supported_schemes: Sequence[type[CustomAuthScheme]],
 ) -> CustomAuthScheme:
   """Rehydrate a CustomAuthScheme into one of the given supported_schemes."""
   incoming_type = scheme.type_
@@ -231,10 +231,11 @@ async def get_auth_credential(
     await self._validate_credential()
 
     # Step 2: Check if credential is already ready (no processing needed)
-    if self._is_credential_ready():
+    raw_auth_credential = self._auth_config.raw_auth_credential
+    if self._is_credential_ready() and raw_auth_credential is not None:
       # Return a copy to avoid leaking mutations across invocations/users when
       # tools share a long-lived AuthConfig instance.
-      return self._auth_config.raw_auth_credential.model_copy(deep=True)
+      return raw_auth_credential.model_copy(deep=True)
 
     # Step 3: Try to load existing processed credential
     credential = await self._load_existing_credential(context)
@@ -438,7 +439,7 @@ def _missing_oauth_info(self) -> bool:
     auth_scheme = self._auth_config.auth_scheme
     if isinstance(auth_scheme, OAuth2):
       flows = auth_scheme.flows
-      return (
+      return bool(
           flows.implicit
           and not flows.implicit.authorizationUrl
           or flows.password

src/google/adk/auth/exchanger/credential_exchanger_registry.py

@@ -28,7 +28,7 @@
 class CredentialExchangerRegistry:
   """Registry for credential exchanger instances."""
 
-  def __init__(self):
+  def __init__(self) -> None:
     self._exchangers: Dict[AuthCredentialTypes, BaseCredentialExchanger] = {}
 
   def register(

src/google/adk/auth/exchanger/oauth2_credential_exchanger.py

@@ -186,7 +186,7 @@ async def _exchange_authorization_code(
         boolean indicating whether the credential was exchanged.
     """
     client, token_endpoint = create_oauth2_session(auth_scheme, auth_credential)
-    if not client:
+    if not client or not auth_credential.oauth2:
       logger.warning(
           "Could not create OAuth2 session for authorization code exchange"
       )

src/google/adk/auth/refresher/credential_refresher_registry.py

@@ -29,7 +29,7 @@
 class CredentialRefresherRegistry:
   """Registry for credential refresher instances."""
 
-  def __init__(self):
+  def __init__(self) -> None:
     self._refreshers: Dict[AuthCredentialTypes, BaseCredentialRefresher] = {}
 
   def register(