feat(webdav): expose file tags

nextcloud_mcp_server/client/webdav.py

@@ -2684,3 +2684,104 @@ async def remove_tag_from_file(self, file_id: int, tag_id: int) -> bool:
 
         response.raise_for_status()
         return True
+
+    # -- File tags (systemtags) ---------------------------------------------
+    #
+    # The client already covered tags (create/assign/remove/search); only two
+    # read paths were missing: list every tag, and list the tags of one file.
+
+    _TAG_PROPFIND = """<?xml version="1.0"?>
+<d:propfind xmlns:d="DAV:" xmlns:oc="http://owncloud.org/ns">
+  <d:prop>
+    <oc:id/>
+    <oc:display-name/>
+    <oc:user-visible/>
+    <oc:user-assignable/>
+  </d:prop>
+</d:propfind>"""
+
+    async def list_tags(self) -> List[Dict[str, Any]]:
+        """Every system tag defined on this instance."""
+        response = await self._make_request(
+            "PROPFIND",
+            "/remote.php/dav/systemtags/",
+            headers={
+                "Depth": "1",
+                "Content-Type": "text/xml",
+                "OCS-APIRequest": "true",
+            },
+            content=self._TAG_PROPFIND,
+        )
+
+        return self._tags_from_multistatus(ET.fromstring(response.content))
+
+    @staticmethod
+    def _tags_from_multistatus(root: Any) -> List[Dict[str, Any]]:
+        """Extract tags from a systemtags PROPFIND response.
+
+        The collection itself comes back as a response element and is skipped.
+        An entry without an id or display name cannot be used for anything, so
+        it is dropped rather than surfaced with placeholder values.
+        """
+        tags: List[Dict[str, Any]] = []
+        for response_elem in root.findall("{DAV:}response"):
+            href = response_elem.find("{DAV:}href")
+            if href is None or href.text == "/remote.php/dav/systemtags/":
+                continue
+            name_elem = response_elem.find(".//{http://owncloud.org/ns}display-name")
+            id_elem = response_elem.find(".//{http://owncloud.org/ns}id")
+            # An entry without a usable id or name cannot be acted on, and an
+            # empty display name would also make the sort key ambiguous.
+            if id_elem is None or not id_elem.text:
+                continue
+            if name_elem is None or not (name_elem.text or "").strip():
+                continue
+            assignable = response_elem.find(
+                ".//{http://owncloud.org/ns}user-assignable"
+            )
+            tags.append(
+                {
+                    "id": int(id_elem.text),
+                    "name": name_elem.text or "",
+                    "assignable": (assignable is None or assignable.text != "false"),
+                }
+            )
+        return sorted(tags, key=lambda t: t["name"].lower())
+
+    async def get_file_tags(self, path: str) -> Dict[str, Any]:
+        """The tags assigned to one file.
+
+        Nextcloud only reports tag *ids* on the file, so the names are looked
+        up from the full list. That is one extra call, but ids alone are of
+        little use to the caller -- and unguessable for a language model.
+        """
+        file_id = await self.get_fileid(path)
+        if not file_id:
+            raise ValueError(f"No file id for path: {path}")
+
+        response = await self._make_request(
+            "PROPFIND",
+            f"/remote.php/dav/systemtags-relations/files/{file_id}",
+            headers={
+                "Depth": "1",
+                "Content-Type": "text/xml",
+                "OCS-APIRequest": "true",
+            },
+            content=self._TAG_PROPFIND,
+        )
+
+        root = ET.fromstring(response.content)
+        ids: List[int] = []
+        for id_elem in root.findall(".//{http://owncloud.org/ns}id"):
+            if id_elem.text and id_elem.text.isdigit():
+                ids.append(int(id_elem.text))
+
+        names = {t["id"]: t["name"] for t in await self.list_tags()}
+        return {
+            "path": path,
+            "file_id": file_id,
+            "tags": [
+                {"id": i, "name": names.get(i, f"(unknown tag {i})")}
+                for i in sorted(set(ids))
+            ],
+        }

nextcloud_mcp_server/models/webdav.py

