mirror of
https://github.com/crewAIInc/crewAI.git
synced 2026-01-09 08:08:32 +00:00
Merge pull request #162 from crewAIInc/devin/1735422935-file-read-tool-fix
Fix FileReadTool infinite loop by maintaining original schema
This commit is contained in:
@@ -4,39 +4,78 @@ from crewai.tools import BaseTool
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class FixedFileReadToolSchema(BaseModel):
|
||||
"""Input for FileReadTool."""
|
||||
|
||||
pass
|
||||
|
||||
|
||||
class FileReadToolSchema(FixedFileReadToolSchema):
|
||||
class FileReadToolSchema(BaseModel):
|
||||
"""Input for FileReadTool."""
|
||||
|
||||
file_path: str = Field(..., description="Mandatory file full path to read the file")
|
||||
|
||||
|
||||
class FileReadTool(BaseTool):
|
||||
"""A tool for reading file contents.
|
||||
|
||||
This tool inherits its schema handling from BaseTool to avoid recursive schema
|
||||
definition issues. The args_schema is set to FileReadToolSchema which defines
|
||||
the required file_path parameter. The schema should not be overridden in the
|
||||
constructor as it would break the inheritance chain and cause infinite loops.
|
||||
|
||||
The tool supports two ways of specifying the file path:
|
||||
1. At construction time via the file_path parameter
|
||||
2. At runtime via the file_path parameter in the tool's input
|
||||
|
||||
Args:
|
||||
file_path (Optional[str]): Path to the file to be read. If provided,
|
||||
this becomes the default file path for the tool.
|
||||
**kwargs: Additional keyword arguments passed to BaseTool.
|
||||
|
||||
Example:
|
||||
>>> tool = FileReadTool(file_path="/path/to/file.txt")
|
||||
>>> content = tool.run() # Reads /path/to/file.txt
|
||||
>>> content = tool.run(file_path="/path/to/other.txt") # Reads other.txt
|
||||
"""
|
||||
name: str = "Read a file's content"
|
||||
description: str = "A tool that can be used to read a file's content."
|
||||
description: str = "A tool that reads the content of a file. To use this tool, provide a 'file_path' parameter with the path to the file you want to read."
|
||||
args_schema: Type[BaseModel] = FileReadToolSchema
|
||||
file_path: Optional[str] = None
|
||||
|
||||
def __init__(self, file_path: Optional[str] = None, **kwargs):
|
||||
def __init__(self, file_path: Optional[str] = None, **kwargs: Any) -> None:
|
||||
"""Initialize the FileReadTool.
|
||||
|
||||
Args:
|
||||
file_path (Optional[str]): Path to the file to be read. If provided,
|
||||
this becomes the default file path for the tool.
|
||||
**kwargs: Additional keyword arguments passed to BaseTool.
|
||||
"""
|
||||
super().__init__(**kwargs)
|
||||
if file_path is not None:
|
||||
self.file_path = file_path
|
||||
self.description = f"A tool that can be used to read {file_path}'s content."
|
||||
self.args_schema = FixedFileReadToolSchema
|
||||
self._generate_description()
|
||||
self.description = f"A tool that reads file content. The default file is {file_path}, but you can provide a different 'file_path' parameter to read another file."
|
||||
|
||||
def _run(
|
||||
self,
|
||||
**kwargs: Any,
|
||||
) -> Any:
|
||||
) -> str:
|
||||
file_path = kwargs.get("file_path", self.file_path)
|
||||
if file_path is None:
|
||||
return "Error: No file path provided. Please provide a file path either in the constructor or as an argument."
|
||||
|
||||
try:
|
||||
file_path = kwargs.get("file_path", self.file_path)
|
||||
with open(file_path, "r") as file:
|
||||
return file.read()
|
||||
except FileNotFoundError:
|
||||
return f"Error: File not found at path: {file_path}"
|
||||
except PermissionError:
|
||||
return f"Error: Permission denied when trying to read file: {file_path}"
|
||||
except Exception as e:
|
||||
return f"Fail to read the file {file_path}. Error: {e}"
|
||||
return f"Error: Failed to read file {file_path}. {str(e)}"
|
||||
|
||||
def _generate_description(self) -> None:
|
||||
"""Generate the tool description based on file path.
|
||||
|
||||
This method updates the tool's description to include information about
|
||||
the default file path while maintaining the ability to specify a different
|
||||
file at runtime.
|
||||
|
||||
Returns:
|
||||
None
|
||||
"""
|
||||
self.description = f"A tool that can be used to read {self.file_path}'s content."
|
||||
|
||||
84
tests/file_read_tool_test.py
Normal file
84
tests/file_read_tool_test.py
Normal file
@@ -0,0 +1,84 @@
|
||||
import os
|
||||
import pytest
|
||||
from crewai_tools import FileReadTool
|
||||
|
||||
def test_file_read_tool_constructor():
|
||||
"""Test FileReadTool initialization with file_path."""
|
||||
# Create a temporary test file
|
||||
test_file = "/tmp/test_file.txt"
|
||||
test_content = "Hello, World!"
|
||||
with open(test_file, "w") as f:
|
||||
f.write(test_content)
|
||||
|
||||
# Test initialization with file_path
|
||||
tool = FileReadTool(file_path=test_file)
|
||||
assert tool.file_path == test_file
|
||||
assert "test_file.txt" in tool.description
|
||||
|
||||
# Clean up
|
||||
os.remove(test_file)
|
||||
|
||||
def test_file_read_tool_run():
|
||||
"""Test FileReadTool _run method with file_path at runtime."""
|
||||
# Create a temporary test file
|
||||
test_file = "/tmp/test_file.txt"
|
||||
test_content = "Hello, World!"
|
||||
with open(test_file, "w") as f:
|
||||
f.write(test_content)
|
||||
|
||||
# Test reading file with runtime file_path
|
||||
tool = FileReadTool()
|
||||
result = tool._run(file_path=test_file)
|
||||
assert result == test_content
|
||||
|
||||
# Clean up
|
||||
os.remove(test_file)
|
||||
|
||||
def test_file_read_tool_error_handling():
|
||||
"""Test FileReadTool error handling."""
|
||||
# Test missing file path
|
||||
tool = FileReadTool()
|
||||
result = tool._run()
|
||||
assert "Error: No file path provided" in result
|
||||
|
||||
# Test non-existent file
|
||||
result = tool._run(file_path="/nonexistent/file.txt")
|
||||
assert "Error: File not found at path:" in result
|
||||
|
||||
# Test permission error (create a file without read permissions)
|
||||
test_file = "/tmp/no_permission.txt"
|
||||
with open(test_file, "w") as f:
|
||||
f.write("test")
|
||||
os.chmod(test_file, 0o000)
|
||||
|
||||
result = tool._run(file_path=test_file)
|
||||
assert "Error: Permission denied" in result
|
||||
|
||||
# Clean up
|
||||
os.chmod(test_file, 0o666) # Restore permissions to delete
|
||||
os.remove(test_file)
|
||||
|
||||
def test_file_read_tool_constructor_and_run():
|
||||
"""Test FileReadTool using both constructor and runtime file paths."""
|
||||
# Create two test files
|
||||
test_file1 = "/tmp/test1.txt"
|
||||
test_file2 = "/tmp/test2.txt"
|
||||
content1 = "File 1 content"
|
||||
content2 = "File 2 content"
|
||||
|
||||
with open(test_file1, "w") as f1, open(test_file2, "w") as f2:
|
||||
f1.write(content1)
|
||||
f2.write(content2)
|
||||
|
||||
# Test that constructor file_path works
|
||||
tool = FileReadTool(file_path=test_file1)
|
||||
result = tool._run()
|
||||
assert result == content1
|
||||
|
||||
# Test that runtime file_path overrides constructor
|
||||
result = tool._run(file_path=test_file2)
|
||||
assert result == content2
|
||||
|
||||
# Clean up
|
||||
os.remove(test_file1)
|
||||
os.remove(test_file2)
|
||||
Reference in New Issue
Block a user