PyKAOSAI Agent 的操作系统抽象层——本地与 SSH 远程文件操作和命令执行的统一接口【免费下载链接】kimi-cliKimi Code CLI is your next CLI agent.项目地址: https://gitcode.com/GitHub_Trending/ki/kimi-cliPyKAOS 是 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.核心价值可以拆成三点面向 Agent接口设计以 Agent 工具Shell、ReadFile、WriteFile 等的调用方式为出发点全部为异步 API便于融入事件循环驱动的 Agent 运行时统一抽象文件操作与命令执行都通过Kaos协议接口表达调用方不关心底层是本地进程还是远程 SSH 会话可切换后端通过 contextvars 将当前 KAOS 实例绑定到当前任务上下文运行时可以在本地实现与 SSH 实现之间灵活切换。从仓库结构看PyKAOS 位于 packages/kaos/ 目录由src/kaos/下的 5 个核心模块组成模块职责init.pyKaos协议、KaosProcess、AsyncReadable/AsyncWritable、StatResult及全部模块级便捷函数local.pyLocalKaos直接操作本地文件系统的默认实现ssh.pySSHKaos通过 asyncssh 的 SSH/SFTP 与远程主机交互的实现path.pyKaosPath跨后端统一的路径抽象_current.py基于contextvars.ContextVar的当前 KAOS 实例管理安装与依赖PyKAOS 作为一个独立的 Python 包发布元数据见 packages/kaos/pyproject.toml包名pykaos当前版本0.9.0Python 版本要求3.12运行时依赖aiofiles24.0,26.0本地异步文件 IO、asyncssh2.21.1SSH/SFTP 客户端。开发依赖包括pytest、pytest-asyncio、ruff、pyright严格类型检查模式、inline-snapshot等。安装可直接使用 uv 或 pipuv pip install pykaos # 或 pip install pykaos核心架构Kaos 协议与统一接口Kaos 协议Kaos是一个runtime_checkable的Protocol见init.py它定义了所有后端实现必须满足的接口契约。协议包含两个层面的能力路径与目录操作方法说明pathclass() - type[PurePath]返回该后端使用的路径类本地为PurePosixPath/PureWindowsPathSSH 为PurePosixPathnormpath(path) - KaosPath规范化路径消除双斜杠等冗余gethome() - KaosPath获取 home 目录getcwd() - KaosPath获取当前工作目录chdir(path)切换当前工作目录stat(path, follow_symlinksTrue) - StatResult获取路径的 stat 信息iterdir(path)异步迭代目录下的条目glob(path, pattern, case_sensitiveTrue)按模式匹配目录下的文件/子目录文件与命令操作方法说明readbytes(path, nNone)读取整个文件为 bytes或只读取前 n 字节readtext(path, encodingutf-8, errorsstrict)按文本读取整个文件readlines(path, ...)逐行迭代文件内容writebytes(path, data)写入二进制数据writetext(path, data, modew, ...)写入文本mode支持w与a返回写入字符数mkdir(path, parentsFalse, exist_okFalse)创建目录exec(*args, envNone) - 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.pyLocalKaos的Process直接包装asyncio.subprocess.Processlocal.py创建时要求 stdin/stdout/stderr 均为管道否则抛出ValueError。值得注意的是SSHKaos.Process.wait()刻意使用wait_closed()而非原生wait()并在注释中说明原因原生wait()会通过communicate()排空 stdout/stderr 内部接收缓冲区导致 wait 之后流不可读使用wait_closed()可以保持与LocalKaos一致的行为——wait 之后仍能读到输出ssh.py。StatResultstat返回的StatResult是一个dataclassinit.py字段对齐os.stat_resultst_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跨后端的路径抽象KaosPathpath.py是 PyKAOS 的路径门面内部委托给当前后端返回的PurePath子类本地是PurePosixPath或PureWindowsPathSSH 始终是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(nNone)、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_sensitiveTrue)模式匹配本地实现基于pathlib.Path.glob支持**递归匹配且*会匹配隐藏文件SSH 实现基于 SFTP 的 glob。一个典型的读写示例import kaos from kaos.path import KaosPath p KaosPath.home() / notes / todo.txt await p.mkdir(parentsTrue, exist_okTrue) 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本地后端LocalKaoslocal.py是直接操作本地文件系统的实现模块底部导出的local_kaos LocalKaos()单例同时是 contextvar 的默认值见 _current.py即默认情况下所有kaos.*调用都落到本地。它的实现细节包括路径类自适应在 Windows 上使用ntpathPureWindowsPath其他平台使用posixpathPurePosixPathlocal.py异步文件 IO全部基于aiofilesstat使用aiofiles.os.statWindows 上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远程后端SSHKaosssh.py通过 asyncssh 建立 SSH 连接与 SFTP 会话让同一套Kaos接口操作远程主机。建立连接使用类方法SSHKaos.create()异步创建ssh.pyfrom kaos.ssh import SSHKaos ssh await SSHKaos.create( host192.168.1.10, port22, usernameagent, password******, # 密码与密钥二选一或同时提供 key_paths[~/.ssh/id_ed25519], key_contents[-----BEGIN OPENSSH PRIVATE KEY-----...], # 密钥内容直传 cwd/home/agent/workspace, # 初始工作目录 )关键参数与内部行为参数默认值说明host必填远程主机地址port22SSH 端口usernameNone登录用户名passwordNone密码认证key_pathsNone私钥文件路径列表key_contentsNone私钥内容字符串列表会经asyncssh.import_private_key解析cwdNone初始工作目录缺省为远端 home创建流程内部固定做了三件事encodingNone保证字节级读写、known_hostsNone跳过主机密钥信任校验避免 Host key is not trusted 报错、随后启动 SFTP 客户端并通过realpath(.)确定 home 与 cwd。连接用完后必须调用await ssh.unsafe_close()关闭ssh.py。SSH 后端的实现取舍路径恒为PurePosixPathnormpath使用posixpath.normpathstat把 SFTP 属性中的类型regular/directory/symlink/socket 等映射为stat模块的S_IF*位并与权限位合并构造st_mode_build_st_modessh.py纳秒时间戳会被折算进浮点秒_sec_with_nanositerdir基于sftp.listdir并过滤.与..ssh.pyglob不支持大小写不敏感匹配传入case_sensitiveFalse会直接抛ValueErrorssh.pyreadlinesSFTP 文件对象不支持逐行迭代因此通过readtextsplitlines实现ssh.pymkdirparentsTrue时走sftp.makedirs否则先检查存在性目录已存在且exist_okFalse时抛FileExistsErrorssh.pyexec命令通过shlex.quote逐参数转义后拼接并显式加上cd cwd 前缀——原因是 SFTP 的工作目录概念不影响 SSH exec为了让 exec 与其他后端行为一致必须显式进入跟踪中的 cwdssh.py。这条规则是刻意严格的若 cwd 不存在命令直接失败。测试与验证SSH 后端的集成测试位于 test_ssh_kaos.py测试通过环境变量注入连接参数未配置有效 SSH 环境时自动跳过KAOS_SSH_HOST默认127.0.0.1、KAOS_SSH_PORT默认22、KAOS_SSH_USERNAME、KAOS_SSH_PASSWORDKAOS_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 ContextVarKaoscontextvar 是任务级task-local的在一个 asyncio 任务里设置后仅影响该任务及其子任务天然适合并发场景下每个 Agent 会话绑定不同后端的需求。__init__.py暴露了三个管理函数import kaos from kaos.local import LocalKaos from kaos.ssh import SSHKaos # 绑定远程后端返回 token用于恢复 ssh await SSHKaos.create(hostmy-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(hostprod-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.pyprocess await kaos.exec(sys.executable, -c, print(hello)) stdout, stderr await asyncio.gather(process.stdout.read(), process.stderr.read()) assert await process.wait() 02. 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 中的生态位置ACPKaosPyKAOS 并非孤立存在。仓库中的 klip-2-acpkaos.md状态Implemented记录了 ACPKaos 的设计它作为LocalKaos的近亲变体把exec、readtext、writetext等少数操作重定向到 ACPAgent 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 不区分 stderrstderr 被保持为空流src/kimi_cli/acp/kaos.py。这从侧面印证了 PyKAOS 协议设计的前瞻性只要满足Kaos协议第三方实现即可无缝接入现有工具链。版本演进脉络packages/kaos/CHANGELOG.md 记录了库的演进历程可以帮你理解当前 API 形态的来由0.2.0初始版本提供Kaos协议、LocalKaos与KaosPath0.3.0iterdir/glob/read_lines改为同步函数返回异步迭代器0.4.0新增Kaos.exec命令执行能力0.5.0KaosProcess移入Kaos.Process新增AsyncReadable/AsyncWritable协议与SSHKaosPython 版本要求降至 3.120.6.0readbytes支持n参数读取前 n 字节0.7.0exec支持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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
PyKAOS：AI Agent 的操作系统抽象层——本地与 SSH 远程文件操作和命令执行的统一接口
PyKAOSAI Agent 的操作系统抽象层——本地与 SSH 远程文件操作和命令执行的统一接口【免费下载链接】kimi-cliKimi Code CLI is your next CLI agent.项目地址: https://gitcode.com/GitHub_Trending/ki/kimi-cliPyKAOS 是 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.核心价值可以拆成三点面向 Agent接口设计以 Agent 工具Shell、ReadFile、WriteFile 等的调用方式为出发点全部为异步 API便于融入事件循环驱动的 Agent 运行时统一抽象文件操作与命令执行都通过Kaos协议接口表达调用方不关心底层是本地进程还是远程 SSH 会话可切换后端通过 contextvars 将当前 KAOS 实例绑定到当前任务上下文运行时可以在本地实现与 SSH 实现之间灵活切换。从仓库结构看PyKAOS 位于 packages/kaos/ 目录由src/kaos/下的 5 个核心模块组成模块职责init.pyKaos协议、KaosProcess、AsyncReadable/AsyncWritable、StatResult及全部模块级便捷函数local.pyLocalKaos直接操作本地文件系统的默认实现ssh.pySSHKaos通过 asyncssh 的 SSH/SFTP 与远程主机交互的实现path.pyKaosPath跨后端统一的路径抽象_current.py基于contextvars.ContextVar的当前 KAOS 实例管理安装与依赖PyKAOS 作为一个独立的 Python 包发布元数据见 packages/kaos/pyproject.toml包名pykaos当前版本0.9.0Python 版本要求3.12运行时依赖aiofiles24.0,26.0本地异步文件 IO、asyncssh2.21.1SSH/SFTP 客户端。开发依赖包括pytest、pytest-asyncio、ruff、pyright严格类型检查模式、inline-snapshot等。安装可直接使用 uv 或 pipuv pip install pykaos # 或 pip install pykaos核心架构Kaos 协议与统一接口Kaos 协议Kaos是一个runtime_checkable的Protocol见init.py它定义了所有后端实现必须满足的接口契约。协议包含两个层面的能力路径与目录操作方法说明pathclass() - type[PurePath]返回该后端使用的路径类本地为PurePosixPath/PureWindowsPathSSH 为PurePosixPathnormpath(path) - KaosPath规范化路径消除双斜杠等冗余gethome() - KaosPath获取 home 目录getcwd() - KaosPath获取当前工作目录chdir(path)切换当前工作目录stat(path, follow_symlinksTrue) - StatResult获取路径的 stat 信息iterdir(path)异步迭代目录下的条目glob(path, pattern, case_sensitiveTrue)按模式匹配目录下的文件/子目录文件与命令操作方法说明readbytes(path, nNone)读取整个文件为 bytes或只读取前 n 字节readtext(path, encodingutf-8, errorsstrict)按文本读取整个文件readlines(path, ...)逐行迭代文件内容writebytes(path, data)写入二进制数据writetext(path, data, modew, ...)写入文本mode支持w与a返回写入字符数mkdir(path, parentsFalse, exist_okFalse)创建目录exec(*args, envNone) - 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.pyLocalKaos的Process直接包装asyncio.subprocess.Processlocal.py创建时要求 stdin/stdout/stderr 均为管道否则抛出ValueError。值得注意的是SSHKaos.Process.wait()刻意使用wait_closed()而非原生wait()并在注释中说明原因原生wait()会通过communicate()排空 stdout/stderr 内部接收缓冲区导致 wait 之后流不可读使用wait_closed()可以保持与LocalKaos一致的行为——wait 之后仍能读到输出ssh.py。StatResultstat返回的StatResult是一个dataclassinit.py字段对齐os.stat_resultst_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跨后端的路径抽象KaosPathpath.py是 PyKAOS 的路径门面内部委托给当前后端返回的PurePath子类本地是PurePosixPath或PureWindowsPathSSH 始终是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(nNone)、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_sensitiveTrue)模式匹配本地实现基于pathlib.Path.glob支持**递归匹配且*会匹配隐藏文件SSH 实现基于 SFTP 的 glob。一个典型的读写示例import kaos from kaos.path import KaosPath p KaosPath.home() / notes / todo.txt await p.mkdir(parentsTrue, exist_okTrue) 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本地后端LocalKaoslocal.py是直接操作本地文件系统的实现模块底部导出的local_kaos LocalKaos()单例同时是 contextvar 的默认值见 _current.py即默认情况下所有kaos.*调用都落到本地。它的实现细节包括路径类自适应在 Windows 上使用ntpathPureWindowsPath其他平台使用posixpathPurePosixPathlocal.py异步文件 IO全部基于aiofilesstat使用aiofiles.os.statWindows 上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远程后端SSHKaosssh.py通过 asyncssh 建立 SSH 连接与 SFTP 会话让同一套Kaos接口操作远程主机。建立连接使用类方法SSHKaos.create()异步创建ssh.pyfrom kaos.ssh import SSHKaos ssh await SSHKaos.create( host192.168.1.10, port22, usernameagent, password******, # 密码与密钥二选一或同时提供 key_paths[~/.ssh/id_ed25519], key_contents[-----BEGIN OPENSSH PRIVATE KEY-----...], # 密钥内容直传 cwd/home/agent/workspace, # 初始工作目录 )关键参数与内部行为参数默认值说明host必填远程主机地址port22SSH 端口usernameNone登录用户名passwordNone密码认证key_pathsNone私钥文件路径列表key_contentsNone私钥内容字符串列表会经asyncssh.import_private_key解析cwdNone初始工作目录缺省为远端 home创建流程内部固定做了三件事encodingNone保证字节级读写、known_hostsNone跳过主机密钥信任校验避免 Host key is not trusted 报错、随后启动 SFTP 客户端并通过realpath(.)确定 home 与 cwd。连接用完后必须调用await ssh.unsafe_close()关闭ssh.py。SSH 后端的实现取舍路径恒为PurePosixPathnormpath使用posixpath.normpathstat把 SFTP 属性中的类型regular/directory/symlink/socket 等映射为stat模块的S_IF*位并与权限位合并构造st_mode_build_st_modessh.py纳秒时间戳会被折算进浮点秒_sec_with_nanositerdir基于sftp.listdir并过滤.与..ssh.pyglob不支持大小写不敏感匹配传入case_sensitiveFalse会直接抛ValueErrorssh.pyreadlinesSFTP 文件对象不支持逐行迭代因此通过readtextsplitlines实现ssh.pymkdirparentsTrue时走sftp.makedirs否则先检查存在性目录已存在且exist_okFalse时抛FileExistsErrorssh.pyexec命令通过shlex.quote逐参数转义后拼接并显式加上cd cwd 前缀——原因是 SFTP 的工作目录概念不影响 SSH exec为了让 exec 与其他后端行为一致必须显式进入跟踪中的 cwdssh.py。这条规则是刻意严格的若 cwd 不存在命令直接失败。测试与验证SSH 后端的集成测试位于 test_ssh_kaos.py测试通过环境变量注入连接参数未配置有效 SSH 环境时自动跳过KAOS_SSH_HOST默认127.0.0.1、KAOS_SSH_PORT默认22、KAOS_SSH_USERNAME、KAOS_SSH_PASSWORDKAOS_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 ContextVarKaoscontextvar 是任务级task-local的在一个 asyncio 任务里设置后仅影响该任务及其子任务天然适合并发场景下每个 Agent 会话绑定不同后端的需求。__init__.py暴露了三个管理函数import kaos from kaos.local import LocalKaos from kaos.ssh import SSHKaos # 绑定远程后端返回 token用于恢复 ssh await SSHKaos.create(hostmy-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(hostprod-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.pyprocess await kaos.exec(sys.executable, -c, print(hello)) stdout, stderr await asyncio.gather(process.stdout.read(), process.stderr.read()) assert await process.wait() 02. 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 中的生态位置ACPKaosPyKAOS 并非孤立存在。仓库中的 klip-2-acpkaos.md状态Implemented记录了 ACPKaos 的设计它作为LocalKaos的近亲变体把exec、readtext、writetext等少数操作重定向到 ACPAgent 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 不区分 stderrstderr 被保持为空流src/kimi_cli/acp/kaos.py。这从侧面印证了 PyKAOS 协议设计的前瞻性只要满足Kaos协议第三方实现即可无缝接入现有工具链。版本演进脉络packages/kaos/CHANGELOG.md 记录了库的演进历程可以帮你理解当前 API 形态的来由0.2.0初始版本提供Kaos协议、LocalKaos与KaosPath0.3.0iterdir/glob/read_lines改为同步函数返回异步迭代器0.4.0新增Kaos.exec命令执行能力0.5.0KaosProcess移入Kaos.Process新增AsyncReadable/AsyncWritable协议与SSHKaosPython 版本要求降至 3.120.6.0readbytes支持n参数读取前 n 字节0.7.0exec支持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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考