fix(client): Strip cookies from responses to avoid falsely raising CSRF errors

chore(deps): update ghcr.io/astral-sh/uv docker tag to v0.8.6

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

@@ -16,7 +16,7 @@ jobs:
 
       - name: Docker meta
         id: meta
-        uses: docker/metadata-action@902fa8ec7d6ecbf8d84d538b9b233a880e428804 # v5
+        uses: docker/metadata-action@c1e51972afc2121e065aed6d45c65596fe445f3f # v5
         with:
           # list of Docker images to use as base name for tags
           images: |
@@ -37,7 +37,7 @@ jobs:
 
       - name: Log in to GitHub Container Registry
         if: github.event_name != 'pull_request'
-        uses: docker/login-action@74a5d142397b4f367a81961eba4e8cd7edddf772 # v3
+        uses: docker/login-action@184bdaa0721073962dff0199f1fb9940f07167d1 # v3
         with:
           registry: ghcr.io
           username: ${{ github.actor }}

CHANGELOG.md

@@ -1,3 +1,16 @@
+## v0.7.0 (2025-08-03)
+
+### Feat
+
+- **contacts**: Initialize Contacts App
+
+## v0.6.1 (2025-08-01)
+
+### Fix
+
+- **calendar**: Fix iCalendar date vs datetime format
+- **calendar**: Remove try/except in calendar API
+
 ## v0.6.0 (2025-07-29)
 
 ### Feat

Dockerfile

@@ -1,4 +1,4 @@
-FROM ghcr.io/astral-sh/uv:0.8.4-python3.11-alpine@sha256:f2c5b953b713f455bcac4429303bb21d7d2547d56a64e1a7b2517cc9f0563f0f
+FROM ghcr.io/astral-sh/uv:0.8.6-python3.11-alpine@sha256:15e7f7ee738f1b8a0e1ed95da7cf821c58b77d3d15275bf8f4605fbbf36679f4
 
 WORKDIR /app
 

README.md

@@ -16,6 +16,7 @@ The server provides integration with multiple Nextcloud apps, enabling LLMs to i
 | **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. |
+| **Contacts** | ✅ Full Support | Create, read, update, and delete contacts and address books via CardDAV. |
 
 ## Available Tools
 
