Files Module

The files module reads and writes files from workflows. It supports two storage modes:

• cloud (default) — S3 storage, scoped per organization
• local — Local filesystem under the CLI working directory and /tmp/bifrost/

And three storage locations:

Location	Cloud (S3 prefix)	Local path
workspace	bucket root	current working directory
temp	_tmp/	/tmp/bifrost/temp
uploads	uploads/	/tmp/bifrost/uploads

uploads holds files submitted via form file fields. Paths are sandboxed — reads/writes outside the configured locations raise ValueError.

Import

from bifrost import files

Method Index

Method	Returns	Description
files.read()	str	Read a text file
files.read_bytes()	bytes	Read a binary file
files.write()	None	Write text
files.write_bytes()	None	Write binary data
files.list()	list[str]	List files in a directory
files.delete()	None	Delete a file
files.exists()	bool	Check if a file exists
files.get_signed_url()	dict	Generate a presigned S3 URL for direct upload/download

All methods accept location and mode keyword arguments (defaults: "workspace", "cloud").

files.read()

async def read(
    path: str,
    location: Literal["workspace", "temp", "uploads"] = "workspace",
    mode: Literal["local", "cloud"] = "cloud",
) -> str

Returns the file contents as a string. Raises FileNotFoundError if missing, ValueError if path escapes the location root.

content = await files.read("data/customers.csv")
uploaded = await files.read("form_id/uuid/upload.txt", location="uploads")

files.read_bytes()

async def read_bytes(
    path: str,
    location: Location = "workspace",
    mode: Mode = "cloud",
) -> bytes

Same as read() but returns raw bytes. Use for binary formats (PDFs, images).

pdf = await files.read_bytes("reports/q4.pdf")

files.write()

async def write(
    path: str,
    content: str,
    location: Location = "workspace",
    mode: Mode = "cloud",
) -> None

Write text content. Creates parent directories as needed.

await files.write("output/report.txt", "Report data\n")

files.write_bytes()

async def write_bytes(
    path: str,
    content: bytes,
    location: Location = "workspace",
    mode: Mode = "cloud",
) -> None

Write binary data.

await files.write_bytes("uploads/logo.png", image_bytes)

files.list()

async def list(
    directory: str = "",
    location: Location = "workspace",
    mode: Mode = "cloud",
) -> list[str]

List file and directory names directly under directory (defaults to the location root).

items = await files.list("uploads")
for name in items:
    print(name)

files.delete()

async def delete(
    path: str,
    location: Location = "workspace",
    mode: Mode = "cloud",
) -> None

Delete a single file. Raises FileNotFoundError if it doesn’t exist.

await files.delete("temp/old_export.csv", location="temp")

files.exists()

async def exists(
    path: str,
    location: Location = "workspace",
    mode: Mode = "cloud",
) -> bool

if await files.exists("data/customers.csv"):
    customers = await files.read("data/customers.csv")

files.get_signed_url()

Generate a presigned S3 URL so a client (browser, external service) can upload or download a file directly without proxying through Bifrost. Always cloud-mode and always scoped to the current org.

async def get_signed_url(
    path: str,
    method: Literal["PUT", "GET"] = "PUT",
    content_type: str = "application/octet-stream",
) -> dict

Parameter	Type	Description
path	str	File path within the org’s S3 namespace
method	"PUT" | "GET"	PUT for uploads, GET for downloads
content_type	str	MIME type (used for PUT only)

Returns:

Key	Type	Description
url	str	Presigned URL
path	str	Full scoped S3 path
expires_in	int	Lifetime in seconds

upload = await files.get_signed_url("incoming/photo.jpg", method="PUT", content_type="image/jpeg")
# Hand `upload["url"]` to the browser; it can PUT directly to S3.