简体中文

文件操作

这个示例模拟一个常见开发任务：创建一个小项目，找到代码中的 TODO，修复后读取修改结果；随后监听目录变化并处理错误响应。

本示例使用以下路由：

文件：/v2/fs/*
事件：/v2/watch

文件和事件：/v1/file/*

要求

GET /v2/sandbox 的 capabilities 中应包含 files。

GET /v1/capabilities 的返回结果中应包含 files。

文件 API 与命令、终端、代码执行共享同一个文件系统；隔离由容器负责，而不是由 API 负责。

大多数 /v2/fs/* 路由还接受可选的 ?user=，用于指定新建文件和目录的归属账户。

它不是执行身份，文件操作仍使用 daemon 的权限。字节流路由（download 和作为 GET 的 tree）以及 /v2/watch 不支持该参数。

创建、搜索和编辑

建一个目录，往里写两个文件，列出结果，先按文件名再按内容找到 TODO，替换它，再读取修改后的行：

共享 Aio helper（见约定）负责发请求并解析返回结构：

Python

TypeScript

BASE_URL = "http://127.0.0.1:18091"
sb = Aio(BASE_URL)

# 1. Scaffold: one directory, two files
sb.post("/v2/fs/mkdir", path="/tmp/demo/src", parents=True)
sb.post("/v2/fs/write", path="/tmp/demo/src/app.py", content=(
    "import sys\n\n"
    "def main():\n"
    "    # TODO: read the input path from argv\n"
    "    print('hello')\n\n"
    "if __name__ == '__main__':\n"
    "    main()\n"
))
sb.post("/v2/fs/write", path="/tmp/demo/README.md", content="# demo\n")
listed = sb.get("/v2/fs/list", path="/tmp/demo", recursive=True)
print([f["path"] for f in listed["files"]])

# 2. Locate the TODO: name glob, then content grep
found = sb.get("/v2/fs/search", path="/tmp/demo", pattern="**/*.py")
target = found["files"][0]
hit = sb.post("/v2/fs/grep", path=target, pattern="TODO.*")
for m in hit["matches"]:
    print(m["line_number"], m["line_content"])
# 4     # TODO: read the input path from argv

# 3. Edit with the editor tool; old_str matches literally
sb.post("/v2/fs/edit", command="str_replace", path=target,
        old_str="    # TODO: read the input path from argv\n    print('hello')",
        new_str="    path = sys.argv[1]\n    print(open(path).read())")

# 4. Read back the changed lines: 0-based, end exclusive
result = sb.get("/v2/fs/read", path=target, start_line=2, end_line=5)
print(result["content"])
# def main():
#     path = sys.argv[1]
#     print(open(path).read())

除了创建目录，其余步骤都有 1.x SDK 方法。创建目录这一步通过共享 Aio helper（见约定）调用：

Python

TypeScript

from agent_sandbox import Sandbox

BASE_URL = "http://127.0.0.1:18091"
client = Sandbox(base_url=BASE_URL)
sb = Aio(BASE_URL)

# 1. Scaffold: one directory, two files
# The 1.x SDK has no mkdir, so the helper posts the route itself.
sb.post("/v1/file/mkdir", path="/tmp/demo/src", parents=True)
client.file.write_file(file="/tmp/demo/src/app.py", content=(
    "import sys\n\n"
    "def main():\n"
    "    # TODO: read the input path from argv\n"
    "    print('hello')\n\n"
    "if __name__ == '__main__':\n"
    "    main()\n"
))
client.file.write_file(file="/tmp/demo/README.md", content="# demo\n")
listed = client.file.list_path(path="/tmp/demo", recursive=True).data
print([f.path for f in listed.files])

# 2. Locate the TODO: name glob, then content grep
found = client.file.glob_files(path="/tmp/demo", pattern="**/*.py").data
target = found.files[0].path
hit = client.file.grep_files(path=target, pattern="TODO.*").data
for m in hit.matches:
    print(m.line_number, m.line_content)
# 4     # TODO: read the input path from argv

