Chris Coutinho
e1412320a7
Add ToolAnnotations to all 105+ MCP tools across 13 modules to enable better client-side UX with human-readable titles and behavioral hints. Changes: - Add title and ToolAnnotations to all @mcp.tool() decorators - Apply correct idempotency classification per ADR-017 - Add destructiveHint for delete operations - Set openWorldHint=False for semantic search (internal data only) Modules updated: - OAuth (4 tools): Authentication and provisioning - Notes (7 tools): Note management - WebDAV (11 tools): File operations - Semantic (3 tools): Semantic search and RAG - Calendar (16 tools): Events and todos - Contacts (7 tools): Address book management - Sharing (5 tools): File/folder sharing - Tables (6 tools): Structured data - Deck (25 tools): Kanban board management - Cookbook (13 tools): Recipe management - News (8 tools): RSS feed reader Annotation patterns: - Read operations: readOnlyHint=True, openWorldHint=True - Create operations: idempotentHint=False, openWorldHint=True - Update operations: idempotentHint=False, openWorldHint=True - Delete operations: destructiveHint=True, idempotentHint=True, openWorldHint=True See docs/ADR-017-mcp-tool-annotations.md for rationale and implementation details. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
456 lines
17 KiB
Python
456 lines
17 KiB
Python
import logging
|
|
|
|
from httpx import HTTPStatusError, RequestError
|
|
from mcp.server.fastmcp import Context, FastMCP
|
|
from mcp.shared.exceptions import McpError
|
|
from mcp.types import ErrorData, ToolAnnotations
|
|
|
|
from nextcloud_mcp_server.auth import require_scopes
|
|
from nextcloud_mcp_server.context import get_client
|
|
from nextcloud_mcp_server.models.notes import (
|
|
AppendContentResponse,
|
|
CreateNoteResponse,
|
|
DeleteNoteResponse,
|
|
Note,
|
|
NoteSearchResult,
|
|
NotesSettings,
|
|
SearchNotesResponse,
|
|
UpdateNoteResponse,
|
|
)
|
|
from nextcloud_mcp_server.observability.metrics import instrument_tool
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def configure_notes_tools(mcp: FastMCP):
|
|
@mcp.resource("notes://settings")
|
|
async def notes_get_settings():
|
|
"""Get the Notes App settings"""
|
|
ctx: Context = (
|
|
mcp.get_context()
|
|
) # https://github.com/modelcontextprotocol/python-sdk/issues/244
|
|
client = await get_client(ctx)
|
|
settings_data = await client.notes.get_settings()
|
|
return NotesSettings(**settings_data)
|
|
|
|
@mcp.resource("nc://Notes/{note_id}/attachments/{attachment_filename}")
|
|
async def nc_notes_get_attachment_resource(note_id: int, attachment_filename: str):
|
|
"""Get a specific attachment from a note"""
|
|
ctx: Context = mcp.get_context()
|
|
client = await get_client(ctx)
|
|
# Assuming a method get_note_attachment exists in the client
|
|
# This method should return the raw content and determine the mime type
|
|
content, mime_type = await client.webdav.get_note_attachment(
|
|
note_id=note_id, filename=attachment_filename
|
|
)
|
|
return {
|
|
"contents": [
|
|
{
|
|
# Use uppercase 'Notes' to match the decorator
|
|
"uri": f"nc://Notes/{note_id}/attachments/{attachment_filename}",
|
|
"mimeType": mime_type, # Client needs to determine this
|
|
"data": content, # Return raw bytes/data
|
|
}
|
|
]
|
|
}
|
|
|
|
@mcp.resource("nc://Notes/{note_id}")
|
|
async def nc_get_note_resource(note_id: int):
|
|
"""Get user note using note id"""
|
|
|
|
ctx: Context = mcp.get_context()
|
|
client = await get_client(ctx)
|
|
try:
|
|
note_data = await client.notes.get_note(note_id)
|
|
return Note(**note_data)
|
|
except RequestError as e:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1,
|
|
message=f"Network error retrieving note {note_id}: {str(e)}",
|
|
)
|
|
)
|
|
except HTTPStatusError as e:
|
|
if e.response.status_code == 404:
|
|
raise McpError(ErrorData(code=-1, message=f"Note {note_id} not found"))
|
|
elif e.response.status_code == 403:
|
|
raise McpError(
|
|
ErrorData(code=-1, message=f"Access denied to note {note_id}")
|
|
)
|
|
else:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1,
|
|
message=f"Failed to retrieve note {note_id}: {e.response.reason_phrase}",
|
|
)
|
|
)
|
|
|
|
@mcp.tool(
|
|
title="Create Note",
|
|
annotations=ToolAnnotations(
|
|
idempotentHint=False, # Multiple calls create multiple notes
|
|
openWorldHint=True,
|
|
),
|
|
)
|
|
@require_scopes("notes:write")
|
|
@instrument_tool
|
|
async def nc_notes_create_note(
|
|
title: str, content: str, category: str, ctx: Context
|
|
) -> CreateNoteResponse:
|
|
"""Create a new note (requires notes:write scope)"""
|
|
client = await get_client(ctx)
|
|
try:
|
|
note_data = await client.notes.create_note(
|
|
title=title,
|
|
content=content,
|
|
category=category,
|
|
)
|
|
note = Note(**note_data)
|
|
return CreateNoteResponse(
|
|
id=note.id, title=note.title, category=note.category, etag=note.etag
|
|
)
|
|
except RequestError as e:
|
|
raise McpError(
|
|
ErrorData(code=-1, message=f"Network error creating note: {str(e)}")
|
|
)
|
|
except HTTPStatusError as e:
|
|
if e.response.status_code == 403:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1,
|
|
message="Access denied: insufficient permissions to create notes",
|
|
)
|
|
)
|
|
elif e.response.status_code == 413:
|
|
raise McpError(ErrorData(code=-1, message="Note content too large"))
|
|
elif e.response.status_code == 409:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1,
|
|
message=f"A note with title '{title}' already exists in this category",
|
|
)
|
|
)
|
|
else:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1,
|
|
message=f"Failed to create note: server error ({e.response.status_code})",
|
|
)
|
|
)
|
|
|
|
@mcp.tool(
|
|
title="Update Note",
|
|
annotations=ToolAnnotations(
|
|
idempotentHint=False, # Requires etag which changes = not idempotent
|
|
openWorldHint=True,
|
|
),
|
|
)
|
|
@require_scopes("notes:write")
|
|
@instrument_tool
|
|
async def nc_notes_update_note(
|
|
note_id: int,
|
|
etag: str,
|
|
title: str | None,
|
|
content: str | None,
|
|
category: str | None,
|
|
ctx: Context,
|
|
) -> UpdateNoteResponse:
|
|
"""Update an existing note's title, content, or category (requires notes:write scope).
|
|
|
|
REQUIRED: etag parameter must be provided to prevent overwriting concurrent changes.
|
|
Get the current ETag by first retrieving the note using nc_notes_get_note tool.
|
|
If the note has been modified by someone else since you retrieved it,
|
|
the update will fail with a 412 error."""
|
|
logger.info("Updating note %s", note_id)
|
|
client = await get_client(ctx)
|
|
try:
|
|
note_data = await client.notes.update(
|
|
note_id=note_id,
|
|
etag=etag,
|
|
title=title,
|
|
content=content,
|
|
category=category,
|
|
)
|
|
note = Note(**note_data)
|
|
return UpdateNoteResponse(
|
|
id=note.id, title=note.title, category=note.category, etag=note.etag
|
|
)
|
|
except RequestError as e:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1, message=f"Network error updating note {note_id}: {str(e)}"
|
|
)
|
|
)
|
|
except HTTPStatusError as e:
|
|
if e.response.status_code == 404:
|
|
raise McpError(ErrorData(code=-1, message=f"Note {note_id} not found"))
|
|
elif e.response.status_code == 412:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1,
|
|
message=f"Note {note_id} has been modified by someone else. Please refresh and try again.",
|
|
)
|
|
)
|
|
elif e.response.status_code == 403:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1,
|
|
message=f"Access denied: insufficient permissions to update note {note_id}",
|
|
)
|
|
)
|
|
elif e.response.status_code == 413:
|
|
raise McpError(
|
|
ErrorData(code=-1, message="Updated note content is too large")
|
|
)
|
|
else:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1,
|
|
message=f"Failed to update note {note_id}: server error ({e.response.status_code})",
|
|
)
|
|
)
|
|
|
|
@mcp.tool(
|
|
title="Append to Note",
|
|
annotations=ToolAnnotations(
|
|
idempotentHint=False, # Each call adds content = not idempotent
|
|
openWorldHint=True,
|
|
),
|
|
)
|
|
@require_scopes("notes:write")
|
|
@instrument_tool
|
|
async def nc_notes_append_content(
|
|
note_id: int, content: str, ctx: Context
|
|
) -> AppendContentResponse:
|
|
"""Append content to an existing note. The tool adds a `\n---\n`
|
|
between the note and what will be appended."""
|
|
|
|
logger.info("Appending content to note %s", note_id)
|
|
client = await get_client(ctx)
|
|
try:
|
|
note_data = await client.notes.append_content(
|
|
note_id=note_id, content=content
|
|
)
|
|
note = Note(**note_data)
|
|
return AppendContentResponse(
|
|
id=note.id, title=note.title, category=note.category, etag=note.etag
|
|
)
|
|
except RequestError as e:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1,
|
|
message=f"Network error appending to note {note_id}: {str(e)}",
|
|
)
|
|
)
|
|
except HTTPStatusError as e:
|
|
if e.response.status_code == 404:
|
|
raise McpError(ErrorData(code=-1, message=f"Note {note_id} not found"))
|
|
elif e.response.status_code == 403:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1,
|
|
message=f"Access denied: insufficient permissions to modify note {note_id}",
|
|
)
|
|
)
|
|
elif e.response.status_code == 413:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1,
|
|
message="Content to append would make the note too large",
|
|
)
|
|
)
|
|
else:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1,
|
|
message=f"Failed to append content to note {note_id}: server error ({e.response.status_code})",
|
|
)
|
|
)
|
|
|
|
@mcp.tool(
|
|
title="Search Notes",
|
|
annotations=ToolAnnotations(
|
|
readOnlyHint=True, # Search doesn't modify data
|
|
openWorldHint=True,
|
|
),
|
|
)
|
|
@require_scopes("notes:read")
|
|
@instrument_tool
|
|
async def nc_notes_search_notes(query: str, ctx: Context) -> SearchNotesResponse:
|
|
"""Search notes by title or content, returning only id, title, and category (requires notes:read scope)."""
|
|
client = await get_client(ctx)
|
|
try:
|
|
search_results_raw = await client.notes_search_notes(query=query)
|
|
|
|
# Convert to NoteSearchResult models, including the _score field
|
|
results = [
|
|
NoteSearchResult(
|
|
id=result["id"],
|
|
title=result["title"],
|
|
category=result["category"],
|
|
score=result.get("_score"), # Include search score if available
|
|
)
|
|
for result in search_results_raw
|
|
]
|
|
|
|
return SearchNotesResponse(
|
|
results=results, query=query, total_found=len(results)
|
|
)
|
|
except RequestError as e:
|
|
raise McpError(
|
|
ErrorData(code=-1, message=f"Network error searching notes: {str(e)}")
|
|
)
|
|
except HTTPStatusError as e:
|
|
if e.response.status_code == 403:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1,
|
|
message="Access denied: insufficient permissions to search notes",
|
|
)
|
|
)
|
|
elif e.response.status_code == 400:
|
|
raise McpError(
|
|
ErrorData(code=-1, message="Invalid search query format")
|
|
)
|
|
else:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1,
|
|
message=f"Search failed: server error ({e.response.status_code})",
|
|
)
|
|
)
|
|
|
|
@mcp.tool(
|
|
title="Get Note",
|
|
annotations=ToolAnnotations(
|
|
readOnlyHint=True, # Read operation only
|
|
openWorldHint=True,
|
|
),
|
|
)
|
|
@require_scopes("notes:read")
|
|
@instrument_tool
|
|
async def nc_notes_get_note(note_id: int, ctx: Context) -> Note:
|
|
"""Get a specific note by its ID (requires notes:read scope)"""
|
|
client = await get_client(ctx)
|
|
try:
|
|
note_data = await client.notes.get_note(note_id)
|
|
return Note(**note_data)
|
|
except RequestError as e:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1, message=f"Network error getting note {note_id}: {str(e)}"
|
|
)
|
|
)
|
|
except HTTPStatusError as e:
|
|
if e.response.status_code == 404:
|
|
raise McpError(ErrorData(code=-1, message=f"Note {note_id} not found"))
|
|
elif e.response.status_code == 403:
|
|
raise McpError(
|
|
ErrorData(code=-1, message=f"Access denied to note {note_id}")
|
|
)
|
|
else:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1,
|
|
message=f"Failed to retrieve note {note_id}: {e.response.reason_phrase}",
|
|
)
|
|
)
|
|
|
|
@mcp.tool(
|
|
title="Get Note Attachment",
|
|
annotations=ToolAnnotations(
|
|
readOnlyHint=True, # Read operation only
|
|
openWorldHint=True,
|
|
),
|
|
)
|
|
@require_scopes("notes:read")
|
|
@instrument_tool
|
|
async def nc_notes_get_attachment(
|
|
note_id: int, attachment_filename: str, ctx: Context
|
|
) -> dict[str, str]:
|
|
"""Get a specific attachment from a note"""
|
|
client = await get_client(ctx)
|
|
try:
|
|
content, mime_type = await client.webdav.get_note_attachment(
|
|
note_id=note_id, filename=attachment_filename
|
|
)
|
|
return { # type: ignore
|
|
"uri": f"nc://Notes/{note_id}/attachments/{attachment_filename}",
|
|
"mimeType": mime_type,
|
|
"data": content,
|
|
}
|
|
except RequestError as e:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1,
|
|
message=f"Network error getting attachment {attachment_filename} for note {note_id}: {str(e)}",
|
|
)
|
|
)
|
|
except HTTPStatusError as e:
|
|
if e.response.status_code == 404:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1,
|
|
message=f"Attachment {attachment_filename} not found for note {note_id}",
|
|
)
|
|
)
|
|
elif e.response.status_code == 403:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1,
|
|
message=f"Access denied to attachment {attachment_filename} for note {note_id}",
|
|
)
|
|
)
|
|
else:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1,
|
|
message=f"Failed to retrieve attachment: {e.response.reason_phrase}",
|
|
)
|
|
)
|
|
|
|
@mcp.tool(
|
|
title="Delete Note",
|
|
annotations=ToolAnnotations(
|
|
destructiveHint=True, # Permanently deletes data
|
|
idempotentHint=True, # Deleting deleted note = same end state
|
|
openWorldHint=True,
|
|
),
|
|
)
|
|
@require_scopes("notes:write")
|
|
@instrument_tool
|
|
async def nc_notes_delete_note(note_id: int, ctx: Context) -> DeleteNoteResponse:
|
|
"""Delete a note permanently"""
|
|
logger.info("Deleting note %s", note_id)
|
|
client = await get_client(ctx)
|
|
try:
|
|
await client.notes.delete_note(note_id)
|
|
return DeleteNoteResponse(
|
|
status_code=200,
|
|
message=f"Note {note_id} deleted successfully",
|
|
deleted_id=note_id,
|
|
)
|
|
except RequestError as e:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1, message=f"Network error deleting note {note_id}: {str(e)}"
|
|
)
|
|
)
|
|
except HTTPStatusError as e:
|
|
if e.response.status_code == 404:
|
|
raise McpError(ErrorData(code=-1, message=f"Note {note_id} not found"))
|
|
elif e.response.status_code == 403:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1,
|
|
message=f"Access denied: insufficient permissions to delete note {note_id}",
|
|
)
|
|
)
|
|
else:
|
|
raise McpError(
|
|
ErrorData(
|
|
code=-1,
|
|
message=f"Failed to delete note {note_id}: server error ({e.response.status_code})",
|
|
)
|
|
)
|