lambdagent/agentexample/agent67/tools/shell_executor.py

"""
agent67.tools.shell_executor — Shell 命令执行器

Lambda: λ cmd_json. exec(cmd)
"""
import json
import os
import subprocess

from .safety import is_dangerous, needs_confirmation, ask_user_confirm


def execute_shell(input_json: str) -> str:
    """
    执行 shell 命令。

    输入格式: {"command": "ls -la ~/Desktop"}
    """
    try:
        data = json.loads(input_json)
        cmd = data.get("command", input_json)
    except (json.JSONDecodeError, AttributeError):
        cmd = str(input_json).strip()

    if is_dangerous(cmd):
        return f"🚫 危险命令已拦截: {cmd}"

    if needs_confirmation(cmd):
        if not ask_user_confirm(f"执行命令: {cmd}"):
            return "❌ 用户取消了操作"

    try:
        result = subprocess.run(
            cmd, shell=True, capture_output=True, text=True,
            timeout=30, cwd=os.path.expanduser("~"),
        )
        output = result.stdout.strip()
        if result.returncode != 0 and result.stderr:
            output += f"\n[stderr] {result.stderr.strip()}"
        return output[:3000] if output else "(命令执行成功，无输出)"
    except subprocess.TimeoutExpired:
        return f"⏰ 命令超时 (30s): {cmd}"
    except Exception as e:
        return f"❌ 执行错误: {e}"