# 3. Edit with the editor tool; old_str matches literally
client.file.str_replace_editor(
    command="str_replace", path=target,
    old_str="    # TODO: read the input path from argv\n    print('hello')",
    new_str="    path = sys.argv[1]\n    print(open(path).read())")

# 4. Read back the changed lines: 0-based, end exclusive
result = client.file.read_file(file=target, start_line=2, end_line=5).data
print(result.content)
# def main():
#     path = sys.argv[1]
#     print(open(path).read())

编辑器工具只保留这些操作：str_replace、insert 和 undo_edit。

view 返回 400，改用 GET /v2/fs/read。
create 返回 400，改用 POST /v2/fs/write。

view 和 create 在 v1 中仍然可用。

old_str 不存在时，str_replace 返回 400 old_str not found in file。output 是工具展示给模型的 cat -n 片段；old_content 和 new_content 是修改前后的完整文件。

按文件名搜索和按文件内容搜索使用不同的路由。

GET /v2/fs/search：合并 v1 的 glob 和 find，返回扁平的路径列表。
POST /v2/fs/grep：在 path 下搜索内容，每个匹配返回从 1 开始的 line_number，也可以请求 context_before 和 context_after。

POST /v1/file/glob：返回包含 path 和元数据的记录。
POST /v1/file/grep：在 path 下搜索内容，每个匹配返回从 1 开始的 line_number，也可以请求 context_before 和 context_after。
POST /v1/file/search：较早的单文件搜索路由，行号从 0 开始。

监听文件变化

watcher 从创建到停止之间记录路径下的变更。

轮询事件

一次 poll 最多长轮询 timeout 秒，并返回下一次调用用的 cursor：

Python

TypeScript

watcher = sb.post("/v2/watch", path="/tmp/demo", recursive=True)
watcher = watcher["watcher_id"]

def poll(watcher_id: str, **params) -> dict:
    return sb.get(f"/v2/watch/{watcher_id}/poll", **params)

sb.post("/v2/commands", command="echo 'X = 1' > /tmp/demo/src/util.py")

for e in poll(watcher, timeout=2)["events"]:
    print(e["seq"], e["type"], e["relative_path"])
# 1 create src/util.py
# 2 write src/util.py

sb.delete(f"/v2/watch/{watcher}")

Python

TypeScript

watcher = client.file.watch_create(path="/tmp/demo", recursive=True)
watcher = watcher["data"]["watcher_id"]

def poll(watcher_id: str, **params) -> dict:
    return client.file.watch_poll(watcher_id, **params)["data"]

client.bash.exec(command="echo 'X = 1' > /tmp/demo/src/util.py")

for e in poll(watcher, timeout=2)["events"]:
    print(e["seq"], e["type"], e["relative_path"])
# 1 create src/util.py
# 2 write src/util.py

client.file.watch_stop(watcher)

过滤暂存文件

上传不会直接写入目标文件，而是先写入目标旁边的 .aiod-upload-<id>.part，完成后再重命名覆盖目标。

指定归属身份时，写入也会先放在 .aiod-write-<id>.part 中；不指定归属身份时，则直接写入目标并保留原 inode。

下面创建两个监听器监听同一目录，其中一个过滤掉暂存文件：

Python

TypeScript

plain = sb.post("/v2/watch", path="/tmp/demo", recursive=True)
plain = plain["watcher_id"]
filtered = sb.post("/v2/watch", path="/tmp/demo", recursive=True,
                    exclude=["*.part"])["watcher_id"]

sb.http.post(f"{BASE_URL}/v2/fs/upload",
             files={"file": ("data.csv", b"a,b\n1,2\n")},
             data={"path": "/tmp/demo/data.csv"})

for name, w in (("plain", plain), ("filtered", filtered)):
    print(name)
    for e in poll(w, timeout=2)["events"]:
        print(" ", e["seq"], e["type"], e["relative_path"])

sb.delete(f"/v2/watch/{plain}")
sb.delete(f"/v2/watch/{filtered}")
# plain
#   1 create .aiod-upload-da85dbba-34b4-45cb-8364-0a72009b9cb3.part
#   2 rename .aiod-upload-da85dbba-34b4-45cb-8364-0a72009b9cb3.part
#   3 write .aiod-upload-da85dbba-34b4-45cb-8364-0a72009b9cb3.part
#   4 rename data.csv
# filtered
#   1 rename data.csv

