Round-3 review polish on PR #885: - deck_move_card_to_board now captures the moved DeckCard and returns its post-move label titles in CardOperationResponse.labels, so LLM clients can confirm the cross-board label remap (the tool's headline behaviour) without a follow-up deck_get_card. The field is optional and defaults to None for the other card operations that share this response model. - Tighten test_move_card_to_board_restores_done_state to assert the returned card reflects the restored done state. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
1780 lines
65 KiB
Python
1780 lines
65 KiB
Python
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=<cardId>. 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 ``<notes_folder>/<category>/<title>.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), along
|
|
with its archived state and due date. 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.
|
|
|
|
Two caveats from Deck's move route: the card's owner is reassigned to
|
|
the user performing the move (the original owner is not preserved), and
|
|
a card marked done keeps its done state but its done timestamp is reset
|
|
to the time of the move.
|
|
|
|
target_stack_id must be a stack on target_board_id; the move is
|
|
rejected otherwise.
|
|
|
|
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
|
|
target_stack_id: The ID of the destination stack (must be on target_board_id)
|
|
order: Position within the destination stack (default 0 = top)
|
|
"""
|
|
client = await get_client(ctx)
|
|
moved = await client.deck.move_card_to_board(
|
|
source_board_id,
|
|
source_stack_id,
|
|
card_id,
|
|
target_board_id,
|
|
target_stack_id,
|
|
order,
|
|
)
|
|
# Surface the post-move labels so callers can confirm the remap without
|
|
# a follow-up get_card (label remapping is this tool's whole point).
|
|
return CardOperationResponse(
|
|
success=True,
|
|
message="Card moved to board successfully",
|
|
card_id=card_id,
|
|
stack_id=target_stack_id,
|
|
board_id=target_board_id,
|
|
labels=[label.title for label in (moved.labels or [])],
|
|
)
|
|
|
|
# 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,
|
|
)
|