ai: page list_collections

A library with a few hundred channels returned 137k characters from one
list_collections call, far past any tool-result budget.  The endpoint
now takes limit/offset (clamped like the other AI listings) and a
search_str name filter, and reports total and next_offset; the MCP tool
passes them through and tells the caller how to continue.

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

modules/ai/api.py

@@ -137,17 +137,26 @@ async def get_zim_entry(request: Request, zim_id: int):
 @ai_bp.get('/collections')
 @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 name or id, or a domain\'s name, as the'
-                ' channel or domain filter of search_files.',
+    description='List collections in the library, paged. Filter with kind: "channel" (video channels),'
+                ' "domain" (archived websites), or "playlist"; search_str matches names. 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')
+@openapi.parameter('search_str', str, 'query')
+@openapi.parameter('limit', int, 'query')
+@openapi.parameter('offset', int, 'query')
 async def list_collections(request: Request):
     kind = request.args.get('kind')
-    collections = search_collections(request.ctx.session, kind=kind)
+    search_str = request.args.get('search_str') or None
+    try:
+        limit = ai_limiter(int(request.args.get('limit', 0)) or None)
+    except ValueError:
+        raise ValidationError('limit must be an integer')
+    offset = _offset(request)
+    collections = search_collections(request.ctx.session, kind=kind, search_str=search_str)
     results = []
-    for collection in collections:
+    for collection in collections[offset:offset + limit]:
         # Channels have a browsable page; a real link stops the model inventing one.
         link = f"/videos/channel/{collection['id']}/video" if collection.get('kind') == 'channel' else None
         results.append({k: v for k, v in dict(
@@ -157,7 +166,8 @@ async def list_collections(request: Request):
             directory=collection.get('directory'),
             link=link,
         ).items() if v is not None})
-    return json_response(dict(results=results, total=len(results)))
+    next_offset = offset + limit if offset + limit < len(collections) else None
+    return json_response(dict(results=results, total=len(collections), next_offset=next_offset))
 
 
 def _lean_inventories() -> list:

modules/ai/test/test_api.py

@@ -40,9 +40,11 @@ async def test_ai_zims(async_client, test_session, test_zim):
 
 
 @pytest.mark.asyncio
-async def test_ai_list_collections(async_client, test_session, archive_factory):
-    """Collections can be listed and filtered by kind."""
+async def test_ai_list_collections(async_client, test_session, archive_factory, channel_factory):
+    """Collections are listed paged, filtered by kind, and searchable by name."""
     archive_factory(domain='example.com', title='a page', contents='contents')
+    for name in ('Alpha Cooking', 'Beta Cooking', 'Gamma Woodwork'):
+        channel_factory(name=name)
     test_session.commit()
 
     request, response = await async_client.get('/api/ai/collections?kind=domain')
@@ -50,10 +52,21 @@ async def test_ai_list_collections(async_client, test_session, archive_factory):
     assert response.json['total'] == 1
     assert response.json['results'][0]['name'] == 'example.com'
     assert response.json['results'][0]['kind'] == 'domain'
+    assert response.json['next_offset'] is None
 
-    request, response = await async_client.get('/api/ai/collections?kind=channel')
+    # Paged, with the full total so the model knows how many there are.
+    request, response = await async_client.get('/api/ai/collections?kind=channel&limit=2')
     assert response.status_code == HTTPStatus.OK
-    assert response.json['total'] == 0
+    assert response.json['total'] == 3 and len(response.json['results']) == 2
+    assert response.json['next_offset'] == 2
+    request, response = await async_client.get('/api/ai/collections?kind=channel&limit=2&offset=2')
+    assert [i['name'] for i in response.json['results']] == ['Gamma Woodwork']
+    assert response.json['next_offset'] is None
+
+    # Name search.
+    request, response = await async_client.get('/api/ai/collections?search_str=cooking')
+    assert response.status_code == HTTPStatus.OK
+    assert sorted(i['name'] for i in response.json['results']) == ['Alpha Cooking', 'Beta Cooking']
 
 
 @pytest.mark.asyncio

wrolpi_mcp/server.py

@@ -276,13 +276,25 @@ async def read_content(file_group_id: int, part: str = "text") -> str:
 # ---------------------------------------------------------------------------
 
 @mcp.tool(annotations={"readOnlyHint": True})
-async def list_collections(kind: str | None = None) -> str:
-    """List all collections (channels, domains, etc.) in the library.
+async def list_collections(
+    kind: str | None = None,
+    query: str | None = None,
+    limit: int = DEFAULT_LIMIT,
+    offset: int = 0,
+) -> str:
+    """List collections (video channels, archived domains, playlists) in the library, paged.
 
     Args:
-        kind: Filter by collection kind: "channel", "domain", or None for all.
+        kind: Filter by collection kind: "channel", "domain", "playlist", or None for all.
+        query: Match collection names containing this text.
+        limit: Maximum results.
+        offset: Pagination offset (from a previous listing's "next offset").
     """
-    params = {"kind": kind} if kind else None
+    params = {"limit": limit, "offset": offset}
+    if kind:
+        params["kind"] = kind
+    if query:
+        params["search_str"] = query
     data = await api_get("/api/ai/collections", params=params)
     collections = data.get("results") or []
     if not collections:
@@ -294,6 +306,8 @@ async def list_collections(kind: str | None = None) -> str:
             parts.append(f"  Directory: {collection['directory']}")
         lines.append("\n".join(parts))
     lines.append(f"\nTotal: {data.get('total', len(collections))}")
+    if (next_offset := data.get("next_offset")) is not None:
+        lines.append(f"More available: call list_collections again with offset={next_offset}")
     return "\n\n".join(lines)