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/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/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py
index 6f359fe..c34971d 100644
--- a/nextcloud_mcp_server/app.py
+++ b/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():
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/client/contacts.py b/nextcloud_mcp_server/client/contacts.py
new file mode 100644
index 0000000..11ecc3f
--- /dev/null
+++ b/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 = """
+
+
+
+
+
+ """
+
+ 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"""
+
+
+
+
+
+
+
+ {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."""
+
+ 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
+ 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
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/nextcloud_mcp_server/server/contacts.py b/nextcloud_mcp_server/server/contacts.py
new file mode 100644
index 0000000..3ee9844
--- /dev/null
+++ b/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)
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/tests/conftest.py b/tests/conftest.py
index ff28f07..6bf3ea8 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.
@@ -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}"
+ )
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")
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
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.
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"