Method run_async

src/google/adk/tools/bash_tool.py:148–247

(
      self, *, args: dict[str, Any], tool_context: ToolContext
  )

    )

  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."}

    stdout = None
    stderr = None
    try:
      process = await asyncio.create_subprocess_exec(
          *shlex.split(command),
          cwd=str(self._workspace),
          stdout=asyncio.subprocess.PIPE,
          stderr=asyncio.subprocess.PIPE,
          start_new_session=True,
          preexec_fn=lambda: _set_resource_limits(self._policy),
      )

      try:
        stdout, stderr = await asyncio.wait_for(
            process.communicate(), timeout=self._policy.timeout_seconds
        )
      except asyncio.TimeoutError:
        try:
          if process.pid:
            os.killpg(process.pid, signal.SIGKILL)
        except ProcessLookupError:
          pass
        stdout, stderr = await process.communicate()
        return {
            "error": (
                f"Command timed out after {self._policy.timeout_seconds}"
                " seconds."
            ),
            "stdout": (
                stdout.decode(errors="replace")
                if stdout
                else "<no stdout captured>"