rt/wrolpi
1
0
Fork 0
mirror of https://github.com/lrnselfreliance/wrolpi.git synced 2026-10-26 19:01:39 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

ai: research mode runs on the compact tool catalog

Browse source 
Research mode switches to nine kind-generic tools: search_files,
get_file, read_content, search_zims, get_zim_entry, get_inventory,
list_files, read_file, search_places.  Tool definitions drop from 20
tools / 8,947 chars (~2,240 tokens) to 9 / 3,881 (~970) on every
request, a sixth of the 8k context the small tier runs with instead of
a quarter.  The remaining descriptions are trimmed toward 300 chars.

The replaced per-kind endpoints are removed (four searches, three
getters, three readers, search_suggestions, the slug inventory route);
their tests move onto the generic routes.  The research prompt, its
worked example, and the empty-search hint no longer tell the model to
list collections first: channels resolve by name from the library map
that is already in the prompt.  list_collections, list_tags, and
get_map_overview stay as endpoints for the MCP wrappers but leave the
chat allowlist; the MCP search_suggestions tool goes since search_files
returns the same matches inline.

scripts/ai_eval.py expects search_files and treats "what channels do I
have" as answerable from context.  On the 1.7B model against the dev
library both catalogs score 14/18, but the compact one answers with one
tool call where the old one needed up to five, no longer invents the
example's channel, and keeps kinds straight when asked for videos.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Roland 2026-09-10 16:28:16 -06:00
parent 6dcd5d323b
commit e65e28bfc2
No known key found for this signature in database
GPG key ID: 1BA620CE6C9A4680
13 changed files with 102 additions and 681 deletions
Download patch file Download diff file Expand all files Collapse all files

6
app/src/components/AiChat.test.js
View file

