Backends API

LocalBackend

pydantic_ai_backends.backends.local.LocalBackend

Local filesystem backend with optional shell execution.

Combines file operations (Python native) with optional shell command execution (subprocess). File operations can be restricted to specific directories using allowed_directories.

Example

Python

from pydantic_ai_backends import LocalBackend

# Full access with shell execution
backend = LocalBackend(root_dir="/workspace")
backend.write("/src/app.py", "print('hello')")
result = backend.execute("python /src/app.py")

# Restricted directories, no shell
backend = 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.

    Combines file operations (Python native) with optional shell command
    execution (subprocess). File operations can be restricted to specific
    directories using `allowed_directories`.

    Example:
        ```python
        from pydantic_ai_backends import LocalBackend

        # Full access with shell execution
        backend = LocalBackend(root_dir="/workspace")
        backend.write("/src/app.py", "print('hello')")
        result = backend.execute("python /src/app.py")

        # Restricted directories, no shell
        backend = 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. If not provided,
                uses first allowed_directory or current working directory.
            allowed_directories: List of directories that file operations are
                restricted to. If None, only root_dir is accessible.
                Paths are resolved to absolute paths.
            enable_execute: Whether shell execution is enabled. Default True.
            sandbox_id: Unique identifier for this backend instance.
            permissions: Optional permission ruleset for fine-grained access control.
                If provided, operations are checked against this ruleset after
                the allowed_directories check passes.
            ask_callback: Async callback for "ask" permission actions. Receives
                (operation, target, reason) and returns True to allow.
            ask_fallback: What to do when ask_callback is None but needed.
                "deny" denies the operation, "error" raises PermissionError.
        """
        self._id = sandbox_id or str(uuid.uuid4())
        self._enable_execute = enable_execute
        self._permissions = permissions
        self._ask_callback = ask_callback
        self._ask_fallback = ask_fallback
        self._permission_checker: PermissionChecker | None = None

        # Background (long-lived) process registry — see execute_background().
        self._bg: dict[str, _BackgroundProcess] = {}
        self._bg_counter = 0
        self._bg_dir: Path | None = None

        # Initialize permission checker if ruleset provided
        if permissions is not None:
            from pydantic_ai_backends.permissions.checker import PermissionChecker

            self._permission_checker = PermissionChecker(
                ruleset=permissions,
                ask_callback=ask_callback,
                ask_fallback=ask_fallback,
            )

        # Resolve allowed directories
        self._allowed_directories: list[Path] | None = None
        if allowed_directories is not None:
            self._allowed_directories = [Path(d).resolve() for d in allowed_directories]
            # Create directories if they don't exist
            for d in self._allowed_directories:
                d.mkdir(parents=True, exist_ok=True)

        # Resolve root directory
        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()  # pragma: no cover

        # Ensure root exists
        self._root.mkdir(parents=True, exist_ok=True)

        # If no allowed_directories specified, restrict to root_dir only
        if self._allowed_directories is None:
            self._allowed_directories = [self._root]

    @staticmethod
    def _shell_cmd(command: str) -> list[str]:
        return ["cmd", "/c", command] if sys.platform == "win32" else ["sh", "-c", command]

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

    @property
    def root_dir(self) -> Path:
        """Get the root directory."""
        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._permission_checker

    def _check_permission_sync(self, operation: PermissionOperation, target: str) -> str | None:
        """Check permission synchronously.

        Returns None if allowed, or an error message if denied.
        For "ask" actions, returns an error since this is sync.
        """
        if self._permission_checker is None:
            return None

        from pydantic_ai_backends.permissions.checker import (
            PermissionAskError,
        )

        action = self._permission_checker.check_sync(operation, target)
        if action == "allow":
            return None
        if action == "deny":
            rule = self._permission_checker._find_matching_rule(operation, target)
            if rule and rule.description:
                return f"Permission denied: {rule.description}"
            return f"Permission denied for {operation} on '{target}'"
        # action == "ask" - in sync context, we can't ask
        if self._ask_fallback == "deny":
            return f"Permission denied for {operation} on '{target}' (approval required)"
        # ask_fallback == "error"
        raise PermissionAskError(operation, target, "Approval required but no callback")

    def _is_denied_sync(self, operation: PermissionOperation, target: str) -> bool:
        """True when the ruleset explicitly resolves to "deny" for this target.

        Used by listing/search operations (`ls`, `glob`, `grep`) where a sync
        "ask" cannot be answered: those operations only hide explicitly denied
        targets and treat "ask" as visible, so a ruleset whose global default
        is "ask" doesn't blank out every listing.
        """
        if self._permission_checker is None:
            return False
        return self._permission_checker.check_sync(operation, target) == "deny"

    def _grep_file_hidden(self, path: str) -> bool:
        """True when a file must not contribute grep matches.

        A file is hidden from grep when it is explicitly denied for "grep",
        or denied for "read" — grep returns file content, so a read deny
        must also stop content from leaking through search results.
        """
        return self._is_denied_sync("grep", path) or self._is_denied_sync("read", path)

    def _command_path_targets(self, command: str) -> set[str]:
        """Best-effort extraction of filesystem paths referenced by a command.

        Tokenizes the command, expands `~`, and resolves each token (and the
        value side of `--flag=value` tokens) against the backend root — the
        cwd commands actually run in. Non-path tokens resolve to harmless
        paths that simply won't match any deny rule.
        """
        try:
            tokens = shlex.split(command, posix=True)
        except ValueError:
            # Unbalanced quotes etc. — fall back to whitespace splitting so a
            # malformed command can't dodge the guard entirely.
            tokens = [token.strip("\"'") for token in command.split()]

        candidates: set[str] = set()
        for token in tokens:
            candidates.add(token)
            if "=" in token:
                candidates.add(token.split("=", 1)[1])

        targets: set[str] = set()
        for candidate in candidates:
            if not candidate:
                continue
            expanded = os.path.expanduser(candidate)
            p = Path(expanded)
            resolved = p.resolve() if p.is_absolute() else (self._root / expanded).resolve()
            targets.add(str(resolved))
        return targets

    def _check_execute_permission_sync(self, command: str) -> str | None:
        """Check a command against execute rules plus a best-effort path guard.

        After the command-pattern check, path-looking tokens in the command are
        resolved and denied when one hits a "read" or "write" deny rule, so the
        straightforward bypass (`cat restricted/secret.txt`) is caught. This is
        defense-in-depth, not a security boundary — a shell can always reach a
        file in ways command-string inspection cannot see. For enforced
        isolation use a sandboxed backend (e.g. `DockerSandbox`).
        """
        perm_error = self._check_permission_sync("execute", command)
        if perm_error:
            return perm_error
        if self._permission_checker is None:
            return None

        guarded_ops: tuple[PermissionOperation, ...] = ("read", "write")
        for target in sorted(self._command_path_targets(command)):
            for op in guarded_ops:
                if self._is_denied_sync(op, target):
                    return (
                        f"Permission denied: command references '{target}', "
                        f"which is denied for {op}"
                    )
        return None

    def _validate_path(self, path: str) -> Path:
        """Validate and resolve path within allowed directories.

        Args:
            path: Path to validate (absolute or relative to root).

        Returns:
            Resolved absolute Path.

        Raises:
            PermissionError: If path is outside allowed directories.
        """
        # Handle relative paths
        if not Path(path).is_absolute():
            resolved = (self._root / path).resolve()
        else:
            resolved = Path(path).resolve()

        # Check against allowed directories
        assert self._allowed_directories is not None
        for allowed in self._allowed_directories:
            try:
                resolved.relative_to(allowed)
                return resolved
            except ValueError:
                continue

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

    def exists(self, path: str) -> bool:
        """Check whether a file exists on the filesystem.

        Returns `False` for: missing files, directories, paths outside
        the allowed directory set, and any other invalid path the
        filesystem refuses to stat. The protocol contract is "invalid
        paths, directories, and permission errors all return False" —
        callers needing to distinguish those reasons should use
        `ls_info()`.

        Exception sources:
        - `PermissionError` from `_validate_path` when the resolved
          path escapes the allowed directory set.
        - `ValueError` from `Path.is_file()` on paths the OS rejects
          before it can stat them (notably embedded null bytes — POSIX
          rejects them at the syscall boundary).
        - `OSError` covers the remaining edge cases (filename too long,
          ELOOP on a symlink cycle, etc.). `Path.is_file()` swallows
          most of these already; the explicit catch is belt-and-braces.
        """
        try:
            validated = self._validate_path(path)
            return validated.is_file()
        except (PermissionError, ValueError, OSError):
            return False

    def ls_info(self, path: str) -> list[FileInfo]:
        """List files and directories at the given path.

        Entries (and the path itself) with an explicit "ls" deny rule are
        omitted; "ask" is treated as visible (listings can't prompt).
        """
        try:
            full_path = self._validate_path(path)
        except PermissionError:  # pragma: no cover
            return []

        if self._is_denied_sync("ls", str(full_path)):
            return []

        if not full_path.exists():  # pragma: no cover
            return []

        if full_path.is_file():  # pragma: no cover
            return [
                FileInfo(
                    name=full_path.name,
                    path=str(full_path),
                    is_dir=False,
                    size=full_path.stat().st_size,
                )
            ]

        results: list[FileInfo] = []
        try:
            for entry in full_path.iterdir():
                try:
                    # Validate each entry is within allowed dirs
                    self._validate_path(str(entry))
                    if self._is_denied_sync("ls", str(entry)):
                        continue
                    results.append(
                        FileInfo(
                            name=entry.name,
                            path=str(entry),
                            is_dir=entry.is_dir(),
                            size=entry.stat().st_size if entry.is_file() else None,
                        )
                    )
                except PermissionError:  # pragma: no cover
                    continue  # Skip entries outside allowed directories
        except PermissionError:  # pragma: no cover
            return []

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

    def read_bytes(self, path: str) -> bytes:
        """Read raw bytes from a file.

        Applies the same "read" permission rules as `read` — a denied path
        returns `b""`, matching the documented "empty bytes on failure"
        contract (an "ask" that can't be answered follows `ask_fallback`,
        exactly like `read`).
        """
        try:
            full_path = self._validate_path(path)
        except PermissionError:
            return b""

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

        if not full_path.exists() or not full_path.is_file():
            return b""

        try:
            return full_path.read_bytes()
        except (PermissionError, OSError):  # pragma: no cover
            return b""

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

        # Check permissions
        perm_error = self._check_permission_sync("read", str(full_path))
        if perm_error:
            return f"Error: {perm_error}"

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

        if full_path.is_dir():  # pragma: no cover
            return f"Error: '{path}' is a directory"

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

        total_lines = len(lines)

        if offset >= total_lines:  # pragma: no cover
            return f"Error: Offset {offset} exceeds file length ({total_lines} lines)"

        end = min(offset + limit, total_lines)
        result_lines = []

        for i in range(offset, end):
            line_num = i + 1
            line = lines[i].rstrip("\n\r")
            result_lines.append(f"{line_num:>6}\t{line}")

        result = "\n".join(result_lines)

        if end < total_lines:
            result += f"\n\n... ({total_lines - end} more lines)"

        # Guard against a single read flooding the context. A request is
        # "explicit" when the caller asked for a specific range; in that case an
        # over-large result errors so the agent narrows it. A default read just
        # truncates to a page, so a plain `read_file(path)` never hard-fails.
        if len(result) > MAX_READ_OUTPUT:
            explicit = offset != 0 or limit != DEFAULT_READ_LIMIT
            if explicit:
                return (
                    f"Error: The requested range is too large to return "
                    f"({len(result):,} chars, limit {MAX_READ_OUTPUT:,}). "
                    "Read a smaller slice with a lower `limit`, or use `grep` "
                    "to locate the part you need."
                )
            result = (
                result[:MAX_READ_OUTPUT]
                + f"\n\n... (truncated at {MAX_READ_OUTPUT:,} chars — pass a smaller "
                "`limit`/`offset` or use `grep` to read the rest)"
            )

        return result

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

        # Check permissions
        perm_error = self._check_permission_sync("write", str(full_path))
        if perm_error:
            return WriteResult(error=perm_error)

        try:
            full_path.parent.mkdir(parents=True, exist_ok=True)

            if isinstance(content, bytes):  # pragma: no cover
                full_path.write_bytes(content)
            else:
                full_path.write_text(_normalize_newlines(content), encoding="utf-8")

            return WriteResult(path=str(full_path))
        except PermissionError:  # pragma: no cover
            return WriteResult(error=f"Permission denied for '{path}'")
        except OSError as e:  # pragma: no cover
            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 strings."""
        try:
            full_path = self._validate_path(path)
        except PermissionError as e:  # pragma: no cover
            return EditResult(error=str(e))

        # Check permissions
        perm_error = self._check_permission_sync("edit", str(full_path))
        if perm_error:
            return EditResult(error=perm_error)

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

        try:
            content = full_path.read_text(encoding="utf-8")
        except PermissionError:  # pragma: no cover
            return EditResult(error=f"Permission denied for '{path}'")
        except OSError as e:  # pragma: no cover
            return EditResult(error=str(e))

        occurrences = content.count(old_string)

        if occurrences == 0:  # pragma: no cover
            return EditResult(error=f"String '{old_string}' not found in file")

        if occurrences > 1 and not replace_all:  # pragma: no cover
            return EditResult(
                error=f"String '{old_string}' found {occurrences} times. "
                "Use replace_all=True to replace all, or provide more context."
            )

        if replace_all:  # pragma: no cover
            new_content = content.replace(old_string, new_string)
        else:
            new_content = content.replace(old_string, new_string, 1)

        try:
            full_path.write_text(_normalize_newlines(new_content), encoding="utf-8")
            return EditResult(path=str(full_path), occurrences=occurrences if replace_all else 1)
        except PermissionError:  # pragma: no cover
            return EditResult(error=f"Permission denied for '{path}'")
        except OSError as e:  # pragma: no cover
            return EditResult(error=str(e))

    def glob_info(self, pattern: str, path: str = ".") -> list[FileInfo]:
        """Find files matching a glob pattern.

        Matches (and the base path itself) with an explicit "glob" deny rule
        are omitted; "ask" is treated as visible (listings can't prompt).
        """
        try:
            base_path = self._validate_path(path)
        except PermissionError:  # pragma: no cover
            return []

        if self._is_denied_sync("glob", str(base_path)):
            return []

        if not base_path.exists():  # pragma: no cover
            return []

        # Collect (mtime, path, FileInfo) so results order by recency —
        # most-recently-modified first, which is usually what an agent wants
        # (matches ripgrep/Claude Code glob ordering). Path is the tie-break for
        # a stable order when mtimes collide.
        collected: list[tuple[float, str, FileInfo]] = []

        try:
            for match in base_path.glob(pattern):  # pragma: no branch
                if match.is_file():
                    try:
                        self._validate_path(str(match))
                        if self._is_denied_sync("glob", str(match)):
                            continue
                        stat = match.stat()
                        collected.append(
                            (
                                stat.st_mtime,
                                str(match),
                                FileInfo(
                                    name=match.name,
                                    path=str(match),
                                    is_dir=False,
                                    size=stat.st_size,
                                ),
                            )
                        )
                    except PermissionError:  # pragma: no cover
                        continue
        except (PermissionError, OSError):  # pragma: no cover
            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 for pattern in files.

        Uses ripgrep if available, falls back to Python regex.

        Files denied for "grep" — or for "read", since matches leak file
        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_path = self._validate_path(search_path)
        except PermissionError as e:  # pragma: no cover
            return str(e)

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

        # Try ripgrep first when searching directories for better performance
        use_ripgrep = shutil.which("rg") is not None and not validated_path.is_file()
        if use_ripgrep:  # pragma: no cover
            return self._grep_ripgrep(pattern, validated_path, glob, ignore_hidden)

        return self._grep_python(pattern, validated_path, glob, ignore_hidden)  # pragma: no cover

    def _grep_ripgrep(  # pragma: no cover
        self, pattern: str, search_path: Path, glob: str | None = None, ignore_hidden: bool = True
    ) -> list[GrepMatch] | str:
        """Use ripgrep for fast searching."""
        cmd = ["rg", "--line-number", "--no-heading", pattern]

        if glob:
            cmd.extend(["--glob", glob])

        if not ignore_hidden:
            cmd.append("--hidden")

        cmd.append(".")

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

        results: list[GrepMatch] = []

        for line in result.stdout.strip().split("\n"):
            if not line:
                continue

            parts = line.split(":", 2)
            if len(parts) >= 3:
                file_path = parts[0]
                try:
                    line_num = int(parts[1])
                except ValueError:
                    continue
                content = parts[2]

                # Convert to absolute path and validate
                try:
                    base_path = search_path.parent if search_path.is_file() else search_path
                    full_path = (base_path / file_path).resolve()
                    self._validate_path(str(full_path))
                    if self._grep_file_hidden(str(full_path)):
                        continue
                    results.append(
                        GrepMatch(
                            path=str(full_path),
                            line_number=line_num,
                            line=content,
                        )
                    )
                except PermissionError:
                    continue

        return results

    def _grep_python(  # pragma: no cover
        self,
        pattern: str,
        search_path: Path,
        glob_pattern: str | None = None,
        ignore_hidden: bool = True,
    ) -> list[GrepMatch] | str:
        """Use Python regex for searching (fallback)."""
        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"

        results: list[GrepMatch] = []

        if search_path.is_file():
            files = [search_path]
        else:
            if glob_pattern:
                files = list(search_path.glob(glob_pattern))
            else:
                files = list(search_path.rglob("*"))
            # Skip build/cache dirs (and hidden, when asked) so the fallback
            # doesn't trawl node_modules/__pycache__ the way ripgrep wouldn't.
            files = [f for f in files if not _grep_path_ignored(f.parts, ignore_hidden)]

        for file_path in files:
            if not file_path.is_file():
                continue

            try:
                self._validate_path(str(file_path))
            except PermissionError:
                continue

            if self._grep_file_hidden(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):
                            results.append(
                                GrepMatch(
                                    path=str(file_path),
                                    line_number=i + 1,
                                    line=line.rstrip("\n\r"),
                                )
                            )
            except (PermissionError, OSError):
                continue

        return results

    def execute(self, command: str, timeout: int | None = None) -> ExecuteResponse:
        """Execute a shell command.

        Args:
            command: Command to execute.
            timeout: Maximum execution time in seconds (default 120).

        Returns:
            ExecuteResponse with output, exit code, and truncation status.

        Raises:
            RuntimeError: If execute is disabled for this backend.
        """
        if not self._enable_execute:
            raise RuntimeError(
                "Shell execution is disabled for this backend. "
                "Initialize with enable_execute=True to enable."
            )

        # Check permissions
        perm_error = self._check_execute_permission_sync(command)
        if perm_error:
            return ExecuteResponse(
                output=f"Error: {perm_error}",
                exit_code=1,
                truncated=False,
            )

        try:
            result = subprocess.run(
                self._shell_cmd(command),
                cwd=self._root,
                capture_output=True,
                text=True,
                timeout=timeout if timeout is not None else 120,
            )

            output = result.stdout + result.stderr

            # Truncate if too long
            truncated = len(output) > MAX_EXECUTE_OUTPUT
            if truncated:  # pragma: no cover
                output = output[:MAX_EXECUTE_OUTPUT]

            return ExecuteResponse(
                output=output,
                exit_code=result.returncode,
                truncated=truncated,
            )
        except subprocess.TimeoutExpired:
            return ExecuteResponse(
                output="Error: Command timed out",
                exit_code=124,
                truncated=False,
            )
        except Exception as e:  # pragma: no cover
            return ExecuteResponse(
                output=f"Error: {e}",
                exit_code=1,
                truncated=False,
            )

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

        Uses `asyncio.create_subprocess_exec` so that cancelling the calling
        task immediately kills the subprocess rather than waiting for a thread
        to finish. On Unix, the subprocess runs in its own session so the entire
        process tree (including any grandchildren the shell forked) is reaped
        on cancellation or timeout. Defaults to a 120-second timeout when
        `timeout` is `None`.
        """
        if not self._enable_execute:
            raise RuntimeError(
                "Shell execution is disabled for this backend. "
                "Initialize with enable_execute=True to enable."
            )

        perm_error = self._check_execute_permission_sync(command)
        if perm_error:
            return ExecuteResponse(output=f"Error: {perm_error}", exit_code=1, truncated=False)

        try:
            proc = await asyncio.create_subprocess_exec(
                *self._shell_cmd(command),
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                cwd=self._root,
                # New session so we can kill the entire process group on Unix.
                # Windows kills the process tree via the cmd /c lifecycle.
                start_new_session=(sys.platform != "win32"),
            )

            try:
                stdout_b, stderr_b = await asyncio.wait_for(
                    proc.communicate(), timeout=timeout if timeout is not None else 120
                )
            except asyncio.CancelledError:
                self._kill_proc_tree(proc)
                # Shield cleanup so a second cancel can't leave pipes dangling.
                with contextlib.suppress(BaseException):
                    await asyncio.shield(asyncio.ensure_future(proc.communicate()))
                raise
            except asyncio.TimeoutError:
                self._kill_proc_tree(proc)
                with contextlib.suppress(BaseException):
                    await proc.communicate()
                return ExecuteResponse(
                    output="Error: Command timed out",
                    exit_code=124,
                    truncated=False,
                )

            output = stdout_b.decode("utf-8", errors="replace") + stderr_b.decode(
                "utf-8", errors="replace"
            )

            truncated = len(output) > MAX_EXECUTE_OUTPUT
            if truncated:  # pragma: no cover
                output = output[:MAX_EXECUTE_OUTPUT]

            return ExecuteResponse(
                output=output,
                exit_code=proc.returncode if proc.returncode is not None else 1,
                truncated=truncated,
            )
        except Exception as e:  # pragma: no cover
            return ExecuteResponse(output=f"Error: {e}", exit_code=1, truncated=False)

    @staticmethod
    def _kill_proc_tree(proc: asyncio.subprocess.Process) -> None:
        """Kill the subprocess and (on Unix) every grandchild it forked.

        On Unix the process is launched with `start_new_session=True`, so we
        can `killpg` the whole tree. On Windows `proc.kill()` is sufficient
        because `cmd /c` already terminates child processes when it dies.
        """
        if sys.platform == "win32":
            with contextlib.suppress(ProcessLookupError):
                proc.kill()
            return
        with contextlib.suppress(ProcessLookupError):
            os.killpg(proc.pid, signal.SIGKILL)

    # ── Background (long-lived) processes ────────────────────────────

    @staticmethod
    def _kill_popen_tree(popen: subprocess.Popen[bytes]) -> None:
        """Kill a background `Popen` and (on Unix) its whole process group."""
        if popen.poll() is not None:
            return
        if sys.platform == "win32":  # pragma: no cover - exercised on Windows only
            with contextlib.suppress(ProcessLookupError):
                popen.kill()
            return
        with contextlib.suppress(ProcessLookupError, OSError):
            os.killpg(os.getpgid(popen.pid), signal.SIGKILL)

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

        Returns immediately with a handle. The process keeps running after this
        call (it is NOT reaped like `execute`); drain its output with
        `read_background` and stop it with `kill_background`.
        """
        if not self._enable_execute:
            raise RuntimeError(
                "Shell execution is disabled for this backend. "
                "Initialize with enable_execute=True to enable."
            )
        perm_error = self._check_execute_permission_sync(command)
        if perm_error:
            raise PermissionError(perm_error)

        if self._bg_dir is None:
            self._bg_dir = Path(tempfile.mkdtemp(prefix="pad_bg_"))
        self._bg_counter += 1
        shell_id = f"bg_{self._bg_counter}"
        stdout_path = self._bg_dir / f"{shell_id}.out"
        stderr_path = self._bg_dir / f"{shell_id}.err"

        # The child writes straight to these files; the parent's handles can be
        # closed right after spawn (the child keeps its own dup'd descriptors).
        with open(stdout_path, "wb") as out_fh, open(stderr_path, "wb") as err_fh:
            popen = subprocess.Popen(
                self._shell_cmd(command),
                cwd=self._root,
                stdout=out_fh,
                stderr=err_fh,
                stdin=subprocess.DEVNULL,
                start_new_session=(sys.platform != "win32"),
            )
        self._bg[shell_id] = _BackgroundProcess(
            shell_id=shell_id,
            command=command,
            popen=popen,
            stdout_path=stdout_path,
            stderr_path=stderr_path,
        )
        return BackgroundHandle(shell_id=shell_id, pid=popen.pid, command=command)

    @staticmethod
    def _drain(path: Path, pos: int) -> tuple[str, int]:
        """Read new bytes from `path` starting at `pos`; return (text, new_pos)."""
        try:
            with open(path, "rb") as f:
                f.seek(pos)
                data = f.read()
        except OSError:  # pragma: no cover - file removed mid-read
            return "", pos
        return data.decode("utf-8", errors="replace"), pos + len(data)

    def read_background(self, shell_id: str) -> BackgroundOutput:
        """Return output produced since the previous read, plus run status."""
        proc = self._bg.get(shell_id)
        if proc is None:
            return BackgroundOutput(
                shell_id=shell_id,
                stdout="",
                stderr=f"No such background shell: {shell_id}",
                running=False,
                exit_code=None,
            )
        exit_code = proc.popen.poll()
        out, proc.stdout_pos = self._drain(proc.stdout_path, proc.stdout_pos)
        err, proc.stderr_pos = self._drain(proc.stderr_path, proc.stderr_pos)
        return BackgroundOutput(
            shell_id=shell_id,
            stdout=out,
            stderr=err,
            running=exit_code is None,
            exit_code=exit_code,
        )

    def kill_background(self, shell_id: str) -> bool:
        """Stop a background process. Returns True if it was still running."""
        proc = self._bg.get(shell_id)
        if proc is None:
            return False
        was_running = proc.popen.poll() is None
        self._kill_popen_tree(proc.popen)
        with contextlib.suppress(Exception):
            proc.popen.wait(timeout=2)
        return was_running

    def list_background(self) -> list[BackgroundProcessInfo]:
        """Return status for every tracked background process."""
        infos: list[BackgroundProcessInfo] = []
        for proc in self._bg.values():
            exit_code = proc.popen.poll()
            infos.append(
                BackgroundProcessInfo(
                    shell_id=proc.shell_id,
                    command=proc.command,
                    pid=proc.popen.pid,
                    running=exit_code is None,
                    exit_code=exit_code,
                )
            )
        return infos

    def kill_all_background(self) -> None:
        """Stop every background process and remove its on-disk output."""
        for proc in list(self._bg.values()):
            self._kill_popen_tree(proc.popen)
            with contextlib.suppress(Exception):
                proc.popen.wait(timeout=2)
        self._bg.clear()
        if self._bg_dir is not None:
            shutil.rmtree(self._bg_dir, ignore_errors=True)
            self._bg_dir = None

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

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. If not provided, uses first allowed_directory or current working directory.	None
allowed_directories	list[str] | None	List of directories that file operations are restricted to. If None, only root_dir is accessible. Paths are resolved to absolute paths.	None
enable_execute	bool	Whether shell execution is enabled. Default True.	True
sandbox_id	str | None	Unique identifier for this backend instance.	None
permissions	PermissionRuleset | None	Optional permission ruleset for fine-grained access control. If provided, operations are checked against this ruleset after the allowed_directories check passes.	None
ask_callback	AskCallback | None	Async callback for "ask" permission actions. Receives (operation, target, reason) and returns True to allow.	None
ask_fallback	AskFallback	What to do when ask_callback is None but needed. "deny" denies the operation, "error" raises PermissionError.	'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. If not provided,
            uses first allowed_directory or current working directory.
        allowed_directories: List of directories that file operations are
            restricted to. If None, only root_dir is accessible.
            Paths are resolved to absolute paths.
        enable_execute: Whether shell execution is enabled. Default True.
        sandbox_id: Unique identifier for this backend instance.
        permissions: Optional permission ruleset for fine-grained access control.
            If provided, operations are checked against this ruleset after
            the allowed_directories check passes.
        ask_callback: Async callback for "ask" permission actions. Receives
            (operation, target, reason) and returns True to allow.
        ask_fallback: What to do when ask_callback is None but needed.
            "deny" denies the operation, "error" raises PermissionError.
    """
    self._id = sandbox_id or str(uuid.uuid4())
    self._enable_execute = enable_execute
    self._permissions = permissions
    self._ask_callback = ask_callback
    self._ask_fallback = ask_fallback
    self._permission_checker: PermissionChecker | None = None

    # Background (long-lived) process registry — see execute_background().
    self._bg: dict[str, _BackgroundProcess] = {}
    self._bg_counter = 0
    self._bg_dir: Path | None = None

    # Initialize permission checker if ruleset provided
    if permissions is not None:
        from pydantic_ai_backends.permissions.checker import PermissionChecker

        self._permission_checker = PermissionChecker(
            ruleset=permissions,
            ask_callback=ask_callback,
            ask_fallback=ask_fallback,
        )

    # Resolve allowed directories
    self._allowed_directories: list[Path] | None = None
    if allowed_directories is not None:
        self._allowed_directories = [Path(d).resolve() for d in allowed_directories]
        # Create directories if they don't exist
        for d in self._allowed_directories:
            d.mkdir(parents=True, exist_ok=True)

    # Resolve root directory
    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()  # pragma: no cover

    # Ensure root exists
    self._root.mkdir(parents=True, exist_ok=True)

    # If no allowed_directories specified, restrict to root_dir only
    if self._allowed_directories is None:
        self._allowed_directories = [self._root]

ls_info(path)

List files and directories at the given path.

Entries (and the path itself) with an explicit "ls" deny rule are omitted; "ask" is treated as visible (listings can't prompt).

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

def ls_info(self, path: str) -> list[FileInfo]:
    """List files and directories at the given path.

    Entries (and the path itself) with an explicit "ls" deny rule are
    omitted; "ask" is treated as visible (listings can't prompt).
    """
    try:
        full_path = self._validate_path(path)
    except PermissionError:  # pragma: no cover
        return []

    if self._is_denied_sync("ls", str(full_path)):
        return []

    if not full_path.exists():  # pragma: no cover
        return []

    if full_path.is_file():  # pragma: no cover
        return [
            FileInfo(
                name=full_path.name,
                path=str(full_path),
                is_dir=False,
                size=full_path.stat().st_size,
            )
        ]

    results: list[FileInfo] = []
    try:
        for entry in full_path.iterdir():
            try:
                # Validate each entry is within allowed dirs
                self._validate_path(str(entry))
                if self._is_denied_sync("ls", str(entry)):
                    continue
                results.append(
                    FileInfo(
                        name=entry.name,
                        path=str(entry),
                        is_dir=entry.is_dir(),
                        size=entry.stat().st_size if entry.is_file() else None,
                    )
                )
            except PermissionError:  # pragma: no cover
                continue  # Skip entries outside allowed directories
    except PermissionError:  # pragma: no cover
        return []

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

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

Read file content with line numbers.

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

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

    # Check permissions
    perm_error = self._check_permission_sync("read", str(full_path))
    if perm_error:
        return f"Error: {perm_error}"

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

    if full_path.is_dir():  # pragma: no cover
        return f"Error: '{path}' is a directory"

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

    total_lines = len(lines)

    if offset >= total_lines:  # pragma: no cover
        return f"Error: Offset {offset} exceeds file length ({total_lines} lines)"

    end = min(offset + limit, total_lines)
    result_lines = []

    for i in range(offset, end):
        line_num = i + 1
        line = lines[i].rstrip("\n\r")
        result_lines.append(f"{line_num:>6}\t{line}")

    result = "\n".join(result_lines)

    if end < total_lines:
        result += f"\n\n... ({total_lines - end} more lines)"

    # Guard against a single read flooding the context. A request is
    # "explicit" when the caller asked for a specific range; in that case an
    # over-large result errors so the agent narrows it. A default read just
    # truncates to a page, so a plain `read_file(path)` never hard-fails.
    if len(result) > MAX_READ_OUTPUT:
        explicit = offset != 0 or limit != DEFAULT_READ_LIMIT
        if explicit:
            return (
                f"Error: The requested range is too large to return "
                f"({len(result):,} chars, limit {MAX_READ_OUTPUT:,}). "
                "Read a smaller slice with a lower `limit`, or use `grep` "
                "to locate the part you need."
            )
        result = (
            result[:MAX_READ_OUTPUT]
            + f"\n\n... (truncated at {MAX_READ_OUTPUT:,} chars — pass a smaller "
            "`limit`/`offset` or use `grep` to read the rest)"
        )

    return result

write(path, content)

Write content to a file.

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

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

    # Check permissions
    perm_error = self._check_permission_sync("write", str(full_path))
    if perm_error:
        return WriteResult(error=perm_error)

    try:
        full_path.parent.mkdir(parents=True, exist_ok=True)

        if isinstance(content, bytes):  # pragma: no cover
            full_path.write_bytes(content)
        else:
            full_path.write_text(_normalize_newlines(content), encoding="utf-8")

        return WriteResult(path=str(full_path))
    except PermissionError:  # pragma: no cover
        return WriteResult(error=f"Permission denied for '{path}'")
    except OSError as e:  # pragma: no cover
        return WriteResult(error=str(e))

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

Edit a file by replacing strings.

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 strings."""
    try:
        full_path = self._validate_path(path)
    except PermissionError as e:  # pragma: no cover
        return EditResult(error=str(e))

    # Check permissions
    perm_error = self._check_permission_sync("edit", str(full_path))
    if perm_error:
        return EditResult(error=perm_error)

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

    try:
        content = full_path.read_text(encoding="utf-8")
    except PermissionError:  # pragma: no cover
        return EditResult(error=f"Permission denied for '{path}'")
    except OSError as e:  # pragma: no cover
        return EditResult(error=str(e))

    occurrences = content.count(old_string)

    if occurrences == 0:  # pragma: no cover
        return EditResult(error=f"String '{old_string}' not found in file")

    if occurrences > 1 and not replace_all:  # pragma: no cover
        return EditResult(
            error=f"String '{old_string}' found {occurrences} times. "
            "Use replace_all=True to replace all, or provide more context."
        )

    if replace_all:  # pragma: no cover
        new_content = content.replace(old_string, new_string)
    else:
        new_content = content.replace(old_string, new_string, 1)

    try:
        full_path.write_text(_normalize_newlines(new_content), encoding="utf-8")
        return EditResult(path=str(full_path), occurrences=occurrences if replace_all else 1)
    except PermissionError:  # pragma: no cover
        return EditResult(error=f"Permission denied for '{path}'")
    except OSError as e:  # pragma: no cover
        return EditResult(error=str(e))

glob_info(pattern, path='.')

Find files matching a glob pattern.

Matches (and the base path itself) with an explicit "glob" deny rule are omitted; "ask" is treated as visible (listings can't prompt).

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

def glob_info(self, pattern: str, path: str = ".") -> list[FileInfo]:
    """Find files matching a glob pattern.

    Matches (and the base path itself) with an explicit "glob" deny rule
    are omitted; "ask" is treated as visible (listings can't prompt).
    """
    try:
        base_path = self._validate_path(path)
    except PermissionError:  # pragma: no cover
        return []

    if self._is_denied_sync("glob", str(base_path)):
        return []

    if not base_path.exists():  # pragma: no cover
        return []

    # Collect (mtime, path, FileInfo) so results order by recency —
    # most-recently-modified first, which is usually what an agent wants
    # (matches ripgrep/Claude Code glob ordering). Path is the tie-break for
    # a stable order when mtimes collide.
    collected: list[tuple[float, str, FileInfo]] = []

    try:
        for match in base_path.glob(pattern):  # pragma: no branch
            if match.is_file():
                try:
                    self._validate_path(str(match))
                    if self._is_denied_sync("glob", str(match)):
                        continue
                    stat = match.stat()
                    collected.append(
                        (
                            stat.st_mtime,
                            str(match),
                            FileInfo(
                                name=match.name,
                                path=str(match),
                                is_dir=False,
                                size=stat.st_size,
                            ),
                        )
                    )
                except PermissionError:  # pragma: no cover
                    continue
    except (PermissionError, OSError):  # pragma: no cover
        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 for pattern in files.

Uses ripgrep if available, falls back to Python regex.

Files denied for "grep" — or for "read", since matches leak file 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 for pattern in files.

    Uses ripgrep if available, falls back to Python regex.

    Files denied for "grep" — or for "read", since matches leak file
    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_path = self._validate_path(search_path)
    except PermissionError as e:  # pragma: no cover
        return str(e)

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

    # Try ripgrep first when searching directories for better performance
    use_ripgrep = shutil.which("rg") is not None and not validated_path.is_file()
    if use_ripgrep:  # pragma: no cover
        return self._grep_ripgrep(pattern, validated_path, glob, ignore_hidden)

    return self._grep_python(pattern, validated_path, glob, ignore_hidden)  # pragma: no cover

execute(command, timeout=None)

Execute a shell command.

Parameters:

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

Returns:

Type	Description
ExecuteResponse	ExecuteResponse with output, exit code, and truncation status.

Raises:

Type	Description
RuntimeError	If execute 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:
    """Execute a shell command.

    Args:
        command: Command to execute.
        timeout: Maximum execution time in seconds (default 120).

    Returns:
        ExecuteResponse with output, exit code, and truncation status.

    Raises:
        RuntimeError: If execute is disabled for this backend.
    """
    if not self._enable_execute:
        raise RuntimeError(
            "Shell execution is disabled for this backend. "
            "Initialize with enable_execute=True to enable."
        )

    # Check permissions
    perm_error = self._check_execute_permission_sync(command)
    if perm_error:
        return ExecuteResponse(
            output=f"Error: {perm_error}",
            exit_code=1,
            truncated=False,
        )

    try:
        result = subprocess.run(
            self._shell_cmd(command),
            cwd=self._root,
            capture_output=True,
            text=True,
            timeout=timeout if timeout is not None else 120,
        )

        output = result.stdout + result.stderr

        # Truncate if too long
        truncated = len(output) > MAX_EXECUTE_OUTPUT
        if truncated:  # pragma: no cover
            output = output[:MAX_EXECUTE_OUTPUT]

        return ExecuteResponse(
            output=output,
            exit_code=result.returncode,
            truncated=truncated,
        )
    except subprocess.TimeoutExpired:
        return ExecuteResponse(
            output="Error: Command timed out",
            exit_code=124,
            truncated=False,
        )
    except Exception as e:  # pragma: no cover
        return ExecuteResponse(
            output=f"Error: {e}",
            exit_code=1,
            truncated=False,
        )

StateBackend

pydantic_ai_backends.backends.state.StateBackend

In-memory file storage backend.

Files are stored in a dictionary and are ephemeral (lost when the process ends). Useful for testing and temporary file operations.

Example

Python

from pydantic_ai_backends import StateBackend

backend = StateBackend()

# Write a file
backend.write("/src/app.py", "print('hello')")

# Read it back
content = backend.read("/src/app.py")
print(content)  # "     1\tprint('hello')"

# Search files
matches = backend.grep_raw("print")

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

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

    Files are stored in a dictionary and are ephemeral (lost when the
    process ends). Useful for testing and temporary file operations.

    Example:
        ```python
        from pydantic_ai_backends import StateBackend

        backend = StateBackend()

        # Write a file
        backend.write("/src/app.py", "print('hello')")

        # Read it back
        content = backend.read("/src/app.py")
        print(content)  # "     1\\tprint('hello')"

        # Search files
        matches = backend.grep_raw("print")
        ```
    """

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

        Args:
            files: Optional initial file dictionary.
        """
        self._files: dict[str, FileData] = files if files is not None else {}

    @property
    def _files_not_hidden(self) -> dict[str, FileData]:
        return {path: data for path, data in self._files.items() if _is_not_hidden_path(path)}

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

    def _get_timestamp(self) -> str:
        """Get current ISO 8601 timestamp."""
        return datetime.now(timezone.utc).isoformat()

    def exists(self, path: str) -> bool:
        """Check whether a file exists in the in-memory store."""
        if _validate_path(path) is not None:
            return False
        return _normalize_path(path) in self._files

    def ls_info(self, path: str) -> list[FileInfo]:
        """List files and directories at the given path."""
        error = _validate_path(path)
        if error:
            return []

        path = _normalize_path(path)

        # Collect all entries at this level
        entries: dict[str, FileInfo] = {}
        prefix = path if path == "/" else path + "/"

        for file_path, file_data in self._files.items():
            if not file_path.startswith(prefix) and file_path != path:
                continue  # pragma: no cover

            # Get the relative path from the directory
            if file_path == path:
                # This is a file, not a directory
                name = file_path.split("/")[-1]
                entries[name] = FileInfo(
                    name=name,
                    path=file_path,
                    is_dir=False,
                    size=sum(len(line) for line in file_data["content"]),
                )
            else:  # pragma: no cover
                rel_path = file_path[len(prefix) :]
                parts = rel_path.split("/")
                name = parts[0]

                if name not in entries:
                    if len(parts) == 1:
                        # Direct child file
                        entries[name] = FileInfo(
                            name=name,
                            path=file_path,
                            is_dir=False,
                            size=sum(len(line) for line in file_data["content"]),
                        )
                    else:
                        # Directory (has more parts)
                        entries[name] = FileInfo(
                            name=name,
                            path=prefix + name,
                            is_dir=True,
                            size=None,
                        )

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

    def read_bytes(self, path: str) -> bytes:
        """Read raw bytes from a file.

        Args:
            path: File path to read.

        Returns:
            File content as bytes.
        """
        error = _validate_path(path)
        if error:  # pragma: no cover
            return f"Error: {error}".encode()

        path = _normalize_path(path)

        if path not in self._files:
            return b""

        content = "\n".join(self._files[path]["content"])
        return content.encode("utf-8", errors="replace")  # pragma: no cover

    def read(self, path: str, offset: int = 0, limit: int = 2000) -> str:
        """Read file content with line numbers."""
        error = _validate_path(path)
        if error:  # pragma: no cover
            return f"Error: {error}"

        path = _normalize_path(path)

        if path not in self._files:
            return f"Error: File '{path}' not found"

        lines = self._files[path]["content"]
        total_lines = len(lines)

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

        end = min(offset + limit, total_lines)
        result_lines = []

        for i in range(offset, end):
            line_num = i + 1  # 1-indexed
            result_lines.append(f"{line_num:>6}\t{lines[i]}")

        result = "\n".join(result_lines)

        if end < total_lines:
            result += f"\n\n... ({total_lines - end} more lines)"

        return result

    def write(self, path: str, content: str | bytes) -> WriteResult:
        """Write content to a file."""
        error = _validate_path(path)
        if error:
            return WriteResult(error=error)

        path = _normalize_path(path)
        now = self._get_timestamp()

        # Convert bytes to string if needed
        if isinstance(content, bytes):
            content = content.decode("utf-8", errors="replace")

        # Split content into lines, preserving empty lines
        lines = content.split("\n")

        existing = self._files.get(path)
        created_at = existing["created_at"] if existing else now
        self._files[path] = FileData(
            content=lines,
            created_at=created_at,
            modified_at=now,
        )

        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 strings."""
        error = _validate_path(path)
        if error:
            return EditResult(error=error)

        path = _normalize_path(path)

        if path not in self._files:
            return EditResult(error=f"File '{path}' not found")  # pragma: no cover

        content = "\n".join(self._files[path]["content"])
        occurrences = content.count(old_string)

        if occurrences == 0:
            return EditResult(error=f"String '{old_string}' not found in file")  # pragma: no cover

        if occurrences > 1 and not replace_all:
            return EditResult(
                error=f"String '{old_string}' found {occurrences} times. "
                "Use replace_all=True to replace all, or provide more context."
            )

        if replace_all:
            new_content = content.replace(old_string, new_string)
        else:
            new_content = content.replace(old_string, new_string, 1)

        self._files[path]["content"] = new_content.split("\n")
        self._files[path]["modified_at"] = self._get_timestamp()

        return EditResult(path=path, occurrences=occurrences if replace_all else 1)

    def glob_info(self, pattern: str, path: str = "/") -> list[FileInfo]:
        """Find files matching a glob pattern."""
        error = _validate_path(path)
        if error:
            return []

        path = _normalize_path(path)

        # Combine path and pattern
        if path == "/":
            full_pattern = "/" + pattern.lstrip("/")
        else:
            full_pattern = path + "/" + pattern.lstrip("/")

        results: list[FileInfo] = []

        for file_path, file_data in self._files.items():
            # Use wcmatch for glob matching
            if wcglob.globmatch(file_path, full_pattern, flags=wcglob.GLOBSTAR):
                name = file_path.split("/")[-1]
                results.append(
                    FileInfo(
                        name=name,
                        path=file_path,
                        is_dir=False,
                        size=sum(len(line) for line in file_data["content"]),
                    )
                )

        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 for pattern in files."""
        try:
            regex = re.compile(pattern)
        except re.error as e:
            return f"Error: Invalid regex pattern: {e}"

        results: list[GrepMatch] = []
        files = self._files_not_hidden if ignore_hidden else self._files
        # Determine which files to search
        files_to_search: list[str] = []

        if path:
            error = _validate_path(path)
            if error:
                return f"Error: {error}"
            path = _normalize_path(path)

            # An explicitly named file is searched even when hidden: look it up
            # in the full file set, not the ignore_hidden-filtered view. The
            # filter only applies to directory walks below.
            if path in self._files:
                files_to_search = [path]
            else:
                # Path is a directory - search all files under it
                prefix = path if path == "/" else path + "/"
                files_to_search = [f for f in files if f.startswith(prefix)]
        else:
            files_to_search = list(files.keys())

        # Filter by glob if provided
        if glob:
            glob_pattern = "/" + glob.lstrip("/")
            files_to_search = [
                f
                for f in files_to_search
                if wcglob.globmatch(f, glob_pattern, flags=wcglob.GLOBSTAR)
            ]

        # Search each file. Index into the full file set so an explicitly
        # named hidden file (added to files_to_search above) is found even
        # when ignore_hidden filtered it out of the directory-walk view.
        for file_path in files_to_search:
            lines = self._files[file_path]["content"]
            for i, line in enumerate(lines):
                if regex.search(line):
                    results.append(
                        GrepMatch(
                            path=file_path,
                            line_number=i + 1,
                            line=line,
                        )
                    )

        return results

files property

Get the internal files dictionary.

__init__(files=None)

Initialize the backend.

Parameters:

Name	Type	Description	Default
files	dict[str, FileData] | None	Optional initial file dictionary.	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.
    """
    self._files: dict[str, FileData] = files if files is not None else {}

ls_info(path)

List files and directories at the given path.

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

def ls_info(self, path: str) -> list[FileInfo]:
    """List files and directories at the given path."""
    error = _validate_path(path)
    if error:
        return []

    path = _normalize_path(path)

    # Collect all entries at this level
    entries: dict[str, FileInfo] = {}
    prefix = path if path == "/" else path + "/"

    for file_path, file_data in self._files.items():
        if not file_path.startswith(prefix) and file_path != path:
            continue  # pragma: no cover

        # Get the relative path from the directory
        if file_path == path:
            # This is a file, not a directory
            name = file_path.split("/")[-1]
            entries[name] = FileInfo(
                name=name,
                path=file_path,
                is_dir=False,
                size=sum(len(line) for line in file_data["content"]),
            )
        else:  # pragma: no cover
            rel_path = file_path[len(prefix) :]
            parts = rel_path.split("/")
            name = parts[0]

            if name not in entries:
                if len(parts) == 1:
                    # Direct child file
                    entries[name] = FileInfo(
                        name=name,
                        path=file_path,
                        is_dir=False,
                        size=sum(len(line) for line in file_data["content"]),
                    )
                else:
                    # Directory (has more parts)
                    entries[name] = FileInfo(
                        name=name,
                        path=prefix + name,
                        is_dir=True,
                        size=None,
                    )

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

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

Read file content with line numbers.

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

def read(self, path: str, offset: int = 0, limit: int = 2000) -> str:
    """Read file content with line numbers."""
    error = _validate_path(path)
    if error:  # pragma: no cover
        return f"Error: {error}"

    path = _normalize_path(path)

    if path not in self._files:
        return f"Error: File '{path}' not found"

    lines = self._files[path]["content"]
    total_lines = len(lines)

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

    end = min(offset + limit, total_lines)
    result_lines = []

    for i in range(offset, end):
        line_num = i + 1  # 1-indexed
        result_lines.append(f"{line_num:>6}\t{lines[i]}")

    result = "\n".join(result_lines)

    if end < total_lines:
        result += f"\n\n... ({total_lines - end} more lines)"

    return result

write(path, content)

Write content to a file.

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

def write(self, path: str, content: str | bytes) -> WriteResult:
    """Write content to a file."""
    error = _validate_path(path)
    if error:
        return WriteResult(error=error)

    path = _normalize_path(path)
    now = self._get_timestamp()

    # Convert bytes to string if needed
    if isinstance(content, bytes):
        content = content.decode("utf-8", errors="replace")

    # Split content into lines, preserving empty lines
    lines = content.split("\n")

    existing = self._files.get(path)
    created_at = existing["created_at"] if existing else now
    self._files[path] = FileData(
        content=lines,
        created_at=created_at,
        modified_at=now,
    )

    return WriteResult(path=path)

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

Edit a file by replacing strings.

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 strings."""
    error = _validate_path(path)
    if error:
        return EditResult(error=error)

    path = _normalize_path(path)

    if path not in self._files:
        return EditResult(error=f"File '{path}' not found")  # pragma: no cover

    content = "\n".join(self._files[path]["content"])
    occurrences = content.count(old_string)

    if occurrences == 0:
        return EditResult(error=f"String '{old_string}' not found in file")  # pragma: no cover

    if occurrences > 1 and not replace_all:
        return EditResult(
            error=f"String '{old_string}' found {occurrences} times. "
            "Use replace_all=True to replace all, or provide more context."
        )

    if replace_all:
        new_content = content.replace(old_string, new_string)
    else:
        new_content = content.replace(old_string, new_string, 1)

    self._files[path]["content"] = new_content.split("\n")
    self._files[path]["modified_at"] = self._get_timestamp()

    return EditResult(path=path, occurrences=occurrences if replace_all else 1)

glob_info(pattern, path='/')

Find files matching a glob pattern.

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

def glob_info(self, pattern: str, path: str = "/") -> list[FileInfo]:
    """Find files matching a glob pattern."""
    error = _validate_path(path)
    if error:
        return []

    path = _normalize_path(path)

    # Combine path and pattern
    if path == "/":
        full_pattern = "/" + pattern.lstrip("/")
    else:
        full_pattern = path + "/" + pattern.lstrip("/")

    results: list[FileInfo] = []

    for file_path, file_data in self._files.items():
        # Use wcmatch for glob matching
        if wcglob.globmatch(file_path, full_pattern, flags=wcglob.GLOBSTAR):
            name = file_path.split("/")[-1]
            results.append(
                FileInfo(
                    name=name,
                    path=file_path,
                    is_dir=False,
                    size=sum(len(line) for line in file_data["content"]),
                )
            )

    return sorted(results, key=lambda x: x["path"])

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

Search for pattern in files.

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 for pattern in files."""
    try:
        regex = re.compile(pattern)
    except re.error as e:
        return f"Error: Invalid regex pattern: {e}"

    results: list[GrepMatch] = []
    files = self._files_not_hidden if ignore_hidden else self._files
    # Determine which files to search
    files_to_search: list[str] = []

    if path:
        error = _validate_path(path)
        if error:
            return f"Error: {error}"
        path = _normalize_path(path)

        # An explicitly named file is searched even when hidden: look it up
        # in the full file set, not the ignore_hidden-filtered view. The
        # filter only applies to directory walks below.
        if path in self._files:
            files_to_search = [path]
        else:
            # Path is a directory - search all files under it
            prefix = path if path == "/" else path + "/"
            files_to_search = [f for f in files if f.startswith(prefix)]
    else:
        files_to_search = list(files.keys())

    # Filter by glob if provided
    if glob:
        glob_pattern = "/" + glob.lstrip("/")
        files_to_search = [
            f
            for f in files_to_search
            if wcglob.globmatch(f, glob_pattern, flags=wcglob.GLOBSTAR)
        ]

    # Search each file. Index into the full file set so an explicitly
    # named hidden file (added to files_to_search above) is found even
    # when ignore_hidden filtered it out of the directory-walk view.
    for file_path in files_to_search:
        lines = self._files[file_path]["content"]
        for i, line in enumerate(lines):
            if regex.search(line):
                results.append(
                    GrepMatch(
                        path=file_path,
                        line_number=i + 1,
                        line=line,
                    )
                )

    return results

CompositeBackend

pydantic_ai_backends.backends.composite.CompositeBackend

Backend that routes operations to different backends by path prefix.

Allows combining multiple backends (e.g., memory for temp files, filesystem for persistent storage) under a unified interface.

Note

Routes are matched using exact-or-child semantics: a path matches a prefix if it equals the prefix exactly or starts with prefix + "/". Paths are passed to the matched backend as-is (no prefix stripping), so each backend must accept the full virtual path.

StateBackend accepts any virtual path, making it the natural choice for routes. LocalBackend validates paths against its root_dir and will reject virtual paths that differ from the real filesystem path — use it as the default backend, not inside routes.

Example

Python

from pydantic_ai_backends import CompositeBackend, StateBackend, LocalBackend

# LocalBackend as default (real filesystem), StateBackend for ephemeral space
backend = CompositeBackend(
    default=LocalBackend(root_dir="/home/user/project"),
    routes={
        "/scratch/": StateBackend(),  # Ephemeral virtual space
    },
)

# Routes to LocalBackend (real filesystem, relative to root_dir)
backend.write("src/app.py", "...")

# Routes to StateBackend (ephemeral)
backend.write("/scratch/temp.txt", "...")

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

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

    Allows combining multiple backends (e.g., memory for temp files,
    filesystem for persistent storage) under a unified interface.

    Note:
        Routes are matched using exact-or-child semantics: a path matches a prefix
        if it equals the prefix exactly or starts with `prefix + "/"`.
        Paths are passed to the matched backend **as-is** (no prefix stripping),
        so each backend must accept the full virtual path.

        `StateBackend` accepts any virtual path, making it the natural choice for
        routes. `LocalBackend` validates paths against its `root_dir` and will
        reject virtual paths that differ from the real filesystem path — use it as
        the **default** backend, not inside `routes`.

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

        # LocalBackend as default (real filesystem), StateBackend for ephemeral space
        backend = CompositeBackend(
            default=LocalBackend(root_dir="/home/user/project"),
            routes={
                "/scratch/": StateBackend(),  # Ephemeral virtual space
            },
        )

        # Routes to LocalBackend (real filesystem, relative to root_dir)
        backend.write("src/app.py", "...")

        # Routes to StateBackend (ephemeral)
        backend.write("/scratch/temp.txt", "...")
        ```
    """

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

        Args:
            default: Default backend for paths that don't match any route.
            routes: Dictionary mapping path prefixes to backends.
                    e.g., {"/memories/": store_backend, "/temp/": state_backend}
        """
        self._default = default
        self._routes = routes or {}

        # Sort routes by length (longest first) for correct matching
        self._sorted_prefixes = sorted(self._routes.keys(), key=len, reverse=True)

    @staticmethod
    def _normalize_path(path: str) -> str:
        """Normalize path to consistent format."""
        return _normalize_path(path)

    def _get_backend(self, path: str) -> BackendProtocol:
        """Get the appropriate backend for a path."""
        normalized = self._normalize_path(path)
        for prefix in self._sorted_prefixes:
            normalized_prefix = self._normalize_path(prefix)
            if normalized == normalized_prefix or normalized.startswith(normalized_prefix + "/"):
                return self._routes[prefix]
        return self._default

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

    def ls_info(self, path: str) -> list[FileInfo]:
        """List files, aggregating from all relevant backends."""
        normalized = self._normalize_path(path)

        # For root path, aggregate from all backends
        if normalized == "/":
            all_entries: dict[str, FileInfo] = {}

            # First, get entries from default backend
            for entry in self._default.ls_info(normalized):
                all_entries[entry["path"]] = entry

            # Then, add virtual directories for route prefixes
            for prefix in self._routes:
                # Extract first directory from prefix
                parts = prefix.strip("/").split("/")
                if parts[0]:
                    dir_name = parts[0]
                    dir_path = "/" + dir_name
                    if dir_path not in all_entries:
                        all_entries[dir_path] = FileInfo(
                            name=dir_name,
                            path=dir_path,
                            is_dir=True,
                            size=None,
                        )

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

        # Route to appropriate backend
        backend = self._get_backend(normalized)
        return backend.ls_info(normalized)

    def read_bytes(self, path: str) -> bytes:
        """Read bytes from the appropriate backend."""
        return self._get_backend(path).read_bytes(path)

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

    def write(self, path: str, content: str | bytes) -> WriteResult:
        """Write to the appropriate backend."""
        return self._get_backend(path).write(path, content)

    def edit(
        self, path: str, old_string: str, new_string: str, replace_all: bool = False
    ) -> EditResult:
        """Edit using the appropriate backend."""
        return self._get_backend(path).edit(path, old_string, new_string, replace_all)

    def glob_info(self, pattern: str, path: str = "/") -> list[FileInfo]:
        """Glob across all backends if searching from root."""
        if path == "/" or path == "":
            all_results: list[FileInfo] = []

            # Search in default backend
            all_results.extend(self._default.glob_info(pattern, path))

            # Search in each routed backend
            for prefix, backend in self._routes.items():
                results = backend.glob_info(pattern, prefix)
                all_results.extend(results)

            return sorted(all_results, key=lambda x: x["path"])

        return self._get_backend(path).glob_info(pattern, path)

    def grep_raw(
        self,
        pattern: str,
        path: str | None = None,
        glob: str | None = None,
        ignore_hidden: bool = True,
    ) -> list[GrepMatch] | str:
        """Grep across all backends if no specific path."""
        if path is None or path == "/" or path == "":
            all_results: list[GrepMatch] = []

            # Search in default backend. Propagate an error string (e.g. an
            # invalid regex) instead of silently dropping it, so callers can
            # tell "no matches" apart from "search failed".
            result = self._default.grep_raw(pattern, path, glob, ignore_hidden)
            if isinstance(result, list):
                all_results.extend(result)
            else:
                return result

            # Search in each routed backend
            for prefix, backend in self._routes.items():
                result = backend.grep_raw(pattern, prefix, glob, ignore_hidden)
                if isinstance(result, list):
                    all_results.extend(result)
                else:
                    return result

            return all_results

        return self._get_backend(path).grep_raw(pattern, path, glob, ignore_hidden)

