pydantic_ai.workspaces
Workspace API, backend protocols, and implementations.
Bases: WorkspaceBackend
The workspace API tools and hooks use as ctx.workspace: the backend’s operations, relative paths, and text.
The provider backend underneath every wrapper, for access to provider-specific functionality.
Type: WorkspaceBackend
Whether this workspace refuses commands and file changes, so tools can leave those out.
Type: bool
Whether this workspace reaches an environment.
False for an UnavailableWorkspace and for a run
with no workspace attached.
Type: bool
The environment’s WorkspaceRef, None until it exists.
Type: WorkspaceRef | None
@async
def run(
command: WorkspaceCommand,
*,
shell: bool = False,
env: Mapping[str, str] | None = None,
timeout: float | None = None,
) -> CommandResult
Run a command in the working directory and wait for it.
To start elsewhere, prefix a shell command with cd <shlex-quoted dir> && .
There is no default timeout. Raises UserError if the backend can’t run commands.
CommandResult
@async
def working_dir() -> str
The default working directory, which relative paths resolve against.
@async
def resolve(path: str, *, base: str | None = None) -> str
Join path onto base (default: the working directory) and normalize it as text.
The only I/O is asking the backend for its working directory when base is omitted.
Symlinks are not followed and .. can escape base, so this confines nothing; use
realpath to learn where a path actually leads.
@async
def read_bytes(path: str) -> bytes
Read a file’s contents as bytes.
@async
def write_bytes(path: str, data: bytes) -> None
Write bytes to a file, creating missing parents and replacing existing contents.
@async
def stat(path: str) -> FileEntry
Return metadata for a file or directory.
FileEntry
@async
def list_dir(path: str) -> Sequence[FileEntry]
List the entries of a directory (non-recursive).
Sequence[FileEntry]
@async
def make_dir(path: str) -> None
Create a directory, including missing parents.
@async
def remove(path: str) -> None
Remove a file, or a directory and its contents.
@async
def exists(path: str) -> bool
Whether a file or directory exists at the path.
@async
def realpath(path: str) -> str
Follow every symlink in path in the environment, like os.path.realpath(path, strict=False).
Uses the backend’s SupportsRealpath, else readlink in
its shell. Without either, it only normalizes the path as text: symlinks are not followed, so a
path check built on it can be escaped through a link.
.. climbs from a symlink’s target, as it does for commands. File methods open
resolve(path), which collapses .. as text, so
realpath(await ws.resolve(path)) names the file they open.
@async
def read_text(path: str, *, encoding: str = 'utf-8') -> str
Read a file as text; undecodable bytes raise UnicodeDecodeError.
@async
def write_text(path: str, content: str, *, encoding: str = 'utf-8') -> None
Write text to a file.
Bases: Workspace
A workspace facade that composes another workspace.
Bases: WrapperWorkspace
A Workspace that allows reads and refuses commands and file changes.
Commands are refused too, since they could change files. This restricts the workspace API; it is not isolation.
Bases: WorkspaceBackend, SupportsCommands, SupportsFilesystem, SupportsRealpath
Run commands as subprocesses on this machine and use its filesystem (POSIX only).
This isolates nothing: commands and absolute paths reach anywhere this process can. Commands
inherit only PATH, HOME and locale (LANG, LC_ALL, LC_CTYPE), so they find the host’s
tools and use its text encoding without inheriting arbitrary secrets.
Background jobs outlive run(); redirect their output to avoid waiting up to two seconds
for inherited output pipes. The caller manages those jobs when the host exits.
The directory is the environment: its ref
exists from construction, and the first operation raises
WorkspaceUnavailableError if it is missing.
working_dir : str | Path
Where commands start and relative paths resolve; ~ is expanded and a relative
path is taken from the current directory. The caller creates and removes it.
Environment variables for every command, on top of the inherited PATH, HOME, LANG,
LC_ALL and LC_CTYPE; the per-call env goes on top.
WorkspaceRef(provider='local', id=<absolute working_dir>), available from construction.
Type: WorkspaceRef
@async
def write_bytes(path: str, data: bytes) -> None
Write bytes to a file, creating missing parents and writing through an existing symlink.
The file is rewritten in place. Reads and writes of one file through this backend are serialized; other processes and commands can still interleave with them.
Bases: WorkspaceBackend, SupportsCommands
A WorkspaceBackend whose operations raise WorkspaceUnavailableError with a configured reason.
Always None: there is no environment to name, so nothing can be reconnected to later.
Type: None
Serializable identity of a workspace environment, without credentials.
Pass it to a later run as workspace= to continue in that environment.
Provider that owns the environment.
Type: str
Provider-specific identifier for the environment.
Type: str
Bases: Protocol
The environment an agent run works in; any object with these members conforms.
Built without I/O from configuration and an optional WorkspaceRef;
the first operation creates the environment, or attaches to the one the ref names. The backend never
tears it down; whoever holds the ref does. See the module docstring for the full contract.
The environment’s identity: None until a fresh one is created, then set for good.
Type: WorkspaceRef | None
@async
def working_dir() -> str
The stable working directory for this environment: absolute, symlinks resolved, no ./.. segments.
Bases: Protocol
Optional command execution.
Without it, Workspace.run raises UserError.
@async
def run(
command: WorkspaceCommand,
*,
shell: bool = False,
env: Mapping[str, str] | None = None,
timeout: float | None = None,
) -> CommandResult
Execute a command with stdin at EOF, returning complete output or raising an error.
Undecodable stdout/stderr bytes are replaced with U+FFFD, never dropped.
The command starts in working_dir(). A missing argv program exits 127.
If the environment is destroyed while the command runs, raise WorkspaceUnavailableError;
a command killed by a signal in a live environment returns its exit code.
On timeout or cancellation, stop the foreground process tree on a best-effort basis;
background jobs may continue if they detach. Return when the direct command exits,
after at most a short output-drain grace even if a background child keeps stdout open.
CommandResult
An argv sequence, or a shell string with shell=True; a mismatch raises TypeError.
Workspace.run rejects an empty argv with ValueError.
shell : bool Default: False
Whether to interpret command with the workspace’s shell.
Extra environment variables, layered over the backend’s own.
A positive finite number of seconds before
WorkspaceTimeoutError; no timeout by default.
Invalid values raise ValueError.
Bases: Protocol
Optional native file access; without it, Workspace uses the shell.
Paths are absolute POSIX paths. A missing path raises FileNotFoundError (except in exists),
and reading a directory raises IsADirectoryError.
@async
def read_bytes(path: str) -> bytes
Read a file’s contents as bytes.
@async
def write_bytes(path: str, data: bytes) -> None
Write bytes to a file, creating missing parents and writing through an existing symlink.
@async
def stat(path: str) -> FileEntry
Return metadata for a file or directory.
FileEntry
@async
def list_dir(path: str) -> Sequence[FileEntry]
List the entries of a directory (non-recursive).
Sequence[FileEntry]
@async
def make_dir(path: str) -> None
Create a directory, including missing parents (mkdir -p semantics).
@async
def remove(path: str) -> None
Remove a file, or a directory and its contents.
Refuses the working directory and its ancestors with ValueError; a symlink is removed itself.
@async
def exists(path: str) -> bool
Whether a file or directory exists at the path.
Bases: Protocol
Native symlink resolution for path boundaries such as harness FileSystem’s root_dir.
Without it, Workspace.realpath uses the backend’s shell, and on a backend without commands it
only normalizes the path as text, so path checks cannot see through symlinks. Implement it on a
filesystem-only backend whose storage can hold symlinks.
@async
def realpath(path: str) -> str
Resolve inside the environment, like os.path.realpath(path, strict=False).
Relative link targets start in the link’s directory; .. after a link climbs from its
target. Keep missing components as written. A loop must not hang; a returned path through a loop
must stay inside the directory holding it (or raise OSError). Return an absolute, normalized path.
The result of a completed command.
The real exit code of the process. Non-zero is a normal result, not an error.
Type: int
Captured standard output.
Type: str
Captured standard error.
Type: str
Metadata about a file or directory.
Base name of the entry.
Type: str
Absolute POSIX path of the entry inside the workspace.
Type: str
Whether the entry is a directory, following a symlink to its target.
Type: bool
Size in bytes for a regular file when known; None is allowed (the shell fallback can measure it).
Bases: RuntimeError
The workspace layer deliberately failed an operation.
Bases: WorkspaceError
The environment is gone or unusable (terminated, expired, not found), so retrying can’t succeed.
Bases: WorkspaceError, TimeoutError
A command exceeded its timeout=; stdout/stderr hold partial output, like subprocess.TimeoutExpired.
Bases: WorkspaceError
A command exceeded its output cap; stdout and stderr hold their captured beginnings.
Bases: WorkspaceError, PermissionError
A mutation was refused because the workspace is read-only.
Raised by ReadOnlyWorkspace and wrappers like it.
An argv sequence (['python', '-c', 'print(1)']), or a shell string with shell=True.
Type: TypeAlias Default: str | Sequence[str]
The conformance suite for Pydantic AI workspace backends.
Subclass WorkspaceBackendSuite in a pytest module and provide a backend fixture: each test checks
one rule of the WorkspaceBackend contract, the same rules the built-in and provider backends pass.
Requires pytest and the anyio pytest plugin.
Subclass in your test suite and provide the backend fixture.
Each test checks one rule of the backend contract. Command rules skip for a backend without
SupportsCommands; filesystem rules run through Workspace, so a
command-only backend is checked on the file operations derived through its shell. The reattach
rules need the optional fixtures below and skip without them.
def fresh_backend() -> Callable[[], WorkspaceBackend] | None
Build an uninitialized backend to check concurrent first use, if supported.
Callable[[], WorkspaceBackend] | None
def destructive_backend(
fresh_backend: Callable[[], WorkspaceBackend] | None,
) -> Callable[[], WorkspaceBackend] | None
Build an independent environment for destruction; defaults to fresh_backend.
Callable[[], WorkspaceBackend] | None
def attach_backend() -> Callable[[WorkspaceRef], WorkspaceBackend] | None
Build a second backend that attaches to ref. Enables the reattach rules.
Callable[[WorkspaceRef], WorkspaceBackend] | None
def destroy_environment() -> Callable[[WorkspaceBackend], Awaitable[None]] | None
Destroy the environment behind backend. Enables the reattach-after-destroy rule.
Callable[[WorkspaceBackend], Awaitable[None]] | None
@async
def test_command_form_must_match_shell(backend: WorkspaceBackend) -> None
A string needs shell=True and an argv sequence needs shell=False; a mismatch is a TypeError.
@async
def test_stdin_is_at_eof(backend: WorkspaceBackend) -> None
Noninteractive commands never wait for input from the caller.
@async
def test_command_output_is_complete(backend: WorkspaceBackend) -> None
If output cannot be collected in full, the backend must raise rather than return a truncated success.
def can_detect_exit_with_inherited_output_pipes() -> bool
Override only if the SDK cannot report exit independently of pipe EOF (E2B currently cannot).
@async
def test_result_reports_exit_code_stdout_and_stderr(backend: WorkspaceBackend) -> None
A non-zero exit is a normal result, not an error.
@async
def test_a_missing_program_exits_127(backend: WorkspaceBackend) -> None
Like sh, a program that doesn’t exist is a normal result with exit code 127, not an error.
@async
def test_working_dir_is_canonical(backend: WorkspaceBackend) -> None
Absolute, symlinks resolved, no ./..: the directory commands actually start in.
@async
def test_large_file_round_trip(backend: WorkspaceBackend) -> None
A shell-derived filesystem must page reads rather than hit a command-output cap.
def has_real_posix_shell() -> bool
Only a test double with no POSIX process/filesystem can opt out.
def enforces_parent_file_errors() -> bool
Opt out only for an in-memory test double without real path traversal.
def filesystem_honors_shell_permissions() -> bool
Override only for provider file APIs that bypass the command user’s permissions (e.g. E2B envd).
@async
def test_native_realpath_keeps_the_working_dir_and_missing_names(
backend: WorkspaceBackend,
) -> None
Checked without commands, so a filesystem-only backend’s realpath is covered too.
@async
def test_a_backend_attached_by_ref_reaches_the_same_environment(
backend: WorkspaceBackend,
attach_backend: Callable[[WorkspaceRef], WorkspaceBackend] | None,
) -> None
Durable execution rebuilds the backend from its ref for every call, so this is what it relies on.