@ -66,8 +66,8 @@ describe('AiChat', () => {
streamChat.mockImplementation(async ({mode, messages}, onEvent) => {
expect(mode).toBe('research');
expect(messages[messages.length - 1]).toEqual({role: 'user', content: 'find canning'});
onEvent('tool_call', {tool: 'search_videos', args: {search_str: 'canning'}});
onEvent('tool_result', {tool: 'search_videos', success: true});
onEvent('tool_call', {tool: 'search_files', args: {search_str: 'canning', kind: 'video'}});
onEvent('tool_result', {tool: 'search_files', success: true});
onEvent('token', {content: 'One video: '});
onEvent('token', {content: '/videos/123'});
onEvent('done', {content: 'One video: /videos/123'});
@ -80,7 +80,7 @@ describe('AiChat', () => {
fireEvent.change(screen.getByLabelText('Chat message'), {target: {value: 'find canning'}});
fireEvent.click(screen.getByText('Send'));
await waitFor(() => expect(screen.getByText(/using search videos/)).toBeInTheDocument());
await waitFor(() => expect(screen.getByText(/using search files/)).toBeInTheDocument());
await waitFor(() => expect(screen.getByRole('link', {name: '/videos/123'})).toBeInTheDocument());
expect(streamChat).toHaveBeenCalledTimes(1);
});

6
app/src/components/ThemeSamplePage.js
View file

@ -728,15 +728,15 @@ export function ThemeSamplePage() {
<ChatMessage message={{
role: 'assistant',
content: 'I found one video: Pressure Canning Basics — /videos/123',
activity: [{tool: 'search_videos', pending: false, success: true}],
activity: [{tool: 'search_files', pending: false, success: true}],
}}/>
<ChatMessage message={{
role: 'assistant', content: '',
activity: [{tool: 'search_archives', pending: true}], streaming: true,
activity: [{tool: 'search_files', pending: true}], streaming: true,
}}/>
<ChatMessage message={{
role: 'assistant', content: '',
activity: [{tool: 'get_video_captions', pending: false, success: false}],
activity: [{tool: 'read_content', pending: false, success: false}],
error: 'The AI failed to answer. Check the logs.',
}}/>
</Section>

309
modules/ai/api.py
View file

@ -22,7 +22,6 @@ 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.videos.models import Video
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
@ -35,7 +34,7 @@ from wrolpi.common import api_param_limiter, get_media_directory, get_relative_t
wrol_mode_enabled
from wrolpi.downloader import download_manager
from wrolpi.errors import InvalidFile, SearchEmpty, UnknownFile, ValidationError
from wrolpi.files.lib import search_files, search_file_suggestion_count, HIDDEN_DIRECTORIES, HIDDEN_FILES
from wrolpi.files.lib import search_files, HIDDEN_DIRECTORIES, HIDDEN_FILES
from wrolpi.files.models import FileGroup
from wrolpi.vars import DOCKERIZED, IS_RPI4
from wrolpi.version import __version__
@ -64,205 +63,6 @@ def _offset(request: Request) -> int:
return 0
@ai_bp.post('/search')
@openapi.definition(
summary='Search all WROLPi content',
description='Search everything in the library at once: videos, archived web pages, ebooks, PDFs, and other'
' files. search_str is optional: omit it to browse the newest items. Prefer the type-specific search'
' (videos/archives/docs) when the user asks about one kind of content. When total is large,'
' narrow the query instead of paging.',
body=schema.AISearchRequest,
)
@openapi.operation('search_all')
@openapi.response(HTTPStatus.OK, schema.AISearchResponse)
@validate(schema.AISearchRequest)
async def search_all(_: Request, body: schema.AISearchRequest):
file_groups, total = search_files(
body.search_str,
ai_limiter(body.limit),
body.offset or 0,
mimetypes=body.mimetypes,
tag_names=body.tag_names,
headline=True,
)
return json_response(lib.format_file_groups(file_groups, total, searched=bool(body.search_str)))
@ai_bp.post('/videos/search')
@openapi.definition(
summary='Search videos',
description='Search downloaded videos (and audio) by title and captions. Returns lean results with a'
' WROLPi link and a captions_link for each video. search_str is optional: omit it to browse the'
' newest videos, or pass channel_id (from list_collections) to browse one channel.',
body=schema.AIVideoSearchRequest,
)
@openapi.operation('search_videos')
@openapi.response(HTTPStatus.OK, schema.AISearchResponse)
@validate(schema.AIVideoSearchRequest)
async def search_videos(_: Request, body: schema.AIVideoSearchRequest):
file_groups, total = videos_lib.search_videos(
search_str=body.search_str,
offset=body.offset or 0,
limit=ai_limiter(body.limit),
channel_id=body.channel_id,
tag_names=body.tag_names,
headline=True,
)
return json_response(lib.format_file_groups(file_groups, total, searched=bool(body.search_str)))
@ai_bp.get('/videos/<file_group_id:int>')
@openapi.definition(
summary='Get one video',
description='Get details about one video by its ID from search results: title, channel, description,'
' duration, link, and a captions_link to read what is said in the video.',
)
@openapi.operation('get_video')
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
async def get_video(_: Request, file_group_id: int):
video, _prev, _next = videos_lib.get_video_for_app(file_group_id, skip_viewed=True)
result = lib.format_file_group(video, description_length=lib.DETAIL_DESCRIPTION_LENGTH, detail=True)
return json_response(result)
@ai_bp.get('/videos/<file_group_id:int>/captions')
@openapi.definition(
summary='Get video captions',
description='Read the timestamped captions of a video by its ID; use this to learn what is said in a video'
' without watching it. Long captions are paged: request again with offset=next_offset to'
' continue reading.',
)
@openapi.operation('get_video_captions')
@openapi.parameter('offset', int, 'query')
@openapi.response(HTTPStatus.OK, schema.AIPagedTextResponse)
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
async def get_video_captions(request: Request, file_group_id: int):
video = videos_lib.get_video(file_group_id)
text = lib.format_caption_chunks(video.get_caption_chunks())
return json_response(lib.paginate_text(text, _offset(request)))
@ai_bp.get('/videos/<file_group_id:int>/comments')
@openapi.definition(
summary='Get video comments',
description='Read the downloaded comments of a video by its ID. Long comment threads are paged:'
' request again with offset=next_offset to continue reading.',
)
@openapi.operation('get_video_comments')
@openapi.parameter('offset', int, 'query')
@openapi.response(HTTPStatus.OK, schema.AIPagedTextResponse)
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
async def get_video_comments(request: Request, file_group_id: int):
video = videos_lib.get_video(file_group_id)
comments = video.get_comments() or []
lines = []
for comment in comments:
author = comment.get('author') or 'unknown'
text = (comment.get('text') or '').strip()
if text:
lines.append(f'{author}: {text}')
return json_response(lib.paginate_text('\n'.join(lines), _offset(request)))
@ai_bp.post('/archives/search')
@openapi.definition(
summary='Search archived web pages',
description='Search archived web pages (saved with SingleFile) by title and content. Returns lean results'
' with a WROLPi link and a text_link to read each page. search_str is optional: omit it to browse'
' the newest pages. Filter with domain (the exact name from list_collections) for one site.',
body=schema.AIArchiveSearchRequest,
)
@openapi.operation('search_archives')
@openapi.response(HTTPStatus.OK, schema.AISearchResponse)
@validate(schema.AIArchiveSearchRequest)
async def search_archives(_: Request, body: schema.AIArchiveSearchRequest):
file_groups, total = archive_lib.search_archives(
body.search_str,
body.domain,
ai_limiter(body.limit),
body.offset or 0,
None, # order: default (most recently published)
body.tag_names,
headline=True,
)
return json_response(lib.format_file_groups(file_groups, total, searched=bool(body.search_str)))
@ai_bp.get('/archives/<file_group_id:int>')
@openapi.definition(
summary='Get one archived web page',
description='Get details about one archived web page by its ID from search results: title, source URL,'
' its WROLPi link, a text_link to read it, and earlier snapshots of the same page.',
)
@openapi.operation('get_archive')
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
async def get_archive(request: Request, file_group_id: int):
archive = archive_lib.get_archive_by_file_group_id(request.ctx.session, file_group_id, skip_viewed=True)
return json_response(_archive_detail(archive))
@ai_bp.get('/archives/<file_group_id:int>/text')
@openapi.definition(
summary='Read an archived web page',
description='Read the plain text content of an archived web page by its ID. Long pages are paged: request'
' again with offset=next_offset to continue reading.',
)
@openapi.operation('get_archive_text')
@openapi.parameter('offset', int, 'query')
@openapi.response(HTTPStatus.OK, schema.AIPagedTextResponse)
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
async def get_archive_text(request: Request, file_group_id: int):
archive = archive_lib.get_archive_by_file_group_id(request.ctx.session, file_group_id, skip_viewed=True)
text = lib.read_archive_text(archive)
if text is None:
raise UnknownFile(f'No readable text for archive {file_group_id}')
return json_response(lib.paginate_text(text, _offset(request)))
@ai_bp.post('/docs/search')
@openapi.definition(
summary='Search documents and ebooks',
description='Search documents (ebooks, PDFs, comics, office docs) by title, content, author, or subject.'
' Returns lean results with a WROLPi link for each document. Provide at least one of'
' search_str, author, subject, or tag_names.',
body=schema.AIDocSearchRequest,
)
@openapi.operation('search_docs')
@openapi.response(HTTPStatus.OK, schema.AISearchResponse)
@validate(schema.AIDocSearchRequest)
async def search_docs(_: Request, body: schema.AIDocSearchRequest):
file_groups, total = _search_docs(
search_str=body.search_str,
author=body.author,
subject=body.subject,
language=body.language,
mimetype=body.mimetype,
limit=ai_limiter(body.limit),
offset=body.offset or 0,
order_by='rank' if body.search_str else 'published_datetime',
tag_names=body.tag_names,
)
return json_response(lib.format_file_groups(file_groups, total, searched=bool(body.search_str)))
@ai_bp.get('/docs/<file_group_id:int>')
@openapi.definition(
summary='Get one document',
description='Get details about one document (ebook, PDF, comic, etc.) by its ID from search results:'
' title, author, subject, description, and its WROLPi link.',
)
@openapi.operation('get_doc')
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
async def get_doc(request: Request, file_group_id: int):
doc = _get_doc(request.ctx.session, file_group_id)
response = _doc_response(doc)
file_group = response['file_group']
# Merge the Doc's details into the FileGroup dict so the lean formatter can use them.
file_group['doc'] = response['doc']
result = lib.format_file_group(file_group, description_length=lib.DETAIL_DESCRIPTION_LENGTH, detail=True)
return json_response(result)
@ai_bp.get('/zims')
@openapi.definition(
summary='List Zim encyclopedias',
@ -289,9 +89,8 @@ async def list_zims(request: Request):
@ai_bp.post('/zims/search')
@openapi.definition(
summary='Search Zim encyclopedias',
description='Search Zim encyclopedias (Wikipedia, etc.). Without zim_id this searches only the Zims that'
' have search-by-default enabled; pass zim_id (from list_zims) to search one specific Zim.'
' Use the returned link (or zim_id and path) to read a full entry.',
description='Search the Zim encyclopedias (Wikipedia, etc.) enabled for search; pass zim_id to search one.'
' Read a result in full with get_zim_entry (zim_id and path).',
body=schema.AIZimSearchRequest,
)
@openapi.operation('search_zims')
@ -316,9 +115,8 @@ async def search_zims(request: Request, body: schema.AIZimSearchRequest):
@ai_bp.get('/zims/<zim_id:int>/entry')
@openapi.definition(
summary='Read a Zim entry',
description='Read one article from a Zim encyclopedia as plain text. Pass the path from Zim search'
' results. Long articles are paged: request again with offset=next_offset to continue'
' reading.',
description='Read one Zim article as plain text by zim_id and the path from search_zims. Paged: request'
' again with offset=next_offset.',
)
@openapi.operation('get_zim_entry')
@openapi.parameter('path', str, 'query')
@ -340,8 +138,8 @@ async def get_zim_entry(request: Request, zim_id: int):
@openapi.definition(
summary='List collections',
description='List collections in the library. Filter with kind: "channel" (video channels), "domain"'
' (archived websites), or "playlist". Use a channel\'s id with search_videos, or a domain\'s'
' name with search_archives.',
' (archived websites), or "playlist". Use a channel\'s name or id, or a domain\'s name, as the'
' channel or domain filter of search_files.',
)
@openapi.operation('list_collections')
@openapi.parameter('kind', str, 'query')
@ -394,17 +192,6 @@ async def get_inventory(request: Request):
return json_response(dict(results=results, total=len(results)))
@ai_bp.get('/inventories/<slug:str>')
@openapi.definition(
summary='Get one inventory',
description='Get one inventory in full (its fields and every item) by its slug.',
)
@openapi.operation('get_inventory_items')
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
async def get_inventory_items(_: Request, slug: str):
return json_response(dict(inventory=_inventory_by_slug(slug)))
# ---------------------------------------------------------------------------
# Help endpoints (Help mode). Help markdown is searched in-process; see help_docs.py.
# ---------------------------------------------------------------------------
@ -696,9 +483,8 @@ async def get_map_overview(request: Request):
@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).',
description='Find towns, cities, and landmarks by name in the downloaded maps. Each result has coordinates'
' and a map link. Pass lat and lon to rank nearest first.',
)
@openapi.operation('search_places')
@openapi.parameter('q', str, 'query')
@ -730,50 +516,6 @@ async def search_places(request: Request):
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),
))
# ---------------------------------------------------------------------------
# Consolidated, kind-generic endpoints. Small models carry every tool definition on every request,
# so one search / one detail / one reader replaces the per-kind trio.
@ -835,12 +577,10 @@ async def _resolve_channel_id(session, channel: str) -> Optional[int]:
@ai_bp.post('/files/search')
@openapi.definition(
summary='Search the library',
description='Search videos, archived web pages, and documents/ebooks by title and content. Omit search_str to'
' browse the newest items. kind narrows to video, archive, or doc. channel (a channel name from'
' the library list, or its id) applies to videos; domain (a site name) to archives; author and'
' subject to documents; a filter for another kind is ignored. Results carry an id for get_file'
' and read_content, and a WROLPi link. matches lists channels/domains/authors whose names fit'
' the term: use one to narrow when total is large.',
description='Search videos, archived web pages, and documents by title and content; omit search_str to'
' browse the newest. kind: video, archive, or doc. channel (name or id) is for videos, domain for'
' archives, author/subject for docs. Each result has an id for get_file/read_content and a link.'
' matches names channels/domains/authors that fit the term: narrow with one when total is large.',
body=schema.AIFileSearchRequest,
)
@openapi.operation('search_files')
@ -917,9 +657,8 @@ async def search_files_endpoint(request: Request, body: schema.AIFileSearchReque
@ai_bp.get('/files/<file_group_id:int>')
@openapi.definition(
summary='Get one file',
description='Details of one item by the id from search results, whatever its kind: title, link, date,'
' channel or author, tags, size, full description; for videos whether captions and comments'
' exist; for archived pages the source URL and earlier snapshots.',
description='Details of one item by its id, any kind: title, link, date, channel/author, tags, size, full'
' description; videos say whether captions/comments exist; archives include earlier snapshots.',
)
@openapi.operation('get_file')
@openapi.response(HTTPStatus.NOT_FOUND, JSONErrorResponse)
@ -943,9 +682,8 @@ CONTENT_PARTS = ('text', 'comments')
@ai_bp.get('/files/<file_group_id:int>/content')
@openapi.definition(
summary='Read a file\'s content',
description='Read what an item says, by its id: part=text gives a video\'s timestamped captions or an'
' archived page\'s plain text; part=comments gives a video\'s downloaded comments. Long content'
' is paged: request again with offset=next_offset to continue.',
description='Read an item by its id: part=text is a video\'s captions or an archived page\'s text;'
' part=comments is a video\'s comments. Paged: request again with offset=next_offset.',
)
@openapi.operation('read_content')
@openapi.parameter('part', str, 'query')
@ -1063,11 +801,9 @@ def _resolve_media_path(relative: str | None, verb: str) -> pathlib.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.',
description='List one directory of the media directory by relative path (omit for the top level):'
' directories first, then files, each with a path for list_files or read_file. Paged by offset.'
' For "what is in this folder"; use search_files for content by topic.',
)
@openapi.operation('list_files')
@openapi.parameter('path', str, 'query')
@ -1120,9 +856,8 @@ async def list_files(request: Request):
@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, collection directories, or list_files). Only text files can be read. Long files are'
' paged: request again with offset=next_offset to continue reading.',
description='Read a plain-text file from the media directory by its relative path (from list_files).'
' Text files only. Paged: request again with offset=next_offset.',
)
@openapi.operation('read_file')
@openapi.parameter('path', str, 'query')

