From 37b1057d2a73e7364d517d8ad1f9571d0abfeaa4 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sun, 3 Aug 2025 14:15:37 +0200 Subject: [PATCH 1/7] feat(contacts): Initialize Contacts App --- nextcloud_mcp_server/app.py | 2 ++ nextcloud_mcp_server/client/__init__.py | 2 ++ nextcloud_mcp_server/server/__init__.py | 2 ++ pyproject.toml | 3 ++- uv.lock | 11 +++++++++++ 5 files changed, 19 insertions(+), 1 deletion(-) diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index 6f359fe..e66fa65 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -12,6 +12,7 @@ from nextcloud_mcp_server.server import ( configure_notes_tools, configure_tables_tools, configure_webdav_tools, + configure_contacts_tools, ) setup_logging() @@ -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(): diff --git a/nextcloud_mcp_server/client/__init__.py b/nextcloud_mcp_server/client/__init__.py index d641a4d..b17162e 100644 --- a/nextcloud_mcp_server/client/__init__.py +++ b/nextcloud_mcp_server/client/__init__.py @@ -5,6 +5,7 @@ from httpx import AsyncClient, Auth, BasicAuth, Request, Response 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 @@ 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() diff --git a/nextcloud_mcp_server/server/__init__.py b/nextcloud_mcp_server/server/__init__.py index 925c987..f3b510b 100644 --- a/nextcloud_mcp_server/server/__init__.py +++ b/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", ] diff --git a/pyproject.toml b/pyproject.toml index 6113df5..da11b60 100644 --- a/pyproject.toml +++ b/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] diff --git a/uv.lock b/uv.lock index a2ba43b..defd8fe 100644 --- a/uv.lock +++ b/uv.lock @@ -512,6 +512,7 @@ dependencies = [ { 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" From 70f01bf40aeaab2b2d808890230b98b3fbf346da Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sun, 3 Aug 2025 14:16:55 +0200 Subject: [PATCH 2/7] Add files --- .../post-installation/install-contacts-app.sh | 3 + main.py | 25 +++ nextcloud_mcp_server/client/contacts.py | 177 ++++++++++++++++++ nextcloud_mcp_server/server/contacts.py | 22 +++ 4 files changed, 227 insertions(+) create mode 100755 app-hooks/post-installation/install-contacts-app.sh create mode 100644 main.py create mode 100644 nextcloud_mcp_server/client/contacts.py create mode 100644 nextcloud_mcp_server/server/contacts.py diff --git a/app-hooks/post-installation/install-contacts-app.sh b/app-hooks/post-installation/install-contacts-app.sh new file mode 100755 index 0000000..7a97d68 --- /dev/null +++ b/app-hooks/post-installation/install-contacts-app.sh @@ -0,0 +1,3 @@ +#!/bin/bash + +php /var/www/html/occ app:enable contacts diff --git a/main.py b/main.py new file mode 100644 index 0000000..302cb46 --- /dev/null +++ b/main.py @@ -0,0 +1,25 @@ +import asyncio +import logging + +from nextcloud_mcp_server.client import NextcloudClient + +logging.basicConfig(level="INFO") +logger = logging.getLogger(__name__) + +client = NextcloudClient.from_env() + + +async def main(): + addressbooks = await client.contacts.list_addressbooks() + # print(addressbooks) + + for addressbook in addressbooks: + contacts = await client.contacts.list_contacts(addressbook=addressbook["name"]) + for contact in contacts: + logger.info( + "Contact etag: %s, details: %s", contact["getetag"], contact["contact"] + ) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/nextcloud_mcp_server/client/contacts.py b/nextcloud_mcp_server/client/contacts.py new file mode 100644 index 0000000..ce8801a --- /dev/null +++ b/nextcloud_mcp_server/client/contacts.py @@ -0,0 +1,177 @@ +"""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 = """ + + + + + + """ + + 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 list_contacts(self, *, addressbook: str): + """List all available contacts for addressbook.""" + + carddav_path = self._get_carddav_base_path() + + report_body = """ + + + + + + """ + + 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 + + # 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( + { + "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 diff --git a/nextcloud_mcp_server/server/contacts.py b/nextcloud_mcp_server/server/contacts.py new file mode 100644 index 0000000..d749010 --- /dev/null +++ b/nextcloud_mcp_server/server/contacts.py @@ -0,0 +1,22 @@ +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) From 0a97357a9c51f0bee4961c19012249340aae74d5 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sun, 3 Aug 2025 14:17:29 +0200 Subject: [PATCH 3/7] remove main.py --- main.py | 25 ------------------------- 1 file changed, 25 deletions(-) delete mode 100644 main.py diff --git a/main.py b/main.py deleted file mode 100644 index 302cb46..0000000 --- a/main.py +++ /dev/null @@ -1,25 +0,0 @@ -import asyncio -import logging - -from nextcloud_mcp_server.client import NextcloudClient - -logging.basicConfig(level="INFO") -logger = logging.getLogger(__name__) - -client = NextcloudClient.from_env() - - -async def main(): - addressbooks = await client.contacts.list_addressbooks() - # print(addressbooks) - - for addressbook in addressbooks: - contacts = await client.contacts.list_contacts(addressbook=addressbook["name"]) - for contact in contacts: - logger.info( - "Contact etag: %s, details: %s", contact["getetag"], contact["contact"] - ) - - -if __name__ == "__main__": - asyncio.run(main()) From ad3e2882033ec5a331710f7dcc0a5e782844e28a Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sun, 3 Aug 2025 14:22:45 +0200 Subject: [PATCH 4/7] test: Replace test_*_clients with single nc_client for tests --- tests/conftest.py | 2 +- tests/integration/test_calendar_operations.py | 114 ++++++------------ 2 files changed, 40 insertions(+), 76 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index ff28f07..cb45de8 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -13,7 +13,7 @@ from nextcloud_mcp_server.client import NextcloudClient logger = logging.getLogger(__name__) -@pytest.fixture(scope="session") +@pytest.fixture(scope="module") async def nc_client() -> AsyncGenerator[NextcloudClient, Any]: """ Fixture to create a NextcloudClient instance for integration tests. diff --git a/tests/integration/test_calendar_operations.py b/tests/integration/test_calendar_operations.py index 1de3d27..94b2aa5 100644 --- a/tests/integration/test_calendar_operations.py +++ b/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") From 21fc55320b0080953f953692e3c02916658bb31a Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sun, 3 Aug 2025 14:25:01 +0200 Subject: [PATCH 5/7] Fix scoping --- tests/integration/test_tables_api.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/integration/test_tables_api.py b/tests/integration/test_tables_api.py index 39a9d05..802bc1e 100644 --- a/tests/integration/test_tables_api.py +++ b/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. From 72cb62a1018747bf29af25323c7f1e7764b35321 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sun, 3 Aug 2025 14:36:16 +0200 Subject: [PATCH 6/7] test(contacts): Add unit/integration tests for a few tools --- nextcloud_mcp_server/app.py | 2 +- nextcloud_mcp_server/client/contacts.py | 58 ++++++++++++ nextcloud_mcp_server/server/contacts.py | 32 +++++++ tests/conftest.py | 80 +++++++++++++++++ tests/integration/test_contacts_mcp.py | 86 ++++++++++++++++++ tests/integration/test_contacts_operations.py | 88 +++++++++++++++++++ 6 files changed, 345 insertions(+), 1 deletion(-) create mode 100644 tests/integration/test_contacts_mcp.py create mode 100644 tests/integration/test_contacts_operations.py diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index e66fa65..c34971d 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -9,10 +9,10 @@ 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, - configure_contacts_tools, ) setup_logging() diff --git a/nextcloud_mcp_server/client/contacts.py b/nextcloud_mcp_server/client/contacts.py index ce8801a..11ecc3f 100644 --- a/nextcloud_mcp_server/client/contacts.py +++ b/nextcloud_mcp_server/client/contacts.py @@ -87,6 +87,62 @@ class ContactsClient(BaseNextcloudClient): 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""" + + + + + + + + {display_name} + + + """ + + 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.""" @@ -135,6 +191,7 @@ class ContactsClient(BaseNextcloudClient): 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) @@ -162,6 +219,7 @@ class ContactsClient(BaseNextcloudClient): contacts.append( { + "vcard_id": vcard_id, "getetag": getetag, "contact": { "fullname": contact.fn, diff --git a/nextcloud_mcp_server/server/contacts.py b/nextcloud_mcp_server/server/contacts.py index d749010..13f89a3 100644 --- a/nextcloud_mcp_server/server/contacts.py +++ b/nextcloud_mcp_server/server/contacts.py @@ -20,3 +20,35 @@ def configure_contacts_tools(mcp: FastMCP): """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.""" + 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.""" + 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) diff --git a/tests/conftest.py b/tests/conftest.py index cb45de8..bf43c95 100644 --- a/tests/conftest.py +++ b/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 +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}" + ) diff --git a/tests/integration/test_contacts_mcp.py b/tests/integration/test_contacts_mcp.py new file mode 100644 index 0000000..f391fa8 --- /dev/null +++ b/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 diff --git a/tests/integration/test_contacts_operations.py b/tests/integration/test_contacts_operations.py new file mode 100644 index 0000000..85628dc --- /dev/null +++ b/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 From 1dfdad5fad37408b8cfef45d906acccdb58a7d72 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sun, 3 Aug 2025 14:42:16 +0200 Subject: [PATCH 7/7] Update README, docstrings, and test scope for temporary_addressbook --- README.md | 12 ++++++++++++ nextcloud_mcp_server/server/contacts.py | 15 +++++++++++++-- tests/conftest.py | 2 +- 3 files changed, 26 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 779089c..550a380 100644 --- a/README.md +++ b/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 | diff --git a/nextcloud_mcp_server/server/contacts.py b/nextcloud_mcp_server/server/contacts.py index 13f89a3..3ee9844 100644 --- a/nextcloud_mcp_server/server/contacts.py +++ b/nextcloud_mcp_server/server/contacts.py @@ -25,7 +25,12 @@ def configure_contacts_tools(mcp: FastMCP): async def nc_contacts_create_addressbook( ctx: Context, *, name: str, display_name: str ): - """Create a new addressbook.""" + """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 @@ -41,7 +46,13 @@ def configure_contacts_tools(mcp: FastMCP): async def nc_contacts_create_contact( ctx: Context, *, addressbook: str, uid: str, contact_data: dict ): - """Create a new contact.""" + """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 diff --git a/tests/conftest.py b/tests/conftest.py index bf43c95..6bf3ea8 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -172,7 +172,7 @@ async def temporary_note_with_attachment( # which should also trigger the WebDAV directory deletion attempt. -@pytest.fixture +@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.