diff --git a/surfsense_mcp/src/surfsense_mcp/features/knowledge_base/__init__.py b/surfsense_mcp/src/surfsense_mcp/features/knowledge_base/__init__.py new file mode 100644 index 000000000..2ff3f571c --- /dev/null +++ b/surfsense_mcp/src/surfsense_mcp/features/knowledge_base/__init__.py @@ -0,0 +1,280 @@ +"""Knowledge-base tools: search the KB and manage its documents. + +Semantic search plus the document lifecycle — list, read, add text, upload a +file, update, and delete — over a workspace's knowledge base. Search and reads +default to the active workspace; document ids identify a single document across +the whole account, so id-addressed tools need no workspace. +""" + +from __future__ import annotations + +import mimetypes +from pathlib import Path + +from mcp.server.fastmcp import FastMCP +from mcp.types import ToolAnnotations + +from ...core.client import SurfSenseClient +from ...core.errors import ToolError +from ...core.rendering import ResponseFormat, clip, to_json +from ...core.workspace_context import WorkspaceContext +from .note_ingestion import build_note_document + +_READ = ToolAnnotations( + readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=False +) +_WRITE = ToolAnnotations( + readOnlyHint=False, destructiveHint=False, idempotentHint=False, openWorldHint=False +) +_DELETE = ToolAnnotations( + readOnlyHint=False, destructiveHint=True, idempotentHint=False, openWorldHint=False +) + + +def register( + mcp: FastMCP, client: SurfSenseClient, context: WorkspaceContext +) -> None: + """Register the knowledge-base tools on the server.""" + + @mcp.tool( + name="surfsense_search_knowledge_base", + annotations=_READ, + structured_output=False, + ) + async def search_knowledge_base( + query: str, + top_k: int = 5, + document_types: list[str] | None = None, + workspace: str | None = None, + response_format: ResponseFormat = "markdown", + ) -> str: + """Search the workspace's knowledge base by meaning and keyword. + + Use this to answer questions from stored content: it returns the most + relevant documents with the passages that matched, ranked by relevance. + top_k caps documents (1–20). Optionally restrict to document_types. + """ + resolved = await context.resolve(workspace) + hits = await client.request( + "POST", + "/documents/search-semantic", + json={ + "workspace_id": resolved.id, + "query": query, + "top_k": max(1, min(top_k, 20)), + "document_types": document_types, + }, + ) + items = (hits or {}).get("items", []) + if response_format == "json": + return to_json(items) + return _render_search(query, items) + + @mcp.tool( + name="surfsense_list_documents", annotations=_READ, structured_output=False + ) + async def list_documents( + document_types: list[str] | None = None, + folder_id: int | None = None, + page: int = 0, + page_size: int = 20, + workspace: str | None = None, + response_format: ResponseFormat = "markdown", + ) -> str: + """List documents in the workspace's knowledge base, newest first. + + Use this to browse or inventory what is stored. Optionally filter by + document_types or a folder_id. Paginated: returns page_size items and a + has_more flag; request the next page by increasing page. + """ + resolved = await context.resolve(workspace) + result = await client.request( + "GET", + "/documents", + params={ + "workspace_id": resolved.id, + "page": page, + "page_size": page_size, + "document_types": _join(document_types), + "folder_id": folder_id, + }, + ) + if response_format == "json": + return to_json(result) + return _render_document_list(result) + + @mcp.tool( + name="surfsense_get_document", annotations=_READ, structured_output=False + ) + async def get_document( + document_id: int, response_format: ResponseFormat = "markdown" + ) -> str: + """Read one document's full content and metadata by id. + + Use this after search or list to open a specific document. The id comes + from those tools' results. + """ + document = await client.request("GET", f"/documents/{document_id}") + if response_format == "json": + return clip(to_json(document)) + return _render_document(document) + + @mcp.tool( + name="surfsense_add_document", annotations=_WRITE, structured_output=False + ) + async def add_document( + title: str, + content: str, + source_url: str | None = None, + workspace: str | None = None, + ) -> str: + """Add a text or markdown note to the workspace's knowledge base. + + Use this to save notes, summaries, or snippets so they become + searchable. The content is indexed asynchronously, so it may take a + moment to appear in search. source_url optionally records where the text + came from. + """ + resolved = await context.resolve(workspace) + await client.request( + "POST", + "/documents", + json=build_note_document( + workspace_id=resolved.id, + title=title, + content=content, + source_url=source_url, + ), + ) + return ( + f"Queued '{title}' for indexing in '{resolved.name}'. " + "It will be searchable once processing completes." + ) + + @mcp.tool( + name="surfsense_upload_file", annotations=_WRITE, structured_output=False + ) + async def upload_file( + file_path: str, + use_vision_llm: bool = False, + workspace: str | None = None, + ) -> str: + """Upload a local file (PDF, doc, etc.) into the knowledge base. + + Use this to ingest a file from disk; it is parsed, chunked, and indexed + asynchronously. Set use_vision_llm to read scanned or image-heavy files + with a vision model (slower). + """ + resolved = await context.resolve(workspace) + payload = _read_upload(file_path) + result = await client.request( + "POST", + "/documents/fileupload", + data={ + "workspace_id": str(resolved.id), + "use_vision_llm": str(use_vision_llm).lower(), + "processing_mode": "basic", + }, + files=[("files", payload)], + ) + pending = (result or {}).get("pending_files", 0) + skipped = (result or {}).get("skipped_duplicates", 0) + note = " (already present, skipped)" if skipped and not pending else "" + return ( + f"Uploaded '{Path(file_path).name}' to '{resolved.name}'{note}. " + "It will be searchable once processing completes." + ) + + @mcp.tool( + name="surfsense_update_document", annotations=_WRITE, structured_output=False + ) + async def update_document(document_id: int, content: str) -> str: + """Replace a document's stored content by id. + + Use this to correct or rewrite a document's text. Note: this updates the + stored content; re-indexing of search chunks is not triggered by this + call. + """ + existing = await client.request("GET", f"/documents/{document_id}") + await client.request( + "PUT", + f"/documents/{document_id}", + json={ + "document_type": existing["document_type"], + "workspace_id": existing["workspace_id"], + "content": content, + }, + ) + return f"Updated document {document_id} ('{existing.get('title', '')}')." + + @mcp.tool( + name="surfsense_delete_document", annotations=_DELETE, structured_output=False + ) + async def delete_document(document_id: int) -> str: + """Delete a document from the knowledge base by id. + + Use this to permanently remove a document. Deletion runs in the + background; the document stops appearing in searches immediately. + """ + await client.request("DELETE", f"/documents/{document_id}") + return f"Deleted document {document_id}." + + +def _read_upload(file_path: str) -> tuple[str, bytes, str]: + path = Path(file_path).expanduser() + if not path.is_file(): + raise ToolError(f"No file at '{file_path}'.") + mime, _ = mimetypes.guess_type(path.name) + return (path.name, path.read_bytes(), mime or "application/octet-stream") + + +def _join(values: list[str] | None) -> str | None: + return ",".join(values) if values else None + + +def _render_search(query: str, items: list[dict]) -> str: + if not items: + return f'No matches for "{query}".' + lines = [f'# {len(items)} result(s) for "{query}"', ""] + for hit in items: + lines.append( + f"## {hit.get('title', 'Untitled')} " + f"(id {hit.get('document_id')}) — score {hit.get('score', 0):.3f}" + ) + for chunk in hit.get("chunks", []): + excerpt = clip(chunk.get("content", "").strip(), 500) + lines.append(f"> {excerpt}") + lines.append("") + return "\n".join(lines).strip() + + +def _render_document_list(result: dict | None) -> str: + items = (result or {}).get("items", []) + if not items: + return "No documents found." + lines = ["# Documents", ""] + for doc in items: + lines.append( + f"- **{doc.get('title', 'Untitled')}** (id {doc.get('id')}) · " + f"{doc.get('document_type')} · updated {doc.get('updated_at')}" + ) + total = (result or {}).get("total", len(items)) + page = (result or {}).get("page", 0) + has_more = (result or {}).get("has_more", False) + lines.append("") + lines.append( + f"_Page {page} · showing {len(items)} of {total}" + + (" · more available_" if has_more else "_") + ) + return "\n".join(lines) + + +def _render_document(document: dict) -> str: + content = clip(document.get("content", "") or "(empty)") + return ( + f"# {document.get('title', 'Untitled')} (id {document.get('id')})\n" + f"- type: {document.get('document_type')}\n" + f"- workspace: {document.get('workspace_id')}\n" + f"- updated: {document.get('updated_at')}\n\n" + f"{content}" + ) diff --git a/surfsense_mcp/src/surfsense_mcp/features/knowledge_base/note_ingestion.py b/surfsense_mcp/src/surfsense_mcp/features/knowledge_base/note_ingestion.py new file mode 100644 index 000000000..a02406c62 --- /dev/null +++ b/surfsense_mcp/src/surfsense_mcp/features/knowledge_base/note_ingestion.py @@ -0,0 +1,45 @@ +"""Translate a plain note into SurfSense's document-ingestion envelope. + +The REST API ingests free text through the browser-extension document shape +(title + page content + visit metadata); the backend then chunks and embeds it +like any saved page. Isolating that mapping lets the KB tool offer a simple +title+content surface without leaking the envelope's shape. +""" + +from __future__ import annotations + +import re +from datetime import datetime, timezone + + +def build_note_document( + *, workspace_id: int, title: str, content: str, source_url: str | None +) -> dict: + """Wrap a note in the EXTENSION document payload the create endpoint expects.""" + # ponytail: reuses the extension-ingestion path to add free text. Ceiling — + # visit metadata is synthesized; the "real page URL" is a stable synthetic + # link derived from the title. Upgrade path: a first-class note endpoint. + captured_at = datetime.now(timezone.utc).isoformat() + return { + "document_type": "EXTENSION", + "workspace_id": workspace_id, + "content": [ + { + "metadata": { + "BrowsingSessionId": "surfsense-mcp", + "VisitedWebPageURL": source_url or _synthetic_url(title), + "VisitedWebPageTitle": title, + "VisitedWebPageDateWithTimeInISOString": captured_at, + "VisitedWebPageReffererURL": "", + "VisitedWebPageVisitDurationInMilliseconds": "0", + }, + "pageContent": content, + } + ], + } + + +def _synthetic_url(title: str) -> str: + slug = re.sub(r"[^a-z0-9]+", "-", title.casefold()).strip("-") or "note" + stamp = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S") + return f"https://surfsense.local/mcp-note/{slug}-{stamp}"