2
modules/ai/lib.py
View file

@ -158,7 +158,7 @@ def format_file_groups(file_groups: list, total: int, searched: bool = False) ->
if searched and total == 0:
# Small models give up on one empty search; steer the retry.
result['hint'] = ('No matches. Names may be spelled differently: try fewer or different'
' words, list_collections for exact channel/domain names, or omit'
' words, use the exact channel/website names from the library list, or omit'
' search_str to browse the newest items.')
return result

43
modules/ai/modes.py
View file

@ -29,39 +29,26 @@ MODES = {
system_prompt=(
f'{COMMON_PROMPT}'
' Your job is to be the librarian: find and summarize content from the user\'s library.'
' Search first, then read captions, page text, or entries when the user wants detail.'
' When the user asks about a channel, website, or their whole library, call'
' list_collections first to learn the exact names (spellings often differ).'
' list_collections only gives names — it never tells you what content exists. You have'
' not summarized a channel until you have called search_videos with its channel_id (and'
' NO search_str) and read the resulting titles.'
' To browse everything of one kind (e.g. "summarize my archives"), call the search tool'
' WITHOUT search_str — that lists the newest items.'
' When a search finds nothing, do not give up: retry with fewer or different words, or'
' browse with list_collections.'
' Search first with search_files, then read captions, page text, or entries with read_content'
' when the user wants detail. The library list above has the exact channel, website, and'
' Zim names (spellings often differ): pass a channel or domain name to search_files directly.'
' To summarize a channel or browse everything of one kind, call search_files WITHOUT search_str'
' and with the channel, domain, or kind; that lists the newest items.'
' When a search finds nothing, do not give up: retry with fewer or different words, or drop the'
' filters. When it offers matches, narrow with one of them.'
' ALWAYS include the WROLPi link for every item you mention, exactly as the tools return it.'
' Never give external URLs. When a search returns a large total, narrow the query instead'
' of paging.'
),
tools=(
'search_all',
'search_videos',
'get_video',
'get_video_captions',
'search_archives',
'get_archive_text',
'search_docs',
'get_doc',
'list_zims',
'search_files',
'get_file',
'read_content',
'search_zims',
'get_zim_entry',
'list_collections',
'get_inventory',
'list_files',
'read_file',
'list_tags',
'search_suggestions',
'get_map_overview',
'search_places',
),
),
@ -127,15 +114,11 @@ MODE_EXAMPLES = {
),
'research': _example(
dict(role='user', content='Summarize the bakehouse channel'),
_tool_call('ex_res_1', 'list_collections', '{"kind": "channel"}'),
_tool_call('ex_res_1', 'search_files', '{"channel": "Bakehouse"}'),
_tool_result('ex_res_1',
'{"results": [{"id": 7, "name": "Bakehouse", "kind": "channel",'
' "link": "/videos/channel/7/video"}], "total": 1}'),
_tool_call('ex_res_2', 'search_videos', '{"channel_id": 7}'),
_tool_result('ex_res_2',
'{"results": [{"id": 12, "kind": "video", "title": "Sourdough Basics",'
' "link": "/videos/12", "channel": "Bakehouse", "duration": 900}], "total": 1}'),
dict(role='assistant', content='The Bakehouse channel (/videos/channel/7/video) has one video:\\n'
' "link": "/videos/12", "channel": "Bakehouse", "duration": "15:00"}], "total": 1}'),
dict(role='assistant', content='The Bakehouse channel has one video:\\n'
'1. **Sourdough Basics** (/videos/12) — a 15 minute introduction'
' to sourdough baking.'),
),

