rt/wrolpi
1
0
Fork 0
mirror of https://github.com/lrnselfreliance/wrolpi.git synced 2026-10-25 18:40:06 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 5

ai: wrolpi_mcp exposes only the compact tool set

Browse source 
The per-kind MCP wrappers (search/search_videos/search_archives/
search_docs, get_video/get_archive/get_doc, get_video_captions/
get_video_comments/get_archive_text) duplicated search_files, get_file,
and read_content and cost an external client ten redundant definitions
on every request.  They are gone.  search_default_zims folds into
search_zim (zim_id optional), get_inventory_items into get_inventory
(slug optional), and read_file is exposed for the first time.  16 tools,
matching what the built-in assistant sees plus the read-only listings.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Roland 2026-09-10 16:35:04 -06:00
parent 10f0a4a46c
commit c6d2ec783e
No known key found for this signature in database
GPG key ID: 1BA620CE6C9A4680
2 changed files with 53 additions and 249 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -1,29 +1,26 @@
"""Guards for the wrolpi_mcp thin proxy.
The MCP server's venv (with the `mcp` package) is separate from the API's, so these are source
scans: external clients depend on the tool names, and the proxy must only reach the tested
/api/ai blueprint (plus the two whole-API reads that predate it)."""
scans: the tool set is deliberate (no per-kind duplicates of the generic tools), and the proxy
must only reach the tested /api/ai blueprint (plus the whole-API statistics read)."""
import re
from wrolpi.vars import PROJECT_DIR
MCP_SERVER = PROJECT_DIR / 'wrolpi_mcp/server.py'
# The tool names external clients (Claude etc.) already use. Renaming one breaks them.
# The MCP mirrors the compact catalog the built-in assistant uses, plus the read-only listings
# (collections, zims, tags, downloads, map, statistics, status) that only external clients need.
EXPECTED_TOOLS = {
'search', 'search_videos', 'search_archives', 'search_docs',
'search_zim', 'search_default_zims',
'get_video', 'get_video_captions', 'get_video_comments',
'get_archive', 'get_archive_text',
'get_doc', 'get_zim_entry',
'list_collections', 'list_zim_files', 'list_files',
'get_statistics', 'get_inventory', 'get_inventory_items', 'get_status',
'list_tags', 'list_downloads', 'get_map_overview', 'search_places',
'search_files', 'get_file', 'read_content',
'search_zim', 'get_zim_entry', 'list_zim_files',
'list_collections', 'list_tags', 'list_files', 'read_file',
'list_downloads', 'get_map_overview', 'search_places',
'get_statistics', 'get_inventory', 'get_status',
}
def test_mcp_tool_names_unchanged():
def test_mcp_tool_set():
text = MCP_SERVER.read_text()
tools = set(re.findall(r'@mcp\.tool[^\n]*\)\nasync def (\w+)\(', text))
assert tools == EXPECTED_TOOLS

281
wrolpi_mcp/server.py
View file

@ -1,8 +1,8 @@
"""WROLPi MCP Server — exposes WROLPi content to LLMs via the Model Context Protocol.
A thin proxy: every tool is one call to the WROLPi API's /api/ai blueprint, which owns (and
tests) the lean result shapes, links, and paging that used to live here. External clients
(Claude etc.) and WROLPi's local assistant share that one behavior."""
tests) the lean result shapes, links, and paging. External clients (Claude etc.) and WROLPi's
local assistant share the same compact, kind-generic tools."""
import json
import logging
import sys
@ -23,7 +23,8 @@ mcp = FastMCP(
instructions=(
"WROLPi is an offline digital library containing videos, archived web pages, "
"ebooks, Zim encyclopedias (Wikipedia, etc.), maps, and documents. "
"Use these tools to search and retrieve content from the library. "
"Use search_files to find content of any kind, get_file for details, and read_content "
"for captions, page text, or comments. "
"IMPORTANT: When presenting results to the user, ALWAYS include the 'WROLPi Link' "
"for every item. Never use the 'Source URL' — only use the local WROLPi Link. "
"When calling tools, use the top-level ID (FileGroup ID)."
@ -156,154 +157,27 @@ def _render_zim_entries(data: dict) -> str:
return text
# ---------------------------------------------------------------------------
# Search tools
# ---------------------------------------------------------------------------
@mcp.tool(annotations={"readOnlyHint": True})
async def search(
query: str,
limit: int = DEFAULT_LIMIT,
offset: int = 0,
mimetypes: list[str] | None = None,
tag_names: list[str] | None = None,
) -> str:
"""Search all WROLPi content (videos, archives, ebooks, and other files).
Args:
query: Text to search for across titles, content, and metadata.
limit: Maximum number of results to return.
offset: Number of results to skip (for pagination).
mimetypes: Filter by MIME types, e.g. ["video/mp4", "text/html"].
tag_names: Filter by tag names.
"""
body = {"search_str": query, "limit": limit, "offset": offset}
if tag_names:
body["tag_names"] = tag_names
# mimetypes are no longer a filter; a kind is. Map the common prefixes.
if mimetypes:
kinds = {("video" if m.startswith("video/") else "archive" if m == "text/html" else "doc") for m in mimetypes}
if len(kinds) == 1:
body["kind"] = kinds.pop()
return _render_results(await api_post("/api/ai/files/search", json=body))
@mcp.tool(annotations={"readOnlyHint": True})
async def search_videos(
query: str,
limit: int = DEFAULT_LIMIT,
offset: int = 0,
channel_id: int | None = None,
tag_names: list[str] | None = None,
) -> str:
"""Search downloaded videos by title and captions.
Args:
query: Text to search for in video titles and captions.
limit: Maximum number of results.
offset: Pagination offset.
channel_id: Filter to a specific channel by ID.
tag_names: Filter by tag names.
"""
body = {"search_str": query, "limit": limit, "offset": offset, "kind": "video"}
if channel_id is not None:
body["channel"] = str(channel_id)
if tag_names:
body["tag_names"] = tag_names
return _render_results(await api_post("/api/ai/files/search", json=body))
@mcp.tool(annotations={"readOnlyHint": True})
async def search_archives(
query: str,
limit: int = DEFAULT_LIMIT,
offset: int = 0,
domain: str | None = None,
tag_names: list[str] | None = None,
) -> str:
"""Search archived web pages (saved with SingleFile).
Args:
query: Text to search for in archived page titles and content.
limit: Maximum number of results.
offset: Pagination offset.
domain: Filter to a specific domain (e.g. "example.com").
tag_names: Filter by tag names.
"""
body = {"search_str": query, "limit": limit, "offset": offset, "kind": "archive"}
if domain:
body["domain"] = domain
if tag_names:
body["tag_names"] = tag_names
return _render_results(await api_post("/api/ai/files/search", json=body))
@mcp.tool(annotations={"readOnlyHint": True})
async def search_docs(
query: str | None = None,
author: str | None = None,
subject: str | None = None,
language: str | None = None,
mimetype: str | None = None,
limit: int = DEFAULT_LIMIT,
offset: int = 0,
tag_names: list[str] | None = None,
) -> str:
"""Search documents (ebooks, PDFs, comics, office docs) in the library.
At least one of query, author, subject, or tag_names should be provided.
Args:
query: Text to search for in document titles and content.
author: Filter by author name (partial match).
subject: Filter by subject (partial match).
language: Filter by language code (exact match, e.g. "en").
mimetype: Filter by MIME type prefix (e.g. "application/epub", "application/pdf").
limit: Maximum number of results.
offset: Pagination offset.
tag_names: Filter by tag names.
"""
body = {"limit": limit, "offset": offset, "kind": "doc"}
if query:
body["search_str"] = query
if author:
body["author"] = author
if subject:
body["subject"] = subject
if tag_names:
body["tag_names"] = tag_names
# language and mimetype are accepted for compatibility; the consolidated search does not filter on them.
return _render_results(await api_post("/api/ai/files/search", json=body))
@mcp.tool(annotations={"readOnlyHint": True})
async def get_doc(file_group_id: int) -> str:
"""Get detailed information about a specific document (ebook, PDF, comic, etc.).
Args:
file_group_id: The top-level ID from search results (this is the FileGroup ID).
"""
return _render_result(await api_get(f"/api/ai/files/{file_group_id}"))
@mcp.tool(annotations={"readOnlyHint": True})
async def search_zim(
zim_id: int,
query: str,
zim_id: int | None = None,
limit: int = DEFAULT_LIMIT,
offset: int = 0,
) -> str:
"""Search within a Zim encyclopedia (e.g. Wikipedia, Wiktionary).
"""Search the Zim encyclopedias (Wikipedia, Wiktionary, etc.).
Use list_zim_files first to find the zim_id.
Without zim_id this searches the Zims that have 'search by default' enabled; pass a zim_id
(from list_zim_files) to search one specific Zim. Read a result with get_zim_entry.
Args:
zim_id: ID of the Zim file to search.
query: Text to search for.
limit: Maximum number of results.
zim_id: ID of one Zim file to search; omit to search the default Zims.
limit: Maximum results (per Zim when searching several).
offset: Pagination offset.
"""
body = {"search_str": query, "zim_id": zim_id, "limit": limit, "offset": offset}
body = {"search_str": query, "limit": limit, "offset": offset}
if zim_id is not None:
body["zim_id"] = zim_id
try:
data = await api_post("/api/ai/zims/search", json=body)
except httpx.HTTPStatusError as e:
@ -313,90 +187,6 @@ async def search_zim(
return _render_zim_entries(data)
@mcp.tool(annotations={"readOnlyHint": True})
async def search_default_zims(
query: str,
limit: int = DEFAULT_LIMIT,
offset: int = 0,
) -> str:
"""Search only the Zim encyclopedias that have 'search by default' enabled.
This does NOT search all Zim files. To search a specific Zim, use list_zim_files
to find its ID, then use search_zim.
Args:
query: Text to search for.
limit: Maximum results per Zim file.
offset: Pagination offset.
"""
body = {"search_str": query, "limit": limit, "offset": offset}
return _render_zim_entries(await api_post("/api/ai/zims/search", json=body))
# ---------------------------------------------------------------------------
# Content retrieval tools
# ---------------------------------------------------------------------------
@mcp.tool(annotations={"readOnlyHint": True})
async def get_video(file_group_id: int) -> str:
"""Get detailed information about a specific video.
Args:
file_group_id: The FileGroup ID from search results.
"""
return _render_result(await api_get(f"/api/ai/files/{file_group_id}"))
@mcp.tool(annotations={"readOnlyHint": True})
async def get_video_captions(file_group_id: int) -> str:
"""Get the subtitle/caption text of a video. Useful for understanding video content without watching.
Args:
file_group_id: The FileGroup ID from search results.
"""
text = await _read_paged(f"/api/ai/files/{file_group_id}/content", params={"part": "text"})
return text or "No captions available for this video."
@mcp.tool(annotations={"readOnlyHint": True})
async def get_video_comments(file_group_id: int) -> str:
"""Get comments/discussion for a video.
Args:
file_group_id: The FileGroup ID from search results.
"""
text = await _read_paged(f"/api/ai/files/{file_group_id}/content", params={"part": "comments"})
return text or "No comments available for this video."
@mcp.tool(annotations={"readOnlyHint": True})
async def get_archive(file_group_id: int) -> str:
"""Get details about an archived web page, including its history of snapshots.
Args:
file_group_id: The FileGroup ID from search results.
"""
return _render_result(await api_get(f"/api/ai/files/{file_group_id}"))
@mcp.tool(annotations={"readOnlyHint": True})
async def get_archive_text(file_group_id: int) -> str:
"""Get the readable plain text content of an archived web page.
This returns the Readability-extracted text, which is clean and suitable for reading.
Args:
file_group_id: The FileGroup ID from search results.
"""
try:
text = await _read_paged(f"/api/ai/files/{file_group_id}/content", params={"part": "text"})
except httpx.HTTPStatusError as e:
if e.response.status_code == 404:
return f"Could not retrieve text content for archive {file_group_id}. Try get_archive for metadata."
raise
return text
@mcp.tool(annotations={"readOnlyHint": True})
async def get_zim_entry(zim_id: int, entry_path: str) -> str:
"""Read a specific article/entry from a Zim file (e.g. a Wikipedia article).
@ -423,7 +213,7 @@ async def get_zim_entry(zim_id: int, entry_path: str) -> str:
# ---------------------------------------------------------------------------
# Kind-generic tools (the compact set the built-in assistant uses)
# Library tools (the same compact set the built-in assistant uses)
# ---------------------------------------------------------------------------
@mcp.tool(annotations={"readOnlyHint": True})
@ -541,6 +331,27 @@ async def list_files(path: str = "", offset: int = 0) -> str:
return "\n".join(lines)
@mcp.tool(annotations={"readOnlyHint": True})
async def read_file(path: str) -> str:
"""Read a plain-text file from the media directory by its relative path (from list_files).
Args:
path: File path relative to the media directory, e.g. "notes/todo.txt". Text files only.
"""
try:
text = await _read_paged("/api/ai/files/read", params={"path": path})
except httpx.HTTPStatusError as e:
if e.response.status_code in (400, 404):
detail = ""
try:
detail = e.response.json().get("error") or ""
except Exception:
pass
return f"Could not read {path}: {detail or 'not a readable text file'}"
raise
return text or f"{path} is empty."
@mcp.tool(annotations={"readOnlyHint": True})
async def list_zim_files() -> str:
"""List all available Zim encyclopedias (Wikipedia, Wiktionary, etc.).
@ -698,8 +509,15 @@ async def get_statistics() -> str:
@mcp.tool(annotations={"readOnlyHint": True})
async def get_inventory() -> str:
"""List all inventories (emergency supplies, food storage, etc.)."""
async def get_inventory(inventory_slug: str | None = None) -> str:
"""List the inventories (emergency supplies, food storage, etc.), or read one in full.
Args:
inventory_slug: The slug of one inventory to read with every item; omit to list them all.
"""
if inventory_slug:
data = await api_get("/api/ai/inventories", params={"slug": inventory_slug})
return json.dumps(data, indent=2, default=str)
data = await api_get("/api/ai/inventories")
inventories = data.get("results") or []
if not inventories:
@ -713,17 +531,6 @@ async def get_inventory() -> str:
return "\n\n".join(lines)
@mcp.tool(annotations={"readOnlyHint": True})
async def get_inventory_items(inventory_slug: str) -> str:
"""Get all items in a specific inventory.
Args:
inventory_slug: The inventory slug (from get_inventory results).
"""
data = await api_get("/api/ai/inventories", params={"slug": inventory_slug})
return json.dumps(data, indent=2, default=str)
@mcp.tool(annotations={"readOnlyHint": True})
async def get_status() -> str:
"""Get WROLPi system status (version, services, CPU, disks, flags)."""