__init__(default, routes=None)

Initialize composite backend.

Parameters:

Name	Type	Description	Default
default	BackendProtocol	Default backend for paths that don't match any route.	required
routes	dict[str, BackendProtocol] | None	Dictionary mapping path prefixes to backends. e.g., {"/memories/": store_backend, "/temp/": state_backend}	None

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

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

    Args:
        default: Default backend for paths that don't match any route.
        routes: Dictionary mapping path prefixes to backends.
                e.g., {"/memories/": store_backend, "/temp/": state_backend}
    """
    self._default = default
    self._routes = routes or {}

    # Sort routes by length (longest first) for correct matching
    self._sorted_prefixes = sorted(self._routes.keys(), key=len, reverse=True)

ls_info(path)

List files, aggregating from all relevant backends.

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

def ls_info(self, path: str) -> list[FileInfo]:
    """List files, aggregating from all relevant backends."""
    normalized = self._normalize_path(path)

    # For root path, aggregate from all backends
    if normalized == "/":
        all_entries: dict[str, FileInfo] = {}

        # First, get entries from default backend
        for entry in self._default.ls_info(normalized):
            all_entries[entry["path"]] = entry

        # Then, add virtual directories for route prefixes
        for prefix in self._routes:
            # Extract first directory from prefix
            parts = prefix.strip("/").split("/")
            if parts[0]:
                dir_name = parts[0]
                dir_path = "/" + dir_name
                if dir_path not in all_entries:
                    all_entries[dir_path] = FileInfo(
                        name=dir_name,
                        path=dir_path,
                        is_dir=True,
                        size=None,
                    )

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

    # Route to appropriate backend
    backend = self._get_backend(normalized)
    return backend.ls_info(normalized)

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

Read from the appropriate backend.

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 appropriate backend."""
    return self._get_backend(path).read(path, offset, limit)