45
modules/ai/schema.py
View file

@ -2,45 +2,6 @@ import dataclasses
from typing import List, Optional
@dataclasses.dataclass
class AISearchRequest:
search_str: Optional[str] = None
limit: int = 5
offset: int = 0
mimetypes: Optional[List[str]] = None
tag_names: Optional[List[str]] = None
@dataclasses.dataclass
class AIVideoSearchRequest:
search_str: Optional[str] = None
limit: int = 5
offset: int = 0
channel_id: Optional[int] = None
tag_names: Optional[List[str]] = None
@dataclasses.dataclass
class AIArchiveSearchRequest:
search_str: Optional[str] = None
limit: int = 5
offset: int = 0
domain: Optional[str] = None
tag_names: Optional[List[str]] = None
@dataclasses.dataclass
class AIDocSearchRequest:
search_str: Optional[str] = None
author: Optional[str] = None
subject: Optional[str] = None
language: Optional[str] = None
mimetype: Optional[str] = None
limit: int = 5
offset: int = 0
tag_names: Optional[List[str]] = None
@dataclasses.dataclass
class AIZimSearchRequest:
search_str: Optional[str] = None
@ -85,12 +46,6 @@ class AIListFilesResponse:
next_offset: Optional[int] = None
@dataclasses.dataclass
class AISearchSuggestionsRequest:
search_str: Optional[str] = None
tag_names: List[str] = dataclasses.field(default_factory=list)
@dataclasses.dataclass
class AIFileSearchRequest:
search_str: Optional[str] = None

