archil 0.8.19

Pure-Python client for Archil disks

archil

Python client for Archil disks. Create disks, list and inspect them, manage who can mount them, run commands against them, and read/write their contents through the S3-compatible object API — all from scripts, CI, or notebooks. It also ships drop-in agent tools that turn a disk into a ready-made filesystem toolset for OpenAI Agents, LangChain, and other frameworks.

archil talks to the Archil control plane over HTTPS and has no native dependencies.

Every method works both synchronously and asynchronously from a single implementation: disk.put_object(...) blocks, while disk.put_object.aio(...) returns a coroutine you can await. This is powered by synchronicity — the same approach Modal uses — so there's one source of truth and no duplicated sync/async logic.

Install

pip install archil

Library

import archil

# Configure once per process — falls back to ARCHIL_API_KEY / ARCHIL_REGION env vars.
archil.configure(api_key="key-...", region="aws-us-east-1")

# Create a disk. `token` here is the disk token — the one-time credential for mounting.
result = archil.create_disk(name="my-disk")
print(f"Created {result.disk.id}, disk token: {result.token}")

# A freshly-created disk starts in "creating"; block until it's usable.
disk = result.disk.wait_until_ready()  # raises on terminal failure / timeout

# List and look up disks
all_disks = archil.list_disks()
d = archil.get_disk(result.disk.id)

Per-disk operations are methods on the Disk object itself, not top-level functions:

d = archil.get_disk("dsk-abc123")

# Run a command in a container with the disk mounted
res = d.exec("ls -la /mnt && cat /mnt/config.json")
print(res.stdout, res.stderr, res.exit_code)

# Manage who can mount the disk
from archil import TokenUser
user = d.add_user(TokenUser(nickname="ci"))
d.remove_user("token", user.identifier)

# Delete
d.delete()

Account-level API keys are top-level helpers:

archil.list_api_keys()
archil.create_api_key(name="ci-bot", description="GitHub Actions")
archil.delete_api_key("key-abc123")

Reading and writing objects

A Disk doubles as an S3-compatible bucket: read, write, delete, and list its files by key without mounting it. These methods talk to Archil's S3 endpoint using your same API key (no separate S3 credentials or SigV4 signing on your part).

import json
d = archil.get_disk("dsk-abc123")
report = {"generated": "2026-01", "rows": 1234}

# Write — accepts str or bytes. content_type is optional. Returns the etag.
result = d.put_object("reports/2026-01/data.json", json.dumps(report), "application/json")

# Read — returns bytes.
data = d.get_object("reports/2026-01/data.json")
text = data.decode("utf-8")

# Metadata / existence without downloading the body
meta = d.head_object("reports/2026-01/data.json")  # None if absent
if d.object_exists("reports/2026-01/data.json"):
    ...

# Delete (idempotent — deleting a missing key succeeds)
d.delete_object("reports/2026-01/data.json")

list_objects auto-paginates by default, returning every matching key. The first argument is a key prefix; a non-recursive listing (the default) returns the immediate level as objects plus subdirectory common_prefixes:

result = d.list_objects("reports/")                       # one level
all_keys = d.list_objects("reports/", recursive=True)     # whole subtree
first_100 = d.list_objects("reports/", limit=100)         # cap the total

# Stream pages instead of buffering everything (large listings):
for page in d.list_objects_pages("reports/"):
    for obj in page.objects:
        print(obj.key, obj.size, obj.last_modified)

# Or drive pagination yourself:
page = d.list_objects("reports/", single_page=True)
if page.is_truncated:
    nxt = d.list_objects("reports/", single_page=True, continuation_token=page.next_continuation_token)

Large uploads and bulk delete

put_object handles any size with one call. Small bodies go through a single request; large ones are uploaded as a multipart upload automatically — split into parts, uploaded with bounded concurrency, and assembled, aborting the upload if any part fails so nothing is left half-staged. You don't pick a different method for big files. For very large objects the part size is grown automatically so the upload never exceeds S3's 10,000-part limit.

# Small or multi-gigabyte — same call.
d.put_object("reports/2026-01/data.json", json.dumps(report), "application/json")

result = d.put_object(
    "backups/2026-01.tar",
    big_bytes,
    content_type="application/x-tar",
    multipart_threshold=5 * 1024 * 1024,  # switch to multipart above 5 MiB; default = part_size
    part_size=32 * 1024 * 1024,           # >= 5 MiB; default 16 MiB
    concurrency=8,                        # parts in flight at once; default 4
)
print(result.etag)

For manual control over the multipart lifecycle (e.g. uploading parts from separate processes), the raw S3 primitives live in the opt-in d.multipart namespace — create, upload_part, complete, abort, list_parts, list_uploads. Most code never needs these.

upload = d.multipart.create("big.bin")
p1 = d.multipart.upload_part("big.bin", upload.upload_id, 1, first_chunk)
p2 = d.multipart.upload_part("big.bin", upload.upload_id, 2, second_chunk)
d.multipart.complete("big.bin", upload.upload_id, [p1, p2])

