Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion griptape/tools/file_manager/tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,10 @@ def load_files_from_disk(self, params: dict) -> ListArtifact | ErrorArtifact:
# Fetch the file to try and determine the appropriate loader
abs_path = os.path.join(self.file_manager_driver.workdir, path)
mime_type = get_mime_type(abs_path)
loader = next((loader for key, loader in self.loaders.items() if mime_type.startswith(key)))
loader = next((loader for key, loader in self.loaders.items() if mime_type.startswith(key)), None)

if loader is None:
return ErrorArtifact(f"Failed to load file '{path}': no loader available for MIME type '{mime_type}'")

artifact = loader.load(path)
if isinstance(artifact, ListArtifact):
Expand Down
12 changes: 11 additions & 1 deletion tests/unit/tools/test_file_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@

import pytest

from griptape.artifacts import ListArtifact, TextArtifact
from griptape.artifacts import ErrorArtifact, ListArtifact, TextArtifact
from griptape.drivers.file_manager.local_file_manager_driver import LocalFileManagerDriver
from griptape.tools import FileManagerTool
from tests.utils import defaults
Expand Down Expand Up @@ -44,6 +44,16 @@ def test_load_files_from_disk_with_encoding(self, file_manager):
assert len(result.value) == 1
assert isinstance(result.value[0], TextArtifact)

def test_load_files_from_disk_with_unsupported_file_type(self, temp_dir):
with open(os.path.join(temp_dir, "config.json"), "w") as f:
f.write('{"a": 1}')

file_manager = FileManagerTool(file_manager_driver=LocalFileManagerDriver(workdir=temp_dir))
result = file_manager.load_files_from_disk({"values": {"paths": ["config.json"]}})

assert isinstance(result, ErrorArtifact)
assert "no loader available" in result.value

def test_save_memory_artifacts_to_disk_for_one_artifact(self, temp_dir):
memory = defaults.text_task_memory("Memory1")
artifact = TextArtifact("foobar")
Expand Down
Loading