From 703d933f220feac655029585d000fcd975471fb6 Mon Sep 17 00:00:00 2001 From: Roland Date: Thu, 10 Sep 2026 15:32:27 -0600 Subject: [PATCH] =?UTF-8?q?ai:=20list=5Ffiles=20tool=20=E2=80=94=20browse?= =?UTF-8?q?=20the=20media=20directory=20tree?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- modules/ai/api.py | 88 ++++++++++++++++++++++++++---- modules/ai/modes.py | 1 + modules/ai/schema.py | 9 +++ modules/ai/test/test_api.py | 77 ++++++++++++++++++++++++++ modules/ai/test/test_mcp_parity.py | 2 +- wrolpi_mcp/server.py | 34 ++++++++++++ 6 files changed, 200 insertions(+), 11 deletions(-) diff --git a/modules/ai/api.py b/modules/ai/api.py index 6092373f..0ae2351f 100644 --- a/modules/ai/api.py +++ b/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: diff --git a/modules/ai/modes.py b/modules/ai/modes.py index d2a32be9..ab0a8a6e 100644 --- a/modules/ai/modes.py +++ b/modules/ai/modes.py @@ -58,6 +58,7 @@ MODES = { 'list_collections', 'list_inventories', 'get_inventory', + 'list_files', 'read_file', ), ), diff --git a/modules/ai/schema.py b/modules/ai/schema.py index 6f6485d1..cec5c797 100644 --- a/modules/ai/schema.py +++ b/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 diff --git a/modules/ai/test/test_api.py b/modules/ai/test/test_api.py index b4742d1a..ccaab373 100644 --- a/modules/ai/test/test_api.py +++ b/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 diff --git a/modules/ai/test/test_mcp_parity.py b/modules/ai/test/test_mcp_parity.py index d6ea8950..65b5fb5d 100644 --- a/modules/ai/test/test_mcp_parity.py +++ b/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', } diff --git a/wrolpi_mcp/server.py b/wrolpi_mcp/server.py index a4399f07..9c496fe1 100644 --- a/wrolpi_mcp/server.py +++ b/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.).