Toolsets API

create_console_toolset

pydantic_ai_backends.toolsets.console.create_console_toolset(id=None, include_execute=True, require_write_approval=False, require_execute_approval=True, default_ignore_hidden=True, permissions=None, max_retries=1, image_support=False, max_image_bytes=DEFAULT_MAX_IMAGE_BYTES, document_support=False, max_document_bytes=DEFAULT_MAX_DOCUMENT_BYTES, edit_format='str_replace', descriptions=None)

Create a console toolset for file operations and shell execution.

This toolset provides tools for interacting with the filesystem and executing shell commands. It works with any backend that implements BackendProtocol (LocalBackend, DockerSandbox, StateBackend, etc.)

Parameters:

Name	Type	Description	Default
id	str | None	Optional unique ID for the toolset.	None
include_execute	bool	Whether to include the execute tool. Requires backend to have execute() method.	True
require_write_approval	bool	Whether write_file and edit_file require approval. Ignored if permissions is provided.	False
require_execute_approval	bool	Whether execute requires approval. Ignored if permissions is provided.	True
default_ignore_hidden	bool	Default behavior for grep regarding hidden files.	True
permissions	PermissionRuleset | None	Optional permission ruleset to determine tool approval requirements. If provided, overrides require_write_approval and require_execute_approval based on whether the operation's default action is "ask".	None
max_retries	int	Maximum number of retries for each tool during a run. When the model sends invalid arguments (e.g. missing required fields), the validation error is fed back and the model can retry up to this many times. Defaults to 1.	1
image_support	bool	Whether to enable image file handling in read_file. When True, reading image files (.png, .jpg, .jpeg, .gif, .webp) returns a BinaryContent object that multimodal models can see, instead of garbled text. Defaults to False.	False
max_image_bytes	int	Maximum image file size in bytes (default: 50MB). Images larger than this will return an error message instead. Only used when image_support is True.	DEFAULT_MAX_IMAGE_BYTES
document_support	bool	Whether to enable binary document handling in read_file. When True, reading document files (.pdf) returns a BinaryContent object that models capable of document understanding (OpenAI/Anthropic/Gemini) can read, instead of the empty/garbled text produced by reading a binary file as text. Kept separate from image_support so document handling can evolve independently (e.g. OCR or text-extraction fallbacks). Defaults to False.	False
max_document_bytes	int	Maximum document file size in bytes (default: 50MB). Documents larger than this will return an error message instead. Only used when document_support is True.	DEFAULT_MAX_DOCUMENT_BYTES
edit_format	EditFormat	File editing format to use. "str_replace" (default) uses exact string matching. "hashline" tags each line with a content hash so models can reference lines by number:hash instead of reproducing text.	'str_replace'
descriptions	dict[str, str] | None	Optional mapping of tool name to custom description override. When provided, the description for a tool is looked up as descriptions.get("tool_name", DEFAULT_DESCRIPTION). Valid keys are: ls, read_file, write_file, edit_file, hashline_edit, glob, grep, execute.	None

Returns:

Type	Description
FunctionToolset[ConsoleDeps]	FunctionToolset with console tools.

Example

Python

from dataclasses import dataclass
from pydantic_ai_backends import LocalBackend, create_console_toolset

@dataclass
class MyDeps:
    backend: LocalBackend

toolset = create_console_toolset()
deps = MyDeps(backend=LocalBackend("/workspace"))

# With hashline edit format (better accuracy, fewer tokens)
toolset = create_console_toolset(edit_format="hashline")

# With image support for multimodal models
toolset = create_console_toolset(image_support=True)

# With document (PDF) support for document-understanding models
toolset = create_console_toolset(document_support=True)

# Or with permissions
from pydantic_ai_backends.permissions import DEFAULT_RULESET

toolset = create_console_toolset(permissions=DEFAULT_RULESET)

Source code in src/pydantic_ai_backends/toolsets/console.py

