diff --git a/veadk/agent.py b/veadk/agent.py
index 28066635..18dee834 100644
--- a/veadk/agent.py
+++ b/veadk/agent.py
@@ -15,7 +15,7 @@
 from __future__ import annotations
 
 import os
-from typing import Optional, Union
+from typing import Dict, Optional, Union
 
 # If user didn't set LITELLM_LOCAL_MODEL_COST_MAP, set it to True
 # to enable local model cost map.
@@ -297,26 +297,45 @@ def update_model(self, model_name: str):
 
     def load_skills(self):
         from pathlib import Path
+        from veadk.skills.skill import Skill
+        from veadk.skills.utils import (
+            load_skills_from_directory,
+            load_skills_from_cloud,
+        )
+        from veadk.tools.builtin_tools.playwright import playwright_tools
+        from veadk.tools.skills_tools import (
+            SkillsTool,
+            read_file_tool,
+            write_file_tool,
+            edit_file_tool,
+            bash_tool,
+        )
 
-        from veadk.skills.utils import load_skills_from_directory
+        skills: Dict[str, Skill] = {}
 
-        skills = []
-        for skill in self.skills:
-            path = Path(skill)
-            if path.is_dir():
-                skills.extend(load_skills_from_directory(path))
+        for item in self.skills:
+            path = Path(item)
+            if path.exists() and path.is_dir():
+                for skill in load_skills_from_directory(path):
+                    skills[skill.name] = skill
             else:
-                logger.error(
-                    f"Skill {skill} is not a directory, skip. Loading skills from cloud is WIP."
-                )
+                for skill in load_skills_from_cloud(item):
+                    skills[skill.name] = skill
         if skills:
             self.instruction += "\nYou have the following skills:\n"
 
