bump: version 0.5.0 → 0.6.0

feat(calendar): add comprehensive Calendar app support via CalDAV protocol

.github/workflows/bump-version.yml

@@ -0,0 +1,32 @@
+name: Bump version
+
+on:
+  push:
+    branches:
+      - master
+
+jobs:
+  bump-version:
+    if: "!startsWith(github.event.head_commit.message, 'bump:')"
+    runs-on: ubuntu-latest
+    name: "Bump version and create changelog with commitizen"
+    permissions:
+      contents: write
+      packages: write
+    steps:
+      - name: Check out
+        uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4
+        with:
+          fetch-depth: 0
+          token: "${{ secrets.PERSONAL_ACCESS_TOKEN }}"
+      - name: Create bump and changelog
+        uses: commitizen-tools/commitizen-action@5b0848cd060263e24602d1eba03710e056ef7711 # 0.24.0
+        with:
+          github_token: ${{ secrets.PERSONAL_ACCESS_TOKEN }}
+          changelog_increment_filename: body.md
+      - name: Release
+        uses: softprops/action-gh-release@72f2c25fcb47643c292f7107632f7a47c1df5cd8 # v2
+        with:
+          body_path: "body.md"
+          tag_name: v${{ env.REVISION }}
+          token: ${{ secrets.GITHUB_TOKEN }}

.github/workflows/docker-build-publish.yml

@@ -2,7 +2,6 @@ name: Build and Publish Docker Image
 
 on:
   push:
-    branches: [ "master" ]
     tags: ["*"]
 
 jobs:
@@ -11,7 +10,6 @@ jobs:
     permissions:
       contents: read
       packages: write
-
     steps:
       - name: Checkout repository
         uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4
@@ -22,7 +20,6 @@ jobs:
         with:
           # list of Docker images to use as base name for tags
           images: |
-            #cbcoutinho/nextcloud-mcp-server
             ghcr.io/cbcoutinho/nextcloud-mcp-server
           # generate Docker tags based on the following events/attributes
           tags: |
@@ -36,7 +33,7 @@ jobs:
             type=raw,value=latest,enable={{is_default_branch}}
 
       - name: Set up Docker Buildx
-        uses: docker/setup-buildx-action@b5ca514318bd6ebac0fb2aedd5d36ec1b5c232a2 # v3
+        uses: docker/setup-buildx-action@e468171a9de216ec08956ac3ada2f0791b6bd435 # v3
 
       - name: Log in to GitHub Container Registry
         if: github.event_name != 'pull_request'
@@ -47,7 +44,7 @@ jobs:
           password: ${{ secrets.GITHUB_TOKEN }}
 
       - name: Build and push Docker image
-        uses: docker/build-push-action@14487ce63c7a62a4a324b0bfb37086795e31c6c1 # v6
+        uses: docker/build-push-action@263435318d21b8e681c14492fe198d362a7d2c83 # v6
         with:
           push: ${{ github.event_name != 'pull_request' }}
           tags: ${{ steps.meta.outputs.tags }}

.github/workflows/test.yml

@@ -6,16 +6,32 @@ on:
       - master
 
 jobs:
-  build:
+  linting:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
+      - name: Install the latest version of uv
+        uses: astral-sh/setup-uv@e92bafb6253dcd438e0484186d7669ea7a8ca1cc # v6
+      - name: Check format
+        run: |
+          uv run --frozen ruff format --diff
+      - name: Linting
+        run: |
+          uv run --frozen ruff check
+
+
+  integration-test:
     runs-on: ubuntu-latest
 
     steps:
       - uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683 # v4.2.2
 
       - name: Run docker compose
-        uses: hoverkraft-tech/compose-action@8be2d741e891ac9b8ac20825e6f3904149599925 # v2.2.0
+        uses: hoverkraft-tech/compose-action@40041ff1b97dbf152cd2361138c2b03fa29139df # v2.3.0
         with:
           compose-file: "./docker-compose.yml"
+      - name: Install the latest version of uv
+        uses: astral-sh/setup-uv@e92bafb6253dcd438e0484186d7669ea7a8ca1cc # v6
 
       - name: Wait for service to be ready
         run: |
@@ -33,18 +49,6 @@ jobs:
           done
           echo "Service is ready (returned 401)."
 
-      - name: Install notes app
-        run: |
-          docker compose exec app php occ app:enable notes
-
-      - name: Install dependencies
-        run: |
-          sudo apt update -y && sudo apt install -y pipx
-          pipx install poetry
-          poetry install
-        env:
-          DEBIAN_FRONTEND: "noninteractive"
-
       # Add subsequent steps here, e.g., running tests
       - name: Run tests
         env:
@@ -52,4 +56,4 @@ jobs:
           NEXTCLOUD_USERNAME: "admin"
           NEXTCLOUD_PASSWORD: "admin"
         run: |
-          poetry run python -m pytest
+          uv run --frozen python -m pytest

.gitignore

@@ -1 +1,6 @@
 __pycache__/
+.coverage
+.env
+*.env
+.env.local
+.env.*.local

.pre-commit-config.yaml

@@ -0,0 +1,13 @@
+repos:
+- repo: https://github.com/commitizen-tools/commitizen
+  rev: v4.8.2
+  hooks:
+  - id: commitizen
+  - id: commitizen-branch
+    stages:
+    - pre-push
+- repo: https://github.com/astral-sh/ruff-pre-commit
+  rev: v0.12.2
+  hooks:
+    - id: ruff-check
+    - id: ruff-format

CHANGELOG.md

@@ -0,0 +1,102 @@
+## v0.6.0 (2025-07-29)
+
+### Feat
+
+- **calendar**: add comprehensive Calendar app support via CalDAV protocol
+
+### Fix
+
+- apply ruff formatting to pass CI checks
+- **calendar**: address PR feedback from maintainer
+
+### Refactor
+
+- **calendar**: optimize logging for production readiness
+
+## v0.5.0 (2025-07-26)
+
+### Feat
+
+- Update webdav client create_directory method to handle recursive directories
+- **webdav**: add complete file system support
+
+### Fix
+
+- apply ruff formatting to test_webdav_operations.py
+
+## v0.4.1 (2025-07-10)
+
+### Fix
+
+- **deps**: update dependency mcp to >=1.10,<1.11
+
+## v0.4.0 (2025-07-06)
+
+### Feat
+
+- Add TablesClient and associated tools
+
+### Fix
+
+- update tests
+
+### Refactor
+
+- Modularize NC and Notes app client
+
+## v0.3.0 (2025-06-06)
+
+### Feat
+
+- Switch to using async client
+
+## v0.2.5 (2025-05-25)
+
+### Fix
+
+- Commitizen release process
+
+## v0.2.4 (2025-05-25)
+
+### Fix
+
+- Do not update dependencies when running in Dockerfile
+- Configure logging
+
+## v0.2.3 (2025-05-25)
+
+### Fix
+
+- Limit search results to notes with score > 0.5
+
+## v0.2.2 (2025-05-24)
+
+### Fix
+
+- Install deps before checking service
+
+## v0.2.1 (2025-05-24)
+
+### Fix
+
+- Install deps before checking service
+
+## v0.2.1 (2025-05-24)
+
+## v0.2.0 (2025-05-24)
+
+### Feat
+
+- **notes**: Add append to note functionality
+
+### Fix
+
+- **deps**: update dependency mcp to >=1.9,<1.10
+
+## v0.1.3 (2025-05-16)
+
+## v0.1.2 (2025-05-05)
+
+## v0.1.1 (2025-05-05)
+
+## v0.1.0 (2025-05-05)

Dockerfile

@@ -1,13 +1,9 @@
-FROM ghcr.io/astral-sh/uv:python3.11-alpine@sha256:c77e10ca22ef1021e1cafcbaee9595b5f9d8d9f2b1fe4cc7e908b981bab73ee7
+FROM ghcr.io/astral-sh/uv:0.8.3-python3.11-alpine@sha256:886c19178558b951bbb9cb242deb94e7e37f9cba5d0dc018cd210ccd6b5116db
 
 WORKDIR /app
 
 COPY . .
 
-RUN uv sync --locked
+RUN uv sync --locked --no-dev
 
-ENV VIRTUAL_ENV=/app/.venv
-ENV PATH="$VIRTUAL_ENV/bin:$PATH"
-ENV FASTMCP_LOG_LEVEL=DEBUG
-
-CMD ["mcp", "run", "--transport", "sse", "nextcloud_mcp_server/server.py:mcp"]
+CMD ["/app/.venv/bin/mcp", "run", "--transport", "sse", "/app/nextcloud_mcp_server/server.py:mcp"]

README.md

