feat: support 3 modes to execute skills (#451)

Author: BhAem

veadk/agent.py

@@ -15,7 +15,7 @@
 from __future__ import annotations
 
 import os
-from typing import Dict, Optional, Union
+from typing import Dict, Optional, Union, Literal
 
 # If user didn't set LITELLM_LOCAL_MODEL_COST_MAP, set it to True
 # to enable local model cost map.
@@ -146,6 +146,8 @@ class Agent(LlmAgent):
 
     skills: list[str] = Field(default_factory=list)
 
+    skills_mode: Literal["skills_sandbox", "aio_sandbox", "local"] = "skills_sandbox"
+
     example_store: Optional[BaseExampleProvider] = None
 
     def model_post_init(self, __context: Any) -> None:
@@ -309,14 +311,7 @@ def load_skills(self):
             load_skills_from_cloud,
             load_skills_from_directory,
         )
-        from veadk.tools.skills_tools import (
-            SkillsTool,
-            bash_tool,
-            edit_file_tool,
-            read_file_tool,
-            write_file_tool,
-            register_skills_tool,
-        )
+        from veadk.tools.skills_tools.skills_toolset import SkillsToolset
 
         skills: Dict[str, Skill] = {}
 
@@ -338,12 +333,14 @@ def load_skills(self):
                     f"- name: {skill.name}\n- description: {skill.description}\n\n"
                 )
 
-            self.tools.append(SkillsTool(skills))
-            self.tools.append(read_file_tool)
-            self.tools.append(write_file_tool)
-            self.tools.append(edit_file_tool)
-            self.tools.append(bash_tool)
-            self.tools.append(register_skills_tool)
+            if self.skills_mode not in ["skills_sandbox", "aio_sandbox", "local"]:
+                raise ValueError(
+                    f"Unsupported skill mode {self.skills_mode}, use `skills_sandbox`, `aio_sandbox` or `local` instead."
+                )
+
+            self.tools.append(SkillsToolset(skills, self.skills_mode))
+        else:
+            logger.warning("No skills loaded.")
 
     def _prepare_tracers(self):
         enable_apmplus_tracer = os.getenv("ENABLE_APMPLUS", "false").lower() == "true"

veadk/tools/builtin_tools/execute_skills.py

@@ -70,8 +70,9 @@ def execute_skills(
     workflow_prompt: str,
     tool_context: ToolContext = None,
 ) -> str:
-    """execute skills in a code sandbox and return the output.
-    For C++ code, don't execute it directly, compile and execute via Python; write sources and object files to /tmp.
+    """Execute skills in a sandbox and return the output.
+
+    Execute skills in a remote sandbox amining to provide isolation and security.
 
     Args:
         workflow_prompt (str): instruction of workflow

veadk/tools/skills_tools/register_skills_tool.py

