Adding a New VFS

A VFS is a class that extends BaseVFS. Define readdir, read and stat, and ls, cat, grep, find, head, wc and the other read commands work on the service. Custom VFS lists every method, with examples for a writable VFS and for grep on the service’s own search.

Write one

from pydantic import BaseModel, ConfigDict, SecretStr

from mirage import Accessor, BaseVFS, FileStat, PathSpec, Workspace


class JiraConfig(BaseModel):
    model_config = ConfigDict(extra="forbid")

    site: str
    token: SecretStr


class JiraAccessor(Accessor):
    def __init__(self, config: JiraConfig) -> None:
        self.client = make_client(config)

    async def close(self) -> None:
        await self.client.close()


class JiraVFS(BaseVFS):
    CONFIG_CLS = JiraConfig
    accessor: JiraAccessor

    def __init__(self, config: JiraConfig) -> None:
        super().__init__(
            name="jira",
            accessor=JiraAccessor(config),
            vfs_md="Issues rendered as .json files.",
        )

    async def readdir(self, path: PathSpec) -> list[FileStat]: ...

    async def read(self, path: PathSpec) -> bytes: ...

    async def stat(self, path: PathSpec) -> FileStat: ...


ws = Workspace({"/jira/": JiraVFS(JiraConfig(site="acme", token="..."))})
await ws.shell("grep -rl outage /jira/")

readdir returns each child’s FileStat with name set to the base name. The mount keeps the listing and answers a later stat from it, so ls -l costs one request. Use path.vfs_path to address the service and path.virtual for the workspace path. Set FileStat.size to the length of the bytes read returns, or None when it is unknown.

Add functions as the service grows

Every other function is optional. Define one, and the commands that need it start working; leave it out, and those commands answer Operation not supported.

Functions	Enables
`write`, `append`, `create`, `truncate`	`>`, `>>`, `touch`, `truncate`, `sed -i`
`mkdir`, `rmdir`, `unlink`, `rm_r`, `rename`, `copy`, `dir_copy`, `setattr`	`mkdir`, `rm`, `mv`, `cp`, `chmod`, `touch -d`
`files_containing`, `lines_containing`	Search for `grep` and `rg` from the service’s own index
`first_lines`, `last_lines`, `line_count`, `checksum`, `read_tree`	Fast `head`, `tail`, `wc -l`, `md5sum`, `find` (Lines, Walks)
`delta_hook`	Notifications from the service’s change feed

The mount mode still decides whether a write may run. read may return an async iterator of bytes for a large file, and write receives one when a program’s output streams into the mount.

Plug in the service’s search

files_containing and lines_containing hand Mirage the service’s own search. grep and rg ask them which files or lines can hold the text and read only those; they still match every line themselves, so the output is the same as a full read. before_full_walk can refuse a command that would walk the whole tree. For example, an S3 bucket whose objects are indexed in Elasticsearch under their keys, by whatever indexes the bucket (such as a function on S3 events). Subclass the built-in S3VFS and answer from the index:

from elasticsearch import AsyncElasticsearch

from mirage import MountMode, PathSpec, Workspace
from mirage.utils.key_prefix import mounted_path
from mirage.vfs.s3 import S3Config, S3VFS


class IndexedS3VFS(S3VFS):
    whole_word_search = True

    def __init__(self, config: S3Config, es: AsyncElasticsearch) -> None:
        super().__init__(config)
        self.es = es

    async def files_containing(
        self, path: PathSpec, text: str, *, whole_word: bool, ignore_case: bool
    ) -> list[PathSpec] | None:
        if not whole_word:
            return None
        found = await self.es.search(
            index=self.config.bucket,
            query={"match_phrase": {"body": text}},
            size=1000,
            track_total_hits=True,
        )
        hits = found["hits"]
        if hits["total"]["value"] > len(hits["hits"]):
            return None
        return [mounted_path(path, "/" + hit["_id"]) for hit in hits["hits"]]

    async def close(self) -> None:
        if self.is_closed:
            return
        try:
            await self.es.close()
        finally:
            await super().close()


vfs = IndexedS3VFS(S3Config(bucket="docs"), AsyncElasticsearch("http://localhost:9200"))
ws = Workspace({"/s3/": (vfs, MountMode.READ)})
await ws.shell("grep -rlw outage /s3/")

grep -rlw outage /s3/ reads only the objects Elasticsearch names. Elasticsearch ignores case, so it may name more than grep prints; grep checks each line and drops the rest. grep -rl outage without -w gets None and reads every object, so outages still matches. The mount is read-only because a write through Mirage would leave the index behind. The answer is only as complete as the index: grep skips an object written elsewhere that Elasticsearch has not indexed yet, so return None while the indexer lags.

Function	Asked by	Answers
`files_containing(path, text, *, whole_word, ignore_case)`	`grep -r`, `rg` on a folder	The files under `path` that may hold `text`, or `None` to read every file
`lines_containing(path, text, *, ignore_case)`	`grep`, `rg` on one file	The lines that may hold `text`, in file order, or `None` to read the file
`before_full_walk(path, *, files)`	`tar -c`, `cp -r`, `rm -r`, `chmod -R`, `diff -r`	Nothing to let the walk run; raise `ValueError` to refuse it

Answer more files than needed, never fewer: an extra file costs one read, a missing one is a wrong answer. Return None whenever the answer may be incomplete, such as a truncated result or an index that lags writes.
A search that matches only whole words, as Elasticsearch does, returns None unless whole_word is set, and sets whole_word_search = True so the mount’s VFS.md tells the agent to use grep -w.
Set searchable to the globs the search covers, such as ("*/*.json",); a file outside them is always read.
walk_limit caps how many files one command walks, 1000 by default.

Search lists which patterns and flags reach these functions, and which built-in VFSs answer them. A service action that is not a file, such as a JQL search or adding a comment to an issue, belongs in a CLI, next to the mount.

Use it in YAML

A workspace YAML mounts the class by its file and name:

mounts:
  /jira:
    vfs: ./jira.py:JiraVFS
    config:
      site: acme
      token: ${JIRA_TOKEN}

To write vfs: jira instead, register the class once:

from mirage import register_vfs

register_vfs("jira", JiraVFS, JiraConfig)

Examples

Runnable, each with a TypeScript twin:

Example	Shows
`custom_vfs.py`	A complete backend in one file: grep on its search, a snapshot and a load
`basevfs_views.py`	A read-only VFS beside a nested mount, a symlink and a CLI
`custom_command.py`	A command of the VFS’s own
`filetype.py`	A renderer for one file type

Getting Started

Access

AI Agents

VFS

CLIs

Runtimes

Adding a New VFS

Write one

Add functions as the service grows

Plug in the service’s search

Use it in YAML

Examples

​Write one

​Add functions as the service grows

​Plug in the service’s search

​Use it in YAML

​Examples

Write one

Add functions as the service grows

Plug in the service’s search

Use it in YAML

Examples