import logging from typing import Literal, cast import anyio from mcp.server.fastmcp import Context, FastMCP from mcp.types import ToolAnnotations from nextcloud_mcp_server.auth import require_scopes from nextcloud_mcp_server.client import NextcloudClient from nextcloud_mcp_server.context import get_client from nextcloud_mcp_server.models.deck import ( AttachFileResponse, AttachmentOperationResponse, BoardOverviewResponse, CardCommentOperationResponse, CardCommentResponse, CardOperationResponse, CreateBoardResponse, CreateCardResponse, CreateLabelResponse, CreateStackResponse, DeckAssignedUser, DeckBoard, DeckCard, DeckCardSummary, DeckComment, DeckCommentSummary, DeckLabel, DeckStack, DeckUser, LabelOperationResponse, ListAttachmentsResponse, ListBoardsResponse, ListCardCommentsResponse, ListCardsResponse, ListLabelsResponse, ListStacksResponse, StackOperationResponse, StackOverview, ) from nextcloud_mcp_server.observability.metrics import instrument_tool logger = logging.getLogger(__name__) # Card status filter applied before serialization. "open" (the default for # list tools) hides archived and explicitly-done cards — the actionable set. CardStatus = Literal["all", "open", "done", "archived"] # Per-card detail level. "summary" (the default for list tools) projects each # card to a compact DeckCardSummary; "full" returns the heavy DeckCard. DetailLevel = Literal["summary", "full"] # Default length for the description preview carried in card summaries. _DEFAULT_DESCRIPTION_PREVIEW = 140 def _validate_positive_length( value: int | None, name: str = "description_max_length" ) -> None: """Tool-layer guard: reject zero/negative length thresholds. Reused for every positive-length knob (description truncation/preview, comment message truncation); ``name`` keeps the error message pointed at the parameter the caller actually passed. """ if value is not None and value <= 0: raise ValueError(f"{name} must be positive, got {value}") def _truncate_card_descriptions( cards: list[DeckCard], description_max_length: int | None ) -> None: """Truncate descriptions strictly longer than the limit; appends "…" so the truncated result is ``description_max_length + 1`` chars.""" if description_max_length is None: return for card in cards: if card.description and len(card.description) > description_max_length: card.description = card.description[:description_max_length] + "…" def _apply_board_filters( board: DeckBoard, *, include_acl: bool, include_users: bool, include_labels: bool, ) -> DeckBoard: """Drop board sub-fields the caller didn't request; returns the board.""" if not include_acl: board.acl = [] if not include_users: board.users = [] if not include_labels: board.labels = [] return board def _extract_uid(user: "DeckUser | DeckAssignedUser") -> str | None: """Pull the bare UID out of either assigned-user shape the API returns.""" if isinstance(user, DeckAssignedUser): return user.participant.uid if isinstance(user, DeckUser): return user.uid return None def _filter_cards( cards: list[DeckCard], *, status: CardStatus, label: str | None, assigned_to: str | None, ) -> list[DeckCard]: """Narrow a flat card list by status/label/assignee before serialization. The upstream Deck API returns every card (including archived ones) inline, so this filtering reduces the tokens the caller sees but not network bandwidth. ``open``/``done``/``archived`` partition the cards (no overlap): a card that is both done and archived is reported only under ``archived``, since archiving is the stronger "off the active board" state. """ if status == "open": cards = [c for c in cards if not c.archived and c.done is None] elif status == "done": cards = [c for c in cards if c.done is not None and not c.archived] elif status == "archived": cards = [c for c in cards if c.archived] # status == "all": no status filter if label is not None: cards = [ c for c in cards if any(lbl.title == label for lbl in (c.labels or [])) ] if assigned_to is not None: cards = [ c for c in cards if assigned_to in {_extract_uid(u) for u in (c.assignedUsers or [])} ] return cards def _summarize_card(card: DeckCard, description_preview_length: int) -> DeckCardSummary: """Project a full DeckCard down to its compact DeckCardSummary.""" description = card.description or "" has_description = bool(description.strip()) preview: str | None = None if has_description: preview = description[:description_preview_length] if len(description) > description_preview_length: preview += "…" assignees = [ uid for u in (card.assignedUsers or []) if (uid := _extract_uid(u)) is not None ] return DeckCardSummary( id=card.id, title=card.title, stackId=card.stackId, archived=card.archived, duedate=card.duedate, done=card.done, labels=[lbl.title for lbl in (card.labels or [])], assignedUsers=assignees, attachmentCount=card.attachmentCount, commentsUnread=card.commentsUnread, hasDescription=has_description, descriptionPreview=preview, ) def _shape_cards( cards: list[DeckCard], *, detail: DetailLevel, status: CardStatus, label: str | None, assigned_to: str | None, description_max_length: int | None, description_preview_length: int, ) -> list[DeckCard | DeckCardSummary]: """Filter then project a card list according to the requested detail level.""" filtered = _filter_cards(cards, status=status, label=label, assigned_to=assigned_to) if detail == "full": _truncate_card_descriptions(filtered, description_max_length) return list(filtered) return [_summarize_card(c, description_preview_length) for c in filtered] def _apply_stack_filters( stack: DeckStack, *, include_cards: bool, detail: DetailLevel, status: CardStatus, label: str | None, assigned_to: str | None, description_max_length: int | None, description_preview_length: int, ) -> DeckStack: """Apply card filtering + projection to a single stack; returns the stack.""" if not include_cards: stack.cards = None elif stack.cards: # Cards come straight from the client as DeckCard; the field type is a # union only because summary projection writes summaries back into it. stack.cards = _shape_cards( cast(list[DeckCard], stack.cards), detail=detail, status=status, label=label, assigned_to=assigned_to, description_max_length=description_max_length, description_preview_length=description_preview_length, ) return stack # Statuses whose result set can contain archived cards. The active Deck # listing endpoints (get_stacks/get_stack) exclude archived cards at the SQL # level — only the /stacks/archived endpoint returns them — so these statuses # need a second fetch and merge. See issue #842. _ARCHIVED_STATUSES: frozenset[str] = frozenset({"all", "archived"}) async def _archived_cards_by_stack( client: NextcloudClient, board_id: int ) -> dict[int, list[DeckCard]]: """Map stack_id -> archived DeckCards for a board. The active stack/card listing endpoints filter out archived cards in SQL; this hits ``/stacks/archived`` (the only endpoint that returns them) and keys the cards by their stack so the list tools can merge them back in. """ archived_stacks = await client.deck.get_archived_stacks(board_id) return { stack.id: cast(list[DeckCard], stack.cards or []) for stack in archived_stacks } def _append_archived_cards(stack: DeckStack, extra: list[DeckCard]) -> None: """Append archived cards onto a stack's existing card list, in place. Kept separate so the assignment stays correctly typed against ``DeckStack.cards`` (``list[DeckCard | DeckCardSummary] | None``). """ merged: list[DeckCard | DeckCardSummary] = list(stack.cards or []) merged.extend(extra) stack.cards = merged def _truncate_comment_message(message: str, message_max_length: int | None) -> str: """Truncate a comment strictly longer than the limit; appends "…".""" if message_max_length is not None and len(message) > message_max_length: return message[:message_max_length] + "…" return message def _shape_comments( comments: list[DeckComment], *, detail: DetailLevel, message_max_length: int | None, order: Literal["newest", "oldest"], ) -> list[DeckComment | DeckCommentSummary]: """Order, truncate and (optionally) project a page of card comments.""" ordered = sorted( comments, key=lambda c: c.creationDateTime, reverse=(order == "newest") ) if detail == "full": for comment in ordered: comment.message = _truncate_comment_message( comment.message, message_max_length ) return list(ordered) return [ DeckCommentSummary( id=c.id, actorId=c.actorId, message=_truncate_comment_message(c.message, message_max_length), creationDateTime=c.creationDateTime, ) for c in ordered ] # Card attachments — file shares ("Share from Files" picker in the Deck UI). # # Mechanism: a Deck card attachment of type="file" is just a Nextcloud share # with shareType=12 (IShare::TYPE_DECK) and shareWith=. The Deck UI # fires this exact request — see Deck app's # src/components/card/AttachmentList.vue:223-238 and lib/Service/FilesAppService.php. # The file is NOT copied; the share row binds the file's existing path to the card. _SHARE_TYPE_DECK = 12 def _resolve_note_path(notes_folder: str, category: str, title: str) -> str: """Reconstruct a note's file path from Notes API metadata. Notes are stored as ``//.md`` in the user's Files; ``<category>`` may be empty or nested (``"Foo/Bar"``). """ parts = [notes_folder.strip("/")] if category: parts.append(category.strip("/")) parts.append(f"{title}.md") return "/" + "/".join(p for p in parts if p) async def _resolve_note_attach_path(client, note_id: int) -> str: """Resolve a Notes-app note ID to its filesystem path for sharing. Hits the Notes API twice (settings + note metadata) and reconstructs the path. Encapsulates the camelCase key (``notesPath``, see ``models/notes.py:43``) so a typo there can't silently route to the default ``"Notes"`` folder for users who've configured a non-default notes location — that bug is exactly what this helper exists to make testable. """ async with anyio.create_task_group() as tg: settings_holder: list[dict] = [] note_holder: list[dict] = [] async def _get_settings() -> None: settings_holder.append(await client.notes.get_settings()) async def _get_note() -> None: note_holder.append(await client.notes.get_note(note_id)) tg.start_soon(_get_settings) tg.start_soon(_get_note) settings = settings_holder[0] note = note_holder[0] notes_folder = settings.get("notesPath") or "Notes" return _resolve_note_path( notes_folder=notes_folder, category=note.get("category") or "", title=note["title"], ) def configure_deck_tools(mcp: FastMCP): """Configure Nextcloud Deck tools and resources for the MCP server.""" # Resources @mcp.resource("nc://Deck/boards") async def deck_boards_resource(): """List all Nextcloud Deck boards""" ctx: Context = mcp.get_context() await ctx.warning("This message is deprecated, use the deck_get_board instead") client = await get_client(ctx) boards = await client.deck.get_boards() return [board.model_dump() for board in boards] @mcp.resource("nc://Deck/boards/{board_id}") async def deck_board_resource(board_id: int): """Get details of a specific Nextcloud Deck board""" ctx: Context = mcp.get_context() await ctx.warning( "This resource is deprecated, use the deck_get_board tool instead" ) client = await get_client(ctx) board = await client.deck.get_board(board_id) return board.model_dump() @mcp.resource("nc://Deck/boards/{board_id}/stacks") async def deck_stacks_resource(board_id: int): """List all stacks in a Nextcloud Deck board""" ctx: Context = mcp.get_context() await ctx.warning( "This resource is deprecated, use the deck_get_stacks tool instead" ) client = await get_client(ctx) stacks = await client.deck.get_stacks(board_id) return [stack.model_dump() for stack in stacks] @mcp.resource("nc://Deck/boards/{board_id}/stacks/{stack_id}") async def deck_stack_resource(board_id: int, stack_id: int): """Get details of a specific Nextcloud Deck stack""" ctx: Context = mcp.get_context() await ctx.warning( "This resource is deprecated, use the deck_get_stack tool instead" ) client = await get_client(ctx) stack = await client.deck.get_stack(board_id, stack_id) return stack.model_dump() @mcp.resource("nc://Deck/boards/{board_id}/stacks/{stack_id}/cards") async def deck_cards_resource(board_id: int, stack_id: int): """List all cards in a Nextcloud Deck stack""" ctx: Context = mcp.get_context() await ctx.warning( "This resource is deprecated, use the deck_get_cards tool instead" ) client = await get_client(ctx) stack = await client.deck.get_stack(board_id, stack_id) if stack.cards: return [card.model_dump() for card in stack.cards] return [] @mcp.resource("nc://Deck/boards/{board_id}/stacks/{stack_id}/cards/{card_id}") async def deck_card_resource(board_id: int, stack_id: int, card_id: int): """Get details of a specific Nextcloud Deck card""" ctx: Context = mcp.get_context() await ctx.warning( "This resource is deprecated, use the deck_get_card tool instead" ) client = await get_client(ctx) card = await client.deck.get_card(board_id, stack_id, card_id) return card.model_dump() @mcp.resource("nc://Deck/boards/{board_id}/labels") async def deck_labels_resource(board_id: int): """List all labels in a Nextcloud Deck board""" ctx: Context = mcp.get_context() await ctx.warning( "This resource is deprecated, use the deck_get_labels tool instead" ) client = await get_client(ctx) board = await client.deck.get_board(board_id) return [label.model_dump() for label in (board.labels or [])] @mcp.resource("nc://Deck/boards/{board_id}/labels/{label_id}") async def deck_label_resource(board_id: int, label_id: int): """Get details of a specific Nextcloud Deck label""" ctx: Context = mcp.get_context() await ctx.warning( "This resource is deprecated, use the deck_get_label tool instead" ) client = await get_client(ctx) label = await client.deck.get_label(board_id, label_id) return label.model_dump() # Read Tools (converted from resources) @mcp.tool( title="List Deck Boards", annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), ) @require_scopes("deck.read") @instrument_tool async def deck_get_boards(ctx: Context) -> ListBoardsResponse: """Get all Nextcloud Deck boards""" client = await get_client(ctx) boards = await client.deck.get_boards() return ListBoardsResponse(boards=boards, total=len(boards)) @mcp.tool( title="Get Deck Board", annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), ) @require_scopes("deck.read") @instrument_tool async def deck_get_board( ctx: Context, board_id: int, include_acl: bool = True, include_users: bool = True, include_labels: bool = True, ) -> DeckBoard: """Get details of a specific Nextcloud Deck board. Args: board_id: The ID of the board include_acl: Include the board's ACL entries (default True). Set False to reduce response size when ACLs are not needed. include_users: Include the board's user list (default True). Set False to reduce response size when users are not needed. include_labels: Include the board's label definitions (default True). Set False to reduce response size; labels can still be retrieved via deck_get_labels. """ client = await get_client(ctx) board = await client.deck.get_board(board_id) return _apply_board_filters( board, include_acl=include_acl, include_users=include_users, include_labels=include_labels, ) @mcp.tool( title="List Deck Stacks", annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), ) @require_scopes("deck.read") @instrument_tool async def deck_get_stacks( ctx: Context, board_id: int, include_cards: bool = True, detail: DetailLevel = "summary", status: CardStatus = "open", label: str | None = None, assigned_to: str | None = None, description_max_length: int | None = None, description_preview_length: int = _DEFAULT_DESCRIPTION_PREVIEW, ) -> ListStacksResponse: """Get all stacks in a Nextcloud Deck board. Cards are returned as compact summaries by default to keep the response small on large boards. Filtering/projection happen client-side after the API returns the full board, so they reduce the tokens the caller sees but not network bandwidth. Args: board_id: The ID of the board include_cards: Include cards inside each stack (default True). Set False for a lightweight stack listing; fetch cards separately via deck_get_cards. detail: "summary" (default) returns compact card rows; "full" returns the complete card objects (the old behavior). status: Which cards to include — "open" (default), "done", "archived", or "all". The first three partition the board (a card that is both done and archived counts as "archived"). "archived"/"all" include archived cards, which the active listing endpoint omits — this costs one extra API call. label: If set, only cards carrying a label with this exact title. assigned_to: If set, only cards assigned to this user UID. description_max_length: In detail="full", truncate each card's description to this many characters. description_preview_length: In detail="summary", length of the description preview carried on each card (default 140). """ _validate_positive_length(description_max_length) _validate_positive_length( description_preview_length, "description_preview_length" ) client = await get_client(ctx) # Fetch active stacks and (when archived cards are in scope) the # archived endpoint concurrently, then merge archived cards onto each # stack by id before filtering. The active endpoint omits archived # cards, so without this merge status="archived"/"all" would drop them. stacks_holder: list[list[DeckStack]] = [] archived_by_stack: dict[int, list[DeckCard]] = {} merge_archived = include_cards and status in _ARCHIVED_STATUSES async def _get_active() -> None: stacks_holder.append(await client.deck.get_stacks(board_id)) async def _get_archived() -> None: archived_by_stack.update(await _archived_cards_by_stack(client, board_id)) async with anyio.create_task_group() as tg: tg.start_soon(_get_active) if merge_archived: tg.start_soon(_get_archived) stacks = stacks_holder[0] if merge_archived: for stack in stacks: extra = archived_by_stack.get(stack.id) if extra: _append_archived_cards(stack, extra) stacks = [ _apply_stack_filters( stack, include_cards=include_cards, detail=detail, status=status, label=label, assigned_to=assigned_to, description_max_length=description_max_length, description_preview_length=description_preview_length, ) for stack in stacks ] return ListStacksResponse(stacks=stacks, total=len(stacks)) @mcp.tool( title="Get Deck Stack", annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), ) @require_scopes("deck.read") @instrument_tool async def deck_get_stack( ctx: Context, board_id: int, stack_id: int, include_cards: bool = True, detail: DetailLevel = "summary", status: CardStatus = "open", label: str | None = None, assigned_to: str | None = None, description_max_length: int | None = None, description_preview_length: int = _DEFAULT_DESCRIPTION_PREVIEW, ) -> DeckStack: """Get details of a specific Nextcloud Deck stack. Cards are returned as compact summaries by default; see deck_get_stacks for the shared parameter semantics. Args: board_id: The ID of the board stack_id: The ID of the stack include_cards: Include cards in the stack (default True). detail: "summary" (default) or "full". status: "open" (default), "done", "archived", or "all" (non-overlapping; a done+archived card counts as "archived"). "archived"/"all" include archived cards, which the active listing endpoint omits — this costs one extra API call. label: If set, only cards carrying a label with this exact title. assigned_to: If set, only cards assigned to this user UID. description_max_length: In detail="full", truncate descriptions. description_preview_length: In detail="summary", preview length. """ _validate_positive_length(description_max_length) _validate_positive_length( description_preview_length, "description_preview_length" ) client = await get_client(ctx) if status == "archived" and include_cards: # Archived-only: the /stacks/archived endpoint already returns the # stack (metadata + archived cards) in one call, so skip the active # fetch whose open cards would all be filtered out anyway. archived = await client.deck.get_archived_stacks(board_id) stack = next((s for s in archived if s.id == stack_id), None) if stack is None: # findAllArchived returns every stack, so this is defensive; # fall back to the active endpoint for the stack metadata. stack = await client.deck.get_stack(board_id, stack_id) else: # Active stack always needed (for metadata + open cards); fetch the # archived cards concurrently when status="all" needs both sets. stack_holder: list[DeckStack] = [] archived_by_stack: dict[int, list[DeckCard]] = {} merge_archived = include_cards and status == "all" async def _get_active() -> None: stack_holder.append(await client.deck.get_stack(board_id, stack_id)) async def _get_archived() -> None: archived_by_stack.update( await _archived_cards_by_stack(client, board_id) ) async with anyio.create_task_group() as tg: tg.start_soon(_get_active) if merge_archived: tg.start_soon(_get_archived) stack = stack_holder[0] extra = archived_by_stack.get(stack_id) if extra: _append_archived_cards(stack, extra) return _apply_stack_filters( stack, include_cards=include_cards, detail=detail, status=status, label=label, assigned_to=assigned_to, description_max_length=description_max_length, description_preview_length=description_preview_length, ) @mcp.tool( title="List Archived Deck Stacks", annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), ) @require_scopes("deck.read") @instrument_tool async def deck_get_archived_stacks( ctx: Context, board_id: int, detail: DetailLevel = "summary", label: str | None = None, assigned_to: str | None = None, description_max_length: int | None = None, description_preview_length: int = _DEFAULT_DESCRIPTION_PREVIEW, ) -> ListStacksResponse: """List archived stacks (with their archived cards) for a Nextcloud Deck board. This is the archived-only shortcut: it returns *only* archived cards in a single call. The active list tools (deck_get_cards, deck_get_stacks, deck_get_board_overview) also include archived cards when called with status="archived"/"all"; use this tool when you want archived cards exclusively and don't need the open ones. Typical use: auditing completed work archived off the active board (e.g. cards moved through a "Done" stack and then archived via deck_archive_card). The shape mirrors deck_get_stacks. Cards are always included on the returned stacks (an archived stack without its cards would have no audit value) and returned as compact summaries by default. There is no ``status`` filter — every card here is archived by definition — but ``label``/``assigned_to`` narrow the set just like the active-stack tools. Args: board_id: The ID of the board detail: "summary" (default) or "full". label: If set, only cards carrying a label with this exact title. assigned_to: If set, only cards assigned to this user UID. description_max_length: In detail="full", truncate descriptions. description_preview_length: In detail="summary", preview length. """ _validate_positive_length(description_max_length) _validate_positive_length( description_preview_length, "description_preview_length" ) client = await get_client(ctx) stacks = await client.deck.get_archived_stacks(board_id) # All cards in archived stacks are themselves archived; status="all" # keeps them (an "open"/"done" filter would drop the whole point). # label/assigned_to still apply for targeted audits. stacks = [ _apply_stack_filters( stack, include_cards=True, detail=detail, status="all", label=label, assigned_to=assigned_to, description_max_length=description_max_length, description_preview_length=description_preview_length, ) for stack in stacks ] return ListStacksResponse(stacks=stacks, total=len(stacks)) @mcp.tool( title="List Deck Cards", annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), ) @require_scopes("deck.read") @instrument_tool async def deck_get_cards( ctx: Context, board_id: int, stack_id: int, detail: DetailLevel = "summary", status: CardStatus = "open", label: str | None = None, assigned_to: str | None = None, description_max_length: int | None = None, description_preview_length: int = _DEFAULT_DESCRIPTION_PREVIEW, ) -> ListCardsResponse: """Get all cards in a Nextcloud Deck stack. Cards are returned as compact summaries by default. Filtering and projection are applied client-side after the API returns the full stack, so they reduce the tokens the caller sees but not network bandwidth — network-wise this tool is equivalent to deck_get_stack(include_cards=True). Args: board_id: The ID of the board stack_id: The ID of the stack detail: "summary" (default) returns compact card rows; "full" returns the complete card objects. status: "open" (default), "done", "archived", or "all". The first three partition the board (a done+archived card counts as "archived"). "archived"/"all" include archived cards, which the active listing endpoint omits — this costs one extra API call. label: If set, only cards carrying a label with this exact title. assigned_to: If set, only cards assigned to this user UID. description_max_length: In detail="full", truncate descriptions. description_preview_length: In detail="summary", preview length. """ _validate_positive_length(description_max_length) _validate_positive_length( description_preview_length, "description_preview_length" ) client = await get_client(ctx) # Archived cards are excluded by the active stack endpoint, so for # statuses that can include them we also fetch /stacks/archived and # merge. "open"/"done" need only the active stack (no extra call). active_cards: list[DeckCard] = [] archived_cards: list[DeckCard] = [] need_active = status != "archived" need_archived = status in _ARCHIVED_STATUSES async def _get_active() -> None: stack = await client.deck.get_stack(board_id, stack_id) active_cards.extend(cast(list[DeckCard], stack.cards or [])) async def _get_archived() -> None: by_stack = await _archived_cards_by_stack(client, board_id) archived_cards.extend(by_stack.get(stack_id, [])) async with anyio.create_task_group() as tg: if need_active: tg.start_soon(_get_active) if need_archived: tg.start_soon(_get_archived) cards = _shape_cards( active_cards + archived_cards, detail=detail, status=status, label=label, assigned_to=assigned_to, description_max_length=description_max_length, description_preview_length=description_preview_length, ) return ListCardsResponse(cards=cards, total=len(cards)) @mcp.tool( title="Get Deck Board Overview", annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), ) @require_scopes("deck.read") @instrument_tool async def deck_get_board_overview( ctx: Context, board_id: int, status: CardStatus = "open", label: str | None = None, assigned_to: str | None = None, description_preview_length: int = _DEFAULT_DESCRIPTION_PREVIEW, ) -> BoardOverviewResponse: """Get a compact, whole-board snapshot in a single call. Returns the board title, its label legend, and every stack with its cards projected to compact summary rows. Prefer it for "show me the board" / "what's in progress" style requests on large boards — it is the token-efficient way to view board *state*. It intentionally omits the board-management fields (ACL, user list, full label objects) that deck_get_board exposes; reach for deck_get_board when you need those. Args: board_id: The ID of the board status: Which cards to include — "open" (default), "done", "archived", or "all". The first three partition the board (a card that is both done and archived counts as "archived"). "archived"/"all" include archived cards, which the active listing endpoint omits — this costs one extra API call. label: If set, only cards carrying a label with this exact title. assigned_to: If set, only cards assigned to this user UID. description_preview_length: Length of the description preview carried on each card summary (default 140). """ _validate_positive_length( description_preview_length, "description_preview_length" ) client = await get_client(ctx) board_holder: list[DeckBoard] = [] stacks_holder: list[list[DeckStack]] = [] archived_by_stack: dict[int, list[DeckCard]] = {} merge_archived = status in _ARCHIVED_STATUSES async def _get_board() -> None: board_holder.append(await client.deck.get_board(board_id)) async def _get_stacks() -> None: stacks_holder.append(await client.deck.get_stacks(board_id)) async def _get_archived() -> None: archived_by_stack.update(await _archived_cards_by_stack(client, board_id)) async with anyio.create_task_group() as tg: tg.start_soon(_get_board) tg.start_soon(_get_stacks) if merge_archived: tg.start_soon(_get_archived) board = board_holder[0] stacks = stacks_holder[0] stack_overviews: list[StackOverview] = [] total_cards = 0 for stack in stacks: cards = cast(list[DeckCard], stack.cards or []) if merge_archived: cards = cards + archived_by_stack.get(stack.id, []) summaries = [ _summarize_card(c, description_preview_length) for c in _filter_cards( cards, status=status, label=label, assigned_to=assigned_to, ) ] total_cards += len(summaries) stack_overviews.append( StackOverview( id=stack.id, title=stack.title, order=stack.order, card_count=len(summaries), cards=summaries, ) ) return BoardOverviewResponse( board_id=board.id, title=board.title, labels=[lbl.title for lbl in (board.labels or [])], stacks=stack_overviews, total_cards=total_cards, ) @mcp.tool( title="Get Deck Card", annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), ) @require_scopes("deck.read") @instrument_tool async def deck_get_card( ctx: Context, board_id: int, stack_id: int, card_id: int ) -> DeckCard: """Get details of a specific Nextcloud Deck card""" client = await get_client(ctx) card = await client.deck.get_card(board_id, stack_id, card_id) return card @mcp.tool( title="List Deck Labels", annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), ) @require_scopes("deck.read") @instrument_tool async def deck_get_labels(ctx: Context, board_id: int) -> ListLabelsResponse: """Get all labels in a Nextcloud Deck board""" client = await get_client(ctx) board = await client.deck.get_board(board_id) labels = board.labels or [] return ListLabelsResponse(labels=labels, total=len(labels)) @mcp.tool( title="Get Deck Label", annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), ) @require_scopes("deck.read") @instrument_tool async def deck_get_label(ctx: Context, board_id: int, label_id: int) -> DeckLabel: """Get details of a specific Nextcloud Deck label""" client = await get_client(ctx) label = await client.deck.get_label(board_id, label_id) return label # Create/Update/Delete Tools @mcp.tool( title="Create Deck Board", annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True), ) @require_scopes("deck.write") @instrument_tool async def deck_create_board( ctx: Context, title: str, color: str ) -> CreateBoardResponse: """Create a new Nextcloud Deck board Args: title: The title of the new board color: The hexadecimal color of the new board (e.g. FF0000) """ client = await get_client(ctx) board = await client.deck.create_board(title, color) return CreateBoardResponse(id=board.id, title=board.title, color=board.color) # Stack Tools @mcp.tool( title="Create Deck Stack", annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True), ) @require_scopes("deck.write") @instrument_tool async def deck_create_stack( ctx: Context, board_id: int, title: str, order: int ) -> CreateStackResponse: """Create a new stack in a Nextcloud Deck board Args: board_id: The ID of the board title: The title of the new stack order: Order for sorting the stacks """ client = await get_client(ctx) stack = await client.deck.create_stack(board_id, title, order) return CreateStackResponse(id=stack.id, title=stack.title, order=stack.order) @mcp.tool( title="Update Deck Stack", annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True), ) @require_scopes("deck.write") @instrument_tool async def deck_update_stack( ctx: Context, board_id: int, stack_id: int, title: str | None = None, order: int | None = None, ) -> StackOperationResponse: """Update a Nextcloud Deck stack Args: board_id: The ID of the board stack_id: The ID of the stack title: New title for the stack order: New order for the stack """ client = await get_client(ctx) await client.deck.update_stack(board_id, stack_id, title, order) return StackOperationResponse( success=True, message="Stack updated successfully", stack_id=stack_id, board_id=board_id, ) @mcp.tool( title="Delete Deck Stack", annotations=ToolAnnotations( destructiveHint=True, idempotentHint=True, openWorldHint=True ), ) @require_scopes("deck.write") @instrument_tool async def deck_delete_stack( ctx: Context, board_id: int, stack_id: int ) -> StackOperationResponse: """Delete a Nextcloud Deck stack Args: board_id: The ID of the board stack_id: The ID of the stack """ client = await get_client(ctx) await client.deck.delete_stack(board_id, stack_id) return StackOperationResponse( success=True, message="Stack deleted successfully", stack_id=stack_id, board_id=board_id, ) # Card Tools @mcp.tool( title="Create Deck Card", annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True), ) @require_scopes("deck.write") @instrument_tool async def deck_create_card( ctx: Context, board_id: int, stack_id: int, title: str, type: str = "plain", order: int = 999, description: str | None = None, duedate: str | None = None, ) -> CreateCardResponse: """Create a new card in a Nextcloud Deck stack Args: board_id: The ID of the board stack_id: The ID of the stack title: The title of the new card type: Type of the card (default: plain) order: Order for sorting the cards description: Description of the card duedate: Due date of the card (ISO-8601 format) """ client = await get_client(ctx) card = await client.deck.create_card( board_id, stack_id, title, type, order, description, duedate ) return CreateCardResponse( id=card.id, title=card.title, description=card.description, stackId=card.stackId, ) @mcp.tool( title="Update Deck Card", annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True), ) @require_scopes("deck.write") @instrument_tool async def deck_update_card( ctx: Context, board_id: int, stack_id: int, card_id: int, title: str | None = None, description: str | None = None, type: str | None = None, owner: str | None = None, order: int | None = None, duedate: str | None = None, archived: bool | None = None, done: str | None = None, ) -> CardOperationResponse: """Update a Nextcloud Deck card Args: board_id: The ID of the board stack_id: The ID of the stack card_id: The ID of the card title: New title for the card description: New description for the card type: New type for the card owner: New owner for the card order: New order for the card duedate: New due date for the card (ISO-8601 format) archived: Whether the card should be archived done: Completion date for the card (ISO-8601 format) """ client = await get_client(ctx) await client.deck.update_card( board_id, stack_id, card_id, title, description, type, owner, order, duedate, archived, done, ) return CardOperationResponse( success=True, message="Card updated successfully", card_id=card_id, stack_id=stack_id, board_id=board_id, ) @mcp.tool( title="Delete Deck Card", annotations=ToolAnnotations( destructiveHint=True, idempotentHint=True, openWorldHint=True ), ) @require_scopes("deck.write") @instrument_tool async def deck_delete_card( ctx: Context, board_id: int, stack_id: int, card_id: int ) -> CardOperationResponse: """Delete a Nextcloud Deck card Args: board_id: The ID of the board stack_id: The ID of the stack card_id: The ID of the card """ client = await get_client(ctx) await client.deck.delete_card(board_id, stack_id, card_id) return CardOperationResponse( success=True, message="Card deleted successfully", card_id=card_id, stack_id=stack_id, board_id=board_id, ) @mcp.tool( title="Archive Deck Card", annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True), ) @require_scopes("deck.write") @instrument_tool async def deck_archive_card( ctx: Context, board_id: int, stack_id: int, card_id: int ) -> CardOperationResponse: """Archive a Nextcloud Deck card Args: board_id: The ID of the board stack_id: The ID of the stack card_id: The ID of the card """ client = await get_client(ctx) await client.deck.archive_card(board_id, stack_id, card_id) return CardOperationResponse( success=True, message="Card archived successfully", card_id=card_id, stack_id=stack_id, board_id=board_id, ) @mcp.tool( title="Unarchive Deck Card", annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True), ) @require_scopes("deck.write") @instrument_tool async def deck_unarchive_card( ctx: Context, board_id: int, stack_id: int, card_id: int ) -> CardOperationResponse: """Unarchive a Nextcloud Deck card Args: board_id: The ID of the board stack_id: The ID of the stack card_id: The ID of the card """ client = await get_client(ctx) await client.deck.unarchive_card(board_id, stack_id, card_id) return CardOperationResponse( success=True, message="Card unarchived successfully", card_id=card_id, stack_id=stack_id, board_id=board_id, ) @mcp.tool( title="Reorder Deck Card", annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True), ) @require_scopes("deck.write") @instrument_tool async def deck_reorder_card( ctx: Context, board_id: int, stack_id: int, card_id: int, order: int, target_stack_id: int, ) -> CardOperationResponse: """Reorder a Nextcloud Deck card within a board. Moves a card to a new position, optionally into a different stack on the SAME board. To move a card to a stack on a DIFFERENT board, use deck_move_card_to_board instead — reordering across boards is rejected because it would orphan the card's board-scoped labels. Args: board_id: The ID of the board stack_id: The ID of the current stack card_id: The ID of the card order: New position in the target stack target_stack_id: The ID of the target stack (must be on board_id) """ client = await get_client(ctx) await client.deck.reorder_card( board_id, stack_id, card_id, order, target_stack_id ) return CardOperationResponse( success=True, message="Card reordered successfully", card_id=card_id, stack_id=target_stack_id, board_id=board_id, ) @mcp.tool( title="Move Deck Card to Another Board", annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True), ) @require_scopes("deck.write") @instrument_tool async def deck_move_card_to_board( ctx: Context, source_board_id: int, source_stack_id: int, card_id: int, target_board_id: int, target_stack_id: int, order: int = 0, ) -> CardOperationResponse: """Move a Nextcloud Deck card to a stack on a different board. The card keeps its identity (same id, comments, attachments). Deck remaps the card's board-scoped labels to the destination board by title — reusing a same-titled label there, or cloning it when you have board-manage permission. Use deck_reorder_card for moves within a single board. The destination is determined by target_stack_id; target_board_id must be the board that owns it (it is used to report the resulting location). Args: source_board_id: The ID of the board the card currently lives on source_stack_id: The ID of the stack the card currently lives in card_id: The ID of the card to move target_board_id: The ID of the destination board (must own target_stack_id) target_stack_id: The ID of the destination stack order: Position within the destination stack (default 0 = top) """ client = await get_client(ctx) await client.deck.move_card_to_board( source_board_id, source_stack_id, card_id, target_stack_id, order ) return CardOperationResponse( success=True, message="Card moved to board successfully", card_id=card_id, stack_id=target_stack_id, board_id=target_board_id, ) # Label Tools @mcp.tool( title="Create Deck Label", annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True), ) @require_scopes("deck.write") @instrument_tool async def deck_create_label( ctx: Context, board_id: int, title: str, color: str ) -> CreateLabelResponse: """Create a new label in a Nextcloud Deck board Args: board_id: The ID of the board title: The title of the new label color: The color of the new label (hex format without #) """ client = await get_client(ctx) label = await client.deck.create_label(board_id, title, color) return CreateLabelResponse(id=label.id, title=label.title, color=label.color) @mcp.tool( title="Update Deck Label", annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True), ) @require_scopes("deck.write") @instrument_tool async def deck_update_label( ctx: Context, board_id: int, label_id: int, title: str | None = None, color: str | None = None, ) -> LabelOperationResponse: """Update a Nextcloud Deck label Args: board_id: The ID of the board label_id: The ID of the label title: New title for the label color: New color for the label (hex format without #) """ client = await get_client(ctx) await client.deck.update_label(board_id, label_id, title, color) return LabelOperationResponse( success=True, message="Label updated successfully", label_id=label_id, board_id=board_id, ) @mcp.tool( title="Delete Deck Label", annotations=ToolAnnotations( destructiveHint=True, idempotentHint=True, openWorldHint=True ), ) @require_scopes("deck.write") @instrument_tool async def deck_delete_label( ctx: Context, board_id: int, label_id: int ) -> LabelOperationResponse: """Delete a Nextcloud Deck label Args: board_id: The ID of the board label_id: The ID of the label """ client = await get_client(ctx) await client.deck.delete_label(board_id, label_id) return LabelOperationResponse( success=True, message="Label deleted successfully", label_id=label_id, board_id=board_id, ) # Card-Label Assignment Tools @mcp.tool( title="Assign Label to Deck Card", annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True), ) @require_scopes("deck.write") @instrument_tool async def deck_assign_label_to_card( ctx: Context, board_id: int, stack_id: int, card_id: int, label_id: int ) -> CardOperationResponse: """Assign a label to a Nextcloud Deck card Args: board_id: The ID of the board stack_id: The ID of the stack card_id: The ID of the card label_id: The ID of the label to assign """ client = await get_client(ctx) await client.deck.assign_label_to_card(board_id, stack_id, card_id, label_id) return CardOperationResponse( success=True, message="Label assigned to card successfully", card_id=card_id, stack_id=stack_id, board_id=board_id, ) @mcp.tool( title="Remove Label from Deck Card", annotations=ToolAnnotations(idempotentHint=True, openWorldHint=True), ) @require_scopes("deck.write") @instrument_tool async def deck_remove_label_from_card( ctx: Context, board_id: int, stack_id: int, card_id: int, label_id: int ) -> CardOperationResponse: """Remove a label from a Nextcloud Deck card Args: board_id: The ID of the board stack_id: The ID of the stack card_id: The ID of the card label_id: The ID of the label to remove """ client = await get_client(ctx) await client.deck.remove_label_from_card(board_id, stack_id, card_id, label_id) return CardOperationResponse( success=True, message="Label removed from card successfully", card_id=card_id, stack_id=stack_id, board_id=board_id, ) # Card-User Assignment Tools @mcp.tool( title="Assign User to Deck Card", annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True), ) @require_scopes("deck.write") @instrument_tool async def deck_assign_user_to_card( ctx: Context, board_id: int, stack_id: int, card_id: int, user_id: str ) -> CardOperationResponse: """Assign a user to a Nextcloud Deck card Args: board_id: The ID of the board stack_id: The ID of the stack card_id: The ID of the card user_id: The user ID to assign """ client = await get_client(ctx) await client.deck.assign_user_to_card(board_id, stack_id, card_id, user_id) return CardOperationResponse( success=True, message="User assigned to card successfully", card_id=card_id, stack_id=stack_id, board_id=board_id, ) @mcp.tool( title="Unassign User from Deck Card", annotations=ToolAnnotations( destructiveHint=True, idempotentHint=True, openWorldHint=True ), ) @require_scopes("deck.write") @instrument_tool async def deck_unassign_user_from_card( ctx: Context, board_id: int, stack_id: int, card_id: int, user_id: str ) -> CardOperationResponse: """Unassign a user from a Nextcloud Deck card Args: board_id: The ID of the board stack_id: The ID of the stack card_id: The ID of the card user_id: The user ID to unassign """ client = await get_client(ctx) await client.deck.unassign_user_from_card(board_id, stack_id, card_id, user_id) return CardOperationResponse( success=True, message="User unassigned from card successfully", card_id=card_id, stack_id=stack_id, board_id=board_id, ) # Card Comment Tools _COMMENT_MAX_LENGTH = 1000 def _validate_comment_message(message: str) -> None: if len(message) > _COMMENT_MAX_LENGTH: raise ValueError( f"Comment message too long: {len(message)} characters " f"(max {_COMMENT_MAX_LENGTH})" ) @mcp.tool( title="List Deck Card Comments", annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), ) @require_scopes("deck.read") @instrument_tool async def deck_get_card_comments( ctx: Context, card_id: int, limit: int = 20, offset: int = 0, detail: DetailLevel = "summary", message_max_length: int | None = None, order: Literal["newest", "oldest"] = "newest", ) -> ListCardCommentsResponse: """List comments on a Nextcloud Deck card. Returns compact comments by default (dropping mentions, actor type and display name). Ordering and truncation apply within the returned page. Args: card_id: The ID of the card limit: Maximum number of comments to return (default 20, max 200) offset: Pagination offset (default 0) detail: "summary" (default) returns compact comments; "full" returns the complete comment objects. message_max_length: If set, truncate each comment message to this many characters. order: "newest" (default) or "oldest" — sort the page by creation time. """ _validate_positive_length(message_max_length, "message_max_length") client = await get_client(ctx) comments = await client.deck.get_comments(card_id, limit=limit, offset=offset) shaped = _shape_comments( comments, detail=detail, message_max_length=message_max_length, order=order, ) return ListCardCommentsResponse(results=shaped, count=len(shaped)) @mcp.tool( title="Create Deck Card Comment", annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True), ) @require_scopes("deck.write") @instrument_tool async def deck_create_card_comment( ctx: Context, card_id: int, message: str, parent_id: int | None = None, ) -> CardCommentResponse: """Create a comment on a Nextcloud Deck card Supports @-mentions (e.g. "@alice"). Pass parent_id to reply to an existing comment on the same card. Message is limited to 1000 characters. Args: card_id: The ID of the card to comment on message: The comment text (max 1000 characters) parent_id: Optional ID of a parent comment to reply to """ _validate_comment_message(message) client = await get_client(ctx) comment = await client.deck.create_comment( card_id, message, parent_id=parent_id ) return CardCommentResponse(comment=comment) @mcp.tool( title="Update Deck Card Comment", annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True), ) @require_scopes("deck.write") @instrument_tool async def deck_update_card_comment( ctx: Context, card_id: int, comment_id: int, message: str ) -> CardCommentResponse: """Update a Nextcloud Deck card comment Only the comment's author can update it; the server returns 403 otherwise. Args: card_id: The ID of the card the comment belongs to comment_id: The ID of the comment to update message: The new comment text (max 1000 characters) """ _validate_comment_message(message) client = await get_client(ctx) comment = await client.deck.update_comment(card_id, comment_id, message) return CardCommentResponse(comment=comment) @mcp.tool( title="Delete Deck Card Comment", annotations=ToolAnnotations( destructiveHint=True, idempotentHint=True, openWorldHint=True ), ) @require_scopes("deck.write") @instrument_tool async def deck_delete_card_comment( ctx: Context, card_id: int, comment_id: int ) -> CardCommentOperationResponse: """Delete a Nextcloud Deck card comment Only the comment's author can delete it; the server returns 403 otherwise. Args: card_id: The ID of the card the comment belongs to comment_id: The ID of the comment to delete """ client = await get_client(ctx) await client.deck.delete_comment(card_id, comment_id) return CardCommentOperationResponse( success=True, message="Comment deleted successfully", card_id=card_id, comment_id=comment_id, ) @mcp.tool( title="Attach File to Deck Card", annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True), ) @require_scopes("deck.write", "files.read") @instrument_tool async def deck_attach_file( ctx: Context, card_id: int, path: str ) -> AttachFileResponse: """Attach an existing Nextcloud file to a Deck card without copying. Creates a share of ``path`` with the card (``shareType=12``, ``shareWith=<card_id>``). The file stays in its original location; clicking the attachment in the Deck UI opens the file in place. Generic over the user's Files: works for any file the caller can read — markdown notes, PDFs, images, spreadsheets, etc. Use :func:`deck_attach_note` if you have a Notes-app note ID and want the path resolved automatically. Calling twice with the same ``path`` creates two distinct shares — caller is responsible for de-duping. Args: card_id: The ID of the Deck card to attach to path: Path to the file in the user's Nextcloud Files (must start with "/", e.g. "/Documents/spec.pdf" or "/Notes/My Note.md") """ if not path.startswith("/"): raise ValueError( f"path must start with '/', got: {path!r} " "(paths are relative to the user's Files root)" ) client = await get_client(ctx) share = await client.sharing.create_share( path=path, share_with=str(card_id), share_type=_SHARE_TYPE_DECK, permissions=1, ) return AttachFileResponse( attachment_id=int(share["id"]), card_id=card_id, path=path, ) @mcp.tool( title="Attach Note to Deck Card", annotations=ToolAnnotations(idempotentHint=False, openWorldHint=True), ) @require_scopes("deck.write", "files.read", "notes.read") @instrument_tool async def deck_attach_note( ctx: Context, card_id: int, note_id: int ) -> AttachFileResponse: """Attach a Nextcloud Note to a Deck card without copying. Convenience wrapper: looks up the note's filesystem path from the Notes app settings + note metadata, then shares the file with the card (same mechanism as :func:`deck_attach_file`). The note remains editable in the Notes app; the card just shows a clickable link to it. Path is reconstructed as ``<notes_folder>/<category>/<title>.md``. If the note's title contains characters that the Notes app sanitises differently (rare), use :func:`deck_attach_file` with the explicit path instead. Args: card_id: The ID of the Deck card to attach to note_id: The ID of the Note to attach """ client = await get_client(ctx) path = await _resolve_note_attach_path(client, note_id) share = await client.sharing.create_share( path=path, share_with=str(card_id), share_type=_SHARE_TYPE_DECK, permissions=1, ) return AttachFileResponse( attachment_id=int(share["id"]), card_id=card_id, path=path, ) @mcp.tool( title="List Deck Card Attachments", annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=True), ) @require_scopes("deck.read") @instrument_tool async def deck_list_attachments( ctx: Context, board_id: int, stack_id: int, card_id: int ) -> ListAttachmentsResponse: """List attachments on a Nextcloud Deck card. Returns both shared-file attachments (``type="file"``, created via :func:`deck_attach_file` / :func:`deck_attach_note`) and uploaded binary attachments (``type="deck_file"``). Args: board_id: The ID of the board stack_id: The ID of the stack card_id: The ID of the card """ client = await get_client(ctx) attachments = await client.deck.get_attachments(board_id, stack_id, card_id) return ListAttachmentsResponse(results=attachments, count=len(attachments)) @mcp.tool( title="Delete Deck Card Attachment", annotations=ToolAnnotations( destructiveHint=True, idempotentHint=True, openWorldHint=True ), ) @require_scopes("deck.write") @instrument_tool async def deck_delete_attachment( ctx: Context, board_id: int, stack_id: int, card_id: int, attachment_id: int, ) -> AttachmentOperationResponse: """Delete an attachment from a Nextcloud Deck card. For ``type="file"`` attachments this removes the share linking the file to the card; the underlying file in the user's Files is left untouched. For ``type="deck_file"`` blobs the binary is deleted from Deck's storage. Args: board_id: The ID of the board stack_id: The ID of the stack card_id: The ID of the card attachment_id: The ID of the attachment to delete """ client = await get_client(ctx) await client.deck.delete_attachment(board_id, stack_id, card_id, attachment_id) return AttachmentOperationResponse( success=True, message="Attachment deleted successfully", card_id=card_id, attachment_id=attachment_id, )