ai: list_files tool — browse the media directory tree

New read-only GET /api/ai/files/list lists one directory of the media
directory (directories first, then files with size and guessed
mimetype), paged at 100 entries.  Every entry carries its relative
path for list_files or read_file.  Same guards as read_file via a
shared resolver: no escaping the media directory, and the config
directory is refused and hidden from the root listing.

The agent loop picks the operation up from the spec; library mode
allowlists it.  wrolpi_mcp gains the matching list_files proxy tool.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

modules/ai/api.py

@@ -4,6 +4,8 @@ This blueprint IS the AI tool catalog: the agent loop generates its tool definit
 endpoints' OpenAPI operations, and the external MCP server proxies them.  Every endpoint is
 read-only; descriptions are written for the model, not for humans.
 """
+import mimetypes
+import pathlib
 from http import HTTPStatus
 
 from sanic import Blueprint, Request
@@ -22,7 +24,7 @@ from wrolpi.collections.lib import search_collections
 from wrolpi.common import api_param_limiter, get_media_directory, logger, wrol_mode_enabled
 from wrolpi.downloader import download_manager
 from wrolpi.errors import InvalidFile, SearchEmpty, UnknownFile, ValidationError
-from wrolpi.files.lib import search_files
+from wrolpi.files.lib import search_files, HIDDEN_DIRECTORIES, HIDDEN_FILES
 from wrolpi.vars import DOCKERIZED, IS_RPI4
 from wrolpi.version import __version__
 from wrolpi.schema import JSONErrorResponse
@@ -39,6 +41,8 @@ ai_limiter = api_param_limiter(25, default=5)
 
 # Refuse to serve files this large through the text reader; real content is paged anyway.
 MAX_TEXT_FILE_SIZE = 10 * 1024 * 1024
+# Directory listings are paged by entry count so one big channel can't fill the model's context.
+LIST_PAGE_SIZE = 100
 
 
 def _offset(request: Request) -> int:
@@ -639,12 +643,82 @@ async def manage_settings(_: Request, body: schema.AIManageSettingsRequest):
     ))
 
 
+def _resolve_media_path(relative: str | None, verb: str) -> pathlib.Path:
+    """Resolve a relative path inside the media directory, refusing escapes and the config directory."""
+    media_directory = get_media_directory().resolve()
+    path = (media_directory / (relative or '').strip('/')).resolve()
+    if path != media_directory and not str(path).startswith(f'{media_directory}/'):
+        raise InvalidFile(f'Cannot {verb} outside the media directory')
+    # The config directory can contain secrets (Wi-Fi credentials, API keys); never serve it.
+    config_directory = media_directory / 'config'
+    if path == config_directory or str(path).startswith(f'{config_directory}/'):
+        raise InvalidFile(f'Cannot {verb} config files')
+    return path
+
+
+@ai_bp.get('/files/list')
+@openapi.definition(
+    summary='List a directory',
+    description='List the directories and files directly inside one directory of the media directory, by its'
+                ' relative path (omit path for the top level). Directories come first, then files, each with its'
+                ' relative path to pass to list_files or read_file. Big directories are paged: request again'
+                ' with offset=next_offset to continue. Use search when looking for content by topic; use this'
+                ' when the user asks what is in a folder or how files are organized.',
+)
+@openapi.operation('list_files')
+@openapi.parameter('path', str, 'query')
+@openapi.parameter('offset', int, 'query')
+@openapi.response(HTTPStatus.OK, schema.AIListFilesResponse)
+@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
+async def list_files(request: Request):
+    relative = request.args.get('path') or ''
+    media_directory = get_media_directory().resolve()
+    directory = _resolve_media_path(relative, 'list')
+    if not directory.is_dir():
+        raise UnknownFile(f'No such directory: {relative}')
+
+    def relative_path(path: pathlib.Path) -> str:
+        rel = str(path.relative_to(media_directory))
+        return f'{rel}/' if path.is_dir() else rel
+
+    directories, files = [], []
+    for child in directory.iterdir():
+        if child.is_dir():
+            # The config directory is never served (see _resolve_media_path), so don't advertise it.
+            if child.name not in HIDDEN_DIRECTORIES and child != media_directory / 'config':
+                directories.append(child)
+        elif child.name not in HIDDEN_FILES:
+            files.append(child)
+    directories.sort(key=lambda i: i.name.lower())
+    files.sort(key=lambda i: i.name.lower())
+
+    entries = [dict(name=i.name, path=relative_path(i)) for i in directories]
+    for file in files:
+        try:
+            size = file.stat().st_size
+        except OSError:
+            size = None
+        entries.append(dict(name=file.name, path=relative_path(file), size=size,
+                            mimetype=mimetypes.guess_type(file.name)[0]))
+
+    offset = _offset(request)
+    page = entries[offset:offset + LIST_PAGE_SIZE]
+    next_offset = offset + LIST_PAGE_SIZE if offset + LIST_PAGE_SIZE < len(entries) else None
+    return json_response(dict(
+        path=relative_path(directory) if directory != media_directory else '',
+        directories=[i for i in page if 'size' not in i],
+        files=[i for i in page if 'size' in i],
+        total=len(entries),
+        next_offset=next_offset,
+    ))
+
+
 @ai_bp.get('/files/read')
 @openapi.definition(
     summary='Read a text file',
     description='Read a plain-text file from the media directory by its relative path (from search result'
-                ' links or collection directories). Only text files can be read. Long files are paged:'
-                ' request again with offset=next_offset to continue reading.',
+                ' links, collection directories, or list_files). Only text files can be read. Long files are'
+                ' paged: request again with offset=next_offset to continue reading.',
 )
 @openapi.operation('read_file')
 @openapi.parameter('path', str, 'query')
@@ -656,13 +730,7 @@ async def read_file(request: Request):
     if not relative:
         raise InvalidFile('path query parameter is required')
 
-    media_directory = get_media_directory().resolve()
-    path = (media_directory / relative.lstrip('/')).resolve()
-    if not str(path).startswith(f'{media_directory}/'):
-        raise InvalidFile('Cannot read outside the media directory')
-    # The config directory can contain secrets (Wi-Fi credentials, API keys); never serve it.
-    if str(path).startswith(f'{media_directory / "config"}/'):
-        raise InvalidFile('Cannot read config files')
+    path = _resolve_media_path(relative, 'read')
     if not path.is_file():
         raise UnknownFile(f'No such file: {relative}')
     if path.stat().st_size > MAX_TEXT_FILE_SIZE:

modules/ai/modes.py

@@ -58,6 +58,7 @@ MODES = {
             'list_collections',
             'list_inventories',
             'get_inventory',
+            'list_files',
             'read_file',
         ),
     ),

modules/ai/schema.py

@@ -74,3 +74,12 @@ class AIPagedTextResponse:
     content: str = ''
     next_offset: Optional[int] = None
     total_chars: int = 0
+
+
+@dataclasses.dataclass
+class AIListFilesResponse:
+    path: str = ''
+    directories: list = dataclasses.field(default_factory=list)
+    files: list = dataclasses.field(default_factory=list)
+    total: int = 0
+    next_offset: Optional[int] = None

modules/ai/test/test_api.py

@@ -306,3 +306,80 @@ async def test_ai_read_file_refuses_escapes(async_client, test_directory):
     (test_directory / 'config/wrolpi.yaml').write_text('secret: hunter2')
     request, response = await async_client.get('/api/ai/files/read?path=config/wrolpi.yaml')
     assert response.status_code == HTTPStatus.BAD_REQUEST
+
+
+@pytest.mark.asyncio
+async def test_ai_list_files(async_client, test_directory):
+    """Directories in the media directory can be listed: directories first, then files, both sorted."""
+    (test_directory / 'videos/channel').mkdir(parents=True)
+    (test_directory / 'videos/channel/b.mp4').write_bytes(b'x' * 10)
+    (test_directory / 'videos/channel/a.mp4').write_bytes(b'x' * 20)
+    (test_directory / 'videos/channel/a.en.vtt').write_text('WEBVTT')
+    (test_directory / 'videos/channel/.DS_Store').write_bytes(b'\x00')  # macOS junk is hidden
+    (test_directory / 'videos/other').mkdir()
+    (test_directory / 'notes.txt').write_text('hi')
+    (test_directory / 'lost+found').mkdir()  # hidden
+    (test_directory / 'config').mkdir(exist_ok=True)  # hidden: may hold secrets
+
+    # The root is listed when no path is given.
+    request, response = await async_client.get('/api/ai/files/list')
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['path'] == ''
+    assert [i['path'] for i in response.json['directories']] == ['videos/']
+    assert [i['path'] for i in response.json['files']] == ['notes.txt']
+    assert response.json['total'] == 2
+    assert response.json['next_offset'] is None
+
+    request, response = await async_client.get('/api/ai/files/list?path=videos/channel')
+    assert response.status_code == HTTPStatus.OK
+    assert response.json['path'] == 'videos/channel/'
+    assert response.json['directories'] == []
+    assert [(i['path'], i['size']) for i in response.json['files']] == [
+        ('videos/channel/a.en.vtt', 6),
+        ('videos/channel/a.mp4', 20),
+        ('videos/channel/b.mp4', 10),
+    ]
+    assert response.json['files'][1]['mimetype'] == 'video/mp4'
+
+    # Trailing slashes and leading slashes are tolerated.
+    request, response = await async_client.get('/api/ai/files/list?path=/videos/')
+    assert response.status_code == HTTPStatus.OK
+    assert [i['path'] for i in response.json['directories']] == ['videos/channel/', 'videos/other/']
+    assert response.json['files'] == []
+
+    # Missing directory, and a file rather than a directory.
+    request, response = await async_client.get('/api/ai/files/list?path=nope')
+    assert response.status_code == HTTPStatus.NOT_FOUND
+    request, response = await async_client.get('/api/ai/files/list?path=notes.txt')
+    assert response.status_code == HTTPStatus.NOT_FOUND
+
+
+@pytest.mark.asyncio
+async def test_ai_list_files_paged(async_client, test_directory):
+    """Big directories are paged so a single listing cannot blow the model's context."""
+    from modules.ai.api import LIST_PAGE_SIZE
+    (test_directory / 'big').mkdir()
+    for i in range(LIST_PAGE_SIZE + 5):
+        (test_directory / f'big/{i:04d}.txt').write_text('x')
+
+    request, response = await async_client.get('/api/ai/files/list?path=big')
+    assert response.status_code == HTTPStatus.OK
+    assert len(response.json['files']) == LIST_PAGE_SIZE
+    assert response.json['total'] == LIST_PAGE_SIZE + 5
+    assert response.json['next_offset'] == LIST_PAGE_SIZE
+
+    request, response = await async_client.get(f'/api/ai/files/list?path=big&offset={LIST_PAGE_SIZE}')
+    assert response.status_code == HTTPStatus.OK
+    assert [i['name'] for i in response.json['files']] == [f'{i:04d}.txt' for i in range(LIST_PAGE_SIZE, LIST_PAGE_SIZE + 5)]
+    assert response.json['next_offset'] is None
+
+
+@pytest.mark.asyncio
+async def test_ai_list_files_refuses_escapes(async_client, test_directory):
+    """Listings outside the media directory, and inside the config directory, are refused."""
+    request, response = await async_client.get('/api/ai/files/list?path=../')
+    assert response.status_code == HTTPStatus.BAD_REQUEST
+
+    (test_directory / 'config').mkdir(exist_ok=True)
+    request, response = await async_client.get('/api/ai/files/list?path=config')
+    assert response.status_code == HTTPStatus.BAD_REQUEST

