feat: add read_many_files tool for batch file reading
Reduces LLM round-trips by allowing multiple files to be read in a single tool call. Uses best-effort error handling so partial failures still return successful reads. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -23,6 +23,12 @@ class ReadFileParams(BaseModel):
|
||||
file_path: str = Field(description="Path to the file to read (relative to workspace root)")
|
||||
|
||||
|
||||
class ReadManyFilesParams(BaseModel):
|
||||
"""Parameters for the read_many_files tool."""
|
||||
|
||||
file_paths: list[str] = Field(description="List of file paths to read (relative to workspace root)")
|
||||
|
||||
|
||||
class ReadFileTool(BaseTool):
|
||||
"""Read the contents of a file within the workspace."""
|
||||
|
||||
@@ -76,6 +82,58 @@ class ReadFileTool(BaseTool):
|
||||
)
|
||||
|
||||
|
||||
class ReadManyFilesTool(BaseTool):
|
||||
"""Read contents of multiple files at once."""
|
||||
|
||||
name = "read_many_files"
|
||||
description = (
|
||||
"Read contents of multiple files at once. Returns each file's content "
|
||||
"prefixed with its path header."
|
||||
)
|
||||
params_model = ReadManyFilesParams
|
||||
|
||||
def execute(self, *, tool_call_id: str, file_paths: list[str], **kwargs: Any) -> ToolResult:
|
||||
if not file_paths:
|
||||
return ToolResult(
|
||||
tool_call_id=tool_call_id,
|
||||
tool_name=self.name,
|
||||
status=ToolResultStatus.ERROR,
|
||||
error="file_paths list is empty",
|
||||
)
|
||||
|
||||
fs_config = self.config.tools.filesystem
|
||||
sections: list[str] = []
|
||||
success_count = 0
|
||||
|
||||
for fp in file_paths:
|
||||
try:
|
||||
content = safe_read_file(
|
||||
fp,
|
||||
self.workspace_root,
|
||||
max_size_bytes=fs_config.max_file_size_bytes,
|
||||
check_binary=fs_config.binary_detection,
|
||||
)
|
||||
sections.append(f"=== {fp} ===\n{content}")
|
||||
success_count += 1
|
||||
except (PathSecurityError, FileNotFoundError, FileSizeError, BinaryFileError) as exc:
|
||||
sections.append(f"=== {fp} ===\n[ERROR] {exc}")
|
||||
|
||||
if success_count == 0:
|
||||
return ToolResult(
|
||||
tool_call_id=tool_call_id,
|
||||
tool_name=self.name,
|
||||
status=ToolResultStatus.ERROR,
|
||||
error="All files failed to read:\n" + "\n".join(sections),
|
||||
)
|
||||
|
||||
return ToolResult(
|
||||
tool_call_id=tool_call_id,
|
||||
tool_name=self.name,
|
||||
status=ToolResultStatus.SUCCESS,
|
||||
output="\n".join(sections),
|
||||
)
|
||||
|
||||
|
||||
class ListDirParams(BaseModel):
|
||||
"""Parameters for the list_dir tool."""
|
||||
|
||||
|
||||
@@ -99,7 +99,7 @@ def create_default_registry(
|
||||
skill_runner: Optional SkillRunner for package skill activation.
|
||||
"""
|
||||
# Read tools
|
||||
from app.tools.filesystem import ListDirTool, ReadFileTool
|
||||
from app.tools.filesystem import ListDirTool, ReadFileTool, ReadManyFilesTool
|
||||
|
||||
# Write tools
|
||||
from app.tools.filesystem import DeleteFileTool, MakeDirTool, WriteFileTool
|
||||
@@ -120,6 +120,7 @@ def create_default_registry(
|
||||
|
||||
# Read
|
||||
registry.register(ReadFileTool(workspace_root, config))
|
||||
registry.register(ReadManyFilesTool(workspace_root, config))
|
||||
registry.register(ListDirTool(workspace_root, config))
|
||||
|
||||
# Search
|
||||
|
||||
69
tests/unit/test_filesystem_read_many.py
Normal file
69
tests/unit/test_filesystem_read_many.py
Normal file
@@ -0,0 +1,69 @@
|
||||
"""Tests for the read_many_files tool."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from app.models.config import AppConfig, load_config
|
||||
from app.models.tool_call import ToolResultStatus
|
||||
from app.tools.filesystem import ReadManyFilesTool
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def config() -> AppConfig:
|
||||
return load_config()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tmp_workspace(tmp_path: Path, config: AppConfig) -> tuple[Path, AppConfig]:
|
||||
"""Create a temporary workspace for read_many_files tests."""
|
||||
config.agent.workspace_root = tmp_path
|
||||
return tmp_path, config
|
||||
|
||||
|
||||
class TestReadManyFilesTool:
|
||||
def test_read_multiple_files(self, tmp_workspace: tuple[Path, AppConfig]) -> None:
|
||||
ws, cfg = tmp_workspace
|
||||
(ws / "a.txt").write_text("alpha")
|
||||
(ws / "b.txt").write_text("bravo")
|
||||
tool = ReadManyFilesTool(ws, cfg)
|
||||
result = tool.run("tc-1", {"file_paths": ["a.txt", "b.txt"]})
|
||||
assert result.status == ToolResultStatus.SUCCESS
|
||||
assert "=== a.txt ===" in result.output
|
||||
assert "alpha" in result.output
|
||||
assert "=== b.txt ===" in result.output
|
||||
assert "bravo" in result.output
|
||||
|
||||
def test_partial_failure(self, tmp_workspace: tuple[Path, AppConfig]) -> None:
|
||||
ws, cfg = tmp_workspace
|
||||
(ws / "exists.txt").write_text("hello")
|
||||
tool = ReadManyFilesTool(ws, cfg)
|
||||
result = tool.run("tc-2", {"file_paths": ["exists.txt", "missing.txt"]})
|
||||
assert result.status == ToolResultStatus.SUCCESS
|
||||
assert "hello" in result.output
|
||||
assert "[ERROR]" in result.output
|
||||
assert "=== missing.txt ===" in result.output
|
||||
|
||||
def test_all_files_fail(self, tmp_workspace: tuple[Path, AppConfig]) -> None:
|
||||
ws, cfg = tmp_workspace
|
||||
tool = ReadManyFilesTool(ws, cfg)
|
||||
result = tool.run("tc-3", {"file_paths": ["no1.txt", "no2.txt"]})
|
||||
assert result.status == ToolResultStatus.ERROR
|
||||
assert "All files failed" in (result.error or "")
|
||||
|
||||
def test_empty_file_paths(self, tmp_workspace: tuple[Path, AppConfig]) -> None:
|
||||
ws, cfg = tmp_workspace
|
||||
tool = ReadManyFilesTool(ws, cfg)
|
||||
result = tool.run("tc-4", {"file_paths": []})
|
||||
assert result.status == ToolResultStatus.ERROR
|
||||
assert "empty" in (result.error or "").lower()
|
||||
|
||||
def test_path_security_inline_error(self, tmp_workspace: tuple[Path, AppConfig]) -> None:
|
||||
ws, cfg = tmp_workspace
|
||||
(ws / "safe.txt").write_text("ok")
|
||||
tool = ReadManyFilesTool(ws, cfg)
|
||||
result = tool.run("tc-5", {"file_paths": ["safe.txt", "../../etc/passwd"]})
|
||||
assert result.status == ToolResultStatus.SUCCESS
|
||||
assert "ok" in result.output
|
||||
assert "[ERROR]" in result.output
|
||||
assert "outside" in result.output.lower()
|
||||
@@ -108,7 +108,7 @@ class TestToolRegistry:
|
||||
registry = create_default_registry(workspace, config)
|
||||
names = set(registry.get_all().keys())
|
||||
assert names == {
|
||||
"read_file", "list_dir", "grep_files", "find_files",
|
||||
"read_file", "read_many_files", "list_dir", "grep_files", "find_files",
|
||||
"write_file", "make_dir", "delete_file",
|
||||
"str_replace", "patch_apply",
|
||||
"run_command",
|
||||
@@ -118,7 +118,7 @@ class TestToolRegistry:
|
||||
def test_schema_export(self, workspace: Path, config: AppConfig) -> None:
|
||||
registry = create_default_registry(workspace, config)
|
||||
schemas = registry.get_openai_tools_schema()
|
||||
assert len(schemas) == 11
|
||||
assert len(schemas) == 12
|
||||
assert all(s["type"] == "function" for s in schemas)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user