Terminal Tool — Agent 安全命令执行设计分析

发表于 更新于 分类于

Terminal Tool 实现分析

来源: /AWorld/examples/gaia/mcp_collections/tools/terminal.py

概述

Terminal MCP Server 是一个基于 Python asyncio 的终端命令执行工具，作为 MCP 服务对外暴露。它让 LLM Agent 能够安全地执行 shell 命令并获取格式化结果。

核心功能

方法 功能
mcp_execute_command(command, timeout, output_format) 执行 shell 命令，支持超时控制和格式化输出
mcp_get_command_history(count, output_format) 获取命令执行历史
mcp_get_terminal_capabilities() 获取终端服务能力信息

安全机制

1. 危险命令拦截

_check_command_safety() 维护一个黑名单，拦截如下命令：

1
2
3
4
5
[
    "rm -rf /", "mkfs", "dd if=", ":(){ :|:& };:",  # fork bomb
    "del /f /s /q", "diskpart",                      # Windows 危险命令
    "sudo rm", "sudo dd", "sudo mkfs",               # sudo 变体
]

2. 交互式命令拦截

_check_interactive_command() 通过正则匹配禁止交互式命令（vim, vi, nano, emacs, ftp, telnet），要求用户改用非交互式替代方案（如 --yes、CI=1、DEBIAN_FRONTEND=noninteractive 等）。

3. 超时保护

asyncio.wait_for(process.communicate(), timeout) 实现超时控制，超时后通过 os.killpg() 杀死整个进程组，避免遗留子进程。

核心执行流程

1
2
3
4
5
6
7
8
9
10
11
12
13
mcp_execute_command()
├── 参数校验（FieldInfo → 实际值）
├── 超时值校验（1s ≤ timeout ≤ 3600s）
├── 安全检测
│   ├── _check_command_safety()          # 危险命令检查
│   └── _check_interactive_command()     # 交互命令检查
├── _execute_command_async()
│   ├── asyncio.create_subprocess_shell()  # 创建子进程
│   │   ├── stdin=subprocess.DEVNULL       # 禁止标准输入
│   │   └── start_new_session=True         # 独立进程组
│   └── asyncio.wait_for(process.communicate(), timeout)
│       └── 超时 → os.killpg() 杀死整个进程组
└── _format_command_output()               # 格式化输出

为什么要屏蔽交互式命令

这是 Terminal Tool 设计中最关键的边界情况。交互式命令（如 vim、nano、less、ftp）在非交互环境中执行时会产生严重问题，原因如下：

1. 进程阻塞：父子进程的管道僵局

_execute_command_async() 中设置 stdin=subprocess.DEVNULL，意味着子进程的 stdin 是一个空设备（/dev/null）。当交互式命令启动后，它会尝试从 stdin 读取用户输入：

1
2
3
4
5
6
7
process = await asyncio.create_subprocess_shell(
    command,
    stdin=subprocess.DEVNULL,  # ← 关键：子进程的输入被关闭
    stdout=asyncio.subprocess.PIPE,
    stderr=asyncio.subprocess.PIPE,
    shell=True,
)

交互式命令的行为：

  1. 尝试读取 stdin → 读到 EOF（因为 DEVNULL 会立即返回 EOF）
  2. 部分交互式工具检测到 EOF 后会自行退出
  3. 但并非所有工具都如此——很多工具（如 less、未加 -E 的 vim）的行为：
    • 陷入空轮询或等待状态
    • 不退出的情况下，process.communicate() 永远不会返回
    • asyncio.wait_for() 最终触发超时

2. Python 子进程模型中的超时连锁问题

当 asyncio.wait_for() 超时触发时：

1
2
3
4
5
except asyncio.TimeoutError:
    if self.platform_info["system"] != "Windows" and process.pid:
        os.killpg(os.getpgid(process.pid), 9)  # 杀死整个进程组
    else:
        process.kill()                          # 仅杀子进程本身

Python 父子进程模型下的超时连锁问题：

1
2
3
4
5
6
7
8
9
父进程（Agent/MCP Server）
├── asyncio event loop
│   └── 创建子进程（shell=True）
│       └── shell
│           └── 交互式命令（vim/nano/less）
│               ├── 从 stdin 读取 → EOF（DEVNULL）
│               └── 陷入空轮询 → 永不退出 ×
│
└── asyncio.wait_for() 超时 → os.killpg()

具体问题：

  1. subprocess.DEVNULL → 子进程 stdin 为 /dev/null：父进程无法向子进程写入任何输入，交互式命令永远得不到用户响应。

  2. 无法唤醒/响应：即使 Agent 想”按 :q 退出 vim”，DEVNULL 让这个操作根本无法完成。子进程的 stdin 通道被永久关闭。

  3. 超时带来的级联影响：

    • 超时如果只调用 process.kill()，仅杀 shell 子进程本身
    • 交互式命令可能派生自己的子进程（如 vim 的交换文件写入进程），成为孤儿进程继续消耗资源
    • 设置 start_new_session=True 后使用 os.killpg() 可以杀整个进程组，但前提是交互命令没有创建新的 session（setsid()）
    • 极端情况下（如 fork bomb 变体 :(){ :|:& };:）：会不断 fork，即使 kill 当前进程，新进程已在其他进程组继续运行

  4. 超时的累积效应：

    • 默认超时 300 秒，如果 Agent 连续执行交互式命令，每次都要等待超时
    • 每个超时命令浪费 5 分钟，多个命令叠加导致 Agent 任务整体超时
    • Agent 的有限上下文窗口被无意义的超时错误信息挤占

3. 输出混乱与管道压力

交互式命令的输出（如 vim 的终端转义序列）会污染 Agent 的上下文：

1
2
stdout: ^[[H^[[2J^[[3J（ANSI 转义码）
stderr: Vim: Warning: Input is not from a terminal

这些内容对 LLM 的分析毫无价值，反而占用了宝贵的 token 预算，且结构化解析也会被中断。

4. 为什么不直接用 pipes 模拟输入

从设计上看，虽然可以用 process.stdin.write(b":q\n") 的方式向进程发送命令，但这个方案的可靠性很差：

  • 需要为每个交互式命令编写不同的输入序列（极难维护）
  • 无法处理交互过程中的非预期状态（如 vim 的 swap 文件冲突弹窗）
  • 对于某些全屏 TUI 应用，发送输入后输出仍包含大量转义控制字符，对 LLM 不友好
  • 与 stdin=subprocess.DEVNULL 的基本设计矛盾（DEVNULL 意味着子进程的 stdin 是只读 /dev/null，无法写入）

最佳实践总结

情况 做法
非交互命令 直接执行，subprocess.DEVNULL 安全
交互式命令（vim, less 等） 禁止，提示改用非交互参数（--yes、CI=1）
需输入确认的命令 Agent 应提前构造好非交互参数
需长时间执行的命令 配置合理的 timeout 值
网络请求（curl/wget） 添加 --no-progress-meter（curl）或 -q（wget）减少 stderr 管道压力

关键设计决策

  1. **stdin=subprocess.DEVNULL**：杜绝所有需要标准输入的命令，从根本上避免输入阻塞
  2. **start_new_session=True**（Unix）：子进程独立进程组，超时后可安全杀掉整个子树
  3. **shell=True 配合 /bin/bash**：支持管道、重定向等 shell 特性
  4. os.killpg() 超时清理：如果超时，kill 整个进程组避免孤儿进程
  5. ExceptionGroup 兼容：兼容 Python 3.11+ 的异常组，避免 anyio TaskGroup 崩溃级联