1  # Copyright (c) 2024-2026 Tencent Zhuque Lab. All rights reserved.
 2  #
 3  # Licensed under the Apache License, Version 2.0 (the "License");
 4  # you may not use this file except in compliance with the License.
 5  # You may obtain a copy of the License at
 6  #
 7  #     http://www.apache.org/licenses/LICENSE-2.0
 8  #
 9  # Unless required by applicable law or agreed to in writing, software
10  # distributed under the License is distributed on an "AS IS" BASIS,
11  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12  # See the License for the specific language governing permissions and
13  # limitations under the License.
14  #
15  # Requirement: Any integration or derivative work must explicitly attribute
16  # Tencent Zhuque Lab (https://github.com/Tencent/AI-Infra-Guard) in its
17  # documentation or user interface, as detailed in the NOTICE file.
18  
19  import subprocess
20  from typing import Any
21  
22  from tools.registry import register_tool
23  from utils.loging import logger
24  
25  
26  @register_tool
27  def execute_shell(command: str, timeout: int = 30, cwd: str | None = None) -> dict[str, Any]:
28      """执行 Shell 命令
29  
30      Args:
31          command: 要执行的 Shell 命令
32          timeout: 超时时间（秒），默认 30 秒
33          cwd: 执行命令的工作目录（可选）
34  
35      Returns:
36          包含执行结果的字典
37      """
38      try:
39          # 确保 timeout 是数字，防止 float + str 错误 (subprocess 内部会进行 time.time() + timeout)
40          try:
41              timeout = float(timeout) if timeout else 60 * 60 * 6
42          except Exception:
43              timeout = 60 * 60 * 6
44  
45          result = subprocess.run(
46              str(command), shell=True, capture_output=True, text=True, timeout=timeout, cwd=cwd
47          )
48  
49          output = {
50              "success": result.returncode == 0,
51              "stdout": result.stdout,
52              "stderr": result.stderr,
53              "return_code": result.returncode,
54          }
55          return output
56  
57      except subprocess.TimeoutExpired:
58          logger.error(f"Shell command timeout after {timeout}s")
59          return {
60              "success": False,
61              "message": f"Execution timeout after {timeout} seconds",
62              "stdout": "",
63              "stderr": "Timeout",
64              "return_code": -1,
65          }
66      except Exception as e:
67          logger.error(f"Error executing shell command: {e}")
68          return {
69              "success": False,
70              "message": f"Error executing shell command: {str(e)}",
71              "stdout": "",
72              "stderr": str(e),
73              "return_code": -1,
74          }