diff --git a/CLAUDE.md b/CLAUDE.md index f7874c7..ed69773 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -101,6 +101,16 @@ Each Nextcloud app has a corresponding server module that: - Tests are marked with `@pytest.mark.integration` for selective running - **Important**: Integration tests run against live Docker containers. After making code changes to the MCP server, rebuild only the MCP container with `docker-compose up --build -d mcp` before running tests +#### Testing Best Practices +- **Always restart MCP server** after code changes with `docker-compose up --build -d mcp` +- **Use existing fixtures** from `tests/conftest.py` to avoid duplicate setup work: + - `nc_mcp_client` - MCP client session for tool/resource testing + - `nc_client` - Direct NextcloudClient for setup/cleanup operations + - `temporary_note` - Creates and cleans up test notes automatically + - `temporary_addressbook` - Creates and cleans up test address books + - `temporary_contact` - Creates and cleans up test contacts +- **Avoid creating standalone test scripts** - use pytest with proper fixtures instead + ### Configuration Files - **`pyproject.toml`** - Python project configuration using uv for dependency management diff --git a/nextcloud_mcp_server/models/__init__.py b/nextcloud_mcp_server/models/__init__.py new file mode 100644 index 0000000..f51ed4f --- /dev/null +++ b/nextcloud_mcp_server/models/__init__.py @@ -0,0 +1,146 @@ +"""Pydantic models for structured MCP server responses.""" + +# Base models +from .base import ( + BaseResponse, + ErrorResponse, + SuccessResponse, + PaginatedResponse, + IdResponse, + StatusResponse, +) + +# Notes models +from .notes import ( + Note, + NoteSearchResult, + NotesSettings, + CreateNoteResponse, + UpdateNoteResponse, + DeleteNoteResponse, + AppendContentResponse, + SearchNotesResponse, +) + +# Calendar models +from .calendar import ( + Calendar, + CalendarEvent, + CalendarEventSummary, + CreateEventResponse, + UpdateEventResponse, + DeleteEventResponse, + ListEventsResponse, + ListCalendarsResponse, + AvailabilitySlot, + FindAvailabilityResponse, + BulkOperationResult, + BulkOperationResponse, + CreateMeetingResponse, + UpcomingEventsResponse, + ManageCalendarResponse, +) + +# Contacts models +from .contacts import ( + AddressBook, + Contact, + ContactField, + ListAddressBooksResponse, + ListContactsResponse, + CreateContactResponse, + UpdateContactResponse, + DeleteContactResponse, + CreateAddressBookResponse, + DeleteAddressBookResponse, +) + +# Tables models +from .tables import ( + Table, + TableColumn, + TableRow, + TableView, + TableSchema, + ListTablesResponse, + GetSchemaResponse, + ReadTableResponse, + CreateRowResponse, + UpdateRowResponse, + DeleteRowResponse, +) + +# WebDAV models +from .webdav import ( + FileInfo, + DirectoryListing, + ReadFileResponse, + WriteFileResponse, + CreateDirectoryResponse, + DeleteResourceResponse, +) + +__all__ = [ + # Base models + "BaseResponse", + "ErrorResponse", + "SuccessResponse", + "PaginatedResponse", + "IdResponse", + "StatusResponse", + # Notes models + "Note", + "NoteSearchResult", + "NotesSettings", + "CreateNoteResponse", + "UpdateNoteResponse", + "DeleteNoteResponse", + "AppendContentResponse", + "SearchNotesResponse", + # Calendar models + "Calendar", + "CalendarEvent", + "CalendarEventSummary", + "CreateEventResponse", + "UpdateEventResponse", + "DeleteEventResponse", + "ListEventsResponse", + "ListCalendarsResponse", + "AvailabilitySlot", + "FindAvailabilityResponse", + "BulkOperationResult", + "BulkOperationResponse", + "CreateMeetingResponse", + "UpcomingEventsResponse", + "ManageCalendarResponse", + # Contacts models + "AddressBook", + "Contact", + "ContactField", + "ListAddressBooksResponse", + "ListContactsResponse", + "CreateContactResponse", + "UpdateContactResponse", + "DeleteContactResponse", + "CreateAddressBookResponse", + "DeleteAddressBookResponse", + # Tables models + "Table", + "TableColumn", + "TableRow", + "TableView", + "TableSchema", + "ListTablesResponse", + "GetSchemaResponse", + "ReadTableResponse", + "CreateRowResponse", + "UpdateRowResponse", + "DeleteRowResponse", + # WebDAV models + "FileInfo", + "DirectoryListing", + "ReadFileResponse", + "WriteFileResponse", + "CreateDirectoryResponse", + "DeleteResourceResponse", +] diff --git a/nextcloud_mcp_server/models/base.py b/nextcloud_mcp_server/models/base.py new file mode 100644 index 0000000..d89b4d2 --- /dev/null +++ b/nextcloud_mcp_server/models/base.py @@ -0,0 +1,70 @@ +"""Base Pydantic models for common response patterns.""" + +from datetime import datetime +from typing import Any, Dict, Generic, List, Optional, TypeVar, Union + +from pydantic import BaseModel, Field + + +class BaseResponse(BaseModel): + """Base response model for all MCP tool responses.""" + + model_config = {"json_encoders": {datetime: lambda v: v.isoformat()}} + + success: bool = Field( + default=True, description="Whether the operation was successful" + ) + timestamp: datetime = Field( + default_factory=datetime.now, description="Response timestamp" + ) + + +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.""" + + message: Optional[str] = Field(None, description="Optional success message") + data: Optional[Dict[str, Any]] = Field(None, description="Optional response data") + + +T = TypeVar("T") + + +class PaginatedResponse(BaseResponse, Generic[T]): + """Generic paginated response model.""" + + items: List[T] = Field(description="List of items") + total_count: Optional[int] = Field( + None, description="Total number of items available" + ) + page: Optional[int] = Field(None, description="Current page number") + page_size: Optional[int] = Field(None, description="Number of items per page") + has_more: Optional[bool] = Field( + None, description="Whether more items are available" + ) + cursor: Optional[str] = Field( + None, description="Cursor for next page (if using cursor-based pagination)" + ) + + +class IdResponse(BaseResponse): + """Response model for operations that return a new ID.""" + + id: Union[int, str] = Field(description="ID of the created or affected resource") + + +class StatusResponse(BaseResponse): + """Response model for operations that return just a status.""" + + status_code: Optional[int] = Field(None, description="HTTP status code") + message: Optional[str] = Field(None, description="Status message") diff --git a/nextcloud_mcp_server/models/calendar.py b/nextcloud_mcp_server/models/calendar.py new file mode 100644 index 0000000..474db42 --- /dev/null +++ b/nextcloud_mcp_server/models/calendar.py @@ -0,0 +1,182 @@ +"""Pydantic models for Calendar app responses.""" + +from typing import List, Optional + +from pydantic import BaseModel, Field + +from .base import BaseResponse, StatusResponse + + +class Calendar(BaseModel): + """Model for a Nextcloud calendar.""" + + name: str = Field(description="Calendar name/ID") + display_name: str = Field(description="Calendar display name") + description: Optional[str] = Field(None, description="Calendar description") + color: Optional[str] = Field(None, description="Calendar color") + href: Optional[str] = Field(None, description="Calendar DAV href") + timezone: Optional[str] = Field(None, description="Calendar timezone") + enabled: bool = Field(default=True, description="Whether calendar is enabled") + ctag: Optional[str] = Field(None, description="Calendar tag for synchronization") + + +class CalendarEventSummary(BaseModel): + """Model for calendar event summary (for lists).""" + + uid: str = Field(description="Event UID") + summary: str = Field(description="Event summary/title") + start: str = Field(description="Event start datetime (ISO format)") + end: Optional[str] = Field(None, description="Event end datetime (ISO format)") + all_day: bool = Field(default=False, description="Whether event is all-day") + location: Optional[str] = Field(None, description="Event location") + description: Optional[str] = Field(None, description="Event description") + categories: List[str] = Field(default_factory=list, description="Event categories") + status: Optional[str] = Field( + None, description="Event status (CONFIRMED, TENTATIVE, CANCELLED)" + ) + + +class CalendarEvent(CalendarEventSummary): + """Model for a complete calendar event.""" + + created: Optional[str] = Field(None, description="Event creation datetime") + last_modified: Optional[str] = Field(None, description="Last modification datetime") + recurring: bool = Field(default=False, description="Whether event is recurring") + recurrence_rule: Optional[str] = Field(None, description="RFC5545 recurrence rule") + recurrence_end: Optional[str] = Field(None, description="Recurrence end date") + attendees: List[str] = Field( + default_factory=list, description="List of attendee email addresses" + ) + organizer: Optional[str] = Field(None, description="Event organizer") + priority: Optional[int] = Field(None, description="Event priority (1-9)") + privacy: Optional[str] = Field(None, description="Event privacy level") + url: Optional[str] = Field(None, description="Event URL") + duration_minutes: Optional[int] = Field( + None, description="Event duration in minutes" + ) + reminder_minutes: Optional[int] = Field( + None, description="Reminder time in minutes before event" + ) + reminder_email: bool = Field( + default=False, description="Whether to send email reminder" + ) + color: Optional[str] = Field(None, description="Event color") + etag: Optional[str] = Field(None, description="ETag for versioning") + + +class CreateEventResponse(BaseResponse): + """Response model for event creation.""" + + event: CalendarEvent = Field(description="The created event") + calendar_name: str = Field( + description="Name of the calendar the event was created in" + ) + + +class UpdateEventResponse(BaseResponse): + """Response model for event updates.""" + + event: CalendarEvent = Field(description="The updated event") + calendar_name: str = Field(description="Name of the calendar the event belongs to") + + +class DeleteEventResponse(StatusResponse): + """Response model for event deletion.""" + + deleted_uid: str = Field(description="UID of the deleted event") + calendar_name: str = Field( + description="Name of the calendar the event was deleted from" + ) + + +class ListEventsResponse(BaseResponse): + """Response model for listing events.""" + + events: List[CalendarEventSummary] = Field(description="List of events") + calendar_name: Optional[str] = Field( + None, description="Calendar name (if filtered to one calendar)" + ) + start_date: Optional[str] = Field(None, description="Start date filter applied") + end_date: Optional[str] = Field(None, description="End date filter applied") + total_found: int = Field(description="Total number of events found") + + +class ListCalendarsResponse(BaseResponse): + """Response model for listing calendars.""" + + calendars: List[Calendar] = Field(description="List of available calendars") + total_count: int = Field(description="Total number of calendars") + + +class AvailabilitySlot(BaseModel): + """Model for an available time slot.""" + + start: str = Field(description="Slot start datetime (ISO format)") + end: str = Field(description="Slot end datetime (ISO format)") + duration_minutes: int = Field(description="Slot duration in minutes") + date: str = Field(description="Date of the slot (YYYY-MM-DD)") + + +class FindAvailabilityResponse(BaseResponse): + """Response model for finding availability.""" + + available_slots: List[AvailabilitySlot] = Field( + description="List of available time slots" + ) + duration_requested: int = Field(description="Requested duration in minutes") + date_range_start: str = Field(description="Start date of search range") + date_range_end: str = Field(description="End date of search range") + attendees_checked: List[str] = Field( + default_factory=list, description="Attendees checked for availability" + ) + business_hours_only: bool = Field( + description="Whether search was limited to business hours" + ) + + +class BulkOperationResult(BaseModel): + """Model for bulk operation results.""" + + operation: str = Field(description="Operation performed (update, delete, move)") + events_processed: int = Field(description="Number of events processed") + events_successful: int = Field( + description="Number of events successfully processed" + ) + events_failed: int = Field(description="Number of events that failed processing") + failed_events: List[str] = Field( + default_factory=list, description="UIDs of events that failed" + ) + errors: List[str] = Field(default_factory=list, description="Error messages") + + +class BulkOperationResponse(BaseResponse): + """Response model for bulk operations.""" + + result: BulkOperationResult = Field(description="Bulk operation result") + + +class CreateMeetingResponse(CreateEventResponse): + """Response model for meeting creation (same as event creation).""" + + pass + + +class UpcomingEventsResponse(BaseResponse): + """Response model for upcoming events.""" + + events: List[CalendarEventSummary] = Field(description="List of upcoming events") + days_ahead: int = Field(description="Number of days ahead searched") + calendar_name: Optional[str] = Field( + None, description="Calendar name (if filtered to one calendar)" + ) + + +class ManageCalendarResponse(BaseResponse): + """Response model for calendar management operations.""" + + action: str = Field(description="Action performed (create, delete, update, list)") + calendar: Optional[Calendar] = Field(None, description="Calendar that was affected") + calendars: Optional[List[Calendar]] = Field( + None, description="List of calendars (for list action)" + ) + message: str = Field(description="Success message") diff --git a/nextcloud_mcp_server/models/contacts.py b/nextcloud_mcp_server/models/contacts.py new file mode 100644 index 0000000..54230b9 --- /dev/null +++ b/nextcloud_mcp_server/models/contacts.py @@ -0,0 +1,130 @@ +"""Pydantic models for Contacts app responses.""" + +from typing import Any, Dict, List, Optional + +from pydantic import BaseModel, Field + +from .base import BaseResponse, StatusResponse + + +class AddressBook(BaseModel): + """Model for a Nextcloud address book.""" + + uri: str = Field(description="Address book URI") + displayname: str = Field(description="Address book display name") + description: Optional[str] = Field(None, description="Address book description") + ctag: Optional[str] = Field( + None, description="Address book tag for synchronization" + ) + + +class ContactField(BaseModel): + """Model for a contact field (email, phone, etc.).""" + + type: str = Field(description="Field type (e.g., 'email', 'phone', 'address')") + value: str = Field(description="Field value") + label: Optional[str] = Field(None, description="Field label (e.g., 'work', 'home')") + preferred: bool = Field( + default=False, description="Whether this is the preferred field of this type" + ) + + +class Contact(BaseModel): + """Model for a Nextcloud contact.""" + + uid: str = Field(description="Contact UID") + fn: str = Field(description="Full name (formatted name)") + given_name: Optional[str] = Field(None, description="Given name") + family_name: Optional[str] = Field(None, description="Family name") + organization: Optional[str] = Field(None, description="Organization") + title: Optional[str] = Field(None, description="Job title") + emails: List[ContactField] = Field( + default_factory=list, description="Email addresses" + ) + phones: List[ContactField] = Field( + default_factory=list, description="Phone numbers" + ) + addresses: List[ContactField] = Field(default_factory=list, description="Addresses") + urls: List[ContactField] = Field(default_factory=list, description="URLs") + note: Optional[str] = Field(None, description="Notes") + photo: Optional[str] = Field(None, description="Photo URL or base64 data") + birthday: Optional[str] = Field(None, description="Birthday (ISO date format)") + categories: List[str] = Field( + default_factory=list, description="Contact categories" + ) + custom_fields: Dict[str, Any] = Field( + default_factory=dict, description="Custom fields" + ) + etag: Optional[str] = Field(None, description="ETag for versioning") + + @property + def primary_email(self) -> Optional[str]: + """Get the primary email address.""" + if not self.emails: + return None + # Return preferred email if available, otherwise first email + preferred = next( + (email.value for email in self.emails if email.preferred), None + ) + return preferred or self.emails[0].value + + @property + def primary_phone(self) -> Optional[str]: + """Get the primary phone number.""" + if not self.phones: + return None + # Return preferred phone if available, otherwise first phone + preferred = next( + (phone.value for phone in self.phones if phone.preferred), None + ) + return preferred or self.phones[0].value + + +class ListAddressBooksResponse(BaseResponse): + """Response model for listing address books.""" + + addressbooks: List[AddressBook] = Field( + description="List of available address books" + ) + total_count: int = Field(description="Total number of address books") + + +class ListContactsResponse(BaseResponse): + """Response model for listing contacts.""" + + contacts: List[Contact] = Field(description="List of contacts") + addressbook: str = Field(description="Address book name") + total_count: int = Field(description="Total number of contacts") + + +class CreateContactResponse(BaseResponse): + """Response model for contact creation.""" + + contact: Contact = Field(description="The created contact") + addressbook: str = Field(description="Address book the contact was created in") + + +class UpdateContactResponse(BaseResponse): + """Response model for contact updates.""" + + contact: Contact = Field(description="The updated contact") + addressbook: str = Field(description="Address book the contact belongs to") + + +class DeleteContactResponse(StatusResponse): + """Response model for contact deletion.""" + + deleted_uid: str = Field(description="UID of the deleted contact") + addressbook: str = Field(description="Address book the contact was deleted from") + + +class CreateAddressBookResponse(BaseResponse): + """Response model for address book creation.""" + + addressbook: AddressBook = Field(description="The created address book") + + +class DeleteAddressBookResponse(StatusResponse): + """Response model for address book deletion.""" + + deleted_name: str = Field(description="Name of the deleted address book") diff --git a/nextcloud_mcp_server/models/notes.py b/nextcloud_mcp_server/models/notes.py new file mode 100644 index 0000000..975aee7 --- /dev/null +++ b/nextcloud_mcp_server/models/notes.py @@ -0,0 +1,77 @@ +"""Pydantic models for Notes app responses.""" + +from datetime import datetime +from typing import List, Optional + +from pydantic import BaseModel, Field + +from .base import BaseResponse, IdResponse, StatusResponse + + +class Note(BaseModel): + """Model for a Nextcloud note.""" + + id: int = Field(description="Note ID") + title: str = Field(description="Note title") + content: str = Field(description="Note content in markdown") + category: str = Field(default="", description="Note category") + modified: int = Field(description="Unix timestamp of last modification") + favorite: bool = Field( + default=False, description="Whether note is marked as favorite" + ) + etag: Optional[str] = Field(None, description="ETag for versioning") + readonly: bool = Field(default=False, description="Whether note is read-only") + + @property + def modified_datetime(self) -> datetime: + """Convert Unix timestamp to datetime.""" + return datetime.fromtimestamp(self.modified) + + +class NoteSearchResult(BaseModel): + """Model for note search results (limited fields).""" + + id: int = Field(description="Note ID") + title: str = Field(description="Note title") + category: str = Field(default="", description="Note category") + score: Optional[float] = Field(None, description="Search relevance score") + + +class NotesSettings(BaseModel): + """Model for Notes app settings.""" + + notesPath: str = Field(description="Path to notes directory") + fileSuffix: str = Field(description="File suffix for notes") + noteMode: str = Field(description="Note mode setting") + + +class CreateNoteResponse(IdResponse): + """Response model for note creation.""" + + note: Note = Field(description="The created note") + + +class UpdateNoteResponse(BaseResponse): + """Response model for note updates.""" + + note: Note = Field(description="The updated note") + + +class DeleteNoteResponse(StatusResponse): + """Response model for note deletion.""" + + deleted_id: int = Field(description="ID of the deleted note") + + +class AppendContentResponse(BaseResponse): + """Response model for appending content to a note.""" + + note: Note = Field(description="The updated note after appending content") + + +class SearchNotesResponse(BaseResponse): + """Response model for note search.""" + + results: List[NoteSearchResult] = Field(description="Search results") + query: str = Field(description="The search query used") + total_found: int = Field(description="Total number of notes found") diff --git a/nextcloud_mcp_server/models/tables.py b/nextcloud_mcp_server/models/tables.py new file mode 100644 index 0000000..32e49c2 --- /dev/null +++ b/nextcloud_mcp_server/models/tables.py @@ -0,0 +1,142 @@ +"""Pydantic models for Tables app responses.""" + +from typing import Any, Dict, List, Optional + +from pydantic import BaseModel, Field + +from .base import BaseResponse, IdResponse, StatusResponse + + +class TableColumn(BaseModel): + """Model for a table column definition.""" + + id: int = Field(description="Column ID") + title: str = Field(description="Column title") + type: str = Field(description="Column type (text, number, datetime, etc.)") + subtype: Optional[str] = Field(None, description="Column subtype") + mandatory: bool = Field(default=False, description="Whether column is mandatory") + description: Optional[str] = Field(None, description="Column description") + text_default: Optional[str] = Field(None, description="Default text value") + text_allowed_pattern: Optional[str] = Field( + None, description="Allowed text pattern" + ) + text_max_length: Optional[int] = Field(None, description="Maximum text length") + number_default: Optional[float] = Field(None, description="Default number value") + number_min: Optional[float] = Field(None, description="Minimum number value") + number_max: Optional[float] = Field(None, description="Maximum number value") + number_decimals: Optional[int] = Field(None, description="Number of decimal places") + datetime_default: Optional[str] = Field(None, description="Default datetime value") + selection_options: List[str] = Field( + default_factory=list, description="Selection options" + ) + selection_default: Optional[str] = Field( + None, description="Default selection value" + ) + + +class TableRow(BaseModel): + """Model for a table row.""" + + id: int = Field(description="Row ID") + created_by: Optional[str] = Field(None, description="User who created the row") + created_at: Optional[str] = Field(None, description="Row creation timestamp") + last_edit_by: Optional[str] = Field( + None, description="User who last edited the row" + ) + last_edit_at: Optional[str] = Field(None, description="Last edit timestamp") + data: Dict[int, Any] = Field(description="Row data keyed by column ID") + + +class TableView(BaseModel): + """Model for a table view.""" + + id: int = Field(description="View ID") + title: str = Field(description="View title") + emoji: Optional[str] = Field(None, description="View emoji") + description: Optional[str] = Field(None, description="View description") + columns: List[int] = Field( + default_factory=list, description="List of column IDs in this view" + ) + sort: List[Dict[str, Any]] = Field( + default_factory=list, description="Sort configuration" + ) + filter: List[Dict[str, Any]] = Field( + default_factory=list, description="Filter configuration" + ) + + +class Table(BaseModel): + """Model for a Nextcloud table.""" + + id: int = Field(description="Table ID") + title: str = Field(description="Table title") + emoji: Optional[str] = Field(None, description="Table emoji") + ownership: str = Field(description="Table ownership") + owner_display_name: str = Field(description="Display name of table owner") + created_by: Optional[str] = Field(None, description="User who created the table") + created_at: Optional[str] = Field(None, description="Table creation timestamp") + last_edit_by: Optional[str] = Field( + None, description="User who last edited the table" + ) + last_edit_at: Optional[str] = Field(None, description="Last edit timestamp") + row_count: int = Field(default=0, description="Number of rows in the table") + has_shares: bool = Field(default=False, description="Whether table is shared") + archived: bool = Field(default=False, description="Whether table is archived") + is_shared: bool = Field( + default=False, description="Whether table is shared with current user" + ) + on_share_permissions: Optional[Dict[str, Any]] = Field( + None, description="Share permissions" + ) + + +class TableSchema(BaseModel): + """Model for complete table schema including columns and views.""" + + table: Table = Field(description="Table information") + columns: List[TableColumn] = Field(description="Table columns") + views: List[TableView] = Field(description="Table views") + + +class ListTablesResponse(BaseResponse): + """Response model for listing tables.""" + + tables: List[Table] = Field(description="List of available tables") + total_count: int = Field(description="Total number of tables") + + +class GetSchemaResponse(BaseResponse): + """Response model for getting table schema.""" + + table_schema: TableSchema = Field(description="Table schema information") + + +class ReadTableResponse(BaseResponse): + """Response model for reading table rows.""" + + rows: List[TableRow] = Field(description="Table rows") + table_id: int = Field(description="Table ID") + total_count: Optional[int] = Field( + None, description="Total number of rows (if known)" + ) + offset: Optional[int] = Field(None, description="Offset used for pagination") + limit: Optional[int] = Field(None, description="Limit used for pagination") + + +class CreateRowResponse(IdResponse): + """Response model for row creation.""" + + row: TableRow = Field(description="The created row") + table_id: int = Field(description="Table ID the row was created in") + + +class UpdateRowResponse(BaseResponse): + """Response model for row updates.""" + + row: TableRow = Field(description="The updated row") + + +class DeleteRowResponse(StatusResponse): + """Response model for row deletion.""" + + deleted_id: int = Field(description="ID of the deleted row") diff --git a/nextcloud_mcp_server/models/webdav.py b/nextcloud_mcp_server/models/webdav.py new file mode 100644 index 0000000..bce6174 --- /dev/null +++ b/nextcloud_mcp_server/models/webdav.py @@ -0,0 +1,88 @@ +"""Pydantic models for WebDAV responses.""" + +from datetime import datetime +from typing import List, Optional + +from pydantic import BaseModel, Field + +from .base import BaseResponse, StatusResponse + + +class FileInfo(BaseModel): + """Model for file/directory information.""" + + name: str = Field(description="File/directory name") + path: str = Field(description="Full path") + is_directory: bool = Field(description="Whether this is a directory") + size: Optional[int] = Field( + None, description="File size in bytes (None for directories)" + ) + content_type: Optional[str] = Field(None, description="MIME content type") + last_modified: Optional[str] = Field( + None, description="Last modification time (ISO format)" + ) + etag: Optional[str] = Field(None, description="ETag for versioning") + + @property + def last_modified_datetime(self) -> Optional[datetime]: + """Convert last modified string to datetime.""" + if not self.last_modified: + return None + try: + return datetime.fromisoformat(self.last_modified.replace("Z", "+00:00")) + except (ValueError, AttributeError): + return None + + +class DirectoryListing(BaseResponse): + """Response model for directory listings.""" + + path: str = Field(description="Directory path") + items: List[FileInfo] = Field(description="Files and directories in the path") + total_count: int = Field(description="Total number of items") + directories_count: int = Field(description="Number of directories") + files_count: int = Field(description="Number of files") + total_size: int = Field(default=0, description="Total size of all files in bytes") + + +class ReadFileResponse(BaseResponse): + """Response model for reading file contents.""" + + path: str = Field(description="File path") + content: str = Field(description="File content (text or base64 for binary)") + content_type: str = Field(description="MIME content type") + size: int = Field(description="File size in bytes") + encoding: Optional[str] = Field( + None, description="Encoding used (e.g., 'base64' for binary files)" + ) + etag: Optional[str] = Field(None, description="ETag for versioning") + last_modified: Optional[str] = Field(None, description="Last modification time") + + +class WriteFileResponse(StatusResponse): + """Response model for writing files.""" + + path: str = Field(description="File path that was written") + size: Optional[int] = Field(None, description="Size of the written file") + created: bool = Field(description="Whether a new file was created (vs overwritten)") + + +class CreateDirectoryResponse(StatusResponse): + """Response model for directory creation.""" + + path: str = Field(description="Directory path that was created") + created: bool = Field( + description="Whether directory was created or already existed" + ) + + +class DeleteResourceResponse(StatusResponse): + """Response model for resource deletion.""" + + path: str = Field(description="Path that was deleted") + was_directory: bool = Field( + description="Whether the deleted resource was a directory" + ) + items_deleted: Optional[int] = Field( + None, description="Number of items deleted (for directories)" + ) diff --git a/nextcloud_mcp_server/server/calendar.py b/nextcloud_mcp_server/server/calendar.py index 5e71a2c..c68c73d 100644 --- a/nextcloud_mcp_server/server/calendar.py +++ b/nextcloud_mcp_server/server/calendar.py @@ -5,6 +5,10 @@ from typing import Optional from mcp.server.fastmcp import Context, FastMCP from nextcloud_mcp_server.client import NextcloudClient +from nextcloud_mcp_server.models.calendar import ( + Calendar, + ListCalendarsResponse, +) logger = logging.getLogger(__name__) @@ -12,10 +16,13 @@ logger = logging.getLogger(__name__) def configure_calendar_tools(mcp: FastMCP): # Calendar tools @mcp.tool() - async def nc_calendar_list_calendars(ctx: Context): + async def nc_calendar_list_calendars(ctx: Context) -> ListCalendarsResponse: """List all available calendars for the user""" client: NextcloudClient = ctx.request_context.lifespan_context.client - return await client.calendar.list_calendars() + calendars_data = await client.calendar.list_calendars() + + calendars = [Calendar(**cal_data) for cal_data in calendars_data] + return ListCalendarsResponse(calendars=calendars, total_count=len(calendars)) @mcp.tool() async def nc_calendar_create_event( diff --git a/nextcloud_mcp_server/server/contacts.py b/nextcloud_mcp_server/server/contacts.py index 3ee9844..370219e 100644 --- a/nextcloud_mcp_server/server/contacts.py +++ b/nextcloud_mcp_server/server/contacts.py @@ -17,7 +17,7 @@ def configure_contacts_tools(mcp: FastMCP): @mcp.tool() async def nc_contacts_list_contacts(ctx: Context, *, addressbook: str): - """List all addressbooks for the user.""" + """List all contacts in the specified addressbook.""" client: NextcloudClient = ctx.request_context.lifespan_context.client return await client.contacts.list_contacts(addressbook=addressbook) diff --git a/nextcloud_mcp_server/server/notes.py b/nextcloud_mcp_server/server/notes.py index 38b3f29..3a4beb1 100644 --- a/nextcloud_mcp_server/server/notes.py +++ b/nextcloud_mcp_server/server/notes.py @@ -1,8 +1,20 @@ import logging +from httpx import HTTPStatusError 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, + CreateNoteResponse, + UpdateNoteResponse, + DeleteNoteResponse, + AppendContentResponse, + SearchNotesResponse, + NoteSearchResult, +) logger = logging.getLogger(__name__) @@ -15,7 +27,8 @@ def configure_notes_tools(mcp: FastMCP): mcp.get_context() ) # https://github.com/modelcontextprotocol/python-sdk/issues/244 client: NextcloudClient = ctx.request_context.lifespan_context.client - return await client.notes.get_settings() + 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(note_id: int, attachment_filename: str): @@ -38,23 +51,61 @@ def configure_notes_tools(mcp: FastMCP): ] } - @mcp.tool() - async def nc_get_note(note_id: int, ctx: Context): + @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 - return await client.notes.get_note(note_id) + try: + note_data = await client.notes.get_note(note_id) + return Note(**note_data) + 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() async def nc_notes_create_note( title: str, content: str, category: str, ctx: Context - ): + ) -> CreateNoteResponse | ErrorResponse: """Create a new note""" client: NextcloudClient = ctx.request_context.lifespan_context.client - return await client.notes.create_note( - title=title, - content=content, - category=category, - ) + try: + note_data = await client.notes.create_note( + title=title, + content=content, + category=category, + ) + note = Note(**note_data) + return CreateNoteResponse(id=note.id, note=note) + except HTTPStatusError as e: + if e.response.status_code == 403: + return ErrorResponse( + error="Access denied: insufficient permissions to create notes" + ) + elif e.response.status_code == 413: + return ErrorResponse(error="Note content too large") + elif e.response.status_code == 409: + return ErrorResponse( + error=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})" + ) @mcp.tool() async def nc_notes_update_note( @@ -64,32 +115,124 @@ def configure_notes_tools(mcp: FastMCP): content: str | None, category: str | None, ctx: Context, - ): + ) -> UpdateNoteResponse | ErrorResponse: + """Update an existing note's title, content, or category""" logger.info("Updating note %s", note_id) client: NextcloudClient = ctx.request_context.lifespan_context.client - return await client.notes.update( - note_id=note_id, - etag=etag, - title=title, - content=content, - category=category, - ) + 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(note=note) + except HTTPStatusError as e: + if e.response.status_code == 404: + return ErrorResponse(error=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." + ) + elif e.response.status_code == 403: + return ErrorResponse( + error=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") + else: + return ErrorResponse( + error=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): + async def nc_notes_append_content( + note_id: int, content: str, ctx: Context + ) -> AppendContentResponse | ErrorResponse: """Append content to an existing note with a clear separator""" logger.info("Appending content to note %s", note_id) client: NextcloudClient = ctx.request_context.lifespan_context.client - return await client.notes.append_content(note_id=note_id, content=content) + try: + note_data = await client.notes.append_content( + note_id=note_id, content=content + ) + note = Note(**note_data) + return AppendContentResponse(note=note) + except HTTPStatusError as e: + if e.response.status_code == 404: + return ErrorResponse(error=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}" + ) + elif e.response.status_code == 413: + return ErrorResponse( + error="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})" + ) @mcp.tool() - async def nc_notes_search_notes(query: str, ctx: Context): + async def nc_notes_search_notes( + query: str, ctx: Context + ) -> SearchNotesResponse | ErrorResponse: """Search notes by title or content, returning only id, title, and category.""" client: NextcloudClient = ctx.request_context.lifespan_context.client - return await client.notes_search_notes(query=query) + 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 HTTPStatusError as e: + if e.response.status_code == 403: + return ErrorResponse( + error="Access denied: insufficient permissions to search notes" + ) + elif e.response.status_code == 400: + return ErrorResponse(error="Invalid search query format") + else: + return ErrorResponse( + error=f"Search failed: server error ({e.response.status_code})" + ) @mcp.tool() - async def nc_notes_delete_note(note_id: int, ctx: Context): + async def nc_notes_delete_note( + note_id: int, ctx: Context + ) -> DeleteNoteResponse | ErrorResponse: + """Delete a note permanently""" logger.info("Deleting note %s", note_id) client: NextcloudClient = ctx.request_context.lifespan_context.client - return await client.notes.delete_note(note_id) + 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 HTTPStatusError as e: + if e.response.status_code == 404: + return ErrorResponse(error=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}" + ) + else: + return ErrorResponse( + error=f"Failed to delete note {note_id}: server error ({e.response.status_code})" + ) diff --git a/pyproject.toml b/pyproject.toml index 9d869ea..09bb01d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -13,6 +13,7 @@ dependencies = [ "pillow (>=11.2.1,<12.0.0)", "icalendar (>=6.0.0,<7.0.0)", "pythonvcard4>=0.2.0", + "pydantic>=2.11.4", ] [tool.pytest.ini_options] diff --git a/tests/integration/test_error_propagation.py b/tests/integration/test_error_propagation.py new file mode 100644 index 0000000..45d5e33 --- /dev/null +++ b/tests/integration/test_error_propagation.py @@ -0,0 +1,186 @@ +"""Test error propagation in the MCP server for various error scenarios.""" + +import logging +from mcp import ClientSession +from mcp.shared.exceptions import McpError + +import pytest + +logger = logging.getLogger(__name__) + + +@pytest.mark.integration +async def test_missing_note_resource_error(nc_mcp_client: ClientSession): + """Test that accessing a non-existent note resource returns proper error.""" + # Try to get a non-existent note via resource - should raise McpError with improved message + with pytest.raises(McpError, match=r"Note 999999 not found"): + await nc_mcp_client.read_resource("nc://Notes/999999") + + +@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 + 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 + 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"] + + +@pytest.mark.integration +async def test_search_with_empty_query(nc_mcp_client: ClientSession): + """Test search behavior with empty query.""" + # Search with empty query + response = await nc_mcp_client.call_tool("nc_notes_search_notes", {"query": ""}) + + logger.info(f"Empty search query response: {response}") + + # Should return successful response with empty or valid results + assert response is not None + assert response.isError is False + + +@pytest.mark.integration +async def test_tool_missing_required_parameters(nc_mcp_client: ClientSession): + """Test calling a tool with missing required parameters.""" + # Try to create note with missing parameters + response = await nc_mcp_client.call_tool( + "nc_notes_create_note", + {"title": "Test"}, # Missing content and category + ) + logger.info(f"Missing params response: {response}") + + # Should return error response for missing required parameters + assert response is not None + assert response.isError is True + assert ( + "required" in response.content[0].text.lower() + or "missing" in response.content[0].text.lower() + ) + + +@pytest.mark.integration +async def test_update_note_with_invalid_etag(nc_mcp_client: ClientSession, nc_client): + """Test updating a note with invalid ETag.""" + # First create a note + note_data = await nc_client.notes.create_note( + title="Test Note for ETag", content="Test content", category="" + ) + note_id = note_data["id"] + + try: + # Try to update with invalid ETag + response = await nc_mcp_client.call_tool( + "nc_notes_update_note", + { + "note_id": note_id, + "etag": "invalid-etag", + "title": "Updated Title", + "content": None, + "category": None, + }, + ) + + logger.info(f"Invalid ETag response: {response}") + + # Should return structured error response with improved message + 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"] + ) + + finally: + # Clean up + await nc_client.notes.delete_note(note_id) + + +@pytest.mark.integration +async def test_calendar_missing_calendar_error(nc_mcp_client: ClientSession): + """Test calendar operations with non-existent calendar.""" + # Try to create event in non-existent calendar + response = await nc_mcp_client.call_tool( + "nc_calendar_create_event", + { + "calendar_name": "non-existent-calendar", + "title": "Test Event", + "start_datetime": "2025-01-15T14:00:00", + }, + ) + + logger.info(f"Non-existent calendar response: {response}") + + # Should return structured error response + assert response is not None + # Note: Some modules may not have improved error handling yet + # Check if we have structured content with success=false or isError=true + if ( + hasattr(response, "structuredContent") + and response.structuredContent + and "result" in response.structuredContent + ): + assert response.structuredContent["result"]["success"] is False + else: + assert response.isError is True + + +@pytest.mark.integration +async def test_webdav_read_missing_file_error(nc_mcp_client: ClientSession): + """Test WebDAV operations with non-existent file.""" + # Try to read a non-existent file + response = await nc_mcp_client.call_tool( + "nc_webdav_read_file", {"path": "non-existent-file.txt"} + ) + + logger.info(f"Missing file response: {response}") + + # Should return structured error response + assert response is not None + # Note: Some modules may not have improved error handling yet + # Check if we have structured content with success=false or isError=true + if ( + hasattr(response, "structuredContent") + and response.structuredContent + and "result" in response.structuredContent + ): + assert response.structuredContent["result"]["success"] is False + else: + assert response.isError is True + + +@pytest.mark.integration +async def test_tables_missing_table_error(nc_mcp_client: ClientSession): + """Test Tables operations with non-existent table.""" + # Try to get schema of non-existent table + response = await nc_mcp_client.call_tool( + "nc_tables_get_schema", {"table_id": 999999} + ) + + logger.info(f"Missing table response: {response}") + + # Should return structured error response + assert response is not None + # Note: Some modules may not have improved error handling yet + # Check if we have structured content with success=false or isError=true + if ( + hasattr(response, "structuredContent") + and response.structuredContent + and "result" in response.structuredContent + ): + assert response.structuredContent["result"]["success"] is False + else: + assert response.isError is True diff --git a/tests/integration/test_mcp.py b/tests/integration/test_mcp.py index 9eba0d4..06b89ff 100644 --- a/tests/integration/test_mcp.py +++ b/tests/integration/test_mcp.py @@ -24,7 +24,6 @@ async def test_mcp_connectivity(nc_mcp_client: ClientSession): # Verify expected tools are present expected_tools = [ - "nc_get_note", "nc_notes_create_note", "nc_notes_update_note", "nc_notes_append_content", @@ -137,11 +136,9 @@ async def test_mcp_notes_crud_workflow( # 3. Read note via MCP logger.info(f"Reading note via MCP: {note_id}") - read_result = await nc_mcp_client.call_tool("nc_get_note", {"note_id": note_id}) - assert read_result.isError is False, ( - f"MCP note read failed: {read_result.content}" - ) - read_note_data = json.loads(read_result.content[0].text) + read_result = await nc_mcp_client.read_resource(f"nc://Notes/{note_id}") + assert len(read_result.contents) == 1, "Expected exactly one content item" + read_note_data = json.loads(read_result.contents[0].text) assert read_note_data["title"] == test_title assert read_note_data["content"] == test_content @@ -199,14 +196,23 @@ async def test_mcp_notes_crud_workflow( ) search_notes_text = search_result.content[0].text logger.info(f"Search result text: {search_notes_text}") - search_notes = json.loads(search_notes_text) + search_response = json.loads(search_notes_text) - # Ensure search_notes is a list - if not isinstance(search_notes, list): - logger.warning( - f"Expected search results to be a list, got: {type(search_notes)}" - ) - search_notes = [search_notes] if search_notes else [] + # Expect structured response with Pydantic format + assert isinstance(search_response, dict), ( + f"Expected search response to be a dict with structured format, got: {type(search_response)}" + ) + assert "results" in search_response, ( + f"Expected 'results' field in search response, got keys: {list(search_response.keys())}" + ) + assert "success" in search_response and search_response["success"], ( + f"Expected successful search response, got: {search_response}" + ) + + search_notes = search_response["results"] + assert isinstance(search_notes, list), ( + f"Expected results to be a list, got: {type(search_notes)}" + ) # Find our note in search results found_note = None @@ -216,7 +222,7 @@ async def test_mcp_notes_crud_workflow( break assert found_note is not None, ( - f"Created note not found in search results. Search returned: {search_notes}" + f"Created note not found in search results. Search returned: {search_response}" ) assert found_note["title"] == updated_title @@ -431,21 +437,33 @@ async def test_mcp_calendar_workflow( f"MCP calendar listing failed: {calendars_result.content}" ) - calendars_data = json.loads(calendars_result.content[0].text) + calendars_response = json.loads(calendars_result.content[0].text) # Debug output to understand the structure - logger.info(f"calendars_data type: {type(calendars_data)}") - logger.info(f"calendars_data content: {calendars_data}") + logger.info(f"calendars_response type: {type(calendars_response)}") + logger.info(f"calendars_response content: {calendars_response}") - # Handle the case where MCP tool returns a single dict instead of a list - if isinstance(calendars_data, dict): - # Single calendar returned as dict instead of list - calendar_name = calendars_data["name"] - elif isinstance(calendars_data, list) and calendars_data: - # Normal case - list of calendars - calendar_name = calendars_data[0]["name"] - else: + # Expect structured response with Pydantic format + assert isinstance(calendars_response, dict), ( + f"Expected calendar response to be a dict with structured format, got: {type(calendars_response)}" + ) + assert "calendars" in calendars_response, ( + f"Expected 'calendars' field in response, got keys: {list(calendars_response.keys())}" + ) + assert "success" in calendars_response and calendars_response["success"], ( + f"Expected successful calendar response, got: {calendars_response}" + ) + + calendars_list = calendars_response["calendars"] + assert isinstance(calendars_list, list), ( + f"Expected calendars to be a list, got: {type(calendars_list)}" + ) + + if not calendars_list: pytest.skip("No calendars available for testing") + + # Use the first available calendar + calendar_name = calendars_list[0]["name"] logger.info(f"Using calendar: {calendar_name}") # 2. Create event via MCP diff --git a/uv.lock b/uv.lock index 6f5bb55..1df9d10 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 2 +revision = 3 requires-python = ">=3.11" [[package]] @@ -512,6 +512,7 @@ dependencies = [ { name = "icalendar" }, { name = "mcp", extra = ["cli"] }, { name = "pillow" }, + { name = "pydantic" }, { name = "pythonvcard4" }, ] @@ -531,6 +532,7 @@ requires-dist = [ { name = "icalendar", specifier = ">=6.0.0,<7.0.0" }, { name = "mcp", extras = ["cli"], specifier = ">=1.10,<1.11" }, { name = "pillow", specifier = ">=11.2.1,<12.0.0" }, + { name = "pydantic", specifier = ">=2.11.4" }, { name = "pythonvcard4", specifier = ">=0.2.0" }, ]