-            for skill in skills:
+            for skill in skills.values():
                 self.instruction += (
                     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(playwright_tools)
+
     async def _run(
         self,
         runner,
diff --git a/veadk/runner.py b/veadk/runner.py
index 888a9a1f..7c901978 100644
--- a/veadk/runner.py
+++ b/veadk/runner.py
@@ -472,6 +472,11 @@ async def run(
             )
         logger.info(f"Run config: {run_config}")
 
+        if self.agent.skills:
+            from veadk.tools.skills_tools.session_path import initialize_session_path
+
+            initialize_session_path(session_id)
+
         user_id = user_id or self.user_id
 
         converted_messages: list = _convert_messages(
diff --git a/veadk/skills/skill.py b/veadk/skills/skill.py
index 8953c612..27184fa2 100644
--- a/veadk/skills/skill.py
+++ b/veadk/skills/skill.py
@@ -13,9 +13,12 @@
 # limitations under the License.
 
 from pydantic import BaseModel
+from typing import Optional
 
 
 class Skill(BaseModel):
     name: str
     description: str
-    path: str
+    path: str  # local path or tos path
+    skill_space_id: Optional[str] = None
+    bucket_name: Optional[str] = None
diff --git a/veadk/skills/utils.py b/veadk/skills/utils.py
index b64d7d56..deb69569 100644
--- a/veadk/skills/utils.py
+++ b/veadk/skills/utils.py
@@ -12,12 +12,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import json
 from pathlib import Path
-
+import os
 import frontmatter
 
 from veadk.skills.skill import Skill
 from veadk.utils.logger import get_logger
+from veadk.utils.volcengine_sign import ve_request
 
 logger = get_logger(__name__)
 
@@ -25,6 +27,10 @@
 def load_skill_from_directory(skill_directory: Path) -> Skill:
     logger.info(f"Load skill from {skill_directory}")
     skill_readme = skill_directory / "SKILL.md"
+    if not skill_readme.exists():
+        logger.error(f"Skill '{skill_directory}' has no SKILL.md file.")
+        raise ValueError(f"Skill '{skill_directory}' has no SKILL.md file")
+
     skill = frontmatter.load(str(skill_readme))
 
     skill_name = skill.get("name", "")
@@ -39,7 +45,7 @@ def load_skill_from_directory(skill_directory: Path) -> Skill:
         )
 
     logger.info(
-        f"Successfully loaded skill from {skill_readme}, name={skill['name']}, description={skill['description']}"
+        f"Successfully loaded skill {skill_name} locally from {skill_readme}, name={skill_name}, description={skill_description}"
     )
     return Skill(
         name=skill_name,  # type: ignore
@@ -58,4 +64,76 @@ def load_skills_from_directory(skills_directory: Path) -> list[Skill]:
     return skills
 
 
-def load_skills_from_cloud(space_name: str) -> list[Skill]: ...
+def load_skills_from_cloud(skill_space_ids: str) -> list[Skill]:
+    skill_space_ids_list = [x.strip() for x in skill_space_ids.split(",")]
+    logger.info(f"Load skills from skill spaces: {skill_space_ids_list}")
+
+    from veadk.auth.veauth.utils import get_credential_from_vefaas_iam
+
+    skills = []
+
+    for skill_space_id in skill_space_ids_list:
+        try:
+            service = os.getenv("AGENTKIT_TOOL_SERVICE_CODE", "agentkit")
+            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 = ""
+
+            if not (access_key and secret_key):
+                # Try to get from vefaas iam
+                cred = get_credential_from_vefaas_iam()
+                access_key = cred.access_key_id
+                secret_key = cred.secret_access_key
+                session_token = cred.session_token
+
+            response = ve_request(
+                request_body={
+                    "SkillSpaceId": skill_space_id,
+                    "InnerTags": {"source": "sandbox"},
+                },
+                action="ListSkillsBySpaceId",
+                ak=access_key,
+                sk=secret_key,
+                service=service,
+                version="2025-10-30",
+                region=region,
+                host=host,
+                header={"X-Security-Token": session_token},
+            )
+
+            if isinstance(response, str):
+                response = json.loads(response)
+
+            list_skills_result = response.get("Result")
+            items = list_skills_result.get("Items")
+
+            for item in items:
+                if not isinstance(item, dict):
+                    continue
+                skill_name = item.get("Name")
+                skill_description = item.get("Description")
+                tos_bucket = item.get("BucketName")
+                tos_path = item.get("TosPath")
+                if not skill_name:
+                    continue
+
+                skill = Skill(
+                    name=skill_name,  # type: ignore
+                    description=skill_description,  # type: ignore
+                    path=tos_path,
+                    skill_space_id=skill_space_id,
+                    bucket_name=tos_bucket,
+                )
+
+                skills.append(skill)
+
+                logger.info(
+                    f"Successfully loaded skill {skill_name} from skill space={skill_space_id}, name={skill_name}, description={skill_description}"
+                )
+        except Exception as e:
+            logger.error(f"Failed to load skill from skill space: {e}")
+
+    return skills
diff --git a/veadk/tools/builtin_tools/execute_skills.py b/veadk/tools/builtin_tools/execute_skills.py
index 6cd5a377..95bb9142 100644
--- a/veadk/tools/builtin_tools/execute_skills.py
+++ b/veadk/tools/builtin_tools/execute_skills.py
@@ -71,7 +71,6 @@ def execute_skills(
     workflow_prompt: str,
     skills: Optional[List[str]] = None,
     tool_context: ToolContext = None,
-    timeout: int = 900,
 ) -> 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.
@@ -79,11 +78,11 @@ def execute_skills(
     Args:
         workflow_prompt (str): instruction of workflow
         skills (Optional[List[str]]): The skills will be invoked
-        timeout (int, optional): The timeout in seconds for the code execution, less than or equal to 900. Defaults to 900.
 
     Returns:
         str: The output of the code execution.
     """
+    timeout = 900  # The timeout in seconds for the code execution, less than or equal to 900. Defaults to 900. Hard-coded to prevent the Agent from adjusting this parameter.
 
     tool_id = getenv("AGENTKIT_TOOL_ID")
 
@@ -131,26 +130,12 @@ def execute_skills(
     if skills:
         cmd.extend(["--skills"] + skills)
 
-    # TODO: remove after agentkit supports custom environment variables setting
-    res = ve_request(
-        request_body={},
-        action="GetCallerIdentity",
-        ak=ak,
-        sk=sk,
-        service="sts",
-        version="2018-01-01",
-        region=region,
-        host="sts.volcengineapi.com",
-        header=header,
-    )
-    try:
-        account_id = res["Result"]["AccountId"]
-    except KeyError as e:
-        logger.error(f"Error occurred while getting account id: {e}, response is {res}")
-        return res
+    skill_space_id = os.getenv("SKILL_SPACE_ID", "")
+    if not skill_space_id:
+        logger.warning("SKILL_SPACE_ID environment variable is not set")
 
     env_vars = {
-        "TOS_SKILLS_DIR": f"tos://agentkit-platform-{account_id}/skills/",
+        "SKILL_SPACE_ID": skill_space_id,
         "TOOL_USER_SESSION_ID": tool_user_session_id,
     }
 
diff --git a/veadk/tools/skills_tools/__init__.py b/veadk/tools/skills_tools/__init__.py
new file mode 100644
index 00000000..b44d44f8
--- /dev/null
+++ b/veadk/tools/skills_tools/__init__.py
@@ -0,0 +1,30 @@
+# 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 .bash_tool import bash_tool
+from .file_tool import edit_file_tool, read_file_tool, write_file_tool
+from .skills_tool import SkillsTool
+from .session_path import initialize_session_path, get_session_path, clear_session_cache
+
+
+__all__ = [
+    "bash_tool",
+    "edit_file_tool",
+    "read_file_tool",
+    "write_file_tool",
+    "SkillsTool",
+    "initialize_session_path",
+    "get_session_path",
+    "clear_session_cache",
+]
diff --git a/veadk/tools/skills_tools/bash_tool.py b/veadk/tools/skills_tools/bash_tool.py
new file mode 100644
index 00000000..1a27cd05
--- /dev/null
+++ b/veadk/tools/skills_tools/bash_tool.py
@@ -0,0 +1,158 @@
+# 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
+
+import asyncio
+import os
+
+from google.adk.tools import ToolContext
+from veadk.tools.skills_tools.session_path import get_session_path
+from veadk.utils.logger import get_logger
+
+logger = get_logger(__name__)
+
+
+async def bash_tool(command: str, description: str, tool_context: ToolContext):
+    """Execute bash commands in the skills environment with local shell.
+
+    This tool uses the local bash shell to execute commands with:
+    - Filesystem restrictions (controlled read/write access)
+    - Network restrictions (controlled domain access)
+    - Process isolation at the OS level
+
+    Use it for command-line operations like running scripts, installing packages, etc.
+    For file operations (read/write/edit), use the dedicated file tools instead.
+
+    Execute bash commands in the skills environment with local shell.
+    Working Directory & Structure:
+    - Commands run in a temporary session directory: /tmp/veadk/{session_id}/
+    - Working directory structure:
+        /tmp/veadk/{session_id}/
+        ├── skills/     -> all skills are available here (read-only).
+        ├── uploads/    -> staged user files (temporary)
+        └── outputs/    -> generated files for return
+    - Your current working directory is added to PYTHONPATH.
+
+    Python Imports (CRITICAL):
+    - To import from a skill, use the full path from the 'skills' root.
+      Example: from skills.skills_name.module import function
+    - If the skills name contains a dash '-', you need to use importlib to import it.
+      Example:
+        import importlib
+        skill_module = importlib.import_module('skills.skill-name.module')
+
+    For file operations:
+    - Use read_file, write_file, and edit_file for interacting with the filesystem.
+
+    Timeouts:
+    - pip install: 120s
+    - python scripts: 60s
+    - other commands: 30s
+
+    Args:
+        command: Bash command to execute. Use && to chain commands.
+        description: Clear, concise description of what this command does (5-10 words)
+        tool_context: The context of the tool execution, including session info.
+
+    Returns:
+        The output of the bash command or error message.
+    """
+
+    if not command:
+        return "Error: No command provided"
+
+    try:
+        # Get session working directory (initialized by SkillsPlugin)
+        working_dir = get_session_path(session_id=tool_context.session.id)
+        logger.info(f"Session working directory: {working_dir}")
+
+        # Determine timeout based on command
+        timeout = _get_command_timeout_seconds(command)
+
+        # Prepare environment with PYTHONPATH including skills directory
+        # This allows imports like: from skills.slack_gif_creator.core import something
+        env = os.environ.copy()
+        # Add root for 'from skills...' and working_dir for local scripts
+        pythonpath_additions = [str(working_dir), "/"]
+        if "PYTHONPATH" in env:
+            pythonpath_additions.append(env["PYTHONPATH"])
+        env["PYTHONPATH"] = ":".join(pythonpath_additions)
+
+        # Check for BASH_VENV_PATH to use a specific virtual environment
+        provided = os.environ.get("BASH_VENV_PATH")
+        if provided and os.path.isdir(provided):
+            bash_venv_path = provided
+            bash_venv_bin = os.path.join(bash_venv_path, "bin")
+            logger.info(f"Using provided BASH_VENV_PATH: {bash_venv_path}")
+            # Prepend bash venv to PATH so its python and pip are used
+            env["PATH"] = f"{bash_venv_bin}:{env.get('PATH', '')}"
+            env["VIRTUAL_ENV"] = bash_venv_path
+
+        # Execute with local bash shell
+        local_bash_command = f"{command}"
+
+        process = await asyncio.create_subprocess_shell(
+            local_bash_command,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+            cwd=working_dir,
+            env=env,  # Pass the modified environment
+        )
+
+        try:
+            stdout, stderr = await asyncio.wait_for(
+                process.communicate(), timeout=timeout
+            )
+        except asyncio.TimeoutError:
+            process.kill()
+            await process.wait()
+            return f"Error: Command timed out after {timeout}s"
+
+        stdout_str = stdout.decode("utf-8", errors="replace") if stdout else ""
+        stderr_str = stderr.decode("utf-8", errors="replace") if stderr else ""
+
+        # Handle command failure
+        if process.returncode != 0:
+            error_msg = f"Command failed with exit code {process.returncode}"
+            if stderr_str:
+                error_msg += f":\n{stderr_str}"
+            elif stdout_str:
+                error_msg += f":\n{stdout_str}"
+            return error_msg
+
+        # Return output
+        output = stdout_str
+        if stderr_str and "WARNING" not in stderr_str:
+            output += f"\n{stderr_str}"
+
+        result = output.strip() if output.strip() else "Command completed successfully."
+
+        logger.info(f"Executed bash command: {command}, description: {description}")
+        logger.info(f"Command result: {result}")
+        return result
+    except Exception as e:
+        error_msg = f"Error executing command '{command}': {e}"
+        logger.error(error_msg)
+        return error_msg
+
+
+def _get_command_timeout_seconds(command: str) -> float:
+    """Determine appropriate timeout for command in seconds."""
+    if "pip install" in command or "pip3 install" in command:
+        return 120.0
+    elif "python " in command or "python3 " in command:
+        return 60.0
+    else:
+        return 30.0
diff --git a/veadk/tools/skills_tools/file_tool.py b/veadk/tools/skills_tools/file_tool.py
new file mode 100644
index 00000000..3fdb8bc4
--- /dev/null
+++ b/veadk/tools/skills_tools/file_tool.py
@@ -0,0 +1,234 @@
+# 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 pathlib import Path
+
+from google.adk.tools import ToolContext
+from veadk.tools.skills_tools.session_path import get_session_path
+
+from veadk.utils.logger import get_logger
+
+logger = get_logger(__name__)
+
+
+def read_file_tool(file_path: str, offset: int, limit: int, tool_context: ToolContext):
+    """Read files with line numbers for precise editing.
+
+    Reads a file from the filesystem with line numbers.
+
+    Working directory structure:
+        /tmp/veadk/{session_id}/
+        ├── skills/     -> all skills are available here (read-only).
+        ├── uploads/    -> staged user files (temporary)
+        └── outputs/    -> generated files for return
+
+    Usage:
+    - Provide a path to the file (absolute or relative to your working directory)
+    - Returns content with line numbers (format: LINE_NUMBER|CONTENT)
+    - Optional offset and limit parameters for reading specific line ranges
+    - Lines longer than 2000 characters are truncated
+    - Always read a file before editing it
+    - You can read from skills/ directory, uploads/, outputs/, or any file in your session
+
+    Args:
+        file_path: Path to the file to read (absolute or relative to working directory)
+        offset: Optional line number to start reading from (1-indexed)
+        limit: Optional number of lines to read
+        tool_context: Context of the tool execution
+
+    Returns:
+        Content of the file with line numbers, or error message.
+    """
+    if not file_path:
+        return "Error: No file path provided"
+
+    # Resolve path relative to session working directory
+    working_dir = get_session_path(session_id=tool_context.session.id)
+    path = Path(file_path)
+    if not path.is_absolute():
+        path = working_dir / path
+    path = path.resolve()
+
+    if not path.exists():
+        return f"Error: File not found: {file_path}"
+
+    if not path.is_file():
+        return f"Error: Path is not a file: {file_path}\nThis tool can only read files, not directories."
+
+    try:
+        lines = path.read_text().splitlines()
+    except Exception as e:
+        return f"Error reading file {file_path}: {e}"
+
+    # Handle offset and limit
+    start = (offset - 1) if offset and offset > 0 else 0
+    end = (start + limit) if limit else len(lines)
+
+    # Format with line numbers
+    result_lines = []
+    for i, line in enumerate(lines[start:end], start=start + 1):
+        # Truncate long lines
+        if len(line) > 2000:
+            line = line[:2000] + "..."
+        result_lines.append(f"{i:6d}|{line}")
+
+    if not result_lines:
+        return "File is empty."
+
+    return "\n".join(result_lines)
+
+
+def write_file_tool(file_path: str, content: str, tool_context: ToolContext):
+    """Write content to files (overwrites existing files).
+
+    Writes content to a file on the filesystem.
+
+    Working directory structure:
+        /tmp/veadk/{session_id}/
+        ├── skills/     -> all skills are available here (read-only).
+        ├── uploads/    -> staged user files (temporary)
+        └── outputs/    -> generated files for return
+
+    Usage:
+    - Provide a path (absolute or relative to working directory) and content to write
+    - Overwrites existing files
+    - Creates parent directories if needed
+    - For existing files, read them first using read_file
+    - Prefer editing existing files over writing new ones
+    - You can write to your working directory, outputs/, or any writable location
+    - Note: skills/ directory is read-only
+
+    Args:
+        file_path: Path to the file to write (absolute or relative to working directory)
+        content: Content to write to the file
+        tool_context: Context of the tool execution
+
+    Returns:
+        Success message or error message.
+    """
+    if not file_path:
+        return "Error: No file path provided"
+
+    # Resolve path relative to session working directory
+    working_dir = get_session_path(session_id=tool_context.session.id)
+    path = Path(file_path)
+    if not path.is_absolute():
+        path = working_dir / path
+    path = path.resolve()
+
+    try:
+        # Create parent directories if needed
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(content)
+        logger.info(f"Successfully wrote to {file_path}")
+        return f"Successfully wrote to {file_path}"
+    except Exception as e:
+        error_msg = f"Error writing file {file_path}: {e}"
+        logger.error(error_msg)
+        return error_msg
+
+
+def edit_file_tool(
+    file_path: str,
+    old_string: str,
+    new_string: str,
+    replace_all: bool,
+    tool_context: ToolContext,
+):
+    """Edit files by replacing exact string matches.
+
+    Working directory structure:
+        /tmp/veadk/{session_id}/
+        ├── skills/     -> all skills are available here (read-only).
+        ├── uploads/    -> staged user files (temporary)
+        └── outputs/    -> generated files for return
+
+    Performs exact string replacements in files.
+    Usage:
+    - You must read the file first using read_file
+    - Provide path (absolute or relative to working directory)
+    - When editing, preserve exact indentation from the file content
+    - Do NOT include line number prefixes in old_string or new_string
+    - old_string must be unique unless replace_all=true
+    - Use replace_all to rename variables/strings throughout the file
+    - old_string and new_string must be different
+    - Note: skills/ directory is read-only
+
+    Args:
+        file_path: Path to the file to edit (absolute or relative to working directory)
+        old_string: The exact text to replace (must exist in file)
+        new_string: The text to replace it with (must be different from old_string)
+        replace_all: Replace all occurrences (default: false, only replaces first occurrence)
+        tool_context: Context of the tool execution
+
+    Returns:
+        Success message or error message.
+    """
+    if not file_path:
+        return "Error: No file path provided"
+
+    if old_string == new_string:
+        return "Error: old_string and new_string must be different"
+
+    # Resolve path relative to session working directory
+    working_dir = get_session_path(session_id=tool_context.session.id)
+    path = Path(file_path)
+    if not path.is_absolute():
+        path = working_dir / path
+    path = path.resolve()
+
+    if not path.exists():
+        return f"Error: File not found: {file_path}"
+
+    if not path.is_file():
+        return f"Error: Path is not a file: {file_path}"
+
+    try:
+        content = path.read_text()
+    except Exception as e:
+        return f"Error reading file {file_path}: {e}"
+
+    # Check if old_string exists
+    if old_string not in content:
+        return (
+            f"Error: old_string not found in {file_path}.\n"
+            f"Make sure you've read the file first and are using the exact string."
+        )
+
+    # Count occurrences
+    count = content.count(old_string)
+
+    if not replace_all and count > 1:
+        return (
+            f"Error: old_string appears {count} times in {file_path}.\n"
+            f"Either provide more context to make it unique, or set "
+            f"replace_all=true to replace all occurrences."
+        )
+
+    # Perform replacement
+    if replace_all:
+        new_content = content.replace(old_string, new_string)
+    else:
+        new_content = content.replace(old_string, new_string, 1)
+
+    try:
+        path.write_text(new_content)
+        logger.info(f"Successfully replaced {count} occurrence(s) in {file_path}")
+        return f"Successfully replaced {count} occurrence(s) in {file_path}"
+    except Exception as e:
+        error_msg = f"Error writing file {file_path}: {e}"
+        logger.error(error_msg)
+        return error_msg
diff --git a/veadk/tools/skills_tools/session_path.py b/veadk/tools/skills_tools/session_path.py
new file mode 100644
index 00000000..2e985d41
--- /dev/null
+++ b/veadk/tools/skills_tools/session_path.py
@@ -0,0 +1,107 @@
+# 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.
+
+import tempfile
+from pathlib import Path
+from veadk.utils.logger import get_logger
+
+logger = get_logger(__name__)
+
+# Cache of initialized session paths to avoid re-creating symlinks
+_session_path_cache: dict[str, Path] = {}
+
+
+def initialize_session_path(session_id: str) -> Path:
+    """Initialize a session's working directory with skills symlink.
+
+    This is called by SkillsPlugin.before_agent_callback() to ensure the session
+    is set up before any tools run. Creates the directory structure and symlink
+    to the skills directory.
+
+    Directory structure:
+        /tmp/veadk/{session_id}/
+        ├── skills/     -> symlink to skills_directory (read-only shared skills)
+        ├── uploads/    -> staged user files (temporary)
+        └── outputs/    -> generated files for return
+
+    Args:
+        session_id: The unique ID of the current session.
+        skills_directory: Path to the shared skills directory.
+
+    Returns:
+        The resolved path to the session's root directory.
+    """
+    # Return cached path if already initialized
+    if session_id in _session_path_cache:
+        return _session_path_cache[session_id]
+
+    # Initialize new session path
+    base_path = Path(tempfile.gettempdir()) / "veadk"
+    session_path = base_path / session_id
+
+    # Create working directories
+    (session_path / "skills").mkdir(parents=True, exist_ok=True)
+    (session_path / "uploads").mkdir(parents=True, exist_ok=True)
+    (session_path / "outputs").mkdir(parents=True, exist_ok=True)
+
+    # Cache and return
+    resolved_path = session_path.resolve()
+    _session_path_cache[session_id] = resolved_path
+    return resolved_path
+
+
+def get_session_path(session_id: str) -> Path:
+    """Get the working directory path for a session.
+
+    This function retrieves the cached session path that was initialized by
+    SkillsPlugin. If the session hasn't been initialized (plugin not used),
+    it falls back to auto-initialization with default /skills directory.
+
+    Tools should call this function to get their working directory. The session
+    must be initialized by SkillsPlugin before tools run, which happens automatically
+    via the before_agent_callback() hook.
+
+    Args:
+        session_id: The unique ID of the current session.
+
+    Returns:
+        The resolved path to the session's root directory.
+
+    Note:
+        If session is not initialized, automatically initializes with /skills.
+        For custom skills directories, ensure SkillsPlugin is installed.
+    """
+    # Return cached path if already initialized
+    if session_id in _session_path_cache:
+        return _session_path_cache[session_id]
+
+    # Fallback: auto-initialize with default /skills
+    logger.warning(
+        f"Session {session_id} not initialized by SkillsPlugin. "
+        f"Auto-initializing with default /skills. "
+        f"Install SkillsPlugin for custom skills directories."
+    )
+    return initialize_session_path(session_id, "/skills")
+
+
+def clear_session_cache(session_id: str | None = None) -> None:
+    """Clear cached session path(s).
+
+    Args:
+        session_id: Specific session to clear. If None, clears all cached sessions.
+    """
+    if session_id:
+        _session_path_cache.pop(session_id, None)
+    else:
+        _session_path_cache.clear()
diff --git a/veadk/tools/skills_tools/skills_tool.py b/veadk/tools/skills_tools/skills_tool.py
new file mode 100644
index 00000000..06854bd1
--- /dev/null
+++ b/veadk/tools/skills_tools/skills_tool.py
@@ -0,0 +1,236 @@
+# 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
+
+import os
+import tempfile
+from pathlib import Path
+from typing import Any, Dict
+
+from google.adk.tools import BaseTool, ToolContext
+from google.genai import types
+
+from veadk.skills.skill import Skill
+from veadk.utils.logger import get_logger
+
+logger = get_logger(__name__)
+
+
+class SkillsTool(BaseTool):
+    """Discover and load skill instructions.
+
+    This tool dynamically discovers available skills and embeds their metadata in the
+    tool description. Agent invokes a skill by name to load its full instructions.
+    """
+
+    def __init__(self, skills: Dict[str, Skill]):
+        self.skills = skills
+
+        # Generate description with available skills embedded
+        description = self._generate_description_with_skills()
+
+        super().__init__(
+            name="skills",
+            description=description,
+        )
+
+    def _generate_description_with_skills(self) -> str:
+        """Generate tool description with available skills embedded."""
+        base_description = (
+            "Execute a skill within the main conversation\n\n"
+            "<skills_instructions>\n"
+            "When users ask you to perform tasks, check if any of the available skills below can help "
+            "complete the task more effectively. Skills provide specialized capabilities and domain knowledge.\n\n"
+            "How to use skills:\n"
+            "- Invoke skills using this tool with the skill name only (no arguments)\n"
+            "- When you invoke a skill, the skill's full SKILL.md will load with detailed instructions\n"
+            "- Follow the skill's instructions and use the bash tool to execute commands\n"
+            "- Examples:\n"
+            '  - command: "data-analysis" - invoke the data-analysis skill\n'
+            '  - command: "pdf-processing" - invoke the pdf-processing skill\n\n'
+            "Important:\n"
+            "- Do not invoke a skill that is already loaded in the conversation\n"
+            "- After loading a skill, use the bash tool for execution\n"
+            "- If not specified, scripts are located in the skill-name/scripts subdirectory\n"
+            "</skills_instructions>\n\n"
+        )
+
+        return base_description
+
+    def _get_declaration(self) -> types.FunctionDeclaration:
+        return types.FunctionDeclaration(
+            name=self.name,
+            description=self.description,
+            parameters=types.Schema(
+                type=types.Type.OBJECT,
+                properties={
+                    "command": types.Schema(
+                        type=types.Type.STRING,
+                        description='The skill name (no arguments). E.g., "data-analysis" or "pdf-processing"',
+                    ),
+                },
+                required=["command"],
+            ),
+        )
+
+    async def run_async(
+        self, *, args: Dict[str, Any], tool_context: ToolContext
+    ) -> str:
+        """Execute skill loading by name."""
+        skill_name = args.get("command", "").strip()
+
+        if not skill_name:
+            return "Error: No skill name provided"
+
+        return self._invoke_skill(skill_name, tool_context)
+
+    def _invoke_skill(self, skill_name: str, tool_context: ToolContext) -> str:
+        """Load and return the full content of a skill."""
+        if skill_name not in self.skills:
+            return f"Error: Skill '{skill_name}' does not exist."
+
+        skill = self.skills[skill_name]
+        skill_dir = (
+            Path(tempfile.gettempdir()) / "veadk" / tool_context.session.id / "skills"
+        )
+
+        if skill.skill_space_id:
+            logger.info(f"Attempting to download skill '{skill_name}' from skill space")
+            try:
+                from veadk.auth.veauth.utils import get_credential_from_vefaas_iam
+                from veadk.integrations.ve_tos.ve_tos import VeTOS
+
+                access_key = os.getenv("VOLCENGINE_ACCESS_KEY")
+                secret_key = os.getenv("VOLCENGINE_SECRET_KEY")
+                session_token = ""
+
+                if not (access_key and secret_key):
+                    # Try to get from vefaas iam
+                    cred = get_credential_from_vefaas_iam()
+                    access_key = cred.access_key_id
+                    secret_key = cred.secret_access_key
+                    session_token = cred.session_token
+
+                tos_bucket, tos_path = skill.bucket_name, skill.path
+
+                # Initialize VeTOS client
+                tos_client = VeTOS(
+                    ak=access_key,
+                    sk=secret_key,
+                    session_token=session_token,
+                    bucket_name=tos_bucket,
+                )
+
+                save_path = skill_dir / f"{skill_name}.zip"
+
+                success = tos_client.download(
+                    bucket_name=tos_bucket,
+                    object_key=tos_path,
+                    save_path=save_path,
+                )
+
+                if not success:
+                    return f"Error: Failed to download skill '{skill_name}' from TOS."
+
+                # Extract downloaded zip into the skill directory
+                import zipfile
+                import shutil
+
+                # Remove existing skill directory to ensure clean extraction
+                target_skill_dir = skill_dir / skill_name
+                if target_skill_dir.exists():
+                    try:
+                        shutil.rmtree(target_skill_dir)
+                        logger.info(
+                            f"Removed existing skill directory: {target_skill_dir}"
+                        )
+                    except Exception as e:
+                        logger.warning(
+                            f"Failed to remove existing skill directory {target_skill_dir}: {e}"
+                        )
+
+                try:
+                    with zipfile.ZipFile(save_path, "r") as z:
+                        z.extractall(path=str(skill_dir))
+                except zipfile.BadZipFile:
+                    logger.error(
+                        f"Downloaded file for '{skill_name}' is not a valid zip"
+                    )
+                    return f"Error: Downloaded file for skill '{skill_name}' is not a valid zip archive."
+                except Exception as e:
+                    logger.error(f"Failed to extract skill zip for '{skill_name}': {e}")
+                    return (
+                        f"Error: Failed to extract skill '{skill_name}' from zip: {e}"
+                    )
+
+                logger.info(
+                    f"Successfully downloaded skill '{skill_name}' from skill space"
+                )
+
+            except Exception as e:
+                logger.error(
+                    f"Failed to download skill '{skill_name}' from skill space: {e}"
+                )
+                return (
+                    f"Error: Skill '{skill_name}' not found locally and failed to download from skill space: {e}. "
+                    f"Check the available skills list in the tool description."
+                )
+        else:
+            # Create symlink to skills directory
+            skills_mount = Path(skill.path)
+            skills_link = skill_dir / skill_name
+            if skills_mount.exists() and not skills_link.exists():
+                try:
+                    skills_link.symlink_to(skills_mount)
+                    logger.debug(f"Created symlink: {skills_link} -> {skills_mount}")
+                except FileExistsError:
+                    # Symlink already exists (race condition from concurrent session setup)
+                    pass
+                except Exception as e:
+                    # Log but don't fail - skills can still be accessed via absolute path
+                    logger.warning(
+                        f"Failed to create skills symlink for {str(skills_mount)}: {e}"
+                    )
+
+        skill_file = skill_dir / skill_name / "SKILL.md"
+        if not skill_file.exists():
+            return f"Error: Skill '{skill_name}' has no SKILL.md file."
+
+        try:
+            with open(skill_file, "r", encoding="utf-8") as f:
+                content = f.read()
+
+            formatted_content = self._format_skill_content(
+                skill_name, content, str(skill_dir)
+            )
+
+            logger.info(f"Invoke skill '{skill_name}' successfully.")
+            return formatted_content
+
+        except Exception as e:
+            logger.error(f"Failed to invoke skill {skill_name}: {e}")
+            return f"Error invoking skill '{skill_name}': {e}"
+
+    def _format_skill_content(self, skill_name: str, content: str, skill_dir) -> str:
+        """Format skill content for display to the agent."""
+        header = (
+            f'<command-message>The "{skill_name}" skill is loading</command-message>\n\n'
+            f"Base directory for this skill: {skill_dir}/{skill_name}\n\n"
+        )
+        footer = (
+            "\n\n---\n"
+            "The skill has been loaded. Follow the instructions above and use the bash tool to execute commands."
+        )
+        return header + content + footer