Python

TypeScript

plain = client.file.watch_create(path="/tmp/demo", recursive=True)
plain = plain["data"]["watcher_id"]
filtered = client.file.watch_create(path="/tmp/demo", recursive=True,
                                    exclude=["*.part"])["data"]["watcher_id"]

client.file.upload_file(file=("data.csv", b"a,b\n1,2\n"),
                        path="/tmp/demo/data.csv")

for name, w in (("plain", plain), ("filtered", filtered)):
    print(name)
    for e in poll(w, timeout=2)["events"]:
        print(" ", e["seq"], e["type"], e["relative_path"])

client.file.watch_stop(plain)
client.file.watch_stop(filtered)
# plain
#   1 create .aiod-upload-da85dbba-34b4-45cb-8364-0a72009b9cb3.part
#   2 rename .aiod-upload-da85dbba-34b4-45cb-8364-0a72009b9cb3.part
#   3 write .aiod-upload-da85dbba-34b4-45cb-8364-0a72009b9cb3.part
#   4 rename data.csv
# filtered
#   1 rename data.csv

exclude 的 glob 匹配被监听根目录下的单个路径片段，因此 *.part 挡掉暂存文件，只留下落在目标上的 rename。传入的列表会替换默认排除项，而不是在其上追加。

等待下一次变更

把上一次的 cursor 传回去，下一次 poll 只返回它之后的事件。没有新事件时，会阻塞到 timeout 结束。

下面的循环用于等待变更：后台构建会先写报告，再写标记文件。

Python

TypeScript

watcher = sb.post("/v2/watch", path="/tmp/demo", recursive=True,
                   exclude=["*.part"])["watcher_id"]
sb.post("/v2/commands", mode="async", command=(
    "sleep 1; echo report > /tmp/demo/report.txt; touch /tmp/demo/build.done"))

cursor = 0
while True:
    page = poll(watcher, cursor=cursor, timeout=30)
    for e in page["events"]:
        print(e["seq"], e["type"], e["relative_path"])
    cursor = page["cursor"]
    if any(e["relative_path"] == "build.done" for e in page["events"]):
        break
# 1 create report.txt
# 2 write report.txt
# 3 create build.done

sb.delete(f"/v2/watch/{watcher}")

Python

TypeScript

watcher = client.file.watch_create(path="/tmp/demo", recursive=True,
                                   exclude=["*.part"])["data"]["watcher_id"]
client.bash.exec(async_mode=True, command=(
    "sleep 1; echo report > /tmp/demo/report.txt; touch /tmp/demo/build.done"))

cursor = 0
while True:
    page = poll(watcher, cursor=cursor, timeout=30)
    for e in page["events"]:
        print(e["seq"], e["type"], e["relative_path"])
    cursor = page["cursor"]
    if any(e["relative_path"] == "build.done" for e in page["events"]):
        break
# 1 create report.txt
# 2 write report.txt
# 3 create build.done

client.file.watch_stop(watcher)

以 SSE 推送事件

events 路由用 SSE 推送同样的事件。每条 file_change 的内容与 poll 相同，并带 <watcher_id>:<seq> 形式的 id。

因此，EventSource 重连时会从 Last-Event-ID 继续，而不是从头重放：

curl -N "$BASE_URL/v2/watch/$WATCHER/events"

curl -N "$BASE_URL/v1/file/watch/$WATCHER/events"

event: watch_started
data: {"watcher_id":"7fd196bd528d"}

id: 7fd196bd528d:1
event: file_change
data: {"seq":1,"type":"create","path":"/tmp/demo/src/util.py","relative_path":"src/util.py","is_dir":false,"timestamp":1788892055.64768,"old_path":null,"mtime":1788892055.6368937,"size":6,"inode":31524000}

传输文件

download 以流返回字节并支持 Range；upload 是 multipart，目标路径放在 path 字段：