delete_objects removes many keys in one round trip (auto-batched at S3's 1000-key limit). Unlike delete_object, per-key failures are returned rather than raised:

result = d.delete_objects(["a.txt", "logs/b.txt", "c.txt"])
for e in result.errors:
    print(f"{e.key}: {e.code} {e.message}")

append_object appends bytes to an existing object (creating it if absent) — handy for log-style writes. Each call may append at most 1 MiB; append in chunks to grow past that.

d.append_object("logs/app.log", "first line\n")
d.append_object("logs/app.log", "second line\n")  # concatenated

Transient failures (HTTP 429 and 5xx, plus network errors) are retried automatically with jittered exponential backoff before surfacing; caller errors (other 4xx) are not retried. The two non-idempotent operations — complete (multipart) and append_object — are not auto-retried, since a retry after a succeeded-but-unacknowledged call would return a spurious NoSuchUpload (complete) or duplicate the appended bytes (append).

Failures raise ArchilS3Error with status (HTTP status), code (the S3 error code, e.g. "NoSuchKey"), request_id, and the raw body on raw. get_object on a missing key raises a 404 — use head_object / object_exists to probe without catching. All SDK errors extend ArchilError, so except ArchilError handles control-plane and S3 failures uniformly.

The S3 endpoint is derived from your region automatically. To target a custom environment, pass s3_base_url to Archil(...) (or set the ARCHIL_S3_BASE_URL env var).

Sharing files

share mints a signed, time-limited link to a single file. Anyone with the link can download that file — no API key, no mounting. The link carries a cryptographically signed token (disk + key + expiry); when it expires it stops working.

d = archil.get_disk("dsk-abc123")

# Default lifetime is 24 hours.
link = d.share("reports/2026-01/summary.pdf")
print(link.url)         # https://control.…/api/shared/<token>
print(link.expires_in)  # 86400

# Set the lifetime in seconds (any positive integer, up to 604800 = 7 days):
week_link = d.share("reports/2026-01/summary.pdf", expires_in=604800)

Async

Every method on Archil, Disks, Disk, and Tokens has an .aio variant that returns a coroutine. (The module-level helpers — configure, create_disk, get_disk, etc. — are synchronous convenience wrappers; from async code, construct Archil(...) directly and use .aio.) Construct the client directly and await:

import asyncio
from archil import Archil

async def main():
    async with Archil(api_key="key-...", region="aws-us-east-1") as client:
        d = await client.disks.get.aio("dsk-abc123")
        await d.put_object.aio("a/b.txt", b"hello")
        data = await d.get_object.aio("a/b.txt")
        async for page in d.list_objects_pages.aio("a/"):
            for obj in page.objects:
                print(obj.key)

asyncio.run(main())

Multiple accounts or regions

For multi-tenant scripts, instantiate Archil directly instead of using the module-level configure:

from archil import Archil

prod = Archil(api_key=prod_key, region="aws-us-east-1")
staging = Archil(api_key=staging_key, region="aws-us-east-1")

prod_disks = prod.disks.list()
staging_disks = staging.disks.list()

Agent tools

Turn a disk — or a multi-disk workspace — into a ready-made filesystem toolset for popular agent frameworks, so you can hand an LLM a real filesystem to work in. The agent gets six tools: read_file, write_file, delete_file, list_files, grep, and run_bash (an arbitrary command in a container with the disk mounted).

import archil
from agents import Agent  # OpenAI Agents SDK

d = archil.get_disk("dsk-abc123")

# Single disk → tools for your framework. The disk is the filesystem root (/).
agent = Agent(name="assistant", tools=d.agent_tools().for_openai_agents())

Or span several disks in one workspace (the same shape as exec's mounts). Each disk is a top-level directory (e.g. /data/…, /cache/…); file operations route to the right disk by path, and grep / list_files fan out across all of them. run_bash starts at the common root, so relative paths line up:

ws = archil.workspace({
    "data": archil.get_disk("dsk-data"),
    "cache": archil.get_disk("dsk-cache"),
})
tools = ws.agent_tools().for_langchain()  # LangChain / LangGraph

A Workspace is a full filesystem in its own right — the same object API a Disk has (get_object / put_object / delete_object / list_objects / grep / exec; both satisfy the FileSystem protocol), so you can use it directly without the agent tools, and add or remove disks at runtime with ws.add_disk(name, disk) / ws.remove_disk(name). A workspace's keys carry the disk name as their first segment:

data = ws.get_object("data/reports/q1.csv")         # routes to the "data" disk
ws.add_disk("scratch", archil.get_disk("dsk-tmp"))  # mount another disk live

Pick the adapter for your framework — each returns that framework's native tool objects:

Method	Framework	Returns
.for_openai_agents()	OpenAI Agents SDK	list[FunctionTool]
.for_langchain()	LangChain / LangGraph	list[StructuredTool]

Frameworks are optional — install the matching extra:

pip install "archil[openai-agents]"   # OpenAI Agents SDK
pip install "archil[langchain]"       # LangChain / LangGraph

Expose a subset of tools with tools=[...] (e.g. read-only research: d.agent_tools(tools=["read_file", "list_files", "grep"])). To give an agent read-only access to a disk in a workspace, mount it with ExecMountSpec(disk=d, read_only=True) — write_file / delete_file then return an error instead of mutating it.

Connecting to a disk's data plane

To run a command against a disk, use Disk.exec() — it returns stdout, stderr, and an exit code from an Archil-managed container with the disk pre-mounted. No local filesystem involved.

To mount a disk as a real filesystem on your machine, use the archil CLI — it mounts through the OS kernel via FUSE, so any program can read and write files with standard APIs. Mounting from Python is not supported; use exec() or the S3-compatible object API instead.

Supported regions

Region	Provider
aws-us-east-1	AWS
aws-us-west-2	AWS
aws-eu-west-1	AWS
gcp-us-central1	GCP

FAQ

What's the difference between an API key and a disk token?

• API key — account-level credential for the control plane. You use one whenever you call archil. Create and manage them at console.archil.com. Goes in the ARCHIL_API_KEY env var or the api_key argument.
• Disk token — per-disk credential that lets a client mount a specific disk. Created automatically when you create_disk(...) (the value is shown once; save it).

Support

Questions, feature requests, or issues? Reach us at [e-mail].