Merge pull request #104 from cbcoutinho/feature/vcard

Initialize Contacts App

Author: cbcoutinho

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-contacts-app.sh

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

nextcloud_mcp_server/app.py

@@ -9,6 +9,7 @@
 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 @@ async def nc_get_capabilities():
 configure_tables_tools(mcp)
 configure_webdav_tools(mcp)
 configure_calendar_tools(mcp)
+configure_contacts_tools(mcp)
 
 
 def run():

nextcloud_mcp_server/client/__init__.py

@@ -5,6 +5,7 @@
 
 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
@@ -43,6 +44,7 @@ def __init__(self, base_url: str, username: str, auth: Auth | None = None):
         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/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 .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": "[email protected]"}.
+        """
+        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

@@ -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]