write(path, content)

Write to the appropriate backend.

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

def write(self, path: str, content: str | bytes) -> WriteResult:
    """Write to the appropriate backend."""
    return self._get_backend(path).write(path, content)

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

Edit using the appropriate backend.

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 using the appropriate backend."""
    return self._get_backend(path).edit(path, old_string, new_string, replace_all)

glob_info(pattern, path='/')

Glob across all backends if searching from root.

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

def glob_info(self, pattern: str, path: str = "/") -> list[FileInfo]:
    """Glob across all backends if searching from root."""
    if path == "/" or path == "":
        all_results: list[FileInfo] = []

        # Search in default backend
        all_results.extend(self._default.glob_info(pattern, path))

        # Search in each routed backend
        for prefix, backend in self._routes.items():
            results = backend.glob_info(pattern, prefix)
            all_results.extend(results)

        return sorted(all_results, key=lambda x: x["path"])

    return self._get_backend(path).glob_info(pattern, path)

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

Grep across all backends if no specific path.

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:
    """Grep across all backends if no specific path."""
    if path is None or path == "/" or path == "":
        all_results: list[GrepMatch] = []

        # Search in default backend. Propagate an error string (e.g. an
        # invalid regex) instead of silently dropping it, so callers can
        # tell "no matches" apart from "search failed".
        result = self._default.grep_raw(pattern, path, glob, ignore_hidden)
        if isinstance(result, list):
            all_results.extend(result)
        else:
            return result

        # Search in each routed backend
        for prefix, backend in self._routes.items():
            result = backend.grep_raw(pattern, prefix, glob, ignore_hidden)
            if isinstance(result, list):
                all_results.extend(result)
            else:
                return result

        return all_results

    return self._get_backend(path).grep_raw(pattern, path, glob, ignore_hidden)