Add FileManager.list_files_from_disk

CHANGELOG.md

@@ -6,6 +6,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+### Added
+- `list_files_from_disk` activity to `FileManager` Tool.
+
 ### Changed
 - Improved RAG performance in `VectorQueryEngine`.
 - **BREAKING**: Secret fields (ex: api_key) removed from serialized Drivers.

griptape/tools/file_manager/tool.py

@@ -1,10 +1,9 @@
 from __future__ import annotations
-
 import logging
 import os
 from pathlib import Path
 from attr import define, field, Factory
-from griptape.artifacts import ErrorArtifact, InfoArtifact, ListArtifact, BaseArtifact
+from griptape.artifacts import ErrorArtifact, InfoArtifact, ListArtifact, BaseArtifact, TextArtifact
 from griptape.tools import BaseTool
 from griptape.utils.decorators import activity
 from griptape.loaders import FileLoader, BaseLoader, PdfLoader, CsvLoader, TextLoader, ImageLoader
@@ -54,14 +53,33 @@ def validate_workdir(self, _, workdir: str) -> None:
         if not Path(workdir).is_absolute():
             raise ValueError("workdir has to be absolute absolute")
 
+    @activity(
+        config={
+            "description": "Can be used to list files on disk",
+            "schema": Schema(
+                {Literal("path", description="Relative path in the POSIX format. For example, 'foo/bar'"): str}
+            ),
+        }
+    )
+    def list_files_from_disk(self, params: dict) -> TextArtifact | ErrorArtifact:
+        path = params["values"]["path"].lstrip("/")
+        full_path = Path(os.path.join(self.workdir, path))
+
+        if os.path.exists(full_path):
+            entries = os.listdir(full_path)
+
+            return TextArtifact("\n".join([e for e in entries]))
+        else:
+            return ErrorArtifact("Path not found")
+
     @activity(
         config={
             "description": "Can be used to load files from disk",
             "schema": Schema(
                 {
                     Literal(
                         "paths",
-                        description="Paths to files to be loaded in the POSIX format. For example, ['foo/bar/file.txt']",
+                        description="Relative paths to files to be loaded in the POSIX format. For example, ['foo/bar/file.txt']",
                     ): []
                 }
             ),
@@ -71,6 +89,7 @@ def load_files_from_disk(self, params: dict) -> ListArtifact | ErrorArtifact:
         artifacts = []
 
         for path in params["values"]["paths"]:
+            path = path.lstrip("/")
             full_path = Path(os.path.join(self.workdir, path))
             extension = path.split(".")[-1]
             loader = self.loaders.get(extension) or self.default_loader
@@ -92,7 +111,7 @@ def load_files_from_disk(self, params: dict) -> ListArtifact | ErrorArtifact:
                 {
                     Literal(
                         "dir_name",
-                        description="Destination directory name on disk in the POSIX format. For example, 'foo/bar'",
+                        description="Relative destination path name on disk in the POSIX format. For example, 'foo/bar'",
                     ): str,
                     Literal("file_name", description="Destination file name. For example, 'baz.txt'"): str,
                     "memory_name": str,
@@ -146,7 +165,7 @@ def save_memory_artifacts_to_disk(self, params: dict) -> ErrorArtifact | InfoArt
     )
     def save_content_to_file(self, params: dict) -> ErrorArtifact | InfoArtifact:
         content = params["values"]["content"]
-        new_path = params["values"]["path"]
+        new_path = params["values"]["path"].lstrip("/")
         full_path = os.path.join(self.workdir, new_path)
 
         try:

tests/unit/tools/test_file_manager.py

@@ -15,6 +15,15 @@ def test_validate_workdir(self):
         with pytest.raises(ValueError):
             FileManager(workdir="foo")
 
+    def test_list_files_from_disk(self):
+        result = FileManager(
+            input_memory=[defaults.text_task_memory("Memory1")], workdir=os.path.abspath(os.path.dirname(__file__))
+        ).list_files_from_disk({"values": {"path": "../../resources"}})
+
+        assert isinstance(result, TextArtifact)
+        assert "bitcoin.pdf" in result.value
+        assert "small.png" in result.value
+
     def test_load_files_from_disk(self):
         result = FileManager(
             input_memory=[defaults.text_task_memory("Memory1")], workdir=os.path.abspath(os.path.dirname(__file__))