@@ -46,6 +47,17 @@ The server provides integration with multiple Nextcloud apps, enabling LLMs to i
 | `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 |
 
+### Contacts Tools
+
+| Tool | Description |
+|------|-------------|
+| `nc_contacts_list_addressbooks` | List all available addressbooks for the user |
+| `nc_contacts_list_contacts` | List all contacts in a specific addressbook |
+| `nc_contacts_create_addressbook` | Create a new addressbook |
+| `nc_contacts_delete_addressbook` | Delete an addressbook |
+| `nc_contacts_create_contact` | Create a new contact in an addressbook |
+| `nc_contacts_delete_contact` | Delete a contact from an addressbook |
+
 ### Tables Tools
 
 | Tool | Description |

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

@@ -11,6 +11,10 @@ php /var/www/html/occ app:enable calendar
 echo "Waiting for calendar app to initialize..."
 sleep 5
 
+# Increase limits on calendar creation for integration tests (100 in 60s)
+php occ config:app:set dav rateLimitCalendarCreation --type=integer --value=100
+php occ config:app:set dav rateLimitPeriodCalendarCreation --type=integer --value=60
+
 # Ensure maintenance mode is off before calendar operations
 php /var/www/html/occ maintenance:mode --off
 

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

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

docker-compose.yml

@@ -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:25c0ae32c6c2301798579f5944af53729766a18eff5660bbef196fc2e6214a9c
+    image: redis:alpine@sha256:7521abdff715d396aa482183942f3fe643344287c29ccb66eee16ac08a92190f
     restart: always
 
   app:
-    image: nextcloud:31.0.7@sha256:81dc361f8f216d8acff20bd3dea2226fb6cea883c277505cbb2ddd6327c867fa
+    image: nextcloud:31.0.7@sha256:a834f43b159890322d471862a3d8ab48602d4a619499aa7bc94b662cf4463e66
     #user: www-data:www-data
     restart: always
     #post_start:

nextcloud_mcp_server/app.py

@@ -9,6 +9,7 @@ from nextcloud_mcp_server.client import NextcloudClient
 from nextcloud_mcp_server.config import setup_logging
 from nextcloud_mcp_server.server import (
     configure_calendar_tools,
+    configure_contacts_tools,
     configure_notes_tools,
     configure_tables_tools,
     configure_webdav_tools,
@@ -56,6 +57,7 @@ configure_notes_tools(mcp)
 configure_tables_tools(mcp)
 configure_webdav_tools(mcp)
 configure_calendar_tools(mcp)
+configure_contacts_tools(mcp)
 
 
 def run():

nextcloud_mcp_server/client/__init__.py

@@ -1,10 +1,19 @@
 import logging
 import os
 
-from httpx import AsyncClient, Auth, BasicAuth, Request, Response
+from httpx import (
+    AsyncClient,
+    Auth,
+    BasicAuth,
+    Request,
+    Response,
+    AsyncBaseTransport,
+    AsyncHTTPTransport,
+)
 
 from ..controllers.notes_search import NotesSearchController
 from .calendar import CalendarClient
+from .contacts import ContactsClient
 from .notes import NotesClient
 from .tables import TablesClient
 from .webdav import WebDAVClient
@@ -12,19 +21,34 @@ from .webdav import WebDAVClient
 logger = logging.getLogger(__name__)
 
 
-def log_request(request: Request):
-    logger.info(
+async def log_request(request: Request):
+    logger.debug(
         "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)
+    logger.debug("Request body: %s", request.content)
+    logger.debug("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)
+async def log_response(response: Response):
+    await response.aread()
+    logger.debug("Response [%s] %s", response.status_code, response.text)
+
+
+class AsyncDisableCookieTransport(AsyncBaseTransport):
+    """This Transport disable cookies from accumulating in the httpx AsyncClient
+
+    Thanks to: https://github.com/encode/httpx/issues/2992#issuecomment-2133258994
+    """
+
+    def __init__(self, transport: AsyncBaseTransport):
+        self.transport = transport
+
+    async def handle_async_request(self, request: Request) -> Response:
+        response = await self.transport.handle_async_request(request)
+        response.headers.pop("set-cookie", None)
+        return response
 
 
 class NextcloudClient:
@@ -35,7 +59,8 @@ class NextcloudClient:
         self._client = AsyncClient(
             base_url=base_url,
             auth=auth,
-            # event_hooks={"request": [log_request], "response": [log_response]},
+            transport=AsyncDisableCookieTransport(AsyncHTTPTransport()),
+            event_hooks={"request": [log_request], "response": [log_response]},
         )
 
         # Initialize app clients
@@ -43,6 +68,7 @@ class NextcloudClient:
         self.webdav = WebDAVClient(self._client, username)
         self.tables = TablesClient(self._client, username)
         self.calendar = CalendarClient(self._client, username)
+        self.contacts = ContactsClient(self._client, username)
 
         # Initialize controllers
         self._notes_search = NotesSearchController()

nextcloud_mcp_server/client/base.py

@@ -3,11 +3,65 @@
 import logging
 from abc import ABC
 
-from httpx import AsyncClient
+from functools import wraps
+import time
+from httpx import HTTPStatusError, codes, RequestError, AsyncClient
 
 logger = logging.getLogger(__name__)
 
 
+def retry_on_429(func):
+    """This decorator handles the 429 response from REST APIs
+
+    The `func` is assumed to be a method that is similar to `httpx.Client.get`,
+    and returns an `httpx.Response` object. In the case of `Too Many Requests` HTTP
+    response, the function will wait for a couple of seconds and retry the request.
+    """
+
+    MAX_RETRIES = 5
+
+    @wraps(func)
+    async def wrapper(*args, **kwargs):
+        retries = 0
+
+        while retries < MAX_RETRIES:
+            try:
+                # Make GET API call
+                retries += 1
+                response = await func(*args, **kwargs)
+                break
+
+            except HTTPStatusError as e:
+                # If we get a '429 Client Error: Too Many Requests'
+                # error we wait a couple of seconds and do a retry
+                if e.response.status_code == codes.TOO_MANY_REQUESTS:
+                    logger.warning(
+                        f"429 Client Error: Too Many Requests, Number of attempts: {retries}"
+                    )
+                    time.sleep(5)
+                else:
+                    logger.warning(
+                        f"HTTPStatusError {e.response.status_code}: {e}, Number of attempts: {retries}"
+                    )
+                    raise
+            except RequestError as e:
+                logger.warning(
+                    f"RequestError {e.request.url}: {e}, Number of attempts: {retries}"
+                )
+                raise
+
+        # If for loop ends without break statement
+        else:
+            logger.warning("All API call retries failed")
+            raise RuntimeError(
+                f"Maximum number of retries ({MAX_RETRIES}) exceeded without success"
+            )
+
+        return response
+
+    return wrapper
+
+
 class BaseNextcloudClient(ABC):
     """Base class for all Nextcloud app clients."""
 
@@ -25,6 +79,7 @@ class BaseNextcloudClient(ABC):
         """Helper to get the base WebDAV path for the authenticated user."""
         return f"/remote.php/dav/files/{self.username}"
 
+    @retry_on_429
     async def _make_request(self, method: str, url: str, **kwargs):
         """Common request wrapper with logging and error handling.
 

nextcloud_mcp_server/client/contacts.py

@@ -0,0 +1,235 @@
+"""CardDAV client for NextCloud contacts operations."""
+
+import logging
+from .base import BaseNextcloudClient
+import xml.etree.ElementTree as ET
+from pythonvCard4.vcard import Contact
+
+logger = logging.getLogger(__name__)
+
+
+class ContactsClient(BaseNextcloudClient):
+    """Client for NextCloud CardDAV contact operations."""
+
+    def _get_carddav_base_path(self) -> str:
+        """Helper to get the base CardDAV path for contacts."""
+        return f"/remote.php/dav/addressbooks/users/{self.username}"
+
+    async def list_addressbooks(self):
+        """List all available addressbooks for the user."""
+
+        carddav_path = self._get_carddav_base_path()
+
+        propfind_body = """<?xml version="1.0" encoding="utf-8"?>
+        <d:propfind xmlns:d="DAV:" xmlns:cs="http://calendarserver.org/ns/">
+            <d:prop>
+                <d:displayname/>
+                <d:getctag />
+            </d:prop>
+        </d:propfind>"""
+
+        headers = {
+            # "Depth": "0",
+            "Content-Type": "application/xml",
+            "Accept": "application/xml",
+        }
+
+        response = await self._make_request(
+            "PROPFIND", carddav_path, content=propfind_body, headers=headers
+        )
+
+        ns = {"d": "DAV:"}
+
+        # logger.info(response.content)
+        root = ET.fromstring(response.content)
+        addressbooks = []
+        for response_elem in root.findall(".//d:response", ns):
+            href = response_elem.find(".//d:href", ns)
+            if href is None:
+                continue
+
+            href_text = href.text or ""
+            if not href_text.endswith("/"):
+                continue  # Skip non-addressbook resources
+
+            # Extract addressbook name from href
+            addressbook_name = href_text.rstrip("/").split("/")[-1]
+            if not addressbook_name or addressbook_name == self.username:
+                continue
+
+            # Get properties
+            propstat = response_elem.find(".//d:propstat", ns)
+            if propstat is None:
+                continue
+
+            prop = propstat.find(".//d:prop", ns)
+            if prop is None:
+                continue
+
+            displayname_elem = prop.find(".//d:displayname", ns)
+            displayname = (
+                displayname_elem.text
+                if displayname_elem is not None
+                else addressbook_name
+            )
+
+            getctag_elem = prop.find(".//d:getctag", ns)
+            getctag = getctag_elem.text if getctag_elem is not None else None
+
+            addressbooks.append(
+                {
+                    "name": addressbook_name,
+                    "display_name": displayname,
+                    "getctag": getctag,
+                }
+            )
+
+        logger.debug(f"Found {len(addressbooks)} addressbooks")
+        return addressbooks
+
+    async def create_addressbook(self, *, name: str, display_name: str):
+        """Create a new addressbook."""
+        carddav_path = self._get_carddav_base_path()
+        url = f"{carddav_path}/{name}/"
+
+        prop_body = f"""<?xml version="1.0" encoding="utf-8"?>
+        <d:mkcol xmlns:d="DAV:" xmlns:c="urn:ietf:params:xml:ns:carddav">
+            <d:set>
+                <d:prop>
+                    <d:resourcetype>
+                        <d:collection/>
+                        <c:addressbook/>
+                    </d:resourcetype>
+                    <d:displayname>{display_name}</d:displayname>
+                </d:prop>
+            </d:set>
+        </d:mkcol>"""
+
+        headers = {
+            "Content-Type": "application/xml",
+        }
+
+        await self._make_request("MKCOL", url, content=prop_body, headers=headers)
+
+    async def delete_addressbook(self, *, name: str):
+        """Delete an addressbook."""
+        carddav_path = self._get_carddav_base_path()
+        url = f"{carddav_path}/{name}/"
+        await self._make_request("DELETE", url)
+
+    async def create_contact(self, *, addressbook: str, uid: str, contact_data: dict):
+        """Create a new contact."""
+        carddav_path = self._get_carddav_base_path()
+        url = f"{carddav_path}/{addressbook}/{uid}.vcf"
+
+        contact = Contact(fn=contact_data.get("fn"), uid=uid)
+        if "email" in contact_data:
+            contact.email = [{"value": contact_data["email"], "type": ["HOME"]}]
+        if "tel" in contact_data:
+            contact.tel = [{"value": contact_data["tel"], "type": ["HOME"]}]
+
+        vcard = contact.to_vcard()
+
+        headers = {
+            "Content-Type": "text/vcard; charset=utf-8",
+            "If-None-Match": "*",
+        }
+
+        await self._make_request("PUT", url, content=vcard, headers=headers)
+
+    async def delete_contact(self, *, addressbook: str, uid: str):
+        """Delete a contact."""
+        carddav_path = self._get_carddav_base_path()
+        url = f"{carddav_path}/{addressbook}/{uid}.vcf"
+        await self._make_request("DELETE", url)
+
+    async def list_contacts(self, *, addressbook: str):
+        """List all available contacts for addressbook."""
+
+        carddav_path = self._get_carddav_base_path()
+
+        report_body = """<?xml version="1.0" encoding="utf-8"?>
+        <card:addressbook-query xmlns:d="DAV:" xmlns:card="urn:ietf:params:xml:ns:carddav">
+            <d:prop>
+                <d:getetag />
+                <card:address-data />
+            </d:prop>
+        </card:addressbook-query>"""
+
+        headers = {
+            "Depth": "1",
+            "Content-Type": "application/xml",
+            "Accept": "application/xml",
+        }
+
+        response = await self._make_request(
+            "REPORT",
+            f"{carddav_path}/{addressbook}",
+            content=report_body,
+            headers=headers,
+        )
+
+        ns = {"d": "DAV:", "card": "urn:ietf:params:xml:ns:carddav"}
+
+        # logger.info(response.text)
+        root = ET.fromstring(response.content)
+        contacts = []
+        for response_elem in root.findall(".//d:response", ns):
+            href = response_elem.find(".//d:href", ns)
+            if href is None:
+                logger.info("Skip missing href")
+                continue
+
+            href_text = href.text or ""
+            # logger.info("Href text: %s", href_text)
+            # if not href_text.endswith("/"):
+            # logger.info("# Skip non-addressbook resources")
+            # continue
+
+            # Extract vcard id from href
+            vcard_id = href_text.rstrip("/").split("/")[-1]
+            if not vcard_id:
+                logger.info("Skip missing vcard_id")
+                continue
+            vcard_id = vcard_id.replace(".vcf", "")
+
+            # Get properties
+            propstat = response_elem.find(".//d:propstat", ns)
+            if propstat is None:
+                logger.info("Skip missing propstat")
+                continue
+
+            prop = propstat.find(".//d:prop", ns)
+            if prop is None:
+                logger.info("Skip missing prop")
+                continue
+
+            getetag_elem = prop.find(".//d:getetag", ns)
+            getetag = getetag_elem.text if getetag_elem is not None else None
+
+            addressdata_elem = prop.find(".//card:address-data", ns)
+            addressdata = (
+                addressdata_elem.text if addressdata_elem is not None else None
+            )
+            if addressdata is None:
+                logger.info("Skip missing addressdata")
+                continue
+
+            contact = Contact.from_vcard(addressdata)
+
+            contacts.append(
+                {
+                    "vcard_id": vcard_id,
+                    "getetag": getetag,
+                    "contact": {
+                        "fullname": contact.fn,
+                        "nickname": contact.nickname,
+                        "birthday": contact.bday,
+                        "email": contact.email,
+                    },
+                    "addressdata": addressdata,
+                }
+            )
+
+        logger.debug(f"Found {len(contacts)} contacts")
+        return contacts

nextcloud_mcp_server/server/__init__.py

@@ -2,10 +2,12 @@ from .calendar import configure_calendar_tools
 from .notes import configure_notes_tools
 from .tables import configure_tables_tools
 from .webdav import configure_webdav_tools
+from .contacts import configure_contacts_tools
 
 __all__ = [
     "configure_calendar_tools",
     "configure_notes_tools",
     "configure_tables_tools",
     "configure_webdav_tools",
+    "configure_contacts_tools",
 ]

nextcloud_mcp_server/server/contacts.py

@@ -0,0 +1,65 @@
+import logging
+
+from mcp.server.fastmcp import Context, FastMCP
+
+from nextcloud_mcp_server.client import NextcloudClient
+
+logger = logging.getLogger(__name__)
+
+
+def configure_contacts_tools(mcp: FastMCP):
+    # Contacts tools
+    @mcp.tool()
+    async def nc_contacts_list_addressbooks(ctx: Context):
+        """List all addressbooks for the user."""
+        client: NextcloudClient = ctx.request_context.lifespan_context.client
+        return await client.contacts.list_addressbooks()
+
+    @mcp.tool()
+    async def nc_contacts_list_contacts(ctx: Context, *, addressbook: str):
+        """List all addressbooks for the user."""
+        client: NextcloudClient = ctx.request_context.lifespan_context.client
+        return await client.contacts.list_contacts(addressbook=addressbook)
+
+    @mcp.tool()
+    async def nc_contacts_create_addressbook(
+        ctx: Context, *, name: str, display_name: str
+    ):
+        """Create a new addressbook.
+
+        Args:
+            name: The name of the addressbook.
+            display_name: The display name of the addressbook.
+        """
+        client: NextcloudClient = ctx.request_context.lifespan_context.client
+        return await client.contacts.create_addressbook(
+            name=name, display_name=display_name
+        )
+
+    @mcp.tool()
+    async def nc_contacts_delete_addressbook(ctx: Context, *, name: str):
+        """Delete an addressbook."""
+        client: NextcloudClient = ctx.request_context.lifespan_context.client
+        return await client.contacts.delete_addressbook(name=name)
+
+    @mcp.tool()
+    async def nc_contacts_create_contact(
+        ctx: Context, *, addressbook: str, uid: str, contact_data: dict
+    ):
+        """Create a new contact.
+
+        Args:
+            addressbook: The name of the addressbook to create the contact in.
+            uid: The unique ID for the contact.
+            contact_data: A dictionary with the contact's details, e.g. {"fn": "John Doe", "email": "john.doe@example.com"}.
+        """
+        client: NextcloudClient = ctx.request_context.lifespan_context.client
+        return await client.contacts.create_contact(
+            addressbook=addressbook, uid=uid, contact_data=contact_data
+        )
+
+    @mcp.tool()
+    async def nc_contacts_delete_contact(ctx: Context, *, addressbook: str, uid: str):
+        """Delete a contact."""
+        client: NextcloudClient = ctx.request_context.lifespan_context.client
+        return await client.contacts.delete_contact(addressbook=addressbook, uid=uid)

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "nextcloud-mcp-server"
-version = "0.6.0"
+version = "0.7.0"
 description = ""
 authors = [
     {name = "Chris Coutinho",email = "chris@coutinho.io"}
@@ -11,7 +11,8 @@ dependencies = [
     "mcp[cli] (>=1.10,<1.11)",
     "httpx (>=0.28.1,<0.29.0)",
     "pillow (>=11.2.1,<12.0.0)",
-    "icalendar (>=6.0.0,<7.0.0)"
+    "icalendar (>=6.0.0,<7.0.0)",
+    "pythonvcard4>=0.2.0",
 ]
 
 [tool.pytest.ini_options]

tests/conftest.py

@@ -170,3 +170,83 @@ async def temporary_note_with_attachment(
 
     # Note: The temporary_note fixture's finally block will handle note deletion,
     # which should also trigger the WebDAV directory deletion attempt.
+
+
+@pytest.fixture(scope="module")
+async def temporary_addressbook(nc_client: NextcloudClient):
+    """
+    Fixture to create a temporary addressbook for a test and ensure its deletion afterward.
+    Yields the created addressbook dictionary.
+    """
+    addressbook_name = f"test-addressbook-{uuid.uuid4().hex[:8]}"
+    logger.info(f"Creating temporary addressbook: {addressbook_name}")
+    try:
+        await nc_client.contacts.create_addressbook(
+            name=addressbook_name, display_name=f"Test Addressbook {addressbook_name}"
+        )
+        logger.info(f"Temporary addressbook created: {addressbook_name}")
+        yield addressbook_name
+    finally:
+        logger.info(f"Cleaning up temporary addressbook: {addressbook_name}")
+        try:
+            await nc_client.contacts.delete_addressbook(name=addressbook_name)
+            logger.info(
+                f"Successfully deleted temporary addressbook: {addressbook_name}"
+            )
+        except HTTPStatusError as e:
+            if e.response.status_code != 404:
+                logger.error(
+                    f"HTTP error deleting temporary addressbook {addressbook_name}: {e}"
+                )
+            else:
+                logger.warning(
+                    f"Temporary addressbook {addressbook_name} already deleted (404)."
+                )
+        except Exception as e:
+            logger.error(
+                f"Unexpected error deleting temporary addressbook {addressbook_name}: {e}"
+            )
+
+
+@pytest.fixture
+async def temporary_contact(nc_client: NextcloudClient, temporary_addressbook: str):
+    """
+    Fixture to create a temporary contact in a temporary addressbook and ensure its deletion.
+    Yields the created contact's UID.
+    """
+    contact_uid = f"test-contact-{uuid.uuid4().hex[:8]}"
+    addressbook_name = temporary_addressbook
+    contact_data = {
+        "fn": "John Doe",
+        "email": "john.doe@example.com",
+        "tel": "1234567890",
+    }
+    logger.info(f"Creating temporary contact in addressbook: {addressbook_name}")
+    try:
+        await nc_client.contacts.create_contact(
+            addressbook=addressbook_name,
+            uid=contact_uid,
+            contact_data=contact_data,
+        )
+        logger.info(f"Temporary contact created with UID: {contact_uid}")
+        yield contact_uid
+    finally:
+        logger.info(f"Cleaning up temporary contact: {contact_uid}")
+        try:
+            await nc_client.contacts.delete_contact(
+                addressbook=addressbook_name, uid=contact_uid
+            )
+            logger.info(f"Successfully deleted temporary contact: {contact_uid}")
+        except HTTPStatusError as e:
+            if e.response.status_code != 404:
+                logger.error(
+                    f"HTTP error deleting temporary contact {contact_uid}: {e}"
+                )
+            else:
+                logger.warning(
+                    f"Temporary contact {contact_uid} already deleted (404)."
+                )
+        except Exception as e:
+            logger.error(
+                f"Unexpected error deleting temporary contact {contact_uid}: {e}"
+            )

tests/integration/test_calendar_operations.py

@@ -15,16 +15,6 @@ logger = logging.getLogger(__name__)
 pytestmark = pytest.mark.integration
 
 
-@pytest.fixture
-async def calendar_test_client():
-    """Create a new, isolated NextcloudClient for calendar tests."""
-    client = NextcloudClient.from_env()
-    try:
-        yield client
-    finally:
-        await client.close()
-
-
 @pytest.fixture
 def test_calendar_name():
     """Unique calendar name for testing."""
@@ -32,16 +22,14 @@ def test_calendar_name():
 
 
 @pytest.fixture
-async def temporary_calendar(
-    calendar_test_client: NextcloudClient, test_calendar_name: str
-):
+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
         logger.info(f"Creating temporary calendar: {calendar_name}")
-        result = await calendar_test_client.calendar.create_calendar(
+        result = await nc_client.calendar.create_calendar(
             calendar_name=calendar_name,
             display_name=f"Test Calendar {calendar_name}",
             description="Temporary calendar for integration testing",
@@ -62,16 +50,14 @@ async def temporary_calendar(
         # Cleanup: Delete the temporary calendar
         try:
             logger.info(f"Cleaning up temporary calendar: {calendar_name}")
-            await calendar_test_client.calendar.delete_calendar(calendar_name)
+            await nc_client.calendar.delete_calendar(calendar_name)
             logger.info(f"Successfully deleted temporary calendar: {calendar_name}")
         except Exception as e:
             logger.error(f"Error deleting temporary calendar {calendar_name}: {e}")
 
 
 @pytest.fixture
-async def temporary_event(
-    calendar_test_client: NextcloudClient, temporary_calendar: str
-):
+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
@@ -91,9 +77,7 @@ async def temporary_event(
 
     try:
         logger.info(f"Creating temporary event in calendar: {calendar_name}")
-        result = await calendar_test_client.calendar.create_event(
-            calendar_name, event_data
-        )
+        result = await nc_client.calendar.create_event(calendar_name, event_data)
         event_uid = result.get("uid")
 
         if not event_uid:
@@ -107,9 +91,7 @@ async def temporary_event(
         if event_uid:
             try:
                 logger.info(f"Cleaning up temporary event: {event_uid}")
-                await calendar_test_client.calendar.delete_event(
-                    calendar_name, 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:
@@ -120,9 +102,9 @@ async def temporary_event(
                 )
 
 
-async def test_list_calendars(calendar_test_client: NextcloudClient):
+async def test_list_calendars(nc_client: NextcloudClient):
     """Test listing available calendars."""
-    calendars = await calendar_test_client.calendar.list_calendars()
+    calendars = await nc_client.calendar.list_calendars()
 
     assert isinstance(calendars, list)
 
@@ -144,7 +126,7 @@ async def test_list_calendars(calendar_test_client: NextcloudClient):
 
 
 async def test_create_and_delete_event(
-    calendar_test_client: NextcloudClient, temporary_calendar: str
+    nc_client: NextcloudClient, temporary_calendar: str
 ):
     """Test creating and deleting a basic event."""
     calendar_name = temporary_calendar
@@ -163,9 +145,7 @@ async def test_create_and_delete_event(
     }
 
     try:
-        result = await calendar_test_client.calendar.create_event(
-            calendar_name, event_data
-        )
+        result = await nc_client.calendar.create_event(calendar_name, event_data)
         assert "uid" in result
         assert result["status_code"] in [200, 201, 204]
 
@@ -173,7 +153,7 @@ async def test_create_and_delete_event(
         logger.info(f"Created event with UID: {event_uid}")
 
         # Verify event was created by retrieving it
-        retrieved_event, etag = await calendar_test_client.calendar.get_event(
+        retrieved_event, etag = await nc_client.calendar.get_event(
             calendar_name, event_uid
         )
         assert retrieved_event["uid"] == event_uid
@@ -181,9 +161,7 @@ async def test_create_and_delete_event(
         assert retrieved_event["location"] == "Test Room"
 
         # Delete event
-        delete_result = await calendar_test_client.calendar.delete_event(
-            calendar_name, event_uid
-        )
+        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}")
@@ -194,7 +172,7 @@ async def test_create_and_delete_event(
 
 
 async def test_create_all_day_event(
-    calendar_test_client: NextcloudClient, temporary_calendar: str
+    nc_client: NextcloudClient, temporary_calendar: str
 ):
     """Test creating an all-day event."""
     calendar_name = temporary_calendar
@@ -209,21 +187,19 @@ async def test_create_all_day_event(
     }
 
     try:
-        result = await calendar_test_client.calendar.create_event(
-            calendar_name, event_data
-        )
+        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 calendar_test_client.calendar.get_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 calendar_test_client.calendar.delete_event(calendar_name, event_uid)
+        await nc_client.calendar.delete_event(calendar_name, event_uid)
 
     except Exception as e:
         logger.error(f"All-day event test failed: {e}")
@@ -231,7 +207,7 @@ async def test_create_all_day_event(
 
 
 async def test_create_recurring_event(
-    calendar_test_client: NextcloudClient, temporary_calendar: str
+    nc_client: NextcloudClient, temporary_calendar: str
 ):
     """Test creating a recurring event."""
     calendar_name = temporary_calendar
@@ -248,30 +224,26 @@ async def test_create_recurring_event(
     }
 
     try:
-        result = await calendar_test_client.calendar.create_event(
-            calendar_name, event_data
-        )
+        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 calendar_test_client.calendar.get_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 calendar_test_client.calendar.delete_event(calendar_name, event_uid)
+        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(
-    calendar_test_client: NextcloudClient, temporary_event: dict
-):
+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"]
 
@@ -279,7 +251,7 @@ async def test_list_events_in_range(
     start_datetime = datetime.now()
     end_datetime = datetime.now() + timedelta(days=7)
 
-    events = await calendar_test_client.calendar.get_calendar_events(
+    events = await nc_client.calendar.get_calendar_events(
         calendar_name=calendar_name,
         start_datetime=start_datetime,
         end_datetime=end_datetime,
@@ -300,9 +272,7 @@ async def test_list_events_in_range(
         assert "start_datetime" in event
 
 
-async def test_update_event(
-    calendar_test_client: NextcloudClient, temporary_event: dict
-):
+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"]
@@ -316,15 +286,13 @@ async def test_update_event(
     }
 
     try:
-        result = await calendar_test_client.calendar.update_event(
+        result = await nc_client.calendar.update_event(
             calendar_name, event_uid, updated_data
         )
         assert result["uid"] == event_uid
 
         # Verify updates
-        updated_event, _ = await calendar_test_client.calendar.get_event(
-            calendar_name, event_uid
-        )
+        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"
@@ -338,7 +306,7 @@ async def test_update_event(
 
 
 async def test_create_event_with_attendees(
-    calendar_test_client: NextcloudClient, temporary_calendar: str
+    nc_client: NextcloudClient, temporary_calendar: str
 ):
     """Test creating an event with attendees."""
     calendar_name = temporary_calendar
@@ -356,14 +324,12 @@ async def test_create_event_with_attendees(
     }
 
     try:
-        result = await calendar_test_client.calendar.create_event(
-            calendar_name, event_data
-        )
+        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 calendar_test_client.calendar.get_event(
+        retrieved_event, _ = await nc_client.calendar.get_event(
             calendar_name, event_uid
         )
         assert retrieved_event["title"] == "Meeting with Attendees"
@@ -371,7 +337,7 @@ async def test_create_event_with_attendees(
         assert retrieved_event["status"] == "TENTATIVE"
 
         # Cleanup
-        await calendar_test_client.calendar.delete_event(calendar_name, event_uid)
+        await nc_client.calendar.delete_event(calendar_name, event_uid)
 
     except Exception as e:
         logger.error(f"Event with attendees test failed: {e}")
@@ -379,33 +345,33 @@ async def test_create_event_with_attendees(
 
 
 async def test_get_nonexistent_event(
-    calendar_test_client: NextcloudClient, temporary_calendar: str
+    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 calendar_test_client.calendar.get_event(calendar_name, fake_uid)
+        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(
-    calendar_test_client: NextcloudClient, temporary_calendar: str
+    nc_client: NextcloudClient, temporary_calendar: str
 ):
     """Test deleting a non-existent event."""
     calendar_name = temporary_calendar
     fake_uid = f"nonexistent-{uuid.uuid4()}"
 
-    result = await calendar_test_client.calendar.delete_event(calendar_name, fake_uid)
+    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(
-    calendar_test_client: NextcloudClient, temporary_calendar: str
+    nc_client: NextcloudClient, temporary_calendar: str
 ):
     """Test creating an event with URL and multiple categories."""
     calendar_name = temporary_calendar
@@ -423,14 +389,12 @@ async def test_event_with_url_and_categories(
     }
 
     try:
-        result = await calendar_test_client.calendar.create_event(
-            calendar_name, event_data
-        )
+        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 calendar_test_client.calendar.get_event(
+        retrieved_event, _ = await nc_client.calendar.get_event(
             calendar_name, event_uid
         )
         assert retrieved_event["title"] == "Event with URL and Categories"
@@ -441,7 +405,7 @@ async def test_event_with_url_and_categories(
         assert retrieved_event.get("priority") == 2
 
         # Cleanup
-        await calendar_test_client.calendar.delete_event(calendar_name, event_uid)
+        await nc_client.calendar.delete_event(calendar_name, event_uid)
 
     except Exception as e:
         logger.error(f"Event with metadata test failed: {e}")
@@ -449,7 +413,7 @@ async def test_event_with_url_and_categories(
 
 
 async def test_calendar_operations_error_handling(
-    calendar_test_client: NextcloudClient,
+    nc_client: NextcloudClient,
 ):
     """Test error handling for calendar operations."""
 
@@ -457,6 +421,6 @@ async def test_calendar_operations_error_handling(
     fake_calendar = f"nonexistent_calendar_{uuid.uuid4().hex}"
 
     with pytest.raises(HTTPStatusError):
-        await calendar_test_client.calendar.get_calendar_events(fake_calendar)
+        await nc_client.calendar.get_calendar_events(fake_calendar)
 
     logger.info("Error handling tests completed successfully")

tests/integration/test_contacts_mcp.py

@@ -0,0 +1,86 @@
+"""Integration tests for Contacts MCP tools."""
+
+import logging
+import uuid
+
+import pytest
+from mcp import ClientSession
+
+from nextcloud_mcp_server.client import NextcloudClient
+
+logger = logging.getLogger(__name__)
+pytestmark = pytest.mark.integration
+
+
+async def test_mcp_contacts_workflow(
+    nc_mcp_client: ClientSession, nc_client: NextcloudClient
+):
+    """Test complete Contacts workflow via MCP tools with verification via NextcloudClient."""
+
+    addressbook_name = f"mcp-test-addressbook-{uuid.uuid4().hex[:8]}"
+    unique_suffix = uuid.uuid4().hex[:8]
+    contact_uid = f"mcp-contact-{unique_suffix}"
+    contact_data = {
+        "fn": f"MCP Contact {unique_suffix}",
+        "email": f"mcp.contact.{unique_suffix}@example.com",
+        "tel": "1234567890",
+    }
+
+    try:
+        # 1. Create address book via MCP
+        logger.info(f"Creating address book via MCP: {addressbook_name}")
+        create_ab_result = await nc_mcp_client.call_tool(
+            "nc_contacts_create_addressbook",
+            {"name": addressbook_name, "display_name": f"MCP Test {addressbook_name}"},
+        )
+        assert create_ab_result.isError is False
+
+        # 2. Verify address book creation
+        addressbooks = await nc_client.contacts.list_addressbooks()
+        assert any(ab["name"] == addressbook_name for ab in addressbooks)
+
+        # 3. Create contact via MCP
+        logger.info(f"Creating contact in {addressbook_name} via MCP")
+        create_c_result = await nc_mcp_client.call_tool(
+            "nc_contacts_create_contact",
+            {
+                "addressbook": addressbook_name,
+                "uid": contact_uid,
+                "contact_data": contact_data,
+            },
+        )
+        assert create_c_result.isError is False
+
+        # 4. Verify contact creation
+        contacts = await nc_client.contacts.list_contacts(addressbook=addressbook_name)
+        assert any(c["vcard_id"] == contact_uid for c in contacts)
+
+        # 5. Delete contact via MCP
+        logger.info(f"Deleting contact {contact_uid} via MCP")
+        delete_c_result = await nc_mcp_client.call_tool(
+            "nc_contacts_delete_contact",
+            {"addressbook": addressbook_name, "uid": contact_uid},
+        )
+        assert delete_c_result.isError is False
+
+        # 6. Verify contact deletion
+        contacts = await nc_client.contacts.list_contacts(addressbook=addressbook_name)
+        assert not any(c["vcard_id"] == contact_uid for c in contacts)
+
+        # 7. Delete address book via MCP
+        logger.info(f"Deleting address book {addressbook_name} via MCP")
+        delete_ab_result = await nc_mcp_client.call_tool(
+            "nc_contacts_delete_addressbook", {"name": addressbook_name}
+        )
+        assert delete_ab_result.isError is False
+
+        # 8. Verify address book deletion
+        addressbooks = await nc_client.contacts.list_addressbooks()
+        assert not any(ab["name"] == addressbook_name for ab in addressbooks)
+
+    finally:
+        # Cleanup in case of failure
+        try:
+            await nc_client.contacts.delete_addressbook(name=addressbook_name)
+        except Exception:
+            pass

tests/integration/test_contacts_operations.py

@@ -0,0 +1,88 @@
+"""Integration tests for Contacts CardDAV operations."""
+
+import logging
+import uuid
+
+import pytest
+
+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_list_addressbooks(nc_client: NextcloudClient):
+    """Test listing available addressbooks."""
+    addressbooks = await nc_client.contacts.list_addressbooks()
+
+    assert isinstance(addressbooks, list)
+
+    if not addressbooks:
+        pytest.skip("No addressbooks available - Contacts app may not be enabled")
+
+    logger.info(f"Found {len(addressbooks)} addressbooks")
+
+    # Check structure of addressbooks
+    for addressbook in addressbooks:
+        assert "name" in addressbook
+        assert "display_name" in addressbook
+        assert "getctag" in addressbook
+
+        logger.info(
+            f"Addressbook: {addressbook['name']} - {addressbook['display_name']}"
+        )
+
+
+async def test_create_and_delete_addressbook(
+    nc_client: NextcloudClient, temporary_addressbook: str
+):
+    """Test creating and deleting a basic addressbook."""
+    addressbooks = await nc_client.contacts.list_addressbooks()
+    addressbook_names = [ab["name"] for ab in addressbooks]
+    assert temporary_addressbook in addressbook_names
+
+
+async def test_list_contacts(
+    nc_client: NextcloudClient, temporary_addressbook: str, temporary_contact: str
+):
+    """Test listing contacts in an addressbook."""
+    contacts = await nc_client.contacts.list_contacts(addressbook=temporary_addressbook)
+    contact_uids = [c["vcard_id"] for c in contacts]
+    assert temporary_contact in contact_uids
+
+
+async def test_full_contact_workflow(
+    nc_client: NextcloudClient, temporary_addressbook: str
+):
+    """Test the full workflow of creating, retrieving, and deleting a contact."""
+    addressbook_name = temporary_addressbook
+    contact_uid = f"test-contact-{uuid.uuid4().hex[:8]}"
+    contact_data = {
+        "fn": "Jane Doe",
+        "email": "jane.doe@example.com",
+        "tel": "9876543210",
+    }
+
+    # Create contact
+    await nc_client.contacts.create_contact(
+        addressbook=addressbook_name,
+        uid=contact_uid,
+        contact_data=contact_data,
+    )
+
+    # Verify contact was created by listing
+    contacts = await nc_client.contacts.list_contacts(addressbook=addressbook_name)
+    contact_uids = [c["vcard_id"] for c in contacts]
+    assert contact_uid in contact_uids
+
+    # Delete contact
+    await nc_client.contacts.delete_contact(
+        addressbook=addressbook_name, uid=contact_uid
+    )
+
+    # Verify contact was deleted
+    contacts = await nc_client.contacts.list_contacts(addressbook=addressbook_name)
+    contact_uids = [c["vcard_id"] for c in contacts]
+    assert contact_uid not in contact_uids

tests/integration/test_tables_api.py

@@ -14,7 +14,7 @@ logger = logging.getLogger(__name__)
 pytestmark = pytest.mark.integration
 
 
-@pytest.fixture(scope="session")
+@pytest.fixture(scope="module")
 async def sample_table_info(nc_client: NextcloudClient) -> Dict[str, Any]:
     """
     Fixture to get information about the sample table that comes with Nextcloud Tables.

uv.lock

@@ -505,13 +505,14 @@ wheels = [
 
 [[package]]
 name = "nextcloud-mcp-server"
-version = "0.6.0"
+version = "0.7.0"
 source = { editable = "." }
 dependencies = [
     { name = "httpx" },
     { name = "icalendar" },
     { name = "mcp", extra = ["cli"] },
     { name = "pillow" },
+    { name = "pythonvcard4" },
 ]
 
 [package.dev-dependencies]
@@ -530,6 +531,7 @@ requires-dist = [
     { name = "icalendar", specifier = ">=6.0.0,<7.0.0" },
     { name = "mcp", extras = ["cli"], specifier = ">=1.10,<1.11" },
     { name = "pillow", specifier = ">=11.2.1,<12.0.0" },
+    { name = "pythonvcard4", specifier = ">=0.2.0" },
 ]
 
 [package.metadata.requires-dev]
@@ -843,6 +845,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/45/58/38b5afbc1a800eeea951b9285d3912613f2603bdf897a4ab0f4bd7f405fc/python_multipart-0.0.20-py3-none-any.whl", hash = "sha256:8a62d3a8335e06589fe01f2a3e178cdcc632f3fbe0d492ad9ee0ec35aab1f104", size = 24546, upload-time = "2024-12-16T19:45:44.423Z" },
 ]
 
+[[package]]
+name = "pythonvcard4"
+version = "0.2.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/34/04/02d5952a9d8cbcb9e62b4fc4f6f842e8d43aead6e307f83e6fd6f7352fbd/pythonvcard4-0.2.0.tar.gz", hash = "sha256:236bba2769e459645cfa776407ff07856aced45b437116bf40ddb39bbcefdb6d", size = 5530, upload-time = "2025-04-26T23:18:48.963Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/0d/2f/ee10d88bbe12e4e9e06f81589d999687038e5cd5fec6c05aed57c50aede6/pythonvcard4-0.2.0-py3-none-any.whl", hash = "sha256:dce31355dd50aee537f8883de86f301510e407bc1755a68ec8d5055b64f5c660", size = 5890, upload-time = "2025-04-26T23:18:48.2Z" },
+]
+
 [[package]]
 name = "pyyaml"
 version = "6.0.2"