chore(ve_identity): Make OAuth2 flow and scopes parameters optional

Updated OAuth2-related classes and functions to make 'auth_flow' and 'scopes' parameters optional, allowing control plane defaults to be used if not provided.

Author: loveyana

tests/test_ve_identity_auth_config.py

@@ -103,7 +103,7 @@ def on_auth_url_callback(url: str):
 
     def test_oauth2_auth_empty_scopes(self):
         """Test that empty scopes raises ValueError."""
-        with pytest.raises(ValueError, match="scopes cannot be empty"):
+        with pytest.raises(ValueError, match="scopes cannot be an empty list"):
             oauth2_auth(
                 provider_name="github",
                 scopes=[],

veadk/agent.py

@@ -298,15 +298,13 @@ async def run(
         final_output = ""
         for _prompt in prompt:
             message = types.Content(role="user", parts=[types.Part(text=_prompt)])
-            final_output = await self._run(
-                runner, user_id, session_id, message, stream, auth_request_processor
-            )
+            final_output = await self._run(runner, user_id, session_id, message, stream, auth_request_processor)
 
         # VeADK features
         if save_session_to_memory:
-            assert (
-                self.long_term_memory is not None
-            ), "Long-term memory is not initialized in agent"
+            assert self.long_term_memory is not None, (
+                "Long-term memory is not initialized in agent"
+            )
             session = await session_service.get_session(
                 app_name=app_name,
                 user_id=user_id,

veadk/integrations/ve_identity/auth_config.py

@@ -21,7 +21,7 @@
 from abc import ABC, abstractmethod
 from typing import Any, Callable, List, Literal, Optional, Union
 
-from pydantic import BaseModel, model_validator, field_validator
+from pydantic import BaseModel, field_validator
 
 from veadk.integrations.ve_identity.models import OAuth2AuthPoller
 from veadk.integrations.ve_identity.identity_client import IdentityClient
@@ -35,6 +35,7 @@ def _get_default_region() -> str:
     """
     try:
         from veadk.config import settings
+
         return settings.veidentity.region
     except Exception:
         # Fallback to default if config loading fails
@@ -52,8 +53,8 @@ class AuthConfig(BaseModel, ABC):
 
     def __init__(self, **data):
         """Initialize AuthConfig with default region from VeADK config if not provided."""
-        if 'region' not in data or data['region'] is None:
-            data['region'] = _get_default_region()
+        if "region" not in data or data["region"] is None:
+            data["region"] = _get_default_region()
         super().__init__(**data)
 
     @field_validator("provider_name")
@@ -82,10 +83,10 @@ def auth_type(self) -> str:
 class OAuth2AuthConfig(AuthConfig):
     """OAuth2 authentication configuration."""
 
-    # Required fields
-    scopes: List[str]
-    auth_flow: Literal["M2M", "USER_FEDERATION"]
-    # Optional fields
+    # Optional fields - control plane will use defaults if not provided
+    scopes: Optional[List[str]] = None
+    auth_flow: Optional[Literal["M2M", "USER_FEDERATION"]] = None
+    # Additional optional fields
     callback_url: Optional[str] = None
     force_authentication: bool = False
     response_for_auth_required: Optional[Union[dict, str]] = None
@@ -94,10 +95,18 @@ class OAuth2AuthConfig(AuthConfig):
 
     @field_validator("scopes")
     @classmethod
-    def validate_scopes_not_empty(cls, v: List[str]) -> List[str]:
-        """Validate that scopes list is not empty and contains valid scope strings."""
+    def validate_scopes_not_empty(cls, v: Optional[List[str]]) -> Optional[List[str]]:
+        """Validate that scopes list is not empty and contains valid scope strings.
+
+        If scopes is None, the control plane will use default scopes.
+        """
+        if v is None:
+            return None
+
         if not v:
-            raise ValueError("scopes cannot be empty")
+            raise ValueError(
+                "scopes cannot be an empty list; use None to use control plane defaults"
+            )
 
         # Validate each scope is not empty
         for scope in v:
@@ -128,15 +137,6 @@ def validate_callback_url(cls, v: Optional[str]) -> Optional[str]:
                 raise ValueError("callback_url must be a valid HTTP/HTTPS URL")
         return v
 
-    @model_validator(mode="after")
-    def _validate_required_fields(self):
-        """Validate required fields."""
-        if not self.scopes:
-            raise ValueError("scopes is required for OAuth2AuthConfig")
-        if not self.auth_flow:
-            raise ValueError("auth_flow is required for OAuth2AuthConfig")
-        return self
-
     @property
     def auth_type(self) -> str:
         return "oauth2"
@@ -201,8 +201,8 @@ def workload_auth(
 
 def oauth2_auth(
     provider_name: str,
-    scopes: List[str],
-    auth_flow: Literal["M2M", "USER_FEDERATION"],
+    scopes: Optional[List[str]] = None,
+    auth_flow: Optional[Literal["M2M", "USER_FEDERATION"]] = None,
     callback_url: Optional[str] = None,
     force_authentication: bool = False,
     response_for_auth_required: Optional[Union[dict, str]] = None,
@@ -215,8 +215,10 @@ def oauth2_auth(
 
     Args:
         provider_name: Name of the credential provider.
-        scopes: List of OAuth2 scopes.
-        auth_flow: Authentication flow type ("M2M" or "USER_FEDERATION").
+        scopes: Optional list of OAuth2 scopes. If not provided, the control plane
+               will use the default configured scopes for the provider.
+        auth_flow: Optional authentication flow type ("M2M" or "USER_FEDERATION").
+                  If not provided, the control plane will use the default configured flow.
         callback_url: Optional callback URL for OAuth2.
         force_authentication: Whether to force authentication.
         response_for_auth_required: Response to return when auth is required.

veadk/integrations/ve_identity/auth_mixins.py

@@ -343,8 +343,8 @@ class OAuth2AuthMixin(BaseAuthMixin):
     def __init__(
         self,
         *,
-        scopes: List[str],
-        auth_flow: Literal["M2M", "USER_FEDERATION"],
+        scopes: Optional[List[str]] = None,
+        auth_flow: Optional[Literal["M2M", "USER_FEDERATION"]] = None,
         callback_url: Optional[str] = None,
         force_authentication: bool = False,
         response_for_auth_required: Optional[str] = None,
@@ -356,9 +356,11 @@ def __init__(
         """Initialize the OAuth2 authentication mixin.
 
         Args:
-            scopes: List of OAuth2 scopes to request.
-            auth_flow: Authentication flow type - "M2M" for machine-to-machine or
-                      "USER_FEDERATION" for user-delegated access.
+            scopes: Optional list of OAuth2 scopes to request. If not provided,
+                   the control plane will use the default configured scopes.
+            auth_flow: Optional authentication flow type - "M2M" for machine-to-machine or
+                      "USER_FEDERATION" for user-delegated access. If not provided,
+                      the control plane will use the default configured flow.
             callback_url: OAuth2 redirect URL (must be pre-registered).
             force_authentication: If True, forces re-authentication even if cached token exists.
             response_for_auth_required: Custom response to return when user authorization is needed.

veadk/integrations/ve_identity/auth_processor.py

@@ -230,21 +230,19 @@ async def process_auth_request(
                 on_auth_url(auth_uri)
 
         # Use custom poller or default poller
-        # Create async token fetcher for default poller
-        async def async_token_fetcher():
-            response = self._identity_client.get_oauth2_token_or_auth_url(
-                **request_dict
-            )
-            return (
-                OAuth2Auth(access_token=response.access_token)
-                if response.access_token and response.access_token.strip()
-                else None
-            )
-
         active_poller = (
             self.config.oauth2_auth_poller(auth_uri, request_dict)
             if self.config.oauth2_auth_poller
-            else _DefaultOauth2AuthPoller(auth_uri, async_token_fetcher)
+            else _DefaultOauth2AuthPoller(
+                auth_uri,
+                lambda: (
+                    lambda response: (
+                        OAuth2Auth(access_token=response.access_token)
+                        if response.access_token and response.access_token.strip()
+                        else None
+                    )
+                )(self._identity_client.get_oauth2_token_or_auth_url(**request_dict)),
+            )
         )
 
         # Poll for the oauth2 auth

veadk/integrations/ve_identity/identity_client.py

@@ -276,7 +276,7 @@ def get_oauth2_token_or_auth_url(
         *,
         provider_name: str,
         agent_identity_token: str,
-        auth_flow: Literal["M2M", "USER_FEDERATION"],
+        auth_flow: Optional[Literal["M2M", "USER_FEDERATION"]] = None,
         scopes: Optional[List[str]] = None,
         callback_url: Optional[str] = None,
         force_authentication: bool = False,
@@ -291,9 +291,11 @@ def get_oauth2_token_or_auth_url(
         Args:
             provider_name: Name of the credential provider configured in the identity service.
             agent_identity_token: Agent's workload access token for authentication.
-            auth_flow: OAuth2 flow type - "M2M" for machine-to-machine or
-                      "USER_FEDERATION" for user-delegated access.
-            scopes: Optional list of OAuth2 scopes to request.
+            auth_flow: Optional OAuth2 flow type - "M2M" for machine-to-machine or
+                      "USER_FEDERATION" for user-delegated access. If not provided,
+                      the control plane will use the default configured value.
+            scopes: Optional list of OAuth2 scopes to request. If not provided,
+                   the control plane will use the default configured scopes.
             callback_url: OAuth2 redirect URL (must be pre-registered with the provider).
             force_authentication: If True, forces re-authentication even if a valid
                                  token exists in the token vault.

veadk/integrations/ve_identity/mcp_tool.py

@@ -15,29 +15,14 @@
 from __future__ import annotations
 
 from typing import Any
-
 from typing_extensions import override
 
+from mcp.types import Tool as McpBaseTool
 from google.genai.types import FunctionDeclaration
 from google.adk.auth.auth_credential import AuthCredential
 from google.adk.tools.base_tool import BaseTool
 from google.adk.tools.tool_context import ToolContext
 from google.adk.tools._gemini_schema_util import _to_gemini_schema
-
-# Attempt to import MCP Tool from the MCP library
-try:
-    from mcp.types import Tool as McpBaseTool
-except ImportError as e:
-    import sys
-
-    if sys.version_info < (3, 10):
-        raise ImportError(
-            "MCP Tool requires Python 3.10 or above. Please upgrade your Python"
-            " version."
-        ) from e
-    else:
-        raise e
-
 from google.adk.tools.mcp_tool.mcp_session_manager import MCPSessionManager
 from google.adk.tools.mcp_tool.mcp_session_manager import retry_on_closed_resource
 
@@ -47,7 +32,6 @@
     AuthRequiredException,
 )
 from veadk.integrations.ve_identity.utils import generate_headers
-
 from veadk.utils.logger import get_logger
 
 logger = get_logger(__name__)

veadk/integrations/ve_identity/mcp_toolset.py

@@ -25,26 +25,8 @@
 from pydantic import model_validator, field_validator
 from typing_extensions import override
 
-from veadk.integrations.ve_identity.auth_config import VeIdentityAuthConfig
-from veadk.integrations.ve_identity.auth_mixins import VeIdentityAuthMixin
-from veadk.integrations.ve_identity.mcp_tool import VeIdentityMcpTool
-from veadk.integrations.ve_identity.utils import generate_headers
-
-# Attempt to import MCP Tool from the MCP library, and hints user to upgrade
-# their Python version to 3.10 if it fails.
-try:
-    from mcp import StdioServerParameters, ClientSession
-    from mcp.types import ListToolsResult
-except ImportError as e:
-    import sys
-
-    if sys.version_info < (3, 10):
-        raise ImportError(
-            "MCP Tool requires Python 3.10 or above. Please upgrade your Python"
-            " version."
-        ) from e
-    else:
-        raise e
+from mcp import StdioServerParameters, ClientSession
+from mcp.types import ListToolsResult
 
 from google.adk.tools.mcp_tool.mcp_session_manager import (
     retry_on_closed_resource,
@@ -58,6 +40,11 @@
 from google.adk.tools.tool_configs import ToolArgsConfig, BaseToolConfig
 from google.adk.agents.readonly_context import ReadonlyContext
 
+from veadk.integrations.ve_identity.auth_config import VeIdentityAuthConfig
+from veadk.integrations.ve_identity.auth_mixins import VeIdentityAuthMixin
+from veadk.integrations.ve_identity.mcp_tool import VeIdentityMcpTool
+from veadk.integrations.ve_identity.utils import generate_headers
+
 logger = logging.getLogger(__name__)
 
 
@@ -388,7 +375,10 @@ class VeIdentityMcpToolsetConfig(BaseToolConfig):
     def _validate_auth_config(cls, v):
         """Convert dict to proper auth config object."""
         if isinstance(v, dict):
-            from veadk.integrations.ve_identity.auth_config import api_key_auth, oauth2_auth
+            from veadk.integrations.ve_identity.auth_config import (
+                api_key_auth,
+                oauth2_auth,
+            )
 
             provider_name = v.get("provider_name")
             if not provider_name: