Backends API

Async adaptation

pydantic_ai_backends.adapter.ensure_async(backend, *, executor=None)

Return an async backend, wrapping sync ones as needed.

An already-async backend is passed through untouched, which is the point: a thread adapter around async code is not merely wasteful but a deadlock waiting to happen, since each call then occupies a worker thread that has to hop back onto the event loop. Two things mark a backend as already async, in this order:

• Subclassing :class:~pydantic_ai_backends.AsyncBaseSandbox. Unambiguous, and the recommended way to write one.
• read_bytes being a coroutine function. The fallback, for a backend that implements the protocol without inheriting from anything. It has to be a method-shape check rather than isinstance against AsyncBackendProtocol, because a runtime-checkable Protocol compares method names — and a sync backend has exactly the same ones.

Parameters:

Name	Type	Description	Default
backend	BackendProtocol | AsyncBackendProtocol	Sync or async backend.	required
executor	Executor | None	Thread pool for a newly created adapter. Ignored when backend is already async or already adapted — this function is idempotent on adapters, so wrap once yourself (AsyncSandboxAdapter(backend, executor=...)) and pass the adapter around when every call site should share one pool.	None

Returns:

Type	Description
AsyncBackendProtocol	An async view of backend.

Source code in src/pydantic_ai_backends/adapter.py

def ensure_async(
    backend: BackendProtocol | AsyncBackendProtocol,
    *,
    executor: Executor | None = None,
) -> AsyncBackendProtocol:
    """Return an async backend, wrapping sync ones as needed.

    An already-async backend is passed through untouched, which is the point: a
    thread adapter around async code is not merely wasteful but a deadlock
    waiting to happen, since each call then occupies a worker thread that has to
    hop back onto the event loop. Two things mark a backend as already async, in
    this order:

    - **Subclassing :class:`~pydantic_ai_backends.AsyncBaseSandbox`.**
      Unambiguous, and the recommended way to write one.
    - **`read_bytes` being a coroutine function.** The fallback, for a backend
      that implements the protocol without inheriting from anything. It has to
      be a method-shape check rather than `isinstance` against
      `AsyncBackendProtocol`, because a runtime-checkable `Protocol` compares
      method *names* — and a sync backend has exactly the same ones.

    Args:
        backend: Sync or async backend.
        executor: Thread pool for a newly created adapter. Ignored when
            `backend` is already async or already adapted — this function is
            idempotent on adapters, so wrap once yourself
            (`AsyncSandboxAdapter(backend, executor=...)`) and pass the adapter
            around when every call site should share one pool.

    Returns:
        An async view of `backend`.
    """
    if isinstance(backend, AsyncBackendAdapter):
        return backend
    if is_async_backend(backend):
        return cast("AsyncBackendProtocol", backend)

    candidate: Any = backend
    if hasattr(candidate, "execute_background"):
        return AsyncBackgroundSandboxAdapter(cast("SandboxProtocol", backend), executor=executor)
    if hasattr(candidate, "execute"):
        return AsyncSandboxAdapter(cast("SandboxProtocol", backend), executor=executor)
    return AsyncBackendAdapter(cast("BackendProtocol", backend), executor=executor)

pydantic_ai_backends.adapter.is_async_backend(backend)

Whether a backend is already asynchronous and must not be thread-wrapped.

Public because it is the contract a third-party backend is judged by, and getting it wrong is expensive in a way that is hard to debug: a natively async backend mistaken for a sync one gets wrapped in a thread adapter, which calls its coroutine function in a worker thread and hands the caller the resulting coroutine object where bytes were expected — no exception, just nonsense.

Parameters:

Name	Type	Description	Default
backend	object	The backend to classify.	required

Returns:

Type	Description
bool	Whether the backend implements the async protocol directly.

Source code in src/pydantic_ai_backends/adapter.py

def is_async_backend(backend: object) -> bool:
    """Whether a backend is already asynchronous and must not be thread-wrapped.

    Public because it is the contract a third-party backend is judged by, and
    getting it wrong is expensive in a way that is hard to debug: a natively
    async backend mistaken for a sync one gets wrapped in a thread adapter,
    which calls its coroutine function in a worker thread and hands the caller
    the resulting coroutine object where bytes were expected — no exception,
    just nonsense.

    Args:
        backend: The backend to classify.

    Returns:
        Whether the backend implements the async protocol directly.
    """
    from pydantic_ai_backends.backends.base import AsyncBaseSandbox

    if isinstance(backend, AsyncBaseSandbox):
        return True
    # Any of the names `read_bytes` is reachable under, since the adapter accepts
    # the legacy one too: a backend async everywhere but spelling it the old way
    # would otherwise be classified as sync and quietly thread-wrapped.
    return any(
        inspect.iscoroutinefunction(getattr(backend, name, None))
        for name in ("read_bytes", LEGACY_READ_BYTES)
    )

Base classes

Subclass one of these to write your own sandbox: implement execute and edit and every other file operation is derived from shell commands. See Writing your own backend for which one to pick.

pydantic_ai_backends.backends.base.BaseSandbox

Bases: _SandboxIdentity, ABC

Base class for synchronous sandboxes that expose a shell.

Parameters:

Name	Type	Description	Default
sandbox_id	str | None	Unique identifier for this sandbox. Generated when omitted.	None

Source code in src/pydantic_ai_backends/backends/base.py

class BaseSandbox(_SandboxIdentity, ABC):
    """Base class for synchronous sandboxes that expose a shell.

    Args:
        sandbox_id: Unique identifier for this sandbox. Generated when omitted.
    """

    def start(self) -> None:
        """Start the sandbox eagerly.

        The default is a no-op, since sandboxes start on first use.
        """

    def is_alive(self) -> bool:
        """Whether the sandbox is running and responsive.

        **Override this.** The default answer is "no", and `SessionManager` reads
        it as "replace the sandbox" — so a subclass that never overrides it has a
        sandbox built, stopped and rebuilt on every single operation. It defaults
        to `False` rather than `True` because a sandbox wrongly believed alive is
        the harder failure to diagnose, but neither default is right for a real
        sandbox: answer it.
        """
        return False

    def stop(self, purge: bool = False) -> None:
        """Stop and clean up the sandbox.

        Args:
            purge: Also discard what the sandbox accumulated — its filesystem,
                and any workspace kept for it outside the container. Left off,
                the sandbox ends while its files survive for the next attach;
                turned on, the thing the sandbox belonged to is gone for good.

                Accepted by every sandbox so one call site can end any of them,
                and the subclasses that cannot tell the two apart say so: a
                Daytona sandbox and a Kubernetes pod are deleted either way,
                because neither keeps a filesystem this library can reattach to.
        """

    @abstractmethod
    def execute(self, command: str, timeout: int | None = None) -> ExecuteResponse:
        """Run a command in the sandbox.

        Args:
            command: Command to execute.
            timeout: Maximum execution time in seconds.
        """
        ...

    @abstractmethod
    def edit(
        self, path: str, old_string: str, new_string: str, replace_all: bool = False
    ) -> EditResult:
        """Edit a file by replacing a string.

        Args:
            path: File path to edit.
            old_string: String to find and replace.
            new_string: Replacement string.
            replace_all: Replace every occurrence instead of only the first.
        """
        ...

    def exists(self, path: str) -> bool:
        """Whether `path` is a regular file, via `test -f`."""
        return self.execute(_shell.exists_command(path), timeout=5).exit_code == 0

    def ls_info(self, path: str) -> list[FileInfo]:
        """List one directory using `ls -la`."""
        command = _shell.ls_command(path)
        return _shell.parse_ls(self.execute(command, timeout=FILE_OP_TIMEOUT), path)

    def read_bytes(self, path: str) -> bytes:
        """Read a whole file with `cat`, or `b""` on any failure."""
        command = _shell.read_bytes_command(path)
        return _shell.parse_read_bytes(self.execute(command, timeout=FILE_OP_TIMEOUT))

    def read(self, path: str, offset: int = 0, limit: int = 2000) -> str:
        """Read a slice of a file, numbered by its real line positions."""
        command = _shell.read_command(path, offset, limit)
        return _shell.parse_read(self.execute(command, timeout=FILE_OP_TIMEOUT))

    def write(self, path: str, content: str | bytes) -> WriteResult:
        """Write a file, carrying the content base64-encoded."""
        command = _shell.write_command(path, content)
        return _shell.parse_write(self.execute(command, timeout=FILE_OP_TIMEOUT), path)

    def glob_info(self, pattern: str, path: str = "/") -> list[FileInfo]:
        """Match files with `find`."""
        command = _shell.glob_command(pattern, path)
        return _shell.parse_glob(self.execute(command, timeout=SEARCH_TIMEOUT))

    def grep_raw(
        self,
        pattern: str,
        path: str | None = None,
        glob: str | None = None,
        ignore_hidden: bool = True,
    ) -> list[GrepMatch] | str:
        """Search file contents with `grep`."""
        command = _shell.grep_command(pattern, path, glob, ignore_hidden)
        return _shell.parse_grep(self.execute(command, timeout=SEARCH_TIMEOUT))

start()

Start the sandbox eagerly.

The default is a no-op, since sandboxes start on first use.

Source code in src/pydantic_ai_backends/backends/base.py

def start(self) -> None:
    """Start the sandbox eagerly.

    The default is a no-op, since sandboxes start on first use.
    """

is_alive()

Whether the sandbox is running and responsive.

Override this. The default answer is "no", and SessionManager reads it as "replace the sandbox" — so a subclass that never overrides it has a sandbox built, stopped and rebuilt on every single operation. It defaults to False rather than True because a sandbox wrongly believed alive is the harder failure to diagnose, but neither default is right for a real sandbox: answer it.

Source code in src/pydantic_ai_backends/backends/base.py

def is_alive(self) -> bool:
    """Whether the sandbox is running and responsive.

    **Override this.** The default answer is "no", and `SessionManager` reads
    it as "replace the sandbox" — so a subclass that never overrides it has a
    sandbox built, stopped and rebuilt on every single operation. It defaults
    to `False` rather than `True` because a sandbox wrongly believed alive is
    the harder failure to diagnose, but neither default is right for a real
    sandbox: answer it.
    """
    return False

stop(purge=False)

Stop and clean up the sandbox.

Parameters:

Name	Type	Description	Default
purge	bool	Also discard what the sandbox accumulated — its filesystem, and any workspace kept for it outside the container. Left off, the sandbox ends while its files survive for the next attach; turned on, the thing the sandbox belonged to is gone for good. Accepted by every sandbox so one call site can end any of them, and the subclasses that cannot tell the two apart say so: a Daytona sandbox and a Kubernetes pod are deleted either way, because neither keeps a filesystem this library can reattach to.	False

Source code in src/pydantic_ai_backends/backends/base.py

def stop(self, purge: bool = False) -> None:
    """Stop and clean up the sandbox.

    Args:
        purge: Also discard what the sandbox accumulated — its filesystem,
            and any workspace kept for it outside the container. Left off,
            the sandbox ends while its files survive for the next attach;
            turned on, the thing the sandbox belonged to is gone for good.

            Accepted by every sandbox so one call site can end any of them,
            and the subclasses that cannot tell the two apart say so: a
            Daytona sandbox and a Kubernetes pod are deleted either way,
            because neither keeps a filesystem this library can reattach to.
    """

execute(command, timeout=None) abstractmethod

Run a command in the sandbox.

Parameters:

Name	Type	Description	Default
command	str	Command to execute.	required
timeout	int | None	Maximum execution time in seconds.	None

Source code in src/pydantic_ai_backends/backends/base.py

@abstractmethod
def execute(self, command: str, timeout: int | None = None) -> ExecuteResponse:
    """Run a command in the sandbox.

    Args:
        command: Command to execute.
        timeout: Maximum execution time in seconds.
    """
    ...

edit(path, old_string, new_string, replace_all=False) abstractmethod

Edit a file by replacing a string.

Parameters:

Name	Type	Description	Default
path	str	File path to edit.	required
old_string	str	String to find and replace.	required
new_string	str	Replacement string.	required
replace_all	bool	Replace every occurrence instead of only the first.	False

Source code in src/pydantic_ai_backends/backends/base.py

@abstractmethod
def edit(
    self, path: str, old_string: str, new_string: str, replace_all: bool = False
) -> EditResult:
    """Edit a file by replacing a string.

    Args:
        path: File path to edit.
        old_string: String to find and replace.
        new_string: Replacement string.
        replace_all: Replace every occurrence instead of only the first.
    """
    ...

pydantic_ai_backends.backends.base.AsyncBaseSandbox

Bases: _SandboxIdentity, ABC

Base class for natively asynchronous sandboxes that expose a shell.

Subclass this when the sandbox is reached over an async transport — asyncssh, an async HTTP client, any async SDK. Implement execute and edit as coroutines and every other operation is derived from shell commands, exactly as the synchronous base does.

Subclassing this rather than wrapping async code in a synchronous facade is not a style preference. ensure_async cannot see through a facade: it wraps the facade in a thread adapter, so each call runs on a worker thread that has to hop back onto the event loop to reach the real async code. A sandbox whose own recovery path also needs a thread — reprovisioning a dead container, say — then waits for a thread that is waiting for the loop, and starves the pool for every other agent sharing it. Being async all the way down means ensure_async passes the backend through untouched and the toolset awaits it directly.

Recognised by ensure_async through this base class, so no method-shape sniffing is involved:

Python

from pydantic_ai_backends import AsyncBaseSandbox, ExecuteResponse


class SSHSandbox(AsyncBaseSandbox):
    async def execute(self, command: str, timeout: int | None = None) -> ExecuteResponse:
        result = await self._connection.run(command, timeout=timeout)
        return ExecuteResponse(output=result.stdout, exit_code=result.exit_status)

    async def edit(self, path, old_string, new_string, replace_all=False) -> EditResult:
        ...

Failures are returned, never raised — see :class:~pydantic_ai_backends.protocol.AsyncBackendProtocol for the contract every method here is held to.

Parameters:

Name	Type	Description	Default
sandbox_id	str | None	Unique identifier for this sandbox. Generated when omitted.	None

Source code in src/pydantic_ai_backends/backends/base.py

class AsyncBaseSandbox(_SandboxIdentity, ABC):
    """Base class for natively asynchronous sandboxes that expose a shell.

    Subclass this when the sandbox is reached over an async transport — asyncssh,
    an async HTTP client, any async SDK. Implement `execute` and `edit` as
    coroutines and every other operation is derived from shell commands, exactly
    as the synchronous base does.

    Subclassing this rather than wrapping async code in a synchronous facade is
    not a style preference. `ensure_async` cannot see through a facade: it
    wraps the facade in a thread adapter, so each call runs on a worker thread
    that has to hop back onto the event loop to reach the real async code. A
    sandbox whose own recovery path also needs a thread — reprovisioning a dead
    container, say — then waits for a thread that is waiting for the loop, and
    starves the pool for every other agent sharing it. Being async all the way
    down means `ensure_async` passes the backend through untouched and the
    toolset awaits it directly.

    Recognised by `ensure_async` through this base class, so no method-shape
    sniffing is involved:

    ```python
    from pydantic_ai_backends import AsyncBaseSandbox, ExecuteResponse


    class SSHSandbox(AsyncBaseSandbox):
        async def execute(self, command: str, timeout: int | None = None) -> ExecuteResponse:
            result = await self._connection.run(command, timeout=timeout)
            return ExecuteResponse(output=result.stdout, exit_code=result.exit_status)

        async def edit(self, path, old_string, new_string, replace_all=False) -> EditResult:
            ...
    ```

    Failures are returned, never raised — see
    :class:`~pydantic_ai_backends.protocol.AsyncBackendProtocol` for the contract
    every method here is held to.

    Args:
        sandbox_id: Unique identifier for this sandbox. Generated when omitted.
    """

    async def start(self) -> None:
        """Start the sandbox eagerly.

        The default is a no-op, since sandboxes start on first use.
        """

    async def is_alive(self) -> bool:
        """Whether the sandbox is running and responsive.

        **Override this.** The default answer is "no", and `SessionManager` reads
        it as "replace the sandbox" — so a subclass that never overrides it has a
        sandbox built, stopped and rebuilt on every single operation. It defaults
        to `False` rather than `True` because a sandbox wrongly believed alive is
        the harder failure to diagnose, but neither default is right for a real
        sandbox: answer it.
        """
        return False

    async def stop(self, purge: bool = False) -> None:
        """Stop and clean up the sandbox.

        Args:
            purge: Also discard what the sandbox accumulated. See
                :meth:`BaseSandbox.stop`.
        """

    @abstractmethod
    async def execute(self, command: str, timeout: int | None = None) -> ExecuteResponse:
        """Run a command in the sandbox.

        Args:
            command: Command to execute.
            timeout: Maximum execution time in seconds.
        """
        ...

    @abstractmethod
    async def edit(
        self, path: str, old_string: str, new_string: str, replace_all: bool = False
    ) -> EditResult:
        """Edit a file by replacing a string.

        Args:
            path: File path to edit.
            old_string: String to find and replace.
            new_string: Replacement string.
            replace_all: Replace every occurrence instead of only the first.
        """
        ...

    async def exists(self, path: str) -> bool:
        """Whether `path` is a regular file, via `test -f`."""
        result = await self.execute(_shell.exists_command(path), timeout=5)
        return result.exit_code == 0

    async def ls_info(self, path: str) -> list[FileInfo]:
        """List one directory using `ls -la`."""
        result = await self.execute(_shell.ls_command(path), timeout=FILE_OP_TIMEOUT)
        return _shell.parse_ls(result, path)

    async def read_bytes(self, path: str) -> bytes:
        """Read a whole file with `cat`, or `b""` on any failure."""
        command = _shell.read_bytes_command(path)
        return _shell.parse_read_bytes(await self.execute(command, timeout=FILE_OP_TIMEOUT))

    async def read(self, path: str, offset: int = 0, limit: int = 2000) -> str:
        """Read a slice of a file, numbered by its real line positions."""
        command = _shell.read_command(path, offset, limit)
        return _shell.parse_read(await self.execute(command, timeout=FILE_OP_TIMEOUT))

    async def write(self, path: str, content: str | bytes) -> WriteResult:
        """Write a file, carrying the content base64-encoded."""
        result = await self.execute(_shell.write_command(path, content), timeout=FILE_OP_TIMEOUT)
        return _shell.parse_write(result, path)

    async def glob_info(self, pattern: str, path: str = "/") -> list[FileInfo]:
        """Match files with `find`."""
        command = _shell.glob_command(pattern, path)
        return _shell.parse_glob(await self.execute(command, timeout=SEARCH_TIMEOUT))

    async def grep_raw(
        self,
        pattern: str,
        path: str | None = None,
        glob: str | None = None,
        ignore_hidden: bool = True,
    ) -> list[GrepMatch] | str:
        """Search file contents with `grep`."""
        command = _shell.grep_command(pattern, path, glob, ignore_hidden)
        return _shell.parse_grep(await self.execute(command, timeout=SEARCH_TIMEOUT))

start() async

Start the sandbox eagerly.

The default is a no-op, since sandboxes start on first use.

Source code in src/pydantic_ai_backends/backends/base.py

async def start(self) -> None:
    """Start the sandbox eagerly.

    The default is a no-op, since sandboxes start on first use.
    """

is_alive() async

Whether the sandbox is running and responsive.

Override this. The default answer is "no", and SessionManager reads it as "replace the sandbox" — so a subclass that never overrides it has a sandbox built, stopped and rebuilt on every single operation. It defaults to False rather than True because a sandbox wrongly believed alive is the harder failure to diagnose, but neither default is right for a real sandbox: answer it.

Source code in src/pydantic_ai_backends/backends/base.py

async def is_alive(self) -> bool:
    """Whether the sandbox is running and responsive.

    **Override this.** The default answer is "no", and `SessionManager` reads
    it as "replace the sandbox" — so a subclass that never overrides it has a
    sandbox built, stopped and rebuilt on every single operation. It defaults
    to `False` rather than `True` because a sandbox wrongly believed alive is
    the harder failure to diagnose, but neither default is right for a real
    sandbox: answer it.
    """
    return False

stop(purge=False) async

Stop and clean up the sandbox.

Parameters:

Name	Type	Description	Default
purge	bool	Also discard what the sandbox accumulated. See :meth:BaseSandbox.stop.	False

Source code in src/pydantic_ai_backends/backends/base.py

async def stop(self, purge: bool = False) -> None:
    """Stop and clean up the sandbox.

    Args:
        purge: Also discard what the sandbox accumulated. See
            :meth:`BaseSandbox.stop`.
    """

execute(command, timeout=None) abstractmethod async

Run a command in the sandbox.

Parameters:

Name	Type	Description	Default
command	str	Command to execute.	required
timeout	int | None	Maximum execution time in seconds.	None

Source code in src/pydantic_ai_backends/backends/base.py

@abstractmethod
async def execute(self, command: str, timeout: int | None = None) -> ExecuteResponse:
    """Run a command in the sandbox.

    Args:
        command: Command to execute.
        timeout: Maximum execution time in seconds.
    """
    ...

edit(path, old_string, new_string, replace_all=False) abstractmethod async

Edit a file by replacing a string.

Parameters:

Name	Type	Description	Default
path	str	File path to edit.	required
old_string	str	String to find and replace.	required
new_string	str	Replacement string.	required
replace_all	bool	Replace every occurrence instead of only the first.	False

Source code in src/pydantic_ai_backends/backends/base.py

@abstractmethod
async def edit(
    self, path: str, old_string: str, new_string: str, replace_all: bool = False
) -> EditResult:
    """Edit a file by replacing a string.

    Args:
        path: File path to edit.
        old_string: String to find and replace.
        new_string: Replacement string.
        replace_all: Replace every occurrence instead of only the first.
    """
    ...

LocalBackend

pydantic_ai_backends.backends.local.LocalBackend

Local filesystem backend with optional shell execution.

File operations are native Python; execute shells out. Both are confined to allowed_directories, and can be narrowed further with a permission ruleset.

Example

Python

from pydantic_ai_backends import LocalBackend

backend = LocalBackend(root_dir="/workspace")
backend.write("/src/app.py", "print('hello')")
result = backend.execute("python /src/app.py")

readonly = LocalBackend(
    allowed_directories=["/home/user/project"],
    enable_execute=False,
)

Source code in src/pydantic_ai_backends/backends/local.py

class LocalBackend:
    """Local filesystem backend with optional shell execution.

    File operations are native Python; `execute` shells out. Both are confined
    to `allowed_directories`, and can be narrowed further with a permission
    ruleset.

    Example:
        ```python
        from pydantic_ai_backends import LocalBackend

        backend = LocalBackend(root_dir="/workspace")
        backend.write("/src/app.py", "print('hello')")
        result = backend.execute("python /src/app.py")

        readonly = LocalBackend(
            allowed_directories=["/home/user/project"],
            enable_execute=False,
        )
        ```
    """

    def __init__(
        self,
        root_dir: str | Path | None = None,
        allowed_directories: list[str] | None = None,
        enable_execute: bool = True,
        sandbox_id: str | None = None,
        permissions: PermissionRuleset | None = None,
        ask_callback: AskCallback | None = None,
        ask_fallback: AskFallback = "error",
    ):
        """Initialize the backend.

        Args:
            root_dir: Base directory for file operations. Defaults to the first
                allowed directory, or the current working directory.
            allowed_directories: Directories file operations are confined to,
                resolved to absolute paths and created when missing. When
                omitted, only `root_dir` is reachable.
            enable_execute: Whether shell execution is available.
            sandbox_id: Unique identifier for this backend instance.
            permissions: Optional ruleset applied after the allowed-directory
                check passes.
            ask_callback: Async callback for "ask" actions, receiving
                `(operation, target, reason)` and returning whether to allow.
            ask_fallback: What an unanswerable "ask" does — `"deny"` refuses the
                operation, `"error"` raises.
        """
        self._id = sandbox_id or str(uuid.uuid4())
        self._enable_execute = enable_execute
        self._permissions = permissions

        self._allowed_directories = [Path(d).resolve() for d in allowed_directories or []]
        for directory in self._allowed_directories:
            directory.mkdir(parents=True, exist_ok=True)

        if root_dir is not None:
            self._root = Path(root_dir).resolve()
        elif self._allowed_directories:
            self._root = self._allowed_directories[0]
        else:
            self._root = Path.cwd()
        self._root.mkdir(parents=True, exist_ok=True)

        if not self._allowed_directories:
            self._allowed_directories = [self._root]

        self._guard = (
            PermissionGuard(permissions, self._root, ask_callback, ask_fallback)
            if permissions is not None
            else None
        )
        self._background = BackgroundProcesses(self._root)

    @property
    def id(self) -> str:
        """Unique identifier for this backend."""
        return self._id

    @property
    def root_dir(self) -> Path:
        """Directory relative paths resolve against, and commands run in."""
        return self._root

    @property
    def execute_enabled(self) -> bool:
        """Whether shell execution is enabled."""
        return self._enable_execute

    @property
    def permissions(self) -> PermissionRuleset | None:
        """The permission ruleset for this backend, if any."""
        return self._permissions

    @property
    def permission_checker(self) -> PermissionChecker | None:
        """The permission checker for this backend, if any."""
        return self._guard.checker if self._guard else None

    def _denial_reason(self, operation: PermissionOperation, target: str) -> str | None:
        return self._guard.denial_reason(operation, target) if self._guard else None

    def _is_denied(self, operation: PermissionOperation, target: str) -> bool:
        return self._guard is not None and self._guard.is_denied(operation, target)

    def _resolve(self, path: str) -> Path:
        """Resolve `path` inside the allowed directories.

        Args:
            path: Absolute path, or one relative to the root directory.

        Raises:
            PermissionError: If the resolved path escapes every allowed
                directory — including via `..` or a symlink, since resolution
                happens before the check.
        """
        candidate = Path(path)
        resolved = candidate.resolve() if candidate.is_absolute() else (self._root / path).resolve()

        for allowed in self._allowed_directories:
            if resolved == allowed or allowed in resolved.parents:
                return resolved

        allowed_str = ", ".join(str(d) for d in self._allowed_directories)
        raise PermissionError(
            f"Access denied: '{path}' is outside allowed directories ({allowed_str})"
        )

    # ── Files ──────────────────────────────────────────────────────────

    def exists(self, path: str) -> bool:
        """Whether `path` is a regular file inside the allowed directories.

        Missing files, directories, paths outside the allowed set and paths the
        filesystem refuses to stat at all (embedded null bytes, `ELOOP`, a name
        that is too long) are all `False`. Use `ls_info` when the reason matters.
        """
        try:
            return self._resolve(path).is_file()
        except (PermissionError, ValueError, OSError):
            return False

    def ls_info(self, path: str) -> list[FileInfo]:
        """List one directory, omitting entries denied for "ls".

        An "ask" counts as visible, since a listing cannot prompt.
        """
        try:
            full_path = self._resolve(path)
        except PermissionError:
            return []

        if self._is_denied("ls", str(full_path)) or not full_path.exists():
            return []

        if full_path.is_file():
            return [_entry_info(full_path)]

        results: list[FileInfo] = []
        try:
            for entry in full_path.iterdir():
                try:
                    self._resolve(str(entry))
                    if self._is_denied("ls", str(entry)):
                        continue
                    info = _entry_info(entry)
                except PermissionError:
                    continue
                except OSError:
                    # Per entry, not per listing, matching `glob_info`: a file
                    # that vanished or cannot be stat'd between `iterdir` and
                    # `_entry_info` used to abort the whole loop and take the
                    # directory's other rows with it.
                    continue
                results.append(info)
        except (PermissionError, OSError):
            # The walk itself failed, so there is nothing left to collect.
            return []

        return sorted(results, key=lambda x: (not x["is_dir"], x["name"]))

    def read_bytes(self, path: str) -> bytes:
        """Read a whole file as bytes, or `b""` when it cannot be read.

        The same "read" rules as :meth:`read` apply; a denied path is `b""`.
        """
        try:
            full_path = self._resolve(path)
        except PermissionError:
            return b""

        if self._denial_reason("read", str(full_path)) is not None:
            return b""
        if not full_path.is_file():
            return b""

        try:
            return full_path.read_bytes()
        except (PermissionError, OSError):
            return b""

    def read(self, path: str, offset: int = 0, limit: int = DEFAULT_READ_LIMIT) -> str:
        """Read a slice of a file with line numbers."""
        try:
            full_path = self._resolve(path)
        except PermissionError as e:
            return f"Error: {e}"

        denial = self._denial_reason("read", str(full_path))
        if denial:
            return f"Error: {denial}"

        if not full_path.exists():
            return f"Error: File '{path}' not found"
        if full_path.is_dir():
            return f"Error: '{path}' is a directory"

        try:
            with open(full_path, encoding="utf-8", errors="replace") as f:
                lines = f.readlines()
        except PermissionError:
            return f"Error: Permission denied for '{path}'"
        except OSError as e:
            return f"Error: {e}"

        if offset >= len(lines):
            return f"Error: Offset {offset} exceeds file length ({len(lines)} lines)"

        end = min(offset + limit, len(lines))
        result = "\n".join(_numbered_line(i + 1, lines[i]) for i in range(offset, end))
        if end < len(lines):
            result += f"\n\n... ({len(lines) - end} more lines)"

        return _within_read_ceiling(result, explicit=offset != 0 or limit != DEFAULT_READ_LIMIT)

    def write(self, path: str, content: str | bytes) -> WriteResult:
        """Write a file, creating parent directories as needed."""
        try:
            full_path = self._resolve(path)
        except PermissionError as e:
            return WriteResult(error=str(e))

        denial = self._denial_reason("write", str(full_path))
        if denial:
            return WriteResult(error=denial)

        try:
            full_path.parent.mkdir(parents=True, exist_ok=True)
            if isinstance(content, bytes):
                full_path.write_bytes(content)
            else:
                full_path.write_text(_normalize_newlines(content), encoding="utf-8")
            return WriteResult(path=str(full_path))
        except PermissionError:
            return WriteResult(error=f"Permission denied for '{path}'")
        except OSError as e:
            return WriteResult(error=str(e))

    def edit(
        self, path: str, old_string: str, new_string: str, replace_all: bool = False
    ) -> EditResult:
        """Edit a file by replacing a string."""
        try:
            full_path = self._resolve(path)
        except PermissionError as e:
            return EditResult(error=str(e))

        denial = self._denial_reason("edit", str(full_path))
        if denial:
            return EditResult(error=denial)

        if not full_path.exists():
            return EditResult(error=f"File '{path}' not found")

        try:
            content = full_path.read_text(encoding="utf-8")
        except UnicodeDecodeError:
            # Refused rather than decoded with replacement characters: writing the
            # result back would substitute every undecodable byte for U+FFFD and
            # destroy the file. `read` can afford `errors="replace"` because it
            # only displays the content; `edit` stores it again.
            return EditResult(error=f"'{path}' is not valid UTF-8 text and cannot be edited")
        except PermissionError:
            return EditResult(error=f"Permission denied for '{path}'")
        except OSError as e:
            return EditResult(error=str(e))

        outcome = replace_in_content(content, old_string, new_string, replace_all)
        if not isinstance(outcome, Replacement):
            return EditResult(error=outcome)

        try:
            full_path.write_text(_normalize_newlines(outcome.content), encoding="utf-8")
        except PermissionError:
            return EditResult(error=f"Permission denied for '{path}'")
        except OSError as e:
            return EditResult(error=str(e))

        return EditResult(path=str(full_path), occurrences=outcome.occurrences)

    def glob_info(self, pattern: str, path: str = ".") -> list[FileInfo]:
        """Match files by glob, most recently modified first.

        That ordering matches ripgrep and Claude Code, and is usually what an
        agent wants; the path breaks ties so the order is stable. Matches denied
        for "glob" are omitted, and "ask" counts as visible.
        """
        try:
            base_path = self._resolve(path)
        except PermissionError:
            return []

        if self._is_denied("glob", str(base_path)) or not base_path.exists():
            return []

        collected: list[tuple[float, str, FileInfo]] = []
        try:
            for match in base_path.glob(pattern):  # pragma: no branch
                try:
                    if not match.is_file():
                        continue
                    self._resolve(str(match))
                    if self._is_denied("glob", str(match)):
                        continue
                    stat = match.stat()
                except PermissionError:
                    continue
                except OSError:
                    # Per entry, not per walk. One file that vanished or cannot
                    # be stat'd between the glob and the stat used to abort the
                    # whole loop and return whatever had been collected — a
                    # silently short answer, which is worse than a missing row
                    # and worse than an error. `ls_info` and grep both skip and
                    # carry on; this now matches them.
                    continue
                info = FileInfo(
                    name=match.name,
                    path=str(match),
                    is_dir=False,
                    size=stat.st_size,
                    modified_at=iso_mtime(stat.st_mtime),
                )
                collected.append((stat.st_mtime, str(match), info))
        except (PermissionError, OSError):
            # The walk itself failed, so there is nothing left to collect.
            pass

        collected.sort(key=lambda item: (-item[0], item[1]))
        return [info for _mtime, _path, info in collected]

    def grep_raw(
        self,
        pattern: str,
        path: str | None = None,
        glob: str | None = None,
        ignore_hidden: bool = True,
    ) -> list[GrepMatch] | str:
        """Search file contents, using ripgrep when it is installed.

        Files denied for "grep" — or for "read", since a match carries content —
        never contribute results. An explicit "grep" deny on the search path
        errors the whole search.
        """
        search_path = path or str(self._root)

        try:
            validated = self._resolve(search_path)
        except PermissionError as e:
            return str(e)

        if self._is_denied("grep", str(validated)):
            return f"Error: Permission denied for grep on '{search_path}'"

        if shutil.which("rg") is not None and not validated.is_file():
            return self._grep_ripgrep(pattern, validated, glob, ignore_hidden)
        return self._grep_python(pattern, validated, glob, ignore_hidden)

    def _grep_ripgrep(
        self, pattern: str, search_path: Path, glob: str | None, ignore_hidden: bool
    ) -> list[GrepMatch] | str:
        argv = ["rg", "--line-number", "--no-heading", pattern]
        if glob:
            argv.extend(["--glob", glob])
        if not ignore_hidden:
            argv.append("--hidden")
        argv.append(".")

        try:
            result = subprocess.run(
                argv,
                cwd=search_path,
                capture_output=True,
                text=True,
                timeout=GREP_TIMEOUT_SECONDS,
            )
        except subprocess.TimeoutExpired:
            return "Error: Search timed out"
        except OSError as e:
            return f"Error: {e}"

        base_path = search_path.parent if search_path.is_file() else search_path
        matches: list[GrepMatch] = []
        for relative_path, line_number, line in _parse_grep_lines(result.stdout):
            full_path = (base_path / relative_path).resolve()
            try:
                self._resolve(str(full_path))
            except PermissionError:
                continue
            if self._hidden_from_grep(str(full_path)):
                continue
            matches.append(GrepMatch(path=str(full_path), line_number=line_number, line=line))
        return matches

    def _grep_python(
        self, pattern: str, search_path: Path, glob: str | None, ignore_hidden: bool
    ) -> list[GrepMatch] | str:
        try:
            regex = re.compile(pattern)
        except re.error as e:
            return f"Error: Invalid regex pattern: {e}"

        if not search_path.exists():
            return f"Error: Path '{search_path}' not found"

        if search_path.is_file():
            files = [search_path]
        else:
            candidates = search_path.glob(glob) if glob else search_path.rglob("*")
            files = [f for f in candidates if not is_ignored_path(f.parts, ignore_hidden)]

        matches: list[GrepMatch] = []
        for file_path in files:
            if not file_path.is_file():
                continue
            try:
                self._resolve(str(file_path))
            except PermissionError:
                continue
            if self._hidden_from_grep(str(file_path)):
                continue

            try:
                with open(file_path, encoding="utf-8", errors="replace") as f:
                    for i, line in enumerate(f):
                        if regex.search(line):
                            matches.append(
                                GrepMatch(
                                    path=str(file_path),
                                    line_number=i + 1,
                                    line=line.rstrip("\n\r"),
                                )
                            )
            except (PermissionError, OSError):
                continue

        return matches

    def _hidden_from_grep(self, path: str) -> bool:
        return self._guard is not None and self._guard.hides_from_grep(path)

    # ── Commands ───────────────────────────────────────────────────────

    def execute(self, command: str, timeout: int | None = None) -> ExecuteResponse:
        """Run a shell command in the root directory.

        Args:
            command: Command to execute.
            timeout: Maximum execution time in seconds. Defaults to
                `DEFAULT_EXECUTE_TIMEOUT`.

        Raises:
            RuntimeError: If execution is disabled for this backend.
        """
        denial = self._execute_denial(command)
        if denial is not None:
            return ExecuteResponse(output=f"Error: {denial}", exit_code=1, truncated=False)

        try:
            result = subprocess.run(
                shell_argv(command),
                cwd=self._root,
                capture_output=True,
                text=True,
                timeout=timeout if timeout is not None else DEFAULT_EXECUTE_TIMEOUT,
            )
        except subprocess.TimeoutExpired:
            return ExecuteResponse(
                output="Error: Command timed out", exit_code=124, truncated=False
            )
        except Exception as e:
            return ExecuteResponse(output=f"Error: {e}", exit_code=1, truncated=False)

        return _execute_response(result.stdout + result.stderr, result.returncode)

    async def async_execute(self, command: str, timeout: int | None = None) -> ExecuteResponse:
        """Cancellable version of :meth:`execute`.

        Cancelling the calling task kills the subprocess immediately instead of
        waiting for a thread to finish. On Unix the process gets its own session
        so the whole tree — including grandchildren the shell forked — is reaped
        on cancellation or timeout.
        """
        denial = self._execute_denial(command)
        if denial is not None:
            return ExecuteResponse(output=f"Error: {denial}", exit_code=1, truncated=False)

        try:
            process = await asyncio.create_subprocess_exec(
                *shell_argv(command),
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                cwd=self._root,
                # New session so the whole group can be killed on Unix; on
                # Windows the `cmd /c` lifecycle already takes the tree down.
                start_new_session=(sys.platform != "win32"),
            )

            try:
                stdout, stderr = await asyncio.wait_for(
                    process.communicate(),
                    timeout=timeout if timeout is not None else DEFAULT_EXECUTE_TIMEOUT,
                )
            except asyncio.CancelledError:
                _kill_process_tree(process)
                # Shielded so a second cancel cannot leave the pipes dangling.
                with contextlib.suppress(BaseException):
                    await asyncio.shield(asyncio.ensure_future(process.communicate()))
                raise
            except asyncio.TimeoutError:
                _kill_process_tree(process)
                with contextlib.suppress(BaseException):
                    await process.communicate()
                return ExecuteResponse(
                    output="Error: Command timed out", exit_code=124, truncated=False
                )

            output = stdout.decode("utf-8", errors="replace")
            output += stderr.decode("utf-8", errors="replace")
            return _execute_response(output, process.returncode)
        except Exception as e:
            return ExecuteResponse(output=f"Error: {e}", exit_code=1, truncated=False)

    def _execute_denial(self, command: str) -> str | None:
        """Why the command may not run, or `None` when it may.

        Raises:
            RuntimeError: If execution is disabled for this backend, which is a
                misconfiguration rather than a refused command.
        """
        if not self._enable_execute:
            raise RuntimeError(
                "Shell execution is disabled for this backend. "
                "Initialize with enable_execute=True to enable."
            )
        return self._guard.execute_denial_reason(command) if self._guard else None

    # ── Background processes ───────────────────────────────────────────

    def execute_background(self, command: str) -> BackgroundHandle:
        """Start `command` as a detached, long-lived process.

        Returns immediately. The process keeps running after this call — drain
        its output with :meth:`read_background` and stop it with
        :meth:`kill_background`.

        Raises:
            RuntimeError: If execution is disabled for this backend.
            PermissionError: If the ruleset refuses the command.
        """
        denial = self._execute_denial(command)
        if denial is not None:
            raise PermissionError(denial)
        return self._background.start(shell_argv(command), command)

    def read_background(self, shell_id: str) -> BackgroundOutput:
        """Return output produced since the previous read, plus run status."""
        return self._background.read(shell_id)

    def kill_background(self, shell_id: str) -> bool:
        """Stop a background process. Returns whether it was still running."""
        return self._background.kill(shell_id)

    def list_background(self) -> list[BackgroundProcessInfo]:
        """Status of every tracked background process."""
        return self._background.list()

    def kill_all_background(self) -> None:
        """Stop every background process and remove its on-disk output."""
        self._background.kill_all()

    def __del__(self) -> None:  # pragma: no cover - best-effort GC cleanup
        with contextlib.suppress(Exception):
            if self._background:
                self._background.kill_all()

execute_enabled property

Whether shell execution is enabled.

__init__(root_dir=None, allowed_directories=None, enable_execute=True, sandbox_id=None, permissions=None, ask_callback=None, ask_fallback='error')

Initialize the backend.

Parameters:

Name	Type	Description	Default
root_dir	str | Path | None	Base directory for file operations. Defaults to the first allowed directory, or the current working directory.	None
allowed_directories	list[str] | None	Directories file operations are confined to, resolved to absolute paths and created when missing. When omitted, only root_dir is reachable.	None
enable_execute	bool	Whether shell execution is available.	True
sandbox_id	str | None	Unique identifier for this backend instance.	None
permissions	PermissionRuleset | None	Optional ruleset applied after the allowed-directory check passes.	None
ask_callback	AskCallback | None	Async callback for "ask" actions, receiving (operation, target, reason) and returning whether to allow.	None
ask_fallback	AskFallback	What an unanswerable "ask" does — "deny" refuses the operation, "error" raises.	'error'

Source code in src/pydantic_ai_backends/backends/local.py

def __init__(
    self,
    root_dir: str | Path | None = None,
    allowed_directories: list[str] | None = None,
    enable_execute: bool = True,
    sandbox_id: str | None = None,
    permissions: PermissionRuleset | None = None,
    ask_callback: AskCallback | None = None,
    ask_fallback: AskFallback = "error",
):
    """Initialize the backend.

    Args:
        root_dir: Base directory for file operations. Defaults to the first
            allowed directory, or the current working directory.
        allowed_directories: Directories file operations are confined to,
            resolved to absolute paths and created when missing. When
            omitted, only `root_dir` is reachable.
        enable_execute: Whether shell execution is available.
        sandbox_id: Unique identifier for this backend instance.
        permissions: Optional ruleset applied after the allowed-directory
            check passes.
        ask_callback: Async callback for "ask" actions, receiving
            `(operation, target, reason)` and returning whether to allow.
        ask_fallback: What an unanswerable "ask" does — `"deny"` refuses the
            operation, `"error"` raises.
    """
    self._id = sandbox_id or str(uuid.uuid4())
    self._enable_execute = enable_execute
    self._permissions = permissions

    self._allowed_directories = [Path(d).resolve() for d in allowed_directories or []]
    for directory in self._allowed_directories:
        directory.mkdir(parents=True, exist_ok=True)

    if root_dir is not None:
        self._root = Path(root_dir).resolve()
    elif self._allowed_directories:
        self._root = self._allowed_directories[0]
    else:
        self._root = Path.cwd()
    self._root.mkdir(parents=True, exist_ok=True)

    if not self._allowed_directories:
        self._allowed_directories = [self._root]

    self._guard = (
        PermissionGuard(permissions, self._root, ask_callback, ask_fallback)
        if permissions is not None
        else None
    )
    self._background = BackgroundProcesses(self._root)

ls_info(path)

List one directory, omitting entries denied for "ls".

An "ask" counts as visible, since a listing cannot prompt.

Source code in src/pydantic_ai_backends/backends/local.py

def ls_info(self, path: str) -> list[FileInfo]:
    """List one directory, omitting entries denied for "ls".

    An "ask" counts as visible, since a listing cannot prompt.
    """
    try:
        full_path = self._resolve(path)
    except PermissionError:
        return []

    if self._is_denied("ls", str(full_path)) or not full_path.exists():
        return []

    if full_path.is_file():
        return [_entry_info(full_path)]

    results: list[FileInfo] = []
    try:
        for entry in full_path.iterdir():
            try:
                self._resolve(str(entry))
                if self._is_denied("ls", str(entry)):
                    continue
                info = _entry_info(entry)
            except PermissionError:
                continue
            except OSError:
                # Per entry, not per listing, matching `glob_info`: a file
                # that vanished or cannot be stat'd between `iterdir` and
                # `_entry_info` used to abort the whole loop and take the
                # directory's other rows with it.
                continue
            results.append(info)
    except (PermissionError, OSError):
        # The walk itself failed, so there is nothing left to collect.
        return []

    return sorted(results, key=lambda x: (not x["is_dir"], x["name"]))

read(path, offset=0, limit=DEFAULT_READ_LIMIT)

Read a slice of a file with line numbers.

Source code in src/pydantic_ai_backends/backends/local.py

def read(self, path: str, offset: int = 0, limit: int = DEFAULT_READ_LIMIT) -> str:
    """Read a slice of a file with line numbers."""
    try:
        full_path = self._resolve(path)
    except PermissionError as e:
        return f"Error: {e}"

    denial = self._denial_reason("read", str(full_path))
    if denial:
        return f"Error: {denial}"

    if not full_path.exists():
        return f"Error: File '{path}' not found"
    if full_path.is_dir():
        return f"Error: '{path}' is a directory"

    try:
        with open(full_path, encoding="utf-8", errors="replace") as f:
            lines = f.readlines()
    except PermissionError:
        return f"Error: Permission denied for '{path}'"
    except OSError as e:
        return f"Error: {e}"

    if offset >= len(lines):
        return f"Error: Offset {offset} exceeds file length ({len(lines)} lines)"

    end = min(offset + limit, len(lines))
    result = "\n".join(_numbered_line(i + 1, lines[i]) for i in range(offset, end))
    if end < len(lines):
        result += f"\n\n... ({len(lines) - end} more lines)"

    return _within_read_ceiling(result, explicit=offset != 0 or limit != DEFAULT_READ_LIMIT)

write(path, content)

Write a file, creating parent directories as needed.

Source code in src/pydantic_ai_backends/backends/local.py

def write(self, path: str, content: str | bytes) -> WriteResult:
    """Write a file, creating parent directories as needed."""
    try:
        full_path = self._resolve(path)
    except PermissionError as e:
        return WriteResult(error=str(e))

    denial = self._denial_reason("write", str(full_path))
    if denial:
        return WriteResult(error=denial)

    try:
        full_path.parent.mkdir(parents=True, exist_ok=True)
        if isinstance(content, bytes):
            full_path.write_bytes(content)
        else:
            full_path.write_text(_normalize_newlines(content), encoding="utf-8")
        return WriteResult(path=str(full_path))
    except PermissionError:
        return WriteResult(error=f"Permission denied for '{path}'")
    except OSError as e:
        return WriteResult(error=str(e))

edit(path, old_string, new_string, replace_all=False)

Edit a file by replacing a string.

Source code in src/pydantic_ai_backends/backends/local.py

def edit(
    self, path: str, old_string: str, new_string: str, replace_all: bool = False
) -> EditResult:
    """Edit a file by replacing a string."""
    try:
        full_path = self._resolve(path)
    except PermissionError as e:
        return EditResult(error=str(e))

    denial = self._denial_reason("edit", str(full_path))
    if denial:
        return EditResult(error=denial)

    if not full_path.exists():
        return EditResult(error=f"File '{path}' not found")

    try:
        content = full_path.read_text(encoding="utf-8")
    except UnicodeDecodeError:
        # Refused rather than decoded with replacement characters: writing the
        # result back would substitute every undecodable byte for U+FFFD and
        # destroy the file. `read` can afford `errors="replace"` because it
        # only displays the content; `edit` stores it again.
        return EditResult(error=f"'{path}' is not valid UTF-8 text and cannot be edited")
    except PermissionError:
        return EditResult(error=f"Permission denied for '{path}'")
    except OSError as e:
        return EditResult(error=str(e))

    outcome = replace_in_content(content, old_string, new_string, replace_all)
    if not isinstance(outcome, Replacement):
        return EditResult(error=outcome)

    try:
        full_path.write_text(_normalize_newlines(outcome.content), encoding="utf-8")
    except PermissionError:
        return EditResult(error=f"Permission denied for '{path}'")
    except OSError as e:
        return EditResult(error=str(e))

    return EditResult(path=str(full_path), occurrences=outcome.occurrences)

glob_info(pattern, path='.')

Match files by glob, most recently modified first.

That ordering matches ripgrep and Claude Code, and is usually what an agent wants; the path breaks ties so the order is stable. Matches denied for "glob" are omitted, and "ask" counts as visible.

Source code in src/pydantic_ai_backends/backends/local.py

def glob_info(self, pattern: str, path: str = ".") -> list[FileInfo]:
    """Match files by glob, most recently modified first.

    That ordering matches ripgrep and Claude Code, and is usually what an
    agent wants; the path breaks ties so the order is stable. Matches denied
    for "glob" are omitted, and "ask" counts as visible.
    """
    try:
        base_path = self._resolve(path)
    except PermissionError:
        return []

    if self._is_denied("glob", str(base_path)) or not base_path.exists():
        return []

    collected: list[tuple[float, str, FileInfo]] = []
    try:
        for match in base_path.glob(pattern):  # pragma: no branch
            try:
                if not match.is_file():
                    continue
                self._resolve(str(match))
                if self._is_denied("glob", str(match)):
                    continue
                stat = match.stat()
            except PermissionError:
                continue
            except OSError:
                # Per entry, not per walk. One file that vanished or cannot
                # be stat'd between the glob and the stat used to abort the
                # whole loop and return whatever had been collected — a
                # silently short answer, which is worse than a missing row
                # and worse than an error. `ls_info` and grep both skip and
                # carry on; this now matches them.
                continue
            info = FileInfo(
                name=match.name,
                path=str(match),
                is_dir=False,
                size=stat.st_size,
                modified_at=iso_mtime(stat.st_mtime),
            )
            collected.append((stat.st_mtime, str(match), info))
    except (PermissionError, OSError):
        # The walk itself failed, so there is nothing left to collect.
        pass

    collected.sort(key=lambda item: (-item[0], item[1]))
    return [info for _mtime, _path, info in collected]

grep_raw(pattern, path=None, glob=None, ignore_hidden=True)

Search file contents, using ripgrep when it is installed.

Files denied for "grep" — or for "read", since a match carries content — never contribute results. An explicit "grep" deny on the search path errors the whole search.

Source code in src/pydantic_ai_backends/backends/local.py

def grep_raw(
    self,
    pattern: str,
    path: str | None = None,
    glob: str | None = None,
    ignore_hidden: bool = True,
) -> list[GrepMatch] | str:
    """Search file contents, using ripgrep when it is installed.

    Files denied for "grep" — or for "read", since a match carries content —
    never contribute results. An explicit "grep" deny on the search path
    errors the whole search.
    """
    search_path = path or str(self._root)

    try:
        validated = self._resolve(search_path)
    except PermissionError as e:
        return str(e)

    if self._is_denied("grep", str(validated)):
        return f"Error: Permission denied for grep on '{search_path}'"

    if shutil.which("rg") is not None and not validated.is_file():
        return self._grep_ripgrep(pattern, validated, glob, ignore_hidden)
    return self._grep_python(pattern, validated, glob, ignore_hidden)

execute(command, timeout=None)

Run a shell command in the root directory.

Parameters:

Name	Type	Description	Default
command	str	Command to execute.	required
timeout	int | None	Maximum execution time in seconds. Defaults to DEFAULT_EXECUTE_TIMEOUT.	None

Raises:

Type	Description
RuntimeError	If execution is disabled for this backend.

Source code in src/pydantic_ai_backends/backends/local.py

def execute(self, command: str, timeout: int | None = None) -> ExecuteResponse:
    """Run a shell command in the root directory.

    Args:
        command: Command to execute.
        timeout: Maximum execution time in seconds. Defaults to
            `DEFAULT_EXECUTE_TIMEOUT`.

    Raises:
        RuntimeError: If execution is disabled for this backend.
    """
    denial = self._execute_denial(command)
    if denial is not None:
        return ExecuteResponse(output=f"Error: {denial}", exit_code=1, truncated=False)

    try:
        result = subprocess.run(
            shell_argv(command),
            cwd=self._root,
            capture_output=True,
            text=True,
            timeout=timeout if timeout is not None else DEFAULT_EXECUTE_TIMEOUT,
        )
    except subprocess.TimeoutExpired:
        return ExecuteResponse(
            output="Error: Command timed out", exit_code=124, truncated=False
        )
    except Exception as e:
        return ExecuteResponse(output=f"Error: {e}", exit_code=1, truncated=False)

    return _execute_response(result.stdout + result.stderr, result.returncode)

StateBackend

pydantic_ai_backends.backends.state.StateBackend

In-memory file storage backend.

Files live in a dictionary and are ephemeral — lost when the process ends, unless a host persists :attr:files and hands it back. Useful for testing, for scratch space alongside a real backend, and as the whole storage layer for an application that keeps the document itself.

Binary content is stored base64 (see :class:~pydantic_ai_backends.FileData) so that document is always JSON. read and grep decline to treat such a file as text rather than showing its encoded form; read_bytes returns exactly what was written.

Example

Python

from pydantic_ai_backends import StateBackend

backend = StateBackend()
backend.write("/src/app.py", "print('hello')")
content = backend.read("/src/app.py")
print(content)  # "     1\tprint('hello')"
matches = backend.grep_raw("print")

restored = StateBackend(files=json.loads(json.dumps(backend.files)))

Source code in src/pydantic_ai_backends/backends/state.py

class StateBackend:
    """In-memory file storage backend.

    Files live in a dictionary and are ephemeral — lost when the process ends,
    unless a host persists :attr:`files` and hands it back. Useful for testing,
    for scratch space alongside a real backend, and as the whole storage layer
    for an application that keeps the document itself.

    Binary content is stored base64 (see :class:`~pydantic_ai_backends.FileData`)
    so that document is always JSON. `read` and `grep` decline to treat such a
    file as text rather than showing its encoded form; `read_bytes` returns
    exactly what was written.

    Example:
        ```python
        from pydantic_ai_backends import StateBackend

        backend = StateBackend()
        backend.write("/src/app.py", "print('hello')")
        content = backend.read("/src/app.py")
        print(content)  # "     1\\tprint('hello')"
        matches = backend.grep_raw("print")

        restored = StateBackend(files=json.loads(json.dumps(backend.files)))
        ```
    """

    def __init__(self, files: dict[str, FileData] | None = None):
        """Initialize the backend.

        Args:
            files: Optional initial file dictionary. A document a previous
                instance produced, including one that has been through JSON,
                loads unchanged — as does one written before `encoding` existed.
        """
        self._files: dict[str, FileData] = files if files is not None else {}

    @property
    def files(self) -> dict[str, FileData]:
        """The internal files dictionary.

        Always a JSON-serialisable document: `json.loads(json.dumps(files))`
        round-trips, and so does storing it in a PostgreSQL `jsonb` column.
        """
        return self._files

    def exists(self, path: str) -> bool:
        """Whether a file is stored at `path`."""
        if unsafe_path_reason(path) is not None:
            return False
        return normalize_path(path) in self._files

    def ls_info(self, path: str) -> list[FileInfo]:
        """List the files and directories directly under `path`."""
        if unsafe_path_reason(path) is not None:
            return []

        path = normalize_path(path)
        prefix = path if path == "/" else path + "/"
        entries: dict[str, FileInfo] = {}

        for file_path, file_data in self._files.items():
            if file_path == path:
                name = file_path.rsplit("/", 1)[-1]
                entries[name] = _file_entry(name, file_path, file_data)
                continue
            if not file_path.startswith(prefix):
                continue

            name, _, rest = file_path[len(prefix) :].partition("/")
            if name in entries:
                continue
            if rest:
                entries[name] = FileInfo(name=name, path=prefix + name, is_dir=True, size=None)
            else:
                entries[name] = _file_entry(name, file_path, file_data)

        return sorted(entries.values(), key=lambda x: (not x["is_dir"], x["name"]))

    def read_bytes(self, path: str) -> bytes:
        """Read a whole file as bytes, or `b""` when there is none at `path`."""
        if unsafe_path_reason(path) is not None:
            return b""

        stored = self._files.get(normalize_path(path))
        if stored is None:
            return b""
        return _content_bytes(stored)

    def read(self, path: str, offset: int = 0, limit: int = 2000) -> str:
        """Read a slice of a file with line numbers.

        A binary file is refused rather than rendered: its stored form is
        base64, and handing that to a model as if it were the file's text is
        worse than saying there is nothing here to read.
        """
        reason = unsafe_path_reason(path)
        if reason is not None:
            return f"Error: {reason}"

        path = normalize_path(path)
        stored = self._files.get(path)
        if stored is None:
            return f"Error: File '{path}' not found"
        if _is_binary(stored):
            return f"Error: File '{path}' is binary; read it as bytes instead"

        lines = stored["content"]
        if offset >= len(lines):
            return f"Error: Offset {offset} exceeds file length ({len(lines)} lines)"

        end = min(offset + limit, len(lines))
        numbered = "\n".join(f"{i + 1:>6}\t{lines[i]}" for i in range(offset, end))
        if end < len(lines):
            return f"{numbered}\n\n... ({len(lines) - end} more lines)"
        return numbered

    def write(self, path: str, content: str | bytes) -> WriteResult:
        """Write a file, replacing any existing content."""
        reason = unsafe_path_reason(path)
        if reason is not None:
            return WriteResult(error=reason)

        path = normalize_path(path)
        lines, encoding = _to_storage(content)

        now = _timestamp()
        existing = self._files.get(path)
        entry = FileData(
            content=lines,
            created_at=existing["created_at"] if existing else now,
            modified_at=now,
        )
        if encoding is not None:
            entry["encoding"] = encoding
        self._files[path] = entry
        return WriteResult(path=path)

    def edit(
        self, path: str, old_string: str, new_string: str, replace_all: bool = False
    ) -> EditResult:
        """Edit a file by replacing a string."""
        reason = unsafe_path_reason(path)
        if reason is not None:
            return EditResult(error=reason)

        path = normalize_path(path)
        stored = self._files.get(path)
        if stored is None:
            return EditResult(error=f"File '{path}' not found")
        if _is_binary(stored):
            return EditResult(error=f"File '{path}' is binary and cannot be edited as text")

        outcome = replace_in_content(
            "\n".join(stored["content"]), old_string, new_string, replace_all
        )
        if not isinstance(outcome, Replacement):
            return EditResult(error=outcome)

        stored["content"] = outcome.content.split("\n")
        stored["modified_at"] = _timestamp()
        return EditResult(path=path, occurrences=outcome.occurrences)

    def glob_info(self, pattern: str, path: str = "/") -> list[FileInfo]:
        """Match stored paths against a glob pattern."""
        if unsafe_path_reason(path) is not None:
            return []

        path = normalize_path(path)
        root = "" if path == "/" else path
        full_pattern = f"{root}/{pattern.lstrip('/')}"

        results = [
            _file_entry(file_path.rsplit("/", 1)[-1], file_path, file_data)
            for file_path, file_data in self._files.items()
            if wcglob.globmatch(file_path, full_pattern, flags=wcglob.GLOBSTAR)
        ]
        return sorted(results, key=lambda x: x["path"])

    def grep_raw(
        self,
        pattern: str,
        path: str | None = None,
        glob: str | None = None,
        ignore_hidden: bool = True,
    ) -> list[GrepMatch] | str:
        """Search stored file contents for a regex."""
        try:
            regex = re.compile(pattern)
        except re.error as e:
            return f"Error: Invalid regex pattern: {e}"

        searchable = self._searchable_paths(path, ignore_hidden)
        if isinstance(searchable, str):
            return searchable

        if glob:
            glob_pattern = "/" + glob.lstrip("/")
            searchable = [
                p for p in searchable if wcglob.globmatch(p, glob_pattern, flags=wcglob.GLOBSTAR)
            ]

        # Binary files are skipped rather than searched: their stored form is
        # base64, so a pattern would be matched against an encoding nobody wrote
        # and the hit would name a line number that does not exist in the file.
        return [
            GrepMatch(path=file_path, line_number=i + 1, line=line)
            for file_path in searchable
            if not _is_binary(self._files[file_path])
            for i, line in enumerate(self._files[file_path]["content"])
            if regex.search(line)
        ]

    def _searchable_paths(self, path: str | None, ignore_hidden: bool) -> list[str] | str:
        """Paths grep should walk, or an error message when `path` is invalid.

        A file named outright is searched even when hidden; `ignore_hidden` only
        filters the directory walk.
        """
        visible = [p for p in self._files if not ignore_hidden or not is_hidden_path(p)]

        if path is None:
            return visible

        reason = unsafe_path_reason(path)
        if reason is not None:
            return f"Error: {reason}"

        path = normalize_path(path)
        if path in self._files:
            return [path]

        prefix = path if path == "/" else path + "/"
        return [p for p in visible if p.startswith(prefix)]

files property

The internal files dictionary.

Always a JSON-serialisable document: json.loads(json.dumps(files)) round-trips, and so does storing it in a PostgreSQL jsonb column.

__init__(files=None)

Initialize the backend.

Parameters:

Name	Type	Description	Default
files	dict[str, FileData] | None	Optional initial file dictionary. A document a previous instance produced, including one that has been through JSON, loads unchanged — as does one written before encoding existed.	None

Source code in src/pydantic_ai_backends/backends/state.py

def __init__(self, files: dict[str, FileData] | None = None):
    """Initialize the backend.

    Args:
        files: Optional initial file dictionary. A document a previous
            instance produced, including one that has been through JSON,
            loads unchanged — as does one written before `encoding` existed.
    """
    self._files: dict[str, FileData] = files if files is not None else {}

ls_info(path)

List the files and directories directly under path.

Source code in src/pydantic_ai_backends/backends/state.py

def ls_info(self, path: str) -> list[FileInfo]:
    """List the files and directories directly under `path`."""
    if unsafe_path_reason(path) is not None:
        return []

    path = normalize_path(path)
    prefix = path if path == "/" else path + "/"
    entries: dict[str, FileInfo] = {}

    for file_path, file_data in self._files.items():
        if file_path == path:
            name = file_path.rsplit("/", 1)[-1]
            entries[name] = _file_entry(name, file_path, file_data)
            continue
        if not file_path.startswith(prefix):
            continue

        name, _, rest = file_path[len(prefix) :].partition("/")
        if name in entries:
            continue
        if rest:
            entries[name] = FileInfo(name=name, path=prefix + name, is_dir=True, size=None)
        else:
            entries[name] = _file_entry(name, file_path, file_data)

    return sorted(entries.values(), key=lambda x: (not x["is_dir"], x["name"]))

read(path, offset=0, limit=2000)

Read a slice of a file with line numbers.

A binary file is refused rather than rendered: its stored form is base64, and handing that to a model as if it were the file's text is worse than saying there is nothing here to read.

Source code in src/pydantic_ai_backends/backends/state.py

def read(self, path: str, offset: int = 0, limit: int = 2000) -> str:
    """Read a slice of a file with line numbers.

    A binary file is refused rather than rendered: its stored form is
    base64, and handing that to a model as if it were the file's text is
    worse than saying there is nothing here to read.
    """
    reason = unsafe_path_reason(path)
    if reason is not None:
        return f"Error: {reason}"

    path = normalize_path(path)
    stored = self._files.get(path)
    if stored is None:
        return f"Error: File '{path}' not found"
    if _is_binary(stored):
        return f"Error: File '{path}' is binary; read it as bytes instead"

    lines = stored["content"]
    if offset >= len(lines):
        return f"Error: Offset {offset} exceeds file length ({len(lines)} lines)"

    end = min(offset + limit, len(lines))
    numbered = "\n".join(f"{i + 1:>6}\t{lines[i]}" for i in range(offset, end))
    if end < len(lines):
        return f"{numbered}\n\n... ({len(lines) - end} more lines)"
    return numbered

write(path, content)

Write a file, replacing any existing content.

Source code in src/pydantic_ai_backends/backends/state.py

def write(self, path: str, content: str | bytes) -> WriteResult:
    """Write a file, replacing any existing content."""
    reason = unsafe_path_reason(path)
    if reason is not None:
        return WriteResult(error=reason)

    path = normalize_path(path)
    lines, encoding = _to_storage(content)

    now = _timestamp()
    existing = self._files.get(path)
    entry = FileData(
        content=lines,
        created_at=existing["created_at"] if existing else now,
        modified_at=now,
    )
    if encoding is not None:
        entry["encoding"] = encoding
    self._files[path] = entry
    return WriteResult(path=path)

edit(path, old_string, new_string, replace_all=False)

Edit a file by replacing a string.

Source code in src/pydantic_ai_backends/backends/state.py

def edit(
    self, path: str, old_string: str, new_string: str, replace_all: bool = False
) -> EditResult:
    """Edit a file by replacing a string."""
    reason = unsafe_path_reason(path)
    if reason is not None:
        return EditResult(error=reason)

    path = normalize_path(path)
    stored = self._files.get(path)
    if stored is None:
        return EditResult(error=f"File '{path}' not found")
    if _is_binary(stored):
        return EditResult(error=f"File '{path}' is binary and cannot be edited as text")

    outcome = replace_in_content(
        "\n".join(stored["content"]), old_string, new_string, replace_all
    )
    if not isinstance(outcome, Replacement):
        return EditResult(error=outcome)

    stored["content"] = outcome.content.split("\n")
    stored["modified_at"] = _timestamp()
    return EditResult(path=path, occurrences=outcome.occurrences)

glob_info(pattern, path='/')

Match stored paths against a glob pattern.

Source code in src/pydantic_ai_backends/backends/state.py

def glob_info(self, pattern: str, path: str = "/") -> list[FileInfo]:
    """Match stored paths against a glob pattern."""
    if unsafe_path_reason(path) is not None:
        return []

    path = normalize_path(path)
    root = "" if path == "/" else path
    full_pattern = f"{root}/{pattern.lstrip('/')}"

    results = [
        _file_entry(file_path.rsplit("/", 1)[-1], file_path, file_data)
        for file_path, file_data in self._files.items()
        if wcglob.globmatch(file_path, full_pattern, flags=wcglob.GLOBSTAR)
    ]
    return sorted(results, key=lambda x: x["path"])

grep_raw(pattern, path=None, glob=None, ignore_hidden=True)

Search stored file contents for a regex.

Source code in src/pydantic_ai_backends/backends/state.py

def grep_raw(
    self,
    pattern: str,
    path: str | None = None,
    glob: str | None = None,
    ignore_hidden: bool = True,
) -> list[GrepMatch] | str:
    """Search stored file contents for a regex."""
    try:
        regex = re.compile(pattern)
    except re.error as e:
        return f"Error: Invalid regex pattern: {e}"

    searchable = self._searchable_paths(path, ignore_hidden)
    if isinstance(searchable, str):
        return searchable

    if glob:
        glob_pattern = "/" + glob.lstrip("/")
        searchable = [
            p for p in searchable if wcglob.globmatch(p, glob_pattern, flags=wcglob.GLOBSTAR)
        ]

    # Binary files are skipped rather than searched: their stored form is
    # base64, so a pattern would be matched against an encoding nobody wrote
    # and the hit would name a line number that does not exist in the file.
    return [
        GrepMatch(path=file_path, line_number=i + 1, line=line)
        for file_path in searchable
        if not _is_binary(self._files[file_path])
        for i, line in enumerate(self._files[file_path]["content"])
        if regex.search(line)
    ]

CompositeBackend

pydantic_ai_backends.backends.composite.CompositeBackend

Backend that routes operations to other backends by path prefix.

Note

Paths reach the matched backend as-is — no prefix is stripped — so every backend must accept the full virtual path. StateBackend accepts any path, which makes it the natural choice for a route. LocalBackend validates paths against its root_dir and will reject virtual paths, so use it as the default, not inside routes.

Example

Python

from pydantic_ai_backends import CompositeBackend, LocalBackend, StateBackend

backend = CompositeBackend(
    default=LocalBackend(root_dir="/home/user/project"),
    routes={"/scratch/": StateBackend()},
)

backend.write("src/app.py", "...")       # real filesystem
backend.write("/scratch/temp.txt", "...")  # ephemeral

Source code in src/pydantic_ai_backends/backends/composite.py

class CompositeBackend:
    """Backend that routes operations to other backends by path prefix.

    Note:
        Paths reach the matched backend **as-is** — no prefix is stripped — so
        every backend must accept the full virtual path. `StateBackend` accepts
        any path, which makes it the natural choice for a route. `LocalBackend`
        validates paths against its `root_dir` and will reject virtual paths, so
        use it as the **default**, not inside `routes`.

    Example:
        ```python
        from pydantic_ai_backends import CompositeBackend, LocalBackend, StateBackend

        backend = CompositeBackend(
            default=LocalBackend(root_dir="/home/user/project"),
            routes={"/scratch/": StateBackend()},
        )

        backend.write("src/app.py", "...")       # real filesystem
        backend.write("/scratch/temp.txt", "...")  # ephemeral
        ```
    """

    def __init__(
        self,
        default: BackendProtocol,
        routes: dict[str, BackendProtocol] | None = None,
    ):
        """Initialize the composite.

        Args:
            default: Backend for paths that match no route.
            routes: Path prefix to backend, e.g. `{"/memories/": store}`.
        """
        self._router: PrefixRouter[BackendProtocol] = PrefixRouter(default, routes)

    def exists(self, path: str) -> bool:
        """Check existence via the backend handling this path."""
        return self._router.for_path(path).exists(path)

    def ls_info(self, path: str) -> list[FileInfo]:
        """List one directory, showing route mount points when listing `/`."""
        normalized = normalize_path(path)
        if normalized != "/":
            return self._router.for_path(normalized).ls_info(normalized)

        entries = {entry["path"]: entry for entry in self._router.default.ls_info(normalized)}
        return _sorted_entries(self._router.route_directories(entries))

    def read_bytes(self, path: str) -> bytes:
        """Read bytes from the backend handling this path."""
        return self._router.for_path(path).read_bytes(path)

    def read(self, path: str, offset: int = 0, limit: int = 2000) -> str:
        """Read from the backend handling this path."""
        return self._router.for_path(path).read(path, offset, limit)

    def write(self, path: str, content: str | bytes) -> WriteResult:
        """Write to the backend handling this path."""
        return self._router.for_path(path).write(path, content)

    def edit(
        self, path: str, old_string: str, new_string: str, replace_all: bool = False
    ) -> EditResult:
        """Edit via the backend handling this path."""
        return self._router.for_path(path).edit(path, old_string, new_string, replace_all)

    def glob_info(self, pattern: str, path: str = "/") -> list[FileInfo]:
        """Match files, searching every backend when starting from the root."""
        if not is_root_path(path):
            return self._router.for_path(path).glob_info(pattern, path)

        results = list(self._router.default.glob_info(pattern, "/"))
        for prefix, backend in self._router.routes.items():
            results.extend(backend.glob_info(pattern, prefix))
        return sorted(results, key=lambda x: x["path"])

    def grep_raw(
        self,
        pattern: str,
        path: str | None = None,
        glob: str | None = None,
        ignore_hidden: bool = True,
    ) -> list[GrepMatch] | str:
        """Search, covering every backend when no specific path is given.

        An error from any backend is returned as-is rather than dropped, so a
        failed search is never mistaken for no matches.
        """
        if path is not None and not is_root_path(path):
            return self._router.for_path(path).grep_raw(pattern, path, glob, ignore_hidden)

        matches: list[GrepMatch] = []
        searches = [(path, self._router.default), *self._router.routes.items()]
        for search_path, backend in searches:
            result = backend.grep_raw(pattern, search_path, glob, ignore_hidden)
            if not isinstance(result, list):
                return result
            matches.extend(result)
        return matches

__init__(default, routes=None)

Initialize the composite.

Parameters:

Name	Type	Description	Default
default	BackendProtocol	Backend for paths that match no route.	required
routes	dict[str, BackendProtocol] | None	Path prefix to backend, e.g. {"/memories/": store}.	None

Source code in src/pydantic_ai_backends/backends/composite.py

def __init__(
    self,
    default: BackendProtocol,
    routes: dict[str, BackendProtocol] | None = None,
):
    """Initialize the composite.

    Args:
        default: Backend for paths that match no route.
        routes: Path prefix to backend, e.g. `{"/memories/": store}`.
    """
    self._router: PrefixRouter[BackendProtocol] = PrefixRouter(default, routes)

ls_info(path)

List one directory, showing route mount points when listing /.

Source code in src/pydantic_ai_backends/backends/composite.py

def ls_info(self, path: str) -> list[FileInfo]:
    """List one directory, showing route mount points when listing `/`."""
    normalized = normalize_path(path)
    if normalized != "/":
        return self._router.for_path(normalized).ls_info(normalized)

    entries = {entry["path"]: entry for entry in self._router.default.ls_info(normalized)}
    return _sorted_entries(self._router.route_directories(entries))

read(path, offset=0, limit=2000)

Read from the backend handling this path.

Source code in src/pydantic_ai_backends/backends/composite.py

def read(self, path: str, offset: int = 0, limit: int = 2000) -> str:
    """Read from the backend handling this path."""
    return self._router.for_path(path).read(path, offset, limit)

write(path, content)

Write to the backend handling this path.

Source code in src/pydantic_ai_backends/backends/composite.py

def write(self, path: str, content: str | bytes) -> WriteResult:
    """Write to the backend handling this path."""
    return self._router.for_path(path).write(path, content)

edit(path, old_string, new_string, replace_all=False)

Edit via the backend handling this path.

Source code in src/pydantic_ai_backends/backends/composite.py

def edit(
    self, path: str, old_string: str, new_string: str, replace_all: bool = False
) -> EditResult:
    """Edit via the backend handling this path."""
    return self._router.for_path(path).edit(path, old_string, new_string, replace_all)

glob_info(pattern, path='/')

Match files, searching every backend when starting from the root.

Source code in src/pydantic_ai_backends/backends/composite.py

def glob_info(self, pattern: str, path: str = "/") -> list[FileInfo]:
    """Match files, searching every backend when starting from the root."""
    if not is_root_path(path):
        return self._router.for_path(path).glob_info(pattern, path)

    results = list(self._router.default.glob_info(pattern, "/"))
    for prefix, backend in self._router.routes.items():
        results.extend(backend.glob_info(pattern, prefix))
    return sorted(results, key=lambda x: x["path"])

grep_raw(pattern, path=None, glob=None, ignore_hidden=True)

Search, covering every backend when no specific path is given.

An error from any backend is returned as-is rather than dropped, so a failed search is never mistaken for no matches.

Source code in src/pydantic_ai_backends/backends/composite.py

def grep_raw(
    self,
    pattern: str,
    path: str | None = None,
    glob: str | None = None,
    ignore_hidden: bool = True,
) -> list[GrepMatch] | str:
    """Search, covering every backend when no specific path is given.

    An error from any backend is returned as-is rather than dropped, so a
    failed search is never mistaken for no matches.
    """
    if path is not None and not is_root_path(path):
        return self._router.for_path(path).grep_raw(pattern, path, glob, ignore_hidden)

    matches: list[GrepMatch] = []
    searches = [(path, self._router.default), *self._router.routes.items()]
    for search_path, backend in searches:
        result = backend.grep_raw(pattern, search_path, glob, ignore_hidden)
        if not isinstance(result, list):
            return result
        matches.extend(result)
    return matches