brandon/nextcloud-mcp-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(server): Replace ErrorResponses with standard McpErrors

Browse Source
This commit is contained in:
Chris Coutinho
2025-08-31 20:58:12 +02:00
parent d712b5487c
commit ef1fb9e9aa
6 changed files with 99 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/models/__init__.py
-2
View File
@@ -3,7 +3,6 @@
# Base models
from .base import (
BaseResponse,
ErrorResponse,
SuccessResponse,
IdResponse,
StatusResponse,
@@ -82,7 +81,6 @@ from .webdav import (
__all__ = [
# Base models
"BaseResponse",
"ErrorResponse",
"SuccessResponse",
"IdResponse",
"StatusResponse",
nextcloud_mcp_server/models/base.py
-11
View File
@@ -35,17 +35,6 @@ class BaseResponse(BaseModel):
return iso_string
class ErrorResponse(BaseResponse):
"""Response model for error cases."""
success: bool = Field(default=False, description="Always False for error responses")
error: str = Field(description="Error message")
error_code: Optional[str] = Field(None, description="Optional error code")
details: Optional[Dict[str, Any]] = Field(
None, description="Additional error details"
)
class SuccessResponse(BaseResponse):
"""Generic success response."""
nextcloud_mcp_server/server/notes.py
+82 -44
View File
@@ -1,10 +1,11 @@
import logging
from httpx import HTTPStatusError
from mcp.shared.exceptions import McpError
from mcp.types import ErrorData
from mcp.server.fastmcp import Context, FastMCP
from nextcloud_mcp_server.client import NextcloudClient
from nextcloud_mcp_server.models.base import ErrorResponse
from nextcloud_mcp_server.models.notes import (
Note,
NotesSettings,
@@ -54,8 +55,6 @@ def configure_notes_tools(mcp: FastMCP):
@mcp.resource("nc://Notes/{note_id}")
async def nc_get_note(note_id: int):
"""Get user note using note id"""
from mcp.shared.exceptions import McpError
from mcp.types import ErrorData
ctx: Context = mcp.get_context()
client: NextcloudClient = ctx.request_context.lifespan_context.client
@@ -80,7 +79,7 @@ def configure_notes_tools(mcp: FastMCP):
@mcp.tool()
async def nc_notes_create_note(
title: str, content: str, category: str, ctx: Context
) -> CreateNoteResponse | ErrorResponse:
) -> CreateNoteResponse:
"""Create a new note"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
try:
@@ -95,18 +94,27 @@ def configure_notes_tools(mcp: FastMCP):
)
except HTTPStatusError as e:
if e.response.status_code == 403:
return ErrorResponse(
error="Access denied: insufficient permissions to create notes"
raise McpError(
ErrorData(
code=-1,
message="Access denied: insufficient permissions to create notes",
)
)
elif e.response.status_code == 413:
return ErrorResponse(error="Note content too large")
raise McpError(ErrorData(code=-1, message="Note content too large"))
elif e.response.status_code == 409:
return ErrorResponse(
error=f"A note with title '{title}' already exists in this category"
raise McpError(
ErrorData(
code=-1,
message=f"A note with title '{title}' already exists in this category",
)
)
else:
return ErrorResponse(
error=f"Failed to create note: server error ({e.response.status_code})"
raise McpError(
ErrorData(
code=-1,
message=f"Failed to create note: server error ({e.response.status_code})",
)
)
@mcp.tool()
@@ -117,7 +125,7 @@ def configure_notes_tools(mcp: FastMCP):
content: str | None,
category: str | None,
ctx: Context,
) -> UpdateNoteResponse | ErrorResponse:
) -> UpdateNoteResponse:
"""Update an existing note's title, content, or category.
REQUIRED: etag parameter must be provided to prevent overwriting concurrent changes.
@@ -140,26 +148,37 @@ def configure_notes_tools(mcp: FastMCP):
)
except HTTPStatusError as e:
if e.response.status_code == 404:
return ErrorResponse(error=f"Note {note_id} not found")
raise McpError(ErrorData(code=-1, message=f"Note {note_id} not found"))
elif e.response.status_code == 412:
return ErrorResponse(
error=f"Note {note_id} has been modified by someone else. Please refresh and try again."
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:
return ErrorResponse(
error=f"Access denied: insufficient permissions to update note {note_id}"
raise McpError(
ErrorData(
code=-1,
message=f"Access denied: insufficient permissions to update note {note_id}",
)
)
elif e.response.status_code == 413:
return ErrorResponse(error="Updated note content is too large")
raise McpError(
ErrorData(code=-1, message="Updated note content is too large")
)
else:
return ErrorResponse(
error=f"Failed to update note {note_id}: server error ({e.response.status_code})"
raise McpError(
ErrorData(
code=-1,
message=f"Failed to update note {note_id}: server error ({e.response.status_code})",
)
)
@mcp.tool()
async def nc_notes_append_content(
note_id: int, content: str, ctx: Context
) -> AppendContentResponse | ErrorResponse:
) -> AppendContentResponse:
"""Append content to an existing note with a clear separator. The tool automatically adds separators between existing and new content - do not include separators in your content."""
logger.info("Appending content to note %s", note_id)
client: NextcloudClient = ctx.request_context.lifespan_context.client
@@ -173,24 +192,31 @@ def configure_notes_tools(mcp: FastMCP):
)
except HTTPStatusError as e:
if e.response.status_code == 404:
return ErrorResponse(error=f"Note {note_id} not found")
raise McpError(ErrorData(code=-1, message=f"Note {note_id} not found"))
elif e.response.status_code == 403:
return ErrorResponse(
error=f"Access denied: insufficient permissions to modify note {note_id}"
raise McpError(
ErrorData(
code=-1,
message=f"Access denied: insufficient permissions to modify note {note_id}",
)
)
elif e.response.status_code == 413:
return ErrorResponse(
error="Content to append would make the note too large"
raise McpError(
ErrorData(
code=-1,
message="Content to append would make the note too large",
)
)
else:
return ErrorResponse(
error=f"Failed to append content to note {note_id}: server error ({e.response.status_code})"
raise McpError(
ErrorData(
code=-1,
message=f"Failed to append content to note {note_id}: server error ({e.response.status_code})",
)
)
@mcp.tool()
async def nc_notes_search_notes(
query: str, ctx: Context
) -> SearchNotesResponse | ErrorResponse:
async def nc_notes_search_notes(query: str, ctx: Context) -> SearchNotesResponse:
"""Search notes by title or content, returning only id, title, and category."""
client: NextcloudClient = ctx.request_context.lifespan_context.client
try:
@@ -212,20 +238,26 @@ def configure_notes_tools(mcp: FastMCP):
)
except HTTPStatusError as e:
if e.response.status_code == 403:
return ErrorResponse(
error="Access denied: insufficient permissions to search notes"
raise McpError(
ErrorData(
code=-1,
message="Access denied: insufficient permissions to search notes",
)
)
elif e.response.status_code == 400:
return ErrorResponse(error="Invalid search query format")
raise McpError(
ErrorData(code=-1, message="Invalid search query format")
)
else:
return ErrorResponse(
error=f"Search failed: server error ({e.response.status_code})"
raise McpError(
ErrorData(
code=-1,
message=f"Search failed: server error ({e.response.status_code})",
)
)
@mcp.tool()
async def nc_notes_delete_note(
note_id: int, ctx: Context
) -> DeleteNoteResponse | ErrorResponse:
async def nc_notes_delete_note(note_id: int, ctx: Context) -> DeleteNoteResponse:
"""Delete a note permanently"""
logger.info("Deleting note %s", note_id)
client: NextcloudClient = ctx.request_context.lifespan_context.client
@@ -238,12 +270,18 @@ def configure_notes_tools(mcp: FastMCP):
)
except HTTPStatusError as e:
if e.response.status_code == 404:
return ErrorResponse(error=f"Note {note_id} not found")
raise McpError(ErrorData(code=-1, message=f"Note {note_id} not found"))
elif e.response.status_code == 403:
return ErrorResponse(
error=f"Access denied: insufficient permissions to delete note {note_id}"
raise McpError(
ErrorData(
code=-1,
message=f"Access denied: insufficient permissions to delete note {note_id}",
)
)
else:
return ErrorResponse(
error=f"Failed to delete note {note_id}: server error ({e.response.status_code})"
raise McpError(
ErrorData(
code=-1,
message=f"Failed to delete note {note_id}: server error ({e.response.status_code})",
)
)
nextcloud_mcp_server/server/webdav.py
+2 -2
View File
@@ -43,11 +43,11 @@ def configure_webdav_tools(mcp: FastMCP):
Examples:
# Read a text file
result = await nc_webdav_read_file("Documents/readme.txt")
print(result['content']) # Decoded text content
logger.info(result['content']) # Decoded text content
# Read a binary file
result = await nc_webdav_read_file("Images/photo.jpg")
print(result['encoding']) # 'base64'
logger.info(result['encoding']) # 'base64'
"""
client: NextcloudClient = ctx.request_context.lifespan_context.client
content, content_type = await client.webdav.read_file(path)
tests/integration/test_error_propagation.py
+8 -22
View File
@@ -20,23 +20,15 @@ async def test_missing_note_resource_error(nc_mcp_client: ClientSession):
@pytest.mark.integration
async def test_delete_missing_note_tool_error(nc_mcp_client: ClientSession):
"""Test that deleting a non-existent note returns proper error."""
# Try to delete a non-existent note
# Try to delete a non-existent note - should return error response
response = await nc_mcp_client.call_tool(
"nc_notes_delete_note", {"note_id": 999999}
)
logger.info(f"Delete missing note response: {response}")
# Should return structured error response with improved message
# Should return error response (not raise exception) for tools
assert response is not None
assert (
response.isError is False
) # Tools now return structured responses, not MCP errors
# Check structured content for error
assert "success" in response.structuredContent["result"]
assert response.structuredContent["result"]["success"] is False
assert "Note 999999 not found" in response.structuredContent["result"]["error"]
assert response.isError is True
assert "Note 999999 not found" in response.content[0].text
@pytest.mark.integration
@@ -81,7 +73,7 @@ async def test_update_note_with_invalid_etag(nc_mcp_client: ClientSession, nc_cl
note_id = note_data["id"]
try:
# Try to update with invalid ETag
# Try to update with invalid ETag - should return error response
response = await nc_mcp_client.call_tool(
"nc_notes_update_note",
{
@@ -93,16 +85,10 @@ async def test_update_note_with_invalid_etag(nc_mcp_client: ClientSession, nc_cl
},
)
logger.info(f"Invalid ETag response: {response}")
# Should return structured error response with improved message
# Should return error response (not raise exception) for tools
assert response is not None
assert response.isError is False # Tools now return structured responses
assert "success" in response.structuredContent["result"]
assert response.structuredContent["result"]["success"] is False
assert (
"modified by someone else" in response.structuredContent["result"]["error"]
)
assert response.isError is True
assert "modified by someone else" in response.content[0].text
finally:
# Clean up
tests/test_models.py
+7 -5
View File
@@ -1,12 +1,14 @@
"""Unit tests for Pydantic models and serialization."""
import json
import re
from datetime import datetime, timezone
import json
import logging
import re
from nextcloud_mcp_server.models.base import BaseResponse, SuccessResponse
logger = logging.getLogger(__name__)
def test_timestamp_format_validation():
"""Test that timestamps in BaseResponse are RFC3339 compliant for MCP validation.
@@ -76,7 +78,7 @@ def test_current_broken_format():
assert "+" not in current_format
assert "-" not in current_format[-6:] # Check last 6 chars for timezone
print(f"Current broken format: {current_format}")
print(
logger.info(f"Current broken format: {current_format}")
logger.info(
"This format causes MCP validation errors because it lacks timezone information"
)