From 2acc1426bf8294f0fd03c5a6e1e7f97b6aba9310 Mon Sep 17 00:00:00 2001 From: CREDO23 Date: Tue, 7 Jul 2026 17:22:37 +0200 Subject: [PATCH] refactor(mcp): split scraper and knowledge-base features into SRP modules --- .../surfsense_mcp/core/workspace_context.py | 43 +- .../surfsense_mcp/core/workspace_matching.py | 51 ++ .../features/knowledge_base/__init__.py | 371 +----------- .../features/knowledge_base/annotations.py | 34 ++ .../features/knowledge_base/document_tools.py | 185 ++++++ .../features/knowledge_base/search_tools.py | 188 ++++++ .../features/scrapers/__init__.py | 562 +----------------- .../features/scrapers/annotations.py | 13 + .../features/scrapers/platforms/__init__.py | 1 + .../scrapers/platforms/google_maps.py | 151 +++++ .../scrapers/platforms/google_search.py | 79 +++ .../features/scrapers/platforms/reddit.py | 98 +++ .../features/scrapers/platforms/web.py | 89 +++ .../features/scrapers/platforms/youtube.py | 131 ++++ .../features/scrapers/run_history.py | 112 ++++ 15 files changed, 1153 insertions(+), 955 deletions(-) create mode 100644 surfsense_mcp/src/surfsense_mcp/core/workspace_matching.py create mode 100644 surfsense_mcp/src/surfsense_mcp/features/knowledge_base/annotations.py create mode 100644 surfsense_mcp/src/surfsense_mcp/features/knowledge_base/document_tools.py create mode 100644 surfsense_mcp/src/surfsense_mcp/features/knowledge_base/search_tools.py create mode 100644 surfsense_mcp/src/surfsense_mcp/features/scrapers/annotations.py create mode 100644 surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/__init__.py create mode 100644 surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/google_maps.py create mode 100644 surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/google_search.py create mode 100644 surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/reddit.py create mode 100644 surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/web.py create mode 100644 surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/youtube.py create mode 100644 surfsense_mcp/src/surfsense_mcp/features/scrapers/run_history.py diff --git a/surfsense_mcp/src/surfsense_mcp/core/workspace_context.py b/surfsense_mcp/src/surfsense_mcp/core/workspace_context.py index cfd90c32b..0c5423e6b 100644 --- a/surfsense_mcp/src/surfsense_mcp/core/workspace_context.py +++ b/surfsense_mcp/src/surfsense_mcp/core/workspace_context.py @@ -16,6 +16,7 @@ from pydantic import Field from .auth.identity import current_identity from .client import SurfSenseClient from .errors import ToolError +from .workspace_matching import as_int, match_by_name, name_list # ponytail: one small entry per distinct caller (API token). Bounded so a flood # of keys can't grow memory without limit; an evicted caller just re-resolves @@ -99,43 +100,20 @@ class WorkspaceContext: ) raise ToolError( "No workspace selected. Choose one first with surfsense_select_workspace, " - f"or pass 'workspace'. Available: {_name_list(workspaces)}." + f"or pass 'workspace'. Available: {name_list(workspaces)}." ) async def _match(self, reference: str | int) -> Workspace: workspaces = await self.fetch_all() - as_id = _as_int(reference) + as_id = as_int(reference) if as_id is not None: found = next((w for w in workspaces if w.id == as_id), None) if found is None: raise ToolError( - f"No workspace with id {as_id}. Available: {_name_list(workspaces)}." + f"No workspace with id {as_id}. Available: {name_list(workspaces)}." ) return found - return _match_by_name(str(reference), workspaces) - - -def _match_by_name(reference: str, workspaces: list[Workspace]) -> Workspace: - """Match on name: exact, then case-insensitive, then unique substring.""" - needle = reference.strip() - exact = [w for w in workspaces if w.name == needle] - if exact: - return exact[0] - lowered = needle.casefold() - insensitive = [w for w in workspaces if w.name.casefold() == lowered] - if insensitive: - return insensitive[0] - partial = [w for w in workspaces if lowered in w.name.casefold()] - if len(partial) == 1: - return partial[0] - if len(partial) > 1: - raise ToolError( - f"'{reference}' matches several workspaces: {_name_list(partial)}. " - "Use a more specific name or the id." - ) - raise ToolError( - f"No workspace named '{reference}'. Available: {_name_list(workspaces)}." - ) + return match_by_name(str(reference), workspaces) def _to_workspace(row: dict) -> Workspace: @@ -146,14 +124,3 @@ def _to_workspace(row: dict) -> Workspace: is_owner=row.get("is_owner", False), member_count=row.get("member_count", 1), ) - - -def _as_int(reference: str | int) -> int | None: - if isinstance(reference, int): - return reference - text = reference.strip() - return int(text) if text.isdigit() else None - - -def _name_list(workspaces: list[Workspace]) -> str: - return ", ".join(f"{w.name} (id {w.id})" for w in workspaces) diff --git a/surfsense_mcp/src/surfsense_mcp/core/workspace_matching.py b/surfsense_mcp/src/surfsense_mcp/core/workspace_matching.py new file mode 100644 index 000000000..33bfc50a5 --- /dev/null +++ b/surfsense_mcp/src/surfsense_mcp/core/workspace_matching.py @@ -0,0 +1,51 @@ +"""Resolve a user-supplied workspace reference to a single workspace. + +Pure matching over an already-fetched list: name (exact, then case-insensitive, +then unique substring) or numeric id. Kept apart from WorkspaceContext so the +resolution rules can be read and tested without the network. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +from .errors import ToolError + +if TYPE_CHECKING: + from .workspace_context import Workspace + + +def match_by_name(reference: str, workspaces: list[Workspace]) -> Workspace: + """Match on name: exact, then case-insensitive, then unique substring.""" + needle = reference.strip() + exact = [w for w in workspaces if w.name == needle] + if exact: + return exact[0] + lowered = needle.casefold() + insensitive = [w for w in workspaces if w.name.casefold() == lowered] + if insensitive: + return insensitive[0] + partial = [w for w in workspaces if lowered in w.name.casefold()] + if len(partial) == 1: + return partial[0] + if len(partial) > 1: + raise ToolError( + f"'{reference}' matches several workspaces: {name_list(partial)}. " + "Use a more specific name or the id." + ) + raise ToolError( + f"No workspace named '{reference}'. Available: {name_list(workspaces)}." + ) + + +def as_int(reference: str | int) -> int | None: + """Return the reference as an id, or None when it isn't numeric.""" + if isinstance(reference, int): + return reference + text = reference.strip() + return int(text) if text.isdigit() else None + + +def name_list(workspaces: list[Workspace]) -> str: + """Render workspaces as a human-readable 'name (id N)' list.""" + return ", ".join(f"{w.name} (id {w.id})" for w in workspaces) diff --git a/surfsense_mcp/src/surfsense_mcp/features/knowledge_base/__init__.py b/surfsense_mcp/src/surfsense_mcp/features/knowledge_base/__init__.py index 7fb1802ae..1a971bfe4 100644 --- a/surfsense_mcp/src/surfsense_mcp/features/knowledge_base/__init__.py +++ b/surfsense_mcp/src/surfsense_mcp/features/knowledge_base/__init__.py @@ -1,379 +1,22 @@ """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. +file, update, and delete — over a workspace's knowledge base. Read tools live in +search_tools, mutations in document_tools. """ from __future__ import annotations -import mimetypes -from pathlib import Path -from typing import Annotated - from mcp.server.fastmcp import FastMCP -from mcp.types import ToolAnnotations -from pydantic import Field from ...core.client import SurfSenseClient -from ...core.errors import ToolError -from ...core.rendering import ResponseFormatParam, clip, to_json -from ...core.workspace_context import WorkspaceContext, WorkspaceParam -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 -) - -_DOCUMENT_ID = Annotated[ - int, - Field( - description="Document id from surfsense_search_knowledge_base or " - "surfsense_list_documents results." - ), -] - -_DOCUMENT_TYPES = Annotated[ - list[str] | None, - Field( - description="Restrict to these document types, e.g. " - "['FILE', 'CRAWLED_URL', 'YOUTUBE_VIDEO']. Omit for all types." - ), -] +from ...core.workspace_context import WorkspaceContext +from . import document_tools, search_tools def register( mcp: FastMCP, client: SurfSenseClient, context: WorkspaceContext ) -> None: - """Register the knowledge-base tools on the server.""" - - @mcp.tool( - name="surfsense_search_knowledge_base", - title="Search knowledge base", - annotations=_READ, - structured_output=False, - ) - async def search_knowledge_base( - query: Annotated[ - str, - Field( - min_length=1, - description="Natural-language search, e.g. " - "'notebooklm user complaints'.", - ), - ], - top_k: Annotated[ - int, Field(ge=1, le=20, description="Maximum documents to return.") - ] = 5, - document_types: _DOCUMENT_TYPES = None, - workspace: WorkspaceParam = None, - response_format: ResponseFormatParam = "markdown", - ) -> str: - """Search the workspace's knowledge base by meaning and keywords. - - Use this FIRST when a question might be answered by content already - stored in SurfSense — notes, uploaded files, saved pages, past - research. Do NOT use it to fetch new data from the web; use the - scraper tools for that. Returns the most relevant documents with the - passages that matched, ranked by relevance score. - Example: query='pricing feedback', top_k=5. - """ - 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", - title="List documents", - annotations=_READ, - structured_output=False, - ) - async def list_documents( - document_types: _DOCUMENT_TYPES = None, - folder_id: Annotated[ - int | None, - Field(description="Only documents in this folder. Omit for all."), - ] = None, - page: Annotated[ - int, Field(ge=0, description="Zero-based page number.") - ] = 0, - page_size: Annotated[ - int, Field(ge=1, description="Documents per page.") - ] = 20, - workspace: WorkspaceParam = None, - response_format: ResponseFormatParam = "markdown", - ) -> str: - """List documents in the workspace's knowledge base, newest first. - - Use this to browse or inventory what is stored; to find documents - about a topic, prefer surfsense_search_knowledge_base. Returns each - document's title, id, type, and update time, plus a has_more flag — - request the next page by increasing page. - Example: document_types=['FILE'], page=0, page_size=20. - """ - 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", - title="Read one document", - annotations=_READ, - structured_output=False, - ) - async def get_document( - document_id: _DOCUMENT_ID, - response_format: ResponseFormatParam = "markdown", - ) -> str: - """Read one document's full content and metadata by id. - - Use this after surfsense_search_knowledge_base or - surfsense_list_documents to open a specific document — search results - only include the matching passages, this returns the whole text. - """ - 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", - title="Add a note", - annotations=_WRITE, - structured_output=False, - ) - async def add_document( - title: Annotated[ - str, - Field(min_length=1, description="Short descriptive title for the note."), - ], - content: Annotated[ - str, - Field( - min_length=1, - description="The note's body; plain text or markdown.", - ), - ], - source_url: Annotated[ - str | None, - Field(description="Where the text came from, if anywhere."), - ] = None, - workspace: WorkspaceParam = None, - ) -> str: - """Save a text or markdown note into the workspace's knowledge base. - - Use this to store notes, summaries, or findings so they become - searchable later — e.g. after finishing a piece of research. For files - on disk use surfsense_upload_file instead. Indexing is asynchronous, - so the note may take a moment to appear in search. - Example: title='NotebookLM subreddits', content='- r/notebooklm ...'. - """ - 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", - title="Upload a file", - annotations=_WRITE, - structured_output=False, - ) - async def upload_file( - file_path: Annotated[ - str, - Field( - description="Path to a local file, e.g. " - "'C:/Users/me/report.pdf' or '~/notes/summary.md'." - ), - ], - use_vision_llm: Annotated[ - bool, - Field( - description="True reads scanned or image-heavy files with a " - "vision model (slower)." - ), - ] = False, - workspace: WorkspaceParam = None, - ) -> str: - """Upload a local file (PDF, docx, markdown, etc.) into the knowledge base. - - Use this to ingest a file from disk so its content becomes searchable; - for text you already have in hand use surfsense_add_document instead. - The file is parsed, chunked, and indexed asynchronously. Duplicate - files are detected and skipped. - Example: file_path='C:/Users/me/report.pdf'. - """ - 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", - title="Replace a document's content", - annotations=_WRITE, - structured_output=False, - ) - async def update_document( - document_id: _DOCUMENT_ID, - content: Annotated[ - str, - Field( - min_length=1, - description="New full text; replaces the existing content " - "entirely.", - ), - ], - ) -> str: - """Replace a document's stored content by id. - - Use this to correct or rewrite a document's text. The new content - REPLACES the old entirely — to append, read the document first with - surfsense_get_document and resend the combined text. Search chunks are - not re-indexed 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", - title="Delete a document", - annotations=_DELETE, - structured_output=False, - ) - async def delete_document(document_id: _DOCUMENT_ID) -> str: - """Permanently delete a document from the knowledge base by id. - - Use this only when the user explicitly asks to remove a document — - deletion cannot be undone. 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}" - ) + """Register every knowledge-base tool on the server.""" + search_tools.register(mcp, client, context) + document_tools.register(mcp, client, context) diff --git a/surfsense_mcp/src/surfsense_mcp/features/knowledge_base/annotations.py b/surfsense_mcp/src/surfsense_mcp/features/knowledge_base/annotations.py new file mode 100644 index 000000000..16322506c --- /dev/null +++ b/surfsense_mcp/src/surfsense_mcp/features/knowledge_base/annotations.py @@ -0,0 +1,34 @@ +"""Tool-call policy hints and shared parameter types for knowledge-base tools.""" + +from __future__ import annotations + +from typing import Annotated + +from mcp.types import ToolAnnotations +from pydantic import Field + +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 +) + +DocumentId = Annotated[ + int, + Field( + description="Document id from surfsense_search_knowledge_base or " + "surfsense_list_documents results." + ), +] + +DocumentTypes = Annotated[ + list[str] | None, + Field( + description="Restrict to these document types, e.g. " + "['FILE', 'CRAWLED_URL', 'YOUTUBE_VIDEO']. Omit for all types." + ), +] diff --git a/surfsense_mcp/src/surfsense_mcp/features/knowledge_base/document_tools.py b/surfsense_mcp/src/surfsense_mcp/features/knowledge_base/document_tools.py new file mode 100644 index 000000000..497a2526c --- /dev/null +++ b/surfsense_mcp/src/surfsense_mcp/features/knowledge_base/document_tools.py @@ -0,0 +1,185 @@ +"""Knowledge-base write tools: add a note, upload a file, update, and delete. + +Add and upload target the active workspace; update and delete address a document +by its account-unique id, so they need no workspace. +""" + +from __future__ import annotations + +import mimetypes +from pathlib import Path +from typing import Annotated + +from mcp.server.fastmcp import FastMCP +from pydantic import Field + +from ...core.client import SurfSenseClient +from ...core.errors import ToolError +from ...core.workspace_context import WorkspaceContext, WorkspaceParam +from .annotations import DELETE, WRITE, DocumentId +from .note_ingestion import build_note_document + + +def register( + mcp: FastMCP, client: SurfSenseClient, context: WorkspaceContext +) -> None: + """Register the knowledge-base write and delete tools.""" + + @mcp.tool( + name="surfsense_add_document", + title="Add a note", + annotations=WRITE, + structured_output=False, + ) + async def add_document( + title: Annotated[ + str, + Field(min_length=1, description="Short descriptive title for the note."), + ], + content: Annotated[ + str, + Field( + min_length=1, + description="The note's body; plain text or markdown.", + ), + ], + source_url: Annotated[ + str | None, + Field(description="Where the text came from, if anywhere."), + ] = None, + workspace: WorkspaceParam = None, + ) -> str: + """Save a text or markdown note into the workspace's knowledge base. + + Use this to store notes, summaries, or findings so they become + searchable later — e.g. after finishing a piece of research. For files + on disk use surfsense_upload_file instead. Indexing is asynchronous, + so the note may take a moment to appear in search. + Example: title='NotebookLM subreddits', content='- r/notebooklm ...'. + """ + 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", + title="Upload a file", + annotations=WRITE, + structured_output=False, + ) + async def upload_file( + file_path: Annotated[ + str, + Field( + description="Path to a local file, e.g. " + "'C:/Users/me/report.pdf' or '~/notes/summary.md'." + ), + ], + use_vision_llm: Annotated[ + bool, + Field( + description="True reads scanned or image-heavy files with a " + "vision model (slower)." + ), + ] = False, + workspace: WorkspaceParam = None, + ) -> str: + """Upload a local file (PDF, docx, markdown, etc.) into the knowledge base. + + Use this to ingest a file from disk so its content becomes searchable; + for text you already have in hand use surfsense_add_document instead. + The file is parsed, chunked, and indexed asynchronously. Duplicate + files are detected and skipped. + Example: file_path='C:/Users/me/report.pdf'. + """ + 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", + title="Replace a document's content", + annotations=WRITE, + structured_output=False, + ) + async def update_document( + document_id: DocumentId, + content: Annotated[ + str, + Field( + min_length=1, + description="New full text; replaces the existing content " + "entirely.", + ), + ], + ) -> str: + """Replace a document's stored content by id. + + Use this to correct or rewrite a document's text. The new content + REPLACES the old entirely — to append, read the document first with + surfsense_get_document and resend the combined text. Search chunks are + not re-indexed 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", + title="Delete a document", + annotations=DELETE, + structured_output=False, + ) + async def delete_document(document_id: DocumentId) -> str: + """Permanently delete a document from the knowledge base by id. + + Use this only when the user explicitly asks to remove a document — + deletion cannot be undone. 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") diff --git a/surfsense_mcp/src/surfsense_mcp/features/knowledge_base/search_tools.py b/surfsense_mcp/src/surfsense_mcp/features/knowledge_base/search_tools.py new file mode 100644 index 000000000..a9e60810d --- /dev/null +++ b/surfsense_mcp/src/surfsense_mcp/features/knowledge_base/search_tools.py @@ -0,0 +1,188 @@ +"""Knowledge-base read tools: semantic search, list, and read one document. + +Search and list default to the active workspace; a document read is addressed by +id, which is unique across the account, so it needs no workspace. +""" + +from __future__ import annotations + +from typing import Annotated + +from mcp.server.fastmcp import FastMCP +from pydantic import Field + +from ...core.client import SurfSenseClient +from ...core.rendering import ResponseFormatParam, clip, to_json +from ...core.workspace_context import WorkspaceContext, WorkspaceParam +from .annotations import READ, DocumentId, DocumentTypes + + +def register( + mcp: FastMCP, client: SurfSenseClient, context: WorkspaceContext +) -> None: + """Register the knowledge-base read tools.""" + + @mcp.tool( + name="surfsense_search_knowledge_base", + title="Search knowledge base", + annotations=READ, + structured_output=False, + ) + async def search_knowledge_base( + query: Annotated[ + str, + Field( + min_length=1, + description="Natural-language search, e.g. " + "'notebooklm user complaints'.", + ), + ], + top_k: Annotated[ + int, Field(ge=1, le=20, description="Maximum documents to return.") + ] = 5, + document_types: DocumentTypes = None, + workspace: WorkspaceParam = None, + response_format: ResponseFormatParam = "markdown", + ) -> str: + """Search the workspace's knowledge base by meaning and keywords. + + Use this FIRST when a question might be answered by content already + stored in SurfSense — notes, uploaded files, saved pages, past + research. Do NOT use it to fetch new data from the web; use the + scraper tools for that. Returns the most relevant documents with the + passages that matched, ranked by relevance score. + Example: query='pricing feedback', top_k=5. + """ + 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", + title="List documents", + annotations=READ, + structured_output=False, + ) + async def list_documents( + document_types: DocumentTypes = None, + folder_id: Annotated[ + int | None, + Field(description="Only documents in this folder. Omit for all."), + ] = None, + page: Annotated[ + int, Field(ge=0, description="Zero-based page number.") + ] = 0, + page_size: Annotated[ + int, Field(ge=1, description="Documents per page.") + ] = 20, + workspace: WorkspaceParam = None, + response_format: ResponseFormatParam = "markdown", + ) -> str: + """List documents in the workspace's knowledge base, newest first. + + Use this to browse or inventory what is stored; to find documents + about a topic, prefer surfsense_search_knowledge_base. Returns each + document's title, id, type, and update time, plus a has_more flag — + request the next page by increasing page. + Example: document_types=['FILE'], page=0, page_size=20. + """ + 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", + title="Read one document", + annotations=READ, + structured_output=False, + ) + async def get_document( + document_id: DocumentId, + response_format: ResponseFormatParam = "markdown", + ) -> str: + """Read one document's full content and metadata by id. + + Use this after surfsense_search_knowledge_base or + surfsense_list_documents to open a specific document — search results + only include the matching passages, this returns the whole text. + """ + document = await client.request("GET", f"/documents/{document_id}") + if response_format == "json": + return clip(to_json(document)) + return _render_document(document) + + +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/scrapers/__init__.py b/surfsense_mcp/src/surfsense_mcp/features/scrapers/__init__.py index aa8265148..dfa2f3ab2 100644 --- a/surfsense_mcp/src/surfsense_mcp/features/scrapers/__init__.py +++ b/surfsense_mcp/src/surfsense_mcp/features/scrapers/__init__.py @@ -1,570 +1,26 @@ """Scraper tools: one MCP surface per SurfSense platform capability. Web crawl, Google Search, Reddit, YouTube, and Google Maps each get a tool that -maps a natural-language request to the workspace's scraper door. Two more tools +maps a natural-language request to the workspace's scraper. Two run-history tools list and fetch past runs, so a large result truncated inline can be retrieved in -full later. +full later. Each platform lives in its own module under platforms/. """ from __future__ import annotations -from typing import Annotated, Literal - from mcp.server.fastmcp import FastMCP -from mcp.types import ToolAnnotations -from pydantic import Field from ...core.client import SurfSenseClient -from ...core.rendering import ResponseFormatParam, clip, to_json -from ...core.workspace_context import WorkspaceContext, WorkspaceParam -from .capability import run_scraper +from ...core.workspace_context import WorkspaceContext +from . import run_history +from .platforms import google_maps, google_search, reddit, web, youtube -# Scrapers reach the open web and record a billable run; they are neither -# read-only nor idempotent, but they do not mutate the knowledge base. -_SCRAPE = ToolAnnotations( - readOnlyHint=False, destructiveHint=False, idempotentHint=False, openWorldHint=True -) -_READ_RUNS = ToolAnnotations( - readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=False -) - -RedditSort = Literal["relevance", "hot", "top", "new", "rising", "comments"] -RedditTime = Literal["hour", "day", "week", "month", "year", "all"] -CommentSort = Literal["TOP_COMMENTS", "NEWEST_FIRST"] -ReviewSort = Literal["newest", "mostRelevant", "highestRanking", "lowestRanking"] +_REGISTRARS = (web, google_search, reddit, youtube, google_maps, run_history) def register( mcp: FastMCP, client: SurfSenseClient, context: WorkspaceContext ) -> None: - """Register the scraper and run-history tools on the server.""" - - @mcp.tool( - name="surfsense_web_crawl", - title="Crawl web pages", - annotations=_SCRAPE, - structured_output=False, - ) - async def web_crawl( - start_urls: Annotated[ - list[str], - Field( - min_length=1, - description="Full URLs to fetch, e.g. " - "['https://example.com/blog/post'].", - ), - ], - max_crawl_depth: Annotated[ - int, - Field( - ge=0, - description="Link-hops to follow from start_urls within the " - "same site. 0 fetches only start_urls.", - ), - ] = 0, - max_crawl_pages: Annotated[ - int, Field(ge=1, description="Stop after this many pages in total.") - ] = 10, - max_length: Annotated[ - int, Field(ge=1, description="Max characters kept per page.") - ] = 50_000, - include_url_patterns: Annotated[ - list[str] | None, - Field( - description="Regexes; only discovered links matching one are " - "followed, e.g. ['/docs/.*']." - ), - ] = None, - exclude_url_patterns: Annotated[ - list[str] | None, - Field(description="Regexes; discovered links matching one are skipped."), - ] = None, - workspace: WorkspaceParam = None, - response_format: ResponseFormatParam = "markdown", - ) -> str: - """Fetch specific web pages and return their cleaned content as markdown. - - Use this to read a page the user names, or to spider a site from a - starting URL. Do NOT use it to find pages on a topic — use - surfsense_google_search for discovery. Returns one item per crawled - page: url, title, and the page text as markdown. - Example: start_urls=['https://blog.example.com'], max_crawl_depth=1, - include_url_patterns=['/2026/']. - """ - return await run_scraper( - client, - context, - platform="web", - verb="crawl", - payload={ - "startUrls": start_urls, - "maxCrawlDepth": max_crawl_depth, - "maxCrawlPages": max_crawl_pages, - "maxLength": max_length, - "includeUrlPatterns": include_url_patterns, - "excludeUrlPatterns": exclude_url_patterns, - }, - workspace=workspace, - response_format=response_format, - ) - - @mcp.tool( - name="surfsense_google_search", - title="Scrape Google Search", - annotations=_SCRAPE, - structured_output=False, - ) - async def google_search( - queries: Annotated[ - list[str], - Field( - min_length=1, - description="Search terms or full Google Search URLs, e.g. " - "['best rss readers 2026'].", - ), - ], - max_pages_per_query: Annotated[ - int, Field(ge=1, description="Result pages to fetch per query.") - ] = 1, - country_code: Annotated[ - str | None, - Field(description="Two-letter country to search from, e.g. 'us'."), - ] = None, - language_code: Annotated[ - str, Field(description="Results language, e.g. 'en'. Empty for default.") - ] = "", - site: Annotated[ - str | None, - Field( - description="Restrict results to one domain, e.g. 'example.com'." - ), - ] = None, - workspace: WorkspaceParam = None, - response_format: ResponseFormatParam = "markdown", - ) -> str: - """Scrape Google Search result pages for one or more queries. - - Use this to discover pages on the open web by topic; follow up with - surfsense_web_crawl to read a result in full. Do NOT use it for - Reddit, YouTube, or Google Maps research — the dedicated tools return - richer data. Returns each query's parsed results: title, url, and - snippet per organic result. - Example: queries=['notebooklm review'], site='news.ycombinator.com'. - """ - return await run_scraper( - client, - context, - platform="google_search", - verb="scrape", - payload={ - "queries": queries, - "max_pages_per_query": max_pages_per_query, - "country_code": country_code, - "language_code": language_code, - "site": site, - }, - workspace=workspace, - response_format=response_format, - ) - - @mcp.tool( - name="surfsense_reddit_scrape", - title="Search or scrape Reddit", - annotations=_SCRAPE, - structured_output=False, - ) - async def reddit_scrape( - urls: Annotated[ - list[str] | None, - Field( - description="Reddit URLs: a post, a subreddit like " - "'https://reddit.com/r/LocalLLaMA', a user page, or a search " - "URL. Provide urls OR search_queries." - ), - ] = None, - search_queries: Annotated[ - list[str] | None, - Field( - description="Terms to search Reddit for, e.g. " - "['NotebookLM alternatives']. Provide search_queries OR urls." - ), - ] = None, - community: Annotated[ - str | None, - Field( - description="Restrict a search to one subreddit, name without " - "'r/', e.g. 'ArtificialInteligence'." - ), - ] = None, - sort: Annotated[RedditSort, Field(description="Post ordering.")] = "new", - time_filter: Annotated[ - RedditTime | None, - Field(description="Time window; only valid with sort='top'."), - ] = None, - max_items: Annotated[ - int, Field(ge=1, description="Maximum posts to return.") - ] = 10, - skip_comments: Annotated[ - bool, - Field( - description="True fetches posts only (faster); False also " - "fetches each post's comment thread." - ), - ] = False, - workspace: WorkspaceParam = None, - response_format: ResponseFormatParam = "markdown", - ) -> str: - """Search or scrape Reddit: posts, comments, subreddits, and users. - - Use this for ANY Reddit research — finding relevant subreddits or - communities for a topic, top posts, or discussions — instead of a - generic web search. Returns posts (title, text, score, subreddit, url) - with comment threads unless skip_comments is set. Every post carries - its subreddit, so to find communities for a topic, search posts and - aggregate their subreddits. - Example: search_queries=['NotebookLM'], sort='top', time_filter='month'. - """ - return await run_scraper( - client, - context, - platform="reddit", - verb="scrape", - payload={ - "urls": urls, - "search_queries": search_queries, - "community": community, - "sort": sort, - "time_filter": time_filter, - "max_items": max_items, - "skip_comments": skip_comments, - }, - workspace=workspace, - response_format=response_format, - ) - - @mcp.tool( - name="surfsense_youtube_scrape", - title="Search or scrape YouTube", - annotations=_SCRAPE, - structured_output=False, - ) - async def youtube_scrape( - urls: Annotated[ - list[str] | None, - Field( - description="YouTube URLs: video, channel, playlist, shorts, " - "or hashtag pages. Provide urls OR search_queries." - ), - ] = None, - search_queries: Annotated[ - list[str] | None, - Field( - description="Terms to search YouTube for, e.g. " - "['NotebookLM tutorial']. Provide search_queries OR urls." - ), - ] = None, - max_results: Annotated[ - int, Field(ge=1, description="Maximum videos to return.") - ] = 10, - download_subtitles: Annotated[ - bool, - Field(description="True also fetches each video's transcript."), - ] = False, - subtitles_language: Annotated[ - str, Field(description="Transcript language code, e.g. 'en'.") - ] = "en", - workspace: WorkspaceParam = None, - response_format: ResponseFormatParam = "markdown", - ) -> str: - """Search or scrape YouTube videos, optionally with transcripts. - - Use this for YouTube research: finding videos on a topic, or reading a - video's details or transcript. For a video's comment section use - surfsense_youtube_comments instead. Returns per-video metadata (title, - channel, views, description, url) and, if requested, the transcript. - Example: search_queries=['NotebookLM tutorial'], download_subtitles=True. - """ - return await run_scraper( - client, - context, - platform="youtube", - verb="scrape", - payload={ - "urls": urls, - "search_queries": search_queries, - "max_results": max_results, - "download_subtitles": download_subtitles, - "subtitles_language": subtitles_language, - }, - workspace=workspace, - response_format=response_format, - ) - - @mcp.tool( - name="surfsense_youtube_comments", - title="Fetch YouTube comments", - annotations=_SCRAPE, - structured_output=False, - ) - async def youtube_comments( - urls: Annotated[ - list[str], - Field( - min_length=1, - description="YouTube video URLs, e.g. " - "['https://www.youtube.com/watch?v=abc123'].", - ), - ], - max_comments: Annotated[ - int, - Field( - ge=1, - description="Maximum comments per video, counting top-level " - "comments and replies together.", - ), - ] = 20, - sort_by: Annotated[ - CommentSort, Field(description="Comment ordering.") - ] = "NEWEST_FIRST", - workspace: WorkspaceParam = None, - response_format: ResponseFormatParam = "markdown", - ) -> str: - """Fetch the comments (and replies) on one or more YouTube videos. - - Use this when the user wants a video's discussion or audience reaction - rather than the video itself; get video URLs from - surfsense_youtube_scrape if you only have a topic. Returns comment - text, author, likes, and replies. - Example: urls=['https://www.youtube.com/watch?v=abc123'], max_comments=50. - """ - return await run_scraper( - client, - context, - platform="youtube", - verb="comments", - payload={ - "urls": urls, - "max_comments": max_comments, - "sort_by": sort_by, - }, - workspace=workspace, - response_format=response_format, - ) - - @mcp.tool( - name="surfsense_google_maps_scrape", - title="Find places on Google Maps", - annotations=_SCRAPE, - structured_output=False, - ) - async def google_maps_scrape( - search_queries: Annotated[ - list[str] | None, - Field( - description="Place searches, e.g. ['coffee shops']. Provide " - "search_queries OR urls OR place_ids." - ), - ] = None, - urls: Annotated[ - list[str] | None, - Field(description="Google Maps URLs of specific places."), - ] = None, - place_ids: Annotated[ - list[str] | None, - Field(description="Google place ids, e.g. ['ChIJj61dQgK6j4AR...']."), - ] = None, - location: Annotated[ - str | None, - Field( - description="Geographic scope for a search, e.g. " - "'Seattle, USA'." - ), - ] = None, - max_places: Annotated[ - int, Field(ge=1, description="Maximum places to return.") - ] = 10, - include_details: Annotated[ - bool, - Field( - description="True adds opening hours and extra contact info " - "(slower)." - ), - ] = False, - workspace: WorkspaceParam = None, - response_format: ResponseFormatParam = "markdown", - ) -> str: - """Find places on Google Maps by search, URL, or place id. - - Use this for local-business and location research: names, addresses, - ratings, categories, coordinates, place ids. For a place's customer - reviews use surfsense_google_maps_reviews instead. - Example: search_queries=['ramen'], location='Osaka, Japan', max_places=5. - """ - return await run_scraper( - client, - context, - platform="google_maps", - verb="scrape", - payload={ - "search_queries": search_queries, - "urls": urls, - "place_ids": place_ids, - "location": location, - "max_places": max_places, - "include_details": include_details, - }, - workspace=workspace, - response_format=response_format, - ) - - @mcp.tool( - name="surfsense_google_maps_reviews", - title="Fetch Google Maps reviews", - annotations=_SCRAPE, - structured_output=False, - ) - async def google_maps_reviews( - urls: Annotated[ - list[str] | None, - Field( - description="Google Maps URLs of places. Provide urls OR " - "place_ids." - ), - ] = None, - place_ids: Annotated[ - list[str] | None, - Field( - description="Google place ids from surfsense_google_maps_scrape." - ), - ] = None, - max_reviews: Annotated[ - int, Field(ge=1, description="Maximum reviews per place.") - ] = 20, - sort_by: Annotated[ - ReviewSort, Field(description="Review ordering.") - ] = "newest", - language: Annotated[ - str, Field(description="Reviews language code, e.g. 'en'.") - ] = "en", - start_date: Annotated[ - str | None, - Field( - description="ISO date like '2026-01-01'; keeps only reviews on " - "or after that day." - ), - ] = None, - workspace: WorkspaceParam = None, - response_format: ResponseFormatParam = "markdown", - ) -> str: - """Fetch customer reviews for Google Maps places by URL or place id. - - Use this to read feedback on specific places; get urls or place_ids - from surfsense_google_maps_scrape first if you only have a name. - Returns review text, rating, author, and date per review. - Example: place_ids=['ChIJj61dQgK6j4AR...'], sort_by='newest'. - """ - return await run_scraper( - client, - context, - platform="google_maps", - verb="reviews", - payload={ - "urls": urls, - "place_ids": place_ids, - "max_reviews": max_reviews, - "sort_by": sort_by, - "language": language, - "start_date": start_date, - }, - workspace=workspace, - response_format=response_format, - ) - - @mcp.tool( - name="surfsense_list_scraper_runs", - title="List past scraper runs", - annotations=_READ_RUNS, - structured_output=False, - ) - async def list_scraper_runs( - limit: Annotated[ - int, Field(ge=1, description="Maximum runs to list.") - ] = 20, - capability: Annotated[ - str | None, - Field( - description="Filter by capability slug, e.g. 'web.crawl' or " - "'reddit.scrape'." - ), - ] = None, - status: Annotated[ - str | None, - Field(description="Filter by run status: 'success' or 'error'."), - ] = None, - workspace: WorkspaceParam = None, - response_format: ResponseFormatParam = "markdown", - ) -> str: - """List recent scraper runs in the workspace, newest first. - - Use this to find the run_id of an earlier scrape — for example when an - inline result was truncated — then fetch it in full with - surfsense_get_scraper_run. Returns each run's id, capability, status, - item count, and creation time. - Example: capability='reddit.scrape', status='success'. - """ - resolved = await context.resolve(workspace) - runs = await client.request( - "GET", - f"/workspaces/{resolved.id}/scrapers/runs", - params={ - "limit": limit, - "capability": capability, - "status": status, - }, - ) - if response_format == "json": - return to_json(runs) - return _render_runs(runs) - - @mcp.tool( - name="surfsense_get_scraper_run", - title="Fetch one scraper run in full", - annotations=_READ_RUNS, - structured_output=False, - ) - async def get_scraper_run( - run_id: Annotated[ - str, - Field( - description="Run id from surfsense_list_scraper_runs or a " - "prior scrape's output." - ), - ], - workspace: WorkspaceParam = None, - response_format: ResponseFormatParam = "markdown", - ) -> str: - """Fetch a single scraper run in full, including its stored output. - - Use this to retrieve the complete, untruncated result of an earlier - scrape. Do NOT re-run a scraper just to recover a truncated result — - fetch the stored run instead. - """ - resolved = await context.resolve(workspace) - run = await client.request( - "GET", f"/workspaces/{resolved.id}/scrapers/runs/{run_id}" - ) - if response_format == "json": - return clip(to_json(run)) - return f"# Run {run.get('id', run_id)}\n\n```json\n{clip(to_json(run))}\n```" - - -def _render_runs(runs: list[dict] | None) -> str: - if not runs: - return "No scraper runs found." - lines = ["# Scraper runs", ""] - for run in runs: - lines.append( - f"- **{run.get('id')}** — {run.get('capability')} · {run.get('status')} · " - f"{run.get('item_count', 0)} item(s) · {run.get('created_at')}" - ) - return "\n".join(lines) + """Register every scraper and run-history tool on the server.""" + for module in _REGISTRARS: + module.register(mcp, client, context) diff --git a/surfsense_mcp/src/surfsense_mcp/features/scrapers/annotations.py b/surfsense_mcp/src/surfsense_mcp/features/scrapers/annotations.py new file mode 100644 index 000000000..cb872cc90 --- /dev/null +++ b/surfsense_mcp/src/surfsense_mcp/features/scrapers/annotations.py @@ -0,0 +1,13 @@ +"""Tool-call policy hints shared across scraper tools.""" + +from __future__ import annotations + +from mcp.types import ToolAnnotations + +SCRAPE = ToolAnnotations( + readOnlyHint=False, destructiveHint=False, idempotentHint=False, openWorldHint=True +) + +READ_RUNS = ToolAnnotations( + readOnlyHint=True, destructiveHint=False, idempotentHint=True, openWorldHint=False +) diff --git a/surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/__init__.py b/surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/__init__.py new file mode 100644 index 000000000..f61704380 --- /dev/null +++ b/surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/__init__.py @@ -0,0 +1 @@ +"""One module per scraper platform; each exposes register(mcp, client, context).""" diff --git a/surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/google_maps.py b/surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/google_maps.py new file mode 100644 index 000000000..e1613ca4e --- /dev/null +++ b/surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/google_maps.py @@ -0,0 +1,151 @@ +"""Google Maps scraper tools: places and reviews.""" + +from __future__ import annotations + +from typing import Annotated, Literal + +from mcp.server.fastmcp import FastMCP +from pydantic import Field + +from ....core.client import SurfSenseClient +from ....core.rendering import ResponseFormatParam +from ....core.workspace_context import WorkspaceContext, WorkspaceParam +from ..annotations import SCRAPE +from ..capability import run_scraper + +ReviewSort = Literal["newest", "mostRelevant", "highestRanking", "lowestRanking"] + + +def register( + mcp: FastMCP, client: SurfSenseClient, context: WorkspaceContext +) -> None: + """Register the Google Maps place and review tools.""" + + @mcp.tool( + name="surfsense_google_maps_scrape", + title="Find places on Google Maps", + annotations=SCRAPE, + structured_output=False, + ) + async def google_maps_scrape( + search_queries: Annotated[ + list[str] | None, + Field( + description="Place searches, e.g. ['coffee shops']. Provide " + "search_queries OR urls OR place_ids." + ), + ] = None, + urls: Annotated[ + list[str] | None, + Field(description="Google Maps URLs of specific places."), + ] = None, + place_ids: Annotated[ + list[str] | None, + Field(description="Google place ids, e.g. ['ChIJj61dQgK6j4AR...']."), + ] = None, + location: Annotated[ + str | None, + Field( + description="Geographic scope for a search, e.g. " + "'Seattle, USA'." + ), + ] = None, + max_places: Annotated[ + int, Field(ge=1, description="Maximum places to return.") + ] = 10, + include_details: Annotated[ + bool, + Field( + description="True adds opening hours and extra contact info " + "(slower)." + ), + ] = False, + workspace: WorkspaceParam = None, + response_format: ResponseFormatParam = "markdown", + ) -> str: + """Find places on Google Maps by search, URL, or place id. + + Use this for local-business and location research: names, addresses, + ratings, categories, coordinates, place ids. For a place's customer + reviews use surfsense_google_maps_reviews instead. + Example: search_queries=['ramen'], location='Osaka, Japan', max_places=5. + """ + return await run_scraper( + client, + context, + platform="google_maps", + verb="scrape", + payload={ + "search_queries": search_queries, + "urls": urls, + "place_ids": place_ids, + "location": location, + "max_places": max_places, + "include_details": include_details, + }, + workspace=workspace, + response_format=response_format, + ) + + @mcp.tool( + name="surfsense_google_maps_reviews", + title="Fetch Google Maps reviews", + annotations=SCRAPE, + structured_output=False, + ) + async def google_maps_reviews( + urls: Annotated[ + list[str] | None, + Field( + description="Google Maps URLs of places. Provide urls OR " + "place_ids." + ), + ] = None, + place_ids: Annotated[ + list[str] | None, + Field( + description="Google place ids from surfsense_google_maps_scrape." + ), + ] = None, + max_reviews: Annotated[ + int, Field(ge=1, description="Maximum reviews per place.") + ] = 20, + sort_by: Annotated[ + ReviewSort, Field(description="Review ordering.") + ] = "newest", + language: Annotated[ + str, Field(description="Reviews language code, e.g. 'en'.") + ] = "en", + start_date: Annotated[ + str | None, + Field( + description="ISO date like '2026-01-01'; keeps only reviews on " + "or after that day." + ), + ] = None, + workspace: WorkspaceParam = None, + response_format: ResponseFormatParam = "markdown", + ) -> str: + """Fetch customer reviews for Google Maps places by URL or place id. + + Use this to read feedback on specific places; get urls or place_ids + from surfsense_google_maps_scrape first if you only have a name. + Returns review text, rating, author, and date per review. + Example: place_ids=['ChIJj61dQgK6j4AR...'], sort_by='newest'. + """ + return await run_scraper( + client, + context, + platform="google_maps", + verb="reviews", + payload={ + "urls": urls, + "place_ids": place_ids, + "max_reviews": max_reviews, + "sort_by": sort_by, + "language": language, + "start_date": start_date, + }, + workspace=workspace, + response_format=response_format, + ) diff --git a/surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/google_search.py b/surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/google_search.py new file mode 100644 index 000000000..cc1a1f8ed --- /dev/null +++ b/surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/google_search.py @@ -0,0 +1,79 @@ +"""Google Search scraper tool.""" + +from __future__ import annotations + +from typing import Annotated + +from mcp.server.fastmcp import FastMCP +from pydantic import Field + +from ....core.client import SurfSenseClient +from ....core.rendering import ResponseFormatParam +from ....core.workspace_context import WorkspaceContext, WorkspaceParam +from ..annotations import SCRAPE +from ..capability import run_scraper + + +def register( + mcp: FastMCP, client: SurfSenseClient, context: WorkspaceContext +) -> None: + """Register the Google Search tool.""" + + @mcp.tool( + name="surfsense_google_search", + title="Scrape Google Search", + annotations=SCRAPE, + structured_output=False, + ) + async def google_search( + queries: Annotated[ + list[str], + Field( + min_length=1, + description="Search terms or full Google Search URLs, e.g. " + "['best rss readers 2026'].", + ), + ], + max_pages_per_query: Annotated[ + int, Field(ge=1, description="Result pages to fetch per query.") + ] = 1, + country_code: Annotated[ + str | None, + Field(description="Two-letter country to search from, e.g. 'us'."), + ] = None, + language_code: Annotated[ + str, Field(description="Results language, e.g. 'en'. Empty for default.") + ] = "", + site: Annotated[ + str | None, + Field( + description="Restrict results to one domain, e.g. 'example.com'." + ), + ] = None, + workspace: WorkspaceParam = None, + response_format: ResponseFormatParam = "markdown", + ) -> str: + """Scrape Google Search result pages for one or more queries. + + Use this to discover pages on the open web by topic; follow up with + surfsense_web_crawl to read a result in full. Do NOT use it for + Reddit, YouTube, or Google Maps research — the dedicated tools return + richer data. Returns each query's parsed results: title, url, and + snippet per organic result. + Example: queries=['notebooklm review'], site='news.ycombinator.com'. + """ + return await run_scraper( + client, + context, + platform="google_search", + verb="scrape", + payload={ + "queries": queries, + "max_pages_per_query": max_pages_per_query, + "country_code": country_code, + "language_code": language_code, + "site": site, + }, + workspace=workspace, + response_format=response_format, + ) diff --git a/surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/reddit.py b/surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/reddit.py new file mode 100644 index 000000000..035193ebc --- /dev/null +++ b/surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/reddit.py @@ -0,0 +1,98 @@ +"""Reddit scraper tool.""" + +from __future__ import annotations + +from typing import Annotated, Literal + +from mcp.server.fastmcp import FastMCP +from pydantic import Field + +from ....core.client import SurfSenseClient +from ....core.rendering import ResponseFormatParam +from ....core.workspace_context import WorkspaceContext, WorkspaceParam +from ..annotations import SCRAPE +from ..capability import run_scraper + +RedditSort = Literal["relevance", "hot", "top", "new", "rising", "comments"] +RedditTime = Literal["hour", "day", "week", "month", "year", "all"] + + +def register( + mcp: FastMCP, client: SurfSenseClient, context: WorkspaceContext +) -> None: + """Register the Reddit tool.""" + + @mcp.tool( + name="surfsense_reddit_scrape", + title="Search or scrape Reddit", + annotations=SCRAPE, + structured_output=False, + ) + async def reddit_scrape( + urls: Annotated[ + list[str] | None, + Field( + description="Reddit URLs: a post, a subreddit like " + "'https://reddit.com/r/LocalLLaMA', a user page, or a search " + "URL. Provide urls OR search_queries." + ), + ] = None, + search_queries: Annotated[ + list[str] | None, + Field( + description="Terms to search Reddit for, e.g. " + "['NotebookLM alternatives']. Provide search_queries OR urls." + ), + ] = None, + community: Annotated[ + str | None, + Field( + description="Restrict a search to one subreddit, name without " + "'r/', e.g. 'ArtificialInteligence'." + ), + ] = None, + sort: Annotated[RedditSort, Field(description="Post ordering.")] = "new", + time_filter: Annotated[ + RedditTime | None, + Field(description="Time window; only valid with sort='top'."), + ] = None, + max_items: Annotated[ + int, Field(ge=1, description="Maximum posts to return.") + ] = 10, + skip_comments: Annotated[ + bool, + Field( + description="True fetches posts only (faster); False also " + "fetches each post's comment thread." + ), + ] = False, + workspace: WorkspaceParam = None, + response_format: ResponseFormatParam = "markdown", + ) -> str: + """Search or scrape Reddit: posts, comments, subreddits, and users. + + Use this for ANY Reddit research — finding relevant subreddits or + communities for a topic, top posts, or discussions — instead of a + generic web search. Returns posts (title, text, score, subreddit, url) + with comment threads unless skip_comments is set. Every post carries + its subreddit, so to find communities for a topic, search posts and + aggregate their subreddits. + Example: search_queries=['NotebookLM'], sort='top', time_filter='month'. + """ + return await run_scraper( + client, + context, + platform="reddit", + verb="scrape", + payload={ + "urls": urls, + "search_queries": search_queries, + "community": community, + "sort": sort, + "time_filter": time_filter, + "max_items": max_items, + "skip_comments": skip_comments, + }, + workspace=workspace, + response_format=response_format, + ) diff --git a/surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/web.py b/surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/web.py new file mode 100644 index 000000000..9c24a4352 --- /dev/null +++ b/surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/web.py @@ -0,0 +1,89 @@ +"""Web crawl scraper tool.""" + +from __future__ import annotations + +from typing import Annotated + +from mcp.server.fastmcp import FastMCP +from pydantic import Field + +from ....core.client import SurfSenseClient +from ....core.rendering import ResponseFormatParam +from ....core.workspace_context import WorkspaceContext, WorkspaceParam +from ..annotations import SCRAPE +from ..capability import run_scraper + + +def register( + mcp: FastMCP, client: SurfSenseClient, context: WorkspaceContext +) -> None: + """Register the web crawl tool.""" + + @mcp.tool( + name="surfsense_web_crawl", + title="Crawl web pages", + annotations=SCRAPE, + structured_output=False, + ) + async def web_crawl( + start_urls: Annotated[ + list[str], + Field( + min_length=1, + description="Full URLs to fetch, e.g. " + "['https://example.com/blog/post'].", + ), + ], + max_crawl_depth: Annotated[ + int, + Field( + ge=0, + description="Link-hops to follow from start_urls within the " + "same site. 0 fetches only start_urls.", + ), + ] = 0, + max_crawl_pages: Annotated[ + int, Field(ge=1, description="Stop after this many pages in total.") + ] = 10, + max_length: Annotated[ + int, Field(ge=1, description="Max characters kept per page.") + ] = 50_000, + include_url_patterns: Annotated[ + list[str] | None, + Field( + description="Regexes; only discovered links matching one are " + "followed, e.g. ['/docs/.*']." + ), + ] = None, + exclude_url_patterns: Annotated[ + list[str] | None, + Field(description="Regexes; discovered links matching one are skipped."), + ] = None, + workspace: WorkspaceParam = None, + response_format: ResponseFormatParam = "markdown", + ) -> str: + """Fetch specific web pages and return their cleaned content as markdown. + + Use this to read a page the user names, or to spider a site from a + starting URL. Do NOT use it to find pages on a topic — use + surfsense_google_search for discovery. Returns one item per crawled + page: url, title, and the page text as markdown. + Example: start_urls=['https://blog.example.com'], max_crawl_depth=1, + include_url_patterns=['/2026/']. + """ + return await run_scraper( + client, + context, + platform="web", + verb="crawl", + payload={ + "startUrls": start_urls, + "maxCrawlDepth": max_crawl_depth, + "maxCrawlPages": max_crawl_pages, + "maxLength": max_length, + "includeUrlPatterns": include_url_patterns, + "excludeUrlPatterns": exclude_url_patterns, + }, + workspace=workspace, + response_format=response_format, + ) diff --git a/surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/youtube.py b/surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/youtube.py new file mode 100644 index 000000000..5582c82bb --- /dev/null +++ b/surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/youtube.py @@ -0,0 +1,131 @@ +"""YouTube scraper tools: videos and comments.""" + +from __future__ import annotations + +from typing import Annotated, Literal + +from mcp.server.fastmcp import FastMCP +from pydantic import Field + +from ....core.client import SurfSenseClient +from ....core.rendering import ResponseFormatParam +from ....core.workspace_context import WorkspaceContext, WorkspaceParam +from ..annotations import SCRAPE +from ..capability import run_scraper + +CommentSort = Literal["TOP_COMMENTS", "NEWEST_FIRST"] + + +def register( + mcp: FastMCP, client: SurfSenseClient, context: WorkspaceContext +) -> None: + """Register the YouTube video and comment tools.""" + + @mcp.tool( + name="surfsense_youtube_scrape", + title="Search or scrape YouTube", + annotations=SCRAPE, + structured_output=False, + ) + async def youtube_scrape( + urls: Annotated[ + list[str] | None, + Field( + description="YouTube URLs: video, channel, playlist, shorts, " + "or hashtag pages. Provide urls OR search_queries." + ), + ] = None, + search_queries: Annotated[ + list[str] | None, + Field( + description="Terms to search YouTube for, e.g. " + "['NotebookLM tutorial']. Provide search_queries OR urls." + ), + ] = None, + max_results: Annotated[ + int, Field(ge=1, description="Maximum videos to return.") + ] = 10, + download_subtitles: Annotated[ + bool, + Field(description="True also fetches each video's transcript."), + ] = False, + subtitles_language: Annotated[ + str, Field(description="Transcript language code, e.g. 'en'.") + ] = "en", + workspace: WorkspaceParam = None, + response_format: ResponseFormatParam = "markdown", + ) -> str: + """Search or scrape YouTube videos, optionally with transcripts. + + Use this for YouTube research: finding videos on a topic, or reading a + video's details or transcript. For a video's comment section use + surfsense_youtube_comments instead. Returns per-video metadata (title, + channel, views, description, url) and, if requested, the transcript. + Example: search_queries=['NotebookLM tutorial'], download_subtitles=True. + """ + return await run_scraper( + client, + context, + platform="youtube", + verb="scrape", + payload={ + "urls": urls, + "search_queries": search_queries, + "max_results": max_results, + "download_subtitles": download_subtitles, + "subtitles_language": subtitles_language, + }, + workspace=workspace, + response_format=response_format, + ) + + @mcp.tool( + name="surfsense_youtube_comments", + title="Fetch YouTube comments", + annotations=SCRAPE, + structured_output=False, + ) + async def youtube_comments( + urls: Annotated[ + list[str], + Field( + min_length=1, + description="YouTube video URLs, e.g. " + "['https://www.youtube.com/watch?v=abc123'].", + ), + ], + max_comments: Annotated[ + int, + Field( + ge=1, + description="Maximum comments per video, counting top-level " + "comments and replies together.", + ), + ] = 20, + sort_by: Annotated[ + CommentSort, Field(description="Comment ordering.") + ] = "NEWEST_FIRST", + workspace: WorkspaceParam = None, + response_format: ResponseFormatParam = "markdown", + ) -> str: + """Fetch the comments (and replies) on one or more YouTube videos. + + Use this when the user wants a video's discussion or audience reaction + rather than the video itself; get video URLs from + surfsense_youtube_scrape if you only have a topic. Returns comment + text, author, likes, and replies. + Example: urls=['https://www.youtube.com/watch?v=abc123'], max_comments=50. + """ + return await run_scraper( + client, + context, + platform="youtube", + verb="comments", + payload={ + "urls": urls, + "max_comments": max_comments, + "sort_by": sort_by, + }, + workspace=workspace, + response_format=response_format, + ) diff --git a/surfsense_mcp/src/surfsense_mcp/features/scrapers/run_history.py b/surfsense_mcp/src/surfsense_mcp/features/scrapers/run_history.py new file mode 100644 index 000000000..9274a1a69 --- /dev/null +++ b/surfsense_mcp/src/surfsense_mcp/features/scrapers/run_history.py @@ -0,0 +1,112 @@ +"""Scraper run history: list past runs and fetch one in full. + +A scrape whose inline result was truncated is retrievable here by run id, so the +model never re-runs a scraper just to recover output. +""" + +from __future__ import annotations + +from typing import Annotated + +from mcp.server.fastmcp import FastMCP +from pydantic import Field + +from ...core.client import SurfSenseClient +from ...core.rendering import ResponseFormatParam, clip, to_json +from ...core.workspace_context import WorkspaceContext, WorkspaceParam +from .annotations import READ_RUNS + + +def register( + mcp: FastMCP, client: SurfSenseClient, context: WorkspaceContext +) -> None: + """Register the run-history tools.""" + + @mcp.tool( + name="surfsense_list_scraper_runs", + title="List past scraper runs", + annotations=READ_RUNS, + structured_output=False, + ) + async def list_scraper_runs( + limit: Annotated[ + int, Field(ge=1, description="Maximum runs to list.") + ] = 20, + capability: Annotated[ + str | None, + Field( + description="Filter by capability slug, e.g. 'web.crawl' or " + "'reddit.scrape'." + ), + ] = None, + status: Annotated[ + str | None, + Field(description="Filter by run status: 'success' or 'error'."), + ] = None, + workspace: WorkspaceParam = None, + response_format: ResponseFormatParam = "markdown", + ) -> str: + """List recent scraper runs in the workspace, newest first. + + Use this to find the run_id of an earlier scrape — for example when an + inline result was truncated — then fetch it in full with + surfsense_get_scraper_run. Returns each run's id, capability, status, + item count, and creation time. + Example: capability='reddit.scrape', status='success'. + """ + resolved = await context.resolve(workspace) + runs = await client.request( + "GET", + f"/workspaces/{resolved.id}/scrapers/runs", + params={ + "limit": limit, + "capability": capability, + "status": status, + }, + ) + if response_format == "json": + return to_json(runs) + return _render_runs(runs) + + @mcp.tool( + name="surfsense_get_scraper_run", + title="Fetch one scraper run in full", + annotations=READ_RUNS, + structured_output=False, + ) + async def get_scraper_run( + run_id: Annotated[ + str, + Field( + description="Run id from surfsense_list_scraper_runs or a " + "prior scrape's output." + ), + ], + workspace: WorkspaceParam = None, + response_format: ResponseFormatParam = "markdown", + ) -> str: + """Fetch a single scraper run in full, including its stored output. + + Use this to retrieve the complete, untruncated result of an earlier + scrape. Do NOT re-run a scraper just to recover a truncated result — + fetch the stored run instead. + """ + resolved = await context.resolve(workspace) + run = await client.request( + "GET", f"/workspaces/{resolved.id}/scrapers/runs/{run_id}" + ) + if response_format == "json": + return clip(to_json(run)) + return f"# Run {run.get('id', run_id)}\n\n```json\n{clip(to_json(run))}\n```" + + +def _render_runs(runs: list[dict] | None) -> str: + if not runs: + return "No scraper runs found." + lines = ["# Scraper runs", ""] + for run in runs: + lines.append( + f"- **{run.get('id')}** — {run.get('capability')} · {run.get('status')} · " + f"{run.get('item_count', 0)} item(s) · {run.get('created_at')}" + ) + return "\n".join(lines)