modules/ai/test/test_mcp_parity.py

@@ -16,7 +16,7 @@ EXPECTED_TOOLS = {
     'get_video', 'get_video_captions', 'get_video_comments',
     'get_archive', 'get_archive_text',
     'get_doc', 'get_zim_entry',
-    'list_collections', 'list_zim_files',
+    'list_collections', 'list_zim_files', 'list_files',
     'get_statistics', 'get_inventory', 'get_inventory_items', 'get_status',
 }
 

wrolpi_mcp/server.py

@@ -441,6 +441,40 @@ async def list_collections(kind: str | None = None) -> str:
     return "\n\n".join(lines)
 
 
+@mcp.tool(annotations={"readOnlyHint": True})
+async def list_files(path: str = "", offset: int = 0) -> str:
+    """List the directories and files inside one directory of the WROLPi media directory.
+
+    Use this to explore how the library is organized on disk (e.g. "videos/", "archive/", a channel's
+    directory).  Directories are listed first, then files.  Every entry's relative path can be passed
+    back to list_files to descend.  Large directories are paged; call again with the returned offset.
+
+    Args:
+        path: Directory relative to the media directory (e.g. "videos/SomeChannel").  Empty for the top level.
+        offset: Number of entries to skip (from a previous listing's "next offset").
+    """
+    try:
+        data = await api_get("/api/ai/files/list", params={"path": path, "offset": offset})
+    except httpx.HTTPStatusError as e:
+        if e.response.status_code == 404:
+            return f"No such directory: {path or '/'}"
+        raise
+    lines = [f"Directory: {data.get('path') or '/'}"]
+    for directory in data.get("directories") or []:
+        lines.append(f"  [dir]  {directory['path']}")
+    for file in data.get("files") or []:
+        size = file.get("size")
+        size_text = f"{size / (1024 * 1024):,.1f} MB" if size and size >= 1024 * 1024 else f"{size or 0:,} B"
+        mimetype = file.get("mimetype") or "unknown"
+        lines.append(f"  [file] {file['path']}  ({mimetype}, {size_text})")
+    if not (data.get("directories") or data.get("files")):
+        lines.append("  (empty)")
+    lines.append(f"\nTotal entries: {data.get('total', 0)}")
+    if (next_offset := data.get("next_offset")) is not None:
+        lines.append(f"More entries available: call list_files again with offset={next_offset}")
+    return "\n".join(lines)
+
+
 @mcp.tool(annotations={"readOnlyHint": True})
 async def list_zim_files() -> str:
     """List all available Zim encyclopedias (Wikipedia, Wiktionary, etc.).