feat(toolkit): support custom cloud build resources

agentkit/toolkit/builders/ve_pipeline.py

@@ -1240,9 +1240,17 @@ def _prepare_pipeline_resources(
             )
             cp_client = VeCodePipeline(region=config.cp_region, provider=provider)
 
-            # Get or create agentkit-cli-workspace
-            workspace_name = "agentkit-cli-workspace"
-            if not cp_client.workspace_exists_by_name(workspace_name):
+            # Get or create the configured workspace.
+            workspace_name = config.cp_workspace_name or DEFAULT_WORKSPACE_NAME
+            workspace_result = cp_client.get_workspaces_by_name(
+                workspace_name, page_size=100
+            )
+            matching_workspaces = [
+                item
+                for item in workspace_result.get("Items", [])
+                if item.get("Name") == workspace_name
+            ]
+            if not matching_workspaces:
                 logger.info(f"Workspace '{workspace_name}' does not exist, creating...")
                 self.reporter.warning(
                     f"Workspace '{workspace_name}' does not exist, creating..."
@@ -1257,16 +1265,11 @@ def _prepare_pipeline_resources(
                     f"Workspace created successfully: {workspace_name}"
                 )
             else:
-                # Workspace exists, get its ID
-                result = cp_client.get_workspaces_by_name(workspace_name, page_size=1)
-                if result.get("Items") and len(result["Items"]) > 0:
-                    workspace_id = result["Items"][0]["Id"]
-                    logger.info(
-                        f"Using existing workspace: {workspace_name} (ID: {workspace_id})"
-                    )
-                    self.reporter.success(f"Using workspace: {workspace_name}")
-                else:
-                    raise Exception(f"Unable to get workspace '{workspace_name}' ID")
+                workspace_id = matching_workspaces[0]["Id"]
+                logger.info(
+                    f"Using existing workspace: {workspace_name} (ID: {workspace_id})"
+                )
+                self.reporter.success(f"Using workspace: {workspace_name}")
 
             logger.info(f"Using workspace: {workspace_name} (ID: {workspace_id})")
 
@@ -1276,8 +1279,6 @@ def _prepare_pipeline_resources(
             # Check if pipeline already exists - try multiple lookup strategies
             # Case 1: If Pipeline ID is configured, use ID for exact lookup
 
-            # tmp: temp fix for pipeline id issue, cp_pipeline_id should be empty string for fix cp name
-            config.cp_pipeline_id = ""
             if config.cp_pipeline_id and config.cp_pipeline_id != AUTO_CREATE_VE:
                 try:
                     # Get pipeline details by ID
@@ -1287,6 +1288,11 @@ def _prepare_pipeline_resources(
 
                     if result.get("Items") and len(result["Items"]) > 0:
                         pipeline_info = result["Items"][0]
+                        if not cp_client.is_agentkit_build_pipeline(pipeline_info):
+                            raise Exception(
+                                f"Pipeline ID '{config.cp_pipeline_id}' is not compatible "
+                                "with AgentKit cloud builds"
+                            )
                         found_pipeline_name = pipeline_info.get("Name", "")
 
                         # If name is also configured, validate name-ID consistency
@@ -1331,8 +1337,8 @@ def _prepare_pipeline_resources(
                         )
 
                 except Exception as e:
-                    if "does not match" in str(e):
-                        raise  # Name-ID mismatch, propagate exception
+                    if "does not match" in str(e) or "not compatible" in str(e):
+                        raise
                     logger.warning(
                         f"Pipeline lookup by ID failed: {str(e)}, will create new pipeline"
                     )
@@ -1344,14 +1350,21 @@ def _prepare_pipeline_resources(
                         workspace_id=workspace_id, name_filter=config.cp_pipeline_name
                     )
 
-                    if (
-                        existing_pipelines.get("Items")
-                        and len(existing_pipelines["Items"]) > 0
-                    ):
+                    matching_pipelines = [
+                        item
+                        for item in existing_pipelines.get("Items", [])
+                        if item.get("Name") == config.cp_pipeline_name
+                    ]
+                    if matching_pipelines:
                         # Found existing pipeline
-                        pipeline_info = existing_pipelines["Items"][0]
+                        pipeline_info = matching_pipelines[0]
                         pipeline_id = pipeline_info["Id"]
                         found_name = pipeline_info.get("Name", "")
+                        if not cp_client.is_agentkit_build_pipeline(pipeline_info):
+                            raise Exception(
+                                f"Pipeline '{found_name}' is not compatible with "
+                                "AgentKit cloud builds"
+                            )
 
                         logger.info(
                             f"Reusing pipeline by name: {found_name} (ID: {pipeline_id})"
@@ -1380,6 +1393,8 @@ def _prepare_pipeline_resources(
                             "Configured pipeline name does not exist, will create new pipeline"
                         )
                 except Exception as e:
+                    if "not compatible" in str(e):
+                        raise
                     logger.warning(
                         f"Pipeline lookup by name failed: {str(e)}, will create new pipeline"
                     )

agentkit/toolkit/strategies/cloud_strategy.py

@@ -120,6 +120,8 @@ def build(
         runtime_name, cp_pipeline_name = self._prepare_runtime_name(
             strategy_config.runtime_name, common_config.agent_name
         )
+        if strategy_config.cp_pipeline_name not in ("", AUTO_CREATE_VE):
+            cp_pipeline_name = strategy_config.cp_pipeline_name
 
         # Track generated names if they differ from config
         if runtime_name != strategy_config.runtime_name:

agentkit/toolkit/volcengine/code_pipeline.py

@@ -25,6 +25,24 @@
 
 
 class VeCodePipeline:
+    AGENTKIT_BUILD_PARAMETER_KEYS = frozenset(
+        {
+            "DOCKERFILE_PATH",
+            "DOWNLOAD_PATH",
+            "PROJECT_ROOT_DIR",
+            "TOS_BUCKET_NAME",
+            "TOS_REGION",
+            "TOS_PROJECT_FILE_NAME",
+            "TOS_PROJECT_FILE_PATH",
+            "CR_NAMESPACE",
+            "CR_INSTANCE",
+            "CR_DOMAIN",
+            "CR_OCI",
+            "CR_TAG",
+            "CR_REGION",
+        }
+    )
+
     def __init__(
         self,
         access_key: str = "",
@@ -316,6 +334,26 @@ def _create_pipeline(
         except KeyError:
             raise Exception(f"Create pipeline failed: {res}")
 
+    @classmethod
+    def is_agentkit_build_pipeline(cls, pipeline: dict) -> bool:
+        """Return whether a ListPipelines item matches AgentKit's build template."""
+        parameters = pipeline.get("Parameters") or []
+        keys = {
+            str(parameter.get("Key") or "")
+            for parameter in parameters
+            if isinstance(parameter, dict)
+        }
+        if cls.AGENTKIT_BUILD_PARAMETER_KEYS.issubset(keys):
+            return True
+
+        spec = pipeline.get("Spec") or ""
+        if not isinstance(spec, str):
+            return False
+        return all(
+            f"parameters.{key}" in spec or f"${key}" in spec
+            for key in cls.AGENTKIT_BUILD_PARAMETER_KEYS
+        )
+
     def run_pipeline(
         self,
         workspace_id: str,

agentkit/toolkit/volcengine/cr.py

@@ -155,6 +155,59 @@ def _check_instance(self, instance_name: str) -> str:
         except Exception as _:
             raise ValueError(f"Error check cr instance {instance_name}: {response}")
 
+    def list_registries(self, page_number: int = 1, page_size: int = 100) -> dict:
+        """List Container Registry instances."""
+        response = self._ve_request(
+            request_body={"PageNumber": page_number, "PageSize": page_size},
+            action="ListRegistries",
+        )
+        try:
+            return response["Result"]
+        except KeyError as error:
+            raise ValueError(f"List CR registries failed: {response}") from error
+
+    def list_namespaces(
+        self,
+        registry: str,
+        page_number: int = 1,
+        page_size: int = 100,
+    ) -> dict:
+        """List namespaces in a Container Registry instance."""
+        response = self._ve_request(
+            request_body={
+                "Registry": registry,
+                "PageNumber": page_number,
+                "PageSize": page_size,
+            },
+            action="ListNamespaces",
+        )
+        try:
+            return response["Result"]
+        except KeyError as error:
+            raise ValueError(f"List CR namespaces failed: {response}") from error
+
+    def list_repositories(
+        self,
+        registry: str,
+        namespace: str,
+        page_number: int = 1,
+        page_size: int = 100,
+    ) -> dict:
+        """List repositories under a Container Registry namespace."""
+        response = self._ve_request(
+            request_body={
+                "Registry": registry,
+                "Namespace": namespace,
+                "PageNumber": page_number,
+                "PageSize": page_size,
+            },
+            action="ListRepositories",
+        )
+        try:
+            return response["Result"]
+        except KeyError as error:
+            raise ValueError(f"List CR repositories failed: {response}") from error
+
     def _create_namespace(
         self,
         instance_name: str = DEFAULT_CR_INSTANCE_NAME,

agentkit/toolkit/volcengine/services/tos_service.py

@@ -16,9 +16,10 @@
 import logging
 from dataclasses import dataclass, field
 from typing import List, Optional
-from agentkit.utils.misc import generate_random_id
+from agentkit.platform import Credentials
 from agentkit.toolkit.config.dataclass_utils import AutoSerializableMixin
 from agentkit.toolkit.config.constants import DEFAULT_TOS_BUCKET_TEMPLATE_NAME
+from agentkit.utils.misc import generate_random_id
 
 try:
     import tos
@@ -76,7 +77,12 @@ class TOSMountConfig:
 class TOSService:
     """Wrapper for Volcano Engine TOS (Object Storage) service."""
 
-    def __init__(self, config: TOSServiceConfig, provider: Optional[str] = None):
+    def __init__(
+        self,
+        config: TOSServiceConfig,
+        provider: Optional[str] = None,
+        credentials: Optional[Credentials] = None,
+    ):
         """Initialize TOS service with configuration.
 
         Args:
@@ -90,6 +96,7 @@ def __init__(self, config: TOSServiceConfig, provider: Optional[str] = None):
 
         self.config = config
         self.provider = provider
+        self.explicit_credentials = credentials
         self.client = None
         self.credentials = None
         self._init_client()
@@ -101,7 +108,14 @@ def _init_client(self) -> None:
 
             # Use configured region if available
             region = self.config.region.strip() if self.config.region else None
-            config = VolcConfiguration(region=region, provider=self.provider)
+            credentials = self.explicit_credentials
+            config = VolcConfiguration(
+                region=region,
+                provider=self.provider,
+                access_key=credentials.access_key if credentials else None,
+                secret_key=credentials.secret_key if credentials else None,
+                session_token=credentials.session_token if credentials else None,
+            )
             creds = config.get_service_credentials("tos")
             ep = config.get_service_endpoint("tos")
 
@@ -346,15 +360,31 @@ def list_bucket_names(self) -> List[str]:
         Returns:
             List[str]: Bucket names under the current account.
         """
+        return [bucket["Name"] for bucket in self.list_buckets()]
+
+    def list_buckets(self) -> List[dict]:
+        """List buckets owned by the current credentials.
+
+        Returns normalized names and locations so callers can avoid selecting a
+        bucket from a different physical TOS region.
+        """
         try:
             out = self.client.list_buckets()
             buckets = getattr(out, "buckets", None) or []
-            names: List[str] = []
-            for b in buckets:
-                name = getattr(b, "name", None)
-                if name:
-                    names.append(name)
-            return names
+            result: List[dict] = []
+            for bucket in buckets:
+                name = getattr(bucket, "name", None)
+                if not name:
+                    continue
+                result.append(
+                    {
+                        "Name": name,
+                        "Location": getattr(bucket, "location", None)
+                        or getattr(self, "actual_region", ""),
+                        "CreationDate": getattr(bucket, "creation_date", None) or "",
+                    }
+                )
+            return result
         except Exception as e:
             logger.error(f"Failed to list buckets: {str(e)}")
             raise