CLI 这个东西用得越多就越觉得它像个“孤岛”。明明功能很强但真正要用它的人往往得坐在终端前面记住一堆参数还得会看输出。我一直在想为什么不能让任何一个命令行工具都能被其他系统随意调用、自动编排、甚至变成图形界面后来我做了个小项目叫 CLI-Anything思路很简单把任何 CLI 命令包一层统一暴露成 REST API、WebSocket 流、桌面快捷键这些通用接口。这篇文章就把这个项目的设计思路、实现细节和踩过的坑完整讲一遍适合那些天天跟命令行打交道、又想让终端能力“走出去”的开发者参考。1. 内容整体设计与思路拆解1.1 CLI 的本质是什么为什么需要一层“适配器”一个命令行程序本质上就是一个输入输出黑盒你给它命令行参数喂它标准输入它往标准输出吐结果往标准错误输出吐日志最后用一个退出码告诉你成没成功。这个模型简单可靠几十年没变过。但问题也出在简单上——它默认服务对象是“坐在终端前的人类”而不是其他程序。举个例子运维同学写了个脚本check_disk.sh平时在服务器上跑得好好的。突然有一天业务方想要一个磁盘监控接口让前端页面能实时看到又有一天他想在钉钉群里发一条命令触发检查。这时候如果每次都要重新写一遍逻辑或者用 SSH 去服务器上执行命令再抓文本就会非常痛苦。CLI-Anything 想解决的就是这个场景你只写一次命令怎么跑剩下的转发、格式化、并发控制、超时管理全交给这一层适配器。这个思路其实很像生活中常见的转换插头。你有个欧标的充电器到了不同的国家不需要把充电器拆了重造只需要换一个插座转换头就能适配当地的电源标准。CLI-Anything 就是那个转换头命令本身不用改外面想接什么协议就插什么协议。1.2 方案选型背后的三个关键决策我第一版其实走了一条弯路给每个 CLI 工具单独写一个 Python 脚本用subprocess去调然后每个脚本来回复制粘贴处理超时和日志的代码。实现到第三个命令的时候我就发现维护成本开始失控。任何一个公共逻辑的改动都要同步到所有脚本而且每个脚本里的参数解析方式还不完全一致。所以第二版我下了三个决心。第一个决心是配置驱动而不是代码驱动。把命令的参数、路径、超时、是否需要交互这些信息全部写进 YAML 配置文件里。新增一个命令时只需要加一段配置不需要写任何胶水代码。这样哪怕是不太会写 Python 的人也能通过配置文件接入新工具。第二个决心是先抓住中间表示层。不管外面是 REST、WebSocket 还是未来可能出现的什么新协议我都先把终端交互过程抽象成一个统一的“命令会话”对象。这个对象负责处理进程生命周期、输入输出流、退出码至于外部用 HTTP 还是 WebSocket 来对接都只是对同一个会话对象的不同视图。第三个决心是流式优先。很多 CLI 工具不是一口气返回结果而是像tail -f、ffmpeg那样源源不断输出。这些工具如果只是等命令结束再一次性返回体验会非常差。所以整个适配器从设计之初就必须支持“边执行边推送”的数据流这个决定直接影响了我后面选择用什么技术栈。2. 核心细节解析与实操要点2.1 进程管理与伪终端为什么必须选 pty要适配“任何” CLI最稳妥的方式是启动一个真正的子进程。Python 里最基础的工具是subprocess.Popen可以直接捕获标准输出和标准错误。但用了一段时间我发现它拿不到两类程序的输出一类是会检测“当前是否在终端里”来决定行为模式的程序比如很多工具只有在 TTY 下才会输出彩色信息和进度条另一类是交互式的程序比如python解释器、ssh、ftp它们需要读写一个终端设备才能正常工作。解决方案是用伪终端PTY。伪终端这个东西很奇妙它模拟了一个真实终端设备让子进程认为自己在跟人说话实际上在跟我们的适配器说话。Python 标准库里有pty模块配合subprocess可以把子进程的输入输出挂到一个伪终端上。选 pty 还有另一个好处很多程序在管道模式下会做块缓冲明明输出了内容却不 flush在伪终端下它们通常会改成行缓冲我们就能更快拿到输出。不过 pty 也带来一个新问题它默认会把子进程收到的输入原样回显echo到输出里。如果你向ssh发送了一个密码输出里就会重复出现这份密码。这个问题我会在下一节详细讲。2.2 输出流处理把“终端屏幕”变成事件流在一个真实终端里屏幕显示的是一个二维平面有光标、有滚动区域、有各种控制字符。如果适配器想把这些信息通过 WebSocket 推给浏览器不能直接把原始字节流发过去——浏览器看到了只会乱码。我采取的办法是把“终端屏幕”抽象成一组事件事件类型触发时机示例场景line收到一行完整文本去掉控制字符普通日志输出data收到无法按行切分的原始片段进度条刷新、密码输入提示exit子进程结束命令退出error进程启动失败或超时路径不存在、执行超时对于大多数工具我只会保留line和exit两类事件因为它们足以覆盖 90% 的自动化场景。对于少数特殊工具比如进度条再单独启用data事件。在代码实现上我用asyncio的事件循环来驱动一个 reader 协程。它从 pty 的文件描述符里读数据按行拆分把每一行送到一个队列里。外部 API 层只需从这个队列里异步读取就能做到边执行边推送。2.3 配置系统像写“菜谱”一样定义命令CLI-Anything 的配置我选的是 YAML。配置结构长成这样commands: disk: cmd: /usr/local/bin/check_disk.sh args: - --path - {{path}} mode: once timeout: 30 tail_log: cmd: tail args: - -f - /var/log/app.log mode: stream timeout: 0这里面有几个关键设计点。第一个是参数模板用{{path}}这种占位符来表示运行时传入的参数。外部 API 收到请求后把请求参数填充到模板里再拼接成完整的命令行。第二个是模式区分once表示命令运行完就结束stream表示命令需要持续输出适配器会一直保持会话。第三个是超时语义timeout: 0表示永不超时通常用于流式命令。我还设计了一个简单的类型转换规则如果请求参数是 JSON 里的true但命令行需要字符串true配置里可以加一个type: string来声明。这类细节看起来小但实际用起来非常影响体验。3. 实操过程与核心环节实现3.1 最小可用原型的代码骨架环境准备很简单Python 3.10 以上装fastapi、uvicorn、pyyaml、websockets。不需要数据库不需要消息队列因为 CLI-Anything 的核心逻辑是进程管理而不是业务编排。第一步我先写一个CommandSession类它是整个系统的核心单元。这个类负责启动子进程、管理伪终端、读取输出、等待退出。import asyncio import os import pty import subprocess class CommandSession: def __init__(self, name, cmd_list, timeout): self.name name self.cmd_list cmd_list self.timeout timeout self.proc None self.fd None self.queue asyncio.Queue() self.exit_code None async def start(self): master_fd, slave_fd pty.openpty() self.proc subprocess.Popen( self.cmd_list, stdinslave_fd, stdoutslave_fd, stderrslave_fd, close_fdsTrue, ) os.close(slave_fd) self.fd master_fd loop asyncio.get_running_loop() loop.add_reader(self.fd, self._read_ready) def _read_ready(self): try: data os.read(self.fd, 4096) except OSError: self._terminate() return if not data: self._terminate() return text data.decode(utf-8, errorsreplace) lines text.splitlines() for line in lines: self.queue.put_nowait({type: line, data: line}) # 处理末尾残留 tail text.splitlines(keependsTrue)[-1:] if text else [] if tail and not tail[0].endswith(\n): self.queue.put_nowait({type: partial, data: tail[0]}) def _terminate(self): if self.fd: loop asyncio.get_running_loop() loop.remove_reader(self.fd) os.close(self.fd) self.fd None if self.proc: self.exit_code self.proc.poll() self.queue.put_nowait({type: exit, code: self.exit_code})这段代码里有几个坑。我在第一版时直接读self.proc.stdout结果遇到很多“卡住”的现象因为子进程把输出写进管道缓冲区后没有退出而管道缓冲区又不够大。换成 pty 后这个问题基本消失因为 pty 本身就模拟了一个可交互的字符设备。3.2 关掉回声解决密码重复显示的问题pty 默认开启回显意味着子进程接收到的输入会原样出现在输出流里。对普通命令没什么影响但如果是交互式命令比如ssh输入密码或者ftp输入用户名这些输入会被打出来既难看也容易泄露。解决办法是在启动子进程之前把 pty 的终端属性设置一下关闭ECHO标志。Python 里可以用termios模块操作slave_fd对应的终端import termios attrs termios.tcgetattr(slave_fd) # 第 3 个元素对应 lflag把 ECHO 位关掉 attrs[3] ~termios.ECHO termios.tcsetattr(slave_fd, termios.TCSANOW, attrs)注意这里的顺序必须先设置好终端属性再启动子进程。如果先Popen再设置子进程可能已经读完了初始终端属性关闭回声就不生效了。这是我实际调试中踩过的一个典型时序问题。关掉全局回声之后普通的非交互命令不受影响因为它们的输出本来就不是回显。但那些需要用户输入的交互命令会变得“看不到输入内容”这其实是正确行为——真实终端在输入密码时也是不显示星号的只显示空白。3.3 超时控制与进程组清理任何适配器都必须处理超时。一个卡死的命令不能永远占用资源。我在CommandSession里加入超时逻辑启动后启动一个计时器如果超时命令还没结束就把整个进程组杀掉。为什么要杀进程组因为很多命令会派生子进程比如bash -c sleep 100 wait只杀掉主进程子进程会变成孤儿继续跑。正确做法是启动时让子进程成为新的进程组组长超时后对这个进程组整体发信号。import signal async def start(self): self.proc subprocess.Popen( self.cmd_list, stdinslave_fd, stdoutslave_fd, stderrslave_fd, start_new_sessionTrue, # 让子进程成为新会话首领 close_fdsTrue, ) async def _timeout_handler(self): await asyncio.sleep(self.timeout) if self.proc and self.proc.poll() is None: os.killpg(os.getpgid(self.proc.pid), signal.SIGKILL)这里选定超时值也有一点讲究。对于交互式命令超时应该是指“等待下一次输出”的空闲超时而不是整个会话的总时长对于批处理命令超时则是指总量时长。为了简单CLI-Anything 第一版只做总量超时把配置写成timeout字段。后续扩展时可以增加idle_timeout来做更精细的控制。3.4 用 FastAPI 把命令暴露成 REST 接口有了CommandSession剩下的工作就是把会话包装成不同的外部接口。REST 是最传统、也最容易对接的。我用 FastAPI 写了一个简单的路由from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI() class RunRequest(BaseModel): args: dict {} app.post(/cmd/{name}/run) async def run_command(name: str, req: RunRequest): config load_config()[name] cmd_list build_command(config, req.args) session CommandSession(name, cmd_list, config[timeout]) await session.start() lines [] async for item in session.queue: if item[type] line: lines.append(item[data]) elif item[type] exit: return {lines: lines, code: item[code]} raise HTTPException(status_code500, detailunexpected exit)这个接口的处理逻辑是创建会话消费输出队列等到退出事件出现把收集到的所有行作为 JSON 返回。为了保持接口简单我在一次性命令模式下会消费完全部输出再返回对于流式命令则要改用 WebSocket。这里有个容易被忽略的点队列必须保证先于退出事件被消费完。因为进程可能先写最后一条输出然后立刻退出如果代码先读了退出事件就来不及读那最后一行的输出了。我处理这个问题用了一个技巧读取输出时先处理队列里已经有的数据再处理退出事件不能让退出事件把队列里的残留数据冲掉。换句话说exit事件不是最高优先级line事件才是。3.5 WebSocket 流式输出与前端演示对于tail -f这类持续输出的命令REST 显然不合适。WebSocket 天然适合流式场景。实现也很直接客户端建立连接时传一个命令名和参数服务端创建一个CommandSession然后把队列里的每条line事件实时发给客户端。from fastapi import WebSocket app.websocket(/cmd/{name}/stream) async def stream_command(ws: WebSocket, name: str): await ws.accept() config load_config()[name] args await ws.receive_json() cmd_list build_command(config, args.get(args, {})) session CommandSession(name, cmd_list, config[timeout]) await session.start() try: while True: item await session.queue.get() if item[type] line: await ws.send_text(item[data]) elif item[type] exit: break finally: await session.close()为了演示我写了一个几十行 HTML 的测试页面。页面顶部是一个命令下拉框中间是一个文本框输出区底部是一个输入框用于向进程发送 stdin。这个页面用原生 JavaScript 的 WebSocket API不用任何框架。实测下来WebSocket 跑tail -f /var/log/syslog非常流畅日志一行行出现在页面上延迟基本可以忽略。我还做了一个小优化把浏览器的输入框和 WebSocket 双向绑定用户往输入框里写内容通过 WS 发给适配器再由适配器写入子进程的 stdin。这样有些交互式工具也能在网页里用起来。3.6 并发控制与背压“任何”命令都可以暴露成接口之后紧接着的问题是如果同时有十个请求来跑十条ffmpeg转码命令机器会不会直接垮掉所以适配器需要一层简单的并发控制。我在配置里加了一个max_concurrency字段表示同一个命令最多有几个会话在同时运行。在启动会话前用一个Semaphore来限制sessions {} async def acquire(name): if name not in sessions: sessions[name] asyncio.Semaphore(3) await sessions[name].acquire() async def release(name): sessions[name].release()如果并发数达到上限新的请求应该尽快失败而不是排队等到超时。我在接口层做了处理尝试获取信号量如果拿不到直接返回 429 Too Many Requests。这样调用方可以立刻感知到压力而不是一直傻等。4. 常见问题与排查技巧实录4.1 pty 模式下输出乱码或者字节不完整用 pty 读数据时读出来的是一段字节流不一定正好按行切分。有时一行文本被切成两半有时两行合并成一段。我在_read_ready里用了splitlines()但这种方法在处理行尾没有换行的片段时会丢数据。一个更稳的方案是维护一个缓冲区每次读入数据后只输出完整行剩下的残片留在缓冲区里等下一次读入再拼接self._buffer data while b\n in self._buffer: line, self._buffer self._buffer.split(b\n, 1) self.queue.put_nowait({type: line, data: line.decode(utf-8, errorsreplace)})对于不按行输出的进度条类工具上面的方案会把进度条硬生生拆成几十条“line”事件体验很差。我最后的处理办法是普通模式用“完整行”输出如果配置里声明了raw: true就把每个读到的片段都原样发出去不做行拆分。两种模式各有适用的场景不能一刀切。4.2 命令明明输出了内容但接口迟迟不返回这是“缓冲干等”问题。ping 127.0.0.1这种工具在管道模式下可能会把多行输出攒到一个缓冲区里等缓冲区满了再一次性吐出来。而ping本身运行时间又长就导致接口一直处于等待状态。如果用 pty大多数程序会改成行缓冲问题是grep、sed这类管线工具依然可能做块缓冲。这种情况最简单的解法是给命令加上stdbuf -oL -eL前缀强制标准输出和标准错误都变成行缓冲commands: grep_log: cmd: stdbuf args: - -oL - -eL - grep - {{pattern}}不过stdbuf对静态链接的程序无效比如某些 Go 语言写的二进制文件。这时候就只能靠 pty 或者给程序设置PYTHONUNBUFFERED1这类环境变量来强制无缓冲。4.3 杀不掉残留进程端口被占住如果一个命令自己 spawn 了子进程而我们只杀了主进程子进程可能还活着并且继续占用某些资源。一个典型案例是运行python3 -m http.server 8080适配器超时杀掉了 python 主进程但 socket 可能还处于监听状态导致下一次启动时报端口占用。我的处理策略很明确第一启动进程时设置start_new_sessionTrue第二清理时用killpg杀整个进程组第三在配置里把端口类命令的可并发数设为 1。关于第三点其实最好的方案是让调用方自己管好端口冲突但作为适配层我也提供了字段environment允许为每个会话单独注入环境变量比如动态端口。4.4 WebSocket 断开后进程仍在跑前端页面如果直接关闭WebSocket 的finally块会执行会话被关闭。但有些客户端是异常断开的await session.close()可能会抛异常导致残留进程。我在实际测试中遇到过几次。解决办法是给session.close()加上 try-except并且把关闭逻辑放在一个独立协程里定时检查连接状态。或者在 WebSocket 路由里用一个disconnect保护的循环捕获任意异常后都强制清理会话。这个细节看起来小但在生产环境非常关键。4.5 常见问题速查表问题现象可能原因排查与解决输出乱码子进程输出非 UTF-8 编码在配置里声明encoding: gbk读取时用对应编码 decode命令找不到子进程 PATH 未继承适配器环境检查 Popen 的env或者把命令行写成绝对路径接口返回 500参数校验失败检查配置里的类型声明和必填字段是否完整进度条在接口里变成一堆碎片适配器按行切分输出对这类命令声明raw: true或者单独走 WebSocket 模式并发峰值时 CPU 突然很高大量 pty 文件描述符被浪费使用asyncio.Semaphore限制并发并设置合理的超时5. 一点心得与可能的扩展方向5.1 设计“中间表示层”的收益远超我最初的预期CLI-Anything 这个项目做下来我最大的体会是不直接拿命令和具体协议绑在一起而是先把“命令运行”抽象成一个统一的会话层表面上多了一层实际上让很多事情变得简单。新增一种接口协议比如后来我加的 MCP 模型上下文协议只需要对同一个CommandSession写一个新的适配器不用改动命令执行的核心逻辑。新增一个命令也只是写一段 YAML。这个设计让整个项目保持了一种很舒服的可扩展性维护起来也不累。5.2 实际动手的一些建议如果你想自己做一个类似的东西我建议不要一上来就追求支持所有协议。先把 REST 和 WebSocket 这两个跑通足够覆盖大多数需求了。配置中心化是必要的但没必要一开始设计得很复杂一个 YAML 文件就够了。等到命令数量超过二十个再考虑按目录拆分配置、加权限控制也不迟。调试 pty 相关问题时建议用script命令或者socat先模拟终端环境看看同样的命令在真实终端里是什么输出行为再回头看适配器代码。很多问题其实是“程序在管道里和在终端里的行为差异”导致的跟适配器本身关系不大。5.3 后续还能怎么玩CLI-Anything 后续的方向我可以想到几个一是接入聊天平台让用户在群里发一条命令机器人执行完再把结果贴回来二是接 MCP让大语言模型直接调用本地的命令行工具相当于给模型加了一双能操作真实系统的手三是做一个简单的 Web 管理界面把系统里的所有命令按权限分给不同团队成员使用。每一个方向都建立在同一个核心抽象上这也是我最兴奋的地方——一个简单的中间层能让存量命令行生态重新焕发生机。最后再分享一个小技巧如果你也想做类似的工具最重要的一件事是把“进程生命周期管理”和“外部接口”彻底分层。我在第一版就是没分层导致后来加 WebSocket 时改动的范围特别大。现在分层清楚之后加任何新功能都像插积木一样简单。这也是这个项目能一直迭代下去的根本原因。