feat(skill): Add BashTool

PiperOrigin-RevId: 875899505

Author: cornellgit

src/google/adk/tools/bash_tool.py

@@ -0,0 +1,150 @@
+# Copyright 2026 Google LLC
+#
+# 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.
+
+"""Tool to execute bash commands."""
+
+from __future__ import annotations
+
+import dataclasses
+import pathlib
+import shlex
+import subprocess
+from typing import Any
+from typing import Optional
+
+from google.genai import types
+
+from .. import features
+from .base_tool import BaseTool
+from .tool_context import ToolContext
+
+
+@dataclasses.dataclass(frozen=True)
+class BashToolPolicy:
+  """Configuration for allowed bash commands based on prefix matching.
+
+  Set allowed_command_prefixes to ("*",) to allow all commands (default),
+  or explicitly list allowed prefixes.
+  """
+
+  allowed_command_prefixes: tuple[str, ...] = ("*",)
+
+
+def _validate_command(command: str, policy: BashToolPolicy) -> Optional[str]:
+  """Validates a bash command against the permitted prefixes."""
+  stripped = command.strip()
+  if not stripped:
+    return "Command is required."
+
+  if "*" in policy.allowed_command_prefixes:
+    return None
+
+  for prefix in policy.allowed_command_prefixes:
+    if stripped.startswith(prefix):
+      return None
+
+  allowed = ", ".join(policy.allowed_command_prefixes)
+  return f"Command blocked. Permitted prefixes are: {allowed}"
+
+
+@features.experimental(features.FeatureName.SKILL_TOOLSET)
+class ExecuteBashTool(BaseTool):
+  """Tool to execute a validated bash command within a workspace directory."""
+
+  def __init__(
+      self,
+      *,
+      workspace: pathlib.Path | None = None,
+      policy: Optional[BashToolPolicy] = None,
+  ):
+    if workspace is None:
+      workspace = pathlib.Path.cwd()
+    policy = policy or BashToolPolicy()
+    allowed_hint = (
+        "any command"
+        if "*" in policy.allowed_command_prefixes
+        else (
+            "commands matching prefixes:"
+            f" {', '.join(policy.allowed_command_prefixes)}"
+        )
+    )
+    super().__init__(
+        name="execute_bash",
+        description=(
+            "Executes a bash command with the working directory set to the"
+            f" workspace. Allowed: {allowed_hint}. All commands require user"
+            " confirmation."
+        ),
+    )
+    self._workspace = workspace
+    self._policy = policy
+
+  def _get_declaration(self) -> Optional[types.FunctionDeclaration]:
+    return types.FunctionDeclaration(
+        name=self.name,
+        description=self.description,
+        parameters_json_schema={
+            "type": "object",
+            "properties": {
+                "command": {
+                    "type": "string",
+                    "description": "The bash command to execute.",
+                },
+            },
+            "required": ["command"],
+        },
+    )
+
+  async def run_async(
+      self, *, args: dict[str, Any], tool_context: ToolContext
+  ) -> Any:
+    command = args.get("command")
+    if not command:
+      return {"error": "Command is required."}
+
+    # Static validation.
+    error = _validate_command(command, self._policy)
+    if error:
+      return {"error": error}
+
+    # Always request user confirmation.
+    if not tool_context.tool_confirmation:
+      tool_context.request_confirmation(
+          hint=f"Please approve or reject the bash command: {command}",
+      )
+      tool_context.actions.skip_summarization = True
+      return {
+          "error": (
+              "This tool call requires confirmation, please approve or reject."
+          )
+      }
+    elif not tool_context.tool_confirmation.confirmed:
+      return {"error": "This tool call is rejected."}
+
+    try:
+      result = subprocess.run(
+          shlex.split(command),
+          shell=False,
+          cwd=str(self._workspace),
+          capture_output=True,
+          text=True,
+          timeout=30,
+      )
+      return {
+          "stdout": result.stdout,
+          "stderr": result.stderr,
+          "returncode": result.returncode,
+      }
+    except subprocess.TimeoutExpired:
+      return {"error": "Command timed out after 30 seconds."}

tests/unittests/tools/test_bash_tool.py

