apunkt/SurfSense
1
0
Fork 0
mirror of https://github.com/MODSetter/SurfSense.git synced 2026-07-10 22:32:16 +02:00
Code Issues Projects Releases Packages Wiki Activity

refactor(mcp): split scraper and knowledge-base features into SRP modules

Browse source
This commit is contained in:
CREDO23 2026-07-07 17:22:37 +02:00
parent 1c01aaeca5
commit ea5b27eaa8
15 changed files with 1153 additions and 955 deletions
Download patch file Download diff file Expand all files Collapse all files

43
surfsense_mcp/src/surfsense_mcp/core/workspace_context.py
View file

@ -16,6 +16,7 @@ from pydantic import Field
from .auth.identity import current_identity from .auth.identity import current_identity
from .client import SurfSenseClient from .client import SurfSenseClient
from .errors import ToolError 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 # 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 # of keys can't grow memory without limit; an evicted caller just re-resolves
@ -99,43 +100,20 @@ class WorkspaceContext:
) )
raise ToolError( raise ToolError(
"No workspace selected. Choose one first with surfsense_select_workspace, " "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: async def _match(self, reference: str | int) -> Workspace:
workspaces = await self.fetch_all() workspaces = await self.fetch_all()
as_id = _as_int(reference) as_id = as_int(reference)
if as_id is not None: if as_id is not None:
found = next((w for w in workspaces if w.id == as_id), None) found = next((w for w in workspaces if w.id == as_id), None)
if found is None: if found is None:
raise ToolError( 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 found
return _match_by_name(str(reference), workspaces) 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)}."
)
def _to_workspace(row: dict) -> Workspace: def _to_workspace(row: dict) -> Workspace:
@ -146,14 +124,3 @@ def _to_workspace(row: dict) -> Workspace:
is_owner=row.get("is_owner", False), is_owner=row.get("is_owner", False),
member_count=row.get("member_count", 1), 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)

51
surfsense_mcp/src/surfsense_mcp/core/workspace_matching.py Normal file
View file

@ -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)

371
surfsense_mcp/src/surfsense_mcp/features/knowledge_base/__init__.py
View file

@ -1,379 +1,22 @@
"""Knowledge-base tools: search the KB and manage its documents. """Knowledge-base tools: search the KB and manage its documents.
Semantic search plus the document lifecycle — list, read, add text, upload a 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 file, update, and delete — over a workspace's knowledge base. Read tools live in
default to the active workspace; document ids identify a single document across search_tools, mutations in document_tools.
the whole account, so id-addressed tools need no workspace.
""" """
from __future__ import annotations from __future__ import annotations
import mimetypes
from pathlib import Path
from typing import Annotated
from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp import FastMCP
from mcp.types import ToolAnnotations
from pydantic import Field
from ...core.client import SurfSenseClient from ...core.client import SurfSenseClient
from ...core.errors import ToolError from ...core.workspace_context import WorkspaceContext
from ...core.rendering import ResponseFormatParam, clip, to_json from . import document_tools, search_tools
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."
),
]
def register( def register(
mcp: FastMCP, client: SurfSenseClient, context: WorkspaceContext mcp: FastMCP, client: SurfSenseClient, context: WorkspaceContext
) -> None: ) -> None:
"""Register the knowledge-base tools on the server.""" """Register every knowledge-base tool on the server."""
search_tools.register(mcp, client, context)
@mcp.tool( document_tools.register(mcp, client, context)
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}"
)

34
surfsense_mcp/src/surfsense_mcp/features/knowledge_base/annotations.py Normal file
View file

@ -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."
),
]

185
surfsense_mcp/src/surfsense_mcp/features/knowledge_base/document_tools.py Normal file
View file

@ -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")

188
surfsense_mcp/src/surfsense_mcp/features/knowledge_base/search_tools.py Normal file
View file

@ -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}"
)

562
surfsense_mcp/src/surfsense_mcp/features/scrapers/__init__.py
View file

@ -1,570 +1,26 @@
"""Scraper tools: one MCP surface per SurfSense platform capability. """Scraper tools: one MCP surface per SurfSense platform capability.
Web crawl, Google Search, Reddit, YouTube, and Google Maps each get a tool that 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 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 __future__ import annotations
from typing import Annotated, Literal
from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp import FastMCP
from mcp.types import ToolAnnotations
from pydantic import Field
from ...core.client import SurfSenseClient from ...core.client import SurfSenseClient
from ...core.rendering import ResponseFormatParam, clip, to_json from ...core.workspace_context import WorkspaceContext
from ...core.workspace_context import WorkspaceContext, WorkspaceParam from . import run_history
from .capability import run_scraper from .platforms import google_maps, google_search, reddit, web, youtube
# Scrapers reach the open web and record a billable run; they are neither _REGISTRARS = (web, google_search, reddit, youtube, google_maps, run_history)
# 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"]
def register( def register(
mcp: FastMCP, client: SurfSenseClient, context: WorkspaceContext mcp: FastMCP, client: SurfSenseClient, context: WorkspaceContext
) -> None: ) -> None:
"""Register the scraper and run-history tools on the server.""" """Register every scraper and run-history tool on the server."""
for module in _REGISTRARS:
@mcp.tool( module.register(mcp, client, context)
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)

13
surfsense_mcp/src/surfsense_mcp/features/scrapers/annotations.py Normal file
View file

@ -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
)

1
surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/__init__.py Normal file
View file

@ -0,0 +1 @@
"""One module per scraper platform; each exposes register(mcp, client, context)."""

151
surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/google_maps.py Normal file
View file

@ -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,
)

79
surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/google_search.py Normal file
View file

@ -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,
)

98
surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/reddit.py Normal file
View file

@ -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,
)

89
surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/web.py Normal file
View file

@ -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,
)

131
surfsense_mcp/src/surfsense_mcp/features/scrapers/platforms/youtube.py Normal file
View file

@ -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,
)

112
surfsense_mcp/src/surfsense_mcp/features/scrapers/run_history.py Normal file
View file

@ -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)