refactor: refactoring service configuration and credential management (#40)

Author: innsd

.gitleaks.toml

@@ -73,4 +73,7 @@ description = "Empty environment variables with KEY pattern"
 regex = '''os\.environ\[".*?KEY"\]\s*=\s*".+"'''
 
 [allowlist]
-paths = ["requirements.txt"]
+paths = [
+    "requirements.txt",
+    "tests/platform/test_configuration_creds.py"
+]

agentkit/apps/agent_server_app/agent_server_app.py

@@ -103,9 +103,7 @@ def __init__(
         @asynccontextmanager
         async def lifespan(app: FastAPI):
             # trigger A2A server app startup
-            logger.info(
-                "Triggering A2A server app startup within API server..."
-            )
+            logger.info("Triggering A2A server app startup within API server...")
             for handler in _a2a_server_app.router.on_startup:
                 await handler()
             yield
@@ -124,18 +122,14 @@ async def _invoke_compat(request: Request):
             # Extract headers (fallback keys supported)
             headers = request.headers
             user_id = (
-                headers.get("user_id")
-                or headers.get("x-user-id")
-                or "agentkit_user"
+                headers.get("user_id") or headers.get("x-user-id") or "agentkit_user"
             )
             session_id = headers.get("session_id") or ""
 
             # Determine app_name from loader
             app_names = self.server.agent_loader.list_agents()
             if not app_names:
-                raise HTTPException(
-                    status_code=404, detail="No agents configured"
-                )
+                raise HTTPException(status_code=404, detail="No agents configured")
             app_name = app_names[0]
 
             # Parse payload and convert to ADK Content
@@ -184,9 +178,7 @@ async def event_generator():
                             user_id=user_id,
                             session_id=session_id,
                             new_message=content,
-                            run_config=RunConfig(
-                                streaming_mode=StreamingMode.SSE
-                            ),
+                            run_config=RunConfig(streaming_mode=StreamingMode.SSE),
                         )
                     ) as agen:
                         async for event in agen:

agentkit/client/base_agentkit_client.py

@@ -20,7 +20,6 @@
 from typing import Any, Dict, Union, Optional
 
 from agentkit.client.base_service_client import BaseServiceClient, ApiConfig
-from agentkit.utils.ve_sign import get_volc_agentkit_host_info, get_volc_agentkit_scheme
 
 
 class BaseAgentkitClient(BaseServiceClient):
