mirror of
https://github.com/MODSetter/SurfSense.git
synced 2026-07-10 22:32:16 +02:00
refactor(mcp): split scraper and knowledge-base features into SRP modules
This commit is contained in:
CREDO23
parent
a7215c09dc
commit
2acc1426bf
15 changed files with 1153 additions and 955 deletions
|
|
@ -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)
|
||||
|
|
|
|||
51
surfsense_mcp/src/surfsense_mcp/core/workspace_matching.py
Normal file
51
surfsense_mcp/src/surfsense_mcp/core/workspace_matching.py
Normal 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)
|
||||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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."
|
||||
),
|
||||
]
|
||||
|
|
@ -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")
|
||||
|
|
@ -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}"
|
||||
)
|
||||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
)
|
||||
|
|
@ -0,0 +1 @@
|
|||
"""One module per scraper platform; each exposes register(mcp, client, context)."""
|
||||
|
|
@ -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,
|
||||
)
|
||||
|
|
@ -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,
|
||||
)
|
||||
|
|
@ -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,
|
||||
)
|
||||
|
|
@ -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,
|
||||
)
|
||||
|
|
@ -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
112
surfsense_mcp/src/surfsense_mcp/features/scrapers/run_history.py
Normal 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)
|
||||
Loading…
Add table
Add a link
Reference in a new issue