def create_console_toolset(  # noqa: C901
    id: str | None = None,
    include_execute: bool = True,
    require_write_approval: bool = False,
    require_execute_approval: bool = True,
    default_ignore_hidden: bool = True,
    permissions: PermissionRuleset | None = None,
    max_retries: int = 1,
    image_support: bool = False,
    max_image_bytes: int = DEFAULT_MAX_IMAGE_BYTES,
    document_support: bool = False,
    max_document_bytes: int = DEFAULT_MAX_DOCUMENT_BYTES,
    edit_format: EditFormat = "str_replace",
    descriptions: dict[str, str] | None = None,
) -> FunctionToolset[ConsoleDeps]:
    """Create a console toolset for file operations and shell execution.

    This toolset provides tools for interacting with the filesystem and
    executing shell commands. It works with any backend that implements
    BackendProtocol (LocalBackend, DockerSandbox, StateBackend, etc.)

    Args:
        id: Optional unique ID for the toolset.
        include_execute: Whether to include the execute tool.
            Requires backend to have execute() method.
        require_write_approval: Whether write_file and edit_file require approval.
            Ignored if permissions is provided.
        require_execute_approval: Whether execute requires approval.
            Ignored if permissions is provided.
        default_ignore_hidden: Default behavior for grep regarding hidden files.
        permissions: Optional permission ruleset to determine tool approval requirements.
            If provided, overrides require_write_approval and require_execute_approval
            based on whether the operation's default action is "ask".
        max_retries: Maximum number of retries for each tool during a run.
            When the model sends invalid arguments (e.g. missing required fields),
            the validation error is fed back and the model can retry up to this
            many times. Defaults to 1.
        image_support: Whether to enable image file handling in read_file.
            When True, reading image files (.png, .jpg, .jpeg, .gif, .webp) returns
            a BinaryContent object that multimodal models can see, instead of garbled
            text. Defaults to False.
        max_image_bytes: Maximum image file size in bytes (default: 50MB).
            Images larger than this will return an error message instead.
            Only used when image_support is True.
        document_support: Whether to enable binary document handling in read_file.
            When True, reading document files (.pdf) returns a BinaryContent object
            that models capable of document understanding (OpenAI/Anthropic/Gemini)
            can read, instead of the empty/garbled text produced by reading a binary
            file as text. Kept separate from image_support so document handling can
            evolve independently (e.g. OCR or text-extraction fallbacks).
            Defaults to False.
        max_document_bytes: Maximum document file size in bytes (default: 50MB).
            Documents larger than this will return an error message instead.
            Only used when document_support is True.
        edit_format: File editing format to use.  `"str_replace"` (default) uses
            exact string matching.  `"hashline"` tags each line with a content hash
            so models can reference lines by number:hash instead of reproducing text.
        descriptions: Optional mapping of tool name to custom description override.
            When provided, the description for a tool is looked up as
            `descriptions.get("tool_name", DEFAULT_DESCRIPTION)`.  Valid keys are:
            `ls`, `read_file`, `write_file`, `edit_file`, `hashline_edit`,
            `glob`, `grep`, `execute`.

    Returns:
        FunctionToolset with console tools.

    Example:
        ```python
        from dataclasses import dataclass
        from pydantic_ai_backends import LocalBackend, create_console_toolset

        @dataclass
        class MyDeps:
            backend: LocalBackend

        toolset = create_console_toolset()
        deps = MyDeps(backend=LocalBackend("/workspace"))

        # With hashline edit format (better accuracy, fewer tokens)
        toolset = create_console_toolset(edit_format="hashline")

        # With image support for multimodal models
        toolset = create_console_toolset(image_support=True)

        # With document (PDF) support for document-understanding models
        toolset = create_console_toolset(document_support=True)

        # Or with permissions
        from pydantic_ai_backends.permissions import DEFAULT_RULESET

        toolset = create_console_toolset(permissions=DEFAULT_RULESET)
        ```
    """

    _descs = descriptions or {}

    # Determine approval requirements and denied operations
    write_approval = _requires_approval_from_ruleset(permissions, "write", require_write_approval)
    execute_approval = _requires_approval_from_ruleset(
        permissions, "execute", require_execute_approval
    )
    # Track denied operations to remove tools after registration
    _denied_tools: set[str] = set()
    if _is_denied_by_ruleset(permissions, "write"):
        _denied_tools.add("write_file")
    if _is_denied_by_ruleset(permissions, "edit"):
        _denied_tools.update({"edit_file", "hashline_edit"})
    if _is_denied_by_ruleset(permissions, "execute"):
        _denied_tools.add("execute")

    toolset: FunctionToolset[ConsoleDeps] = FunctionToolset(id=id, max_retries=max_retries)

    @toolset.tool(description=_descs.get("ls", LS_DESCRIPTION))
    async def ls(  # pragma: no cover
        ctx: RunContext[ConsoleDeps],
        path: str = ".",
    ) -> str:
        """List files and directories at the given path.

        Args:
            path: Directory path to list. Defaults to current directory.
        """
        entries = await asyncio.to_thread(ctx.deps.backend.ls_info, path)

        if not entries:
            return f"Directory '{path}' is empty or does not exist"

        lines = [f"Contents of {path}:"]
        for entry in entries:
            if entry["is_dir"]:
                lines.append(f"  {entry['name']}/")
            else:
                size = entry.get("size")
                size_str = f" ({size} bytes)" if size is not None else ""
                lines.append(f"  {entry['name']}{size_str}")

        return "\n".join(lines)

    # --- read_file tool ---
    if edit_format == "hashline":

        @toolset.tool(description=_descs.get("read_file", HASHLINE_READ_FILE_DESCRIPTION))
        async def read_file(  # pragma: no cover
            ctx: RunContext[ConsoleDeps],
            path: str,
            offset: int = 0,
            limit: int = 2000,
        ) -> Any:
            """Read file content with hashline tags.

            Args:
                path: Absolute or relative path to the file to read.
                offset: Line number to start reading from (0-indexed).
                limit: Maximum number of lines to read. Defaults to 2000.
            """
            if image_support:
                image = await _maybe_image_content(ctx.deps.backend, path, max_image_bytes)
                if image is not None:
                    return image
            if document_support:
                document = await _maybe_document_content(ctx.deps.backend, path, max_document_bytes)
                if document is not None:
                    return document

            from pydantic_ai_backends.hashline import format_hashline_output

            if not await asyncio.to_thread(ctx.deps.backend.exists, path):
                return f"Error: File '{path}' not found"
            raw_bytes = await asyncio.to_thread(ctx.deps.backend.read_bytes, path)
            text = raw_bytes.decode("utf-8", errors="replace")
            return format_hashline_output(text, offset, limit)

    else:

        @toolset.tool(description=_descs.get("read_file", READ_FILE_DESCRIPTION))
        async def read_file(  # pragma: no cover
            ctx: RunContext[ConsoleDeps],
            path: str,
            offset: int = 0,
            limit: int = 2000,
        ) -> Any:
            """Read file content with line numbers.

            Args:
                path: Absolute or relative path to the file to read.
                offset: Line number to start reading from (0-indexed).
                limit: Maximum number of lines to read. Defaults to 2000.
            """
            if image_support:
                image = await _maybe_image_content(ctx.deps.backend, path, max_image_bytes)
                if image is not None:
                    return image
            if document_support:
                document = await _maybe_document_content(ctx.deps.backend, path, max_document_bytes)
                if document is not None:
                    return document
            return await asyncio.to_thread(ctx.deps.backend.read, path, offset, limit)

    # --- write_file tool ---
    @toolset.tool(
        description=_descs.get("write_file", WRITE_FILE_DESCRIPTION),
        requires_approval=write_approval,
    )
    async def write_file(  # pragma: no cover
        ctx: RunContext[ConsoleDeps],
        path: str,
        content: str,
    ) -> str:
        """Write content to a file.

        Args:
            path: Path to the file to write.
            content: Complete content to write to the file.
        """
        result = await asyncio.to_thread(ctx.deps.backend.write, path, content)

        if result.error:
            return f"Error: {result.error}"

        lines = len(content.splitlines())
        return f"Wrote {lines} lines to {result.path}"

    # --- edit tool (str_replace or hashline) ---
    if edit_format == "hashline":

        @toolset.tool(
            description=_descs.get("hashline_edit", HASHLINE_EDIT_DESCRIPTION),
            requires_approval=write_approval,
        )
        async def hashline_edit(  # pragma: no cover
            ctx: RunContext[ConsoleDeps],
            path: str,
            start_line: int,
            start_hash: str,
            new_content: str,
            end_line: int | None = None,
            end_hash: str | None = None,
            insert_after: bool = False,
        ) -> str:
            """Edit a file by referencing lines with their content hashes.

            Args:
                path: Path to the file to edit.
                start_line: 1-indexed line number to start the edit.
                start_hash: 2-char content hash of the start line (from read_file).
                new_content: Replacement text. Empty string deletes line(s).
                end_line: 1-indexed end of range (inclusive). Omit for single-line edit.
                end_hash: 2-char content hash of the end line. Optional validation.
                insert_after: If True, insert new_content after start_line instead \
of replacing it.
            """
            from pydantic_ai_backends.hashline import apply_hashline_edit_with_summary

            backend = ctx.deps.backend
            per_path = _edit_locks.setdefault(backend, {})
            if path not in per_path:
                per_path[path] = asyncio.Lock()
            async with per_path[path]:
                # Read current file content
                if not await asyncio.to_thread(backend.exists, path):
                    return f"Error: File '{path}' not found"
                raw_bytes = await asyncio.to_thread(backend.read_bytes, path)

                text = raw_bytes.decode("utf-8", errors="replace")

                # Apply edit
                new_text, error, summary = apply_hashline_edit_with_summary(
                    text,
                    start_line,
                    start_hash,
                    new_content,
                    end_line,
                    end_hash,
                    insert_after,
                )

                if error:
                    return f"Error: {error}"

                # Write back
                write_result = await asyncio.to_thread(backend.write, path, new_text)
                if write_result.error:
                    return f"Error: {write_result.error}"

                return f"Edited {write_result.path}: {summary}"

    else:

        @toolset.tool(
            description=_descs.get("edit_file", EDIT_FILE_DESCRIPTION),
            requires_approval=write_approval,
        )
        async def edit_file(  # pragma: no cover
            ctx: RunContext[ConsoleDeps],
            path: str,
            old_string: str,
            new_string: str,
            replace_all: bool = False,
        ) -> str:
            """Edit a file by performing exact string replacement.

            Args:
                path: Path to the file to edit.
                old_string: Exact string to find and replace. Must match file content exactly \
including whitespace and indentation.
                new_string: Replacement string. Must be different from old_string.
                replace_all: If True, replace all occurrences. If False (default), \
the old_string must appear exactly once in the file.
            """
            result = await asyncio.to_thread(
                ctx.deps.backend.edit, path, old_string, new_string, replace_all
            )

            if result.error:
                return f"Error: {result.error}"

            return f"Edited {result.path}: replaced {result.occurrences} occurrence(s)"

    @toolset.tool(description=_descs.get("glob", GLOB_DESCRIPTION))
    async def glob(  # pragma: no cover
        ctx: RunContext[ConsoleDeps],
        pattern: str,
        path: str = ".",
    ) -> str:
        """Find files matching a glob pattern.

        Args:
            pattern: Glob pattern to match.
            path: Base directory to search from. Defaults to current directory.
        """
        entries = await asyncio.to_thread(ctx.deps.backend.glob_info, pattern, path)

        if not entries:
            return f"No files matching '{pattern}' in {path}"

        lines = [f"Found {len(entries)} file(s) matching '{pattern}':"]
        for entry in entries[:100]:
            lines.append(f"  {entry['path']}")

        if len(entries) > 100:
            lines.append(f"  ... and {len(entries) - 100} more")

        return "\n".join(lines)

    @toolset.tool(description=_descs.get("grep", GREP_DESCRIPTION))
    async def grep(  # pragma: no cover
        ctx: RunContext[ConsoleDeps],
        pattern: str,
        path: str | None = None,
        glob_pattern: str | None = None,
        output_mode: Literal["content", "files_with_matches", "count"] = "files_with_matches",
        ignore_hidden: bool = default_ignore_hidden,
    ) -> str:
        """Search for a regex pattern across files.

        Args:
            pattern: Regex pattern to search for.
            path: File or directory to search in. If None, searches current directory.
            glob_pattern: Filter files by pattern (e.g., `"*.py"`, `"*.{js,ts}"`).
            output_mode: Output format — `"content"`, `"files_with_matches"`, or `"count"`.
            ignore_hidden: Whether to skip hidden files/directories.
        """
        result = await asyncio.to_thread(
            ctx.deps.backend.grep_raw, pattern, path, glob_pattern, ignore_hidden
        )

        if isinstance(result, str):
            return result  # Error message

        if not result:
            return f"No matches for '{pattern}'"

        matches: list[GrepMatch] = result

        if output_mode == "count":
            return f"Found {len(matches)} match(es) for '{pattern}'"

        if output_mode == "files_with_matches":
            files = sorted(set(m["path"] for m in matches))
            lines = [f"Files containing '{pattern}':"]
            for f in files[:50]:
                lines.append(f"  {f}")
            if len(files) > 50:
                lines.append(f"  ... and {len(files) - 50} more files")
            return "\n".join(lines)

        # content mode
        lines = [f"Matches for '{pattern}':"]
        for m in matches[:50]:
            lines.append(f"  {m['path']}:{m['line_number']}: {m['line'][:100]}")
        if len(matches) > 50:
            lines.append(f"  ... and {len(matches) - 50} more matches")
        return "\n".join(lines)

    # Expose references for testing
    cast(_ConsoleToolsetTestAttrs, toolset)._console_default_ignore_hidden = default_ignore_hidden
    cast(_ConsoleToolsetTestAttrs, toolset)._console_grep_impl = grep

    if include_execute:

        @toolset.tool(
            description=_descs.get("execute", EXECUTE_DESCRIPTION),
            requires_approval=execute_approval,
        )
        async def execute(  # pragma: no cover
            ctx: RunContext[ConsoleDeps],
            command: str,
            timeout: int | None = 120,
        ) -> str:
            """Execute a shell command in the working directory.

            Args:
                command: Shell command to execute.
                timeout: Maximum execution time in seconds. Default 120. Increase \
for long-running builds or test suites.
            """
            backend = ctx.deps.backend

            # Check if backend supports execute
            if not hasattr(backend, "execute"):
                return "Error: Backend does not support command execution"

            # Check if execute is enabled (for LocalBackend)
            if hasattr(backend, "execute_enabled") and not backend.execute_enabled:  # pyright: ignore[reportAttributeAccessIssue]
                return "Error: Shell execution is disabled for this backend"

            try:
                if hasattr(backend, "async_execute"):
                    result = await backend.async_execute(command, timeout)  # pyright: ignore[reportAttributeAccessIssue]
                else:
                    result = await asyncio.to_thread(backend.execute, command, timeout)  # pyright: ignore[reportAttributeAccessIssue]
            except RuntimeError as e:
                return f"Error: {e}"

            output = result.output
            if result.truncated:
                output += "\n\n... (output truncated)"

            if result.exit_code is not None and result.exit_code != 0:
                return f"Command failed (exit code {result.exit_code}):\n{output}"

            return str(output)

    # Remove tools for denied operations (fixes issue #23)
    for tool_name in _denied_tools:
        toolset.tools.pop(tool_name, None)

    return toolset