@@ -255,3 +255,46 @@ class SearchFilesResponse(BaseResponse):
     filters_applied: Optional[dict] = Field(
         None, description="Filters that were applied to the search"
     )
+
+
+class SystemTag(BaseModel):
+    """A system tag defined on the instance."""
+
+    id: int = Field(description="Tag id")
+    name: str = Field(description="Tag display name")
+    assignable: bool = Field(
+        default=True, description="Whether users may assign this tag"
+    )
+
+
+class ListTagsResponse(BaseResponse):
+    """Every system tag on the instance."""
+
+    tags: List[SystemTag] = Field(default_factory=list, description="Tags, by name")
+    total_count: int = Field(description="Number of tags")
+
+
+class FileTagsResponse(BaseResponse):
+    """Tags assigned to one file."""
+
+    path: str = Field(description="Path the tags belong to")
+    file_id: str = Field(description="Nextcloud file id")
+    tags: List[SystemTag] = Field(default_factory=list, description="Assigned tags")
+
+
+class FilesByTagResponse(BaseResponse):
+    """Files carrying a given tag."""
+
+    tag: str = Field(description="Tag name that was searched for")
+    tag_id: Optional[int] = Field(None, description="Tag id, when the tag exists")
+    files: List[FileInfo] = Field(default_factory=list, description="Matching files")
+    total_count: int = Field(description="Number of matching files")
+
+
+class TagFileResponse(BaseResponse):
+    """Outcome of attaching or detaching a tag."""
+
+    path: str = Field(description="Path that was tagged or untagged")
+    tag: str = Field(description="Tag name")
+    tag_id: int = Field(description="Tag id")
+    assigned: bool = Field(description="True after tagging, False after untagging")

nextcloud_mcp_server/server/webdav.py

@@ -27,7 +27,14 @@
     SearchFilesResponse,
     WriteFileResponse,
 )
-from nextcloud_mcp_server.models.webdav import ParseStatus
+from nextcloud_mcp_server.models.webdav import (
+    FilesByTagResponse,
+    FileTagsResponse,
+    ListTagsResponse,
+    ParseStatus,
+    SystemTag,
+    TagFileResponse,
+)
 from nextcloud_mcp_server.observability.metrics import instrument_tool
 from nextcloud_mcp_server.server.tag_exclusion import (
     get_excluded_file_paths,
@@ -164,11 +171,11 @@ def _read_capped() -> bytes | None:
     )
 
 
-async def _resolve_commented_file(client: "NextcloudClient", path: str) -> int:
-    """Resolve ``path`` to the file ID the comments collection is keyed by.
+async def _resolve_file_id(client: "NextcloudClient", path: str) -> int:
+    """Resolve ``path`` to the numeric file ID the DAV collections are keyed by.
 
-    Shared by both comment tools so the excluded-tag guard and the
-    does-it-exist check cannot drift between reading and writing comments.
+    Shared by the comment and tag tools so the excluded-tag guard and the
+    does-it-exist check cannot drift between them.
 
     Raises:
         ToolError: If the path is excluded by tag, resolves to nothing, or
