diff --git a/nextcloud_mcp_server/client/calendar.py b/nextcloud_mcp_server/client/calendar.py
index d462bf27e..b3ebe455d 100644
--- a/nextcloud_mcp_server/client/calendar.py
+++ b/nextcloud_mcp_server/client/calendar.py
@@ -9,6 +9,7 @@
from caldav.async_collection import AsyncCalendar, AsyncEvent
from caldav.async_davclient import AsyncDAVClient
from caldav.elements import cdav, dav
+import httpx
from httpx import Auth
from icalendar import Alarm, Calendar, vDDDTypes, vRecur
from icalendar import Event as ICalEvent
@@ -32,15 +33,120 @@ def __init__(self, base_url: str, username: str, auth: Auth | None = None):
auth: httpx.Auth object (BasicAuth or BearerAuth)
"""
self.username = username
- self.base_url = base_url
+ self.base_url = base_url.rstrip("/")
+ self._auth = auth
+ self._ssl_verify = get_nextcloud_ssl_verify()
# AsyncDAVClient needs the full base URL for proper URL construction
self._dav_client = AsyncDAVClient(
- url=f"{base_url}/remote.php/dav/",
+ url=f"{self.base_url}/remote.php/dav/",
username=username,
auth=auth,
- ssl_verify_cert=get_nextcloud_ssl_verify(), # type: ignore[arg-type] # caldav types say bool|str but passes through to httpx which accepts SSLContext
+ ssl_verify_cert=self._ssl_verify, # type: ignore[arg-type] # caldav types say bool|str but passes through to httpx which accepts SSLContext
)
- self._calendar_home_url = f"{base_url}/remote.php/dav/calendars/{username}/"
+ # Fallback: some servers still support username-based home sets.
+ # Nextcloud typically uses a UUID principal, so we'll discover the
+ # correct home-set lazily via `current-user-principal`.
+ self._calendar_home_url = (
+ f"{self.base_url}/remote.php/dav/calendars/{username}/"
+ )
+ self._calendar_home_url_discovered = False
+
+ async def _ensure_calendar_home_url(self) -> None:
+ """Discover CalDAV calendar-home-set using DAV principal UUID.
+
+ Nextcloud frequently uses UUID principals for `current-user-principal`.
+ The correct calendar home is then exposed via `cal:calendar-home-set`.
+ """
+ if self._calendar_home_url_discovered:
+ return
+
+ timeout = httpx.Timeout(timeout=30, connect=5)
+ try:
+ async with httpx.AsyncClient(
+ auth=self._auth, verify=self._ssl_verify, timeout=timeout
+ ) as client:
+ propfind_principal_body = """
+
+
+
+
+"""
+
+ principal_resp = await client.request(
+ "PROPFIND",
+ f"{self.base_url}/remote.php/dav/",
+ headers={
+ "Depth": "0",
+ "Content-Type": "application/xml",
+ "Accept": "application/xml",
+ },
+ content=propfind_principal_body,
+ )
+ principal_resp.raise_for_status()
+
+ principal_tree = etree.fromstring(principal_resp.content)
+ principal_href_el = principal_tree.find(
+ ".//d:current-user-principal/d:href",
+ namespaces={"d": "DAV:"},
+ )
+ principal_href = (
+ principal_href_el.text.strip()
+ if principal_href_el is not None
+ and principal_href_el.text
+ else None
+ )
+ if not principal_href:
+ raise RuntimeError("current-user-principal href not found")
+
+ propfind_home_body = """
+
+
+
+
+"""
+
+ home_resp = await client.request(
+ "PROPFIND",
+ f"{self.base_url}{principal_href}",
+ headers={
+ "Depth": "0",
+ "Content-Type": "application/xml",
+ "Accept": "application/xml",
+ },
+ content=propfind_home_body,
+ )
+ home_resp.raise_for_status()
+
+ home_tree = etree.fromstring(home_resp.content)
+ home_href_el = home_tree.find(
+ ".//cal:calendar-home-set/d:href",
+ namespaces={
+ "d": "DAV:",
+ "cal": "urn:ietf:params:xml:ns:caldav",
+ },
+ )
+ home_href = (
+ home_href_el.text.strip()
+ if home_href_el is not None and home_href_el.text
+ else None
+ )
+ if home_href:
+ # home_href is a DAV path like /remote.php/dav/calendars//
+ self._calendar_home_url = (
+ f"{self.base_url}{home_href}".rstrip("/") + "/"
+ )
+ logger.info(
+ "Discovered CalDAV calendar-home-set: %s",
+ self._calendar_home_url,
+ )
+ except Exception as e:
+ logger.warning(
+ "CalDAV calendar-home-set discovery failed; using username-based path: %s",
+ e,
+ )
+ finally:
+ # Avoid repeatedly hitting discovery endpoints within a session.
+ self._calendar_home_url_discovered = True
def _get_calendar_url(self, calendar_name: str) -> str:
"""Get the full URL for a calendar."""
@@ -102,6 +208,7 @@ async def _wait_for_calendar_propagation(
async def list_calendars(self) -> List[Dict[str, Any]]:
"""List all available calendars for the user."""
+ await self._ensure_calendar_home_url()
# Use custom PROPFIND with CalendarServer namespace (cs:) for calendar-color.
# caldav library's nsmap lacks "CS" namespace, and its CalendarColor uses
# Apple iCal namespace which Nextcloud doesn't recognize.
@@ -191,13 +298,12 @@ async def create_calendar(
color: str = "#1976D2",
) -> Dict[str, Any]:
"""Create a new calendar with retry on 429 errors."""
+ await self._ensure_calendar_home_url()
# Use custom MKCALENDAR XML instead of caldav library's make_calendar() due to:
# 1. Missing CalendarServer namespace (cs:) in caldav's nsmap
# 2. caldav's CalendarColor uses Apple iCal namespace, not cs:calendar-color
# 3. make_calendar() doesn't support calendar-description or calendar-color params
- calendar_url = (
- f"{self.base_url}/remote.php/dav/calendars/{self.username}/{calendar_name}/"
- )
+ calendar_url = f"{self._calendar_home_url}{calendar_name}/"
mkcalendar_body = f"""
@@ -237,10 +343,9 @@ async def create_calendar(
async def delete_calendar(self, calendar_name: str) -> Dict[str, Any]:
"""Delete a calendar."""
+ await self._ensure_calendar_home_url()
# Use absolute URL for deletion
- calendar_url = (
- f"{self.base_url}/remote.php/dav/calendars/{self.username}/{calendar_name}/"
- )
+ calendar_url = f"{self._calendar_home_url}{calendar_name}/"
await self._dav_client.delete(calendar_url)
logger.debug(f"Deleted calendar: {calendar_name}")
diff --git a/nextcloud_mcp_server/client/contacts.py b/nextcloud_mcp_server/client/contacts.py
index 8f959f66e..f60d6942c 100644
--- a/nextcloud_mcp_server/client/contacts.py
+++ b/nextcloud_mcp_server/client/contacts.py
@@ -2,6 +2,9 @@
import logging
import xml.etree.ElementTree as ET
+from typing import Optional
+
+from xml.sax.saxutils import escape as xml_escape
from pythonvCard4.vcard import Contact
@@ -15,13 +18,107 @@ class ContactsClient(BaseNextcloudClient):
app_name = "contacts"
+ def __init__(self, http_client, username: str):
+ super().__init__(http_client, username)
+ self._carddav_base_path: Optional[str] = None
+ self._principal_id: Optional[str] = None
+ self._carddav_base_path_discovered = False
+
def _get_carddav_base_path(self) -> str:
"""Helper to get the base CardDAV path for contacts."""
+ if self._carddav_base_path:
+ return self._carddav_base_path
+ # Fallback: username-based path. Nextcloud often uses UUID principals.
return f"/remote.php/dav/addressbooks/users/{self.username}"
+ async def _ensure_carddav_base_path(self) -> None:
+ """Discover CardDAV addressbook-home-set via DAV principal UUID."""
+ if self._carddav_base_path_discovered:
+ return
+
+ try:
+ propfind_principal_body = """
+
+
+
+
+"""
+
+ response = await self._make_request(
+ "PROPFIND",
+ "/remote.php/dav/",
+ content=propfind_principal_body,
+ headers={
+ "Depth": "0",
+ "Content-Type": "application/xml",
+ "Accept": "application/xml",
+ },
+ )
+
+ ns = {"d": "DAV:", "card": "urn:ietf:params:xml:ns:carddav"}
+ root = ET.fromstring(response.content)
+ principal_href_el = root.find(
+ ".//d:current-user-principal/d:href",
+ ns,
+ )
+ principal_href = (
+ principal_href_el.text.strip()
+ if principal_href_el is not None and principal_href_el.text
+ else None
+ )
+ if not principal_href:
+ raise RuntimeError("current-user-principal href not found")
+
+ principal_id = principal_href.rstrip("/").split("/")[-1]
+ self._principal_id = principal_id
+
+ propfind_home_body = """
+
+
+
+
+"""
+
+ home_resp = await self._make_request(
+ "PROPFIND",
+ principal_href,
+ content=propfind_home_body,
+ headers={
+ "Depth": "0",
+ "Content-Type": "application/xml",
+ "Accept": "application/xml",
+ },
+ )
+
+ home_root = ET.fromstring(home_resp.content)
+ home_href_el = home_root.find(
+ ".//card:addressbook-home-set/d:href",
+ ns,
+ )
+ home_href = (
+ home_href_el.text.strip()
+ if home_href_el is not None and home_href_el.text
+ else None
+ )
+
+ if home_href:
+ self._carddav_base_path = home_href.rstrip("/")
+ logger.info(
+ "Discovered CardDAV addressbook-home-set: %s",
+ self._carddav_base_path,
+ )
+ except Exception as e:
+ logger.warning(
+ "CardDAV addressbook-home-set discovery failed; using username-based path: %s",
+ e,
+ )
+ finally:
+ self._carddav_base_path_discovered = True
+
async def list_addressbooks(self):
"""List all available addressbooks for the user."""
+ await self._ensure_carddav_base_path()
carddav_path = self._get_carddav_base_path()
propfind_body = """
@@ -58,7 +155,10 @@ async def list_addressbooks(self):
# Extract addressbook name from href
addressbook_name = href_text.rstrip("/").split("/")[-1]
- if not addressbook_name or addressbook_name == self.username:
+ if not addressbook_name:
+ continue
+ # Don't treat the principal container itself as a real addressbook.
+ if self._principal_id and addressbook_name == self._principal_id:
continue
# Get properties
@@ -93,6 +193,7 @@ async def list_addressbooks(self):
async def create_addressbook(self, *, name: str, display_name: str):
"""Create a new addressbook."""
+ await self._ensure_carddav_base_path()
carddav_path = self._get_carddav_base_path()
url = f"{carddav_path}/{name}/"
@@ -117,12 +218,14 @@ async def create_addressbook(self, *, name: str, display_name: str):
async def delete_addressbook(self, *, name: str):
"""Delete an addressbook."""
+ await self._ensure_carddav_base_path()
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."""
+ await self._ensure_carddav_base_path()
carddav_path = self._get_carddav_base_path()
url = f"{carddav_path}/{addressbook}/{uid}.vcf"
@@ -143,6 +246,7 @@ async def create_contact(self, *, addressbook: str, uid: str, contact_data: dict
async def delete_contact(self, *, addressbook: str, uid: str):
"""Delete a contact."""
+ await self._ensure_carddav_base_path()
carddav_path = self._get_carddav_base_path()
url = f"{carddav_path}/{addressbook}/{uid}.vcf"
await self._make_request("DELETE", url)
@@ -151,6 +255,7 @@ async def update_contact(
self, *, addressbook: str, uid: str, contact_data: dict, etag: str = ""
):
"""Update an existing contact while preserving all existing properties."""
+ await self._ensure_carddav_base_path()
carddav_path = self._get_carddav_base_path()
url = f"{carddav_path}/{addressbook}/{uid}.vcf"
@@ -192,16 +297,146 @@ async def update_contact(
await self._make_request("PUT", url, content=vcard_content, headers=headers)
async def list_contacts(self, *, addressbook: str):
- """List all available contacts for addressbook."""
+ """List contacts for addressbook (can be expensive)."""
+ return await self.list_contacts_query(addressbook=addressbook)
+
+ async def list_contacts_query(
+ self,
+ *,
+ addressbook: str,
+ query: str | None = None,
+ limit: int | None = None,
+ ):
+ """List contacts with optional CardDAV server-side filtering."""
+ await self._ensure_carddav_base_path()
carddav_path = self._get_carddav_base_path()
- report_body = """
+ filter_xml = ""
+ if query:
+ def translit_ru_to_lat(s: str) -> str:
+ """Very small best-effort transliteration for Russian -> Latin.
+
+ Used to make CardDAV text-match work when vCard stores names
+ in Latin (e.g. "Agarkov") while the user searches in Cyrillic.
+ """
+ mapping = {
+ "а": "a",
+ "б": "b",
+ "в": "v",
+ "г": "g",
+ "д": "d",
+ "е": "e",
+ "ё": "e",
+ "ж": "zh",
+ "з": "z",
+ "и": "i",
+ "й": "y",
+ "к": "k",
+ "л": "l",
+ "м": "m",
+ "н": "n",
+ "о": "o",
+ "п": "p",
+ "р": "r",
+ "с": "s",
+ "т": "t",
+ "у": "u",
+ "ф": "f",
+ "х": "kh",
+ "ц": "ts",
+ "ч": "ch",
+ "ш": "sh",
+ "щ": "shch",
+ "ъ": "",
+ "ы": "y",
+ "ь": "",
+ "э": "e",
+ "ю": "yu",
+ "я": "ya",
+ "А": "A",
+ "Б": "B",
+ "В": "V",
+ "Г": "G",
+ "Д": "D",
+ "Е": "E",
+ "Ё": "E",
+ "Ж": "Zh",
+ "З": "Z",
+ "И": "I",
+ "Й": "Y",
+ "К": "K",
+ "Л": "L",
+ "М": "M",
+ "Н": "N",
+ "О": "O",
+ "П": "P",
+ "Р": "R",
+ "С": "S",
+ "Т": "T",
+ "У": "U",
+ "Ф": "F",
+ "Х": "Kh",
+ "Ц": "Ts",
+ "Ч": "Ch",
+ "Ш": "Sh",
+ "Щ": "Shch",
+ "Ъ": "",
+ "Ы": "Y",
+ "Ь": "",
+ "Э": "E",
+ "Ю": "Yu",
+ "Я": "Ya",
+ }
+
+ return "".join(mapping.get(ch, ch) for ch in s)
+
+ # Server-side text search on the FN (full name) vCard property.
+ # RFC6352 supports CardDAV text matching via card:text-match.
+ q_raw = str(query)
+ q_translit = translit_ru_to_lat(q_raw)
+
+ q_xml = xml_escape(q_raw, {"\"": """})
+ q_translit_xml = (
+ xml_escape(q_translit, {"\"": """})
+ if q_translit != q_raw
+ else None
+ )
+
+ # Best-effort: match either the original query (if server has Cyrillic)
+ # or its transliteration (common when vCard stores Latin names).
+ match_parts = [
+ f"{q_xml}"
+ ]
+ if q_translit_xml:
+ match_parts.append(
+ f"{q_translit_xml}"
+ )
+
+ filter_xml = f"""
+
+
+ {''.join(match_parts)}
+
+ """
+
+ limit_xml = ""
+ if limit is not None:
+ # Best-effort: use CardDAV limit stanza. Some servers may ignore it,
+ # but it won't hurt correctness when combined with filtering.
+ limit_xml = f"""
+
+ {int(limit)}
+ """
+
+ report_body = f"""
+ {filter_xml}
+ {limit_xml}
"""
headers = {
@@ -263,7 +498,39 @@ async def list_contacts(self, *, addressbook: str):
logger.info("Skip missing addressdata")
continue
- contact = Contact.from_vcard(addressdata)
+ def _remove_bday_lines(vcard: str) -> str:
+ # Some Nextcloud contacts contain invalid BDAY values that break vCard parsing.
+ # Best-effort sanitize by removing all BDAY lines.
+ lines = vcard.splitlines()
+ kept: list[str] = []
+ for line in lines:
+ # Handle both "BDAY:" and "BDAY;PARAM=...:" variants.
+ if line.lstrip().upper().startswith("BDAY"):
+ continue
+ kept.append(line)
+ return "\n".join(kept)
+
+ try:
+ contact = Contact.from_vcard(addressdata)
+ except Exception as e:
+ # Avoid failing the whole addressbook due to a single broken contact.
+ logger.warning(
+ "Failed to parse vCard for '%s' (addressbook='%s'): %s. Retrying without BDAY.",
+ vcard_id,
+ addressbook,
+ e,
+ )
+ try:
+ sanitized = _remove_bday_lines(addressdata)
+ contact = Contact.from_vcard(sanitized)
+ except Exception as e2:
+ logger.warning(
+ "Retry without BDAY failed for vCard '%s' (addressbook='%s'): %s. Skipping contact.",
+ vcard_id,
+ addressbook,
+ e2,
+ )
+ continue
contacts.append(
{
@@ -280,6 +547,9 @@ async def list_contacts(self, *, addressbook: str):
}
)
+ if limit is not None and len(contacts) >= limit:
+ break
+
logger.debug(f"Found {len(contacts)} contacts")
return contacts
diff --git a/nextcloud_mcp_server/client/sharing.py b/nextcloud_mcp_server/client/sharing.py
index 07ec45ae8..156987b2c 100644
--- a/nextcloud_mcp_server/client/sharing.py
+++ b/nextcloud_mcp_server/client/sharing.py
@@ -2,6 +2,7 @@
import logging
from typing import Any
+from urllib.parse import unquote
from .base import BaseNextcloudClient, retry_on_429
@@ -13,6 +14,41 @@ class SharingClient(BaseNextcloudClient):
app_name = "sharing"
+ def __init__(self, http_client, username: str):
+ super().__init__(http_client, username)
+ self._resolved_user_ids: dict[str, str] = {}
+
+ async def _resolve_user_id(self, share_with: str) -> str:
+ """Resolve Nextcloud internal user id (UUID) for `share_type=0`.
+
+ On some Nextcloud instances `files_sharing` expects `shareWith` to be
+ an internal user id rather than the login/username.
+ """
+ # Already looks like a UUID/principal id.
+ if share_with in self._resolved_user_ids:
+ return self._resolved_user_ids[share_with]
+
+ try:
+ resp = await self._client.get(
+ "/ocs/v2.php/cloud/users",
+ params={"search": share_with},
+ headers={"OCS-APIRequest": "true", "Accept": "application/json"},
+ )
+ resp.raise_for_status()
+ data = resp.json()
+ users = data.get("ocs", {}).get("data", {}).get("users") or []
+ if users:
+ self._resolved_user_ids[share_with] = users[0]
+ return users[0]
+ except Exception as e:
+ logger.debug(
+ "Failed to resolve internal user id for '%s': %s",
+ share_with,
+ e,
+ )
+
+ return share_with
+
@retry_on_429
async def create_share(
self,
@@ -42,6 +78,48 @@ async def create_share(
Raises:
HTTPStatusError: If the request fails
"""
+ # Nextcloud can provide the "path" in multiple forms:
+ # - decoded relative file path (what OCS expects)
+ # - percent-encoded path (from WebDAV href segments)
+ # - WebDAV absolute-like href: /remote.php/dav/files//
+ #
+ # Normalize to "relative to user's files root" before calling OCS.
+ path = unquote(path)
+
+ # Strip WebDAV DAV prefix if a href was passed by caller/agent.
+ # Keep everything after /remote.php/dav/files//
+ dav_prefix = "/remote.php/dav/files/"
+ if dav_prefix in path:
+ after = path.split(dav_prefix, 1)[1]
+ # after: "/"
+ parts = after.split("/", 1)
+ if len(parts) == 2:
+ path = parts[1]
+ else:
+ path = ""
+
+ # OCS expects path without leading slash.
+ path = path.lstrip("/")
+
+ # For user shares Nextcloud may expect internal UUID user id.
+ # Additionally, this Nextcloud instance forbids sharing with yourself.
+ # If caller requested `share_type=0` and `share_with` resolves to the
+ # current user, fall back to a public share (share_type=3) so that
+ # `nc_share_create` can still generate a usable link for the caller.
+ if share_type == 0 and share_with:
+ resolved_share_with = await self._resolve_user_id(share_with)
+ resolved_self = await self._resolve_user_id(self.username)
+ if resolved_share_with == resolved_self:
+ logger.info(
+ "Share with yourself is forbidden for internal shares; "
+ "falling back to public share_type=3 for path '%s'",
+ path,
+ )
+ share_type = 3
+ share_with = ""
+ else:
+ share_with = resolved_share_with
+
response = await self._client.post(
"/ocs/v2.php/apps/files_sharing/api/v1/shares",
headers={"OCS-APIRequest": "true", "Accept": "application/json"},
diff --git a/nextcloud_mcp_server/client/webdav.py b/nextcloud_mcp_server/client/webdav.py
index 2d6356e54..562fb5336 100644
--- a/nextcloud_mcp_server/client/webdav.py
+++ b/nextcloud_mcp_server/client/webdav.py
@@ -19,8 +19,81 @@ class WebDAVClient(BaseNextcloudClient):
app_name = "webdav"
+ def __init__(self, http_client, username: str):
+ super().__init__(http_client, username)
+ # Nextcloud often uses UUID principals for WebDAV files home.
+ # Example correct base: /remote.php/dav/files//
+ self._files_principal_id: Optional[str] = None
+ self._files_principal_discovered = False
+
+ async def _ensure_files_principal_id(self) -> None:
+ """Discover WebDAV files principal id via DAV current-user-principal."""
+ if self._files_principal_discovered:
+ return
+
+ try:
+ propfind_principal_body = """
+
+
+
+
+"""
+
+ response = await self._make_request(
+ "PROPFIND",
+ "/remote.php/dav/",
+ content=propfind_principal_body,
+ headers={
+ "Depth": "0",
+ "Content-Type": "application/xml",
+ "Accept": "application/xml",
+ },
+ )
+
+ root = ET.fromstring(response.content)
+
+ principal_href: Optional[str] = None
+ for elem in root.iter():
+ local = elem.tag.split("}")[-1]
+ if local != "current-user-principal":
+ continue
+ for sub in elem.iter():
+ sub_local = sub.tag.split("}")[-1]
+ if sub_local == "href" and sub.text:
+ principal_href = sub.text.strip()
+ break
+ if principal_href:
+ break
+
+ if not principal_href:
+ raise RuntimeError("current-user-principal href not found")
+
+ # principal_href example:
+ # /remote.php/dav/principals/users//
+ principal_id = principal_href.rstrip("/").split("/")[-1]
+ if not principal_id:
+ raise RuntimeError("Unable to extract files principal id")
+
+ self._files_principal_id = principal_id
+ except Exception as e:
+ logger.warning(
+ "WebDAV files principal discovery failed; using username-based path: %s",
+ e,
+ )
+ self._files_principal_id = None
+ finally:
+ self._files_principal_discovered = True
+
+ def _get_files_principal_id(self) -> str:
+ return self._files_principal_id or self.username
+
+ def _get_webdav_base_path(self) -> str:
+ """Helper to get the base WebDAV path for the authenticated user."""
+ return f"/remote.php/dav/files/{self._get_files_principal_id()}"
+
async def delete_resource(self, path: str) -> Dict[str, Any]:
"""Delete a resource (file or directory) via WebDAV DELETE."""
+ await self._ensure_files_principal_id()
# Ensure path ends with a slash if it's a directory
if not path.endswith("/"):
path_with_slash = f"{path}/"
@@ -108,6 +181,7 @@ async def add_note_attachment(
mime_type: Optional[str] = None,
) -> Dict[str, Any]:
"""Add/Update an attachment to a note via WebDAV PUT."""
+ await self._ensure_files_principal_id()
# Construct paths based on provided category
webdav_base = self._get_webdav_base_path()
category_path_part = f"{category}/" if category else ""
@@ -185,6 +259,7 @@ async def get_note_attachment(
self, note_id: int, filename: str, category: Optional[str] = None
) -> Tuple[bytes, str]:
"""Fetch a specific attachment from a note via WebDAV GET."""
+ await self._ensure_files_principal_id()
webdav_base = self._get_webdav_base_path()
category_path_part = f"{category}/" if category else ""
attachment_dir_segment = f".attachments.{note_id}"
@@ -220,6 +295,7 @@ async def get_note_attachment(
async def list_directory(self, path: str = "") -> List[Dict[str, Any]]:
"""List files and directories in the specified path via WebDAV PROPFIND."""
+ await self._ensure_files_principal_id()
webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}"
if not webdav_path.endswith("/"):
webdav_path += "/"
@@ -259,7 +335,10 @@ async def list_directory(self, path: str = "") -> List[Dict[str, Any]]:
# Extract file/directory name from href
href_text = href.text or ""
- name = href_text.rstrip("/").split("/")[-1]
+ # Nextcloud may return percent-encoded href segments.
+ # Decode so downstream tools (e.g. OCS sharing) can use this
+ # name as-is (decoded file/folder path).
+ name = unquote(href_text.rstrip("/").split("/")[-1])
if not name:
continue
@@ -318,6 +397,7 @@ async def list_directory(self, path: str = "") -> List[Dict[str, Any]]:
async def read_file(self, path: str) -> Tuple[bytes, str]:
"""Read a file's content via WebDAV GET."""
+ await self._ensure_files_principal_id()
webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}"
logger.debug(f"Reading file: {path}")
@@ -345,6 +425,7 @@ async def write_file(
self, path: str, content: bytes, content_type: Optional[str] = None
) -> Dict[str, Any]:
"""Write content to a file via WebDAV PUT."""
+ await self._ensure_files_principal_id()
webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}"
logger.debug(f"Writing file: {path}")
@@ -376,6 +457,7 @@ async def create_directory(
self, path: str, recursive: bool = False
) -> Dict[str, Any]:
"""Create a directory via WebDAV MKCOL."""
+ await self._ensure_files_principal_id()
webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}"
if not webdav_path.endswith("/"):
webdav_path += "/"
@@ -433,6 +515,7 @@ async def move_resource(
Returns:
Dict with status_code and optional message
"""
+ await self._ensure_files_principal_id()
source_webdav_path = f"{self._get_webdav_base_path()}/{source_path.lstrip('/')}"
destination_webdav_path = (
f"{self._get_webdav_base_path()}/{destination_path.lstrip('/')}"
@@ -514,6 +597,7 @@ async def copy_resource(
Returns:
Dict with status_code and optional message
"""
+ await self._ensure_files_principal_id()
source_webdav_path = f"{self._get_webdav_base_path()}/{source_path.lstrip('/')}"
destination_webdav_path = (
f"{self._get_webdav_base_path()}/{destination_path.lstrip('/')}"
@@ -595,6 +679,7 @@ async def search_files(
Returns:
List of file/directory dictionaries with requested properties
"""
+ await self._ensure_files_principal_id()
# Default properties if not specified
if properties is None:
properties = [
@@ -651,8 +736,8 @@ def _build_search_xml(
) -> str:
"""Build the XML body for a SEARCH request."""
# Construct the scope path
- username = self.username
- scope_path = f"/files/{username}"
+ files_principal_id = self._get_files_principal_id()
+ scope_path = f"/files/{files_principal_id}"
if scope:
scope_path = f"{scope_path}/{scope.lstrip('/')}"
@@ -785,6 +870,9 @@ def _parse_search_response(
else:
relative_path = href_text.rstrip("/").split("/")[-1]
+ # Decode percent-encoded path segments (e.g. for Cyrillic names).
+ relative_path = unquote(relative_path)
+
# Get properties
propstat = response_elem.find(".//{DAV:}propstat")
if propstat is None:
@@ -1204,6 +1292,7 @@ async def get_files_by_tag(self, tag_id: int) -> list[dict[str, Any]]:
Returns:
List of file info dictionaries with path, size, content_type, etc.
"""
+ await self._ensure_files_principal_id()
# Use WebDAV REPORT method with systemtag filter, requesting all properties
report_body = f"""
@@ -1263,7 +1352,9 @@ async def get_files_by_tag(self, tag_id: int) -> list[dict[str, Any]]:
# Decode href path and extract the file path
href_path = unquote(href_elem.text)
# Remove WebDAV prefix to get user-relative path
- webdav_prefix = f"/remote.php/dav/files/{self.username}/"
+ webdav_prefix = (
+ f"/remote.php/dav/files/{self._get_files_principal_id()}/"
+ )
file_path = href_path.replace(webdav_prefix, "/")
# Parse last modified timestamp
@@ -1308,6 +1399,7 @@ async def get_file_info(self, path: str) -> dict[str, Any] | None:
File info dictionary with id, name, size, content_type, etc.
Returns None if file not found.
"""
+ await self._ensure_files_principal_id()
webdav_path = f"{self._get_webdav_base_path()}/{path.lstrip('/')}"
propfind_body = """
diff --git a/nextcloud_mcp_server/server/contacts.py b/nextcloud_mcp_server/server/contacts.py
index a5e91da91..15384ae25 100644
--- a/nextcloud_mcp_server/server/contacts.py
+++ b/nextcloud_mcp_server/server/contacts.py
@@ -123,7 +123,11 @@ async def nc_contacts_list_addressbooks(ctx: Context) -> ListAddressBooksRespons
@require_scopes("contacts:read")
@instrument_tool
async def nc_contacts_list_contacts(
- ctx: Context, *, addressbook: str
+ ctx: Context,
+ *,
+ addressbook: str,
+ query: str | None = None,
+ limit: int | None = 50,
) -> ListContactsResponse:
"""List all contacts in the specified addressbook.
@@ -131,9 +135,13 @@ async def nc_contacts_list_contacts(
addressbook: The URI slug of the addressbook (e.g. "contacts"),
not the display name. Use nc_contacts_list_addressbooks to
find available URI slugs.
+ query: Optional text to search by (matched server-side against vCard `FN`).
+ limit: Maximum number of contacts to return (best-effort; defaults to 50).
"""
client = await get_client(ctx)
- contacts_data = await client.contacts.list_contacts(addressbook=addressbook)
+ contacts_data = await client.contacts.list_contacts_query(
+ addressbook=addressbook, query=query, limit=limit
+ )
contacts = [_raw_contact_to_model(c) for c in contacts_data]
return ListContactsResponse(
contacts=contacts, addressbook=addressbook, total_count=len(contacts)