CompositeBackend

CompositeBackend

Bases: BackendProtocol

Routes file operations to different backends by path prefix.

Matches paths against route prefixes (longest first) and delegates to the corresponding backend. Unmatched paths use the default backend.

ATTRIBUTE	DESCRIPTION
default	Backend for paths that don't match any route.
routes	Map of path prefixes to backends (e.g., {"/memories/": store_backend}).
sorted_routes	Routes sorted by length (longest first) for correct matching.

Examples:

composite = CompositeBackend(default=StateBackend(runtime), routes={"/memories/": StoreBackend(runtime), "/cache/": StoreBackend(runtime)})

composite.write("/temp.txt", "data")
composite.write("/memories/note.txt", "data")

METHOD	DESCRIPTION
__init__	Initialize composite backend.
ls_info	List directory contents (non-recursive).
als_info	Async version of ls_info.
read	Read file content, routing to appropriate backend.
aread	Async version of read.
grep_raw	Search files for regex pattern.
agrep_raw	Async version of grep_raw.
glob_info	Find files matching a glob pattern.
aglob_info	Async version of glob_info.
write	Create a new file, routing to appropriate backend.
awrite	Async version of write.
edit	Edit a file, routing to appropriate backend.
aedit	Async version of edit.
execute	Execute shell command via default backend.
aexecute	Async version of execute.
upload_files	Upload multiple files, batching by backend for efficiency.
aupload_files	Async version of upload_files.
download_files	Download multiple files, batching by backend for efficiency.
adownload_files	Async version of download_files.

__init__

__init__(
    default: BackendProtocol | StateBackend, routes: dict[str, BackendProtocol]
) -> None

Initialize composite backend.

PARAMETER	DESCRIPTION
default	Backend for paths that don't match any route. TYPE: BackendProtocol | StateBackend
routes	Map of path prefixes to backends. Prefixes must start with "/" and should end with "/" (e.g., "/memories/"). TYPE: dict[str, BackendProtocol]

ls_info

ls_info(path: str) -> list[FileInfo]

List directory contents (non-recursive).

If path matches a route, lists only that backend. If path is "/", aggregates default backend plus virtual route directories. Otherwise lists default backend.

PARAMETER	DESCRIPTION
path	Absolute directory path starting with "/". TYPE: str

RETURNS	DESCRIPTION
list[FileInfo]	List of FileInfo dicts. Directories have trailing "/" and is_dir=True.
list[FileInfo]	Route prefixes are restored in returned paths.

Examples:

infos = composite.ls_info("/")
infos = composite.ls_info("/memories/")

als_info async

als_info(path: str) -> list[FileInfo]

Async version of ls_info.

read

read(file_path: str, offset: int = 0, limit: int = 2000) -> str

Read file content, routing to appropriate backend.

PARAMETER	DESCRIPTION
file_path	Absolute file path. TYPE: str
offset	Line offset to start reading from (0-indexed). TYPE: int DEFAULT: 0
limit	Maximum number of lines to read. TYPE: int DEFAULT: 2000

RETURNS	DESCRIPTION
str	Formatted file content with line numbers, or error message.

aread async

aread(file_path: str, offset: int = 0, limit: int = 2000) -> str

Async version of read.

grep_raw

grep_raw(
    pattern: str, path: str | None = None, glob: str | None = None
) -> list[GrepMatch] | str

Search files for regex pattern.

Routes to backends based on path: specific route searches one backend, "/" or None searches all backends, otherwise searches default backend.

PARAMETER	DESCRIPTION
pattern	Regex pattern to search for. TYPE: str
path	Directory to search. None searches all backends. TYPE: str | None DEFAULT: None
glob	Glob pattern to filter files (e.g., ".py", "**/.txt"). Filters by filename, not content. TYPE: str | None DEFAULT: None

RETURNS	DESCRIPTION
list[GrepMatch] | str	List of GrepMatch dicts with path (route prefix restored), line
list[GrepMatch] | str	(1-indexed), and text. Returns error string on failure.

Examples:

