In-app

Call these when your agent runs in the same process as the workspace. Every other way in ends in the same calls.

Workspace

from mirage import Workspace

class Workspace(
    mounts: dict[str, VFS],
    mode: MountMode = MountMode.READ,
    ...,
)

One terminal over the backends mounted at each prefix.

dict[str, VFS]

required

The backend at each prefix, such as {"/": RAMVFS(), "/s3": S3VFS(config)}.

MountMode

default:"MountMode.READ"

The mode a mount gets when it sets none: READ, WRITE or EXEC.

dict[str, SessionProfile]

Named profiles a session can be created under.

str

The profile a session gets when it names none.

list[Policy]

Policies every line passes through.

AskHandler

Answers an ask while the line waits. Without one, asks wait in ws.decisions.

The other parameters (cache, index, store, runtimes, clis, env, secrets, command_limits, …) are the top-level YAML keys.

from mirage import MountMode, Workspace
from mirage.vfs.ram import RAMVFS

ws = Workspace({"/": RAMVFS()}, mode=MountMode.WRITE)

ws.session

await ws.session(
    session_id, mounts=None, *, profile=None
) -> Session

The Session every call below runs as. It creates the session when the id is new and adopts it when it exists. profile and mounts apply only on create, and raise ValueError for an existing session. ws.create_session is the same call without the handle.

session = await ws.session("agent")

ws.list_sessions, ws.close_session

ws.list_sessions() -> list[SessionState]
await ws.close_session(session_id) -> None

List the sessions, or close one: its lines are cancelled and its jobs end. The default session cannot be closed.

Show SessionState

str

The working directory.

Mapping[str, str]

The exported variables.

str | None

The profile it was created under.

dict[str, MountMode] | None

Its mount modes, when narrower than the workspace’s.

int

$?.

float

Unix time.

print([s.session_id for s in ws.list_sessions()])
await ws.close_session("agent")

ws.cancel, ws.kill

await ws.cancel(session_id=None) -> int
await ws.kill(session_id=None) -> int

Stop a session’s work, or every session’s with no id, and leave the sessions open. cancel cancels the running and queued lines, whichever way they came in, and returns once they have ended; each ends with MirageAbortError. kill kills the background jobs (cmd &) and the processes runtimes started. Each answers how many it stopped.

await ws.cancel("agent")
await ws.kill()

ws.snapshot, Workspace.load

await ws.snapshot(target, *, compress=None, s3=None) -> int
await Workspace.load(source, *, s3=None, ...) -> Workspace

Save the workspace to a tar and rebuild it from one. target and source are a path or a file object, or an object key with s3=. snapshot answers the tar’s size in bytes. While it captures, new lines and file writes wait, and running ones get 30 seconds to end, else it raises OSError (EBUSY); disk mount files are streamed, never held whole. See Snapshots for what a tar holds.

await ws.snapshot("run.tar")
restored = await Workspace.load("run.tar")

ws.copy

await ws.copy() -> Workspace

A deep copy, without a tar in between. RAM and disk mounts are copied, so a write in the copy never reaches the original; a disk mount’s copy lives in a new temporary directory. Sessions, history and profiles come along. Remote backends such as S3 or Redis are shared, so a write through the copy reaches the same bucket. Policies are not copied.

fork = await ws.copy()

ws.decisions

ws.decisions.pending(session_id="") -> tuple[Decision, ...]
await ws.decisions.answer(
    decision_id, outcome, scope=Scope.ONCE, note=""
) -> None

The asks waiting for a host. Each one belongs to the session that ran the line: pending() lists every session’s, pending("agent") one session’s, and an answer never reaches another session. outcome is Outcome.ALLOW or Outcome.DENY. Scope.ONCE answers the one line, so its retry runs or is refused; Scope.SESSION allows every line the rule covers for the rest of that session.

Show Decision

str

The id to answer it by.

str

The session that ran the line.

str

The command name.

tuple[str, ...]

The words after the name, expanded.

str

tuple[str, ...]

The paths the line names.

str

The rule’s reason.

Outcome | None

The answer; None while it waits.

Scope

str

What the host said when answering.

from mirage import Outcome

for ask in ws.decisions.pending():
    await ws.decisions.answer(ask.id, Outcome.ALLOW)

ws.close

await ws.close() -> None

Stops its jobs and closes its mounts. Sessions and history stay in the workspace’s store.

Session

from mirage import Session

class Session(workspace: Workspace, session_id: str)

One session’s handle, from ws.session. Every call on it answers under the session’s profile, working directory and environment. ws.shell, ws.glob, ws.vfs and ws.tools are the same calls as the workspace’s default session.

str

SessionState

The session’s record, as ws.list_sessions answers it.

Ops

The file API, below.

MirageToolOperations

The agent tools, below.

Decisions

The workspace’s asks ledger, which this session’s questions land in.

session.shell

await session.shell(
    command, stdin=None, cwd=None, env=None,
    cancel=None, record=True,
) -> IOResult

Runs a shell line as the session. A non-zero exit is a result, not an exception.

bytes | AsyncIterator[bytes]

The line’s stdin. An async iterator streams it.

str

A working directory for this line only. A cd in the line does not leak.

dict[str, str]

Variables for this line only.

asyncio.Event

Setting it stops the line and raises MirageAbortError.

bool

default:"True"