@@ -60,30 +59,15 @@ def __init__(
             service_name: Service name for logging (e.g., 'knowledge', 'memory')
         """
         super().__init__(
+            service="agentkit",
             access_key=access_key,
             secret_key=secret_key,
             region=region,
             session_token=session_token,
             service_name=service_name,
-            credential_env_prefix="AGENTKIT",
             header=header,
         )
 
-    def _get_service_config(self) -> Dict[str, str]:
-        """
-        Get AgentKit service configuration.
-
-        Returns:
-            Dictionary with host, api_version, and service
-        """
-        host, api_version, service = get_volc_agentkit_host_info()
-        return {
-            "host": host,
-            "api_version": api_version,
-            "service": service,
-            "scheme": get_volc_agentkit_scheme(),
-        }
-
     def _get(self, api_action: str, params: Dict[str, Any] = None) -> str:
         """Legacy method for GET requests."""
         try:

agentkit/client/base_iam_client.py

@@ -20,7 +20,6 @@
 from typing import Dict, Union
 
 from agentkit.client.base_service_client import BaseServiceClient, ApiConfig
-from agentkit.utils.ve_sign import get_volc_iam_host_scheme
 
 
 class BaseIAMClient(BaseServiceClient):
@@ -64,25 +63,10 @@ def __init__(
             service_name: Service name for logging
         """
         super().__init__(
+            service="iam",
             access_key=access_key,
             secret_key=secret_key,
             region=region,
             session_token=session_token,
             service_name=service_name,
-            credential_env_prefix="IAM",
         )
-
-    def _get_service_config(self) -> Dict[str, str]:
-        """
-        Get IAM service configuration.
-
-        Returns:
-            Dictionary with host, api_version, and service
-        """
-        host, scheme = get_volc_iam_host_scheme()
-        return {
-            "host": host or self.IAM_HOST,
-            "api_version": self.IAM_API_VERSION,
-            "service": self.IAM_SERVICE_CODE,
-            "scheme": scheme,
-        }

agentkit/client/base_service_client.py

@@ -26,10 +26,12 @@
 from volcengine.Credentials import Credentials
 from volcengine.ServiceInfo import ServiceInfo
 
-from agentkit.utils.ve_sign import (
-    get_volc_ak_sk_region,
-    ensure_x_custom_source_header,
+from agentkit.platform import (
+    VolcConfiguration,
+    resolve_credentials,
+    resolve_endpoint,
 )
+from agentkit.utils.ve_sign import ensure_x_custom_source_header
 
 T = TypeVar("T")
 
@@ -62,52 +64,61 @@ class BaseServiceClient(Service):
 
     Subclasses should:
     1. Override API_ACTIONS with their API action configurations
-    2. Implement _get_service_config() to provide service-specific configuration
     """
 
     # Subclasses should override this with their API action configurations
     API_ACTIONS: Dict[str, Union[str, ApiConfig]] = {}
 
     def __init__(
         self,
+        service: str,
         access_key: str = "",
         secret_key: str = "",
         region: str = "",
         session_token: str = "",
         service_name: str = "",
-        credential_env_prefix: str = "",
+        platform_config: Optional[VolcConfiguration] = None,
         header: Optional[Dict[str, Any]] = None,
     ) -> None:
         """
         Initialize the service client.
 
         Args:
+            service: Logical service name for signing and endpoint resolution
             access_key: Volcengine access key
             secret_key: Volcengine secret key
-            region: Volcengine region
+            region: Volcengine region override for endpoint
             session_token: Optional session token
             service_name: Service name for logging
-            credential_env_prefix: Environment variable prefix for credentials (e.g., 'AGENTKIT', 'IAM')
+            platform_config: Optional platform-level configuration overrides
         """
-        # Validate and get credentials
-        if not any([access_key, secret_key, region]):
-            access_key, secret_key, region = get_volc_ak_sk_region(
-                credential_env_prefix
-            )
-        else:
-            if not all([access_key, secret_key, region]):
-                raise ValueError(
-                    f"Error creating {service_name} instance: "
-                    "missing access key, secret key or region"
-                )
+        if platform_config is None:
+            platform_config = VolcConfiguration()
+
+        creds = resolve_credentials(
+            service=service,
+            explicit_access_key=access_key or None,
+            explicit_secret_key=secret_key or None,
+            platform_config=platform_config,
+        )
+
+        ep = resolve_endpoint(
+            service=service,
+            region=region or None,
+            platform_config=platform_config,
+        )
 
-        # Store credentials and service info
-        self.access_key = access_key
-        self.secret_key = secret_key
-        self.region = region
+        self.access_key = creds.access_key
+        self.secret_key = creds.secret_key
+        self.region = ep.region
         self.session_token = session_token
         self.service_name = service_name
 
+        self.host = ep.host
+        self.api_version = ep.api_version
+        self.service = ep.service
+        self.scheme = ep.scheme
+
         if header is None:
             effective_header: Dict[str, Any] = {"Accept": "application/json"}
         else:
@@ -116,14 +127,6 @@ def __init__(
                 effective_header = {"Accept": "application/json", **effective_header}
 
         effective_header = ensure_x_custom_source_header(effective_header)
-
-        # Get service-specific configuration from subclass
-        config = self._get_service_config()
-        self.host = config["host"]
-        self.api_version = config["api_version"]
-        self.service = config["service"]
-        self.scheme = config.get("scheme", "https")
-
         # Create ServiceInfo
         self.service_info = ServiceInfo(
             host=self.host,
@@ -145,20 +148,9 @@ def __init__(
 
         # Initialize parent Service class
         Service.__init__(self, service_info=self.service_info, api_info=self.api_info)
-
-    def _get_service_config(self) -> Dict[str, str]:
-        """
-        Get service-specific configuration.
-
-        Subclasses must override this method to provide:
-        - host: API endpoint host
-        - api_version: API version string
-        - service: Service name for signing
-
-        Returns:
-            Dictionary with 'host', 'api_version', and 'service' keys
-        """
-        raise NotImplementedError("Subclasses must implement _get_service_config()")
+        # need setting ak/sk after initializing Service to avoid volcengine SDK bugs
+        self.set_ak(self.access_key)
+        self.set_sk(self.secret_key)
 
     def _build_api_info(self) -> Dict[str, ApiInfo]:
         """

agentkit/platform/__init__.py

@@ -0,0 +1,89 @@
+# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+from typing import Optional
+
+from agentkit.platform.configuration import VolcConfiguration, Endpoint, Credentials
+from agentkit.platform.constants import DEFAULT_REGION_RULES
+
+__all__ = [
+    "VolcConfiguration",
+    "Endpoint",
+    "Credentials",
+    "resolve_endpoint",
+    "resolve_credentials",
+    "DEFAULT_REGION_RULES",
+]
+
+# Backward compatibility wrappers
+
+
+def resolve_endpoint(
+    service: str,
+    *,
+    region: Optional[str] = None,
+    platform_config: Optional[VolcConfiguration] = None,
+) -> Endpoint:
+    """
+    Resolves the endpoint for a service.
+
+    Args:
+        service: Service identifier (e.g. 'agentkit', 'cr')
+        region: Explicit region override
+        platform_config: Optional configuration object. If provided, 'region' arg overrides config.
+    """
+    # If region is provided explicitly, it should override the config's region.
+    # We can achieve this by creating a new ephemeral config or just letting VolcConfiguration handle it?
+    # Since VolcConfiguration is immutable-ish, we can just instantiate one if needed.
+
+    if platform_config:
+        # If explicit region is passed, it overrides the config's region for this call
+        if region and region != platform_config.region:
+            # Create a temporary lightweight clone with new region, keeping creds
+            # Or more simply, VolcConfiguration handles service-specific region overrides internally
+            # but here 'region' is a global override for this call.
+            # Let's create a new config merging both.
+            cfg = VolcConfiguration(
+                region=region,
+                access_key=platform_config._ak,
+                secret_key=platform_config._sk,
+            )
+        else:
+            cfg = platform_config
+    else:
+        cfg = VolcConfiguration(region=region)
+
+    return cfg.get_service_endpoint(service)
+
+
+def resolve_credentials(
+    service: str,
+    *,
+    explicit_access_key: Optional[str] = None,
+    explicit_secret_key: Optional[str] = None,
+    platform_config: Optional[VolcConfiguration] = None,
+) -> Credentials:
+    """
+    Resolves credentials for a service.
+    """
+    # 1. Explicit args take absolute precedence
+    if explicit_access_key and explicit_secret_key:
+        return Credentials(
+            access_key=explicit_access_key, secret_key=explicit_secret_key
+        )
+
+    cfg = platform_config or VolcConfiguration()
+    return cfg.get_service_credentials(service)