matches = composite.grep_raw("TODO", path="/memories/")
matches = composite.grep_raw("error", path="/")
matches = composite.grep_raw("import", path="/", glob="*.py")

agrep_raw async

agrep_raw(
    pattern: str, path: str | None = None, glob: str | None = None
) -> list[GrepMatch] | str

Async version of grep_raw.

See grep_raw() for detailed documentation on routing behavior and parameters.

glob_info

glob_info(pattern: str, path: str = '/') -> list[FileInfo]

Find files matching a glob pattern.

PARAMETER	DESCRIPTION
pattern	Glob pattern with wildcards to match file paths. Supports standard glob syntax: - * matches any characters within a filename/directory - ** matches any directories recursively - ? matches a single character - [abc] matches one character from set TYPE: str
path	Base directory to search from. Default: "/" (root). The pattern is applied relative to this path. TYPE: str DEFAULT: '/'

RETURNS	DESCRIPTION
list[FileInfo]	list of FileInfo

aglob_info async

aglob_info(pattern: str, path: str = '/') -> list[FileInfo]

Async version of glob_info.

write

write(file_path: str, content: str) -> WriteResult

Create a new file, routing to appropriate backend.

PARAMETER	DESCRIPTION
file_path	Absolute file path. TYPE: str
content	File content as a string. TYPE: str

RETURNS	DESCRIPTION
WriteResult	Success message or Command object, or error if file already exists.

awrite async

awrite(file_path: str, content: str) -> WriteResult

Async version of write.

edit

edit(
    file_path: str, old_string: str, new_string: str, replace_all: bool = False
) -> EditResult

Edit a file, routing to appropriate backend.

PARAMETER	DESCRIPTION
file_path	Absolute file path. TYPE: str
old_string	String to find and replace. TYPE: str
new_string	Replacement string. TYPE: str
replace_all	If True, replace all occurrences. TYPE: bool DEFAULT: False

RETURNS	DESCRIPTION
EditResult	Success message or Command object, or error message on failure.

aedit async

aedit(
    file_path: str, old_string: str, new_string: str, replace_all: bool = False
) -> EditResult

Async version of edit.

execute

execute(command: str) -> ExecuteResponse

Execute shell command via default backend.

PARAMETER	DESCRIPTION
command	Shell command to execute. TYPE: str

RETURNS	DESCRIPTION
ExecuteResponse	ExecuteResponse with output, exit code, and truncation flag.

RAISES	DESCRIPTION
NotImplementedError	If default backend doesn't implement SandboxBackendProtocol.

Examples:

composite = CompositeBackend(default=FilesystemBackend(root_dir="/tmp"), routes={"/memories/": StoreBackend(runtime)})

result = composite.execute("ls -la")

aexecute async

aexecute(command: str) -> ExecuteResponse

Async version of execute.

upload_files

upload_files(files: list[tuple[str, bytes]]) -> list[FileUploadResponse]

Upload multiple files, batching by backend for efficiency.

Groups files by their target backend, calls each backend's upload_files once with all files for that backend, then merges results in original order.

PARAMETER	DESCRIPTION
files	List of (path, content) tuples to upload. TYPE: list[tuple[str, bytes]]

RETURNS	DESCRIPTION
list[FileUploadResponse]	List of FileUploadResponse objects, one per input file.
list[FileUploadResponse]	Response order matches input order.

aupload_files async

aupload_files(files: list[tuple[str, bytes]]) -> list[FileUploadResponse]

Async version of upload_files.

download_files

download_files(paths: list[str]) -> list[FileDownloadResponse]

Download multiple files, batching by backend for efficiency.

Groups paths by their target backend, calls each backend's download_files once with all paths for that backend, then merges results in original order.

PARAMETER	DESCRIPTION
paths	List of file paths to download. TYPE: list[str]

RETURNS	DESCRIPTION
list[FileDownloadResponse]	List of FileDownloadResponse objects, one per input path.
list[FileDownloadResponse]	Response order matches input order.

adownload_files async

adownload_files(paths: list[str]) -> list[FileDownloadResponse]

Async version of download_files.