FilesystemToolkit

def read_file(
    self,
    base_path: str | os.PathLike[str] | None = None,
    *,
    name: str = "read_file",
    description: str = "Read the contents of a file. Returns text by default. Set raw=true to read binary content as base64.",
    middleware: Iterable[ToolMiddleware] = (),
) -> FunctionTool:
    base_dir = _resolve_dir(base_path, self.base_dir)

    @tool(name=name, description=description, middleware=middleware)
    def _read_file(
        path: Annotated[
            str,
            Field(description="Relative path to the file to read."),
        ],
        raw: bool = Field(
            default=False,
            description="If true, read the file as binary and return base64-encoded content.",
        ),
    ) -> str:
        target = _resolve_path(base_dir, path)
        if raw:
            return base64.b64encode(target.read_bytes()).decode("ascii")
        # Pin the encoding so the toolkit produces the same bytes on Linux,
        # macOS, and Windows. Without it `Path.read_text()` inherits
        # `locale.getpreferredencoding()`, which is `cp1252` on most
        # Windows installs and raises `UnicodeDecodeError` on any file
        # written with non-cp1252 content (most modern source files,
        # config files, JSON with non-ASCII strings, etc.).
        return target.read_text(encoding="utf-8")

    return _read_file

def find_files(
    self,
    base_path: str | os.PathLike[str] | None = None,
    *,
    name: str = "find_files",
    description: str = "Search for files matching a glob pattern. Use recursive patterns like '**/*.py' to search subdirectories.",
    middleware: Iterable[ToolMiddleware] = (),
) -> FunctionTool:
    base_dir = _resolve_dir(base_path, self.base_dir)

    @tool(name=name, description=description, middleware=middleware)
    def _find_files(
        pattern: Annotated[
            str,
            Field(description="Glob pattern to match files, e.g. '*.py' or '**/*.txt'."),
        ],
        path: str = Field(
            default=".",
            description="Relative path to the directory to search from. Defaults to the base directory.",
        ),
    ) -> list[str]:
        target = _resolve_path(base_dir, path)
        return sorted(str(p.relative_to(base_dir)) for p in _glob(target, pattern))

    return _find_files

def write_file(
    self,
    base_path: str | os.PathLike[str] | None = None,
    *,
    name: str = "write_file",
    description: str = "Create or overwrite a file with the given content. Parent directories are created automatically.",
    middleware: Iterable[ToolMiddleware] = (),
) -> FunctionTool:
    base_dir = _resolve_dir(base_path, self.base_dir)

    @tool(name=name, description=description, middleware=middleware)
    def _write_file(
        path: Annotated[
            str,
            Field(description="Relative path to the file to create or overwrite."),
        ],
        content: Annotated[
            str,
            Field(description="The full content to write to the file."),
        ],
    ) -> str:
        target = _resolve_path(base_dir, path)
        target.parent.mkdir(parents=True, exist_ok=True)
        # Encode the payload as UTF-8 so the toolkit round-trips any
        # text the model writes (smart quotes, emoji, non-Latin scripts,
        # code with non-ASCII docstrings). The platform-default
        # encoding is `cp1252` on most Windows machines and silently
        # raises `UnicodeEncodeError` mid-write.
        target.write_text(content, encoding="utf-8")
        return f"Successfully wrote {len(content)} characters to {path}"

    return _write_file

def update_file(
    self,
    base_path: str | os.PathLike[str] | None = None,
    *,
    name: str = "update_file",
    description: str = "Update a file by replacing the first occurrence of old_content with new_content.",
    middleware: Iterable[ToolMiddleware] = (),
) -> FunctionTool:
    base_dir = _resolve_dir(base_path, self.base_dir)

    @tool(name=name, description=description, middleware=middleware)
    def _update_file(
        path: Annotated[
            str,
            Field(description="Relative path to the file to update."),
        ],
        old_content: Annotated[
            str,
            Field(description="The exact text to find and replace."),
        ],
        new_content: Annotated[
            str,
            Field(description="The text to replace old_content with."),
        ],
    ) -> str:
        target = _resolve_path(base_dir, path)
        # See _read_file / _write_file: pin UTF-8 so update_file reads
        # back the exact bytes it (or any other UTF-8 writer) produced,
        # and writes the modified payload back in the same encoding.
        # Without this, the read/write pair on Windows would silently
        # transcode the file into cp1252 every time the model edits it.
        text = target.read_text(encoding="utf-8")
        if old_content not in text:
            raise ValueError(f"old_content not found in {path}")
        target.write_text(text.replace(old_content, new_content, 1), encoding="utf-8")
        return f"Successfully updated {path}"

    return _update_file

def delete_file(
    self,
    base_path: str | os.PathLike[str] | None = None,
    *,
    name: str = "delete_file",
    description: str = "Delete a file or directory.",
    middleware: Iterable[ToolMiddleware] = (),
) -> FunctionTool:
    base_dir = _resolve_dir(base_path, self.base_dir)

    @tool(name=name, description=description, middleware=middleware)
    def _delete_file(
        path: Annotated[
            str,
            Field(description="Relative path to the file or directory to delete."),
        ],
    ) -> str:
        target = _resolve_path(base_dir, path)
        if target.is_dir():
            shutil.rmtree(target)
        else:
            target.unlink()
        return f"Successfully deleted {path}"

    return _delete_file

def set_provider(self, provider: Provider) -> None:
    for t in self.tools:
        t.set_provider(provider)

async def schemas(self, context: "Context") -> Iterable[ToolSchema]:
    schemas: list[ToolSchema] = []
    for t in self.tools:
        schemas.extend(await t.schemas(context))
    return schemas

def register(
    self,
    stack: "ExitStack | AsyncExitStack",
    context: "Context",
    *,
    middleware: Iterable["BaseMiddleware"] = (),
) -> None:
    for t in self.tools:
        t.register(stack, context, middleware=middleware)

def tool(
    self,
    function: Callable[..., Any] | None = None,
    *,
    name: str | None = None,
    description: str | None = None,
    schema: FunctionParameters | None = None,
    sync_to_thread: bool = True,
    middleware: Iterable[ToolMiddleware] = (),
) -> FunctionTool | Callable[[Callable[..., Any]], FunctionTool]:
    def make_tool(f: Callable[..., Any]) -> FunctionTool:
        t = tool(
            f,
            name=name,
            description=description,
            schema=schema,
            sync_to_thread=sync_to_thread,
            middleware=middleware,
        )
        self._add_tool(t)
        return t

    if function:
        return make_tool(function)

    return make_tool