curl "$BASE_URL/v2/fs/download?path=/tmp/demo/src/app.py" -o app.py
curl "$BASE_URL/v2/fs/download?path=/tmp/demo/src/app.py" -H "Range: bytes=0-9"   # 206

curl -F "file=@data.csv" -F "path=/tmp/demo/data.csv" "$BASE_URL/v2/fs/upload"

curl "$BASE_URL/v1/file/download?path=/tmp/demo/src/app.py" -o app.py
curl "$BASE_URL/v1/file/download?path=/tmp/demo/src/app.py" -H "Range: bytes=0-9"   # 206

curl -F "file=@data.csv" -F "path=/tmp/demo/data.csv" "$BASE_URL/v1/file/upload"

需要需要一次请求搬运整个目录时，切到 v2：

tree 把整个目录以 tar 包的形式流式返回，PUT 则把请求体里的 tar 包写回去：

curl "$BASE_URL/v2/fs/tree?path=/tmp/demo" -o demo.tar

错误处理

文件系统失败仍然使用统一的返回结构：success: false、错误消息，以及一个结构化的 data，其中有 errno、错误种类、操作、路径，以及重试是否有意义。

helper 遇到 success: false 会抛异常，所以要看失败的响应，得用 sb.http 而不是 sb.get/sb.post：

Python

TypeScript

reply = sb.http.get("/v2/fs/read", params={"path": "/tmp/demo/missing.txt"}).json()
print(reply["success"], reply["data"]["error_type"], reply["data"]["errno_name"])
# False not_found ENOENT

SDK 不抛异常，而是把返回结构交回来，所以失败的响应不需要特殊处理：

Python

TypeScript

reply = client.file.read_file(file="/tmp/demo/missing.txt")
print(reply.success, reply.data.error_type, reply.data.errno_name)
# False not_found ENOENT

两种写法在网络上收到的响应是同一个：

{
  "success": false,
  "message": "Failed to read file: No such file or directory (os error 2)",
  "data": {
    "errno": 2,
    "errno_name": "ENOENT",
    "error_type": "not_found",
    "exception_type": "FileNotFoundError",
    "message": "Failed to read file: No such file or directory (os error 2)",
    "operation": "read",
    "path": "/tmp/demo/missing.txt",
    "retryable": false
  },
  "hint": null
}

HTTP 状态码也能帮助判断错误种类：

状态码	`error_type`
`400`	`bad_request`、`invalid_path`、`invalid_target`
`403`	`permission_denied`
`404`	`not_found`
`409`	`already_exists`
`422`	`decode_error`
`507`	`no_space_left`

JSON body 格式错误返回 422，顶层带 errors 列表。查询参数格式错误（例如缺少必填参数或类型不对）返回 400 纯文本，而不是统一返回结构。

情况	状态码	看哪里
文件系统失败（路径不存在、权限不足……）	按种类分 `400`/`403`/`404`/`409`/`422`/`507`	`data.error_type`、`data.errno_name`、`data.retryable`
JSON body 格式错误（字段缺失或类型错误）	`422`	`errors[].location`
查询参数格式错误（必填参数缺失）	`400`，纯文本	不是返回结构
`str_replace` 的 `old_str` 不存在	`400`	`message`

无论哪种失败，HTTP 状态码都是 200。通过 success 和 data.error_type 判断错误类型。

JSON body 格式错误是例外：请求进入 handler 之前就会返回 422，顶层带 errors 列表。

情况	状态码	看哪里
文件系统失败（路径不存在、权限不足……）	`200`	`data.error_type`、`data.errno_name`、`data.retryable`
JSON body 格式错误（字段缺失或类型错误）	`422`	`errors[].location`
`str_replace` 的 `old_str` 不存在	`200`	`message`

先看 HTTP 状态码，再看 success，最后看 data.error_type。

#文件操作

#要求

#创建、搜索和编辑

#监听文件变化

#轮询事件

#过滤暂存文件

#等待下一次变更

#以 SSE 推送事件

#传输文件

#错误处理

#相关页面

文件操作

要求

创建、搜索和编辑

监听文件变化

轮询事件

过滤暂存文件

等待下一次变更

以 SSE 推送事件

传输文件

错误处理

相关页面