Skip to content
Merged
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
44 changes: 34 additions & 10 deletions builtin_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -48,11 +48,22 @@ def _safe_resolve(relative: str | None) -> Path:
async def list_files(
context: dict[str, Any],
path: str | None = None,
recursive: bool = True,
max_depth: int | None = None,
) -> dict[str, Any]:
"""List files and subdirectories at *path* inside the files base directory.

Returns a JSON object with an ``entries`` list; each entry has ``name``,
``type`` (``"file"`` or ``"directory"``), and ``size`` (bytes, files only).
Returns a JSON object with an ``entries`` list; each entry has ``name``
(basename), ``path`` (relative to the listed directory, using ``/`` as the
separator), ``type`` (``"file"`` or ``"directory"``), and ``size`` (bytes,
files only).

When *recursive* is true (default), descends into subdirectories. Each
directory is still emitted as its own entry (with ``type="directory"``)
before its children. *max_depth* limits the recursion depth (``1`` =
immediate children only, same as ``recursive=False``; ``None`` = unlimited).
Symlinks to directories are not followed, to avoid cycles.

If the directory does not exist yet the entries list is empty (not an error).
"""
try:
Expand All @@ -63,23 +74,36 @@ async def list_files(
"ok": True,
"base_dir": str(base),
"path": path or "",
"recursive": recursive,
"entries": [],
}
if not target.is_dir():
return {"ok": False, "error": f"'{path}' is not a directory"}

entries: list[dict[str, Any]] = []
for entry in sorted(target.iterdir()):
entries.append(
{
"name": entry.name,
"type": "directory" if entry.is_dir() else "file",
"size": entry.stat().st_size if entry.is_file() else None,
}
)

def _walk(directory: Path, depth: int) -> None:
for entry in sorted(directory.iterdir()):
is_dir = entry.is_dir() and not entry.is_symlink()
rel = entry.relative_to(target).as_posix()
entries.append(
{
"name": entry.name,
"path": rel,
"type": "directory" if is_dir else "file",
"size": entry.stat().st_size if entry.is_file() else None,
}
)
if recursive and is_dir and (max_depth is None or depth + 1 < max_depth):
_walk(entry, depth + 1)

_walk(target, 0)

return {
"ok": True,
"base_dir": str(base),
"path": path or "",
"recursive": recursive,
"entries": entries,
}
except Exception as exc:
Expand Down
20 changes: 19 additions & 1 deletion server.py
Original file line number Diff line number Diff line change
Expand Up @@ -540,7 +540,25 @@ def register_builtin_tools() -> None:
"Omit or pass an empty string to list the root."
),
"default": "",
}
},
"recursive": {
"type": "boolean",
"description": (
"If true (default), also list files inside subdirectories. "
"Directories themselves are still listed as entries with "
"type='directory'. Symlinks to directories are not followed. "
"Set to false for a shallow (one-level) listing."
),
"default": True,
},
"max_depth": {
"type": "integer",
"description": (
"Maximum recursion depth when recursive=true "
"(1 = immediate children only). Omit for unlimited."
),
"minimum": 1,
},
},
"required": [],
},
Expand Down
73 changes: 73 additions & 0 deletions tests/test_builtin_tools.py
Original file line number Diff line number Diff line change
Expand Up @@ -145,6 +145,79 @@ async def test_list_nonexistent_subdir_returns_empty(self, tmp_path: Path, monke
assert result["ok"] is True
assert result["entries"] == []

@pytest.mark.asyncio
async def test_recursive_lists_nested_entries(self, tmp_path: Path, monkeypatch):
base = tmp_path / "files"
base.mkdir()
(base / "top.txt").write_text("x")
(base / "sub").mkdir()
(base / "sub" / "a.txt").write_text("a")
(base / "sub" / "deep").mkdir()
(base / "sub" / "deep" / "b.txt").write_text("b")
_set_base(monkeypatch, base)
from builtin_tools import list_files
result = await list_files(_ctx(), recursive=True)
assert result["ok"] is True
paths = {e["path"] for e in result["entries"]}
assert paths == {"top.txt", "sub", "sub/a.txt", "sub/deep", "sub/deep/b.txt"}

@pytest.mark.asyncio
async def test_recursive_default_is_recursive(self, tmp_path: Path, monkeypatch):
base = tmp_path / "files"
base.mkdir()
(base / "sub").mkdir()
(base / "sub" / "a.txt").write_text("a")
_set_base(monkeypatch, base)
from builtin_tools import list_files
result = await list_files(_ctx())
paths = {e["path"] for e in result["entries"]}
assert paths == {"sub", "sub/a.txt"}

@pytest.mark.asyncio
async def test_recursive_false_is_shallow(self, tmp_path: Path, monkeypatch):
base = tmp_path / "files"
base.mkdir()
(base / "sub").mkdir()
(base / "sub" / "a.txt").write_text("a")
_set_base(monkeypatch, base)
from builtin_tools import list_files
result = await list_files(_ctx(), recursive=False)
paths = {e["path"] for e in result["entries"]}
assert paths == {"sub"}

@pytest.mark.asyncio
async def test_recursive_max_depth(self, tmp_path: Path, monkeypatch):
base = tmp_path / "files"
base.mkdir()
(base / "sub").mkdir()
(base / "sub" / "a.txt").write_text("a")
(base / "sub" / "deep").mkdir()
(base / "sub" / "deep" / "b.txt").write_text("b")
_set_base(monkeypatch, base)
from builtin_tools import list_files
result = await list_files(_ctx(), recursive=True, max_depth=2)
paths = {e["path"] for e in result["entries"]}
assert paths == {"sub", "sub/a.txt", "sub/deep"}

@pytest.mark.asyncio
async def test_recursive_does_not_follow_dir_symlinks(self, tmp_path: Path, monkeypatch):
base = tmp_path / "files"
base.mkdir()
(base / "real").mkdir()
(base / "real" / "x.txt").write_text("x")
try:
(base / "link").symlink_to(base / "real", target_is_directory=True)
except (OSError, NotImplementedError):
pytest.skip("symlinks not supported on this platform")
_set_base(monkeypatch, base)
from builtin_tools import list_files
result = await list_files(_ctx(), recursive=True)
paths = {e["path"] for e in result["entries"]}
assert "real/x.txt" in paths
assert "link/x.txt" not in paths
link_entry = next(e for e in result["entries"] if e["path"] == "link")
assert link_entry["type"] == "file"


# ---------------------------------------------------------------------------
# get_file
Expand Down
Loading