PyKAOS：AI Agent 的操作系统抽象层——本地与 SSH 远程文件操作和命令执行的统一接口

PyKAOS：AI Agent 的操作系统抽象层——本地与 SSH 远程文件操作和命令执行的统一接口

【免费下载链接】kimi-cliKimi Code CLI is your next CLI agent.项目地址: https://gitcode.com/GitHub_Trending/ki/kimi-cli

PyKAOS 是 kimi-cli 仓库（Kimi Code CLI）中独立发布的轻量级 Python 库（包名pykaos，模块名kaos），它为 AI Agent 提供了一层与操作系统交互的抽象接口：文件读写、目录遍历、路径处理和命令执行等操作，可以在一套统一 API 下自由地在本地环境与SSH 远程主机之间切换。阅读完本篇，你将掌握 PyKAOS 的Kaos协议、KaosPath路径抽象、LocalKaos/SSHKaos两个内置实现，以及基于 contextvars 的"当前后端"切换机制，并能在自己的 Agent 代码中写出"一套代码、本地/远程皆可运行"的文件与命令操作逻辑。

PyKAOS 是什么

根据 packages/kaos/README.md 的定位，PyKAOS 是：

A lightweight Python library providing an abstraction layer for agents to interact with operating systems. File operations and command executions via KAOS can be easily switched between local environment and remote systems over SSH.

核心价值可以拆成三点：

1. 面向 Agent：接口设计以 Agent 工具（Shell、ReadFile、WriteFile 等）的调用方式为出发点，全部为异步 API，便于融入事件循环驱动的 Agent 运行时；
2. 统一抽象：文件操作与命令执行都通过Kaos协议接口表达，调用方不关心底层是本地进程还是远程 SSH 会话；
3. 可切换后端：通过 contextvars 将"当前 KAOS 实例"绑定到当前任务上下文，运行时可以在本地实现与 SSH 实现之间灵活切换。

从仓库结构看，PyKAOS 位于 packages/kaos/ 目录，由src/kaos/下的 5 个核心模块组成：

模块	职责
init.py	Kaos协议、KaosProcess、AsyncReadable/AsyncWritable、StatResult及全部模块级便捷函数
local.py	LocalKaos：直接操作本地文件系统的默认实现
ssh.py	SSHKaos：通过 asyncssh 的 SSH/SFTP 与远程主机交互的实现
path.py	KaosPath：跨后端统一的路径抽象
_current.py	基于contextvars.ContextVar的当前 KAOS 实例管理

安装与依赖

PyKAOS 作为一个独立的 Python 包发布，元数据见 packages/kaos/pyproject.toml：

• 包名：pykaos，当前版本0.9.0；
• Python 版本要求：>=3.12；
• 运行时依赖：aiofiles>=24.0,<26.0（本地异步文件 IO）、asyncssh==2.21.1（SSH/SFTP 客户端）。

开发依赖包括pytest、pytest-asyncio、ruff、pyright（严格类型检查模式）、inline-snapshot等。安装可直接使用 uv 或 pip：

uv pip install pykaos # 或 pip install pykaos

核心架构：Kaos 协议与统一接口

Kaos 协议

Kaos是一个@runtime_checkable的Protocol（见init.py），它定义了所有后端实现必须满足的接口契约。协议包含两个层面的能力：

路径与目录操作

方法	说明
pathclass() -> type[PurePath]	返回该后端使用的路径类（本地为PurePosixPath/PureWindowsPath，SSH 为PurePosixPath）
normpath(path) -> KaosPath	规范化路径，消除双斜杠等冗余
gethome() -> KaosPath	获取 home 目录
getcwd() -> KaosPath	获取当前工作目录
chdir(path)	切换当前工作目录
stat(path, follow_symlinks=True) -> StatResult	获取路径的 stat 信息
iterdir(path)	异步迭代目录下的条目
glob(path, pattern, case_sensitive=True)	按模式匹配目录下的文件/子目录

文件与命令操作

方法	说明
readbytes(path, n=None)	读取整个文件为 bytes，或只读取前 n 字节
readtext(path, encoding="utf-8", errors="strict")	按文本读取整个文件
readlines(path, ...)	逐行迭代文件内容
writebytes(path, data)	写入二进制数据
writetext(path, data, mode="w", ...)	写入文本（mode支持"w"与"a"，返回写入字符数）
mkdir(path, parents=False, exist_ok=False)	创建目录
exec(*args, env=None) -> KaosProcess	执行命令，env可传入子进程环境变量（不传则继承父进程环境）

