How to Give an Agent a Large File Without Filling Its Context

id=sb_wo9x0s9c5c
sbx exec $id "mkdir -p ~/inbox"
time sbx scp ./interview.wav $id:/home/user/inbox/
# real    0m1.871s
time sbx scp ./orders.csv $id:/home/user/inbox/
# real    0m1.680s
sbx exec $id "sha256sum ~/inbox/*"
# e4b2d1cd3983ebe47f929e7af59b97480df165a776f0dccdf3294f2606a6736e  /home/user/inbox/interview.wav
# 274209ff95f2896061a7b8881087fcb44111441c1682d835de8809b59160a1a2  /home/user/inbox/orders.csv

sbx exec $id --timeout 120 "cd ~/inbox && curl -sS -o trips.parquet \
  -w 'HTTP %{http_code} %{size_download} bytes in %{time_total} s\n' \
  https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2024-01.parquet"
# HTTP 200 49961641 bytes in 0.690292 s

import hashlib, time
from sandboxer import Sandboxer

sandbox = Sandboxer.get("sb_wo9x0s9c5c")
data = open("orders.csv", "rb").read()

t = time.time()
CHUNK = 4 * 1024 * 1024
parts = []
for n, off in enumerate(range(0, len(data), CHUNK)):
    name = f"sdk/parts/{n:03d}"
    sandbox.files.write(name, data[off:off + CHUNK])
    parts.append(name)
run = sandbox.exec("cat sdk/parts/* > sdk/orders.csv && rm -r sdk/parts && sha256sum sdk/orders.csv")
print(f"{len(parts)} parts in {time.time() - t:.1f} s")
print(run.stdout.strip())

13 parts in 2.4 s
274209ff95f2896061a7b8881087fcb44111441c1682d835de8809b59160a1a2  sdk/orders.csv

#!/home/user/v/bin/python
"""Small tools an agent calls on big files. Each prints compact JSON, never the file's bytes."""
import hashlib, json, os, re, subprocess, sys

def file_info(path):
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for block in iter(lambda: f.read(1 << 20), b""):
            h.update(block)
    kind = subprocess.run(["file", "-b", path], capture_output=True, text=True).stdout.strip()
    return {"path": path, "bytes": os.path.getsize(path), "sha256": h.hexdigest(), "type": kind[:80]}

def audio_probe(path):
    out = subprocess.run(["ffprobe", "-v", "error", "-show_entries",
        "format=duration,bit_rate:stream=codec_name,sample_rate,channels",
        "-of", "json", path], capture_output=True, text=True, check=True).stdout
    j = json.loads(out); s = j["streams"][0]
    return {"duration_s": round(float(j["format"]["duration"]), 1), "codec": s["codec_name"],
            "sample_rate": int(s["sample_rate"]), "channels": s["channels"]}

def audio_segments(path, outdir):
    # Find silences, cut at their midpoints, write 16 kHz mono WAVs a transcriber can take one at a time.
    log = subprocess.run(["ffmpeg", "-hide_banner", "-i", path, "-af", "silencedetect=n=-40dB:d=1",
        "-f", "null", "-"], capture_output=True, text=True).stderr
    starts = [float(x) for x in re.findall(r"silence_start: ([\d.]+)", log)]
    ends = [float(x) for x in re.findall(r"silence_end: ([\d.]+)", log)]
    cuts = [round((a + b) / 2, 2) for a, b in zip(starts, ends)]
    os.makedirs(outdir, exist_ok=True)
    subprocess.run(["ffmpeg", "-v", "error", "-y", "-i", path, "-ac", "1", "-ar", "16000",
        "-f", "segment", "-segment_times", ",".join(map(str, cuts)) or "99999",
        f"{outdir}/seg_%03d.wav"], check=True)
    segs = sorted(os.listdir(outdir))
    return {"segments": len(segs), "dir": outdir, "first": segs[0], "last": segs[-1],
            "bytes_total": sum(os.path.getsize(f"{outdir}/{s}") for s in segs)}

def csv_head(path, n=5):
    with open(path) as f:
        return {"lines": [next(f).rstrip("\n") for _ in range(int(n))]}

def sql(query, limit=20):
    import duckdb
    rel = duckdb.sql(query)
    rows = rel.limit(int(limit)).fetchall()
    return {"columns": rel.columns, "rows": [list(map(str, r)) for r in rows]}

if __name__ == "__main__":
    tool, *args = sys.argv[1:]
    print(json.dumps(globals()[tool](*args), separators=(",", ":")))

#!/bin/bash
# measure.sh <tool> <args...>
cd ~
s=$EPOCHREALTIME
out=$(./ftools.py "$@")
e=$EPOCHREALTIME
echo "$out"
printf '[%d bytes, %.2f s]\n' "$(printf '%s' "$out" | wc -c)" "$(awk "BEGIN{print $e - $s}")"

sbx exec $id --timeout 120 "~/measure.sh audio_probe inbox/interview.wav"

$ file_info inbox/interview.wav
{"path":"inbox/interview.wav","bytes":52224078,"sha256":"e4b2d1cd…6736e","type":"RIFF (little-endian) data, WAVE audio, Microsoft PCM, 16 bit, stereo 48000 Hz"}
[210 bytes, 0.10 s]
$ audio_probe inbox/interview.wav
{"duration_s":272.0,"codec":"pcm_s16le","sample_rate":48000,"channels":2}
[73 bytes, 0.12 s]
$ audio_segments inbox/interview.wav work/segments
{"segments":7,"dir":"work/segments","first":"seg_000.wav","last":"seg_006.wav","bytes_total":8704546}
[101 bytes, 1.12 s]
$ csv_head inbox/orders.csv 4
{"lines":["order_id,created_at,region,sku,qty,unit_price_cents,customer_id","1,2026-09-11T04:25:00Z,us-east,SKU-0037,9,6367,C095864",…]}
[249 bytes, 0.05 s]

import json
from sandboxer import Sandboxer

sandbox = Sandboxer.get("sb_wo9x0s9c5c")

def call(tool, *args):
    """What the agent's tool handler does: run one small tool on the file in the sandbox."""
    run = sandbox.exec(["/home/user/ftools.py", tool, *map(str, args)], timeout_seconds=120)
    if run.exit_code != 0:
        return {"error": run.stderr[-500:]}          # keep errors small too
    return json.loads(run.stdout)

call("sql", "SELECT region, count(*) AS orders, round(sum(qty * unit_price_cents) / 100) AS revenue_usd "
            "FROM 'inbox/orders.csv' GROUP BY region ORDER BY revenue_usd DESC")
call("sql", "SELECT sku, sum(qty) AS units FROM 'inbox/orders.csv' GROUP BY sku ORDER BY units DESC", 3)
call("sql", "SELECT count(*) AS trips, round(avg(trip_distance), 2) AS avg_miles, "
            "round(avg(total_amount), 2) AS avg_total FROM 'inbox/trips.parquet'")

{"columns":["region","orders","revenue_usd"],"rows":[["ap-south","169590","212815376.0"],["us-east","168990","212597864.0"],…]}
[233 bytes, 0.31 s round trip]
{"columns":["sku","units"],"rows":[["SKU-0248","11354"],["SKU-0338","11305"],["SKU-0165","11260"]]}
[99 bytes, 0.28 s round trip]
{"columns":["trips","avg_miles","avg_total"],"rows":[["2964624","3.65","26.8"]]}
[80 bytes, 0.17 s round trip]