False keeps the line out of history.

Show IOResult

int

async -> str

The output as text. materialize_stdout() answers the bytes.

async -> str

The error output as text. materialize_stderr() answers the bytes.

Refusal | None

Why the line did not run, or None.

Show Refusal

str

deny for a refusal, pending for an ask nobody answered, failed for a policy that raised.

str

The rule’s or the policy’s words.

str

The policy class that refused; empty for an ask.

str

command or operand.

str | None

The ask’s id, for pending.

result = await session.shell("wc -l", stdin=b"a\nb\n")
print(result.exit_code, await result.stdout_str())

session.glob

await session.glob(pattern) -> list[str]

The paths a pattern matches, as the session sees them.

paths = await session.glob("/src/**/*.py")

session.explain

await session.explain.shell(line) -> ShellExplanation
await session.explain.vfs.<call>(...) -> VfsExplanation

What session.shell or a session.vfs call would do, without running it: a line’s verdict and parse tree, a VFS call’s verdict and error, each with every policy’s answer. See Explain.

shell_res = await session.explain.shell("rm /data/x")
print(shell_res.exit_code, shell_res.reason)
for node in shell_res.node.children:
    if isinstance(node, CommandExplanation):
        print(node.command, [a.policy for a in node.answers])
vfs_res = await session.explain.vfs.write("/data/x", b"hi")
print(vfs_res.error, [a.policy for a in vfs_res.answers])

session.vfs

The file API, run as the session. Paths are absolute, bytes are bytes. A missing path raises FileNotFoundError and a refused one PermissionError.

Call	Answers
`read(path, offset=0, size=None)`	The bytes, whole or a range.
`write(path, data)`, `append(path, data)`	`None`
`stat(path, nofollow=False)`	A `FileStat`.
`readdir(path)`	The full path of each entry.
`exists(path)`	`True` or `False`.
`mkdir(path)`, `rmdir(path)`, `unlink(path)`	`None`
`rename(src, dst)`	`None`
`truncate(path, length)`	`None`

Show FileStat

str

FileType

FILE, DIRECTORY, SYMLINK or CHAR_DEVICE.

int | None

The bytes a read answers; None when unknown.

str | None

ISO 8601.

ContentType | None

A file’s content shape, such as text.

int | None

The permission bits.

int | str | None

str | None

The backend’s ETag and version, when it has them.

dict

Backend-specific fields.

await session.vfs.write("/notes.txt", b"hello\n")
print(await session.vfs.read("/notes.txt"))

session.tools

await session.tools.shell(command) -> ToolResult
await session.tools.read(path, offset=0, limit=2000) -> ToolResult
await session.tools.write(path, content) -> ToolResult
await session.tools.edit(
    path, old_string, new_string, replace_all=False
) -> ToolResult
await session.tools.ls(path) -> ToolResult
await session.tools.grep(pattern, path, ignore_case=False, ...)
await session.tools.glob(pattern, path="/") -> ToolResult
await session.tools.call(name, arguments) -> ToolResult
session.tools.names() -> tuple[str, ...]

The seven agent tools, for a model rather than a program: every answer is text, and a failure is is_error, never an exception. call takes a tool’s name and its JSON input, the form an agent loop gets from the model. The session has one table, shared by every caller, so write and edit refuse a file the agent has not read in full, or one that changed since it read it. names() is the tools the session’s profile leaves it, which MCP, RPC and the agent adapters offer: shell needs a command the allow list installs, ls and grep those commands, write and edit a mount the session may write.

Tool	Answers
`shell`	The line’s stdout, stderr and exit status as one text.
`read`	Numbered lines, `limit` of them from `offset`.
`write`	Writes a new file; an existing one only after a full `read`.
`edit`	Replaces `old_string` once, or everywhere with `replace_all`.
`ls`, `grep`	`ls` and `grep -rn`, as text.
`glob`	The files a pattern matches, one per line.

Show ToolResult

str

The text handed back to the agent.

bool

True when the call failed.

mirage.workspace.tools.tool_descriptions has each tool’s description and JSON schema to hand the model, and the agent adapters wire the tools into each framework. MirageToolOperations(session, stale_write_protection=False), from mirage.workspace.tools.tool_operations, is a table without the read guard.

await session.tools.call("read", {"path": "/notes.txt"})
result = await session.tools.call("edit", {
    "path": "/notes.txt",
    "old_string": "hello",
    "new_string": "hi",
})
print(result.text, result.is_error)

Getting Started

Access

AI Agents

VFS

CLIs

Watch

Runtimes

Workspace

ws.session

ws.list_sessions, ws.close_session

ws.cancel, ws.kill

ws.snapshot, Workspace.load

ws.copy

ws.decisions

ws.close

Session

session.shell

session.glob

session.explain

session.vfs

session.tools

​Workspace

​ws.session

​ws.list_sessions, ws.close_session

​ws.cancel, ws.kill

​ws.snapshot, Workspace.load

​ws.copy

​ws.decisions

​ws.close

​Session

​session.shell

​session.glob

​session.explain

​session.vfs

​session.tools

Workspace

ws.session

ws.list_sessions, ws.close_session

ws.cancel, ws.kill

ws.snapshot, Workspace.load

ws.copy

ws.decisions

ws.close

Session

session.shell

session.glob

session.explain

session.vfs

session.tools