@@ -6,20 +6,205 @@ The Nextcloud MCP (Model Context Protocol) server allows Large Language Models (
 
 ## Features
 
-Currently, the server primarily interacts with the Nextcloud Notes API, providing tools and resources to manage notes.
+The server provides integration with multiple Nextcloud apps, enabling LLMs to interact with your Nextcloud data through a rich set of tools and resources.
 
-### Available Tools
+## Supported Nextcloud Apps
 
-*   `nc_notes_create_note`: Create a new note.
-*   `nc_notes_update_note`: Update an existing note by ID.
-*   `nc_notes_delete_note`: Delete a note by ID.
+| App | Support Status | Description |
+|-----|----------------|-------------|
+| **Notes** | ✅ Full Support | Create, read, update, delete, and search notes. Handle attachments via WebDAV. |
+| **Calendar** | ✅ Full Support | Complete calendar integration - create, update, delete events. Support for recurring events, reminders, attendees, and all-day events via CalDAV. |
+| **Tables** | ⚠️ Row Operations | Read table schemas and perform CRUD operations on table rows. Table management not yet supported. |
+| **Files (WebDAV)** | ✅ Full Support | Complete file system access - browse directories, read/write files, create/delete resources. |
 
-### Available Resources
+## Available Tools
 
-*   `notes://{note_id}`: Access a specific note by its ID.
-*   `notes://all`: Access all notes.
-*   `notes://settings`: Access note settings.
-*   `nc://capabilities`: Access Nextcloud server capabilities.
+### Notes Tools
+
+| Tool | Description |
+|------|-------------|
+| `nc_get_note` | Get a specific note by ID |
+| `nc_notes_create_note` | Create a new note with title, content, and category |
+| `nc_notes_update_note` | Update an existing note by ID |
+| `nc_notes_append_content` | Append content to an existing note with a clear separator |
+| `nc_notes_delete_note` | Delete a note by ID |
+| `nc_notes_search_notes` | Search notes by title or content |
+
+### Calendar Tools
+
+| Tool | Description |
+|------|-------------|
+| `nc_calendar_list_calendars` | List all available calendars for the user |
+| `nc_calendar_create_event` | Create a comprehensive calendar event with full feature support (recurring, reminders, attendees, etc.) |
+| `nc_calendar_list_events` | **Enhanced:** List events with advanced filtering (min attendees, duration, categories, status, search across all calendars) |
+| `nc_calendar_get_event` | Get detailed information about a specific event |
+| `nc_calendar_update_event` | Update any aspect of an existing event |
+| `nc_calendar_delete_event` | Delete a calendar event |
+| `nc_calendar_create_meeting` | Quick meeting creation with smart defaults |
+| `nc_calendar_get_upcoming_events` | Get upcoming events in the next N days |
+| `nc_calendar_find_availability` | **New:** Intelligent availability finder - find free time slots for meetings with attendee conflict detection |
+| `nc_calendar_bulk_operations` | **New:** Bulk update, delete, or move events matching filter criteria |
+| `nc_calendar_manage_calendar` | **New:** Create, delete, and manage calendar properties |
+
+### Tables Tools
+
+| Tool | Description |
+|------|-------------|
+| `nc_tables_list_tables` | List all tables available to the user |
+| `nc_tables_get_schema` | Get the schema/structure of a specific table including columns and views |
+| `nc_tables_read_table` | Read rows from a table with optional pagination |
+| `nc_tables_insert_row` | Insert a new row into a table |
+| `nc_tables_update_row` | Update an existing row in a table |
+| `nc_tables_delete_row` | Delete a row from a table |
+
+### WebDAV File System Tools
+
+| Tool | Description |
+|------|-------------|
+| `nc_webdav_list_directory` | List files and directories in any NextCloud path |
+| `nc_webdav_read_file` | Read file content (text files decoded, binary as base64) |
+| `nc_webdav_write_file` | Create or update files in NextCloud |
+| `nc_webdav_create_directory` | Create new directories |
+| `nc_webdav_delete_resource` | Delete files or directories |
+
+## Available Resources
+
+| Resource | Description |
+|----------|-------------|
+| `nc://capabilities` | Access Nextcloud server capabilities |
+| `notes://settings` | Access Notes app settings |
+| `nc://Notes/{note_id}/attachments/{attachment_filename}` | Access attachments for notes |
+
+### WebDAV File System Access
+
+The server provides complete file system access to your NextCloud instance, enabling you to:
+
+- Browse any directory structure
+- Read and write files of any type
+- Create and delete directories
+- Manage your NextCloud files directly through LLM interactions
+
+**Usage Examples:**
+
+```python
+# List files in root directory
+await nc_webdav_list_directory("")
+
+# Browse a specific folder
+await nc_webdav_list_directory("Documents/Projects")
+
+# Read a text file
+content = await nc_webdav_read_file("Documents/readme.txt")
+
+# Create a new directory
+await nc_webdav_create_directory("NewProject/docs")
+
+# Write content to a file
+await nc_webdav_write_file("NewProject/docs/notes.md", "# My Notes\n\nContent here...")
+
+# Delete a file or directory
+await nc_webdav_delete_resource("old_file.txt")
+```
+
+### Calendar Integration
+
+The server provides comprehensive calendar integration through CalDAV, enabling you to:
+
+- List all available calendars
+- Create, read, update, and delete calendar events  
+- Handle recurring events with RRULE support
+- Manage event reminders and notifications
+- Support all-day and timed events
+- Handle attendees and meeting invitations
+- Organize events with categories and priorities
+
+**Usage Examples:**
+
+```python
+# List available calendars
+calendars = await nc_calendar_list_calendars()
+
+# Create a simple event
+await nc_calendar_create_event(
+    calendar_name="personal",
+    title="Team Meeting", 
+    start_datetime="2025-07-28T14:00:00",
+    end_datetime="2025-07-28T15:00:00",
+    description="Weekly team sync",
+    location="Conference Room A"
+)
+
+# Create a recurring weekly meeting
+await nc_calendar_create_event(
+    calendar_name="work",
+    title="Weekly Standup",
+    start_datetime="2025-07-28T09:00:00", 
+    end_datetime="2025-07-28T09:30:00",
+    recurring=True,
+    recurrence_rule="FREQ=WEEKLY;BYDAY=MO"
+)
+
+# Quick meeting creation
+await nc_calendar_create_meeting(
+    title="Client Call",
+    date="2025-07-28",
+    time="15:00",
+    duration_minutes=60,
+    attendees="client@example.com,colleague@company.com"
+)
+
+# Get upcoming events  
+events = await nc_calendar_get_upcoming_events(days_ahead=7)
+
+# Advanced search - find all meetings with 5+ attendees lasting 2+ hours
+long_meetings = await nc_calendar_list_events(
+    calendar_name="",  # Search all calendars
+    search_all_calendars=True,
+    start_date="2025-07-01",
+    end_date="2025-07-31", 
+    min_attendees=5,
+    min_duration_minutes=120,
+    title_contains="meeting"
+)
+
+# Find availability for a 1-hour meeting with specific attendees
+availability = await nc_calendar_find_availability(
+    duration_minutes=60,
+    attendees="sarah@company.com,mike@company.com",
+    date_range_start="2025-07-28",
+    date_range_end="2025-08-04",
+    business_hours_only=True,
+    exclude_weekends=True,
+    preferred_times="09:00-12:00,14:00-17:00"
+)
+
+# Bulk update all team meetings to new location
+bulk_result = await nc_calendar_bulk_operations(
+    operation="update",
+    title_contains="team meeting",
+    start_date="2025-08-01", 
+    end_date="2025-08-31",
+    new_location="Conference Room B",
+    new_reminder_minutes=15
+)
+
+# Create a new project calendar
+new_calendar = await nc_calendar_manage_calendar(
+    action="create",
+    calendar_name="project-alpha",
+    display_name="Project Alpha Calendar",
+    description="Calendar for Project Alpha team",
+    color="#FF5722"
+)
+```
+
+### Note Attachments
+
+This server supports adding and retrieving note attachments via WebDAV. Please note the following behavior regarding attachments:
+
+* When a note is deleted, its attachments remain in the system. This matches the behavior of the official Nextcloud Notes app.
+* Orphaned attachments (attachments whose parent notes have been deleted) may accumulate over time.
+* WebDAV permissions must be properly configured for attachment operations to work correctly.
 
 ## Installation
 
@@ -105,4 +290,6 @@ Contributions are welcome! Please feel free to submit issues or pull requests on
 
 ## License
 
-This project is licensed under the MIT License. See the LICENSE file for details.
+This project is licensed under the AGPL-3.0 License. See the [LICENSE](./LICENSE) file for details.
+
+[![MseeP.ai Security Assessment Badge](https://mseep.net/pr/cbcoutinho-nextcloud-mcp-server-badge.png)](https://mseep.ai/app/cbcoutinho-nextcloud-mcp-server)

app-hooks/post-installation/install-notes-app.sh

@@ -0,0 +1,3 @@
+#!/bin/bash
+
+php /var/www/html/occ app:enable notes

app-hooks/post-installation/install-tables-app.sh

@@ -0,0 +1,3 @@
+#!/bin/bash
+
+php /var/www/html/occ app:enable tables

docker-compose.yml

@@ -3,7 +3,7 @@ services:
   # https://hub.docker.com/_/mariadb
   db:
     # Note: Check the recommend version here: https://docs.nextcloud.com/server/latest/admin_manual/installation/system_requirements.html#server
-    image: mariadb:lts@sha256:663d4d3e652220e3c618564dd401ae33ee5ea2b31aafd13c6d4e8ed29b8df733
+    image: mariadb:lts@sha256:2bcbaec92bd9d4f6591bc8103d3a8e6d0512ee2235506e47a2e129d190444405
     restart: always
     command: --transaction-isolation=READ-COMMITTED
     volumes:
@@ -17,11 +17,11 @@ services:
   # Note: Redis is an external service. You can find more information about the configuration here:
   # https://hub.docker.com/_/redis
   redis:
-    image: redis:alpine@sha256:f773b35a95e170d92dd4214a3ec4859b1b7960bf56896ae687646d695f311187
+    image: redis:alpine@sha256:25c0ae32c6c2301798579f5944af53729766a18eff5660bbef196fc2e6214a9c
     restart: always
 
   app:
-    image: nextcloud@sha256:ad4da6574b6dcb75c185128b091e6ac613f0aabda7ce7f75c9730d9f706e37d0
+    image: nextcloud:31.0.7@sha256:31d564f5f9f43f2aed0633854a2abd39155f85aa156997f7252f5af908efa99b
     #user: www-data:www-data
     restart: always
     #post_start:
@@ -34,7 +34,9 @@ services:
       - db
     volumes:
       - nextcloud:/var/www/html
+      - ./app-hooks/post-installation:/docker-entrypoint-hooks.d/post-installation:ro
     environment:
+      - NEXTCLOUD_TRUSTED_DOMAINS=app
       - NEXTCLOUD_ADMIN_USER=admin
       - NEXTCLOUD_ADMIN_PASSWORD=admin
       - MYSQL_PASSWORD=password
@@ -42,6 +44,15 @@ services:
       - MYSQL_USER=nextcloud
       - MYSQL_HOST=db
 
+  mcp:
+    build: .
+    ports:
+      - 8000:8000
+    environment:
+      - NEXTCLOUD_HOST=http://app:80
+      - NEXTCLOUD_USERNAME=admin
+      - NEXTCLOUD_PASSWORD=admin
+
 volumes:
   nextcloud:
   db:

docs/ADR-001-enhanced-note-search.md

@@ -0,0 +1,168 @@
+# ADR-001: Enhanced Note Search with Token-Based Relevance Ranking
+
+## Status
+Proposed
+
+## Context
+The current search implementation in the Nextcloud MCP server performs simple substring matching without relevance ranking. The existing method:
+1. Fetches all notes
+2. Performs case-insensitive substring matching on title and content
+3. Returns matches without any ordering by relevance
+
+This approach has several limitations:
+- Requires exact substring matches
+- No ranking by relevance
+- Only finds notes where the exact query string appears
+- Cannot prioritize more important matches (e.g., title vs content)
+- Inefficient for large note collections
+
+We need to improve the search functionality without adding external dependencies to enhance the user experience while maintaining simplicity.
+
+## Decision
+We will implement a token-based search with relevance ranking that:
+1. Splits queries and note content into individual tokens (words)
+2. Matches based on tokens rather than complete substrings
+3. Applies weighted scoring with title matches valued higher than content matches
+4. Sorts results by relevance score
+5. Maintains backward compatibility with the existing API
+
+## Implementation Details
+
+### 1. Query Processing
+The search query will be tokenized (split into individual words), normalized (converted to lowercase), and filtered for stop words if necessary:
+
+```python
+def process_query(query: str) -> list[str]:
+    # Convert to lowercase and split into tokens
+    tokens = query.lower().split()
+    # Filter out very short tokens (optional)
+    tokens = [token for token in tokens if len(token) > 1]
+    # Could add stop word removal here
+    return tokens
+```
+
+### 2. Note Content Processing
+Each note's title and content will be processed in a similar way:
+
+```python
+def process_note_content(note: dict) -> tuple[list[str], list[str]]:
+    # Process title
+    title = note.get("title", "").lower()
+    title_tokens = title.split()
+    
+    # Process content
+    content = note.get("content", "").lower()
+    content_tokens = content.split()
+    
+    return title_tokens, content_tokens
+```
+
+### 3. Scoring Algorithm
+We'll implement a scoring function that:
+- Assigns higher weight to title matches (e.g., 3x more important than content matches)
+- Considers the percentage of query tokens that match
+- Factors in the frequency of matches
+
+```python
+def calculate_score(query_tokens: list[str], title_tokens: list[str], content_tokens: list[str]) -> float:
+    # Constants for weighting
+    TITLE_WEIGHT = 3.0
+    CONTENT_WEIGHT = 1.0
+    
+    score = 0.0
+    
+    # Count matches in title
+    title_matches = sum(1 for qt in query_tokens if qt in title_tokens)
+    if query_tokens:  # Avoid division by zero
+        title_match_ratio = title_matches / len(query_tokens)
+        score += TITLE_WEIGHT * title_match_ratio
+    
+    # Count matches in content
+    content_matches = sum(1 for qt in query_tokens if qt in content_tokens)
+    if query_tokens:  # Avoid division by zero
+        content_match_ratio = content_matches / len(query_tokens)
+        score += CONTENT_WEIGHT * content_match_ratio
+    
+    # If no tokens matched at all, return zero
+    if title_matches == 0 and content_matches == 0:
+        return 0.0
+        
+    return score
+```
+
+### 4. Enhanced Search Implementation
+
+```python
+def notes_search_notes(self, *, query: str):
+    """
+    Search notes using token-based matching with relevance ranking.
+    Returns notes sorted by relevance score.
+    """
+    all_notes = self.notes_get_all()
+    search_results = []
+    
+    # Process the query
+    query_tokens = process_query(query)
+    
+    # If empty query after processing, return empty results
+    if not query_tokens:
+        return []
+    
+    # Process and score each note
+    for note in all_notes:
+        title_tokens, content_tokens = process_note_content(note)
+        score = calculate_score(query_tokens, title_tokens, content_tokens)
+        
+        # Only include notes with a non-zero score
+        if score > 0:
+            search_results.append({
+                "id": note.get("id"),
+                "title": note.get("title"),
+                "category": note.get("category"),
+                "modified": note.get("modified"),
+                "_score": score  # Include score for sorting (optional field)
+            })
+    
+    # Sort by score in descending order
+    search_results.sort(key=lambda x: x["_score"], reverse=True)
+    
+    # Remove score field before returning (optional)
+    for result in search_results:
+        if "_score" in result:
+            del result["_score"]
+    
+    return search_results
+```
+
+### 5. Performance Considerations
+- The enhanced search still retrieves all notes from the server, which could be inefficient for large collections
+- Future improvements could include caching or building an in-memory index
+- For very large note collections, consider adding pagination to the API
+
+## Consequences
+
+### Benefits
+1. Better search results with matches on individual words instead of exact phrases
+2. Relevant results appear first due to ranking
+3. Title matches are prioritized, matching user expectations
+4. No additional dependencies required
+5. Maintains backward compatibility with existing API
+
+### Limitations
+1. Slightly increased complexity in the search implementation
+2. Still requires fetching all notes for each search operation
+3. No handling of typos or similar words (would require fuzzy matching)
+4. No stemming/lemmatization to match word variations
+
+### Future Potential Enhancements
+1. Add support for phrase queries (exact matches)
+2. Implement an in-memory index for faster repeated searches
+3. Add basic natural language processing features (stemming, stop words)
+4. Support for fuzzy matching to handle typos
+
+## Alternatives Considered
+1. Implementing a full-text search engine (e.g., integrating with Elasticsearch)
+2. Using vector-based semantic search with embeddings
+3. Adding external NLP libraries for more sophisticated text processing
+
+These alternatives were not selected for the initial implementation due to the desire to maintain simplicity and avoid adding dependencies, but could be considered for future enhancements.

docs/nextcloud_notes_image_embedding.md

@@ -0,0 +1,118 @@
+# Working with Images in Nextcloud Notes
+
+This document explains how to properly work with images and attachments in Nextcloud Notes through the MCP server.
+
+## Adding Image Attachments
+
+Images and other files can be attached to notes using the WebDAV protocol. The Nextcloud MCP server handles this through the `add_note_attachment` method:
+
+```python
+# Example: Adding an image attachment to a note
+client.add_note_attachment(
+    note_id=123,  # The ID of the note
+    filename="image.png",  # The filename for the attachment
+    content=image_bytes,  # The binary content of the image
+    mime_type="image/png"  # The MIME type
+)
+```
+
+## Embedding Images in Notes
+
+For images to display inline within notes, you must reference them correctly in the note content. There are two methods:
+
+### 1. Markdown Syntax (Recommended)
+
+```markdown
+![Image Alt Text](.attachments.{note_id}/{filename})
+```
+
+For example:
+```markdown
+![My Screenshot](.attachments.123/screenshot.png)
+```
+
+### 2. HTML Image Tags
+
+```html
+<img src=".attachments.{note_id}/{filename}" alt="Image description" width="300" />
+```
+
+For example:
+```html
+<img src=".attachments.123/screenshot.png" alt="My Screenshot" width="300" />
+```
+
+## Storage Location
+
+Image attachments are stored in a hidden directory structure:
+
+```
+/Notes/.attachments.{note_id}/{filename}
+```
+
+This path is accessible via WebDAV, allowing direct file operations.
+
+## Orphaned Attachments Behavior
+
+**Important:** When notes are deleted, their attachments remain in the system. This is the expected behavior of the official Nextcloud Notes app, not a bug in the MCP server implementation.
+
+Consequences:
+- Orphaned attachments accumulate over time
+- No automatic cleanup of attachment directories
+- References to attachments in deleted notes become broken links
+
+## Examples
+
+### Complete Example: Creating a Note with Embedded Image
+
+```python
+from nextcloud_mcp_server.client import NextcloudClient
+import os
+
+# Create client
+client = NextcloudClient.from_env()
+
+# 1. Create the note
+note = client.notes_create_note(
+    title="Note with Embedded Image",
+    content="# Image Example\n\nThis note will have an embedded image.",
+    category="Documentation"
+)
+note_id = note["id"]
+note_etag = note["etag"]
+
+# 2. Read image content
+with open("example.png", "rb") as f:
+    image_content = f.read()
+
+# 3. Upload image as attachment
+client.add_note_attachment(
+    note_id=note_id,
+    filename="example.png",
+    content=image_content,
+    mime_type="image/png"
+)
+
+# 4. Update note content to include image reference
+updated_content = f"""# Image Example
+
+This note has an embedded image below:
+
+![Example Image](.attachments.{note_id}/example.png)
+"""
+
+# 5. Update the note with image reference
+client.notes_update_note(
+    note_id=note_id,
+    etag=note_etag,
+    content=updated_content
+)
+```
+
+## Troubleshooting
+
+If you encounter issues with attachments:
+
+1. **401 Unauthorized errors**: Verify WebDAV permissions in Nextcloud
+2. **Images not displaying**: Check the exact path format (`.attachments.{note_id}/{filename}`)
+3. **Attachment access after note deletion**: This is expected - attachments persist after note deletion

nextcloud_mcp_server/client.py

@@ -1,161 +0,0 @@
-import os
-from httpx import (
-    Client,
-    Auth,
-    BasicAuth,
-    Request,
-    Response,
-)
-import logging
-from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
-
-logger = logging.getLogger(__name__)
-
-
-def log_request(request: Request):
-    logger.info(
-        "Request event hook ****: %s %s - Waiting for content",
-        request.method,
-        request.url,
-    )
-    logger.info("Request body: %s", request.content)
-    logger.info("Headers: %s", request.headers)
-
-
-def log_response(response: Response):
-    response.read()  # Explicitly read the stream before accessing .text
-    logger.info("Response [%s] %s", response.status_code, response.text)
-
-
-class NextcloudClient:
-
-    def __init__(self, base_url: str, auth: Auth | None = None):
-
-        self._client = Client(
-            base_url=base_url,
-            auth=auth,
-            event_hooks={"request": [log_request], "response": [log_response]},
-        )
-        HTTPXClientInstrumentor.instrument_client(self._client)
-
-    @classmethod
-    def from_env(cls):
-
-        logger.info("Creating NC Client using env vars")
-
-        host = os.environ["NEXTCLOUD_HOST"]
-        username = os.environ["NEXTCLOUD_USERNAME"]
-        password = os.environ["NEXTCLOUD_PASSWORD"]
-        return cls(base_url=host, auth=BasicAuth(username, password))
-
-    def capabilities(self):
-
-        response = self._client.get(
-            "/ocs/v2.php/cloud/capabilities",
-            headers={"OCS-APIRequest": "true", "Accept": "application/json"},
-        )
-        response.raise_for_status()
-
-        return response.json()
-
-    def notes_get_settings(self):
-        response = self._client.get("/apps/notes/api/v1/settings")
-        response.raise_for_status()
-        return response.json()
-
-    def notes_get_all(self):
-        response = self._client.get("/apps/notes/api/v1/notes")
-        response.raise_for_status()
-        return response.json()
-
-    def notes_get_note(self, *, note_id: int):
-        response = self._client.get(f"/apps/notes/api/v1/notes/{note_id}")
-        response.raise_for_status()
-        return response.json()
-
-    def notes_create_note(
-        self,
-        *,
-        title: str | None = None,
-        content: str | None = None,
-        category: str | None = None,
-    ):
-        body = {}
-        if title:
-            body.update({"title": title})
-        if content:
-            body.update({"content": content})
-        if category:
-            body.update({"category": category})
-
-        response = self._client.post(
-            url="/apps/notes/api/v1/notes",
-            json=body,
-        )
-        response.raise_for_status()
-        return response.json()
-
-    def notes_update_note(
-        self,
-        *,
-        note_id: int,
-        etag: str,
-        title: str | None = None,
-        content: str | None = None,
-        category: str | None = None,
-    ):
-        # body = {"etag": etag} # Removed redundant line
-        body = {}
-        if title:
-            body.update({"title": title})
-        if content:
-            body.update({"content": content})
-        if category:
-            body.update({"category": category})
-
-        logger.info(
-            "Attempting to update note %s with etag %s, Body: %s, Category: %s",
-            note_id,
-            etag,  # This was current_etag in the loop
-            body,
-            category,
-        )
-        # Ensure conditional PUT using If-Match header is active
-        response = self._client.put(
-            url=f"/apps/notes/api/v1/notes/{note_id}",
-            json=body,
-            headers={
-                "If-Match": f'"{etag}"'  # NOTE: The `etag` needs to be surrounded by quotes `""`
-            },
-        )
-        logger.info(
-            "Update response for note %s: Status %s, Headers %s",
-            note_id,
-            response.status_code,
-            response.headers,
-        )
-        response.raise_for_status()
-        return response.json()
-
-    def notes_search_notes(self, *, query: str):
-        all_notes = self.notes_get_all()
-        search_results = []
-        query_lower = query.lower()
-        for note in all_notes:
-            title_lower = note.get("title", "").lower()
-            content_lower = note.get("content", "").lower()
-            if query_lower in title_lower or query_lower in content_lower:
-                search_results.append(
-                    {
-                        "id": note.get("id"),
-                        "title": note.get("title"),
-                        "category": note.get("category"),
-                        "modified": note.get("modified"),
-                    }
-                )
-        return search_results
-
-    def notes_delete_note(self, *, note_id: int):
-        response = self._client.delete(f"/apps/notes/api/v1/notes/{note_id}")
-        response.raise_for_status()
-        return response.json()

nextcloud_mcp_server/client/__init__.py

@@ -0,0 +1,85 @@
+import os
+from httpx import (
+    AsyncClient,
+    Auth,
+    BasicAuth,
+    Request,
+    Response,
+)
+import logging
+
+from .notes import NotesClient
+from .webdav import WebDAVClient
+from .tables import TablesClient
+from .calendar import CalendarClient
+from ..controllers.notes_search import NotesSearchController
+
+logger = logging.getLogger(__name__)
+
+
+def log_request(request: Request):
+    logger.info(
+        "Request event hook: %s %s - Waiting for content",
+        request.method,
+        request.url,
+    )
+    logger.info("Request body: %s", request.content)
+    logger.info("Headers: %s", request.headers)
+
+
+def log_response(response: Response):
+    response.read()  # Explicitly read the stream before accessing .text
+    logger.info("Response [%s] %s", response.status_code, response.text)
+
+
+class NextcloudClient:
+    """Main Nextcloud client that orchestrates all app clients."""
+
+    def __init__(self, base_url: str, username: str, auth: Auth | None = None):
+        self.username = username
+        self._client = AsyncClient(
+            base_url=base_url,
+            auth=auth,
+            # event_hooks={"request": [log_request], "response": [log_response]},
+        )
+
+        # Initialize app clients
+        self.notes = NotesClient(self._client, username)
+        self.webdav = WebDAVClient(self._client, username)
+        self.tables = TablesClient(self._client, username)
+        self.calendar = CalendarClient(self._client, username)
+
+        # Initialize controllers
+        self._notes_search = NotesSearchController()
+
+    @classmethod
+    def from_env(cls):
+        logger.info("Creating NC Client using env vars")
+
+        host = os.environ["NEXTCLOUD_HOST"]
+        username = os.environ["NEXTCLOUD_USERNAME"]
+        password = os.environ["NEXTCLOUD_PASSWORD"]
+        # Pass username to constructor
+        return cls(base_url=host, username=username, auth=BasicAuth(username, password))
+
+    async def capabilities(self):
+        response = await self._client.get(
+            "/ocs/v2.php/cloud/capabilities",
+            headers={"OCS-APIRequest": "true", "Accept": "application/json"},
+        )
+        response.raise_for_status()
+
+        return response.json()
+
+    async def notes_search_notes(self, *, query: str):
+        """Search notes using token-based matching with relevance ranking."""
+        all_notes = await self.notes.get_all_notes()
+        return self._notes_search.search_notes(all_notes, query)
+
+    def _get_webdav_base_path(self) -> str:
+        """Helper to get the base WebDAV path for the authenticated user."""
+        return f"/remote.php/dav/files/{self.username}"
+
+    async def close(self):
+        """Close the HTTP client."""
+        await self._client.aclose()

nextcloud_mcp_server/client/base.py

@@ -0,0 +1,41 @@
+"""Base client for Nextcloud operations with shared authentication."""
+
+from abc import ABC
+from httpx import AsyncClient
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class BaseNextcloudClient(ABC):
+    """Base class for all Nextcloud app clients."""
+
+    def __init__(self, http_client: AsyncClient, username: str):
+        """Initialize with shared HTTP client and username.
+
+        Args:
+            http_client: Authenticated AsyncClient instance
+            username: Nextcloud username for WebDAV operations
+        """
+        self._client = http_client
+        self.username = username
+
+    def _get_webdav_base_path(self) -> str:
+        """Helper to get the base WebDAV path for the authenticated user."""
+        return f"/remote.php/dav/files/{self.username}"
+
+    async def _make_request(self, method: str, url: str, **kwargs):
+        """Common request wrapper with logging and error handling.
+
+        Args:
+            method: HTTP method
+            url: Request URL
+            **kwargs: Additional request parameters
+
+        Returns:
+            Response object
+        """
+        logger.debug(f"Making {method} request to {url}")
+        response = await self._client.request(method, url, **kwargs)
+        response.raise_for_status()
+        return response

nextcloud_mcp_server/client/calendar.py

@@ -0,0 +1,977 @@
+"""CalDAV client for NextCloud calendar operations."""
+
+import xml.etree.ElementTree as ET
+from datetime import datetime, date
+from typing import Dict, Any, List, Optional, Tuple
+import logging
+from httpx import HTTPStatusError
+from icalendar import Calendar, Event as ICalEvent, vRecur, Alarm
+from datetime import timedelta
+import uuid
+
+from .base import BaseNextcloudClient
+
+logger = logging.getLogger(__name__)
+
+
+class CalendarClient(BaseNextcloudClient):
+    """Client for NextCloud CalDAV calendar operations."""
+
+    def _get_caldav_base_path(self) -> str:
+        """Helper to get the base CalDAV path for calendars."""
+        return f"/remote.php/dav/calendars/{self.username}"
+
+    def _get_principals_path(self) -> str:
+        """Helper to get the principals path for the user."""
+        return f"/remote.php/dav/principals/users/{self.username}"
+
+    async def list_calendars(self) -> List[Dict[str, Any]]:
+        """List all available calendars for the user."""
+        caldav_path = self._get_caldav_base_path()
+
+        propfind_body = """<?xml version="1.0" encoding="utf-8"?>
+        <d:propfind xmlns:d="DAV:" xmlns:c="urn:ietf:params:xml:ns:caldav" xmlns:cs="http://calendarserver.org/ns/">
+            <d:prop>
+                <d:displayname/>
+                <d:resourcetype/>
+                <c:calendar-description/>
+                <cs:calendar-color/>
+                <c:supported-calendar-component-set/>
+            </d:prop>
+        </d:propfind>"""
+
+        headers = {
+            "Depth": "1",
+            "Content-Type": "application/xml",
+            "Accept": "application/xml",
+        }
+
+        try:
+            response = await self._client.request(
+                "PROPFIND", caldav_path, content=propfind_body, headers=headers
+            )
+            response.raise_for_status()
+
+            # Parse XML response
+            root = ET.fromstring(response.content)
+            calendars = []
+
+            for response_elem in root.findall(".//{DAV:}response"):
+                href = response_elem.find(".//{DAV:}href")
+                if href is None:
+                    continue
+
+                href_text = href.text or ""
+                if not href_text.endswith("/"):
+                    continue  # Skip non-calendar resources
+
+                # Extract calendar name from href
+                calendar_name = href_text.rstrip("/").split("/")[-1]
+                if not calendar_name or calendar_name == self.username:
+                    continue
+
+                # Get properties
+                propstat = response_elem.find(".//{DAV:}propstat")
+                if propstat is None:
+                    continue
+
+                prop = propstat.find(".//{DAV:}prop")
+                if prop is None:
+                    continue
+
+                # Check if it's a calendar resource
+                resourcetype = prop.find(".//{DAV:}resourcetype")
+                is_calendar = (
+                    resourcetype is not None
+                    and resourcetype.find(".//{urn:ietf:params:xml:ns:caldav}calendar")
+                    is not None
+                )
+
+                if not is_calendar:
+                    continue
+
+                # Extract calendar properties
+                displayname_elem = prop.find(".//{DAV:}displayname")
+                displayname = (
+                    displayname_elem.text
+                    if displayname_elem is not None
+                    else calendar_name
+                )
+
+                description_elem = prop.find(
+                    ".//{urn:ietf:params:xml:ns:caldav}calendar-description"
+                )
+                description = (
+                    description_elem.text if description_elem is not None else ""
+                )
+
+                color_elem = prop.find(
+                    ".//{http://calendarserver.org/ns/}calendar-color"
+                )
+                color = color_elem.text if color_elem is not None else "#1976D2"
+
+                calendars.append(
+                    {
+                        "name": calendar_name,
+                        "display_name": displayname,
+                        "description": description,
+                        "color": color,
+                        "href": href_text,
+                    }
+                )
+
+            logger.debug(f"Found {len(calendars)} calendars")
+            return calendars
+
+        except HTTPStatusError as e:
+            if e.response.status_code == 401:
+                logger.warning(
+                    "Authentication failed for CalDAV - Calendar app may not be enabled for this user"
+                )
+                return []
+            elif e.response.status_code == 404:
+                logger.warning(
+                    "CalDAV endpoint not found - Calendar app may not be installed"
+                )
+                return []
+            logger.error(f"HTTP error listing calendars: {e}")
+            raise e
+        except Exception as e:
+            logger.error(f"Unexpected error listing calendars: {e}")
+            raise e
+
+    async def get_calendar_events(
+        self,
+        calendar_name: str,
+        start_date: str = "",
+        end_date: str = "",
+        limit: int = 50,
+    ) -> List[Dict[str, Any]]:
+        """List events in a calendar within date range."""
+        calendar_path = f"{self._get_caldav_base_path()}/{calendar_name}/"
+
+        # Build time range filter if dates provided
+        time_range_filter = ""
+        if start_date or end_date:
+            start_dt = start_date or "19700101T000000Z"
+            end_dt = end_date or "20301231T235959Z"
+            time_range_filter = f"""
+                <c:time-range start="{start_dt}" end="{end_dt}"/>
+            """
+
+        report_body = f"""<?xml version="1.0" encoding="utf-8"?>
+        <c:calendar-query xmlns:d="DAV:" xmlns:c="urn:ietf:params:xml:ns:caldav">
+            <d:prop>
+                <d:getetag/>
+                <c:calendar-data/>
+            </d:prop>
+            <c:filter>
+                <c:comp-filter name="VCALENDAR">
+                    <c:comp-filter name="VEVENT">
+                        {time_range_filter}
+                    </c:comp-filter>
+                </c:comp-filter>
+            </c:filter>
+        </c:calendar-query>"""
+
+        headers = {
+            "Depth": "1",
+            "Content-Type": "application/xml",
+            "Accept": "application/xml",
+        }
+
+        try:
+            response = await self._client.request(
+                "REPORT", calendar_path, content=report_body, headers=headers
+            )
+            response.raise_for_status()
+
+            # Parse XML response and extract events
+            root = ET.fromstring(response.content)
+            events = []
+
+            for response_elem in root.findall(".//{DAV:}response"):
+                href = response_elem.find(".//{DAV:}href")
+                if href is None:
+                    continue
+
+                propstat = response_elem.find(".//{DAV:}propstat")
+                if propstat is None:
+                    continue
+
+                prop = propstat.find(".//{DAV:}prop")
+                if prop is None:
+                    continue
+
+                calendar_data = prop.find(
+                    ".//{urn:ietf:params:xml:ns:caldav}calendar-data"
+                )
+                etag_elem = prop.find(".//{DAV:}getetag")
+
+                if calendar_data is not None and calendar_data.text:
+                    event_data = self._parse_ical_event(calendar_data.text)
+                    if event_data:
+                        event_data["href"] = href.text
+                        event_data["etag"] = (
+                            etag_elem.text if etag_elem is not None else ""
+                        )
+                        events.append(event_data)
+
+                if len(events) >= limit:
+                    break
+
+            logger.debug(f"Found {len(events)} events")
+            return events
+
+        except HTTPStatusError as e:
+            logger.error(f"HTTP error getting calendar events: {e}")
+            raise e
+        except Exception as e:
+            logger.error(f"Unexpected error getting calendar events: {e}")
+            raise e
+
+    async def create_event(
+        self, calendar_name: str, event_data: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """Create a new calendar event with comprehensive features."""
+        event_uid = str(uuid.uuid4())
+        event_filename = f"{event_uid}.ics"
+        event_path = f"{self._get_caldav_base_path()}/{calendar_name}/{event_filename}"
+
+        # Create iCalendar event
+        ical_content = self._create_ical_event(event_data, event_uid)
+
+        headers = {
+            "Content-Type": "text/calendar; charset=utf-8",
+            "If-None-Match": "*",  # Ensure we're creating, not updating
+        }
+
+        try:
+            response = await self._client.put(
+                event_path, content=ical_content, headers=headers
+            )
+            response.raise_for_status()
+
+            logger.debug(f"Created event {event_uid}")
+            return {
+                "uid": event_uid,
+                "href": event_path,
+                "etag": response.headers.get("etag", ""),
+                "status_code": response.status_code,
+            }
+
+        except HTTPStatusError as e:
+            logger.error(f"HTTP error creating event: {e}")
+            raise e
+        except Exception as e:
+            logger.error(f"Unexpected error creating event: {e}")
+            raise e
+
+    async def update_event(
+        self,
+        calendar_name: str,
+        event_uid: str,
+        event_data: Dict[str, Any],
+        etag: str = "",
+    ) -> Dict[str, Any]:
+        """Update an existing calendar event."""
+        event_filename = f"{event_uid}.ics"
+        event_path = f"{self._get_caldav_base_path()}/{calendar_name}/{event_filename}"
+
+        # Get existing event data to merge with updates
+        existing_event_data = {}
+        if not etag:
+            try:
+                existing_event_data, current_etag = await self.get_event(
+                    calendar_name, event_uid
+                )
+                etag = current_etag
+            except Exception:
+                # Continue without etag if we can't get it
+                pass
+
+        # Merge existing data with new data (new data takes precedence)
+        merged_data = {**existing_event_data, **event_data}
+
+        # Create updated iCalendar event
+        ical_content = self._create_ical_event(merged_data, event_uid)
+
+        headers = {
+            "Content-Type": "text/calendar; charset=utf-8",
+        }
+        if etag:
+            headers["If-Match"] = etag
+
+        try:
+            response = await self._client.put(
+                event_path, content=ical_content, headers=headers
+            )
+            response.raise_for_status()
+
+            logger.debug(f"Updated event {event_uid}")
+            return {
+                "uid": event_uid,
+                "href": event_path,
+                "etag": response.headers.get("etag", ""),
+                "status_code": response.status_code,
+            }
+
+        except HTTPStatusError as e:
+            logger.error(f"HTTP error updating event: {e}")
+            raise e
+        except Exception as e:
+            logger.error(f"Unexpected error updating event: {e}")
+            raise e
+
+    async def delete_event(self, calendar_name: str, event_uid: str) -> Dict[str, Any]:
+        """Delete a calendar event."""
+        event_filename = f"{event_uid}.ics"
+        event_path = f"{self._get_caldav_base_path()}/{calendar_name}/{event_filename}"
+
+        try:
+            response = await self._client.delete(event_path)
+            response.raise_for_status()
+
+            logger.debug(f"Deleted event {event_uid}")
+            return {"status_code": response.status_code}
+
+        except HTTPStatusError as e:
+            if e.response.status_code == 404:
+                logger.debug(f"Event {event_uid} not found")
+                return {"status_code": 404}
+            logger.error(f"HTTP error deleting event: {e}")
+            raise e
+        except Exception as e:
+            logger.error(f"Unexpected error deleting event: {e}")
+            raise e
+
+    async def get_event(
+        self, calendar_name: str, event_uid: str
+    ) -> Tuple[Dict[str, Any], str]:
+        """Get detailed information about a specific event."""
+        event_filename = f"{event_uid}.ics"
+        event_path = f"{self._get_caldav_base_path()}/{calendar_name}/{event_filename}"
+
+        headers = {"Accept": "text/calendar"}
+
+        try:
+            response = await self._client.get(event_path, headers=headers)
+            response.raise_for_status()
+
+            etag = response.headers.get("etag", "")
+            event_data = self._parse_ical_event(response.text)
+
+            if not event_data:
+                raise ValueError(f"Failed to parse event data for {event_uid}")
+
+            event_data["href"] = event_path
+            event_data["etag"] = etag
+
+            logger.debug(f"Retrieved event {event_uid}")
+            return event_data, etag
+
+        except HTTPStatusError as e:
+            logger.error(f"HTTP error getting event: {e}")
+            raise e
+        except Exception as e:
+            logger.error(f"Unexpected error getting event: {e}")
+            raise e
+
+    def _create_ical_event(self, event_data: Dict[str, Any], event_uid: str) -> str:
+        """Create iCalendar content from event data."""
+        cal = Calendar()
+        cal.add("prodid", "-//NextCloud MCP Server//EN")
+        cal.add("version", "2.0")
+
+        event = ICalEvent()
+        event.add("uid", event_uid)
+        event.add("summary", event_data.get("title", ""))
+        event.add("description", event_data.get("description", ""))
+        event.add("location", event_data.get("location", ""))
+
+        # Handle dates/times
+        start_str = event_data.get("start_datetime", "")
+        end_str = event_data.get("end_datetime", "")
+        all_day = event_data.get("all_day", False)
+
+        if start_str:  # Only parse if start_datetime is provided
+            if all_day:
+                start_date = datetime.fromisoformat(start_str.split("T")[0]).date()
+                event.add("dtstart", start_date)
+                if end_str:
+                    end_date = datetime.fromisoformat(end_str.split("T")[0]).date()
+                    event.add("dtend", end_date)
+            else:
+                start_dt = datetime.fromisoformat(start_str.replace("Z", "+00:00"))
+                event.add("dtstart", start_dt)
+                if end_str:
+                    end_dt = datetime.fromisoformat(end_str.replace("Z", "+00:00"))
+                    event.add("dtend", end_dt)
+
+        # Add categories
+        categories = event_data.get("categories", "")
+        if categories:
+            event.add("categories", categories.split(","))
+
+        # Add priority and status
+        priority = event_data.get("priority", 5)
+        event.add("priority", priority)
+
+        status = event_data.get("status", "CONFIRMED")
+        event.add("status", status)
+
+        # Add privacy classification
+        privacy = event_data.get("privacy", "PUBLIC")
+        event.add("class", privacy)
+
+        # Add URL
+        url = event_data.get("url", "")
+        if url:
+            event.add("url", url)
+
+        # Handle recurrence
+        recurring = event_data.get("recurring", False)
+        if recurring:
+            recurrence_rule = event_data.get("recurrence_rule", "")
+            if recurrence_rule:
+                event.add("rrule", vRecur.from_ical(recurrence_rule))
+
+        # Add alarms/reminders
+        reminder_minutes = event_data.get("reminder_minutes", 0)
+        if reminder_minutes > 0:
+            alarm = Alarm()
+            alarm.add("action", "DISPLAY")
+            alarm.add("description", "Event reminder")
+            alarm.add("trigger", timedelta(minutes=-reminder_minutes))
+            event.add_component(alarm)
+
+        # Add attendees
+        attendees = event_data.get("attendees", "")
+        if attendees:
+            for email in attendees.split(","):
+                if email.strip():
+                    event.add("attendee", f"mailto:{email.strip()}")
+
+        # Add timestamps
+        now = datetime.utcnow()
+        event.add("created", now)
+        event.add("dtstamp", now)
+        event.add("last-modified", now)
+
+        cal.add_component(event)
+        return cal.to_ical().decode("utf-8")
+
+    def _parse_ical_event(self, ical_text: str) -> Optional[Dict[str, Any]]:
+        """Parse iCalendar text and extract event data."""
+        try:
+            cal = Calendar.from_ical(ical_text)
+            for component in cal.walk():
+                if component.name == "VEVENT":
+                    event_data = {
+                        "uid": str(component.get("uid", "")),
+                        "title": str(component.get("summary", "")),
+                        "description": str(component.get("description", "")),
+                        "location": str(component.get("location", "")),
+                        "status": str(component.get("status", "CONFIRMED")),
+                        "priority": int(component.get("priority", 5)),
+                        "privacy": str(component.get("class", "PUBLIC")),
+                        "url": str(component.get("url", "")),
+                    }
+
+                    # Handle dates
+                    dtstart = component.get("dtstart")
+                    if dtstart:
+                        if isinstance(dtstart.dt, date) and not isinstance(
+                            dtstart.dt, datetime
+                        ):
+                            event_data["start_datetime"] = dtstart.dt.isoformat()
+                            event_data["all_day"] = True
+                        else:
+                            event_data["start_datetime"] = dtstart.dt.isoformat()
+                            event_data["all_day"] = False
+
+                    dtend = component.get("dtend")
+                    if dtend:
+                        if isinstance(dtend.dt, date) and not isinstance(
+                            dtend.dt, datetime
+                        ):
+                            event_data["end_datetime"] = dtend.dt.isoformat()
+                        else:
+                            event_data["end_datetime"] = dtend.dt.isoformat()
+
+                    # Handle categories
+                    categories = component.get("categories")
+                    if categories:
+                        event_data["categories"] = self._extract_categories(categories)
+
+                    # Handle recurrence
+                    rrule = component.get("rrule")
+                    if rrule:
+                        event_data["recurring"] = True
+                        event_data["recurrence_rule"] = str(rrule)
+
+                    # Handle attendees
+                    attendees = []
+                    for attendee in component.get("attendee", []):
+                        if isinstance(attendee, list):
+                            attendees.extend(
+                                str(a).replace("mailto:", "") for a in attendee
+                            )
+                        else:
+                            attendees.append(str(attendee).replace("mailto:", ""))
+                    if attendees:
+                        event_data["attendees"] = ",".join(attendees)
+
+                    return event_data
+
+            return None
+
+        except Exception as e:
+            logger.error(f"Error parsing iCalendar: {e}")
+            return None
+
+    def _extract_categories(self, categories_obj) -> str:
+        """Extract categories from icalendar object to string."""
+        if not categories_obj:
+            return ""
+
+        try:
+            # Handle icalendar vCategory objects
+            if hasattr(categories_obj, "cats"):
+                # vCategory object has a 'cats' attribute that's a list
+                return ", ".join(str(cat) for cat in categories_obj.cats)
+            elif hasattr(categories_obj, "__iter__") and not isinstance(
+                categories_obj, str
+            ):
+                # Handle lists or other iterables
+                return ", ".join(str(cat) for cat in categories_obj)
+            else:
+                # Handle strings or other objects
+                return str(categories_obj)
+        except Exception:
+            # Fallback to string conversion
+            return str(categories_obj)
+
+    async def search_events_across_calendars(
+        self,
+        start_date: str = "",
+        end_date: str = "",
+        filters: Optional[Dict[str, Any]] = None,
+    ) -> List[Dict[str, Any]]:
+        """Search events across all calendars with advanced filtering."""
+        try:
+            calendars = await self.list_calendars()
+            all_events = []
+
+            for calendar in calendars:
+                try:
+                    events = await self.get_calendar_events(
+                        calendar["name"], start_date, end_date
+                    )
+
+                    # Apply filters if provided
+                    if filters:
+                        events = self._apply_event_filters(events, filters)
+
+                    # Add calendar info to each event
+                    for event in events:
+                        event["calendar_name"] = calendar["name"]
+                        event["calendar_display_name"] = calendar.get(
+                            "display_name", calendar["name"]
+                        )
+
+                    all_events.extend(events)
+                except Exception as e:
+                    logger.warning(
+                        f"Error getting events from calendar {calendar['name']}: {e}"
+                    )
+                    continue
+
+            return all_events
+
+        except Exception as e:
+            logger.error(f"Error searching events across calendars: {e}")
+            raise
+
+    def _apply_event_filters(
+        self, events: List[Dict[str, Any]], filters: Dict[str, Any]
+    ) -> List[Dict[str, Any]]:
+        """Apply advanced filters to event list."""
+        filtered_events = []
+
+        for event in events:
+            # Skip if event doesn't match filters
+            if not self._event_matches_filters(event, filters):
+                continue
+            filtered_events.append(event)
+
+        return filtered_events
+
+    def _event_matches_filters(
+        self, event: Dict[str, Any], filters: Dict[str, Any]
+    ) -> bool:
+        """Check if an event matches the provided filters."""
+        try:
+            # Filter by minimum attendees
+            if "min_attendees" in filters:
+                attendees = event.get("attendees", "")
+                attendee_count = len(attendees.split(",")) if attendees else 0
+                if attendee_count < filters["min_attendees"]:
+                    return False
+
+            # Filter by minimum duration
+            if "min_duration_minutes" in filters:
+                start_str = event.get("start_datetime", "")
+                end_str = event.get("end_datetime", "")
+                if start_str and end_str:
+                    try:
+                        start_dt = datetime.fromisoformat(
+                            start_str.replace("Z", "+00:00")
+                        )
+                        end_dt = datetime.fromisoformat(end_str.replace("Z", "+00:00"))
+                        duration_minutes = (end_dt - start_dt).total_seconds() / 60
+                        if duration_minutes < filters["min_duration_minutes"]:
+                            return False
+                    except Exception:
+                        pass
+
+            # Filter by categories
+            if "categories" in filters:
+                event_categories = event.get("categories", "").lower()
+                required_categories = [cat.lower() for cat in filters["categories"]]
+                if not any(cat in event_categories for cat in required_categories):
+                    return False
+
+            # Filter by status
+            if "status" in filters:
+                if event.get("status", "").upper() != filters["status"].upper():
+                    return False
+
+            # Filter by title contains
+            if "title_contains" in filters:
+                title = event.get("title", "").lower()
+                search_term = filters["title_contains"].lower()
+                if search_term not in title:
+                    return False
+
+            # Filter by location contains
+            if "location_contains" in filters:
+                location = event.get("location", "").lower()
+                search_term = filters["location_contains"].lower()
+                if search_term not in location:
+                    return False
+
+            return True
+
+        except Exception:
+            # If filtering fails, include the event
+            return True
+
+    async def find_availability(
+        self,
+        duration_minutes: int,
+        attendees: Optional[List[str]] = None,
+        date_range_start: str = "",
+        date_range_end: str = "",
+        constraints: Optional[Dict[str, Any]] = None,
+    ) -> List[Dict[str, Any]]:
+        """Find available time slots for scheduling."""
+        try:
+            # Set default date range if not provided
+            if not date_range_start:
+                date_range_start = datetime.now().strftime("%Y-%m-%d")
+            if not date_range_end:
+                end_date = datetime.now() + timedelta(days=7)
+                date_range_end = end_date.strftime("%Y-%m-%d")
+
+            # Get all events in the date range
+            busy_events = await self.search_events_across_calendars(
+                start_date=date_range_start, end_date=date_range_end
+            )
+
+            # Filter events for relevant attendees if specified
+            if attendees:
+                relevant_events = []
+                for event in busy_events:
+                    event_attendees = event.get("attendees", "").lower()
+                    if any(
+                        attendee.lower() in event_attendees for attendee in attendees
+                    ):
+                        relevant_events.append(event)
+                busy_events = relevant_events
+
+            # Apply constraints
+            constraints = constraints or {}
+            business_hours_only = constraints.get("business_hours_only", False)
+            exclude_weekends = constraints.get("exclude_weekends", False)
+            preferred_times = constraints.get("preferred_times", [])
+
+            # Generate time slots
+            available_slots = self._generate_available_slots(
+                busy_events,
+                duration_minutes,
+                date_range_start,
+                date_range_end,
+                business_hours_only,
+                exclude_weekends,
+                preferred_times,
+            )
+
+            return available_slots
+
+        except Exception as e:
+            logger.error(f"Error finding availability: {e}")
+            raise
+
+    def _generate_available_slots(
+        self,
+        busy_events: List[Dict[str, Any]],
+        duration_minutes: int,
+        start_date: str,
+        end_date: str,
+        business_hours_only: bool,
+        exclude_weekends: bool,
+        preferred_times: List[str],
+    ) -> List[Dict[str, Any]]:
+        """Generate available time slots."""
+        available_slots = []
+
+        try:
+            current_date = datetime.fromisoformat(start_date)
+            end_date_dt = datetime.fromisoformat(end_date)
+
+            while current_date <= end_date_dt:
+                # Skip weekends if requested
+                if exclude_weekends and current_date.weekday() >= 5:
+                    current_date += timedelta(days=1)
+                    continue
+
+                # Generate slots for this day
+                day_slots = self._generate_day_slots(
+                    current_date,
+                    busy_events,
+                    duration_minutes,
+                    business_hours_only,
+                    preferred_times,
+                )
+                available_slots.extend(day_slots)
+
+                current_date += timedelta(days=1)
+
+            return available_slots[:10]  # Limit to 10 slots
+
+        except Exception as e:
+            logger.error(f"Error generating available slots: {e}")
+            return []
+
+    def _generate_day_slots(
+        self,
+        date: datetime,
+        busy_events: List[Dict[str, Any]],
+        duration_minutes: int,
+        business_hours_only: bool,
+        preferred_times: List[str],
+    ) -> List[Dict[str, Any]]:
+        """Generate available slots for a specific day."""
+        slots = []
+
+        try:
+            # Define working hours
+            if business_hours_only:
+                start_hour, end_hour = 9, 17
+            else:
+                start_hour, end_hour = 8, 20
+
+            # Get busy periods for this day
+            day_busy_periods = []
+            for event in busy_events:
+                try:
+                    event_start = datetime.fromisoformat(
+                        event["start_datetime"].replace("Z", "+00:00")
+                    )
+                    event_end = datetime.fromisoformat(
+                        event["end_datetime"].replace("Z", "+00:00")
+                    )
+
+                    # Check if event is on this day
+                    if event_start.date() == date.date():
+                        day_busy_periods.append((event_start.time(), event_end.time()))
+                except Exception:
+                    continue
+
+            # Sort busy periods
+            day_busy_periods.sort()
+
+            # Generate potential slots
+            current_time = date.replace(
+                hour=start_hour, minute=0, second=0, microsecond=0
+            )
+            end_time = date.replace(hour=end_hour, minute=0, second=0, microsecond=0)
+            slot_duration = timedelta(minutes=duration_minutes)
+
+            while current_time + slot_duration <= end_time:
+                slot_end = current_time + slot_duration
+
+                # Check if slot conflicts with any busy period
+                if not self._slot_conflicts(
+                    current_time.time(), slot_end.time(), day_busy_periods
+                ):
+                    # Check preferred times if specified
+                    if not preferred_times or self._slot_in_preferred_times(
+                        current_time.time(), preferred_times
+                    ):
+                        slots.append(
+                            {
+                                "start_datetime": current_time.isoformat(),
+                                "end_datetime": slot_end.isoformat(),
+                                "duration_minutes": duration_minutes,
+                                "date": date.date().isoformat(),
+                            }
+                        )
+
+                current_time += timedelta(minutes=30)  # 30-minute increments
+
+            return slots
+
+        except Exception as e:
+            logger.error(f"Error generating day slots: {e}")
+            return []
+
+    def _slot_conflicts(self, slot_start, slot_end, busy_periods):
+        """Check if a time slot conflicts with busy periods."""
+        for busy_start, busy_end in busy_periods:
+            if slot_start < busy_end and slot_end > busy_start:
+                return True
+        return False
+
+    def _slot_in_preferred_times(self, slot_start, preferred_times):
+        """Check if slot falls within preferred time ranges."""
+        if not preferred_times:
+            return True
+
+        for time_range in preferred_times:
+            try:
+                start_str, end_str = time_range.split("-")
+                pref_start = datetime.strptime(start_str, "%H:%M").time()
+                pref_end = datetime.strptime(end_str, "%H:%M").time()
+
+                if pref_start <= slot_start <= pref_end:
+                    return True
+            except Exception:
+                continue
+
+        return False
+
+    async def bulk_update_events(
+        self, filter_criteria: Dict[str, Any], update_data: Dict[str, Any]
+    ) -> Dict[str, Any]:
+        """Bulk update events matching filter criteria."""
+        try:
+            # Find events matching criteria
+            events = await self.search_events_across_calendars(
+                start_date=filter_criteria.get("start_date", ""),
+                end_date=filter_criteria.get("end_date", ""),
+                filters=filter_criteria,
+            )
+
+            updated_count = 0
+            failed_count = 0
+            results = []
+
+            for event in events:
+                try:
+                    # Update the event
+                    await self.update_event(
+                        event["calendar_name"], event["uid"], update_data
+                    )
+                    updated_count += 1
+                    results.append(
+                        {
+                            "uid": event["uid"],
+                            "status": "updated",
+                            "title": event.get("title", ""),
+                        }
+                    )
+                except Exception as e:
+                    failed_count += 1
+                    results.append(
+                        {
+                            "uid": event["uid"],
+                            "status": "failed",
+                            "error": str(e),
+                            "title": event.get("title", ""),
+                        }
+                    )
+
+            return {
+                "total_found": len(events),
+                "updated_count": updated_count,
+                "failed_count": failed_count,
+                "results": results,
+            }
+
+        except Exception as e:
+            logger.error(f"Error in bulk update: {e}")
+            raise
+
+    async def create_calendar(
+        self,
+        calendar_name: str,
+        display_name: str = "",
+        description: str = "",
+        color: str = "#1976D2",
+    ) -> Dict[str, Any]:
+        """Create a new calendar."""
+        try:
+            # Calendar creation via CalDAV MKCALENDAR
+            calendar_path = f"{self._get_caldav_base_path()}/{calendar_name}/"
+
+            # Create MKCALENDAR body
+            mkcol_body = f"""<?xml version="1.0" encoding="utf-8"?>
+            <mkcalendar xmlns="urn:ietf:params:xml:ns:caldav" xmlns:d="DAV:" xmlns:cs="http://calendarserver.org/ns/">
+                <d:set>
+                    <d:prop>
+                        <d:displayname>{display_name or calendar_name}</d:displayname>
+                        <cs:calendar-color>{color}</cs:calendar-color>
+                        <caldav:calendar-description xmlns:caldav="urn:ietf:params:xml:ns:caldav">{description}</caldav:calendar-description>
+                        <caldav:supported-calendar-component-set xmlns:caldav="urn:ietf:params:xml:ns:caldav">
+                            <caldav:comp name="VEVENT"/>
+                        </caldav:supported-calendar-component-set>
+                    </d:prop>
+                </d:set>
+            </mkcalendar>"""
+
+            headers = {"Content-Type": "application/xml", "Depth": "0"}
+
+            response = await self._client.request(
+                "MKCALENDAR", calendar_path, content=mkcol_body, headers=headers
+            )
+            response.raise_for_status()
+
+            logger.debug(f"Created calendar: {calendar_name}")
+            return {
+                "name": calendar_name,
+                "display_name": display_name or calendar_name,
+                "description": description,
+                "color": color,
+                "status_code": response.status_code,
+            }
+
+        except Exception as e:
+            logger.error(f"Error creating calendar {calendar_name}: {e}")
+            raise
+
+    async def delete_calendar(self, calendar_name: str) -> Dict[str, Any]:
+        """Delete a calendar."""
+        try:
+            calendar_path = f"{self._get_caldav_base_path()}/{calendar_name}/"
+
+            response = await self._client.delete(calendar_path)
+            response.raise_for_status()
+
+            logger.debug(f"Deleted calendar: {calendar_name}")
+            return {"status_code": response.status_code}
+
+        except Exception as e:
+            logger.error(f"Error deleting calendar {calendar_name}: {e}")
+            raise

nextcloud_mcp_server/client/notes.py

@@ -0,0 +1,199 @@
+"""Client for Nextcloud Notes app operations."""
+
+from typing import Dict, List, Any, Optional
+import logging
+
+from .base import BaseNextcloudClient
+
+logger = logging.getLogger(__name__)
+
+
+class NotesClient(BaseNextcloudClient):
+    """Client for Nextcloud Notes app operations."""
+
+    async def get_settings(self) -> Dict[str, Any]:
+        """Get Notes app settings."""
+        response = await self._make_request("GET", "/apps/notes/api/v1/settings")
+        return response.json()
+
+    async def get_all_notes(self) -> List[Dict[str, Any]]:
+        """Get all notes."""
+        response = await self._make_request("GET", "/apps/notes/api/v1/notes")
+        return response.json()
+
+    async def get_note(self, note_id: int) -> Dict[str, Any]:
+        """Get a specific note by ID."""
+        response = await self._make_request(
+            "GET", f"/apps/notes/api/v1/notes/{note_id}"
+        )
+        return response.json()
+
+    async def create_note(
+        self,
+        title: Optional[str] = None,
+        content: Optional[str] = None,
+        category: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Create a new note."""
+        body = {}
+        if title:
+            body["title"] = title
+        if content:
+            body["content"] = content
+        if category:
+            body["category"] = category
+
+        response = await self._make_request(
+            "POST", "/apps/notes/api/v1/notes", json=body
+        )
+        return response.json()
+
+    async def update(
+        self,
+        note_id: int,
+        etag: str,
+        title: Optional[str] = None,
+        content: Optional[str] = None,
+        category: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Update an existing note."""
+        # Get current note details to check for category change
+        old_note = None
+        try:
+            if category is not None:
+                old_note = await self.get_note(note_id)
+                old_category = old_note.get("category", "")
+                logger.info(f"Current category for note {note_id}: '{old_category}'")
+        except Exception as e:
+            logger.warning(
+                f"Could not fetch current note {note_id} details before update: {e}"
+            )
+            old_note = None
+
+        # Prepare update body
+        body = {}
+        if title:
+            body["title"] = title
+        if content:
+            body["content"] = content
+        if category:
+            body["category"] = category
+
+        logger.info(
+            f"Attempting to update note {note_id} with etag {etag}. Body: {body}"
+        )
+
+        response = await self._make_request(
+            "PUT",
+            f"/apps/notes/api/v1/notes/{note_id}",
+            json=body,
+            headers={"If-Match": f'"{etag}"'},
+        )
+
+        logger.info(
+            f"Update response for note {note_id}: Status {response.status_code}"
+        )
+        updated_note = response.json()
+
+        # Check for category change and cleanup old attachment directory if needed
+        if (
+            old_note
+            and category is not None
+            and old_note.get("category", "") != category
+        ):
+            logger.info(
+                f"Category changed from '{old_note.get('category', '')}' to '{category}' - cleaning up old attachment directory"
+            )
+            try:
+                # Import here to avoid circular imports
+                from .webdav import WebDAVClient
+
+                webdav_client = WebDAVClient(self._client, self.username)
+                await webdav_client.cleanup_old_attachment_directory(
+                    note_id=note_id, old_category=old_note.get("category", "")
+                )
+            except Exception as e:
+                logger.error(
+                    f"Error cleaning up old attachment directory for note {note_id}: {e}"
+                )
+
+        return updated_note
+
+    async def delete_note(self, note_id: int) -> Dict[str, Any]:
+        """Delete a note and its attachments."""
+        # Fetch note details first to get category for cleanup
+        try:
+            note_details = await self.get_note(note_id)
+            category = note_details.get("category", "")
+
+            # Determine potential categories for cleanup
+            potential_categories = []
+            if category:
+                potential_categories.append(category)
+            if category != "":
+                potential_categories.append("")  # Empty category
+
+            logger.info(
+                f"Note {note_id} has category: '{category}', will check attachment directories in: {potential_categories}"
+            )
+        except Exception as e:
+            logger.warning(
+                f"Could not fetch note {note_id} details before deletion: {e}"
+            )
+            potential_categories = ["", "Unknown"]  # Try common categories
+
+        # Delete the note via API
+        logger.info(f"Deleting note {note_id} via API")
+        response = await self._make_request(
+            "DELETE", f"/apps/notes/api/v1/notes/{note_id}"
+        )
+        logger.info(f"Note {note_id} deleted successfully via API")
+        json_response = response.json()
+
+        # Clean up attachment directories
+        try:
+            from .webdav import WebDAVClient
+
+            webdav_client = WebDAVClient(self._client, self.username)
+
+            for cat in potential_categories:
+                try:
+                    await webdav_client.cleanup_note_attachments(note_id, cat)
+                except Exception as e:
+                    logger.warning(
+                        f"Failed to cleanup attachments for category '{cat}': {e}"
+                    )
+        except Exception as e:
+            logger.warning(f"Error during attachment cleanup: {e}")
+
+        return json_response
+
+    async def append_content(self, note_id: int, content: str) -> Dict[str, Any]:
+        """Append content to an existing note with a separator."""
+        logger.info(f"Appending content to note {note_id}")
+
+        # Get current note
+        current_note = await self.get_note(note_id)
+
+        # Use fixed separator for consistency
+        separator = "\n---\n"
+
+        # Combine content
+        existing_content = current_note.get("content", "")
+        if existing_content:
+            new_content = existing_content + separator + content
+        else:
+            new_content = content  # No separator needed for empty notes
+
+        logger.info(
+            f"Combining existing content ({len(existing_content)} chars) with new content ({len(content)} chars)"
+        )
+
+        # Update with combined content
+        return await self.update(
+            note_id=note_id,
+            etag=current_note["etag"],
+            content=new_content,
+            title=None,  # Keep existing title
+            category=None,  # Keep existing category
+        )

nextcloud_mcp_server/client/tables.py

@@ -0,0 +1,125 @@
+"""Client for Nextcloud Tables app operations."""
+
+from typing import Dict, List, Any, Optional
+import logging
+
+from .base import BaseNextcloudClient
+
+logger = logging.getLogger(__name__)
+
+
+class TablesClient(BaseNextcloudClient):
+    """Client for Nextcloud Tables app operations."""
+
+    async def list_tables(self) -> List[Dict[str, Any]]:
+        """List all tables available to the user."""
+        response = await self._make_request(
+            "GET",
+            "/ocs/v2.php/apps/tables/api/2/tables",
+            headers={"OCS-APIRequest": "true", "Accept": "application/json"},
+        )
+        result = response.json()
+        return result["ocs"]["data"]
+
+    async def get_table_schema(self, table_id: int) -> Dict[str, Any]:
+        """Get the schema/structure of a specific table including columns and views."""
+        # Using v1 API as v2 schema endpoint had issues during testing
+        response = await self._make_request(
+            "GET", f"/index.php/apps/tables/api/1/tables/{table_id}/scheme"
+        )
+        return response.json()
+
+    async def get_table_rows(
+        self, table_id: int, limit: Optional[int] = None, offset: Optional[int] = None
+    ) -> List[Dict[str, Any]]:
+        """Read rows from a table with optional pagination."""
+        params = {}
+        if limit is not None:
+            params["limit"] = limit
+        if offset is not None:
+            params["offset"] = offset
+
+        response = await self._make_request(
+            "GET", f"/index.php/apps/tables/api/1/tables/{table_id}/rows", params=params
+        )
+        return response.json()
+
+    async def create_row(self, table_id: int, data: Dict[str, Any]) -> Dict[str, Any]:
+        """Insert a new row into a table.
+
+        Args:
+            table_id: ID of the table to insert into
+            data: Dictionary mapping column IDs to values, e.g. {1: "text", 2: 42}
+        """
+        # Transform data to API format: {"data": {"1": "text", "2": 42}}
+        api_data = {str(k): v for k, v in data.items()}
+
+        response = await self._make_request(
+            "POST",
+            f"/ocs/v2.php/apps/tables/api/2/tables/{table_id}/rows",
+            headers={"OCS-APIRequest": "true", "Accept": "application/json"},
+            json={"data": api_data},
+        )
+        result = response.json()
+        return result["ocs"]["data"]
+
+    async def update_row(self, row_id: int, data: Dict[str, Any]) -> Dict[str, Any]:
+        """Update an existing row in a table.
+
+        Args:
+            row_id: ID of the row to update
+            data: Dictionary mapping column IDs to new values, e.g. {1: "new text", 2: 99}
+        """
+        # Transform data to API format for v1 endpoint
+        api_data = {str(k): v for k, v in data.items()}
+
+        response = await self._make_request(
+            "PUT",
+            f"/index.php/apps/tables/api/1/rows/{row_id}",
+            json={"data": api_data},
+        )
+        return response.json()
+
+    async def delete_row(self, row_id: int) -> Dict[str, Any]:
+        """Delete a row from a table."""
+        response = await self._make_request(
+            "DELETE", f"/index.php/apps/tables/api/1/rows/{row_id}"
+        )
+        return response.json()
+
+    def transform_row_data(
+        self, rows: List[Dict[str, Any]], columns: List[Dict[str, Any]]
+    ) -> List[Dict[str, Any]]:
+        """Transform raw row data into more readable format using column names.
+
+        Args:
+            rows: Raw row data from the API
+            columns: Column definitions from table schema
+
+        Returns:
+            List of rows with column names as keys instead of column IDs
+        """
+        # Create mapping from column ID to column title
+        column_map = {col["id"]: col["title"] for col in columns}
+
+        transformed_rows = []
+        for row in rows:
+            transformed_row = {
+                "id": row["id"],
+                "tableId": row["tableId"],
+                "createdBy": row["createdBy"],
+                "createdAt": row["createdAt"],
+                "lastEditBy": row["lastEditBy"],
+                "lastEditAt": row["lastEditAt"],
+                "data": {},
+            }
+
+            # Transform data array to column_name: value mapping
+            for item in row["data"]:
+                column_id = item["columnId"]
+                column_name = column_map.get(column_id, f"column_{column_id}")
+                transformed_row["data"][column_name] = item["value"]
+
+            transformed_rows.append(transformed_row)
+
+        return transformed_rows

nextcloud_mcp_server/client/webdav.py

@@ -0,0 +1,417 @@
+"""WebDAV client for Nextcloud file operations."""
+
+import mimetypes
+from typing import Tuple, Dict, Any, Optional, List
+import logging
+from httpx import HTTPStatusError
+import xml.etree.ElementTree as ET
+
+from .base import BaseNextcloudClient
+
+logger = logging.getLogger(__name__)
+
+
+class WebDAVClient(BaseNextcloudClient):
+    """Client for Nextcloud WebDAV operations."""
+
+    async def delete_resource(self, path: str) -> Dict[str, Any]:
+        """Delete a resource (file or directory) via WebDAV DELETE."""
+        # Ensure path ends with a slash if it's a directory
+        if not path.endswith("/"):
+            path_with_slash = f"{path}/"
+        else:
+            path_with_slash = path
+
+        webdav_path = f"{self._get_webdav_base_path()}/{path_with_slash.lstrip('/')}"
+        logger.debug(f"Deleting WebDAV resource: {webdav_path}")
+
+        headers = {"OCS-APIRequest": "true"}
+        try:
+            # First try a PROPFIND to verify resource exists
+            propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
+            try:
+                propfind_resp = await self._client.request(
+                    "PROPFIND", webdav_path, headers=propfind_headers
+                )
+                logger.debug(
+                    f"Resource exists check status: {propfind_resp.status_code}"
+                )
+            except HTTPStatusError as e:
+                if e.response.status_code == 404:
+                    logger.debug(f"Resource '{path}' doesn't exist, no deletion needed")
+                    return {"status_code": 404}
+                # For other errors, continue with deletion attempt
+
+            # Proceed with deletion
+            response = await self._client.delete(webdav_path, headers=headers)
+            response.raise_for_status()
+            logger.debug(f"Successfully deleted WebDAV resource '{path}'")
+            return {"status_code": response.status_code}
+
+        except HTTPStatusError as e:
+            if e.response.status_code == 404:
+                logger.debug(f"Resource '{path}' not found, no deletion needed")
+                return {"status_code": 404}
+            else:
+                logger.error(f"HTTP error deleting WebDAV resource '{path}': {e}")
+                raise e
+        except Exception as e:
+            logger.error(f"Unexpected error deleting WebDAV resource '{path}': {e}")
+            raise e
+
+    async def cleanup_old_attachment_directory(
+        self, note_id: int, old_category: str
+    ) -> Dict[str, Any]:
+        """Clean up the attachment directory for a note in its old category location."""
+        old_category_path_part = f"{old_category}/" if old_category else ""
+        old_attachment_dir_path = (
+            f"Notes/{old_category_path_part}.attachments.{note_id}/"
+        )
+
+        logger.debug(f"Cleaning up old attachment directory: {old_attachment_dir_path}")
+        try:
+            delete_result = await self.delete_resource(path=old_attachment_dir_path)
+            logger.debug(f"Cleanup result: {delete_result}")
+            return delete_result
+        except Exception as e:
+            logger.error(f"Error during cleanup of old attachment directory: {e}")
+            raise e
+
+    async def cleanup_note_attachments(
+        self, note_id: int, category: str
+    ) -> Dict[str, Any]:
+        """Clean up attachment directory for a specific note and category."""
+        cat_path_part = f"{category}/" if category else ""
+        attachment_dir_path = f"Notes/{cat_path_part}.attachments.{note_id}/"
+
+        logger.debug(
+            f"Cleaning up attachments for note {note_id} in category '{category}'"
+        )
+        try:
+            delete_result = await self.delete_resource(path=attachment_dir_path)
+            logger.debug(f"Cleanup result for note {note_id}: {delete_result}")
+            return delete_result
+        except Exception as e:
+            logger.error(f"Failed cleaning up attachments for note {note_id}: {e}")
+            raise e
+
+    async def add_note_attachment(
+        self,
+        note_id: int,
+        filename: str,
+        content: bytes,
+        category: Optional[str] = None,
+        mime_type: Optional[str] = None,
+    ) -> Dict[str, Any]:
+        """Add/Update an attachment to a note via WebDAV PUT."""
+        # Construct paths based on provided category
+        webdav_base = self._get_webdav_base_path()
+        category_path_part = f"{category}/" if category else ""
+        attachment_dir_segment = f".attachments.{note_id}"
+        parent_dir_webdav_rel_path = (
+            f"Notes/{category_path_part}{attachment_dir_segment}"
+        )
+        parent_dir_path = f"{webdav_base}/{parent_dir_webdav_rel_path}"
+        attachment_path = f"{parent_dir_path}/{filename}"
+
+        logger.debug(f"Uploading attachment '{filename}' for note {note_id}")
+
+        if not mime_type:
+            mime_type, _ = mimetypes.guess_type(filename)
+            if not mime_type:
+                mime_type = "application/octet-stream"
+
+        headers = {"Content-Type": mime_type, "OCS-APIRequest": "true"}
+        try:
+            # First check if we can access WebDAV at all
+            notes_dir_path = f"{webdav_base}/Notes"
+            propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
+            notes_dir_response = await self._client.request(
+                "PROPFIND", notes_dir_path, headers=propfind_headers
+            )
+
+            if notes_dir_response.status_code == 401:
+                logger.error("WebDAV authentication failed for Notes directory")
+                raise HTTPStatusError(
+                    f"Authentication error accessing WebDAV Notes directory: {notes_dir_response.status_code}",
+                    request=notes_dir_response.request,
+                    response=notes_dir_response,
+                )
+            elif notes_dir_response.status_code >= 400:
+                logger.error(
+                    f"Error accessing WebDAV Notes directory: {notes_dir_response.status_code}"
+                )
+                notes_dir_response.raise_for_status()
+
+            # Ensure the parent directory exists using MKCOL
+            mkcol_headers = {"OCS-APIRequest": "true"}
+            mkcol_response = await self._client.request(
+                "MKCOL", parent_dir_path, headers=mkcol_headers
+            )
+
+            # MKCOL should return 201 Created or 405 Method Not Allowed (if directory already exists)
+            if mkcol_response.status_code not in [201, 405]:
+                logger.error(
+                    f"Unexpected status code {mkcol_response.status_code} when creating attachments directory"
+                )
+                mkcol_response.raise_for_status()
+
+            # Proceed with the PUT request
+            response = await self._client.put(
+                attachment_path, content=content, headers=headers
+            )
+            response.raise_for_status()
+            logger.debug(
+                f"Successfully uploaded attachment '{filename}' to note {note_id}"
+            )
+            return {"status_code": response.status_code}
+
+        except HTTPStatusError as e:
+            logger.error(
+                f"HTTP error uploading attachment '{filename}' to note {note_id}: {e}"
+            )
+            raise e
+        except Exception as e:
+            logger.error(
+                f"Unexpected error uploading attachment '{filename}' to note {note_id}: {e}"
+            )
+            raise e
+
+    async def get_note_attachment(
+        self, note_id: int, filename: str, category: Optional[str] = None
+    ) -> Tuple[bytes, str]:
+        """Fetch a specific attachment from a note via WebDAV GET."""
+        webdav_base = self._get_webdav_base_path()
+        category_path_part = f"{category}/" if category else ""
+        attachment_dir_segment = f".attachments.{note_id}"
+        attachment_path = f"{webdav_base}/Notes/{category_path_part}{attachment_dir_segment}/{filename}"
+
+        logger.debug(f"Fetching attachment '{filename}' for note {note_id}")
+
+        try:
+            response = await self._client.get(attachment_path)
+            response.raise_for_status()
+
+            content = response.content
+            mime_type = response.headers.get("content-type", "application/octet-stream")
+
+            logger.debug(
+                f"Successfully fetched attachment '{filename}' ({len(content)} bytes)"
+            )
+            return content, mime_type
+
+        except HTTPStatusError as e:
+            if e.response.status_code == 404:
+                logger.debug(f"Attachment '{filename}' not found for note {note_id}")
+            else:
+                logger.error(
+                    f"HTTP error fetching attachment '{filename}' for note {note_id}: {e}"
+                )
+            raise e
+        except Exception as e:
+            logger.error(
+                f"Unexpected error fetching attachment '{filename}' for note {note_id}: {e}"
+            )
+            raise e
+
+    async def list_directory(self, path: str = "") -> List[Dict[str, Any]]:
+        """List files and directories in the specified path via WebDAV PROPFIND."""
+        webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}"
+        if not webdav_path.endswith("/"):
+            webdav_path += "/"
+
+        logger.debug(f"Listing directory: {path}")
+
+        propfind_body = """<?xml version="1.0"?>
+        <d:propfind xmlns:d="DAV:">
+            <d:prop>
+                <d:displayname/>
+                <d:getcontentlength/>
+                <d:getcontenttype/>
+                <d:getlastmodified/>
+                <d:resourcetype/>
+            </d:prop>
+        </d:propfind>"""
+
+        headers = {"Depth": "1", "Content-Type": "text/xml", "OCS-APIRequest": "true"}
+
+        try:
+            response = await self._client.request(
+                "PROPFIND", webdav_path, content=propfind_body, headers=headers
+            )
+            response.raise_for_status()
+
+            # Parse the XML response
+            root = ET.fromstring(response.content)
+            items = []
+
+            # Skip the first response (the directory itself)
+            responses = root.findall(".//{DAV:}response")[1:]
+
+            for response_elem in responses:
+                href = response_elem.find(".//{DAV:}href")
+                if href is None:
+                    continue
+
+                # Extract file/directory name from href
+                href_text = href.text or ""
+                name = href_text.rstrip("/").split("/")[-1]
+                if not name:
+                    continue
+
+                # Get properties
+                propstat = response_elem.find(".//{DAV:}propstat")
+                if propstat is None:
+                    continue
+
+                prop = propstat.find(".//{DAV:}prop")
+                if prop is None:
+                    continue
+
+                # Determine if it's a directory
+                resourcetype = prop.find(".//{DAV:}resourcetype")
+                is_directory = (
+                    resourcetype is not None
+                    and resourcetype.find(".//{DAV:}collection") is not None
+                )
+
+                # Get other properties
+                size_elem = prop.find(".//{DAV:}getcontentlength")
+                size = (
+                    int(size_elem.text)
+                    if size_elem is not None and size_elem.text
+                    else 0
+                )
+
+                content_type_elem = prop.find(".//{DAV:}getcontenttype")
+                content_type = (
+                    content_type_elem.text if content_type_elem is not None else None
+                )
+
+                modified_elem = prop.find(".//{DAV:}getlastmodified")
+                modified = modified_elem.text if modified_elem is not None else None
+
+                items.append(
+                    {
+                        "name": name,
+                        "path": f"{path.rstrip('/')}/{name}" if path else name,
+                        "is_directory": is_directory,
+                        "size": size if not is_directory else None,
+                        "content_type": content_type,
+                        "last_modified": modified,
+                    }
+                )
+
+            logger.debug(f"Found {len(items)} items in directory: {path}")
+            return items
+
+        except HTTPStatusError as e:
+            logger.error(f"HTTP error listing directory '{webdav_path}': {e}")
+            raise e
+        except Exception as e:
+            logger.error(f"Unexpected error listing directory '{webdav_path}': {e}")
+            raise e
+
+    async def read_file(self, path: str) -> Tuple[bytes, str]:
+        """Read a file's content via WebDAV GET."""
+        webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}"
+
+        logger.debug(f"Reading file: {path}")
+
+        try:
+            response = await self._client.get(webdav_path)
+            response.raise_for_status()
+
+            content = response.content
+            content_type = response.headers.get(
+                "content-type", "application/octet-stream"
+            )
+
+            logger.debug(f"Successfully read file '{path}' ({len(content)} bytes)")
+            return content, content_type
+
+        except HTTPStatusError as e:
+            logger.error(f"HTTP error reading file '{path}': {e}")
+            raise e
+        except Exception as e:
+            logger.error(f"Unexpected error reading file '{path}': {e}")
+            raise e
+
+    async def write_file(
+        self, path: str, content: bytes, content_type: Optional[str] = None
+    ) -> Dict[str, Any]:
+        """Write content to a file via WebDAV PUT."""
+        webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}"
+
+        logger.debug(f"Writing file: {path}")
+
+        if not content_type:
+            content_type, _ = mimetypes.guess_type(path)
+            if not content_type:
+                content_type = "application/octet-stream"
+
+        headers = {"Content-Type": content_type, "OCS-APIRequest": "true"}
+
+        try:
+            response = await self._client.put(
+                webdav_path, content=content, headers=headers
+            )
+            response.raise_for_status()
+
+            logger.debug(f"Successfully wrote file '{path}'")
+            return {"status_code": response.status_code}
+
+        except HTTPStatusError as e:
+            logger.error(f"HTTP error writing file '{path}': {e}")
+            raise e
+        except Exception as e:
+            logger.error(f"Unexpected error writing file '{path}': {e}")
+            raise e
+
+    async def create_directory(
+        self, path: str, recursive: bool = False
+    ) -> Dict[str, Any]:
+        """Create a directory via WebDAV MKCOL."""
+        webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}"
+        if not webdav_path.endswith("/"):
+            webdav_path += "/"
+
+        logger.debug(f"Creating directory: {path}")
+
+        headers = {"OCS-APIRequest": "true"}
+
+        try:
+            response = await self._client.request("MKCOL", webdav_path, headers=headers)
+            response.raise_for_status()
+
+            logger.debug(f"Successfully created directory '{path}'")
+            return {"status_code": response.status_code}
+
+        except HTTPStatusError as e:
+            # Method Not Allowed - directory already exists
+            if e.response.status_code == 405:
+                logger.debug(f"Directory '{path}' already exists")
+                return {"status_code": 405, "message": "Directory already exists"}
+
+            # File Conflict - parent directory does not exist
+            if e.response.status_code == 409 and recursive:
+                # Extract parent directory path
+                path_parts = path.strip("/").split("/")
+                if len(path_parts) > 1:
+                    parent_dir = "/".join(path_parts[:-1])
+                    logger.debug(
+                        f"Parent directory '{parent_dir}' doesn't exist, creating recursively"
+                    )
+                    await self.create_directory(parent_dir, recursive)
+                    # Now try to create the original directory again
+                    return await self.create_directory(path, recursive)
+                else:
+                    # This shouldn't happen for single-level directories under root
+                    logger.error(f"409 conflict for single-level directory '{path}'")
+                    raise e
+
+            logger.error(f"HTTP error creating directory '{path}': {e}")
+            raise e
+        except Exception as e:
+            logger.error(f"Unexpected error creating directory '{path}': {e}")
+            raise e

nextcloud_mcp_server/config.py

@@ -4,11 +4,8 @@ LOGGING_CONFIG = {
     "version": 1,
     "handlers": {
         "default": {
-            "class": "logging.FileHandler",
+            "class": "logging.StreamHandler",
             "formatter": "http",
-            # "stream": "ext://sys.stderr"
-            "filename": "/tmp/nextcloud-mcp-server.log",
-            "mode": "a",
         }
     },
     "formatters": {

nextcloud_mcp_server/controllers/__init__.py

@@ -0,0 +1 @@
+"""Controllers for utility operations."""

nextcloud_mcp_server/controllers/notes_search.py

@@ -0,0 +1,102 @@
+"""Controller for notes search functionality."""
+
+from typing import List, Dict, Any
+
+
+class NotesSearchController:
+    """Handles notes search logic and scoring."""
+
+    def search_notes(
+        self, notes: List[Dict[str, Any]], query: str
+    ) -> List[Dict[str, Any]]:
+        """
+        Search notes using token-based matching with relevance ranking.
+        Returns notes sorted by relevance score.
+        """
+        search_results = []
+        query_tokens = self._process_query(query)
+
+        # If empty query after processing, return empty results
+        if not query_tokens:
+            return []
+
+        # Process and score each note
+        for note in notes:
+            title_tokens, content_tokens = self._process_note_content(note)
+            score = self._calculate_score(query_tokens, title_tokens, content_tokens)
+
+            # Only include notes with a non-zero score
+            if score >= 0.5:
+                search_results.append(
+                    {
+                        "id": note.get("id"),
+                        "title": note.get("title"),
+                        "category": note.get("category"),
+                        "modified": note.get("modified"),
+                        "_score": score,  # Include score for sorting
+                    }
+                )
+
+        # Sort by score in descending order
+        search_results.sort(key=lambda x: x["_score"], reverse=True)
+
+        return search_results
+
+    def _process_query(self, query: str) -> List[str]:
+        """
+        Tokenize and normalize the search query.
+        """
+        # Convert to lowercase and split into tokens
+        tokens = query.lower().split()
+        # Filter out very short tokens
+        tokens = [token for token in tokens if len(token) > 1]
+        return tokens
+
+    def _process_note_content(
+        self, note: Dict[str, Any]
+    ) -> tuple[List[str], List[str]]:
+        """
+        Tokenize and normalize note title and content.
+        """
+        # Process title
+        title = note.get("title", "").lower()
+        title_tokens = title.split()
+
+        # Process content
+        content = note.get("content", "").lower()
+        content_tokens = content.split()
+
+        return title_tokens, content_tokens
+
+    def _calculate_score(
+        self,
+        query_tokens: List[str],
+        title_tokens: List[str],
+        content_tokens: List[str],
+    ) -> float:
+        """
+        Calculate a relevance score for a note based on query tokens.
+        """
+        # Constants for weighting
+        TITLE_WEIGHT = 3.0
+        CONTENT_WEIGHT = 1.0
+
+        score = 0.0
+
+        # Count matches in title
+        title_matches = sum(1 for qt in query_tokens if qt in title_tokens)
+        if query_tokens:  # Avoid division by zero
+            title_match_ratio = title_matches / len(query_tokens)
+            score += TITLE_WEIGHT * title_match_ratio
+
+        # Count matches in content
+        content_matches = sum(1 for qt in query_tokens if qt in content_tokens)
+        if query_tokens:  # Avoid division by zero
+            content_match_ratio = content_matches / len(query_tokens)
+            score += CONTENT_WEIGHT * content_match_ratio
+
+        # If no tokens matched at all, return zero
+        if title_matches == 0 and content_matches == 0:
+            return 0.0
+
+        return score

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "nextcloud-mcp-server"
-version = "0.1.0"
+version = "0.6.0"
 description = ""
 authors = [
     {name = "Chris Coutinho",email = "chris@coutinho.io"}
@@ -8,26 +8,44 @@ authors = [
 readme = "README.md"
 requires-python = ">=3.11"
 dependencies = [
-    "mcp[cli] (>=1.7,<1.8)",
+    "mcp[cli] (>=1.10,<1.11)",
     "httpx (>=0.28.1,<0.29.0)",
-    "opentelemetry-instrumentation-httpx (>=0.53b1,<0.54)",
+    "pillow (>=11.2.1,<12.0.0)",
+    "icalendar (>=6.0.0,<7.0.0)"
 ]
 
 [project.scripts]
 nc-mcp-server = "nextcloud_mcp_server.server:run"
 
 [tool.pytest.ini_options]
+asyncio_mode = "auto"
+asyncio_default_test_loop_scope = "session"
+asyncio_default_fixture_loop_scope = "session"
+log_cli = 1
+log_cli_level = "WARN"
+log_level = "WARN"
 markers = [
     "integration: marks tests as slow (deselect with '-m \"not slow\"')"
 ]
 
+[tool.commitizen]
+name = "cz_conventional_commits"
+tag_format = "v$version"
+version_scheme = "pep440"
+version_provider = "uv"
+update_changelog_on_bump = true
+major_version_zero = true
 
 [build-system]
 requires = ["poetry-core>=2.0.0,<3.0.0"]
 build-backend = "poetry.core.masonry.api"
 
-[tool.poetry.group.dev.dependencies]
-black = "25.1.0"
-ipython = "9.2.0"
-pytest = "8.3.5"
-opentelemetry-distro = "^0.53b1"
+[dependency-groups]
+dev = [
+    "commitizen>=4.8.2",
+    "ipython>=9.2.0",
+    "pytest>=8.3.5",
+    "pytest-asyncio>=1.0.0",
+    "pytest-cov>=6.1.1",
+    "ruff>=0.11.13",
+]

renovate.json

@@ -1,7 +1,8 @@
 {
   "$schema": "https://docs.renovatebot.com/renovate-schema.json",
   "extends": [
-    "config:best-practices"
+    "config:best-practices",
+    "mergeConfidence:all-badges"
   ],
   "dependencyDashboard": true
 }

tests/conftest.py

@@ -0,0 +1,128 @@
+import pytest
+import os
+import logging
+import uuid
+from nextcloud_mcp_server.client import NextcloudClient
+from httpx import HTTPStatusError
+import asyncio
+
+logger = logging.getLogger(__name__)
+
+# pytestmark = pytest.mark.asyncio(loop_scope="package")
+
+
+@pytest.fixture(scope="session")
+async def nc_client() -> NextcloudClient:
+    """
+    Fixture to create a NextcloudClient instance for integration tests.
+    Uses environment variables for configuration.
+    """
+
+    assert os.getenv("NEXTCLOUD_HOST"), "NEXTCLOUD_HOST env var not set"
+    assert os.getenv("NEXTCLOUD_USERNAME"), "NEXTCLOUD_USERNAME env var not set"
+    assert os.getenv("NEXTCLOUD_PASSWORD"), "NEXTCLOUD_PASSWORD env var not set"
+    logger.info("Creating session-scoped NextcloudClient from environment variables.")
+    client = NextcloudClient.from_env()
+    # Optional: Perform a quick check like getting capabilities to ensure connection works
+    try:
+        await client.capabilities()
+        logger.info(
+            "NextcloudClient session fixture initialized and capabilities checked."
+        )
+    except Exception as e:
+        logger.error(f"Failed to initialize NextcloudClient session fixture: {e}")
+        pytest.fail(f"Failed to connect to Nextcloud or get capabilities: {e}")
+    return client
+
+
+@pytest.fixture
+async def temporary_note(nc_client: NextcloudClient):
+    """
+    Fixture to create a temporary note for a test and ensure its deletion afterward.
+    Yields the created note dictionary.
+    """
+    asyncio.new_event_loop()
+
+    note_id = None
+    unique_suffix = uuid.uuid4().hex[:8]
+    note_title = f"Temporary Test Note {unique_suffix}"
+    note_content = f"Content for temporary note {unique_suffix}"
+    note_category = "TemporaryTesting"
+    created_note_data = None
+
+    logger.info(f"Creating temporary note: {note_title}")
+    try:
+        created_note_data = await nc_client.notes.create_note(
+            title=note_title, content=note_content, category=note_category
+        )
+        note_id = created_note_data.get("id")
+        if not note_id:
+            pytest.fail("Failed to get ID from created temporary note.")
+
+        logger.info(f"Temporary note created with ID: {note_id}")
+        yield created_note_data  # Provide the created note data to the test
+
+    finally:
+        if note_id:
+            logger.info(f"Cleaning up temporary note ID: {note_id}")
+            try:
+                await nc_client.notes.delete_note(note_id=note_id)
+                logger.info(f"Successfully deleted temporary note ID: {note_id}")
+            except HTTPStatusError as e:
+                # Ignore 404 if note was already deleted by the test itself
+                if e.response.status_code != 404:
+                    logger.error(f"HTTP error deleting temporary note {note_id}: {e}")
+                else:
+                    logger.warning(f"Temporary note {note_id} already deleted (404).")
+            except Exception as e:
+                logger.error(f"Unexpected error deleting temporary note {note_id}: {e}")
+
+
+@pytest.fixture
+async def temporary_note_with_attachment(
+    nc_client: NextcloudClient, temporary_note: dict
+):
+    """
+    Fixture that creates a temporary note, adds an attachment, and cleans up both.
+    Yields a tuple: (note_data, attachment_filename, attachment_content).
+    Depends on the temporary_note fixture.
+    """
+    asyncio.new_event_loop()
+
+    note_data = temporary_note
+    note_id = note_data["id"]
+    note_category = note_data.get("category")  # Get category from the note data
+    unique_suffix = uuid.uuid4().hex[:8]
+    attachment_filename = f"temp_attach_{unique_suffix}.txt"
+    attachment_content = f"Content for {attachment_filename}".encode("utf-8")
+    attachment_mime = "text/plain"
+
+    logger.info(
+        f"Adding attachment '{attachment_filename}' to temporary note ID: {note_id} (category: '{note_category or ''}')"
+    )
+    try:
+        # Pass the category to add_note_attachment
+        upload_response = await nc_client.webdav.add_note_attachment(
+            note_id=note_id,
+            filename=attachment_filename,
+            content=attachment_content,
+            category=note_category,  # Pass the fetched category
+            mime_type=attachment_mime,
+        )
+        assert upload_response.get("status_code") in [
+            201,
+            204,
+        ], f"Failed to upload attachment: {upload_response}"
+        logger.info(f"Attachment '{attachment_filename}' added successfully.")
+
+        yield note_data, attachment_filename, attachment_content
+
+        # Cleanup for the attachment is handled by the notes_delete_note call
+        # in the temporary_note fixture's finally block (which deletes the .attachments dir)
+
+    except Exception as e:
+        logger.error(f"Failed to add attachment in fixture: {e}")
+        pytest.fail(f"Fixture setup failed during attachment upload: {e}")
+
+    # Note: The temporary_note fixture's finally block will handle note deletion,
+    # which should also trigger the WebDAV directory deletion attempt.

tests/integration/test_attachments.py

@@ -0,0 +1,403 @@
+import pytest
+import logging
+import time
+import uuid
+from httpx import HTTPStatusError
+
+from nextcloud_mcp_server.client import NextcloudClient
+
+# Note: nc_client fixture is session-scoped in conftest.py
+# Note: temporary_note and temporary_note_with_attachment fixtures are function-scoped in conftest.py
+
+logger = logging.getLogger(__name__)
+
+# Mark all tests in this module as integration tests
+pytestmark = pytest.mark.integration
+
+
+async def test_attachments_add_and_get(
+    nc_client: NextcloudClient, temporary_note_with_attachment: tuple
+):
+    """
+    Tests adding an attachment (via fixture) and retrieving it.
+    """
+    note_data, attachment_filename, attachment_content = temporary_note_with_attachment
+    note_id = note_data["id"]
+    note_category = note_data.get("category")  # Get category from fixture data
+
+    logger.info(
+        f"Attempting to retrieve attachment '{attachment_filename}' added by fixture for note ID: {note_id}"
+    )
+    # Pass category to get_note_attachment
+    retrieved_content, retrieved_mime = await nc_client.webdav.get_note_attachment(
+        note_id=note_id, filename=attachment_filename, category=note_category
+    )
+    logger.info(
+        f"Attachment retrieved. Mime type: {retrieved_mime}, Size: {len(retrieved_content)} bytes"
+    )
+
+    assert retrieved_content == attachment_content
+    assert "text/plain" in retrieved_mime  # Fixture uses text/plain
+    logger.info("Retrieved attachment content and mime type verified successfully.")
+
+
+async def test_attachments_add_to_note_with_category(
+    nc_client: NextcloudClient, temporary_note: dict
+):
+    """
+    Tests adding and retrieving an attachment specifically for a note that has a category.
+    Uses temporary_note fixture and adds attachment manually within the test.
+    """
+    note_data = (
+        temporary_note  # Note created by fixture (has category 'TemporaryTesting')
+    )
+    note_id = note_data["id"]
+    note_category = note_data["category"]
+    logger.info(
+        f"Using note ID: {note_id} with category '{note_category}' for attachment test."
+    )
+
+    # Add attachment within the test
+    unique_suffix = uuid.uuid4().hex[:8]
+    attachment_filename = f"category_attach_{unique_suffix}.txt"
+    attachment_content = f"Content for {attachment_filename}".encode("utf-8")
+    attachment_mime = "text/plain"
+
+    logger.info(
+        f"Attempting to add attachment '{attachment_filename}' to note ID: {note_id}"
+    )
+    # Pass category to add_note_attachment
+    upload_response = await nc_client.webdav.add_note_attachment(
+        note_id=note_id,
+        filename=attachment_filename,
+        content=attachment_content,
+        category=note_category,  # Pass the note's category
+        mime_type=attachment_mime,
+    )
+    assert upload_response and "status_code" in upload_response
+    assert upload_response["status_code"] in [201, 204]
+    logger.info(
+        f"Attachment '{attachment_filename}' added successfully (Status: {upload_response['status_code']})."
+    )
+    time.sleep(1)
+
+    # Get and Verify Attachment
+    logger.info(
+        f"Attempting to retrieve attachment '{attachment_filename}' from note ID: {note_id}"
+    )
+    # Pass category to get_note_attachment
+    retrieved_content, retrieved_mime = await nc_client.webdav.get_note_attachment(
+        note_id=note_id,
+        filename=attachment_filename,
+        category=note_category,  # Pass the note's category
+    )
+    logger.info(
+        f"Attachment retrieved. Mime type: {retrieved_mime}, Size: {len(retrieved_content)} bytes"
+    )
+
+    assert retrieved_content == attachment_content
+    assert attachment_mime in retrieved_mime
+    logger.info(
+        "Retrieved attachment content and mime type verified successfully for note with category."
+    )
+    # Cleanup is handled by the temporary_note fixture
+
+
+async def test_attachments_cleanup_on_note_delete(
+    nc_client: NextcloudClient, temporary_note_with_attachment: tuple
+):
+    """
+    Tests that the attachment (and its directory) are deleted when the parent note is deleted.
+    Relies on the cleanup mechanism within notes_delete_note and the temporary_note fixture.
+    """
+    note_data, attachment_filename, _ = temporary_note_with_attachment
+    note_id = note_data["id"]
+    note_category = note_data.get("category")  # Get category from fixture data
+
+    # Fixture setup already added the attachment.
+    # Fixture teardown (from temporary_note) will delete the note.
+    # We just need to verify the attachment is gone *after* the test finishes
+    # and the fixture cleanup runs. However, pytest fixtures don't easily allow
+    # checking state *after* cleanup.
+    # Instead, we will manually delete the note here and verify the attachment is gone.
+
+    logger.info(
+        f"Attachment '{attachment_filename}' exists for note {note_id} (added by fixture)."
+    )
+
+    # Manually delete the note
+    logger.info(f"Manually deleting note ID: {note_id} within the test.")
+    await nc_client.notes.delete_note(note_id=note_id)
+    logger.info(f"Note ID: {note_id} deleted successfully.")
+    time.sleep(1)
+
+    # Verify Note Is Deleted
+    with pytest.raises(HTTPStatusError) as excinfo_note:
+        await nc_client.notes.get_note(note_id=note_id)
+    assert excinfo_note.value.response.status_code == 404
+    logger.info(f"Verified note {note_id} deletion (404 received).")
+
+    # Verify Attachment Is Deleted (via 404 on GET)
+    logger.info(
+        f"Verifying attachment '{attachment_filename}' is deleted for note ID: {note_id}"
+    )
+    with pytest.raises(HTTPStatusError) as excinfo_attach:
+        # Pass category to get_note_attachment - although it should fail anyway
+        # because the note (and thus details) are gone.
+        # The client method will raise 404 from the initial notes_get_note call.
+        await nc_client.webdav.get_note_attachment(
+            note_id=note_id,
+            filename=attachment_filename,
+            category=note_category,  # Pass category, though note fetch should fail first
+        )
+    # Expect 404 because the note itself is gone
+    assert excinfo_attach.value.response.status_code == 404
+    logger.info(
+        f"Attachment '{attachment_filename}' correctly not found (404) after note deletion."
+    )
+
+    # Directly verify attachment directory doesn't exist using WebDAV PROPFIND
+    logger.info("Directly verifying attachment directory doesn't exist via PROPFIND")
+    webdav_base = nc_client._get_webdav_base_path()
+    category_path_part = f"{note_category}/" if note_category else ""
+    attachment_dir_path = (
+        f"{webdav_base}/Notes/{category_path_part}.attachments.{note_id}"
+    )
+    propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
+    try:
+        propfind_resp = await nc_client._client.request(
+            "PROPFIND", attachment_dir_path, headers=propfind_headers
+        )
+        status = propfind_resp.status_code
+        if status in [200, 207]:  # Successful PROPFIND means directory exists
+            logger.error(
+                f"Attachment directory still exists! PROPFIND returned {status}"
+            )
+            assert False, (
+                f"Expected attachment directory to be gone, but PROPFIND returned {status}!"
+            )
+    except HTTPStatusError as e:
+        assert e.response.status_code == 404, (
+            f"Expected PROPFIND to fail with 404, got {e.response.status_code}"
+        )
+        logger.info(
+            "Verified attachment directory does not exist via PROPFIND (404 received)"
+        )
+
+    # Note: The temporary_note fixture will still run its cleanup,
+    # but it will find the note already deleted (404) and handle it gracefully.
+
+
+async def test_attachments_category_change_handling(nc_client: NextcloudClient):
+    """
+    Tests attachment handling when a note's category is changed.
+    Verifies attachment retrieval works before and after category change,
+    and that cleanup targets the correct final location.
+    """
+    note_id = None
+    initial_category = "CategoryA"
+    new_category = "CategoryB"
+    unique_suffix = uuid.uuid4().hex[:8]
+    note_title = f"Category Change Test {unique_suffix}"
+    attachment_filename = f"cat_change_{unique_suffix}.txt"
+    attachment_content = f"Content for {attachment_filename}".encode("utf-8")
+
+    try:
+        # 1. Create note with initial category
+        logger.info(f"Creating note '{note_title}' in category '{initial_category}'")
+        created_note = await nc_client.notes.create_note(
+            title=note_title, content="Initial content", category=initial_category
+        )
+        note_id = created_note["id"]
+        etag1 = created_note["etag"]
+        logger.info(f"Note created with ID: {note_id}, Etag: {etag1}")
+        time.sleep(1)
+
+        # 2. Add attachment (passing initial category)
+        logger.info(
+            f"Adding attachment '{attachment_filename}' to note {note_id} (in {initial_category})"
+        )
+        upload_response = await nc_client.webdav.add_note_attachment(
+            note_id=note_id,
+            filename=attachment_filename,
+            content=attachment_content,
+            category=initial_category,
+            mime_type="text/plain",
+        )
+        assert upload_response["status_code"] in [201, 204]
+        logger.info("Attachment added successfully.")
+        time.sleep(1)
+
+        # 3. Verify attachment retrieval from initial category (passing initial category)
+        logger.info(
+            f"Verifying attachment retrieval from initial category '{initial_category}'"
+        )
+        retrieved_content1, _ = await nc_client.webdav.get_note_attachment(
+            note_id=note_id, filename=attachment_filename, category=initial_category
+        )
+        assert retrieved_content1 == attachment_content
+        logger.info("Attachment retrieved successfully from initial category.")
+
+        # 4. Update note category
+        logger.info(
+            f"Updating note {note_id} category from '{initial_category}' to '{new_category}'"
+        )
+        # Need to fetch the latest etag after attachment add (WebDAV ops don't update note etag)
+        current_note_data = await nc_client.notes.get_note(note_id=note_id)
+        current_etag = current_note_data["etag"]
+        updated_note = await nc_client.notes.update(
+            note_id=note_id,
+            etag=current_etag,
+            category=new_category,
+            title=note_title,
+            content="Updated content",  # Pass required fields
+        )
+        etag3 = updated_note["etag"]
+        assert updated_note["category"] == new_category
+        logger.info(f"Note category updated successfully. New Etag: {etag3}")
+        time.sleep(1)
+
+        # 5. Verify attachment retrieval from *new* category (passing new category)
+        logger.info(
+            f"Verifying attachment retrieval from new category '{new_category}'"
+        )
+        retrieved_content2, _ = await nc_client.webdav.get_note_attachment(
+            note_id=note_id, filename=attachment_filename, category=new_category
+        )
+        assert retrieved_content2 == attachment_content
+        logger.info("Attachment retrieved successfully from new category.")
+
+        # 5.1 Verify old category attachment directory is gone via WebDAV PROPFIND
+        logger.info("Directly checking if old attachment directory exists in WebDAV")
+        webdav_base = nc_client._get_webdav_base_path()
+        old_attachment_dir_path = (
+            f"{webdav_base}/Notes/{initial_category}/.attachments.{note_id}"
+        )
+        propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
+        try:
+            propfind_resp = await nc_client._client.request(
+                "PROPFIND", old_attachment_dir_path, headers=propfind_headers
+            )
+            status = propfind_resp.status_code
+            if status in [200, 207]:  # Successful PROPFIND means directory exists
+                logger.error(
+                    f"Old attachment directory still exists! PROPFIND returned {status}"
+                )
+                assert False, (
+                    f"Expected old directory to be gone, but PROPFIND returned {status} - directory still exists!"
+                )
+        except HTTPStatusError as e:
+            assert e.response.status_code == 404, (
+                f"Expected PROPFIND to fail with 404, got {e.response.status_code}"
+            )
+            logger.info(
+                "Verified old attachment directory does not exist via PROPFIND (404 received)"
+            )
+
+        # 5.2 Verify new category attachment directory exists via WebDAV PROPFIND
+        logger.info("Directly checking if new attachment directory exists in WebDAV")
+        new_attachment_dir_path = (
+            f"{webdav_base}/Notes/{new_category}/.attachments.{note_id}"
+        )
+        try:
+            propfind_resp = await nc_client._client.request(
+                "PROPFIND", new_attachment_dir_path, headers=propfind_headers
+            )
+            status = propfind_resp.status_code
+            assert status in [
+                207,
+                200,
+            ], f"Expected PROPFIND to return success (207/200), got {status}"
+            logger.info(
+                f"Verified new attachment directory exists via PROPFIND ({status} received)"
+            )
+        except HTTPStatusError as e:
+            logger.error(
+                f"New attachment directory not found! PROPFIND failed with {e.response.status_code}"
+            )
+            assert False, (
+                f"Expected new attachment directory to exist, but PROPFIND failed with {e.response.status_code}"
+            )
+
+    finally:
+        # 6. Cleanup: Delete the note (client should use the *final* category for cleanup path)
+        if note_id:
+            logger.info(
+                f"Cleaning up note ID: {note_id} (last known category: '{new_category}')"
+            )
+            try:
+                await nc_client.notes.delete_note(note_id=note_id)
+                logger.info(f"Note {note_id} deleted.")
+                time.sleep(1)
+                # Verify note deletion
+                with pytest.raises(HTTPStatusError) as excinfo_note_del:
+                    await nc_client.notes.get_note(note_id=note_id)
+                assert excinfo_note_del.value.response.status_code == 404
+                logger.info("Verified note deleted (404).")
+                # Verify attachment deletion (should fail with 404 on the initial note fetch)
+                with pytest.raises(HTTPStatusError) as excinfo_attach_del:
+                    # Pass the *last known* category, although the note fetch should fail first
+                    await nc_client.webdav.get_note_attachment(
+                        note_id=note_id,
+                        filename=attachment_filename,
+                        category=new_category,
+                    )
+                assert excinfo_attach_del.value.response.status_code == 404
+                logger.info(
+                    "Verified attachment cannot be retrieved after note deletion (404)."
+                )
+
+                # 6.1 Verify both old and new attachment directories are gone via WebDAV PROPFIND
+                logger.info(
+                    "Directly verifying attachment directories don't exist via PROPFIND"
+                )
+                webdav_base = nc_client._get_webdav_base_path()
+
+                # Check new category attachment directory
+                new_attachment_dir_path = (
+                    f"{webdav_base}/Notes/{new_category}/.attachments.{note_id}"
+                )
+                propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
+                try:
+                    resp = await nc_client._client.request(
+                        "PROPFIND", new_attachment_dir_path, headers=propfind_headers
+                    )
+                    if resp.status_code in [
+                        200,
+                        207,
+                    ]:  # Successful PROPFIND means directory exists
+                        assert False, "New category attachment directory still exists!"
+                except HTTPStatusError as e:
+                    assert e.response.status_code == 404, (
+                        f"Expected PROPFIND to fail with 404, got {e.response.status_code}"
+                    )
+                    logger.info(
+                        "Verified new category attachment directory is gone via PROPFIND"
+                    )
+
+                # Check old category attachment directory
+                old_attachment_dir_path = (
+                    f"{webdav_base}/Notes/{initial_category}/.attachments.{note_id}"
+                )
+                try:
+                    resp = await nc_client._client.request(
+                        "PROPFIND", old_attachment_dir_path, headers=propfind_headers
+                    )
+                    if resp.status_code in [
+                        200,
+                        207,
+                    ]:  # Successful PROPFIND means directory exists
+                        assert False, "Old category attachment directory still exists!"
+                except HTTPStatusError as e:
+                    assert e.response.status_code == 404, (
+                        f"Expected PROPFIND to fail with 404, got {e.response.status_code}"
+                    )
+                    logger.info(
+                        "Verified old category attachment directory is gone via PROPFIND"
+                    )
+
+                logger.info(
+                    "Verified all attachment directories are properly cleaned up."
+                )
+            except Exception as e:
+                logger.error(f"Error during cleanup for note {note_id}: {e}")

tests/integration/test_calendar_operations.py

@@ -0,0 +1,409 @@
+"""Integration tests for Calendar CalDAV operations."""
+
+import pytest
+import logging
+import uuid
+from datetime import datetime, timedelta
+from httpx import HTTPStatusError
+
+from nextcloud_mcp_server.client import NextcloudClient
+
+logger = logging.getLogger(__name__)
+
+# Mark all tests in this module as integration tests
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture
+def test_calendar_name():
+    """Unique calendar name for testing."""
+    return f"test_calendar_{uuid.uuid4().hex[:8]}"
+
+
+@pytest.fixture
+async def temporary_calendar(nc_client: NextcloudClient, test_calendar_name: str):
+    """Create a temporary calendar for testing and clean up afterward."""
+    calendar_name = test_calendar_name
+
+    try:
+        # Create a test calendar if possible
+        # Note: Calendar creation might require admin permissions
+        # For now, we'll use an existing calendar or create events in default calendar
+
+        # Try to find an existing calendar to use
+        calendars = await nc_client.calendar.list_calendars()
+        if calendars:
+            calendar_name = calendars[0]["name"]
+            logger.info(f"Using existing calendar: {calendar_name}")
+            yield calendar_name
+        else:
+            pytest.skip("No calendars available for testing")
+
+    except Exception as e:
+        logger.error(f"Error setting up temporary calendar: {e}")
+        pytest.skip(f"Calendar setup failed: {e}")
+
+
+@pytest.fixture
+async def temporary_event(nc_client: NextcloudClient, temporary_calendar: str):
+    """Create a temporary event for testing and clean up afterward."""
+    event_uid = None
+    calendar_name = temporary_calendar
+
+    # Create a test event
+    tomorrow = datetime.now() + timedelta(days=1)
+    event_data = {
+        "title": f"Test Event {uuid.uuid4().hex[:8]}",
+        "start_datetime": tomorrow.strftime("%Y-%m-%dT14:00:00"),
+        "end_datetime": tomorrow.strftime("%Y-%m-%dT15:00:00"),
+        "description": "Test event created by integration tests",
+        "location": "Test Location",
+        "categories": "testing",
+        "status": "CONFIRMED",
+        "priority": 5,
+    }
+
+    try:
+        logger.info(f"Creating temporary event in calendar: {calendar_name}")
+        result = await nc_client.calendar.create_event(calendar_name, event_data)
+        event_uid = result.get("uid")
+
+        if not event_uid:
+            pytest.fail("Failed to create temporary event")
+
+        logger.info(f"Created temporary event with UID: {event_uid}")
+        yield {"uid": event_uid, "calendar_name": calendar_name, "data": event_data}
+
+    finally:
+        # Cleanup
+        if event_uid:
+            try:
+                logger.info(f"Cleaning up temporary event: {event_uid}")
+                await nc_client.calendar.delete_event(calendar_name, event_uid)
+                logger.info(f"Successfully deleted temporary event: {event_uid}")
+            except HTTPStatusError as e:
+                if e.response.status_code != 404:
+                    logger.error(f"Error deleting temporary event {event_uid}: {e}")
+            except Exception as e:
+                logger.error(
+                    f"Unexpected error deleting temporary event {event_uid}: {e}"
+                )
+
+
+async def test_list_calendars(nc_client: NextcloudClient):
+    """Test listing available calendars."""
+    calendars = await nc_client.calendar.list_calendars()
+
+    assert isinstance(calendars, list)
+
+    if not calendars:
+        pytest.skip("No calendars available - Calendar app may not be enabled")
+
+    logger.info(f"Found {len(calendars)} calendars")
+
+    # Check structure of calendars
+    for calendar in calendars:
+        assert "name" in calendar
+        assert "display_name" in calendar
+        assert "href" in calendar
+        # Optional fields
+        assert "description" in calendar
+        assert "color" in calendar
+
+        logger.info(f"Calendar: {calendar['name']} - {calendar['display_name']}")
+
+
+async def test_create_and_delete_event(
+    nc_client: NextcloudClient, temporary_calendar: str
+):
+    """Test creating and deleting a basic event."""
+    calendar_name = temporary_calendar
+
+    # Create event
+    tomorrow = datetime.now() + timedelta(days=1)
+    event_data = {
+        "title": "Integration Test Event",
+        "start_datetime": tomorrow.strftime("%Y-%m-%dT10:00:00"),
+        "end_datetime": tomorrow.strftime("%Y-%m-%dT11:00:00"),
+        "description": "Test event for integration testing",
+        "location": "Test Room",
+        "categories": "testing,integration",
+        "status": "CONFIRMED",
+        "priority": 3,
+    }
+
+    try:
+        result = await nc_client.calendar.create_event(calendar_name, event_data)
+        assert "uid" in result
+        assert result["status_code"] in [200, 201, 204]
+
+        event_uid = result["uid"]
+        logger.info(f"Created event with UID: {event_uid}")
+
+        # Verify event was created by retrieving it
+        retrieved_event, etag = await nc_client.calendar.get_event(
+            calendar_name, event_uid
+        )
+        assert retrieved_event["uid"] == event_uid
+        assert retrieved_event["title"] == "Integration Test Event"
+        assert retrieved_event["location"] == "Test Room"
+
+        # Delete event
+        delete_result = await nc_client.calendar.delete_event(calendar_name, event_uid)
+        assert delete_result["status_code"] in [200, 204, 404]
+
+        logger.info(f"Successfully deleted event: {event_uid}")
+
+    except Exception as e:
+        logger.error(f"Test failed: {e}")
+        raise
+
+
+async def test_create_all_day_event(
+    nc_client: NextcloudClient, temporary_calendar: str
+):
+    """Test creating an all-day event."""
+    calendar_name = temporary_calendar
+
+    tomorrow = datetime.now() + timedelta(days=1)
+    event_data = {
+        "title": "All Day Test Event",
+        "start_datetime": tomorrow.strftime("%Y-%m-%d"),
+        "all_day": True,
+        "description": "Test all-day event",
+        "categories": "testing",
+    }
+
+    try:
+        result = await nc_client.calendar.create_event(calendar_name, event_data)
+        event_uid = result["uid"]
+        logger.info(f"Created all-day event with UID: {event_uid}")
+
+        # Verify event
+        retrieved_event, _ = await nc_client.calendar.get_event(
+            calendar_name, event_uid
+        )
+        assert retrieved_event["title"] == "All Day Test Event"
+        assert retrieved_event.get("all_day") is True
+
+        # Cleanup
+        await nc_client.calendar.delete_event(calendar_name, event_uid)
+
+    except Exception as e:
+        logger.error(f"All-day event test failed: {e}")
+        raise
+
+
+async def test_create_recurring_event(
+    nc_client: NextcloudClient, temporary_calendar: str
+):
+    """Test creating a recurring event."""
+    calendar_name = temporary_calendar
+
+    tomorrow = datetime.now() + timedelta(days=1)
+    event_data = {
+        "title": "Weekly Recurring Test",
+        "start_datetime": tomorrow.strftime("%Y-%m-%dT14:00:00"),
+        "end_datetime": tomorrow.strftime("%Y-%m-%dT15:00:00"),
+        "description": "Test recurring event",
+        "recurring": True,
+        "recurrence_rule": "FREQ=WEEKLY;BYDAY=MO,WE,FR",
+        "reminder_minutes": 30,
+    }
+
+    try:
+        result = await nc_client.calendar.create_event(calendar_name, event_data)
+        event_uid = result["uid"]
+        logger.info(f"Created recurring event with UID: {event_uid}")
+
+        # Verify event
+        retrieved_event, _ = await nc_client.calendar.get_event(
+            calendar_name, event_uid
+        )
+        assert retrieved_event["title"] == "Weekly Recurring Test"
+        assert retrieved_event.get("recurring") is True
+
+        # Cleanup
+        await nc_client.calendar.delete_event(calendar_name, event_uid)
+
+    except Exception as e:
+        logger.error(f"Recurring event test failed: {e}")
+        raise
+
+
+async def test_list_events_in_range(nc_client: NextcloudClient, temporary_event: dict):
+    """Test listing events within a date range."""
+    calendar_name = temporary_event["calendar_name"]
+
+    # Get events for the next week
+    start_date = datetime.now().strftime("%Y%m%dT000000Z")
+    end_date = (datetime.now() + timedelta(days=7)).strftime("%Y%m%dT235959Z")
+
+    events = await nc_client.calendar.get_calendar_events(
+        calendar_name=calendar_name, start_date=start_date, end_date=end_date, limit=50
+    )
+
+    assert isinstance(events, list)
+    logger.info(f"Found {len(events)} events in date range")
+
+    # Our temporary event should be in the list
+    event_uids = [event.get("uid") for event in events]
+    assert temporary_event["uid"] in event_uids
+
+    # Check event structure
+    for event in events:
+        assert "uid" in event
+        assert "title" in event
+        assert "start_datetime" in event
+
+
+async def test_update_event(nc_client: NextcloudClient, temporary_event: dict):
+    """Test updating an existing event."""
+    calendar_name = temporary_event["calendar_name"]
+    event_uid = temporary_event["uid"]
+
+    # Update event data
+    updated_data = {
+        "title": "Updated Test Event Title",
+        "description": "Updated description for test event",
+        "location": "Updated Location",
+        "priority": 1,  # High priority
+    }
+
+    try:
+        result = await nc_client.calendar.update_event(
+            calendar_name, event_uid, updated_data
+        )
+        assert result["uid"] == event_uid
+
+        # Verify updates
+        updated_event, _ = await nc_client.calendar.get_event(calendar_name, event_uid)
+        assert updated_event["title"] == "Updated Test Event Title"
+        assert updated_event["description"] == "Updated description for test event"
+        assert updated_event["location"] == "Updated Location"
+        assert updated_event["priority"] == 1
+
+        logger.info(f"Successfully updated event: {event_uid}")
+
+    except Exception as e:
+        logger.error(f"Event update test failed: {e}")
+        raise
+
+
+async def test_create_event_with_attendees(
+    nc_client: NextcloudClient, temporary_calendar: str
+):
+    """Test creating an event with attendees."""
+    calendar_name = temporary_calendar
+
+    tomorrow = datetime.now() + timedelta(days=1)
+    event_data = {
+        "title": "Meeting with Attendees",
+        "start_datetime": tomorrow.strftime("%Y-%m-%dT16:00:00"),
+        "end_datetime": tomorrow.strftime("%Y-%m-%dT17:00:00"),
+        "description": "Test meeting with multiple attendees",
+        "location": "Conference Room A",
+        "attendees": "test1@example.com,test2@example.com",
+        "reminder_minutes": 15,
+        "status": "TENTATIVE",
+    }
+
+    try:
+        result = await nc_client.calendar.create_event(calendar_name, event_data)
+        event_uid = result["uid"]
+        logger.info(f"Created event with attendees, UID: {event_uid}")
+
+        # Verify event
+        retrieved_event, _ = await nc_client.calendar.get_event(
+            calendar_name, event_uid
+        )
+        assert retrieved_event["title"] == "Meeting with Attendees"
+        assert "test1@example.com" in retrieved_event.get("attendees", "")
+        assert retrieved_event["status"] == "TENTATIVE"
+
+        # Cleanup
+        await nc_client.calendar.delete_event(calendar_name, event_uid)
+
+    except Exception as e:
+        logger.error(f"Event with attendees test failed: {e}")
+        raise
+
+
+async def test_get_nonexistent_event(
+    nc_client: NextcloudClient, temporary_calendar: str
+):
+    """Test retrieving a non-existent event."""
+    calendar_name = temporary_calendar
+    fake_uid = f"nonexistent-{uuid.uuid4()}"
+
+    with pytest.raises(HTTPStatusError) as exc_info:
+        await nc_client.calendar.get_event(calendar_name, fake_uid)
+
+    assert exc_info.value.response.status_code == 404
+    logger.info(f"Correctly got 404 for nonexistent event: {fake_uid}")
+
+
+async def test_delete_nonexistent_event(
+    nc_client: NextcloudClient, temporary_calendar: str
+):
+    """Test deleting a non-existent event."""
+    calendar_name = temporary_calendar
+    fake_uid = f"nonexistent-{uuid.uuid4()}"
+
+    result = await nc_client.calendar.delete_event(calendar_name, fake_uid)
+    assert result["status_code"] == 404
+    logger.info(f"Correctly got 404 for deleting nonexistent event: {fake_uid}")
+
+
+async def test_event_with_url_and_categories(
+    nc_client: NextcloudClient, temporary_calendar: str
+):
+    """Test creating an event with URL and multiple categories."""
+    calendar_name = temporary_calendar
+
+    tomorrow = datetime.now() + timedelta(days=1)
+    event_data = {
+        "title": "Event with URL and Categories",
+        "start_datetime": tomorrow.strftime("%Y-%m-%dT09:00:00"),
+        "end_datetime": tomorrow.strftime("%Y-%m-%dT10:30:00"),
+        "description": "Test event with additional metadata",
+        "categories": "work,meeting,important,quarterly",
+        "url": "https://zoom.us/j/123456789",
+        "privacy": "PRIVATE",
+        "priority": 2,
+    }
+
+    try:
+        result = await nc_client.calendar.create_event(calendar_name, event_data)
+        event_uid = result["uid"]
+        logger.info(f"Created event with metadata, UID: {event_uid}")
+
+        # Verify event
+        retrieved_event, _ = await nc_client.calendar.get_event(
+            calendar_name, event_uid
+        )
+        assert retrieved_event["title"] == "Event with URL and Categories"
+        assert "work" in retrieved_event.get("categories", "")
+        assert "important" in retrieved_event.get("categories", "")
+        assert retrieved_event.get("url") == "https://zoom.us/j/123456789"
+        assert retrieved_event.get("privacy") == "PRIVATE"
+        assert retrieved_event.get("priority") == 2
+
+        # Cleanup
+        await nc_client.calendar.delete_event(calendar_name, event_uid)
+
+    except Exception as e:
+        logger.error(f"Event with metadata test failed: {e}")
+        raise
+
+
+async def test_calendar_operations_error_handling(nc_client: NextcloudClient):
+    """Test error handling for calendar operations."""
+
+    # Test with non-existent calendar
+    fake_calendar = f"nonexistent_calendar_{uuid.uuid4().hex}"
+
+    with pytest.raises(HTTPStatusError):
+        await nc_client.calendar.get_calendar_events(fake_calendar)
+
+    logger.info("Error handling tests completed successfully")

tests/integration/test_embedded_images.py

@@ -0,0 +1,186 @@
+import pytest
+import time
+import uuid
+import logging
+from PIL import Image, ImageDraw
+from io import BytesIO
+from httpx import HTTPStatusError  # Import if needed for specific error checks
+
+from nextcloud_mcp_server.client import NextcloudClient
+
+# Note: nc_client fixture is session-scoped in conftest.py
+# Note: temporary_note fixture is function-scoped in conftest.py
+
+logger = logging.getLogger(__name__)
+
+# Mark all tests in this module as integration tests
+pytestmark = pytest.mark.integration
+
+
+# Keep the test_image fixture as it's specific to generating image data
+@pytest.fixture(scope="module")  # Keep module scope if image generation is slow
+def test_image_data() -> tuple[bytes, str]:
+    """
+    Generate test image data (bytes) and suggest a filename.
+    Returns (image_bytes, suggested_filename).
+    """
+    logger.info("Generating test image data in memory.")
+    img = Image.new("RGB", (300, 200), color=(255, 255, 255))
+    draw = ImageDraw.Draw(img)
+    draw.rectangle([(20, 20), (280, 180)], fill=(0, 120, 212))  # Blue rectangle
+    draw.text(
+        (50, 90), "Nextcloud Notes Test Image", fill=(255, 255, 255)
+    )  # White text
+
+    img_byte_arr = BytesIO()
+    img.save(img_byte_arr, format="PNG")
+    image_bytes = img_byte_arr.getvalue()
+    suggested_filename = "test_image.png"
+    logger.info(f"Generated test image data ({len(image_bytes)} bytes).")
+    return image_bytes, suggested_filename
+
+
+async def test_note_with_embedded_image(
+    nc_client: NextcloudClient, temporary_note: dict, test_image_data: tuple
+):
+    """
+    Tests creating a note, attaching an image, embedding it in the content,
+    and verifying the attachment can be retrieved.
+    """
+    note_data = temporary_note  # Use fixture for note creation/cleanup
+    note_id = note_data["id"]
+    note_etag = note_data["etag"]
+    image_content, suggested_filename = test_image_data  # Get image data from fixture
+
+    unique_suffix = uuid.uuid4().hex[:8]
+    attachment_filename = (
+        f"test_image_{unique_suffix}.png"  # Make filename unique per run
+    )
+
+    # 1. Upload the image as an attachment
+    note_category = note_data.get("category")  # Get category from fixture data
+    logger.info(
+        f"Uploading image attachment '{attachment_filename}' to note {note_id} (category: '{note_category or ''}')..."
+    )
+    upload_response = await nc_client.webdav.add_note_attachment(
+        note_id=note_id,
+        filename=attachment_filename,
+        content=image_content,
+        category=note_category,  # Pass the category
+        mime_type="image/png",
+    )
+    assert upload_response and upload_response.get("status_code") in [201, 204]
+    logger.info(
+        f"Image uploaded successfully (Status: {upload_response.get('status_code')})."
+    )
+    time.sleep(1)  # Allow potential processing time
+
+    # 1.1 Verify attachment directory exists via WebDAV PROPFIND
+    logger.info("Directly checking if attachment directory exists in WebDAV")
+    webdav_base = nc_client._get_webdav_base_path()
+    category_path_part = f"{note_category}/" if note_category else ""
+    attachment_dir_path = (
+        f"{webdav_base}/Notes/{category_path_part}.attachments.{note_id}"
+    )
+    propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
+    try:
+        propfind_resp = await nc_client._client.request(
+            "PROPFIND", attachment_dir_path, headers=propfind_headers
+        )
+        status = propfind_resp.status_code
+        assert status in [
+            207,
+            200,
+        ], f"Expected PROPFIND to return success (207/200), got {status}"
+        logger.info(
+            f"Verified attachment directory exists via PROPFIND ({status} received)"
+        )
+    except HTTPStatusError as e:
+        logger.error(
+            f"Attachment directory not found! PROPFIND failed with {e.response.status_code}"
+        )
+        assert False, (
+            f"Expected attachment directory to exist, but PROPFIND failed with {e.response.status_code}"
+        )
+
+    # 2. Update the note content to include the embedded image references
+    updated_content = f"""{note_data["content"]}
+
+## Image Embedding Test
+
+### Markdown Syntax
+![Test Image MD](.attachments.{note_id}/{attachment_filename})
+
+### HTML Syntax
+<img src=".attachments.{note_id}/{attachment_filename}" alt="Test Image HTML" width="150" />
+"""
+    logger.info("Updating note content with image references...")
+    updated_note = await nc_client.notes.update(
+        note_id=note_id,
+        etag=note_etag,  # Use etag from the created note
+        content=updated_content,
+        title=note_data["title"],  # Pass required fields
+        category=note_data["category"],  # Pass required fields
+    )
+    new_etag = updated_note["etag"]
+    assert new_etag != note_etag
+    logger.info("Note content updated with image references.")
+    time.sleep(1)
+
+    # 3. Verify the updated note content
+    retrieved_note = await nc_client.notes.get_note(note_id=note_id)
+    assert f".attachments.{note_id}/{attachment_filename}" in retrieved_note["content"]
+    logger.info("Verified image reference exists in updated note content.")
+
+    # 4. Verify the image attachment can be retrieved
+    logger.info(
+        f"Retrieving image attachment '{attachment_filename}' (category: '{note_category or ''}')..."
+    )
+    # Pass category to get_note_attachment
+    retrieved_img_content, mime_type = await nc_client.webdav.get_note_attachment(
+        note_id=note_id, filename=attachment_filename, category=note_category
+    )
+    assert retrieved_img_content == image_content
+    assert mime_type.startswith("image/png")
+    logger.info(
+        "Successfully retrieved and verified image attachment content and mime type."
+    )
+
+    # 5. Manually trigger deletion to verify cleanup (instead of waiting for fixture teardown)
+    logger.info(
+        f"Manually deleting note ID: {note_id} to verify proper attachment cleanup"
+    )
+    await nc_client.notes.delete_note(note_id=note_id)
+    logger.info(f"Note ID: {note_id} deleted successfully.")
+    time.sleep(1)
+
+    # 6. Verify note is deleted
+    with pytest.raises(HTTPStatusError) as excinfo_note:
+        await nc_client.notes.get_note(note_id=note_id)
+    assert excinfo_note.value.response.status_code == 404
+    logger.info(f"Verified note {note_id} deletion (404 received).")
+
+    # 7. Verify attachment directory is deleted via WebDAV PROPFIND
+    logger.info("Directly verifying attachment directory doesn't exist via PROPFIND")
+    try:
+        propfind_resp = await nc_client._client.request(
+            "PROPFIND", attachment_dir_path, headers=propfind_headers
+        )
+        status = propfind_resp.status_code
+        if status in [200, 207]:  # Successful PROPFIND means directory exists
+            logger.error(
+                f"Attachment directory still exists! PROPFIND returned {status}"
+            )
+            assert False, (
+                f"Expected attachment directory to be gone, but PROPFIND returned {status}!"
+            )
+    except HTTPStatusError as e:
+        assert e.response.status_code == 404, (
+            f"Expected PROPFIND to fail with 404, got {e.response.status_code}"
+        )
+        logger.info(
+            "Verified attachment directory does not exist via PROPFIND (404 received)"
+        )
+
+    # Note: The temporary_note fixture will still run its cleanup,
+    # but it will find the note already deleted (404) and handle it gracefully.

tests/integration/test_notes_api.py

@@ -0,0 +1,259 @@
+import pytest
+import logging
+import asyncio
+import uuid  # Keep uuid if needed for generating unique data within tests
+from httpx import HTTPStatusError
+
+from nextcloud_mcp_server.client import NextcloudClient
+
+# Note: nc_client fixture is now session-scoped in conftest.py
+
+logger = logging.getLogger(__name__)
+
+# Mark all tests in this module as integration tests
+pytestmark = pytest.mark.integration
+
+
+async def test_notes_api_create_and_read(
+    nc_client: NextcloudClient, temporary_note: dict
+):
+    """
+    Tests creating a note via the API (using fixture) and then reading it back.
+    """
+    created_note_data = temporary_note  # Get data from fixture
+    note_id = created_note_data["id"]
+
+    logger.info(f"Reading note created by fixture, ID: {note_id}")
+    read_note = await nc_client.notes.get_note(note_id=note_id)
+
+    assert read_note["id"] == note_id
+    assert read_note["title"] == created_note_data["title"]
+    assert read_note["content"] == created_note_data["content"]
+    assert read_note["category"] == created_note_data["category"]
+    logger.info(f"Successfully read and verified note ID: {note_id}")
+
+
+async def test_notes_api_update(nc_client: NextcloudClient, temporary_note: dict):
+    """
+    Tests updating a note created by the fixture.
+    """
+    created_note_data = temporary_note
+    note_id = created_note_data["id"]
+    original_etag = created_note_data["etag"]
+    original_category = created_note_data["category"]
+
+    update_title = f"Updated Title {uuid.uuid4().hex[:8]}"
+    update_content = f"Updated Content {uuid.uuid4().hex[:8]}"
+
+    logger.info(f"Attempting to update note ID: {note_id} with etag: {original_etag}")
+    updated_note = await nc_client.notes.update(
+        note_id=note_id,
+        etag=original_etag,
+        title=update_title,
+        content=update_content,
+        # category=original_category # Explicitly pass category if required by update
+    )
+    logger.info(f"Note updated: {updated_note}")
+
+    assert updated_note["id"] == note_id
+    assert updated_note["title"] == update_title
+    assert updated_note["content"] == update_content
+    assert (
+        updated_note["category"] == original_category
+    )  # Verify category didn't change
+    assert "etag" in updated_note
+    assert updated_note["etag"] != original_etag  # Etag must change
+
+    # Optional: Verify update by reading again
+    await asyncio.sleep(1)  # Allow potential propagation delay
+    read_updated_note = await nc_client.notes.get_note(note_id=note_id)
+    assert read_updated_note["title"] == update_title
+    assert read_updated_note["content"] == update_content
+    logger.info(f"Successfully updated and verified note ID: {note_id}")
+
+
+async def test_notes_api_update_conflict(
+    nc_client: NextcloudClient, temporary_note: dict
+):
+    """
+    Tests that attempting to update with an old etag fails with 412.
+    """
+    created_note_data = temporary_note
+    note_id = created_note_data["id"]
+    original_etag = created_note_data["etag"]
+
+    # Perform a first update to change the etag
+    first_update_title = f"First Update {uuid.uuid4().hex[:8]}"
+    logger.info(f"Performing first update on note ID: {note_id} to change etag.")
+    first_updated_note = await nc_client.notes.update(
+        note_id=note_id,
+        etag=original_etag,
+        title=first_update_title,
+        content="First update content",
+        # category=created_note_data["category"] # Pass category if required
+    )
+    new_etag = first_updated_note["etag"]
+    assert new_etag != original_etag
+    logger.info(f"Note ID: {note_id} updated, new etag: {new_etag}")
+    await asyncio.sleep(1)
+
+    # Now attempt update with the *original* etag
+    logger.info(
+        f"Attempting second update on note ID: {note_id} with OLD etag: {original_etag}"
+    )
+    with pytest.raises(HTTPStatusError) as excinfo:
+        await nc_client.notes.update(
+            note_id=note_id,
+            etag=original_etag,  # Use the stale etag
+            title="This update should fail due to conflict",
+            # category=created_note_data["category"] # Pass category if required
+        )
+    assert excinfo.value.response.status_code == 412  # Precondition Failed
+    logger.info("Update with old etag correctly failed with 412 Precondition Failed.")
+
+
+async def test_notes_api_delete_nonexistent(nc_client: NextcloudClient):
+    """
+    Tests deleting a note that doesn't exist fails with 404.
+    """
+    non_existent_id = 999999999  # Use an ID highly unlikely to exist
+    logger.info(f"\nAttempting to delete non-existent note ID: {non_existent_id}")
+    with pytest.raises(HTTPStatusError) as excinfo:
+        await nc_client.notes.delete_note(note_id=non_existent_id)
+    assert excinfo.value.response.status_code == 404
+    logger.info(
+        f"Deleting non-existent note ID: {non_existent_id} correctly failed with 404."
+    )
+
+
+async def test_notes_api_append_content_to_existing_note(
+    nc_client: NextcloudClient, temporary_note: dict
+):
+    """
+    Tests appending content to an existing note using the new append functionality.
+    """
+    created_note_data = temporary_note
+    note_id = created_note_data["id"]
+    original_content = created_note_data["content"]
+
+    append_text = f"Appended content {uuid.uuid4().hex[:8]}"
+
+    logger.info(f"Appending content to note ID: {note_id}")
+    updated_note = await nc_client.notes.append_content(
+        note_id=note_id, content=append_text
+    )
+    logger.info(f"Note after append: {updated_note}")
+
+    # Verify the note was updated
+    assert updated_note["id"] == note_id
+    assert "etag" in updated_note
+    assert updated_note["etag"] != created_note_data["etag"]  # Etag must change
+
+    # Verify content has the separator and appended text
+    expected_content = original_content + "\n---\n" + append_text
+    assert updated_note["content"] == expected_content
+
+    # Verify by reading the note again
+    await asyncio.sleep(1)  # Allow potential propagation delay
+    read_note = await nc_client.notes.get_note(note_id=note_id)
+    assert read_note["content"] == expected_content
+    logger.info(f"Successfully appended content to note ID: {note_id}")
+
+
+async def test_notes_api_append_content_to_empty_note(nc_client: NextcloudClient):
+    """
+    Tests appending content to an empty note (no separator should be added).
+    """
+    # Create an empty note
+    test_title = f"Empty Note {uuid.uuid4().hex[:8]}"
+    test_category = "Test"
+
+    logger.info("Creating empty note for append test")
+    empty_note = await nc_client.notes.create_note(
+        title=test_title,
+        content="",
+        category=test_category,  # Empty content
+    )
+    note_id = empty_note["id"]
+
+    try:
+        append_text = f"First content {uuid.uuid4().hex[:8]}"
+
+        logger.info(f"Appending content to empty note ID: {note_id}")
+        updated_note = await nc_client.notes.append_content(
+            note_id=note_id, content=append_text
+        )
+
+        # For empty notes, content should just be the appended text (no separator)
+        assert updated_note["content"] == append_text
+
+        # Verify by reading the note again
+        await asyncio.sleep(1)
+        read_note = await nc_client.notes.get_note(note_id=note_id)
+        assert read_note["content"] == append_text
+        logger.info(f"Successfully appended content to empty note ID: {note_id}")
+
+    finally:
+        # Clean up the test note
+        try:
+            await nc_client.notes.delete_note(note_id=note_id)
+            logger.info(f"Cleaned up test note ID: {note_id}")
+        except Exception as e:
+            logger.warning(f"Failed to clean up test note ID: {note_id}: {e}")
+
+
+async def test_notes_api_append_content_multiple_times(
+    nc_client: NextcloudClient, temporary_note: dict
+):
+    """
+    Tests appending content multiple times to verify separator behavior.
+    """
+    created_note_data = temporary_note
+    note_id = created_note_data["id"]
+    original_content = created_note_data["content"]
+
+    first_append = f"First append {uuid.uuid4().hex[:8]}"
+    second_append = f"Second append {uuid.uuid4().hex[:8]}"
+
+    logger.info(f"Performing multiple appends to note ID: {note_id}")
+
+    # First append
+    updated_note = await nc_client.notes.append_content(
+        note_id=note_id, content=first_append
+    )
+
+    expected_content_after_first = original_content + "\n---\n" + first_append
+    assert updated_note["content"] == expected_content_after_first
+
+    # Second append
+    updated_note = await nc_client.notes.append_content(
+        note_id=note_id, content=second_append
+    )
+
+    expected_content_after_second = (
+        expected_content_after_first + "\n---\n" + second_append
+    )
+    assert updated_note["content"] == expected_content_after_second
+
+    # Verify by reading the note again
+    await asyncio.sleep(1)
+    read_note = await nc_client.notes.get_note(note_id=note_id)
+    assert read_note["content"] == expected_content_after_second
+    logger.info(f"Successfully performed multiple appends to note ID: {note_id}")
+
+
+async def test_notes_api_append_content_nonexistent_note(nc_client: NextcloudClient):
+    """
+    Tests that appending to a non-existent note fails with 404.
+    """
+    non_existent_id = 999999999
+
+    logger.info(f"Attempting to append to non-existent note ID: {non_existent_id}")
+    with pytest.raises(HTTPStatusError) as excinfo:
+        await nc_client.notes.append_content(
+            note_id=non_existent_id, content="This should fail"
+        )
+    assert excinfo.value.response.status_code == 404
+    logger.info(
+        f"Appending to non-existent note ID: {non_existent_id} correctly failed with 404."
+    )

tests/integration/test_tables_api.py

@@ -0,0 +1,534 @@
+import pytest
+import logging
+import asyncio
+import uuid
+from httpx import HTTPStatusError
+from typing import Dict, Any
+
+from nextcloud_mcp_server.client import NextcloudClient
+
+logger = logging.getLogger(__name__)
+
+# Mark all tests in this module as integration tests
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture(scope="session")
+async def sample_table_info(nc_client: NextcloudClient) -> Dict[str, Any]:
+    """
+    Fixture to get information about the sample table that comes with Nextcloud Tables.
+    This assumes that the sample table exists in the Nextcloud instance.
+    """
+    logger.info("Looking for sample table in Nextcloud Tables app")
+
+    # Get all tables
+    tables = await nc_client.tables.list_tables()
+
+    # Look for a sample table (usually created by default)
+    sample_table = None
+    for table in tables:
+        # Common names for sample tables
+        if any(
+            keyword in table.get("title", "").lower()
+            for keyword in ["sample", "demo", "example", "test"]
+        ):
+            sample_table = table
+            break
+
+    if not sample_table and tables:
+        # If no sample table found, use the first available table
+        sample_table = tables[0]
+        logger.info(
+            f"No sample table found, using first available table: {sample_table.get('title')}"
+        )
+
+    if not sample_table:
+        pytest.skip(
+            "No tables found in Nextcloud Tables app. Please ensure Tables app is installed and has at least one table."
+        )
+
+    # Get the schema for the sample table
+    table_id = sample_table["id"]
+    schema = await nc_client.tables.get_table_schema(table_id)
+
+    logger.info(f"Using sample table: {sample_table.get('title')} (ID: {table_id})")
+
+    return {
+        "table": sample_table,
+        "schema": schema,
+        "table_id": table_id,
+        "columns": schema.get("columns", []),
+    }
+
+
+@pytest.fixture
+async def temporary_table_row(
+    nc_client: NextcloudClient, sample_table_info: Dict[str, Any]
+):
+    """
+    Fixture to create a temporary row in the sample table for testing.
+    Yields the created row data and cleans up afterward.
+    """
+    table_id = sample_table_info["table_id"]
+    columns = sample_table_info["columns"]
+
+    # Create test data based on the table schema
+    test_data = {}
+    unique_suffix = uuid.uuid4().hex[:8]
+
+    for column in columns:
+        column_id = column["id"]
+        column_type = column.get("type", "text")
+        column_title = column.get("title", f"column_{column_id}")
+
+        # Generate test data based on column type
+        if column_type == "text":
+            test_data[column_id] = f"Test {column_title} {unique_suffix}"
+        elif column_type == "number":
+            test_data[column_id] = 42
+        elif column_type == "datetime":
+            test_data[column_id] = "2024-01-01T12:00:00Z"
+        elif column_type == "select":
+            # For select columns, use the first option if available
+            options = column.get("selectOptions", [])
+            if options:
+                test_data[column_id] = options[0].get("label", "Option 1")
+            else:
+                test_data[column_id] = "Test Option"
+        else:
+            # Default to text for unknown types
+            test_data[column_id] = f"Test {column_title} {unique_suffix}"
+
+    logger.info(f"Creating temporary row in table {table_id} with data: {test_data}")
+
+    created_row = None
+    try:
+        created_row = await nc_client.tables.create_row(table_id, test_data)
+        row_id = created_row.get("id")
+
+        if not row_id:
+            pytest.fail("Failed to get ID from created temporary row.")
+
+        logger.info(f"Temporary row created with ID: {row_id}")
+        yield created_row
+
+    finally:
+        if created_row and created_row.get("id"):
+            row_id = created_row["id"]
+            logger.info(f"Cleaning up temporary row ID: {row_id}")
+            try:
+                await nc_client.tables.delete_row(row_id)
+                logger.info(f"Successfully deleted temporary row ID: {row_id}")
+            except HTTPStatusError as e:
+                # Ignore 404 if row was already deleted by the test itself
+                if e.response.status_code != 404:
+                    logger.error(f"HTTP error deleting temporary row {row_id}: {e}")
+                else:
+                    logger.warning(f"Temporary row {row_id} already deleted (404).")
+            except Exception as e:
+                logger.error(f"Unexpected error deleting temporary row {row_id}: {e}")
+
+
+async def test_tables_list_tables(nc_client: NextcloudClient):
+    """
+    Test listing all tables available to the user.
+    """
+    logger.info("Testing list_tables functionality")
+
+    tables = await nc_client.tables.list_tables()
+
+    assert isinstance(tables, list)
+    assert len(tables) > 0, "Expected at least one table to be available"
+
+    # Check that each table has required fields
+    for table in tables:
+        assert "id" in table
+        assert "title" in table
+        assert isinstance(table["id"], int)
+        assert isinstance(table["title"], str)
+
+    logger.info(f"Successfully listed {len(tables)} tables")
+
+
+async def test_tables_get_schema(
+    nc_client: NextcloudClient, sample_table_info: Dict[str, Any]
+):
+    """
+    Test getting the schema/structure of a specific table.
+    """
+    table_id = sample_table_info["table_id"]
+
+    logger.info(f"Testing get_table_schema for table ID: {table_id}")
+
+    schema = await nc_client.tables.get_table_schema(table_id)
+
+    assert isinstance(schema, dict)
+    assert "columns" in schema
+    assert isinstance(schema["columns"], list)
+    assert len(schema["columns"]) > 0, "Expected at least one column in the table"
+
+    # Check that each column has required fields
+    for column in schema["columns"]:
+        assert "id" in column
+        assert "title" in column
+        assert "type" in column
+        assert isinstance(column["id"], int)
+        assert isinstance(column["title"], str)
+        assert isinstance(column["type"], str)
+
+    logger.info(f"Successfully retrieved schema with {len(schema['columns'])} columns")
+
+
+async def test_tables_read_table(
+    nc_client: NextcloudClient, sample_table_info: Dict[str, Any]
+):
+    """
+    Test reading rows from a table.
+    """
+    table_id = sample_table_info["table_id"]
+
+    logger.info(f"Testing get_table_rows for table ID: {table_id}")
+
+    # Test without pagination
+    rows = await nc_client.tables.get_table_rows(table_id)
+
+    assert isinstance(rows, list)
+    # Note: The table might be empty, so we don't assert len > 0
+
+    # Test with pagination
+    rows_limited = await nc_client.tables.get_table_rows(table_id, limit=5, offset=0)
+
+    assert isinstance(rows_limited, list)
+    assert len(rows_limited) <= 5
+
+    # If there are rows, check their structure
+    if rows:
+        row = rows[0]
+        assert "id" in row
+        assert "tableId" in row
+        assert "data" in row
+        assert isinstance(row["id"], int)
+        assert isinstance(row["tableId"], int)
+        assert isinstance(row["data"], list)
+
+    logger.info(f"Successfully read {len(rows)} rows from table")
+
+
+async def test_tables_create_row(
+    nc_client: NextcloudClient, sample_table_info: Dict[str, Any]
+):
+    """
+    Test creating a new row in a table.
+    """
+    table_id = sample_table_info["table_id"]
+    columns = sample_table_info["columns"]
+
+    # Create test data based on the table schema
+    test_data = {}
+    unique_suffix = uuid.uuid4().hex[:8]
+
+    for column in columns:
+        column_id = column["id"]
+        column_type = column.get("type", "text")
+        column_title = column.get("title", f"column_{column_id}")
+
+        # Generate test data based on column type
+        if column_type == "text":
+            test_data[column_id] = f"Test Create {column_title} {unique_suffix}"
+        elif column_type == "number":
+            test_data[column_id] = 123
+        elif column_type == "datetime":
+            test_data[column_id] = "2024-01-01T12:00:00Z"
+        elif column_type == "select":
+            # For select columns, use the first option if available
+            options = column.get("selectOptions", [])
+            if options:
+                test_data[column_id] = options[0].get("label", "Option 1")
+            else:
+                test_data[column_id] = "Test Option"
+        else:
+            # Default to text for unknown types
+            test_data[column_id] = f"Test Create {column_title} {unique_suffix}"
+
+    logger.info(f"Testing create_row for table ID: {table_id} with data: {test_data}")
+
+    created_row = None
+    try:
+        created_row = await nc_client.tables.create_row(table_id, test_data)
+
+        assert isinstance(created_row, dict)
+        assert "id" in created_row
+        assert "tableId" in created_row
+        assert isinstance(created_row["id"], int)
+        assert created_row["tableId"] == table_id
+
+        # Verify the row was created by reading it back
+        await asyncio.sleep(1)  # Allow potential propagation delay
+        rows = await nc_client.tables.get_table_rows(table_id)
+        created_row_id = created_row["id"]
+
+        # Find the created row in the results
+        found_row = None
+        for row in rows:
+            if row["id"] == created_row_id:
+                found_row = row
+                break
+
+        assert found_row is not None, (
+            f"Created row with ID {created_row_id} not found in table"
+        )
+
+        logger.info(f"Successfully created row with ID: {created_row_id}")
+
+    finally:
+        # Clean up the created row
+        if created_row and created_row.get("id"):
+            try:
+                await nc_client.tables.delete_row(created_row["id"])
+                logger.info(f"Cleaned up created row ID: {created_row['id']}")
+            except Exception as e:
+                logger.warning(f"Failed to clean up created row: {e}")
+
+
+async def test_tables_update_row(
+    nc_client: NextcloudClient,
+    temporary_table_row: Dict[str, Any],
+    sample_table_info: Dict[str, Any],
+):
+    """
+    Test updating an existing row in a table.
+    """
+    row_id = temporary_table_row["id"]
+    columns = sample_table_info["columns"]
+
+    # Create updated data
+    update_data = {}
+    unique_suffix = uuid.uuid4().hex[:8]
+
+    for column in columns:
+        column_id = column["id"]
+        column_type = column.get("type", "text")
+        column_title = column.get("title", f"column_{column_id}")
+
+        # Generate updated test data based on column type
+        if column_type == "text":
+            update_data[column_id] = f"Updated {column_title} {unique_suffix}"
+        elif column_type == "number":
+            update_data[column_id] = 456
+        elif column_type == "datetime":
+            update_data[column_id] = "2024-12-31T23:59:59Z"
+        elif column_type == "select":
+            # For select columns, use the first option if available
+            options = column.get("selectOptions", [])
+            if options:
+                update_data[column_id] = options[0].get("label", "Option 1")
+            else:
+                update_data[column_id] = "Updated Option"
+        else:
+            # Default to text for unknown types
+            update_data[column_id] = f"Updated {column_title} {unique_suffix}"
+
+    logger.info(f"Testing update_row for row ID: {row_id} with data: {update_data}")
+
+    updated_row = await nc_client.tables.update_row(row_id, update_data)
+
+    assert isinstance(updated_row, dict)
+    assert "id" in updated_row
+    assert updated_row["id"] == row_id
+
+    # Verify the row was updated by reading it back
+    await asyncio.sleep(1)  # Allow potential propagation delay
+    table_id = sample_table_info["table_id"]
+    rows = await nc_client.tables.get_table_rows(table_id)
+
+    # Find the updated row in the results
+    found_row = None
+    for row in rows:
+        if row["id"] == row_id:
+            found_row = row
+            break
+
+    assert found_row is not None, f"Updated row with ID {row_id} not found in table"
+
+    logger.info(f"Successfully updated row with ID: {row_id}")
+
+
+async def test_tables_delete_row(
+    nc_client: NextcloudClient, sample_table_info: Dict[str, Any]
+):
+    """
+    Test deleting a row from a table.
+    """
+    table_id = sample_table_info["table_id"]
+    columns = sample_table_info["columns"]
+
+    # First create a row to delete
+    test_data = {}
+    unique_suffix = uuid.uuid4().hex[:8]
+
+    for column in columns:
+        column_id = column["id"]
+        column_type = column.get("type", "text")
+        column_title = column.get("title", f"column_{column_id}")
+
+        if column_type == "text":
+            test_data[column_id] = f"Test Delete {column_title} {unique_suffix}"
+        elif column_type == "number":
+            test_data[column_id] = 789
+        elif column_type == "datetime":
+            test_data[column_id] = "2024-06-15T10:30:00Z"
+        elif column_type == "select":
+            options = column.get("selectOptions", [])
+            if options:
+                test_data[column_id] = options[0].get("label", "Option 1")
+            else:
+                test_data[column_id] = "Delete Option"
+        else:
+            test_data[column_id] = f"Test Delete {column_title} {unique_suffix}"
+
+    logger.info(f"Creating row for delete test in table ID: {table_id}")
+
+    created_row = await nc_client.tables.create_row(table_id, test_data)
+    row_id = created_row["id"]
+
+    logger.info(f"Testing delete_row for row ID: {row_id}")
+
+    # Delete the row
+    delete_result = await nc_client.tables.delete_row(row_id)
+
+    assert isinstance(delete_result, dict)
+    # The delete response might vary, but it should be successful
+
+    # Verify the row was deleted by trying to find it
+    await asyncio.sleep(1)  # Allow potential propagation delay
+    rows = await nc_client.tables.get_table_rows(table_id)
+
+    # Ensure the deleted row is not in the results
+    found_row = None
+    for row in rows:
+        if row["id"] == row_id:
+            found_row = row
+            break
+
+    assert found_row is None, f"Deleted row with ID {row_id} still found in table"
+
+    logger.info(f"Successfully deleted row with ID: {row_id}")
+
+
+async def test_tables_delete_nonexistent_row(nc_client: NextcloudClient):
+    """
+    Test that deleting a non-existent row fails appropriately.
+    """
+    non_existent_id = 999999999  # Use an ID highly unlikely to exist
+
+    logger.info(f"Testing delete_row for non-existent row ID: {non_existent_id}")
+
+    with pytest.raises(HTTPStatusError) as excinfo:
+        await nc_client.tables.delete_row(non_existent_id)
+
+    # Accept both 404 and 500 as valid error responses for non-existent rows
+    # The API behavior may vary between Nextcloud versions
+    assert excinfo.value.response.status_code in [404, 500]
+    logger.info(
+        f"Deleting non-existent row ID: {non_existent_id} correctly failed with {excinfo.value.response.status_code}."
+    )
+
+
+async def test_tables_transform_row_data(
+    nc_client: NextcloudClient, sample_table_info: Dict[str, Any]
+):
+    """
+    Test the transform_row_data utility method.
+    """
+    table_id = sample_table_info["table_id"]
+    columns = sample_table_info["columns"]
+
+    logger.info(f"Testing transform_row_data for table ID: {table_id}")
+
+    # Get some rows to transform
+    rows = await nc_client.tables.get_table_rows(table_id, limit=5)
+
+    if not rows:
+        logger.info("No rows to transform, skipping transform_row_data test")
+        return
+
+    # Transform the rows
+    transformed_rows = nc_client.tables.transform_row_data(rows, columns)
+
+    assert isinstance(transformed_rows, list)
+    assert len(transformed_rows) == len(rows)
+
+    # Check the structure of transformed rows
+    for i, transformed_row in enumerate(transformed_rows):
+        original_row = rows[i]
+
+        assert "id" in transformed_row
+        assert "tableId" in transformed_row
+        assert "data" in transformed_row
+        assert transformed_row["id"] == original_row["id"]
+        assert transformed_row["tableId"] == original_row["tableId"]
+        assert isinstance(transformed_row["data"], dict)
+
+        # Check that column IDs were transformed to column names
+        for column in columns:
+            column_title = column["title"]
+            # The transformed data should have column names as keys
+            # (though the column might not have data in this row)
+            if any(item["columnId"] == column["id"] for item in original_row["data"]):
+                assert column_title in transformed_row["data"]
+
+    logger.info(f"Successfully transformed {len(transformed_rows)} rows")
+
+
+async def test_tables_get_nonexistent_table_schema(nc_client: NextcloudClient):
+    """
+    Test that getting schema for a non-existent table fails appropriately.
+    """
+    non_existent_id = 999999999  # Use an ID highly unlikely to exist
+
+    logger.info(
+        f"Testing get_table_schema for non-existent table ID: {non_existent_id}"
+    )
+
+    with pytest.raises(HTTPStatusError) as excinfo:
+        await nc_client.tables.get_table_schema(non_existent_id)
+
+    assert excinfo.value.response.status_code == 404
+    logger.info(
+        f"Getting schema for non-existent table ID: {non_existent_id} correctly failed with 404."
+    )
+
+
+async def test_tables_read_nonexistent_table(nc_client: NextcloudClient):
+    """
+    Test that reading from a non-existent table fails appropriately.
+    """
+    non_existent_id = 999999999  # Use an ID highly unlikely to exist
+
+    logger.info(f"Testing get_table_rows for non-existent table ID: {non_existent_id}")
+
+    with pytest.raises(HTTPStatusError) as excinfo:
+        await nc_client.tables.get_table_rows(non_existent_id)
+
+    assert excinfo.value.response.status_code == 404
+    logger.info(
+        f"Reading from non-existent table ID: {non_existent_id} correctly failed with 404."
+    )
+
+
+async def test_tables_create_row_invalid_table(nc_client: NextcloudClient):
+    """
+    Test that creating a row in a non-existent table fails appropriately.
+    """
+    non_existent_id = 999999999  # Use an ID highly unlikely to exist
+    test_data = {1: "test value"}
+
+    logger.info(f"Testing create_row for non-existent table ID: {non_existent_id}")
+
+    with pytest.raises(HTTPStatusError) as excinfo:
+        await nc_client.tables.create_row(non_existent_id, test_data)
+
+    assert excinfo.value.response.status_code == 404
+    logger.info(
+        f"Creating row in non-existent table ID: {non_existent_id} correctly failed with 404."
+    )

tests/integration/test_webdav_cleanup.py

@@ -0,0 +1,263 @@
+import pytest
+import logging
+import time
+import uuid
+from httpx import HTTPStatusError
+
+from nextcloud_mcp_server.client import NextcloudClient
+
+logger = logging.getLogger(__name__)
+
+# Mark all tests in this module as integration tests
+pytestmark = pytest.mark.integration
+
+
+async def test_category_change_cleans_up_old_attachments_directory(
+    nc_client: NextcloudClient,
+):
+    """
+    Tests that when a note's category is changed, the old attachment directory is properly cleaned up.
+    """
+    note_id = None
+    initial_category = "CategoryTest1"
+    new_category = "CategoryTest2"
+    unique_suffix = uuid.uuid4().hex[:8]
+    note_title = f"Category Cleanup Test {unique_suffix}"
+    attachment_filename = f"cleanup_test_{unique_suffix}.txt"
+    attachment_content = f"Content for {attachment_filename}".encode("utf-8")
+
+    try:
+        # 1. Create note with initial category
+        logger.info(f"Creating note '{note_title}' in category '{initial_category}'")
+        created_note = await nc_client.notes.create_note(
+            title=note_title, content="Initial content", category=initial_category
+        )
+        note_id = created_note["id"]
+        etag1 = created_note["etag"]
+        logger.info(f"Note created with ID: {note_id}, Etag: {etag1}")
+        time.sleep(1)
+
+        # 2. Add attachment (passing initial category)
+        logger.info(
+            f"Adding attachment '{attachment_filename}' to note {note_id} (in {initial_category})"
+        )
+        upload_response = await nc_client.webdav.add_note_attachment(
+            note_id=note_id,
+            filename=attachment_filename,
+            content=attachment_content,
+            category=initial_category,
+            mime_type="text/plain",
+        )
+        assert upload_response["status_code"] in [201, 204]
+        logger.info("Attachment added successfully.")
+        time.sleep(1)
+
+        # 3. Verify attachment retrieval from initial category
+        logger.info(
+            f"Verifying attachment retrieval from initial category '{initial_category}'"
+        )
+        retrieved_content1, _ = await nc_client.webdav.get_note_attachment(
+            note_id=note_id, filename=attachment_filename, category=initial_category
+        )
+        assert retrieved_content1 == attachment_content
+        logger.info("Attachment retrieved successfully from initial category.")
+
+        # 4. Construct and check the WebDAV path for the initial category's attachment directory
+        initial_webdav_path = f"Notes/{initial_category}/.attachments.{note_id}"
+        logger.info(f"Initial WebDAV path for attachments: {initial_webdav_path}")
+        # Here we would check if the directory exists, but the WebDAV client doesn't directly
+        # expose directory listing functionality, so we'll infer from attachment retrieval success
+
+        # 5. Update note category
+        logger.info(
+            f"Updating note {note_id} category from '{initial_category}' to '{new_category}'"
+        )
+        current_note_data = await nc_client.notes.get_note(note_id=note_id)
+        current_etag = current_note_data["etag"]
+        updated_note = await nc_client.notes.update(
+            note_id=note_id,
+            etag=current_etag,
+            category=new_category,
+            title=note_title,
+            content="Updated content",
+        )
+        etag3 = updated_note["etag"]
+        assert updated_note["category"] == new_category
+        logger.info(f"Note category updated successfully. New Etag: {etag3}")
+        time.sleep(1)
+
+        # 6. Verify attachment retrieval from new category
+        logger.info(
+            f"Verifying attachment retrieval from new category '{new_category}'"
+        )
+        retrieved_content2, _ = await nc_client.webdav.get_note_attachment(
+            note_id=note_id, filename=attachment_filename, category=new_category
+        )
+        assert retrieved_content2 == attachment_content
+        logger.info("Attachment retrieved successfully from new category.")
+
+        # 7. Try to retrieve from old category - this should fail
+        logger.info(
+            f"Trying to retrieve attachment from old category '{initial_category}' - should fail"
+        )
+        try:
+            await nc_client.webdav.get_note_attachment(
+                note_id=note_id, filename=attachment_filename, category=initial_category
+            )
+            # If we get here, it means the old directory still exists (a problem)
+            logger.error(
+                "ISSUE DETECTED: Was able to retrieve attachment from old category path!"
+            )
+            assert False, (
+                "Old category attachment directory still exists and accessible!"
+            )
+        except HTTPStatusError as e:
+            # This is the expected outcome - old directory should be gone
+            logger.info(
+                f"Correctly got error accessing old category path: {e.response.status_code}"
+            )
+            assert e.response.status_code == 404, (
+                f"Expected 404, got {e.response.status_code}"
+            )
+            logger.info(
+                "Verified old category attachment directory is not accessible (good!)"
+            )
+
+            # 7.1 Directly check old attachment directory existence using WebDAV PROPFIND
+            logger.info(
+                "Directly checking if old attachment directory exists in WebDAV"
+            )
+            webdav_base = nc_client._get_webdav_base_path()
+            old_attachment_dir_path = (
+                f"{webdav_base}/Notes/{initial_category}/.attachments.{note_id}"
+            )
+            propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
+            try:
+                propfind_resp = await nc_client._client.request(
+                    "PROPFIND", old_attachment_dir_path, headers=propfind_headers
+                )
+                status = propfind_resp.status_code
+                if status in [
+                    200,
+                    207,
+                ]:  # Success codes indicate the directory exists (a problem)
+                    logger.error(
+                        f"Old attachment directory still exists! PROPFIND returned {status}"
+                    )
+                    assert False, (
+                        f"Expected old attachment directory to be gone, but it still exists (PROPFIND returned {status})!"
+                    )
+                # If we got another status code (like 404), it's also good - the directory doesn't exist
+                logger.info(
+                    f"Verified old attachment directory does not exist (PROPFIND returned {status})"
+                )
+            except HTTPStatusError as e:
+                # 404 is expected - directory should not exist
+                assert e.response.status_code == 404, (
+                    f"Expected PROPFIND to fail with 404, got {e.response.status_code}"
+                )
+                logger.info(
+                    "Verified old attachment directory does not exist via PROPFIND (404 received)"
+                )
+
+    finally:
+        # 8. Cleanup: Delete the note
+        if note_id:
+            logger.info(f"Cleaning up note ID: {note_id}")
+            try:
+                await nc_client.notes.delete_note(note_id=note_id)
+                logger.info(f"Note {note_id} deleted.")
+                time.sleep(1)
+
+                # 9. Verify both old and new attachment paths are gone
+                logger.info("Verifying all attachment paths are gone")
+                with pytest.raises(HTTPStatusError) as excinfo_new:
+                    await nc_client.webdav.get_note_attachment(
+                        note_id=note_id,
+                        filename=attachment_filename,
+                        category=new_category,
+                    )
+                assert excinfo_new.value.response.status_code == 404
+
+                with pytest.raises(HTTPStatusError) as excinfo_old:
+                    await nc_client.webdav.get_note_attachment(
+                        note_id=note_id,
+                        filename=attachment_filename,
+                        category=initial_category,
+                    )
+                assert excinfo_old.value.response.status_code == 404
+
+                # 9.1 Directly verify directories don't exist using WebDAV PROPFIND
+                logger.info(
+                    "Directly verifying attachment directories don't exist via PROPFIND"
+                )
+                webdav_base = nc_client._get_webdav_base_path()
+
+                # Check new category attachment directory
+                new_attachment_dir_path = (
+                    f"{webdav_base}/Notes/{new_category}/.attachments.{note_id}"
+                )
+                propfind_headers = {"Depth": "0", "OCS-APIRequest": "true"}
+                try:
+                    propfind_resp = await nc_client._client.request(
+                        "PROPFIND", new_attachment_dir_path, headers=propfind_headers
+                    )
+                    status = propfind_resp.status_code
+                    if status in [
+                        200,
+                        207,
+                    ]:  # Success codes indicate the directory exists (a problem)
+                        logger.error(
+                            f"New category attachment directory still exists! PROPFIND returned {status}"
+                        )
+                        assert False, (
+                            f"Expected new category attachment directory to be gone, but it still exists (PROPFIND returned {status})!"
+                        )
+                    # If we got another status code (like 404), it's also good - the directory doesn't exist
+                    logger.info(
+                        f"Verified new category attachment directory does not exist (PROPFIND returned {status})"
+                    )
+                except HTTPStatusError as e:
+                    assert e.response.status_code == 404, (
+                        f"Expected PROPFIND to fail with 404, got {e.response.status_code}"
+                    )
+                    logger.info(
+                        "Verified new category attachment directory is gone via PROPFIND"
+                    )
+
+                # Check old category attachment directory
+                old_attachment_dir_path = (
+                    f"{webdav_base}/Notes/{initial_category}/.attachments.{note_id}"
+                )
+                try:
+                    propfind_resp = await nc_client._client.request(
+                        "PROPFIND", old_attachment_dir_path, headers=propfind_headers
+                    )
+                    status = propfind_resp.status_code
+                    if status in [
+                        200,
+                        207,
+                    ]:  # Success codes indicate the directory exists (a problem)
+                        logger.error(
+                            f"Old category attachment directory still exists! PROPFIND returned {status}"
+                        )
+                        assert False, (
+                            f"Expected old category attachment directory to be gone, but it still exists (PROPFIND returned {status})!"
+                        )
+                    # If we got another status code (like 404), it's also good - the directory doesn't exist
+                    logger.info(
+                        f"Verified old category attachment directory does not exist (PROPFIND returned {status})"
+                    )
+                except HTTPStatusError as e:
+                    assert e.response.status_code == 404, (
+                        f"Expected PROPFIND to fail with 404, got {e.response.status_code}"
+                    )
+                    logger.info(
+                        "Verified old category attachment directory is gone via PROPFIND"
+                    )
+
+                logger.info(
+                    "Verified all attachment directories are properly cleaned up."
+                )
+            except Exception as e:
+                logger.error(f"Error during cleanup for note {note_id}: {e}")

tests/integration/test_webdav_operations.py

@@ -0,0 +1,272 @@
+"""Integration tests for WebDAV operations."""
+
+import pytest
+import logging
+import uuid
+from httpx import HTTPStatusError
+
+from nextcloud_mcp_server.client import NextcloudClient
+
+logger = logging.getLogger(__name__)
+
+# Mark all tests in this module as integration tests
+pytestmark = pytest.mark.integration
+
+
+@pytest.fixture
+async def test_base_path(nc_client: NextcloudClient):
+    """Base path for test files/directories."""
+    test_dir = f"mcp_test_{uuid.uuid4().hex[:8]}"
+    await nc_client.webdav.create_directory(test_dir)
+    yield test_dir
+    await nc_client.webdav.delete_resource(test_dir)
+
+
+async def test_create_and_delete_directory(
+    nc_client: NextcloudClient, test_base_path: str
+):
+    """Test creating and deleting directories."""
+    test_dir = f"{test_base_path}/test_directory"
+
+    try:
+        # Create directory
+        result = await nc_client.webdav.create_directory(test_dir)
+        assert result["status_code"] == 201  # Created
+        logger.info(f"Created directory: {test_dir}")
+
+        # Verify directory exists by listing parent
+        parent_listing = await nc_client.webdav.list_directory(test_base_path)
+        dir_names = [item["name"] for item in parent_listing]
+        assert "test_directory" in dir_names
+
+        # Delete directory
+        delete_result = await nc_client.webdav.delete_resource(test_dir)
+        assert delete_result["status_code"] in [204, 404]  # No Content or Not Found
+        logger.info(f"Deleted directory: {test_dir}")
+
+    finally:
+        # Cleanup: ensure directory is deleted
+        try:
+            await nc_client.webdav.delete_resource(test_dir)
+        except Exception:
+            pass
+
+
+async def test_write_read_delete_file(nc_client: NextcloudClient, test_base_path: str):
+    """Test writing, reading, and deleting files."""
+    test_file = f"{test_base_path}/test_file.txt"
+    test_content = f"Test content {uuid.uuid4().hex}"
+
+    try:
+        # Create base directory first
+        await nc_client.webdav.create_directory(test_base_path)
+
+        # Write file
+        write_result = await nc_client.webdav.write_file(
+            test_file, test_content.encode("utf-8"), content_type="text/plain"
+        )
+        assert write_result["status_code"] in [200, 201, 204]  # Success codes
+        logger.info(f"Wrote file: {test_file}")
+
+        # Read file back
+        content, content_type = await nc_client.webdav.read_file(test_file)
+        assert content.decode("utf-8") == test_content
+        assert "text/plain" in content_type
+        logger.info(f"Read file: {test_file}")
+
+        # Verify file appears in directory listing
+        listing = await nc_client.webdav.list_directory(test_base_path)
+        file_names = [item["name"] for item in listing]
+        assert "test_file.txt" in file_names
+
+        # Delete file
+        delete_result = await nc_client.webdav.delete_resource(test_file)
+        assert delete_result["status_code"] in [204, 404]  # No Content or Not Found
+        logger.info(f"Deleted file: {test_file}")
+
+    finally:
+        # Cleanup
+        try:
+            await nc_client.webdav.delete_resource(test_file)
+            await nc_client.webdav.delete_resource(test_base_path)
+        except Exception:
+            pass
+
+
+async def test_list_directory_empty_and_populated(
+    nc_client: NextcloudClient, test_base_path: str
+):
+    """Test listing empty and populated directories."""
+    try:
+        # Create base directory
+        await nc_client.webdav.create_directory(test_base_path)
+
+        # List empty directory
+        empty_listing = await nc_client.webdav.list_directory(test_base_path)
+        assert isinstance(empty_listing, list)
+        assert len(empty_listing) == 0
+        logger.info(f"Empty directory listing: {len(empty_listing)} items")
+
+        # Add some files and directories
+        await nc_client.webdav.create_directory(f"{test_base_path}/subdir1")
+        await nc_client.webdav.create_directory(f"{test_base_path}/subdir2")
+        await nc_client.webdav.write_file(
+            f"{test_base_path}/file1.txt", b"content1", content_type="text/plain"
+        )
+        await nc_client.webdav.write_file(
+            f"{test_base_path}/file2.md",
+            b"# Markdown content",
+            content_type="text/markdown",
+        )
+
+        # List populated directory
+        populated_listing = await nc_client.webdav.list_directory(test_base_path)
+        assert len(populated_listing) == 4  # 2 dirs + 2 files
+
+        # Check that we have both files and directories
+        names = [item["name"] for item in populated_listing]
+        assert "subdir1" in names
+        assert "subdir2" in names
+        assert "file1.txt" in names
+        assert "file2.md" in names
+
+        # Check metadata is present
+        for item in populated_listing:
+            assert "name" in item
+            assert "path" in item
+            assert "is_directory" in item
+            assert "size" in item
+            assert "content_type" in item
+            assert "last_modified" in item
+
+        logger.info(f"Populated directory listing: {len(populated_listing)} items")
+
+    finally:
+        # Cleanup
+        try:
+            await nc_client.webdav.delete_resource(f"{test_base_path}/file1.txt")
+            await nc_client.webdav.delete_resource(f"{test_base_path}/file2.md")
+            await nc_client.webdav.delete_resource(f"{test_base_path}/subdir1")
+            await nc_client.webdav.delete_resource(f"{test_base_path}/subdir2")
+            await nc_client.webdav.delete_resource(test_base_path)
+        except Exception:
+            pass
+
+
+async def test_read_nonexistent_file(nc_client: NextcloudClient):
+    """Test reading a file that doesn't exist."""
+    nonexistent_file = f"nonexistent_{uuid.uuid4().hex}.txt"
+
+    with pytest.raises(HTTPStatusError) as exc_info:
+        await nc_client.webdav.read_file(nonexistent_file)
+
+    assert exc_info.value.response.status_code == 404
+    logger.info(f"Correctly got 404 for nonexistent file: {nonexistent_file}")
+
+
+async def test_delete_nonexistent_resource(nc_client: NextcloudClient):
+    """Test deleting a resource that doesn't exist."""
+    nonexistent_resource = f"nonexistent_{uuid.uuid4().hex}"
+
+    result = await nc_client.webdav.delete_resource(nonexistent_resource)
+    assert result["status_code"] == 404
+    logger.info(f"Correctly got 404 for nonexistent resource: {nonexistent_resource}")
+
+
+async def test_create_nested_directories(
+    nc_client: NextcloudClient, test_base_path: str
+):
+    """Test creating nested directory structures."""
+    nested_path = f"{test_base_path}/level1/level2/level3"
+
+    try:
+        # Create nested directories (should create parent directories automatically)
+        result = await nc_client.webdav.create_directory(nested_path, True)
+        assert result["status_code"] == 201
+
+        # Verify the structure was created
+        level1_listing = await nc_client.webdav.list_directory(
+            f"{test_base_path}/level1"
+        )
+        assert len(level1_listing) == 1
+        assert level1_listing[0]["name"] == "level2"
+        assert level1_listing[0]["is_directory"] is True
+
+        level2_listing = await nc_client.webdav.list_directory(
+            f"{test_base_path}/level1/level2"
+        )
+        assert len(level2_listing) == 1
+        assert level2_listing[0]["name"] == "level3"
+        assert level2_listing[0]["is_directory"] is True
+
+        logger.info(f"Created nested directory structure: {nested_path}")
+
+    finally:
+        # Cleanup - delete from deepest to shallowest
+        try:
+            await nc_client.webdav.delete_resource(nested_path)
+            await nc_client.webdav.delete_resource(f"{test_base_path}/level1/level2")
+            await nc_client.webdav.delete_resource(f"{test_base_path}/level1")
+        except Exception:
+            pass
+
+
+async def test_overwrite_existing_file(nc_client: NextcloudClient, test_base_path: str):
+    """Test overwriting an existing file."""
+    test_file = f"{test_base_path}/overwrite_test.txt"
+    original_content = "Original content"
+    new_content = "New content after overwrite"
+
+    try:
+        # Create base directory
+        await nc_client.webdav.create_directory(test_base_path)
+
+        # Write original file
+        await nc_client.webdav.write_file(
+            test_file, original_content.encode("utf-8"), content_type="text/plain"
+        )
+
+        # Verify original content
+        content, _ = await nc_client.webdav.read_file(test_file)
+        assert content.decode("utf-8") == original_content
+
+        # Overwrite with new content
+        overwrite_result = await nc_client.webdav.write_file(
+            test_file, new_content.encode("utf-8"), content_type="text/plain"
+        )
+        assert overwrite_result["status_code"] in [200, 204]  # OK or No Content
+
+        # Verify new content
+        content, _ = await nc_client.webdav.read_file(test_file)
+        assert content.decode("utf-8") == new_content
+
+        logger.info(f"Successfully overwrote file: {test_file}")
+
+    finally:
+        # Cleanup
+        try:
+            await nc_client.webdav.delete_resource(test_file)
+            await nc_client.webdav.delete_resource(test_base_path)
+        except Exception:
+            pass
+
+
+async def test_list_root_directory(nc_client: NextcloudClient):
+    """Test listing the root directory."""
+    root_listing = await nc_client.webdav.list_directory("")
+
+    # Root directory should exist and be listable
+    assert isinstance(root_listing, list)
+    # Should have at least some default folders/files
+    assert len(root_listing) >= 0
+
+    # Check structure of items
+    for item in root_listing:
+        assert "name" in item
+        assert "path" in item
+        assert "is_directory" in item
+        assert "size" in item
+        assert "content_type" in item
+        assert "last_modified" in item
+
+    logger.info(f"Root directory contains {len(root_listing)} items")

tests/test_client.py

@@ -1,158 +0,0 @@
-import pytest
-import os
-import time
-import uuid
-from httpx import HTTPStatusError
-
-from nextcloud_mcp_server.client import NextcloudClient
-
-# Tests assume NEXTCLOUD_HOST, NEXTCLOUD_USERNAME, NEXTCLOUD_PASSWORD env vars are set
-
-
-@pytest.fixture(scope="module")
-def nc_client() -> NextcloudClient:
-    """
-    Fixture to create a NextcloudClient instance for integration tests.
-    Reads credentials from environment variables.
-    Scope is 'module' so the client is reused for all tests in this file.
-    """
-    # Basic check to ensure env vars seem present - tests will fail properly if not
-    assert os.getenv("NEXTCLOUD_HOST"), "NEXTCLOUD_HOST env var not set"
-    assert os.getenv("NEXTCLOUD_USERNAME"), "NEXTCLOUD_USERNAME env var not set"
-    assert os.getenv("NEXTCLOUD_PASSWORD"), "NEXTCLOUD_PASSWORD env var not set"
-    return NextcloudClient.from_env()
-
-
-@pytest.mark.integration
-def test_note_crud_integration(nc_client: NextcloudClient):
-    """
-    Integration test for the complete CRUD (Create, Read, Update, Delete)
-    lifecycle of a note.
-    """
-    # --- Create ---
-    unique_id = str(uuid.uuid4())  # To ensure note is unique for this test run
-    create_title = f"Integration Test Note {unique_id}"
-    create_content = f"Content for integration test {unique_id}"
-    create_category = "IntegrationTesting"
-
-    created_note = (
-        None  # Initialize to ensure cleanup happens even if create fails mid-assert
-    )
-    try:
-        print(f"\nAttempting to create note: {create_title}")
-        created_note = nc_client.notes_create_note(
-            title=create_title, content=create_content, category=create_category
-        )
-        print(f"Note created: {created_note}")
-
-        assert created_note is not None
-        assert "id" in created_note
-        assert created_note["title"] == create_title
-        assert created_note["content"] == create_content
-        assert created_note["category"] == create_category
-        assert "etag" in created_note
-        note_id = created_note["id"]
-        etag = created_note["etag"]
-
-        # Add a small delay to allow Nextcloud to process if needed
-        time.sleep(1)
-
-        # --- Read (Verify Create) ---
-        print(f"Attempting to read note ID: {note_id}")
-        read_note = nc_client.notes_get_note(note_id=note_id)
-        print(f"Note read: {read_note}")
-        assert read_note["id"] == note_id
-        assert read_note["title"] == create_title
-        assert read_note["content"] == create_content
-        assert read_note["category"] == create_category
-        # Etag might change even on read in some systems, so don't assert etag equality here
-
-        # --- Update ---
-        update_title = f"Updated Test Note {unique_id}"
-        update_content = f"Updated content {unique_id}"
-        # Use the etag from the *creation* for the update's If-Match header
-        print(f"Attempting to update note ID: {note_id} with etag: {etag}")
-        updated_note = nc_client.notes_update_note(
-            note_id=note_id,
-            etag=etag,
-            title=update_title,
-            content=update_content,
-            # category=create_category # Keep category same or update if needed
-        )
-        print(f"Note updated: {updated_note}")
-        assert updated_note["id"] == note_id
-        assert updated_note["title"] == update_title
-        assert updated_note["content"] == update_content
-        assert updated_note["category"] == create_category  # Category wasn't updated
-        assert "etag" in updated_note
-        assert updated_note["etag"] != etag  # Etag must change on update
-        new_etag = updated_note["etag"]
-
-        # Add a small delay
-        time.sleep(1)
-
-        # --- Read (Verify Update) ---
-        print(f"Attempting to read updated note ID: {note_id}")
-        read_updated_note = nc_client.notes_get_note(note_id=note_id)
-        print(f"Updated note read: {read_updated_note}")
-        assert read_updated_note["id"] == note_id
-        assert read_updated_note["title"] == update_title
-        assert read_updated_note["content"] == update_content
-        # Don't assert etag equality here either
-
-        # --- Test Update Conflict (Precondition Failed) ---
-        print(f"Attempting to update note ID: {note_id} with OLD etag: {etag}")
-        with pytest.raises(HTTPStatusError) as excinfo:
-            nc_client.notes_update_note(
-                note_id=note_id,
-                etag=etag,  # Use the OLD etag
-                title="This update should fail",
-            )
-        assert excinfo.value.response.status_code == 412  # Precondition Failed
-        print("Update with old etag correctly failed with 412.")
-
-    finally:
-        # --- Delete ---
-        if created_note and "id" in created_note:
-            note_id_to_delete = created_note["id"]
-            print(f"Attempting to delete note ID: {note_id_to_delete}")
-            try:
-                delete_response = nc_client.notes_delete_note(note_id=note_id_to_delete)
-                print(f"Delete response: {delete_response}")
-                # Check if delete returns the deleted object or just status
-                # Assuming it returns the object based on previous tests
-                assert delete_response["id"] == note_id_to_delete
-                print(f"Note ID: {note_id_to_delete} deleted successfully.")
-
-                # --- Verify Delete ---
-                print(f"Attempting to read deleted note ID: {note_id_to_delete}")
-                with pytest.raises(HTTPStatusError) as excinfo_del:
-                    nc_client.notes_get_note(note_id=note_id_to_delete)
-                assert excinfo_del.value.response.status_code == 404
-                print(
-                    f"Reading deleted note ID: {note_id_to_delete} correctly failed with 404."
-                )
-
-            except HTTPStatusError as e:
-                # If deletion fails unexpectedly, log it but don't fail the test here
-                # as the primary goal was CRUD, and cleanup failure is secondary.
-                print(f"Error during cleanup (deleting note {note_id_to_delete}): {e}")
-            except Exception as e:
-                print(f"Unexpected error during cleanup: {e}")
-        else:
-            print(
-                "Skipping delete step as note creation might have failed or ID was not available."
-            )
-
-
-@pytest.mark.integration
-def test_delete_nonexistent_note(nc_client: NextcloudClient):
-    """Test deleting a note that doesn't exist."""
-    non_existent_id = 999999999  # Use an ID highly unlikely to exist
-    print(f"\nAttempting to delete non-existent note ID: {non_existent_id}")
-    with pytest.raises(HTTPStatusError) as excinfo:
-        nc_client.notes_delete_note(note_id=non_existent_id)
-    assert excinfo.value.response.status_code == 404
-    print(
-        f"Deleting non-existent note ID: {non_existent_id} correctly failed with 404."
-    )