get_console_system_prompt

pydantic_ai_backends.toolsets.console.get_console_system_prompt(edit_format='str_replace')

Get the system prompt for console tools.

Parameters:

Name	Type	Description	Default
edit_format	EditFormat	Which edit format to describe in the prompt.	'str_replace'

Returns:

Type	Description
str	System prompt describing available console tools.

Source code in src/pydantic_ai_backends/toolsets/console.py

def get_console_system_prompt(edit_format: EditFormat = "str_replace") -> str:
    """Get the system prompt for console tools.

    Args:
        edit_format: Which edit format to describe in the prompt.

    Returns:
        System prompt describing available console tools.
    """
    if edit_format == "hashline":
        return HASHLINE_CONSOLE_PROMPT
    return CONSOLE_SYSTEM_PROMPT

ConsoleDeps

pydantic_ai_backends.toolsets.console.ConsoleDeps

Bases: Protocol

Protocol for dependencies that provide a backend.

Source code in src/pydantic_ai_backends/toolsets/console.py

@runtime_checkable
class ConsoleDeps(Protocol):
    """Protocol for dependencies that provide a backend."""

    @property
    def backend(self) -> BackendProtocol:
        """The backend for file operations."""
        ...

backend property

The backend for file operations.

Console Tools

The toolset provides these tools:

ls

Python

async def ls(ctx: RunContext[ConsoleDeps], path: str = ".") -> str:
    """List files and directories at the given path.

    Args:
        path: Directory path to list. Defaults to current directory.
    """

read_file

Python

async def read_file(
    ctx: RunContext[ConsoleDeps],
    path: str,
    offset: int = 0,
    limit: int = 2000,
) -> str:
    """Read file content with line numbers.

    Args:
        path: Path to the file to read.
        offset: Line number to start reading from (0-indexed).
        limit: Maximum number of lines to read.
    """

write_file

Python

async def write_file(
    ctx: RunContext[ConsoleDeps],
    path: str,
    content: str,
) -> str:
    """Write content to a file (creates or overwrites).

    Args:
        path: Path to the file to write.
        content: Content to write to the file.
    """

edit_file

Python

async def edit_file(
    ctx: RunContext[ConsoleDeps],
    path: str,
    old_string: str,
    new_string: str,
    replace_all: bool = False,
) -> str:
    """Edit a file by replacing strings.

    Args:
        path: Path to the file to edit.
        old_string: String to find and replace.
        new_string: Replacement string.
        replace_all: If True, replace all occurrences.
    """

glob

Python

async def glob(
    ctx: RunContext[ConsoleDeps],
    pattern: str,
    path: str = ".",
) -> str:
    """Find files matching a glob pattern.

    Args:
        pattern: Glob pattern to match (e.g., "**/*.py").
        path: Base directory to search from.
    """

grep

Python

async def grep(
    ctx: RunContext[ConsoleDeps],
    pattern: str,
    path: str | None = None,
    glob_pattern: str | None = None,
    output_mode: Literal["content", "files_with_matches", "count"] = "files_with_matches",
    ignore_hidden: bool = True,
) -> str:
    """Search for a regex pattern in files.

    Args:
        pattern: Regex pattern to search for.
        path: Specific file or directory to search.
        glob_pattern: Glob pattern to filter files.
        output_mode: Output format.
        ignore_hidden: Whether to skip hidden files (defaults to the toolset setting).
    """

execute

Python

async def execute(
    ctx: RunContext[ConsoleDeps],
    command: str,
    timeout: int | None = 120,
) -> str:
    """Execute a shell command.

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