diff --git a/modules/ai/api.py b/modules/ai/api.py index 0ae2351f..6962b7c6 100644 --- a/modules/ai/api.py +++ b/modules/ai/api.py @@ -4,6 +4,7 @@ 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 asyncio import mimetypes import pathlib from http import HTTPStatus @@ -13,18 +14,26 @@ from sanic_ext import validate from sanic_ext.extensions.openapi import openapi from modules.archive import lib as archive_lib +from modules.archive.lib import search_domains_by_name +from modules.docs.lib import search_authors_by_name, search_subjects_by_name +from modules.map import lib as map_lib +from modules.map import search as map_search +from modules.map.pins import get_map_pins_config +from modules.videos.channel.lib import search_channels_by_name +from modules.zim.models import Zims from modules.docs.lib import _doc_response, _get_doc, _search_docs from modules.inventory.common import get_inventory_configs from modules.inventory.errors import UnknownInventory from modules.videos.video import lib as videos_lib from modules.zim import lib as zim_lib -from wrolpi import flags +from wrolpi import flags, tags from wrolpi.api_utils import json_response from wrolpi.collections.lib import search_collections -from wrolpi.common import api_param_limiter, get_media_directory, logger, wrol_mode_enabled +from wrolpi.common import api_param_limiter, get_media_directory, get_relative_to_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, HIDDEN_DIRECTORIES, HIDDEN_FILES +from wrolpi.files.lib import search_files, search_file_suggestion_count, HIDDEN_DIRECTORIES, HIDDEN_FILES from wrolpi.vars import DOCKERIZED, IS_RPI4 from wrolpi.version import __version__ from wrolpi.schema import JSONErrorResponse @@ -577,6 +586,193 @@ async def ai_list_disks(_: Request): return json_response(ret) +# Errors on downloads can be long tracebacks; the model only needs the gist. +DOWNLOAD_ERROR_LENGTH = 300 +MAP_LINK_ZOOM = 12 + + +def _map_link(lat, lon) -> str: + return f'/map?lat={lat}&lon={lon}&z={MAP_LINK_ZOOM}' + + +@ai_bp.get('/tags') +@openapi.definition( + summary='List tags', + description='List every tag in the library with how many files, Zim entries, channels, and domains carry' + ' it, plus the most recently used tag names. Use a tag name in the tag_names filter of the' + ' search tools. Users tag things they care about, so tags are a good map of their interests.', +) +@openapi.operation('list_tags') +async def list_tags(_: Request): + results = [dict( + name=i['name'], + file_groups=i['file_group_count'], + zim_entries=i['zim_entry_count'], + channels=i['channel_count'], + domains=i['domain_count'], + ) for i in tags.get_tags()] + return json_response(dict(results=results, recent=tags.get_recent_tags(), total=len(results))) + + +def _format_download(download: dict) -> dict: + error = download.get('error') + if error and len(error) > DOWNLOAD_ERROR_LENGTH: + # Errors are usually tracebacks; the exception message at the end is the useful part. + error = '…' + error[-DOWNLOAD_ERROR_LENGTH:] + destination = download.get('destination') + return {k: v for k, v in dict( + id=download.get('id'), + url=download.get('url'), + status=download.get('status'), + downloader=download.get('downloader'), + sub_downloader=download.get('sub_downloader'), + frequency=download.get('frequency'), + destination=str(get_relative_to_media_directory(destination)) if destination else None, + collection_id=download.get('collection_id'), + tag_names=download.get('tag_names') or None, + next_download=download.get('next_download'), + last_successful_download=download.get('last_successful_download'), + error=error, + ).items() if v is not None} + + +@ai_bp.get('/downloads') +@openapi.definition( + summary='List downloads', + description='Read the download queue: a summary (pending count, whether downloading is disabled or' + ' stopped, daily limit), the recurring downloads (channels, feeds; frequency is in seconds),' + ' and the newest one-time downloads. Filter with status: new, pending, failed, deferred, or' + ' complete. Use this when the user asks what is downloading, why a download failed, or what' + ' is scheduled. You cannot start, stop, or retry downloads.', +) +@openapi.operation('list_downloads') +@openapi.parameter('status', str, 'query') +@openapi.parameter('limit', int, 'query') +async def list_downloads(request: Request): + status = request.args.get('status') + try: + limit = ai_limiter(int(request.args.get('limit', 0)) or None) + except ValueError: + raise ValidationError('limit must be an integer') + summary = download_manager.get_summary() + data = download_manager.get_fe_downloads() + + def select(downloads: list) -> list: + if status: + downloads = [i for i in downloads if i.get('status') == status] + return [_format_download(i) for i in downloads[:limit]] + + return json_response(dict( + summary=summary, + recurring=select(data['recurring_downloads']), + once=select(data['once_downloads']), + pending_once=data.get('pending_once_downloads', 0), + )) + + +@ai_bp.get('/map') +@openapi.definition( + summary='Get map overview', + description='What maps this WROLPi has: the downloaded map files (regions) and whether each has a place' + ' search index, the map regions subscribed for updates, and the user\'s saved pins with a' + ' link to each. Use search_places to look up a town or landmark.', +) +@openapi.operation('get_map_overview') +async def get_map_overview(request: Request): + files = [dict(name=i['name'], path=i['path'], size=i['size'], has_search_index=i['has_search_index']) + for i in map_lib.get_pmtiles_files()] + pins = [dict(id=i.get('id'), label=i.get('label'), lat=i.get('lat'), lon=i.get('lon'), + link=_map_link(i.get('lat'), i.get('lon'))) + for i in get_map_pins_config().pins] + return json_response(dict( + files=files, + subscriptions=map_lib.get_map_subscriptions(request.ctx.session), + pins=pins, + search_indexes=map_search.get_search_status(), + )) + + +@ai_bp.get('/map/search') +@openapi.definition( + summary='Search places on the map', + description='Find towns, cities, and landmarks by name in the downloaded maps\' place index. Each result' + ' has coordinates and a WROLPi map link. Pass lat and lon to rank nearest first. Results are' + ' empty when no map has a search index (see get_map_overview).', +) +@openapi.operation('search_places') +@openapi.parameter('q', str, 'query') +@openapi.parameter('limit', int, 'query') +@openapi.parameter('offset', int, 'query') +@openapi.parameter('lat', float, 'query') +@openapi.parameter('lon', float, 'query') +async def search_places(request: Request): + q = (request.args.get('q') or '').strip() + if not q: + raise ValidationError('q query parameter is required') + try: + limit = ai_limiter(int(request.args.get('limit', 0)) or None) + lat = float(request.args.get('lat')) if request.args.get('lat') is not None else None + lon = float(request.args.get('lon')) if request.args.get('lon') is not None else None + except ValueError: + raise ValidationError('limit, lat, and lon must be numbers') + data = await asyncio.to_thread(map_search.search_places, q, limit=limit, offset=_offset(request), + lat=lat, lon=lon) + results = [{k: v for k, v in dict( + name=i.get('name'), + kind=i.get('kind_detail') or i.get('kind'), + region=i.get('region'), + population=i.get('population'), + lat=i.get('lat'), + lon=i.get('lon'), + link=_map_link(i.get('lat'), i.get('lon')), + ).items() if v is not None} for i in data.get('results') or []] + return json_response(dict(results=results, total=data.get('total', len(results)))) + + +@ai_bp.post('/search/suggestions') +@openapi.definition( + summary='Preview what a search would find', + description='Before searching, learn which channels, domains (archived sites), authors, and subjects match' + ' a term, and how many files and Zim entries a search would return. Use the channel id with' + ' search_videos, the domain name with search_archives, the author/subject with search_docs.' + ' tag_names narrows the file estimate to tagged items.', + body=schema.AISearchSuggestionsRequest, +) +@openapi.operation('search_suggestions') +@validate(schema.AISearchSuggestionsRequest) +async def search_suggestions(request: Request, body: schema.AISearchSuggestionsRequest): + if not body.search_str and not body.tag_names: + raise ValidationError('search_str or tag_names is required') + session = request.ctx.session + search_str = body.search_str or '' + + channels = await search_channels_by_name(session, search_str, order_by_video_count=True) + domains = await search_domains_by_name(session, search_str) + authors = await search_authors_by_name(session, search_str) + subjects = await search_subjects_by_name(session, search_str) + + counts = await search_file_suggestion_count(body.search_str, body.tag_names, []) + zims = [] + if body.search_str: + def estimate(): + with_counts = Zims.estimate(session, body.search_str) + return [dict(id=zim.id, title=lib.zim_metadata_dict(zim.zim_metadata).get('title') or zim.path.name, + estimate=count) for zim, count in with_counts.items()] + + try: + zims = await asyncio.to_thread(estimate) + except Exception as e: + logger.debug('AI suggestions could not estimate Zims', exc_info=e) + + return json_response(dict( + channels=[dict(id=i.id, name=i.name, link=f'/videos/channel/{i.id}/video') for i in channels], + domains=[dict(id=i['id'], name=i['domain']) for i in domains], + authors=authors, + subjects=subjects, + estimates=dict(file_groups=counts['file_groups'], file_groups_deep=counts['file_groups_deep'], zims=zims), + )) + + # --------------------------------------------------------------------------- # Manage endpoints — for the AI page's Manage tab, NOT part of the model's tool catalog. # The agent loop excludes everything under /api/ai/manage/ when building tool definitions. diff --git a/modules/ai/modes.py b/modules/ai/modes.py index ab0a8a6e..b6ae7763 100644 --- a/modules/ai/modes.py +++ b/modules/ai/modes.py @@ -60,6 +60,10 @@ MODES = { 'get_inventory', 'list_files', 'read_file', + 'list_tags', + 'search_suggestions', + 'get_map_overview', + 'search_places', ), ), 'system': dict( diff --git a/modules/ai/schema.py b/modules/ai/schema.py index cec5c797..bc5a0f38 100644 --- a/modules/ai/schema.py +++ b/modules/ai/schema.py @@ -83,3 +83,9 @@ class AIListFilesResponse: files: list = dataclasses.field(default_factory=list) total: int = 0 next_offset: Optional[int] = None + + +@dataclasses.dataclass +class AISearchSuggestionsRequest: + search_str: Optional[str] = None + tag_names: List[str] = dataclasses.field(default_factory=list) diff --git a/modules/ai/test/test_api.py b/modules/ai/test/test_api.py index ccaab373..0c648f9f 100644 --- a/modules/ai/test/test_api.py +++ b/modules/ai/test/test_api.py @@ -383,3 +383,139 @@ async def test_ai_list_files_refuses_escapes(async_client, test_directory): (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 + + +@pytest.mark.asyncio +async def test_ai_list_tags(async_client, test_session, tag_factory, video_factory): + """Tags are listed with what they are applied to, plus the recently used names.""" + one = await tag_factory('one') + await tag_factory('two') + video = video_factory(title='tagged video') + video.file_group.add_tag(test_session, one.id) + test_session.commit() + + request, response = await async_client.get('/api/ai/tags') + assert response.status_code == HTTPStatus.OK + assert response.json['total'] == 2 + assert [i['name'] for i in response.json['results']] == ['one', 'two'] + assert response.json['results'][0]['file_groups'] == 1 + assert response.json['results'][1]['file_groups'] == 0 + assert response.json['recent'] == ['one'] + + +@pytest.mark.asyncio +async def test_ai_list_downloads(async_client, test_session, test_download_manager, test_downloader, + test_directory): + """The download queue is summarized with lean, relative-path entries; status can filter.""" + test_download_manager.create_download(test_session, 'https://example.com/once', test_downloader.name, + destination=test_directory / 'archive/example.com') + test_download_manager.recurring_download(test_session, 'https://example.com/feed', 86400, + test_downloader.name) + test_session.commit() + + request, response = await async_client.get('/api/ai/downloads') + assert response.status_code == HTTPStatus.OK + assert response.json['summary']['recurring'] == 1 + assert 'disabled' in response.json['summary'] + once, = response.json['once'] + assert once['url'] == 'https://example.com/once' + assert once['destination'] == 'archive/example.com' + assert once['status'] == 'new' + assert once['downloader'] == test_downloader.name + recurring, = response.json['recurring'] + assert recurring['url'] == 'https://example.com/feed' + assert recurring['frequency'] == 86400 + + # Filter by status. + request, response = await async_client.get('/api/ai/downloads?status=complete') + assert response.status_code == HTTPStatus.OK + assert response.json['once'] == [] and response.json['recurring'] == [] + + +@pytest.mark.asyncio +async def test_ai_map(async_client, test_session, test_directory, make_files_structure): + """The map overview reports map files, subscriptions, pins, and search indexes; places can be searched.""" + from modules.map.pins import get_map_pins_config + from modules.map.test.test_search import _create_test_search_db + + make_files_structure(['map/oregon.pmtiles']) + _create_test_search_db(test_directory / 'map/oregon.search.db', [ + ('Portland', 'city', 45.5, -122.6, 6, 'places', 'city', 650000, 'Oregon'), + ('Port Orford', 'town', 42.7, -124.5, 9, 'places', 'town', 1100, 'Oregon'), + ]) + get_map_pins_config().add_pin(45.1, -122.2, 'Home') + + request, response = await async_client.get('/api/ai/map') + assert response.status_code == HTTPStatus.OK + files, = response.json['files'] + assert files['name'] == 'oregon.pmtiles' and files['has_search_index'] is True + assert response.json['subscriptions'] == [] + pin, = response.json['pins'] + assert pin['label'] == 'Home' and pin['link'] == '/map?lat=45.1&lon=-122.2&z=12' + + request, response = await async_client.get('/api/ai/map/search?q=port') + assert response.status_code == HTTPStatus.OK + assert response.json['total'] == 2 + first = response.json['results'][0] + assert first['name'] == 'Portland' + assert first['region'] == 'Oregon' + assert first['link'] == '/map?lat=45.5&lon=-122.6&z=12' + + # Proximity ranking within an importance tier: equal min_zoom, nearest first. + _create_test_search_db(test_directory / 'map/coast.search.db', [ + ('Port A', 'town', 42.0, -124.0, 9, 'places', 'town', 100, 'Oregon'), + ('Port B', 'town', 44.0, -124.0, 9, 'places', 'town', 100, 'Oregon'), + ]) + request, response = await async_client.get('/api/ai/map/search?q=port%20b&lat=44.0&lon=-124.0') + assert response.status_code == HTTPStatus.OK + assert response.json['results'][0]['name'] == 'Port B' + request, response = await async_client.get('/api/ai/map/search?q=port&lat=42.0&lon=-124.0&limit=2') + assert response.status_code == HTTPStatus.OK + assert len(response.json['results']) == 2 and response.json['total'] == 4 + + # q is required. + request, response = await async_client.get('/api/ai/map/search') + assert response.status_code == HTTPStatus.BAD_REQUEST + + +@pytest.mark.asyncio +async def test_ai_search_suggestions(async_client, test_session, video_factory, archive_factory, channel_factory, + tag_factory): + """Suggestions name the channels/domains/authors/subjects matching a term and estimate result counts.""" + channel = channel_factory(name='Wild Cooking') + video_factory(channel_id=channel.id, title='cooking rice') + archive_factory(domain='cooking.example.com', title='cooking beans', contents='beans') + tag = await tag_factory('recipes') + test_session.commit() + + content = dict(search_str='cooking') + request, response = await async_client.post('/api/ai/search/suggestions', content=json.dumps(content)) + assert response.status_code == HTTPStatus.OK + ch, = response.json['channels'] + assert ch['name'] == 'Wild Cooking' and ch['id'] == channel.id and ch['link'].startswith('/videos/channel/') + dom, = response.json['domains'] + assert dom['name'] == 'cooking.example.com' + assert response.json['authors'] == [] and response.json['subjects'] == [] + assert response.json['estimates']['file_groups'] == 2 + assert 'file_groups_deep' in response.json['estimates'] + assert response.json['estimates']['zims'] == [] + + # Tag names narrow the estimate. + content = dict(search_str='cooking', tag_names=[tag.name]) + request, response = await async_client.post('/api/ai/search/suggestions', content=json.dumps(content)) + assert response.status_code == HTTPStatus.OK + assert response.json['estimates']['file_groups'] == 0 + + # Something to search for is required. + request, response = await async_client.post('/api/ai/search/suggestions', content=json.dumps({})) + assert response.status_code == HTTPStatus.BAD_REQUEST + + +def test_ai_format_download_error_tail(): + """Download errors are tracebacks; the model gets the end (the exception message), not the start.""" + from modules.ai.api import _format_download, DOWNLOAD_ERROR_LENGTH + error = 'Traceback (most recent call last):\n' + (' File "x.py"\n' * 100) + 'ValueError: the real reason' + formatted = _format_download(dict(id=1, url='https://example.com', status='failed', error=error)) + assert formatted['error'].endswith('ValueError: the real reason') + assert len(formatted['error']) == DOWNLOAD_ERROR_LENGTH + 1 + assert 'error' not in _format_download(dict(id=2, url='https://example.com', status='new', error=None)) diff --git a/modules/ai/test/test_mcp_parity.py b/modules/ai/test/test_mcp_parity.py index 65b5fb5d..ac770f60 100644 --- a/modules/ai/test/test_mcp_parity.py +++ b/modules/ai/test/test_mcp_parity.py @@ -18,6 +18,7 @@ EXPECTED_TOOLS = { 'get_doc', 'get_zim_entry', 'list_collections', 'list_zim_files', 'list_files', 'get_statistics', 'get_inventory', 'get_inventory_items', 'get_status', + 'list_tags', 'search_suggestions', 'list_downloads', 'get_map_overview', 'search_places', } diff --git a/wrolpi_mcp/server.py b/wrolpi_mcp/server.py index 9c496fe1..f1abf158 100644 --- a/wrolpi_mcp/server.py +++ b/wrolpi_mcp/server.py @@ -498,6 +498,164 @@ async def list_zim_files() -> str: return "\n\n".join(lines) +@mcp.tool(annotations={"readOnlyHint": True}) +async def list_tags() -> str: + """List every tag in the library with how many files, Zim entries, channels, and domains carry it. + + Tag names can be passed as tag_names to the search tools. Users tag what they care about, so this + is a good map of their interests. Also returns the most recently used tag names. + """ + data = await api_get("/api/ai/tags") + results = data.get("results") or [] + if not results: + return "No tags found." + lines = [] + for tag in results: + counts = ", ".join(f"{tag.get(k, 0)} {k.replace('_', ' ')}" for k in + ("file_groups", "zim_entries", "channels", "domains") if tag.get(k)) + lines.append(f" {tag['name']}" + (f" ({counts})" if counts else " (unused)")) + text = f"Tags ({data.get('total', len(results))}):\n" + "\n".join(lines) + if recent := data.get("recent"): + text += f"\n\nRecently used: {', '.join(recent)}" + return text + + +@mcp.tool(annotations={"readOnlyHint": True}) +async def search_suggestions(query: str, tag_names: list[str] | None = None) -> str: + """Preview what a search would find before running it. + + Returns the channels, domains (archived sites), authors, and subjects whose names match the query, + and estimates of how many files and Zim entries the search would return. Use the channel ID with + search_videos, the domain with search_archives, the author/subject with search_docs. + + Args: + query: The term the user wants to search for. + tag_names: Narrow the file estimate to items with these tags. + """ + body = {"search_str": query} + if tag_names: + body["tag_names"] = tag_names + data = await api_post("/api/ai/search/suggestions", json=body) + lines = [] + for channel in data.get("channels") or []: + lines.append(f" Channel: {channel['name']} (ID: {channel['id']}) LINK: {_link(channel.get('link'))}") + for domain in data.get("domains") or []: + lines.append(f" Domain: {domain['name']}") + for author in data.get("authors") or []: + lines.append(f" Author: {author['name']}") + for subject in data.get("subjects") or []: + lines.append(f" Subject: {subject['name']}") + if not lines: + lines.append(" No matching channels, domains, authors, or subjects.") + estimates = data.get("estimates") or {} + lines.append(f"\nEstimated matching files: {estimates.get('file_groups', 0)}" + f" (deep content search: {estimates.get('file_groups_deep', 0)})") + for zim in estimates.get("zims") or []: + lines.append(f" Zim '{zim.get('title')}' (ID {zim.get('id')}): ~{zim.get('estimate', 0)} entries") + return "\n".join(lines) + + +@mcp.tool(annotations={"readOnlyHint": True}) +async def list_downloads(status: str | None = None, limit: int = DEFAULT_LIMIT) -> str: + """Read the WROLPi download queue: summary, recurring downloads (channels, feeds), and one-time downloads. + + Use this when the user asks what is downloading, what failed and why, or what is scheduled. + Downloads cannot be started, stopped, or retried from here. + + Args: + status: Only show downloads with this status: new, pending, failed, deferred, or complete. + limit: Maximum recurring and one-time downloads to show (each). + """ + params = {"limit": limit} + if status: + params["status"] = status + data = await api_get("/api/ai/downloads", params=params) + summary = data.get("summary") or {} + lines = ["Summary: " + ", ".join(f"{k}={v}" for k, v in summary.items())] + + def render(download: dict) -> str: + parts = [f" [{download.get('status')}] {download.get('url')} (ID: {download.get('id')}," + f" {download.get('downloader')})"] + if download.get("frequency"): + parts.append(f" every {download['frequency']}s, next: {download.get('next_download') or 'unscheduled'}") + if download.get("destination"): + parts.append(f" destination: {download['destination']}") + if download.get("tag_names"): + parts.append(f" tags: {', '.join(download['tag_names'])}") + if download.get("error"): + parts.append(f" error: {download['error']}") + return "\n".join(parts) + + recurring = data.get("recurring") or [] + once = data.get("once") or [] + lines.append(f"\nRecurring downloads ({len(recurring)} shown):") + lines.extend(render(i) for i in recurring) if recurring else lines.append(" none") + lines.append(f"\nOne-time downloads ({len(once)} shown, {data.get('pending_once', 0)} pending):") + lines.extend(render(i) for i in once) if once else lines.append(" none") + return "\n".join(lines) + + +@mcp.tool(annotations={"readOnlyHint": True}) +async def get_map_overview() -> str: + """Describe the maps on this WROLPi: downloaded map regions, whether each has a place-search index, + subscribed regions, and the user's saved pins (with links).""" + data = await api_get("/api/ai/map") + lines = ["Map files:"] + files = data.get("files") or [] + for file in files: + size_gb = (file.get("size") or 0) / (1024 ** 3) + index = "searchable" if file.get("has_search_index") else "no search index" + lines.append(f" {file['name']} ({size_gb:,.2f} GB, {index})") + if not files: + lines.append(" none") + subscriptions = data.get("subscriptions") or [] + names = [i.get("name") or i.get("region") if isinstance(i, dict) else str(i) for i in subscriptions] + lines.append(f"\nSubscribed regions: {', '.join(names) if names else 'none'}") + lines.append("\nPins:") + pins = data.get("pins") or [] + for pin in pins: + lines.append(f" {pin.get('label') or 'Unlabeled'} ({pin.get('lat')}, {pin.get('lon')})" + f" LINK: {_link(pin.get('link'))}") + if not pins: + lines.append(" none") + return "\n".join(lines) + + +@mcp.tool(annotations={"readOnlyHint": True}) +async def search_places( + query: str, + limit: int = DEFAULT_LIMIT, + offset: int = 0, + lat: float | None = None, + lon: float | None = None, +) -> str: + """Find towns, cities, and landmarks by name in the downloaded maps. Each result links to the WROLPi map. + + Args: + query: Place name (prefix match), e.g. "Portland". + limit: Maximum results. + offset: Pagination offset. + lat: Latitude to rank nearby places first. + lon: Longitude to rank nearby places first. + """ + params = {"q": query, "limit": limit, "offset": offset} + if lat is not None and lon is not None: + params.update(lat=lat, lon=lon) + data = await api_get("/api/ai/map/search", params=params) + results = data.get("results") or [] + if not results: + return "No places found. The maps may have no search index (see get_map_overview)." + lines = [] + for i, place in enumerate(results, 1): + detail = ", ".join(str(place[k]) for k in ("kind", "region") if place.get(k)) + if place.get("population"): + detail += f", pop. {place['population']:,}" + lines.append(f"{i}. {place['name']} ({detail}) at {place.get('lat')}, {place.get('lon')}" + f" LINK: {_link(place.get('link'))}") + lines.append(f"\nTotal matching: {data.get('total', len(results))}") + return "\n".join(lines) + + @mcp.tool(annotations={"readOnlyHint": True}) async def get_statistics() -> str: """Get an overview of what content is stored in the WROLPi library.