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

        # 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]

    @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 (
            PermissionError,
        )

        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 PermissionError(operation, target, "Approval required but no callback")

    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 ls_info(self, path: str) -> list[FileInfo]:
        """List files and directories at the given path."""
        try:
            full_path = self._validate_path(path)
        except PermissionError:  # pragma: no cover
            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))
                    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:  # pragma: no cover
        """Read raw bytes from a file."""
        try:
            full_path = self._validate_path(path)
        except PermissionError:
            return b""

        if not full_path.exists() or 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 = 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)"

        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(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(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."""
        try:
            base_path = self._validate_path(path)
        except PermissionError:  # pragma: no cover
            return []

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

        results: list[FileInfo] = []

        try:
            for match in base_path.glob(pattern):  # pragma: no branch
                if match.is_file():
                    try:
                        self._validate_path(str(match))
                        results.append(
                            FileInfo(
                                name=match.name,
                                path=str(match),
                                is_dir=False,
                                size=match.stat().st_size,
                            )
                        )
                    except PermissionError:  # pragma: no cover
                        continue
        except (PermissionError, OSError):  # pragma: no cover
            pass

        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.

        Uses ripgrep if available, falls back to Python regex.
        """
        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)

        # 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))
                    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("*"))
            if ignore_hidden:
                files = [f for f in files if not any(part.startswith(".") for part in f.parts)]

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

            try:
                self._validate_path(str(file_path))
            except PermissionError:
                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_permission_sync("execute", command)
        if perm_error:
            return ExecuteResponse(
                output=f"Error: {perm_error}",
                exit_code=1,
                truncated=False,
            )

        try:
            result = subprocess.run(
                ["sh", "-c", command],
                cwd=self._root,
                capture_output=True,
                text=True,
                timeout=timeout or 120,
            )

            output = result.stdout + result.stderr

            # Truncate if too long
            max_output = 100000
            truncated = len(output) > max_output
            if truncated:  # pragma: no cover
                output = output[:max_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,
            )

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

    # 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.

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."""
    try:
        full_path = self._validate_path(path)
    except PermissionError:  # pragma: no cover
        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))
                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)"

    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(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(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.

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."""
    try:
        base_path = self._validate_path(path)
    except PermissionError:  # pragma: no cover
        return []

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

    results: list[FileInfo] = []

    try:
        for match in base_path.glob(pattern):  # pragma: no branch
            if match.is_file():
                try:
                    self._validate_path(str(match))
                    results.append(
                        FileInfo(
                            name=match.name,
                            path=str(match),
                            is_dir=False,
                            size=match.stat().st_size,
                        )
                    )
                except PermissionError:  # pragma: no cover
                    continue
    except (PermissionError, OSError):  # pragma: no cover
        pass

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

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

Search for pattern in files.

Uses ripgrep if available, falls back to Python regex.

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.
    """
    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)

    # 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_permission_sync("execute", command)
    if perm_error:
        return ExecuteResponse(
            output=f"Error: {perm_error}",
            exit_code=1,
            truncated=False,
        )

    try:
        result = subprocess.run(
            ["sh", "-c", command],
            cwd=self._root,
            capture_output=True,
            text=True,
            timeout=timeout or 120,
        )

        output = result.stdout + result.stderr

        # Truncate if too long
        max_output = 100000
        truncated = len(output) > max_output
        if truncated:  # pragma: no cover
            output = output[:max_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 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)

            if path in 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
        for file_path in files_to_search:
            lines = 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)

        if path in 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
    for file_path in files_to_search:
        lines = 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.

Example

Python

from pydantic_ai_backends import CompositeBackend, StateBackend, FilesystemBackend

backend = CompositeBackend(
    default=StateBackend(),  # Default for unmatched paths
    routes={
        "/project/": FilesystemBackend("/my/project"),
        "/workspace/": FilesystemBackend("/tmp/workspace"),
    },
)

# Routes to FilesystemBackend
backend.write("/project/app.py", "...")

# Routes to StateBackend (default)
backend.write("/temp/scratch.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.

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

        backend = CompositeBackend(
            default=StateBackend(),  # Default for unmatched paths
            routes={
                "/project/": FilesystemBackend("/my/project"),
                "/workspace/": FilesystemBackend("/tmp/workspace"),
            },
        )

        # Routes to FilesystemBackend
        backend.write("/project/app.py", "...")

        # Routes to StateBackend (default)
        backend.write("/temp/scratch.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)

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

    def ls_info(self, path: str) -> list[FileInfo]:
        """List files, aggregating from all relevant backends."""
        # If path matches a specific route, use that backend
        backend = self._get_backend(path)

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

            # First, get entries from default backend
            for entry in self._default.ls_info(path):
                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"]))

        return backend.ls_info(path)

    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
            result = self._default.grep_raw(pattern, path, glob, ignore_hidden)
            if isinstance(result, list):
                all_results.extend(result)
            elif isinstance(result, str) and result.startswith("Error"):
                pass  # Ignore errors from individual backends

            # 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)

            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."""
    # If path matches a specific route, use that backend
    backend = self._get_backend(path)

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

        # First, get entries from default backend
        for entry in self._default.ls_info(path):
            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"]))

    return backend.ls_info(path)

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
        result = self._default.grep_raw(pattern, path, glob, ignore_hidden)
        if isinstance(result, list):
            all_results.extend(result)
        elif isinstance(result, str) and result.startswith("Error"):
            pass  # Ignore errors from individual backends

        # 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)

        return all_results

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