• 简体中文
  • 文件操作

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

    本示例使用以下路由:

    • 文件:/v2/fs/*
    • 事件:/v2/watch
    • 文件和事件:/v1/file/*

    要求

    GET /v2/sandboxcapabilities 中应包含 files

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

    文件 API 与命令、终端、代码执行共享同一个文件系统;隔离由沙箱负责,而不是由 API 负责。

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

    它不是执行身份,文件操作仍使用 daemon 的权限。字节流路由(download 和作为 GETtree)以及 /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_replaceinsertundo_edit

    • view 返回 400,改用 GET /v2/fs/read
    • create 返回 400,改用 POST /v2/fs/write

    viewcreate 在 v1 中仍然可用。

    old_str 不存在时,str_replace 返回 400 old_str not found in fileoutput 是工具展示给模型的 cat -n 片段;old_contentnew_content 是修改前后的完整文件。

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

    • GET /v2/fs/search:合并 v1 的 globfind,返回扁平的路径列表。
    • POST /v2/fs/grep:在 path 下搜索内容,每个匹配返回从 1 开始的 line_number,也可以请求 context_beforecontext_after
    • POST /v1/file/glob:返回包含 path 和元数据的记录。
    • POST /v1/file/grep:在 path 下搜索内容,每个匹配返回从 1 开始的 line_number,也可以请求 context_beforecontext_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 以流返回字节并支持 Rangeupload 是 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
    400bad_requestinvalid_pathinvalid_target
    403permission_denied
    404not_found
    409already_exists
    422decode_error
    507no_space_left

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

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

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

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

    情况状态码看哪里
    文件系统失败(路径不存在、权限不足……)200data.error_typedata.errno_namedata.retryable
    JSON body 格式错误(字段缺失或类型错误)422errors[].location
    str_replaceold_str 不存在200message

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

    相关页面