协议上方的AsyncReadable/AsyncWritable协议描述了异步字节流接口（read/readline/readuntil/write/drain/close等），统一了asyncio.StreamReader/StreamWriter与asyncssh的SSHReader/SSHWriter两种流类型（见init.py），这正是KaosProcess能同时包装本地子进程与远程进程的基础。

KaosProcess：统一的进程接口

KaosProcess协议（init.py）暴露了stdin/stdout/stderr三个异步流、pid、returncode属性以及wait()、kill()两个协程方法：

• wait()返回进程退出码；
• pid在本地实现中返回真实进程号，SSH 实现因asyncssh.SSHClientProcess不暴露 pid 而固定返回-1（见 ssh.py）；
• LocalKaos的Process直接包装asyncio.subprocess.Process（local.py），创建时要求 stdin/stdout/stderr 均为管道，否则抛出ValueError。

值得注意的是，SSHKaos.Process.wait()刻意使用wait_closed()而非原生wait()，并在注释中说明原因：原生wait()会通过communicate()排空 stdout/stderr 内部接收缓冲区，导致 wait 之后流不可读；使用wait_closed()可以保持与LocalKaos一致的行为——wait 之后仍能读到输出（ssh.py）。

StatResult

stat返回的StatResult是一个@dataclass（init.py），字段对齐os.stat_result：st_mode、st_ino、st_dev、st_nlink、st_uid、st_gid、st_size、st_atime、st_mtime、st_ctime。SSH 后端通过 SFTP 属性构造该结果（见下文），其中st_ino/st_dev因 SFTP 协议不支持而固定为 0。

KaosPath：跨后端的路径抽象

KaosPath（path.py）是 PyKAOS 的路径门面，内部委托给当前后端返回的PurePath子类（本地是PurePosixPath或PureWindowsPath，SSH 始终是PurePosixPath）。它提供的方法可分为四组：

构造与转换

• KaosPath(*args)：按当前后端路径语义构造；
• KaosPath.home()/KaosPath.cwd()：类方法，等价于kaos.gethome()/kaos.getcwd()；
• unsafe_from_local_path(Path)/unsafe_to_local_path()：仅在确认使用LocalKaos时才可调用的本地路径互转（方法名的unsafe前缀即为此警告）；
• canonical()：将路径转为绝对并解析掉./..，与pathlib.Path.resolve()不同，它不解析符号链接（path.py）。

路径运算

• name、parent、is_absolute()、joinpath()、/运算符、relative_to()、expanduser()（展开~为后端 home 目录），并实现了完整的比较运算符（</<=/>/>=/==），因此可直接用于排序与集合运算。

文件操作（异步方法）

• read_bytes(n=None)、read_text()、read_lines()；
• write_bytes(data)、write_text(data)、append_text(data)；
• stat()、exists()、is_file()、is_dir()（后三者通过捕获OSError判断，并基于st_mode的S_ISREG/S_ISDIR位判定）；
• mkdir(parents, exist_ok)。

目录遍历

• iterdir()：返回目录的直接子项；
• glob(pattern, case_sensitive=True)：模式匹配，本地实现基于pathlib.Path.glob（支持**递归匹配，且*会匹配隐藏文件），SSH 实现基于 SFTP 的 glob。

一个典型的读写示例：

import kaos from kaos.path import KaosPath p = KaosPath.home() / "notes" / "todo.txt" await p.mkdir(parents=True, exist_ok=True) await p.write_text("buy milk\n") await p.append_text("call doctor\n") async for line in p.read_lines(): print(line) # buy milk # call doctor

这段代码在本地与 SSH 后端下语义一致，无需任何改动。

LocalKaos：本地后端

LocalKaos（local.py）是直接操作本地文件系统的实现，模块底部导出的local_kaos = LocalKaos()单例同时是 contextvar 的默认值（见 _current.py），即默认情况下所有kaos.*调用都落到本地。

它的实现细节包括：

