English

File Operations

This example follows a common development task: scaffold a small project, find and fix a TODO, read the changed lines, then watch the directory while a command changes it and handle the error contract.

This example uses these routes:

Files: /v2/fs/*
Events: /v2/watch

Files and events: /v1/file/*

Requirements

GET /v2/sandbox should include files in capabilities.

GET /v1/capabilities should report files.

The file plane shares one filesystem with commands, terminals, and code execution; confinement is the container's job, not the API's.

Most /v2/fs/* routes also take an optional ?user= — the account attributed as the operation's owner, not an execution identity; the operation still runs with daemon privileges. The byte-stream routes (download, tree as a GET) and /v2/watch have no such parameter.

Scaffold, search, edit

Make a directory, write two files into it, list what landed, find the TODO by name and then by content, replace it, and read the changed lines back:

The shared Aio helper (Conventions) posts to a route and unwraps the envelope:

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())

The 1.x SDK covers every step but the directory, so the shared Aio helper (Conventions) posts that one route:

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())

In v1, the editor tool exposes only the commands that do not already have another route: str_replace, insert, undo_edit.

view answers 400; use GET /v2/fs/read instead.
create answers 400; use POST /v2/fs/write instead.

view and create still work here.

A str_replace whose old_str is absent answers 400 old_str not found in file. output is the cat -n snippet the tool shows a model; old_content and new_content carry the full file before and after.

The name glob and the content grep are two different routes.

GET /v2/fs/search: v1's glob and find collapsed into one; returns a flat list of paths.
POST /v2/fs/grep: matches content under path, reports 1-based line_number per match, and takes optional context_before/context_after.

POST /v1/file/glob: returns an entry per match with path and its metadata.
POST /v1/file/grep: matches content under path, reports 1-based line_number per match, and takes optional context_before/context_after.
POST /v1/file/search: the older file-scoped, 0-based variant.

Watch a directory

A watcher records changes under a path until it is stopped.

Poll for events

A poll long-polls for up to timeout seconds and returns a cursor for the next call:

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)

Filter the staging file

An upload never writes the target directly: it fills a .aiod-upload-<id>.part file next to it, then renames that over the target. A write stages the same way through .aiod-write-<id>.part when an ownership identity applies, as it does on the images; without one it writes the target in place and keeps its inode. Two watchers on the same directory, one of them filtering the staging file:

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 globs match single path components below the watched root, so *.part hides the staging file and keeps the rename onto the target. The list you pass replaces the default excludes rather than extending them.

Wait for the next change

Pass the previous cursor back and the next poll returns only what came after it. With nothing new, it blocks until timeout runs out.

The example below uses that loop to wait for a background build that writes a report and then a marker file:

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)

Stream the events

The events route pushes the same events as SSE. Every file_change carries the poll payload and an id of <watcher_id>:<seq>, so an EventSource reconnect resumes from Last-Event-ID instead of replaying from the start:

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}

Transfer files

download streams the bytes and honours Range; upload is multipart with the target in 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"

When a whole directory has to move in one request, switch to v2:

tree streams a whole directory out as a tar archive, and PUT with an archive as the body puts one back:

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

Error contract

A filesystem failure keeps the envelope: success: false, the message, and a structured data naming the errno, the error kind, the operation, the path, and whether a retry could help.

The helper raises on success: false, so reading a failing reply means going through sb.http instead of 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

The SDK hands back the envelope instead of raising, so a failing reply needs no special handling:

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

The reply on the wire is the same either way:

{
  "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
}

The HTTP status carries the kind too:

Status	`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`

A malformed JSON body answers 422 with a top-level errors list. A malformed query string (a required parameter missing or the wrong type) answers 400 as plain text, not the envelope.

Case	Status	Where to look
Filesystem failure (missing path, permission denied, ...)	`400`/`403`/`404`/`409`/`422`/`507` by kind	`data.error_type`, `data.errno_name`, `data.retryable`
Malformed JSON body (missing or mistyped field)	`422`	`errors[].location`
Malformed query string (missing required param)	`400`, plain text	not the envelope
`str_replace` with an absent `old_str`	`400`	`message`

The HTTP status stays 200 whatever the failure was, so success and data.error_type are the only things that name it. A malformed JSON body is the exception: it answers 422 with a top-level errors list before any handler runs.

Case	Status	Where to look
Filesystem failure (missing path, permission denied, ...)	`200`	`data.error_type`, `data.errno_name`, `data.retryable`
Malformed JSON body (missing or mistyped field)	`422`	`errors[].location`
`str_replace` with an absent `old_str`	`200`	`message`

Check the HTTP status, then success, then data.error_type.

Agent Calls the Sandbox — the file API as one step in a script
Interactive Terminal — a live shell on the same filesystem
Error Handling — the envelope and status codes across every plane
File (FS) — the full v2 route table and usage; the route map has the v1 equivalents

#File Operations

#Requirements

#Scaffold, search, edit

#Watch a directory

#Poll for events

#Filter the staging file

#Wait for the next change

#Stream the events

#Transfer files

#Error contract

#Related

File Operations

Requirements

Scaffold, search, edit

Watch a directory

Poll for events

Filter the staging file

Wait for the next change

Stream the events

Transfer files

Error contract

Related