@@ -80,14 +80,13 @@ def register_skills_tool(
     try:
         from veadk.auth.veauth.utils import get_credential_from_vefaas_iam
 
-        service = os.getenv("AGENTKIT_TOOL_SERVICE_CODE", "agentkit")
+        agentkit_tool_service = os.getenv("AGENTKIT_TOOL_SERVICE_CODE", "agentkit")
+        agentkit_skill_host = os.getenv("AGENTKIT_SKILL_HOST", "open.volcengineapi.com")
         region = os.getenv("AGENTKIT_TOOL_REGION", "cn-beijing")
-        host = os.getenv("AGENTKIT_SKILL_HOST", "open.volcengineapi.com")
 
         access_key = os.getenv("VOLCENGINE_ACCESS_KEY")
         secret_key = os.getenv("VOLCENGINE_SECRET_KEY")
         session_token = ""
-        region = os.getenv("AGENTKIT_TOOL_REGION", "cn-beijing")
 
         if not (access_key and secret_key):
             cred = get_credential_from_vefaas_iam()
@@ -149,10 +148,10 @@ def register_skills_tool(
             action="CreateSkill",
             ak=access_key,
             sk=secret_key,
-            service=service,
+            service=agentkit_tool_service,
             version="2025-10-30",
             region=region,
-            host=host,
+            host=agentkit_skill_host,
             header={"X-Security-Token": session_token},
         )
 

veadk/tools/skills_tools/skills_tool.py

@@ -39,14 +39,14 @@ def __init__(self, skills: Dict[str, Skill]):
         self.skills = skills
 
         # Generate description with available skills embedded
-        description = self._generate_description_with_skills()
+        description = self._generate_description()
 
         super().__init__(
-            name="skills",
+            name="skills_tool",
             description=description,
         )
 
-    def _generate_description_with_skills(self) -> str:
+    def _generate_description(self) -> str:
         """Generate tool description with available skills embedded."""
         base_description = (
             "Execute a skill within the main conversation\n\n"

veadk/tools/skills_tools/skills_toolset.py

@@ -0,0 +1,105 @@
+# 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 Dict, List, Optional
+
+try:
+    from typing_extensions import override
+except ImportError:
+    from typing import override
+
+from google.adk.agents.readonly_context import ReadonlyContext
+from google.adk.tools import BaseTool, FunctionTool
+from google.adk.tools.base_toolset import BaseToolset
+
+from veadk.skills.skill import Skill
+from veadk.tools.skills_tools import (
+    SkillsTool,
+    read_file_tool,
+    write_file_tool,
+    edit_file_tool,
+    bash_tool,
+    register_skills_tool,
+)
+from veadk.utils.logger import get_logger
+
+logger = get_logger(__name__)
+
+
+class SkillsToolset(BaseToolset):
+    """Toolset that provides Skills functionality for domain expertise execution.
+
+    This toolset provides skills access through specialized tools:
+    1. SkillsTool - Discover and load skill instructions
+    2. ReadFileTool - Read files with line numbers
+    3. WriteFileTool - Write/create files
+    4. EditFileTool - Edit files with precise replacements
+    5. BashTool - Execute shell commands
+    6. RegisterSkillsTool - Register new skills into the remote skill space
+
+    Skills provide specialized domain knowledge and scripts that the agent can use
+    to solve complex tasks. The toolset enables discovery of available skills,
+    file manipulation, and command execution.
+    """
+
+    def __init__(self, skills: Dict[str, Skill], skills_mode: str) -> None:
+        """Initialize the skills toolset.
+
+        Args:
+            skills: A dictionary of skill.
+            skills_mode: The mode of skills operation, e.g., "skills_sandbox".
+        """
+        super().__init__()
+
+        self.skills_mode = skills_mode
+
+        self._tools = {
+            "skills": SkillsTool(skills),
+            "read_file": FunctionTool(read_file_tool),
+            "write_file": FunctionTool(write_file_tool),
+            "edit_file": FunctionTool(edit_file_tool),
+            "bash": FunctionTool(bash_tool),
+            "register_skills": FunctionTool(register_skills_tool),
+        }
+
+    @override
+    async def get_tools(
+        self, readonly_context: Optional[ReadonlyContext] = None
+    ) -> List[BaseTool]:
+        """Return tools according to selected skills_mode."""
+
+        match self.skills_mode:
+            case "local":
+                logger.info(
+                    "Skills mode=local, adding skills_tool, read_file_tool, write_file_tool, edit_file_tool, bash_tool and register_skills_tool to the agent."
+                )
+                return list(self._tools.values())
+
+            case "skills_sandbox":
+                logger.info(
+                    "Skills mode=skills_sandbox, no skills tools are added to the agent."
+                )
+                return []
+
+            case "aio_sandbox":
+                logger.info("Skills mode=aio_sandbox: not implemented yet")
+                return []
+
+            case _:
+                logger.warning(
+                    f"Unknown skills_mode: {self.skills_mode}, returning empty tool list."
+                )
+                return []