#文件操作
这个示例模拟一个常见开发任务:创建一个小项目,找到代码中的 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(见 约定)负责发请求并解析返回结构:
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())const BASE_URL = "http://127.0.0.1:18091";
const sb = new Aio(BASE_URL);
// 1. Scaffold: one directory, two files
await sb.post("/v2/fs/mkdir", { path: "/tmp/demo/src", parents: true });
await 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",
});
await sb.post("/v2/fs/write", { path: "/tmp/demo/README.md", content: "# demo\n" });
const listed = await sb.get<any>(
`/v2/fs/list?path=${encodeURIComponent("/tmp/demo")}&recursive=true`
);
console.log(listed.files.map((f: any) => f.path));
// 2. Locate the TODO: name glob, then content grep
const found = await sb.get<any>(
`/v2/fs/search?path=${encodeURIComponent("/tmp/demo")}` +
`&pattern=${encodeURIComponent("**/*.py")}`
);
const target = found.files[0];
const hit = await sb.post<any>("/v2/fs/grep", { path: target, pattern: "TODO.*" });
for (const m of hit.matches) console.log(m.line_number, m.line_content);
// 4 # TODO: read the input path from argv
// 3. Edit with the editor tool; old_str matches literally
await 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
const read = await sb.get<any>(
`/v2/fs/read?path=${encodeURIComponent(target)}&start_line=2&end_line=5`
);
console.log(read.content);
// def main():
// path = sys.argv[1]
// print(open(path).read())除了创建目录,其余步骤都有 1.x SDK 方法。创建目录这一步通过共享 Aio helper(见 约定)调用:
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())import { SandboxClient } from "@agent-infra/sandbox";
const BASE_URL = "http://127.0.0.1:18091";
const client = new SandboxClient({ environment: BASE_URL });
const sb = new Aio(BASE_URL);
// 1. Scaffold: one directory, two files
// The 1.x SDK has no mkdir, so the helper posts the route itself.
await sb.post("/v1/file/mkdir", { path: "/tmp/demo/src", parents: true });
await client.file.writeFile({
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",
});
await client.file.writeFile({ file: "/tmp/demo/README.md", content: "# demo\n" });
const listed = ((await client.file.listPath({
path: "/tmp/demo", recursive: true,
})) as any).body.data;
console.log(listed.files.map((f: any) => f.path));
// 2. Locate the TODO: name glob, then content grep
const found = ((await client.file.globFiles({
path: "/tmp/demo", pattern: "**/*.py",
})) as any).body.data;
const target = found.files[0].path;
const hit = ((await client.file.grepFiles({
path: target, pattern: "TODO.*",
})) as any).body.data;
for (const m of hit.matches) console.log(m.line_number, m.line_content);
// 4 # TODO: read the input path from argv
// 3. Edit with the editor tool; old_str matches literally
await client.file.strReplaceEditor({
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
const read = ((await client.file.readFile({
file: target, start_line: 2, end_line: 5,
})) as any).body.data;
console.log(read.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:
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}")let watcher = (await sb.post<any>("/v2/watch", { path: "/tmp/demo", recursive: true }))
.watcher_id;
async function poll(id: string, qs: string): Promise<any> {
return sb.get<any>(`/v2/watch/${id}/poll?${qs}`);
}
await sb.post("/v2/commands", { command: "echo 'X = 1' > /tmp/demo/src/util.py" });
for (const e of (await poll(watcher, "timeout=2")).events) {
console.log(e.seq, e.type, e.relative_path);
}
// 1 create src/util.py
// 2 write src/util.py
await sb.delete(`/v2/watch/${watcher}`);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)let watcher = ((await client.file.watchCreate({
path: "/tmp/demo", recursive: true,
})) as any).body.data.watcher_id;
async function poll(id: string, params: any): Promise<any> {
return ((await client.file.watchPoll(id, params)) as any).body.data;
}
await client.bash.exec({ command: "echo 'X = 1' > /tmp/demo/src/util.py" });
for (const e of (await poll(watcher, { timeout: 2 })).events) {
console.log(e.seq, e.type, e.relative_path);
}
// 1 create src/util.py
// 2 write src/util.py
await client.file.watchStop(watcher);#过滤暂存文件
上传不会直接写入目标文件,而是先写入目标旁边的 .aiod-upload-<id>.part,完成后再重命名覆盖目标。
指定归属身份时,写入也会先放在 .aiod-write-<id>.part 中;不指定归属身份时,则直接写入目标并保留原 inode。
下面创建两个监听器监听同一目录,其中一个过滤掉暂存文件:
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.csvconst plain = (await sb.post<any>("/v2/watch", { path: "/tmp/demo", recursive: true }))
.watcher_id;
const filtered = (
await sb.post<any>("/v2/watch", {
path: "/tmp/demo", recursive: true, exclude: ["*.part"],
})
).watcher_id;
const form = new FormData();
form.append("file", new Blob(["a,b\n1,2\n"]), "data.csv");
form.append("path", "/tmp/demo/data.csv");
await fetch(`${BASE_URL}/v2/fs/upload`, { method: "POST", body: form });
for (const [name, w] of [["plain", plain], ["filtered", filtered]]) {
console.log(name);
for (const e of (await poll(w, "timeout=2")).events) {
console.log(" ", e.seq, e.type, e.relative_path);
}
}
await sb.delete(`/v2/watch/${plain}`);
await sb.delete(`/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.csvplain = 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.csvconst plain = ((await client.file.watchCreate({
path: "/tmp/demo", recursive: true,
})) as any).body.data.watcher_id;
const filtered = ((await client.file.watchCreate({
path: "/tmp/demo", recursive: true, exclude: ["*.part"],
})) as any).body.data.watcher_id;
await client.file.uploadFile({
file: new File(["a,b\n1,2\n"], "data.csv"),
path: "/tmp/demo/data.csv",
} as any);
for (const [name, w] of [["plain", plain], ["filtered", filtered]]) {
console.log(name);
for (const e of (await poll(w, { timeout: 2 })).events) {
console.log(" ", e.seq, e.type, e.relative_path);
}
}
await client.file.watchStop(plain);
await client.file.watchStop(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.csvexclude 的 glob 匹配被监听根目录下的单个路径片段,因此 *.part 挡掉暂存文件,只留下落在目标上的 rename。传入的列表会替换默认排除项,而不是在其上追加。
#等待下一次变更
把上一次的 cursor 传回去,下一次 poll 只返回它之后的事件。没有新事件时,会阻塞到 timeout 结束。
下面的循环用于等待变更:后台构建会先写报告,再写标记文件。
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}")watcher = (
await sb.post<any>("/v2/watch", {
path: "/tmp/demo", recursive: true, exclude: ["*.part"],
})
).watcher_id;
await sb.post("/v2/commands", {
mode: "async",
command: "sleep 1; echo report > /tmp/demo/report.txt; touch /tmp/demo/build.done",
});
let cursor = 0;
for (;;) {
const page = await poll(watcher, `cursor=${cursor}&timeout=30`);
for (const e of page.events) console.log(e.seq, e.type, e.relative_path);
cursor = page.cursor;
if (page.events.some((e: any) => e.relative_path === "build.done")) break;
}
// 1 create report.txt
// 2 write report.txt
// 3 create build.done
await sb.delete(`/v2/watch/${watcher}`);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)watcher = ((await client.file.watchCreate({
path: "/tmp/demo", recursive: true, exclude: ["*.part"],
})) as any).body.data.watcher_id;
await client.bash.exec({
async_mode: true,
command: "sleep 1; echo report > /tmp/demo/report.txt; touch /tmp/demo/build.done",
} as any);
let cursor = 0;
for (;;) {
const page = await poll(watcher, { cursor, timeout: 30 });
for (const e of page.events) console.log(e.seq, e.type, e.relative_path);
cursor = page.cursor;
if (page.events.some((e: any) => e.relative_path === "build.done")) break;
}
// 1 create report.txt
// 2 write report.txt
// 3 create build.done
await client.file.watchStop(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:
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 ENOENTconst res = await fetch(
`${BASE_URL}/v2/fs/read?path=${encodeURIComponent("/tmp/demo/missing.txt")}`
);
const reply = await res.json();
console.log(reply.success, reply.data.error_type, reply.data.errno_name);
// false not_found ENOENTSDK 不抛异常,而是把返回结构交回来,所以失败的响应不需要特殊处理:
reply = client.file.read_file(file="/tmp/demo/missing.txt")
print(reply.success, reply.data.error_type, reply.data.errno_name)
# False not_found ENOENTconst reply = ((await client.file.readFile({
file: "/tmp/demo/missing.txt",
})) as any).body;
console.log(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。
#相关页面
- Agent 调用沙箱 —— 文件 API 作为脚本中的一步
- 交互式终端 —— 同一个文件系统上的实时 shell
- 错误处理 —— 各 plane 通用的返回结构与状态码
- 文件(FS) —— 完整的 v2 路由表与用法;路由对照 有 v1 的对应关系