• 路径类自适应：在 Windows 上使用ntpath+PureWindowsPath，其他平台使用posixpath+PurePosixPath（local.py）；
• 异步文件 IO：全部基于aiofiles，stat使用aiofiles.os.stat；Windows 上st_ctime取st_birthtime作为替代（local.py）；
• 保持行尾原样：writetext以newline=""打开文件，禁用 Python 写入时的通用换行转换，保证 LF 与 CRLF 都按原文落盘——这正是 CHANGELOG.md 中 0.8.0 版本修复的"Windows 上 writetext 把 LF 转成 CRLF"问题，并由 test_local_kaos.py 中的二进制回读测试锁定；
• 阻塞调用隔离：glob与mkdir等可能阻塞的操作通过asyncio.to_thread放到线程池执行（[local.py](https://link.gitcode.com/i/5789d5787ddd88e614f6f151b41103de#L101-L109, L159-L163)）；
• 命令执行：exec使用asyncio.create_subprocess_exec，三个标准流均以管道方式创建（local.py）。

SSHKaos：远程后端

SSHKaos（ssh.py）通过 asyncssh 建立 SSH 连接与 SFTP 会话，让同一套Kaos接口操作远程主机。

建立连接

使用类方法SSHKaos.create()异步创建（ssh.py）：

from kaos.ssh import SSHKaos ssh = await SSHKaos.create( host="192.168.1.10", port=22, username="agent", password="******", # 密码与密钥二选一或同时提供 key_paths=["~/.ssh/id_ed25519"], key_contents=["-----BEGIN OPENSSH PRIVATE KEY-----..."], # 密钥内容直传 cwd="/home/agent/workspace", # 初始工作目录 )

关键参数与内部行为：

参数	默认值	说明
host	必填	远程主机地址
port	22	SSH 端口
username	None	登录用户名
password	None	密码认证
key_paths	None	私钥文件路径列表
key_contents	None	私钥内容字符串列表（会经asyncssh.import_private_key解析）
cwd	None	初始工作目录，缺省为远端 home

创建流程内部固定做了三件事：encoding=None保证字节级读写、known_hosts=None跳过主机密钥信任校验（避免 "Host key is not trusted" 报错）、随后启动 SFTP 客户端并通过realpath(".")确定 home 与 cwd。连接用完后必须调用await ssh.unsafe_close()关闭（ssh.py）。

SSH 后端的实现取舍

• 路径：恒为PurePosixPath，normpath使用posixpath.normpath；
• stat：把 SFTP 属性中的类型（regular/directory/symlink/socket 等）映射为stat模块的S_IF*位，并与权限位合并构造st_mode（_build_st_mode，ssh.py）；纳秒时间戳会被折算进浮点秒（_sec_with_nanos）；
• iterdir：基于sftp.listdir并过滤.与..（ssh.py）；
• glob：不支持大小写不敏感匹配，传入case_sensitive=False会直接抛ValueError（ssh.py）；
• readlines：SFTP 文件对象不支持逐行迭代，因此通过readtext+splitlines实现（ssh.py）；
• mkdir：parents=True时走sftp.makedirs；否则先检查存在性，目录已存在且exist_ok=False时抛FileExistsError（ssh.py）；
• exec：命令通过shlex.quote逐参数转义后拼接，并显式加上cd <cwd> &&前缀——原因是 SFTP 的工作目录概念不影响 SSH exec，为了让 exec 与其他后端行为一致，必须显式进入跟踪中的 cwd（ssh.py）。这条规则是刻意严格的：若 cwd 不存在，命令直接失败。

测试与验证

SSH 后端的集成测试位于 test_ssh_kaos.py，测试通过环境变量注入连接参数，未配置有效 SSH 环境时自动跳过：

• KAOS_SSH_HOST（默认127.0.0.1）、KAOS_SSH_PORT（默认22）、KAOS_SSH_USERNAME、KAOS_SSH_PASSWORD；
• KAOS_SSH_KEY_PATHS（逗号分隔）与KAOS_SSH_KEY_CONTENTS（|||分隔）。

测试覆盖了chdir与真实路径同步、exec尊重 cwd、mkdir的exist_ok语义、stat 的目录/文件类型判定、KaosPath读写往返、iterdir 过滤、glob 大小写敏感、stdout/stderr 分流、空命令报错以及kill后returncode更新等场景。

当前后端切换：contextvars 机制

PyKAOS 的"切换后端"能力建立在 _current.py 的contextvars.ContextVar之上：

current_kaos = ContextVarKaos

contextvar 是任务级（task-local）的：在一个 asyncio 任务里设置后，仅影响该任务及其子任务，天然适合并发场景下"每个 Agent 会话绑定不同后端"的需求。__init__.py暴露了三个管理函数：

import kaos from kaos.local import LocalKaos from kaos.ssh import SSHKaos # 绑定远程后端（返回 token，用于恢复） ssh = await SSHKaos.create(host="my-server") token = kaos.set_current_kaos(ssh) try: # 此范围内所有 kaos.* 调用都走 SSH print(await kaos.readtext("/etc/hostname")) finally: kaos.reset_current_kaos(token) # 恢复之前的后端 await ssh.unsafe_close()

同时，__init__.py为Kaos协议的每个方法都提供了同名模块级函数（kaos.readtext、kaos.exec、kaos.glob……），它们统一通过get_current_kaos()委托给当前实例（init.py）。这也解释了KaosPath为何能透明地跟随后端切换——它调用的正是这些模块级函数。

一个端到端的"本地/远程切换"示例：

import kaos from kaos.ssh import SSHKaos async def collect_logs(use_ssh: bool): kaos_impl = await SSHKaos.create(host="prod-01") if use_ssh else None token = kaos.set_current_kaos(kaos_impl) if kaos_impl else None try: proc = await kaos.exec("sh", "-c", "cat /var/log/app.log | tail -20") out, _ = await asyncio.gather(proc.stdout.read(), proc.stderr.read()) code = await proc.wait() return code, out.decode() finally: if token is not None: kaos.reset_current_kaos(token) if kaos_impl is not None: await kaos_impl.unsafe_close()

命令执行的三种形态

kaos.exec是 Agent 运行命令的统一入口，仓库测试分别验证了三种典型调用形态：

1. 直接参数形式（test_local_kaos.py）

process = await kaos.exec(sys.executable, "-c", "print('hello')") stdout, stderr = await asyncio.gather(process.stdout.read(), process.stderr.read()) assert await process.wait() == 0

2. POSIX shell 形态（test_local_kaos_sh.py）——通过/bin/sh -c执行管道、条件、环境变量、命令替换等复合命令，测试覆盖&&、;、||、管道、stdin 输入、超时 kill 等场景；此文件在 Windows 上跳过。

3. cmd.exe 形态（test_local_kaos_cmd.py）——通过cmd.exe /c执行，并先执行chcp 65001保证 UTF-8 输出；此文件仅在 Windows 上运行。

行为契约（三个文件共同确认）：

• wait()之前或之后读取 stdout/stderr 均可得到完整输出；
• 非零退出码能通过wait()获取（如sys.exit(7)返回 7）；
• 运行中的进程可被kill()，kill 后wait()返回非零；
• exec()不接受空命令，至少需要一个参数，否则抛ValueError。

在 kimi-cli 中的生态位置：ACPKaos

PyKAOS 并非孤立存在。仓库中的 klip-2-acpkaos.md（状态：Implemented）记录了 ACPKaos 的设计：它作为LocalKaos的近亲变体，把exec、readtext、writetext等少数操作重定向到 ACP（Agent Client Protocol）客户端，让 Zed 等 ACP 客户端能够观察到 Agent 的文件编辑与命令执行，其余操作全部透传给本地实现。该 KLIP 明确指出"KAOS 已经抽象了操作系统操作，ACP 天然适合作为 KAOS 的一个后端"。

实际实现位于 src/kimi_cli/acp/kaos.py：其中的ACPProcess实现了KaosProcess协议，spawn时调用 ACP 的create_terminal创建终端，后台轮询terminal_output增量刷新输出，wait()并发等待退出状态与输出，并在结束时确保terminal/release；由于 ACP 不区分 stderr，stderr 被保持为空流（src/kimi_cli/acp/kaos.py）。这从侧面印证了 PyKAOS 协议设计的前瞻性：只要满足Kaos协议，第三方实现即可无缝接入现有工具链。

版本演进脉络

packages/kaos/CHANGELOG.md 记录了库的演进历程，可以帮你理解当前 API 形态的来由：

• 0.2.0：初始版本，提供Kaos协议、LocalKaos与KaosPath；
• 0.3.0：iterdir/glob/read_lines改为同步函数返回异步迭代器；
• 0.4.0：新增Kaos.exec命令执行能力；
• 0.5.0：KaosProcess移入Kaos.Process，新增AsyncReadable/AsyncWritable协议与SSHKaos，Python 版本要求降至 3.12；
• 0.6.0：readbytes支持n参数读取前 n 字节；
• 0.7.0：exec支持env参数传递子进程环境变量；
• 0.8.0：修复 Windows 上writetext的 LF→CRLF 转换；
• 0.9.0：补充隐藏文件 glob 行为的测试。

小结

PyKAOS 用不到千行的核心源码，为 AI Agent 提供了一套足够简洁、可扩展的操作系统抽象：Kaos协议定义契约，KaosPath抹平路径差异，LocalKaos与SSHKaos提供开箱即用的两端实现，contextvars 机制让运行时切换后端只需一行代码。对于正在构建 Agent 工具链的开发者，无论目标是本地沙箱还是远程执行环境，这套模式都值得直接借鉴或复用。

【免费下载链接】kimi-cliKimi Code CLI is your next CLI agent.项目地址: https://gitcode.com/GitHub_Trending/ki/kimi-cli