@@ -1083,7 +1090,7 @@ async def nc_webdav_list_comments(
             raise ValueError(f"offset must not be negative, got {offset}")
 
         client = await get_client(ctx)
-        file_id = await _resolve_commented_file(client, path)
+        file_id = await _resolve_file_id(client, path)
 
         comments = await client.webdav.list_comments(
             file_id, limit=limit, offset=offset
@@ -1144,7 +1151,7 @@ async def nc_webdav_create_comment(
             )
 
         client = await get_client(ctx)
-        file_id = await _resolve_commented_file(client, path)
+        file_id = await _resolve_file_id(client, path)
 
         comment_id = await client.webdav.create_comment(file_id, message)
         return CreateFileCommentResponse(
@@ -1153,3 +1160,124 @@ async def nc_webdav_create_comment(
             comment_id=comment_id,
             message=message,
         )
+
+    @mcp.tool(
+        title="List Tags",
+        annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True),
+    )
+    @require_scopes("files.read")
+    @instrument_tool
+    async def nc_webdav_list_tags(ctx: Context) -> ListTagsResponse:
+        """List all system tags available for tagging files."""
+        client = await get_client(ctx)
+        tags = [SystemTag(**t) for t in await client.webdav.list_tags()]
+        return ListTagsResponse(tags=tags, total_count=len(tags))
+
+    @mcp.tool(
+        title="Get File Tags",
+        annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True),
+    )
+    @require_scopes("files.read")
+    @instrument_tool
+    async def nc_webdav_get_file_tags(path: str, ctx: Context) -> FileTagsResponse:
+        """List the tags assigned to one file.
+
+        Args:
+            path: Path to the file, relative to the user's files root.
+        """
+        client = await get_client(ctx)
+        # Resolving through the shared helper applies the excluded-tag guard,
+        # so an excluded path cannot be probed for existence via its tags.
+        await _resolve_file_id(client, path)
+
+        data = await client.webdav.get_file_tags(path)
+        return FileTagsResponse(
+            path=data["path"],
+            file_id=str(data["file_id"]),
+            tags=[SystemTag(id=t["id"], name=t["name"]) for t in data["tags"]],
+        )
+
+    @mcp.tool(
+        title="Find Files By Tag",
+        annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True),
+    )
+    @require_scopes("files.read")
+    @with_links
+    @instrument_tool
+    async def nc_webdav_find_by_tag_name(tag: str, ctx: Context) -> FilesByTagResponse:
+        """Find all files carrying a given tag.
+
+        Args:
+            tag: Tag name, case-sensitive.
+        """
+        client = await get_client(ctx)
+        found = await client.webdav.get_tag_by_name(tag)
+        if found is None or found.get("id") is None:
+            return FilesByTagResponse(tag=tag, files=[], total_count=0)
+
+        files = await client.webdav.get_files_by_tag(found["id"])
+
+        # A tag can be attached to an excluded file; the listing must not
+        # surface it any more than a directory listing would.
+        excluded = await get_excluded_file_paths(client.webdav)
+        if excluded:
+            files = [
+                f for f in files if not is_path_excluded(f.get("path", ""), excluded)
+            ]
+
+        return FilesByTagResponse(
+            tag=tag,
+            tag_id=found["id"],
+            files=[FileInfo(**f) for f in files],
+            total_count=len(files),
+        )
+
+    @mcp.tool(
+        title="Tag File",
+        annotations=ToolAnnotations(
+            destructiveHint=False,
+            idempotentHint=True,  # Same tag twice = same end state
+            openWorldHint=True,
+        ),
+    )
+    @require_scopes("files.write")
+    @instrument_tool
+    async def nc_webdav_tag_file(path: str, tag: str, ctx: Context) -> TagFileResponse:
+        """Attach a tag to a file, creating the tag if it does not exist yet.
+
+        Args:
+            path: Path to the file, relative to the user's files root.
+            tag: Tag name.
+        """
+        client = await get_client(ctx)
+        file_id = await _resolve_file_id(client, path)
+        resolved = await client.webdav.get_or_create_tag(tag)
+        await client.webdav.assign_tag_to_file(file_id, resolved["id"])
+        return TagFileResponse(path=path, tag=tag, tag_id=resolved["id"], assigned=True)
+
+    @mcp.tool(
+        title="Untag File",
+        annotations=ToolAnnotations(
+            destructiveHint=False,  # The tag itself survives
+            idempotentHint=True,  # Removing an absent tag = same end state
+            openWorldHint=True,
+        ),
+    )
+    @require_scopes("files.write")
+    @instrument_tool
+    async def nc_webdav_untag_file(
+        path: str, tag: str, ctx: Context
+    ) -> TagFileResponse:
+        """Remove a tag from a file. The tag itself keeps existing.
+
+        Args:
+            path: Path to the file, relative to the user's files root.
+            tag: Tag name.
+        """
+        client = await get_client(ctx)
+        file_id = await _resolve_file_id(client, path)
+        found = await client.webdav.get_tag_by_name(tag)
+        if found is None or found.get("id") is None:
+            raise ToolError(f"No tag named {tag!r} exists")
+        await client.webdav.remove_tag_from_file(file_id, found["id"])
+        return TagFileResponse(path=path, tag=tag, tag_id=found["id"], assigned=False)