@@ -0,0 +1,229 @@
+# Copyright 2026 Google LLC
+#
+# 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 unittest import mock
+
+from google.adk.tools import bash_tool
+from google.adk.tools import tool_context
+from google.adk.tools.tool_confirmation import ToolConfirmation
+import pytest
+
+
+@pytest.fixture
+def workspace(tmp_path):
+  """Creates a workspace mirroring the anthropics/skills PDF skill layout."""
+  # Skill: pdf/
+  skill_dir = tmp_path / "pdf"
+  skill_dir.mkdir()
+  (skill_dir / "SKILL.md").write_text(
+      "---\nname: pdf\n"
+      "description: Use this skill whenever the user wants to do"
+      " anything with PDF files.\n"
+      "---\n# PDF Processing Guide\n\n## Overview\n"
+      "This guide covers PDF processing operations."
+  )
+  scripts = skill_dir / "scripts"
+  scripts.mkdir()
+  (scripts / "extract_form_structure.py").write_text(
+      "import sys; print(f'extracting from {sys.argv[1]}')"
+  )
+  (scripts / "fill_pdf_form_with_annotations.py").write_text(
+      "print('filling form')"
+  )
+  references = skill_dir / "references"
+  references.mkdir()
+  (references / "REFERENCE.md").write_text("# Reference\nDetailed docs.")
+  # A loose file at workspace root (not inside a skill).
+  (tmp_path / "sample.pdf").write_bytes(b"%PDF-1.4 fake")
+  return tmp_path
+
+
+@pytest.fixture
+def tool_context_no_confirmation():
+  """ToolContext with no confirmation (initial call)."""
+  ctx = mock.create_autospec(tool_context.ToolContext, instance=True)
+  ctx.tool_confirmation = None
+  ctx.actions = mock.MagicMock()
+  return ctx
+
+
+@pytest.fixture
+def tool_context_confirmed():
+  """ToolContext with confirmation approved."""
+  ctx = mock.create_autospec(tool_context.ToolContext, instance=True)
+  confirmation = mock.create_autospec(ToolConfirmation, instance=True)
+  confirmation.confirmed = True
+  ctx.tool_confirmation = confirmation
+  ctx.actions = mock.MagicMock()
+  return ctx
+
+
+@pytest.fixture
+def tool_context_rejected():
+  """ToolContext with confirmation rejected."""
+  ctx = mock.create_autospec(tool_context.ToolContext, instance=True)
+  confirmation = mock.create_autospec(ToolConfirmation, instance=True)
+  confirmation.confirmed = False
+  ctx.tool_confirmation = confirmation
+  ctx.actions = mock.MagicMock()
+  return ctx
+
+
+# --- _validate_command tests ---
+
+
+class TestValidateCommand:
+
+  def test_empty_command(self):
+    policy = bash_tool.BashToolPolicy()
+    assert bash_tool._validate_command("", policy) is not None
+    assert bash_tool._validate_command("   ", policy) is not None
+
+  def test_default_policy_allows_everything(self):
+    policy = bash_tool.BashToolPolicy()
+    assert bash_tool._validate_command("rm -rf /", policy) is None
+    assert bash_tool._validate_command("cat /etc/passwd", policy) is None
+    assert bash_tool._validate_command("sudo curl", policy) is None
+
+  def test_restricted_policy_allows_prefixes(self):
+    policy = bash_tool.BashToolPolicy(allowed_command_prefixes=("ls", "cat"))
+    assert bash_tool._validate_command("ls -la", policy) is None
+    assert bash_tool._validate_command("cat file.txt", policy) is None
+
+  def test_restricted_policy_blocks_others(self):
+    policy = bash_tool.BashToolPolicy(allowed_command_prefixes=("ls", "cat"))
+    assert bash_tool._validate_command("rm -rf .", policy) is not None
+    assert bash_tool._validate_command("tree", policy) is not None
+    assert "Permitted prefixes are: ls, cat" in bash_tool._validate_command(
+        "tree", policy
+    )
+
+
+class TestExecuteBashTool:
+
+  @pytest.mark.asyncio
+  async def test_requests_confirmation(
+      self, workspace, tool_context_no_confirmation
+  ):
+    tool = bash_tool.ExecuteBashTool(workspace=workspace)
+    result = await tool.run_async(
+        args={"command": "ls"},
+        tool_context=tool_context_no_confirmation,
+    )
+    assert "error" in result
+    assert "requires confirmation" in result["error"]
+    tool_context_no_confirmation.request_confirmation.assert_called_once()
+
+  @pytest.mark.asyncio
+  async def test_rejected(self, workspace, tool_context_rejected):
+    tool = bash_tool.ExecuteBashTool(workspace=workspace)
+    result = await tool.run_async(
+        args={"command": "ls"}, tool_context=tool_context_rejected
+    )
+    assert result == {"error": "This tool call is rejected."}
+
+  @pytest.mark.asyncio
+  async def test_executes_when_confirmed(
+      self, workspace, tool_context_confirmed
+  ):
+    tool = bash_tool.ExecuteBashTool(workspace=workspace)
+    result = await tool.run_async(
+        args={"command": "ls"},
+        tool_context=tool_context_confirmed,
+    )
+    assert result["returncode"] == 0
+    assert "pdf" in result["stdout"]
+
+  @pytest.mark.asyncio
+  async def test_cat_skill_md(self, workspace, tool_context_confirmed):
+    tool = bash_tool.ExecuteBashTool(workspace=workspace)
+    result = await tool.run_async(
+        args={"command": "cat pdf/SKILL.md"},
+        tool_context=tool_context_confirmed,
+    )
+    assert "PDF Processing Guide" in result["stdout"]
+
+  @pytest.mark.asyncio
+  async def test_python_script(self, workspace, tool_context_confirmed):
+    tool = bash_tool.ExecuteBashTool(workspace=workspace)
+    result = await tool.run_async(
+        args={
+            "command": "python3 pdf/scripts/extract_form_structure.py test.pdf"
+        },
+        tool_context=tool_context_confirmed,
+    )
+    assert "extracting from test.pdf" in result["stdout"]
+    assert result["returncode"] == 0
+
+  @pytest.mark.asyncio
+  async def test_blocks_disallowed_by_policy(
+      self, workspace, tool_context_no_confirmation
+  ):
+    policy = bash_tool.BashToolPolicy(allowed_command_prefixes=("ls",))
+    tool = bash_tool.ExecuteBashTool(workspace=workspace, policy=policy)
+    result = await tool.run_async(
+        args={"command": "rm -rf ."},
+        tool_context=tool_context_no_confirmation,
+    )
+    assert "error" in result
+    assert "Permitted prefixes are: ls" in result["error"]
+    tool_context_no_confirmation.request_confirmation.assert_not_called()
+
+  @pytest.mark.asyncio
+  async def test_captures_stderr(self, workspace, tool_context_confirmed):
+    tool = bash_tool.ExecuteBashTool(workspace=workspace)
+    result = await tool.run_async(
+        args={"command": "python3 -c 'import sys; sys.stderr.write(\"err\")'"},
+        tool_context=tool_context_confirmed,
+    )
+    assert "err" in result["stderr"]
+
+  @pytest.mark.asyncio
+  async def test_nonzero_returncode(self, workspace, tool_context_confirmed):
+    tool = bash_tool.ExecuteBashTool(workspace=workspace)
+    result = await tool.run_async(
+        args={"command": "python3 -c 'exit(42)'"},
+        tool_context=tool_context_confirmed,
+    )
+    assert result["returncode"] == 42
+
+  @pytest.mark.asyncio
+  async def test_timeout(self, workspace, tool_context_confirmed):
+    tool = bash_tool.ExecuteBashTool(workspace=workspace)
+    with mock.patch(
+        "google.adk.tools.bash_tool.subprocess.run",
+        side_effect=__import__("subprocess").TimeoutExpired("cmd", 30),
+    ):
+      result = await tool.run_async(
+          args={"command": "python scripts/do_thing.py"},
+          tool_context=tool_context_confirmed,
+      )
+    assert "error" in result
+    assert "timed out" in result["error"].lower()
+
+  @pytest.mark.asyncio
+  async def test_cwd_is_workspace(self, workspace, tool_context_confirmed):
+    tool = bash_tool.ExecuteBashTool(workspace=workspace)
+    result = await tool.run_async(
+        args={"command": "python3 -c 'import os; print(os.getcwd())'"},
+        tool_context=tool_context_confirmed,
+    )
+    assert result["stdout"].strip() == str(workspace)
+
+  @pytest.mark.asyncio
+  async def test_no_command(self, workspace, tool_context_confirmed):
+    tool = bash_tool.ExecuteBashTool(workspace=workspace)
+    result = await tool.run_async(args={}, tool_context=tool_context_confirmed)
+    assert "error" in result
+    assert "required" in result["error"].lower()