22
modules/ai/test/test_agent.py
View file

@ -36,17 +36,17 @@ async def test_tool_definitions_from_spec(tool_definitions):
assert not any(t.path.startswith('/api/chat') for t in tools.values())
# Body, query, and path parameters all appear in the flat schema.
schema = tools['search_videos'].parameters_schema()
schema = tools['search_files'].parameters_schema()
assert schema['properties']['search_str'] == {'type': 'string'}
assert schema['properties']['tag_names'] == {'type': 'array', 'items': {'type': 'string'}}
schema = tools['get_video_captions'].parameters_schema()
schema = tools['read_content'].parameters_schema()
assert schema['required'] == ['file_group_id']
assert schema['properties']['offset'] == {'type': 'integer'}
# OpenAI tool format.
openai_tool = tools['search_all'].as_openai_tool()
openai_tool = tools['search_files'].as_openai_tool()
assert openai_tool['type'] == 'function'
assert openai_tool['function']['name'] == 'search_all'
assert openai_tool['function']['name'] == 'search_files'
def _endpoint_shim(async_client):
@ -70,8 +70,8 @@ async def test_execute_tool(async_client, tool_definitions, video_factory, test_
with _endpoint_shim(async_client):
# A real search through the blueprint.
success, result = await agent.execute_tool(research_tools, 'search_videos',
dict(search_str='canning'))
success, result = await agent.execute_tool(research_tools, 'search_files',
dict(search_str='canning', kind='video'))
assert success is True
result = json.loads(result)
assert result['total'] == 1
@ -86,20 +86,20 @@ async def test_execute_tool(async_client, tool_definitions, video_factory, test_
assert success is False and 'Unknown tool' in error
# Unknown argument names the accepted ones.
success, error = await agent.execute_tool(research_tools, 'search_videos', dict(query='x'))
success, error = await agent.execute_tool(research_tools, 'search_files', dict(query='x'))
assert success is False and 'accepts' in error and 'search_str' in error
# Missing required path parameter.
success, error = await agent.execute_tool(research_tools, 'get_video_captions', {})
success, error = await agent.execute_tool(research_tools, 'read_content', {})
assert success is False and 'Missing required' in error
# Wrong type.
success, error = await agent.execute_tool(research_tools, 'get_video_captions',
success, error = await agent.execute_tool(research_tools, 'read_content',
dict(file_group_id='banana'))
assert success is False and 'must be a integer' in error
# An endpoint error comes back as a message, not an exception.
success, error = await agent.execute_tool(research_tools, 'get_video_captions',
success, error = await agent.execute_tool(research_tools, 'read_content',
dict(file_group_id=123456))
assert success is False and 'error' in error.lower()
@ -129,7 +129,7 @@ async def test_execute_tool_result_truncated(async_client, tool_definitions, arc
archive = test_session.query(Archive).one()
with _endpoint_shim(async_client):
success, result = await agent.execute_tool(MODES['research']['tools'], 'get_archive_text',
success, result = await agent.execute_tool(MODES['research']['tools'], 'read_content',
dict(file_group_id=archive.file_group_id))
assert success is True
assert len(result) <= agent.TOOL_RESULT_MAX_CHARS + 100

274
modules/ai/test/test_api.py
View file

@ -7,199 +7,6 @@ import pytest
from modules.ai import lib
@pytest.mark.asyncio
async def test_ai_search_all(async_client, test_session, video_factory, archive_factory):
"""Global search returns lean results of every kind with links and a total."""
video_factory(title='canning tomatoes')
archive_factory(domain='example.com', title='canning peppers', contents='all about canning peppers')
test_session.commit()
request, response = await async_client.post('/api/ai/search', content=json.dumps(dict(search_str='canning')))
assert response.status_code == HTTPStatus.OK
assert response.json['total'] == 2
kinds = {i['kind'] for i in response.json['results']}
assert kinds == {'video', 'archive'}
for result in response.json['results']:
assert result['link']
assert result['id']
@pytest.mark.asyncio
async def test_ai_search_all_limit(async_client, test_session, archive_factory):
"""The limit is clamped and the total reports all matches so the model narrows, not pages."""
for i in range(3):
archive_factory(domain='example.com', title=f'gardening {i}', contents='gardening guide')
test_session.commit()
content = dict(search_str='gardening', limit=1)
request, response = await async_client.post('/api/ai/search', content=json.dumps(content))
assert response.status_code == HTTPStatus.OK
assert len(response.json['results']) == 1
assert response.json['total'] == 3
# An excessive limit is clamped to the maximum.
content = dict(search_str='gardening', limit=10_000)
request, response = await async_client.post('/api/ai/search', content=json.dumps(content))
assert response.status_code == HTTPStatus.OK
@pytest.mark.asyncio
async def test_ai_search_videos(async_client, test_session, video_factory):
"""Video search returns lean video results with captions links."""
video_factory(title='sourdough starter', with_caption_file=True)
video_factory(title='unrelated')
test_session.commit()
content = dict(search_str='sourdough')
request, response = await async_client.post('/api/ai/videos/search', content=json.dumps(content))
assert response.status_code == HTTPStatus.OK
assert response.json['total'] == 1
result = response.json['results'][0]
assert result['kind'] == 'video'
assert result['link'] == f'/videos/{result["id"]}'
assert 'captions_link' not in result # derivable from the id; kept out of listings
@pytest.mark.asyncio
async def test_ai_get_video(async_client, test_session, video_factory):
"""A single video can be fetched by its FileGroup ID."""
video = video_factory(title='wood stove install')
test_session.commit()
file_group_id = video.file_group_id
request, response = await async_client.get(f'/api/ai/videos/{file_group_id}')
assert response.status_code == HTTPStatus.OK
assert response.json['id'] == file_group_id
assert response.json['kind'] == 'video'
assert response.json['link'] == f'/videos/{file_group_id}'
# Fetching does not mark the video viewed.
test_session.expire_all()
assert video.file_group.viewed is None
request, response = await async_client.get('/api/ai/videos/123456')
assert response.status_code == HTTPStatus.NOT_FOUND
@pytest.mark.asyncio
async def test_ai_get_video_captions(async_client, test_session, video_factory):
"""Captions are returned as timestamped, paged text."""
video = video_factory(title='captioned', with_caption_file=True)
test_session.commit()
request, response = await async_client.get(f'/api/ai/videos/{video.file_group_id}/captions')
assert response.status_code == HTTPStatus.OK
assert response.json['content'].startswith('[00:00:0')
assert response.json['total_chars'] > 0
# An offset beyond the end returns an empty page, never an error.
offset = response.json['total_chars'] + 100
request, response = await async_client.get(f'/api/ai/videos/{video.file_group_id}/captions?offset={offset}')
assert response.status_code == HTTPStatus.OK
assert response.json['content'] == ''
assert response.json['next_offset'] is None
@pytest.mark.asyncio
async def test_ai_search_archives(async_client, test_session, archive_factory):
"""Archive search supports the domain filter and returns text links."""
archive_factory(domain='example.com', title='pressure canning', contents='pressure canning guide')
archive_factory(domain='other.org', title='water bath canning', contents='water bath canning guide')
test_session.commit()
content = dict(search_str='canning', domain='example.com')
request, response = await async_client.post('/api/ai/archives/search', content=json.dumps(content))
assert response.status_code == HTTPStatus.OK
assert response.json['total'] == 1
result = response.json['results'][0]
assert result['kind'] == 'archive'
assert 'text_link' not in result # derivable from the id; kept out of listings
@pytest.mark.asyncio
async def test_ai_get_archive_text(async_client, test_session, archive_factory):
"""Archive text is served from the readability text file and paged."""
contents = 'word ' * 2_000 # 10,000 chars, more than two pages.
archive = archive_factory(domain='example.com', title='long read', contents=contents)
test_session.commit()
file_group_id = archive.file_group_id
request, response = await async_client.get(f'/api/ai/archives/{file_group_id}/text')
assert response.status_code == HTTPStatus.OK
assert len(response.json['content']) == lib.PAGE_SIZE
assert response.json['next_offset'] == lib.PAGE_SIZE
assert response.json['total_chars'] == len(contents)
# The second page continues where the first ended.
request, response = await async_client.get(
f'/api/ai/archives/{file_group_id}/text?offset={response.json["next_offset"]}')
assert response.status_code == HTTPStatus.OK
assert response.json['content'] == contents[lib.PAGE_SIZE:2 * lib.PAGE_SIZE]
request, response = await async_client.get('/api/ai/archives/123456/text')
assert response.status_code == HTTPStatus.NOT_FOUND
@pytest.mark.asyncio
async def test_ai_get_archive(async_client, test_session, archive_factory):
"""A single archive can be fetched with its history of snapshots."""
archive_factory(domain='example.com', url='https://example.com/a', title='first', contents='one')
archive = archive_factory(domain='example.com', url='https://example.com/a', title='second', contents='two')
test_session.commit()
request, response = await async_client.get(f'/api/ai/archives/{archive.file_group_id}')
assert response.status_code == HTTPStatus.OK
assert response.json['kind'] == 'archive'
assert response.json['link'] == f'/archives/{archive.file_group_id}'
# History holds the OTHER snapshots of the URL, not the archive itself.
assert [i['title'] for i in response.json['history']] == ['first']
request, response = await async_client.get('/api/ai/archives/123456')
assert response.status_code == HTTPStatus.NOT_FOUND
@pytest.mark.asyncio
async def test_ai_get_video_comments(async_client, test_session, video_factory):
"""Video comments render as paged author lines."""
info_json = dict(duration=5, comments=[
dict(author='alice', text='great video'),
dict(author='bob', text='thanks'),
])
video = video_factory(title='commented', with_info_json=info_json)
test_session.commit()
request, response = await async_client.get(f'/api/ai/videos/{video.file_group_id}/comments')
assert response.status_code == HTTPStatus.OK
assert 'alice: great video' in response.json['content']
assert 'bob: thanks' in response.json['content']
# No comments is an empty page, not an error.
no_comments = video_factory(title='quiet')
test_session.commit()
request, response = await async_client.get(f'/api/ai/videos/{no_comments.file_group_id}/comments')
assert response.status_code == HTTPStatus.OK
assert response.json['content'] == ''
@pytest.mark.asyncio
async def test_ai_search_docs(async_client, test_session, test_directory, example_epub, refresh_files):
"""Docs can be searched and fetched with links."""
await refresh_files()
content = dict(search_str='WROLPi')
request, response = await async_client.post('/api/ai/docs/search', content=json.dumps(content))
assert response.status_code == HTTPStatus.OK
assert response.json['total'] >= 1
result = response.json['results'][0]
assert result['kind'] == 'doc'
assert result['link'] == f'/docs/{result["id"]}'
request, response = await async_client.get(f'/api/ai/docs/{result["id"]}')
assert response.status_code == HTTPStatus.OK
assert response.json['id'] == result['id']
assert response.json['kind'] == 'doc'
@pytest.mark.asyncio
async def test_ai_zims(async_client, test_session, test_zim):
"""Zims can be listed, searched, and their entries read as paged text."""
@ -249,26 +56,6 @@ async def test_ai_list_collections(async_client, test_session, archive_factory):
assert response.json['total'] == 0
@pytest.mark.asyncio
async def test_ai_inventories(async_client, food_inventory_factory):
"""Inventories are listed lean, and fetched in full by slug."""
slug = food_inventory_factory(items=[dict(name='rice', count=2)])
request, response = await async_client.get('/api/ai/inventories')
assert response.status_code == HTTPStatus.OK
assert response.json['total'] >= 1
lean = next(i for i in response.json['results'] if i['slug'] == slug)
assert lean['item_count'] == 1
assert 'items' not in lean
request, response = await async_client.get(f'/api/ai/inventories/{slug}')
assert response.status_code == HTTPStatus.OK
assert response.json['inventory']['items'][0]['name'] == 'rice'
request, response = await async_client.get('/api/ai/inventories/no-such-inventory')
assert response.status_code == HTTPStatus.NOT_FOUND
@pytest.mark.asyncio
async def test_ai_read_file(async_client, test_directory):
"""Text files in the media directory can be read; everything else is refused."""
@ -478,39 +265,6 @@ async def test_ai_map(async_client, test_session, test_directory, make_files_str
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
@ -578,11 +332,21 @@ async def test_ai_search_files_kinds(async_client, test_session, video_factory,
assert {'id', 'kind', 'title', 'link'} <= set(result)
assert 'captions_link' not in result and 'size' not in result and 'mimetype' not in result
# Something to search by is required.
# No filters at all browses the newest items.
request, response = await async_client.post('/api/ai/files/search', content=json.dumps({}))
assert response.status_code == HTTPStatus.OK # browse newest
assert response.status_code == HTTPStatus.OK
assert response.json['total'] == 3
# The limit is clamped and total reports every match so the model narrows instead of paging.
data = await search(search_str='cooking', limit=1)
assert len(data['results']) == 1 and data['total'] == 3
data = await search(search_str='cooking', limit=10_000)
assert len(data['results']) == 3
# An unknown kind is refused.
request, response = await async_client.post('/api/ai/files/search', content=json.dumps(dict(kind='song')))
assert response.status_code == HTTPStatus.BAD_REQUEST
@pytest.mark.asyncio
async def test_ai_search_files_docs(async_client, test_session, test_directory, example_epub, refresh_files):
@ -651,6 +415,20 @@ async def test_ai_read_content(async_client, test_session, video_factory, archiv
request, response = await async_client.get(f'/api/ai/files/{video.file_group_id}/content')
assert response.status_code == HTTPStatus.OK
assert response.json['content'].startswith('[00:00:0')
assert response.json['total_chars'] > 0
# An offset beyond the end returns an empty page, never an error.
offset = response.json['total_chars'] + 100
request, response = await async_client.get(f'/api/ai/files/{video.file_group_id}/content?offset={offset}')
assert response.status_code == HTTPStatus.OK
assert response.json['content'] == '' and response.json['next_offset'] is None
# A video without comments is an empty page, not an error.
quiet = video_factory(title='quiet')
test_session.commit()
request, response = await async_client.get(f'/api/ai/files/{quiet.file_group_id}/content?part=comments')
assert response.status_code == HTTPStatus.OK
assert response.json['content'] == ''
request, response = await async_client.get(f'/api/ai/files/{video.file_group_id}/content?part=comments')
assert response.status_code == HTTPStatus.OK

6
modules/ai/test/test_chat_api.py
View file

@ -82,7 +82,7 @@ async def test_chat_tool_call_then_answer(async_client, ai_enabled, video_factor
patch, calls = _llama_script(
[dict(type='stop', tool_calls=[
dict(id='call_1', name='search_videos', arguments='{"search_str": "canning"}')])],
dict(id='call_1', name='search_files', arguments='{"search_str": "canning", "kind": "video"}')])],
[dict(type='token', content='Found '), dict(type='token', content='one video.'),
dict(type='stop', tool_calls=[])],
)
@ -96,8 +96,8 @@ async def test_chat_tool_call_then_answer(async_client, ai_enabled, video_factor
names = [e for e, _ in events if e != 'status']
assert names == ['tool_call', 'tool_result', 'token', 'token', 'done']
tool_call = dict(events)['tool_call']
assert tool_call == dict(tool='search_videos', args=dict(search_str='canning'))
assert dict(events)['tool_result'] == dict(tool='search_videos', success=True)
assert tool_call == dict(tool='search_files', args=dict(search_str='canning', kind='video'))
assert dict(events)['tool_result'] == dict(tool='search_files', success=True)
assert dict(events)['done'] == dict(content='Found one video.')
# The second llama call saw the system prompt (with the live library map), the mode's

2
modules/ai/test/test_mcp_parity.py
View file

@ -18,7 +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',
'list_tags', 'list_downloads', 'get_map_overview', 'search_places',
'search_files', 'get_file', 'read_content',
}

17
scripts/ai_eval.py
View file

@ -26,15 +26,17 @@ import requests
requests.packages.urllib3.disable_warnings() # noqa Self-signed certs.
# (mode, question, expected first tools (any of), expect a WROLPi link in the answer)
# (mode, question, expected first tools (any of; None = answerable from context, no tool expected),
# expect a WROLPi link in the answer)
QUESTIONS = [
('help', 'How do I download videos?', {'search_help'}, True),
('help', 'What are Tags?', {'search_help'}, True),
('help', 'How do I add my own files to WROLPi?', {'search_help'}, True),
('research', 'What are my three newest archives?', {'search_archives'}, True),
('research', 'What video channels do I have?', {'list_collections', 'search_videos'}, False),
('research', 'Find videos about cooking', {'search_videos', 'search_all'}, False),
('research', 'Summarize my newest videos', {'search_videos'}, True),
('research', 'What are my three newest archives?', {'search_files'}, True),
# The library map in the prompt already names the channels; no tool call is needed.
('research', 'What video channels do I have?', None, False),
('research', 'Find videos about cooking', {'search_files'}, False),
('research', 'Summarize my newest videos', {'search_files'}, True),
('system', 'Is everything running?', {'get_system_status', 'list_services'}, False),
('system', 'How full are my drives?', {'get_system_status', 'list_disks'}, False),
('system', 'Why would downloads fail?', {'get_system_status', 'search_help'}, False),
@ -73,10 +75,11 @@ def score(result: dict, expected_tools: set, expect_link: bool) -> dict:
answer = result.get('answer') or ''
checks = {
'done': result.get('error') is None and bool(answer),
'used_tool': bool(result['tools']),
'right_tool': bool(result['tools']) and result['tools'][0] in expected_tools,
'no_external_urls': not EXTERNAL_URL.search(answer),
}
if expected_tools is not None:
checks['used_tool'] = bool(result['tools'])
checks['right_tool'] = bool(result['tools']) and result['tools'][0] in expected_tools
if expect_link:
checks['has_link'] = bool(RELATIVE_LINK.search(answer))
return checks

16
scripts/api_perf.py
View file

@ -198,14 +198,16 @@ CATALOG = [
dict(name='ai/collections', method='GET', path='/api/ai/collections'),
dict(name='ai/inventories', method='GET', path='/api/ai/inventories'),
dict(name='ai/zims', method='GET', path='/api/ai/zims'),
dict(name='ai/search', method='POST', path='/api/ai/search', body={'search_str': '{q}', 'limit': 5}),
dict(name='ai/videos/search', method='POST', path='/api/ai/videos/search', body={'search_str': '{q}', 'limit': 5}),
dict(name='ai/archives/search', method='POST', path='/api/ai/archives/search',
body={'search_str': '{q}', 'limit': 5}),
dict(name='ai/docs/search', method='POST', path='/api/ai/docs/search', body={'search_str': '{q}', 'limit': 5}),
dict(name='ai/files/search', method='POST', path='/api/ai/files/search', body={'search_str': '{q}', 'limit': 5}),
dict(name='ai/files/search video', method='POST', path='/api/ai/files/search',
body={'search_str': '{q}', 'kind': 'video', 'limit': 5}),
dict(name='ai/files/search archive', method='POST', path='/api/ai/files/search',
body={'search_str': '{q}', 'kind': 'archive', 'limit': 5}),
dict(name='ai/files/search doc', method='POST', path='/api/ai/files/search',
body={'search_str': '{q}', 'kind': 'doc', 'limit': 5}),
dict(name='ai/zims/search', method='POST', path='/api/ai/zims/search', body={'search_str': '{q}', 'limit': 5}),
dict(name='ai/videos/<id>', method='GET', path='/api/ai/videos/{video_id}', needs='video_id'),
dict(name='ai/archives/<id>', method='GET', path='/api/ai/archives/{archive_id}', needs='archive_id'),
dict(name='ai/files/<video id>', method='GET', path='/api/ai/files/{video_id}', needs='video_id'),
dict(name='ai/files/<archive id>', method='GET', path='/api/ai/files/{archive_id}', needs='archive_id'),
dict(name='chat/modes', method='GET', path='/api/chat/modes'),
]

35
wrolpi_mcp/server.py
View file

@ -586,41 +586,6 @@ async def list_tags() -> str:
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.