builderadmin
/
NoteDiscovery
1
0
Fork 0
Code Issues Pull Requests 1 Packages Projects Releases Wiki Activity

added pagination to endpoints and MCP tools (max results)

This commit is contained in:
Gamosoft 2026-03-23 17:47:28 +01:00
parent f5147ccc14
commit 8a87ba1a4d
7 changed files with 380 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

117
backend/main.py
View File

@ -42,6 +42,7 @@ from .utils import (
get_templates,
get_template_content,
apply_template_placeholders,
paginate,
)
from .plugins import PluginManager
from .themes import get_available_themes, get_theme_css
@ -876,23 +877,48 @@ async def list_tags():
@api_router.get("/tags/{tag_name}", tags=["Tags"])
async def get_notes_by_tag_endpoint(tag_name: str):
async def get_notes_by_tag_endpoint(
tag_name: str,
limit: Optional[int] = None,
offset: int = 0
):
"""
Get all notes that have a specific tag.
Get all notes that have a specific tag with optional pagination.
Args:
tag_name: The tag to filter by (case-insensitive)
limit: Maximum number of notes to return (optional, no default limit)
offset: Number of notes to skip (default: 0)
Returns:
List of notes matching the tag
Examples:
GET /api/tags/docker -> All notes with #docker tag
GET /api/tags/docker?limit=10 -> First 10 notes with #docker tag
"""
try:
notes = get_notes_by_tag(config['storage']['notes_dir'], tag_name)
return {
# Apply pagination with consistent sorting by path
paginated = paginate(
items=notes,
limit=limit,
offset=offset,
sort_key=lambda x: x.get('path', '').lower()
)
response = {
"tag": tag_name,
"count": len(notes),
"notes": notes
"count": paginated.total,
"notes": paginated.items
}
# Include pagination metadata only when limit is specified
if limit is not None:
response["pagination"] = paginated.to_dict()
return response
except Exception as e:
raise HTTPException(status_code=500, detail=safe_error_message(e, "Failed to get notes by tag"))
@ -1004,11 +1030,47 @@ async def create_note_from_template(request: Request, data: dict):
# --- Notes Endpoints ---
@api_router.get("/notes", tags=["Notes"])
async def list_notes():
"""List all notes with metadata"""
async def list_notes(
limit: Optional[int] = None,
offset: int = 0
):
"""
List all notes with metadata.
Supports optional pagination for API consumers (MCP, scripts):
- No parameters: Returns all notes (frontend compatibility)
- With limit: Returns paginated results with metadata
Args:
limit: Maximum number of notes to return (optional, no default limit)
offset: Number of notes to skip (default: 0)
Examples:
GET /api/notes -> All notes
GET /api/notes?limit=20 -> First 20 notes
GET /api/notes?limit=20&offset=20 -> Notes 21-40
"""
try:
notes, folders = scan_notes_fast_walk(config['storage']['notes_dir'], include_media=True)
return {"notes": notes, "folders": folders}
# Apply pagination with consistent sorting by path for stable results
result = paginate(
items=notes,
limit=limit,
offset=offset,
sort_key=lambda x: x.get('path', '').lower()
)
response = {
"notes": result.items,
"folders": folders
}
# Include pagination metadata only when limit is specified
if limit is not None:
response["pagination"] = result.to_dict()
return response
except Exception as e:
raise HTTPException(status_code=500, detail=safe_error_message(e, "Failed to list notes"))
@ -1161,8 +1223,24 @@ async def remove_note(request: Request, note_path: str):
@api_router.get("/search", tags=["Search"])
async def search(q: str):
"""Search notes by content"""
async def search(
q: str,
limit: Optional[int] = None,
offset: int = 0
):
"""
Search notes by content with optional pagination.
Args:
q: Search query string
limit: Maximum number of results to return (optional, no default limit)
offset: Number of results to skip (default: 0)
Examples:
GET /api/search?q=docker -> All matching results
GET /api/search?q=docker&limit=10 -> First 10 results
GET /api/search?q=docker&limit=10&offset=10 -> Results 11-20
"""
try:
if not config['search']['enabled']:
raise HTTPException(status_code=403, detail="Search is disabled")
@ -1172,7 +1250,24 @@ async def search(q: str):
# Run plugin hooks
plugin_manager.run_hook('on_search', query=q, results=results)
return {"results": results, "query": q}
# Apply pagination with consistent sorting by path
paginated = paginate(
items=results,
limit=limit,
offset=offset,
sort_key=lambda x: x.get('path', '').lower()
)
response = {
"results": paginated.items,
"query": q
}
# Include pagination metadata only when limit is specified
if limit is not None:
response["pagination"] = paginated.to_dict()
return response
except HTTPException:
raise
except Exception as e:

100
backend/utils.py
View File

@ -7,11 +7,109 @@ import re
import shutil
import threading
import time
from dataclasses import dataclass
from pathlib import Path
from typing import List, Dict, Optional, Tuple
from typing import List, Dict, Optional, Tuple, Any, TypeVar, Callable
from datetime import datetime, timezone
# ============================================================================
# Pagination Support
# ============================================================================
@dataclass
class PaginationResult:
"""
Result of applying pagination to a list.
Attributes:
items: The paginated subset of items
total: Total number of items before pagination
limit: The limit that was applied (None if no pagination)
offset: The offset that was applied
has_more: Whether there are more items after this page
"""
items: List[Any]
total: int
limit: Optional[int]
offset: int
has_more: bool
def to_dict(self) -> Dict[str, Any]:
"""Convert pagination info to dict for API response."""
return {
"limit": self.limit,
"offset": self.offset,
"total": self.total,
"has_more": self.has_more
}
T = TypeVar('T')
def paginate(
items: List[T],
limit: Optional[int] = None,
offset: int = 0,
sort_key: Optional[Callable[[T], Any]] = None,
sort_reverse: bool = False
) -> PaginationResult:
"""
Apply optional pagination to a list with consistent sorting.
This function is designed to be backward-compatible:
- If limit is None, returns all items (no pagination)
- If limit is provided, returns a paginated subset
Sorting is always applied (when sort_key is provided) to ensure
stable pagination across requests.
Args:
items: List of items to paginate
limit: Maximum number of items to return (None = no limit)
offset: Number of items to skip (default: 0)
sort_key: Function to extract sort key from item (e.g., lambda x: x['path'])
sort_reverse: If True, sort in descending order
Returns:
PaginationResult with items and pagination metadata
Example:
# No pagination (frontend compatibility)
result = paginate(notes)
# With pagination (MCP usage)
result = paginate(notes, limit=20, offset=0, sort_key=lambda x: x['path'])
"""
total = len(items)
# Apply sorting for consistent ordering (prevents out-of-order issues)
if sort_key is not None:
items = sorted(items, key=sort_key, reverse=sort_reverse)
# Apply pagination only if limit is specified
if limit is not None:
# Clamp offset to valid range
offset = max(0, min(offset, total))
end = offset + limit
paginated_items = items[offset:end]
has_more = end < total
else:
# No pagination - return all items
paginated_items = items
offset = 0
has_more = False
return PaginationResult(
items=paginated_items,
total=total,
limit=limit,
offset=offset,
has_more=has_more
)
# In-memory cache for parsed tags
# Format: {file_path: (mtime, tags)}
_tag_cache: Dict[str, Tuple[float, List[str]]] = {}

43
documentation/API.md
View File

@ -7,12 +7,28 @@ Base URL: `http://localhost:8000`
### List All Notes
```http
GET /api/notes
GET /api/notes?limit=20&offset=0
```
Returns all notes with their metadata and folder structure.
**Example:**
**Optional Pagination:**
| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `limit` | integer | - | Max notes to return (omit for all) |
| `offset` | integer | 0 | Number of notes to skip |
When `limit` is provided, the response includes pagination metadata.
**Examples:**
```bash
# Get all notes (default)
curl http://localhost:8000/api/notes
# Get first 20 notes
curl "http://localhost:8000/api/notes?limit=20"
# Get notes 21-40
curl "http://localhost:8000/api/notes?limit=20&offset=20"
```
### Get Note Content
@ -269,11 +285,22 @@ Content-Type: application/json
### Search Notes
```http
GET /api/search?q={query}
GET /api/search?q={query}&limit=10&offset=0
```
**Example:**
**Optional Pagination:**
| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `limit` | integer | - | Max results to return (omit for all) |
| `offset` | integer | 0 | Number of results to skip |
**Examples:**
```bash
# Search all matches
curl "http://localhost:8000/api/search?q=hello"
# Get first 10 results
curl "http://localhost:8000/api/search?q=hello&limit=10"
```
## 🎨 Themes
@ -409,14 +436,24 @@ Returns all tags found in notes with their usage counts.
```
### Get Notes by Tag
`GET /api/tags/{tag_name}`
```http
GET /api/tags/{tag_name}
GET /api/tags/{tag_name}?limit=10&offset=0
```
Returns all notes that have a specific tag.
**Optional Pagination:**
| Parameter | Type | Default | Description |
|-----------|------|---------|-------------|
| `limit` | integer | - | Max notes to return (omit for all) |
| `offset` | integer | 0 | Number of notes to skip |
**Response:**
```json
{
"tag": "python",
"count": 5,
"notes": [
{
"path": "tutorials/python-basics.md",

22
documentation/MCP.md
View File

@ -114,8 +114,8 @@ The MCP server provides these tools to AI assistants:
| Tool | Description |
|------|-------------|
| `search_notes` | Full-text search across all notes |
| `list_notes` | List all notes with metadata |
| `search_notes` | Full-text search across all notes (supports `max_results`) |
| `list_notes` | List all notes with metadata (supports `max_results`) |
| `get_note` | Read a specific note's content |
| `get_recent_notes` | Get recently modified notes (last N days) |
@ -124,7 +124,7 @@ The MCP server provides these tools to AI assistants:
| Tool | Description |
|------|-------------|
| `list_tags` | List all tags with note counts |
| `get_notes_by_tag` | Find notes with a specific tag |
| `get_notes_by_tag` | Find notes with a specific tag (supports `max_results`) |
| `get_graph` | Get knowledge graph data |
### Note Management
@ -153,6 +153,22 @@ The MCP server provides these tools to AI assistants:
## Tool Details
### Pagination with `max_results`
Some tools support an optional `max_results` parameter to limit results for large vaults:
| Tool | Parameter | Description |
|------|-----------|-------------|
| `search_notes` | `max_results` | Limit number of search results |
| `list_notes` | `max_results` | Limit number of notes returned |
| `get_notes_by_tag` | `max_results` | Limit number of notes returned |
When omitted, all results are returned. This is useful for large note collections where you only need a subset.
**Example prompt:** "Search for notes about Python, but just show me the first 5 results"
---
### `append_to_note`
Append content to an existing note without overwriting. Perfect for journals, logs, or collecting ideas.

50
mcp_server/client.py
View File

@ -139,14 +139,26 @@ class NoteDiscoveryClient:
# Notes API
# =========================================================================
def list_notes(self) -> APIResponse:
def list_notes(
self,
limit: int | None = None,
offset: int = 0
) -> APIResponse:
"""
List all notes with metadata.
Args:
limit: Maximum number of notes to return (None = no limit)
offset: Number of notes to skip (default: 0)
Returns:
APIResponse with notes and folders data
"""
return self._request("GET", "/api/notes")
params: dict[str, str] = {}
if limit is not None:
params["limit"] = str(limit)
params["offset"] = str(offset)
return self._request("GET", "/api/notes", params=params if params else None)
def get_note(self, path: str) -> APIResponse:
"""
@ -255,17 +267,28 @@ class NoteDiscoveryClient:
# Search API
# =========================================================================
def search(self, query: str) -> APIResponse:
def search(
self,
query: str,
limit: int | None = None,
offset: int = 0
) -> APIResponse:
"""
Search notes by query.
Args:
query: Search query string
limit: Maximum number of results to return (None = no limit)
offset: Number of results to skip (default: 0)
Returns:
APIResponse with search results
"""
return self._request("GET", "/api/search", params={"q": query})
params: dict[str, str] = {"q": query}
if limit is not None:
params["limit"] = str(limit)
params["offset"] = str(offset)
return self._request("GET", "/api/search", params=params)
# =========================================================================
# Tags API
@ -280,18 +303,33 @@ class NoteDiscoveryClient:
"""
return self._request("GET", "/api/tags")
def get_notes_by_tag(self, tag: str) -> APIResponse:
def get_notes_by_tag(
self,
tag: str,
limit: int | None = None,
offset: int = 0
) -> APIResponse:
"""
Get notes with a specific tag.
Args:
tag: Tag name
limit: Maximum number of notes to return (None = no limit)
offset: Number of notes to skip (default: 0)
Returns:
APIResponse with matching notes
"""
encoded_tag = urllib.parse.quote(tag, safe="")
return self._request("GET", f"/api/tags/{encoded_tag}")
params: dict[str, str] = {}
if limit is not None:
params["limit"] = str(limit)
params["offset"] = str(offset)
return self._request(
"GET",
f"/api/tags/{encoded_tag}",
params=params if params else None
)
# =========================================================================
# Folders API

60
mcp_server/server.py
View File

@ -207,23 +207,33 @@ class MCPServer:
def _tool_search_notes(self, args: dict) -> str:
"""Search notes by query."""
query = args.get("query", "")
if not query:
return "Error: query is required"
max_results = args.get("max_results")
response = self.client.search(query)
if not query:
return "No search term provided. Please specify a query."
# Use server-side pagination if max_results is specified
response = self.client.search(query, limit=max_results)
if not response.success:
return f"Search failed: {response.error}"
data = response.data or {}
results = data.get("results", [])
pagination = data.get("pagination")
if not results:
return f"No notes found matching '{query}'"
# Build header with pagination info if available
if pagination:
total = pagination.get("total", len(results))
output = [f"Found {total} result(s) for '{query}' (showing {len(results)}):\n"]
else:
output = [f"Found {len(results)} result(s) for '{query}':\n"]
# Format search results
output = [f"Found {len(results)} result(s) for '{query}':\n"]
for i, result in enumerate(results[:20], 1): # Limit to 20 results
for i, result in enumerate(results, 1):
path = result.get("path", "unknown")
snippet = result.get("snippet", "")
output.append(f"{i}. **{path}**")
@ -231,14 +241,18 @@ class MCPServer:
output.append(f" {snippet[:200]}...")
output.append("")
if len(results) > 20:
output.append(f"... and {len(results) - 20} more results")
# Add pagination hint if there are more results
if pagination and pagination.get("has_more"):
output.append(f"... more results available (use max_results and offset to paginate)")
return "\n".join(output)
def _tool_list_notes(self, args: dict) -> str:
"""List all notes."""
response = self.client.list_notes()
max_results = args.get("max_results")
# Use server-side pagination if max_results is specified
response = self.client.list_notes(limit=max_results)
if not response.success:
return f"Failed to list notes: {response.error}"
@ -246,8 +260,14 @@ class MCPServer:
data = response.data or {}
notes = data.get("notes", [])
folders = data.get("folders", [])
pagination = data.get("pagination")
output = [f"Found {len(notes)} note(s) in {len(folders)} folder(s):\n"]
# Build header with pagination info if available
if pagination:
total = pagination.get("total", len(notes))
output = [f"Found {total} note(s) in {len(folders)} folder(s) (showing {len(notes)}):\n"]
else:
output = [f"Found {len(notes)} note(s) in {len(folders)} folder(s):\n"]
# Group notes by folder
notes_by_folder: dict[str, list] = {}
@ -266,6 +286,10 @@ class MCPServer:
output.append(f" 📝 {name} (modified: {modified})")
output.append("")
# Add pagination hint if there are more results
if pagination and pagination.get("has_more"):
output.append(f"... more notes available (use max_results to limit)")
return "\n".join(output)
def _tool_get_note(self, args: dict) -> str:
@ -316,25 +340,39 @@ class MCPServer:
def _tool_get_notes_by_tag(self, args: dict) -> str:
"""Get notes with a specific tag."""
tag = args.get("tag", "")
max_results = args.get("max_results")
if not tag:
return "Error: tag is required"
response = self.client.get_notes_by_tag(tag)
# Use server-side pagination if max_results is specified
response = self.client.get_notes_by_tag(tag, limit=max_results)
if not response.success:
return f"Failed to get notes by tag: {response.error}"
data = response.data or {}
notes = data.get("notes", [])
count = data.get("count", len(notes))
pagination = data.get("pagination")
if not notes:
return f"No notes found with tag '#{tag}'"
output = [f"Notes with tag #{tag}:\n"]
# Build header with pagination info if available
if pagination:
output = [f"Notes with tag #{tag} ({count} total, showing {len(notes)}):\n"]
else:
output = [f"Notes with tag #{tag} ({count} total):\n"]
for note in notes:
path = note.get("path", "unknown")
output.append(f" 📝 {path}")
# Add pagination hint if there are more results
if pagination and pagination.get("has_more"):
output.append(f"\n... more notes available (use max_results to limit)")
return "\n".join(output)
def _tool_get_graph(self, args: dict) -> str:

15
mcp_server/tools.py
View File

@ -21,6 +21,10 @@ TOOLS: list[dict[str, Any]] = [
"query": {
"type": "string",
"description": "Search query. Can be keywords, phrases, or natural language."
},
"max_results": {
"type": "integer",
"description": "Maximum number of results to return. Useful for large vaults. If not specified, returns all matches."
}
},
"required": ["query"]
@ -31,7 +35,12 @@ TOOLS: list[dict[str, Any]] = [
"description": "List all notes in the knowledge base with their metadata (title, path, last modified date, size). Use this to get an overview of available notes or find notes by browsing.",
"inputSchema": {
"type": "object",
"properties": {},
"properties": {
"max_results": {
"type": "integer",
"description": "Maximum number of notes to return. Useful for large vaults. If not specified, returns all notes."
}
},
"required": []
}
},
@ -71,6 +80,10 @@ TOOLS: list[dict[str, Any]] = [
"tag": {
"type": "string",
"description": "Tag name (without the # symbol)"
},
"max_results": {
"type": "integer",
"description": "Maximum number of notes to return. Useful for large vaults. If not specified, returns all matches."
}
},
"required": ["tag"]