diff --git a/changes/+mcp-expand-strips-media.bugfix b/changes/+mcp-expand-strips-media.bugfix new file mode 100644 index 0000000..8c7d0d8 --- /dev/null +++ b/changes/+mcp-expand-strips-media.bugfix @@ -0,0 +1 @@ +Fix `expand` reintroducing avatar/logo fields that `strip_media` otherwise removes from the rest of the response - an expanded block now goes through the same strip pass as everything else. diff --git a/changes/+mcp-invited-by-collapse.bugfix b/changes/+mcp-invited-by-collapse.bugfix new file mode 100644 index 0000000..063ee4a --- /dev/null +++ b/changes/+mcp-invited-by-collapse.bugfix @@ -0,0 +1 @@ +Fix `invited_by` (on membership records) escaping the `payload="compact"`/`"minimal"` collapsing every other user/project block gets, since it doesn't end in `_extra_info`. diff --git a/changes/+mcp-minimal-field-set-fixes.bugfix b/changes/+mcp-minimal-field-set-fixes.bugfix new file mode 100644 index 0000000..79e7682 --- /dev/null +++ b/changes/+mcp-minimal-field-set-fixes.bugfix @@ -0,0 +1 @@ +Fix the MCP server's `minimal` payload projection: milestones now include `project_extra_info.name`/`.slug`, user stories now include the bare `assigned_users` id list, and both user stories and issues drop the redundant per-project `status` id in favour of the cross-project-comparable `status_extra_info.name`. diff --git a/changes/+mcp-payload-reduction.feature b/changes/+mcp-payload-reduction.feature new file mode 100644 index 0000000..3de3382 --- /dev/null +++ b/changes/+mcp-payload-reduction.feature @@ -0,0 +1 @@ +Add opt-in payload-reduction parameters (`payload`, `fields`, `strip_media`, `expand`, `resolve_assigned_users`, `strict_filters`) to the MCP server's read tools, shrinking oversized responses on request while leaving default behaviour byte-for-byte unchanged. diff --git a/changes/+mcp-resolve-assigned-users-field-name-fix.bugfix b/changes/+mcp-resolve-assigned-users-field-name-fix.bugfix new file mode 100644 index 0000000..5114829 --- /dev/null +++ b/changes/+mcp-resolve-assigned-users-field-name-fix.bugfix @@ -0,0 +1 @@ +Fix `resolve_assigned_users` always returning `null` display names by reading the correct `full_name` field from Taiga membership records instead of the nonexistent `full_name_display`. diff --git a/changes/+mcp-surface-real-error-messages.bugfix b/changes/+mcp-surface-real-error-messages.bugfix new file mode 100644 index 0000000..421d115 --- /dev/null +++ b/changes/+mcp-surface-real-error-messages.bugfix @@ -0,0 +1 @@ +Fix the MCP server swallowing anticipated failure messages (a `strict_filters` violation, a stale/missing `version` on a single-item `update_*` call) into a generic "Error executing tool" with no detail. These now surface the real message, matching what `update_work_items` already returned per-row for the same failures. diff --git a/changes/+mcp-write-payload-reduction.feature b/changes/+mcp-write-payload-reduction.feature new file mode 100644 index 0000000..c263997 --- /dev/null +++ b/changes/+mcp-write-payload-reduction.feature @@ -0,0 +1 @@ +Add opt-in `return_representation` (`full`/`minimal`/`none`) to the MCP server's write tools, and a new `update_work_items` batch-write tool, shrinking write-response payloads and collapsing many-item write sequences into one call while leaving default behaviour byte-for-byte unchanged. diff --git a/docs/mcp.rst b/docs/mcp.rst index 3be3a61..e846f33 100644 --- a/docs/mcp.rst +++ b/docs/mcp.rst @@ -153,6 +153,28 @@ not in any particular project. Check it went through with: claude mcp get taiga +.. warning:: **Some MCP clients require every declared parameter to be passed + explicitly, even ones with a documented default.** Every + parameter below that shows a default (e.g. ``payload="full"``, + ``strict_filters=False``) is genuinely optional in this server's + own JSON schema and its runtime validation - confirmed by + calling the real server directly over the MCP protocol, + omitting those parameters entirely. Some MCP clients have + nonetheless been observed rejecting the call outright + (``-32602``, wording resembling Zod's ``nonoptional`` schema + check) when a parameter carrying a schema-level ``default`` is + omitted - for *every* such parameter, not just the ones this + server added recently: `include_user_stories` on + ``list_milestones`` (which predates this server's payload- + reduction work) triggers the same rejection on an affected + client. This is a property of that client's own schema + validation, not of this server, and there is no available hook + in this server's dependencies to change what gets emitted to + work around it. **If your client exhibits this, pass every + parameter explicitly on every call to an affected tool** - + there is no way to make a client-side check like this + optional from the server side. + **************** Available tools **************** @@ -219,6 +241,10 @@ Available tools Link a user story to an epic, identifying both by their per-project ref numbers (primary) or by database id (secondary, see below). +``update_work_items`` + Update a batch of user stories/tasks/issues/epics in one call - see the + tip below for the exact shape and its non-atomic semantics. + .. important:: ``get_user_story``/``get_task``/``get_issue``/``get_epic`` and their ``update_*``/``delete_*`` counterparts take a ``project`` (id or slug) and a ``ref`` - the per-project sequential number Taiga @@ -237,11 +263,23 @@ Available tools you already hold the database id (for example from a prior tool response), not a ref. -``list_milestones``, ``get_milestone``, ``create_milestone``, ``delete_milestone`` - Manage milestones (sprints). +``list_milestones``, ``get_milestone`` + List/get milestones (sprints), optionally scoped to a project (``list_milestones`` + only). Each milestone embeds its full ``user_stories`` - pass + ``include_user_stories=False`` to strip that (potentially large) field from the + result. This embedded route is the cheapest way to get every story in a sprint in + one call (pair it with ``fields=["id", "user_stories.ref", ...]`` to shrink it + further), but Taiga's embedded story serializer carries only the single primary + ``assigned_to`` - it has no ``assigned_users`` key at all, so multi-assignee data is + unavailable through this route regardless of ``fields``/``resolve_assigned_users``. + For complete multi-assignee data, fetch those stories individually with + ``list_user_stories``/``get_user_story`` instead. + +``create_milestone``, ``delete_milestone`` + Create/delete milestones (sprints). ``list_wiki_pages``, ``get_wiki_page``, ``create_wiki_page``, ``update_wiki_page`` - Manage wiki pages. + Manage wiki pages, optionally scoped to a project. .. tip:: Call ``get_project`` first when creating or updating an entity - it returns every status/priority/severity/points id valid for that @@ -252,6 +290,168 @@ Available tools through further pages, and ``order_by`` (e.g. ``-created_date``) to control ordering - for example to fetch the most recent items first. +.. tip:: Every ``list_*``/``get_*`` tool that returns a full resource - projects, + user stories, tasks, issues, epics, milestones, memberships, wiki pages - + accepts four further parameters to shrink an oversized response, all + opt-in (the default reproduces today's full response exactly): + + - ``payload``: ``"full"`` (default, untouched) / ``"compact"`` (drops + avatar/logo URLs and shrinks every ``*_extra_info`` block to its id + plus display name) / ``"minimal"`` (only the fields a sprint-planning + report actually reads - currently defined for user stories, issues, + epics and milestones; for any other resource ``"minimal"`` behaves + the same as ``"compact"`` for now). The exact field set kept by + ``payload="minimal"`` per entity (``MINIMAL_FIELDS`` in + ``taiga/mcp_server/serialize.py``): + + - ``milestone`` (``list_milestones``/``get_milestone``): ``id``, + ``name``, ``slug``, ``project``, ``project_extra_info.name``, + ``project_extra_info.slug``, ``estimated_start``, + ``estimated_finish``, ``closed``. The board's name/slug are + included alongside the bare ``project`` id specifically so a + cross-project report can display and link each board without a + second call. + - ``userstory`` (``list_user_stories``/``get_user_story``/ + ``get_user_story_by_id``): ``id``, ``ref``, ``subject``, + ``version``, ``milestone``, ``milestone_name``, + ``status_extra_info.name``, ``status_extra_info.is_closed``, + ``is_closed``, ``finish_date``, ``is_blocked``, + ``assigned_to_extra_info.full_name_display``, + ``assigned_users``, ``epics.ref``, ``assigned_users_extra_info``. + The numeric ``status`` id is deliberately omitted - it is a + per-project id, not comparable across boards, and redundant + alongside ``status_extra_info.name``. ``assigned_users`` (the + bare secondary-assignee id list) is included so ``minimal`` + never silently under-reports who is assigned; pair it with + ``resolve_assigned_users=True`` for names, not just ids. + - ``issue`` (``list_issues``/``get_issue``/``get_issue_by_id``): + same as ``userstory`` minus ``assigned_users``/ + ``assigned_users_extra_info`` (issues have no ``assigned_users``). + - ``epic`` (``list_epics``/``get_epic``/``get_epic_by_id``): + ``id``, ``ref``, ``subject``, ``status_extra_info.name``, + ``project``. + - ``fields``: an explicit list of field paths, overriding ``payload`` + entirely, e.g. ``["ref", "subject", "status_extra_info.name"]``. A + dotted path keeps only that nested key; if the value at that point is + itself a list (e.g. a story's ``epics``), the remaining path is + applied to every element, e.g. ``"epics.ref"``. + - ``strip_media``: ``True``/``False``, overriding whether avatar/logo + URLs (``photo``, ``big_photo``, ``gravatar_id``, ``logo_small_url``) + are stripped, regardless of ``payload``. These carry rotating + signed-URL signatures, so leaving them in also defeats prompt caching + between otherwise-identical calls. + - ``expand``: a list of top-level block names to add back at full + detail on top of a reduced ``payload``, e.g. + ``payload="minimal", expand=["assigned_to_extra_info"]``. + + ``list_user_stories``, ``get_user_story`` and ``get_user_story_by_id`` + additionally accept ``resolve_assigned_users=True``: ``assigned_users`` + is a bare list of user ids with no names anywhere in the default + payload, so this resolves them into + ``assigned_users_extra_info: [{"id", "full_name_display"}, ...]`` via + that story's project memberships. Off by default because, unlike the + four parameters above, it adds a request rather than removing one (one + ``list_memberships`` call per distinct project touched). + + Every ``list_*`` tool additionally accepts ``strict_filters=False`` + (default): pass ``True`` to raise immediately if ``filters`` nests one + of this server's own parameter names by mistake (e.g. + ``filters={"include_user_stories": False}``) instead of silently + returning an unfiltered response - Taiga's REST backend ignores + unknown query parameters with no error either way, which measured as + much as a 24x size regression with no signal that anything went wrong. + + **`strict_filters` is a narrow guard, not a filter validator.** It + only catches this server's own parameter names appearing inside + `filters` by mistake. It gives **no protection** against a + misspelled or unsupported Taiga filter key (e.g. `milestone__in`, + which Taiga silently ignores rather than erroring - see the note + below) - that class of mistake returns a normal-looking but wrong + result set with no error either way, `strict_filters` or not. Any + filter-based narrowing this server accepts must be re-asserted + client-side (e.g. checking the returned items' own fields match + what the filter was supposed to select) rather than trusted purely + because the call didn't raise. + +.. note:: ``filters`` is forwarded as-is to Taiga's REST endpoint, so + server-side filtering - e.g. ``list_milestones(filters={"closed": + False, "estimated_start__lte": "2026-09-20"})`` - already works + today with no MCP-side change, for any lookup Taiga's own API + filter backend supports. Which lookups that includes is a property + of the Taiga server you're talking to, not of this client. Filters + verified against a live instance: ``closed`` (milestones), + ``page_size`` (all ``list_*`` tools, caps at 100 per page regardless + of the value requested), ``estimated_start__lte``/ + ``estimated_finish__gte`` (milestones - returns exactly the boards + whose window overlaps the given range), and ``milestone`` as a + single int (user stories, issues). A comma-separated list of ids is + **not** supported the same way for either ``milestone`` (errors) or + ``milestone__in`` (silently ignored, returning an arbitrary unrelated + page of results rather than an error) - do not rely on either form; + fetch each milestone's items with its own call instead. A Python + *list* value (e.g. ``filters={"milestone": [1444, 1446]}``) is a + third failure mode, and the most dangerous of the three: it returns + a small, clean, plausible-looking result set - but only for the + *last* id in the list, with every other id's items silently + dropped. This client sends the list correctly (as repeated query + parameters, standard ``requests`` behaviour); the collapse happens + in Taiga's own REST backend, which appears to read only the last + value of a repeated parameter (standard Django ``QueryDict.get()`` + semantics) - nothing on this client's side can change that. Do not + pass a list as a filter value for any key; fetch each id with its + own call instead. Separately, + ``project=None`` (the default on item-listing tools) already returns + results across every project the caller can see - no project scope + is required for a cross-project item query. + +.. note:: Taiga's own data can disagree with itself on closure state: the + top-level ``is_closed`` field and ``status_extra_info.is_closed`` + are two independently-set signals and have been observed to + contradict each other on the same item (e.g. ``is_closed: true`` + while the status is actually "In progress" and + ``status_extra_info.is_closed: false``). This is a property of the + underlying Taiga data, not a serialization defect in this server. + ``status_extra_info.name``/``status_extra_info.is_closed`` are the + more reliable signals if the two disagree - they come directly from + the status Taiga's UI itself displays, rather than a separately + maintained flag on the item. + +.. tip:: Every ``create_*``/``update_*`` tool accepts ``return_representation`` + (``"full"`` default / ``"minimal"`` / ``"none"``), opt-in, to shrink + what a write echoes back: + + - ``"full"`` (default): the complete written resource, exactly as + before. + - ``"minimal"``: ``{"id", "ref" (only if the entity has one), + "version", ...the fields you passed in ``fields``}`` - never the + tool's own required arguments (``subject``, ``status``, etc.), + since you already know those - only ``id``/``ref``/``version`` are + genuinely new information a write produces. + - ``"none"``: ``{"ok": true, "id", "ref" (if any), "version"}`` - a + bare acknowledgement. + + On ``update_*`` tools, ``"minimal"``/``"none"`` also skip the + re-fetch these tools otherwise perform after writing - a latency + win, not just a smaller response. ``version`` is not a separate + parameter on any ``create_*``/``update_*`` tool: if Taiga needs it + for optimistic locking, pass it inside ``fields`` yourself, exactly + as today. (``set_custom_attribute_value``/``set_custom_attribute_value_by_id`` + are the exception - see the note above on their own, unrelated + ``version`` sequence.) + +.. tip:: ``update_work_items(project, updates, return_representation="full")`` + updates a batch of user stories/tasks/issues/epics in one call. + Each entry in ``updates`` is + ``{"entity_type": "user_story"|"task"|"issue"|"epic", "ref": , + "fields": {...}}``. **Not atomic** - items are processed in order, + each succeeds or fails independently, and one failure never rolls + back or blocks any other item. Returns one result row per input + item, in the same order (zip ``updates`` with the result to match + them up); a failed item's row is + ``{"status": "error", "entity_type", "ref", "error"}``. Wiki pages + aren't supported here (no per-project ``ref``) - use + ``update_wiki_page`` directly. + **************** Security notes **************** diff --git a/pyproject.toml b/pyproject.toml index e81f1d1..72ca971 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -55,7 +55,7 @@ commit = true message = "Release {new_version}" commit_args = "--no-verify" tag = false -current_version = "1.3.4.dev1" +current_version = "2.0.0b6" parse = """(?x) (?P[0-9]+) \\.(?P[0-9]+) diff --git a/taiga/__init__.py b/taiga/__init__.py index c802f9e..e0aa5b8 100644 --- a/taiga/__init__.py +++ b/taiga/__init__.py @@ -6,7 +6,7 @@ Taiga Python API library """ -__version__ = "1.3.4.dev1" +__version__ = "2.0.0.b6" __author__ = "Nephila" __license__ = "MIT" __all__ = ["TaigaAPI"] diff --git a/taiga/mcp_server/serialize.py b/taiga/mcp_server/serialize.py index d6c7ca3..c10ada2 100644 --- a/taiga/mcp_server/serialize.py +++ b/taiga/mcp_server/serialize.py @@ -10,6 +10,7 @@ from ..models.base import InstanceResource _SKIPPED_ATTRS = {"requester"} +_AVATAR_KEYS = frozenset({"photo", "big_photo", "gravatar_id", "logo_small_url"}) def to_jsonable(value: Any) -> Any: @@ -25,3 +26,188 @@ def to_jsonable(value: Any) -> Any: if isinstance(value, (list, tuple)): return [to_jsonable(item) for item in value] return str(value) + + +def strip_avatar_fields(data: Any) -> Any: + """Recursively drop avatar/logo-URL keys from a jsonable dict/list structure. + + Removes `photo`, `big_photo`, `gravatar_id` and `logo_small_url` wherever they occur, + regardless of which resource type's `*_extra_info` block they came from - these key + names are never used for anything but a rotating-signature avatar/logo URL. + """ + if isinstance(data, dict): + return {key: strip_avatar_fields(val) for key, val in data.items() if key not in _AVATAR_KEYS} + if isinstance(data, list): + return [strip_avatar_fields(item) for item in data] + return data + + +def select_fields(data: Any, paths: list[str]) -> Any: + """Project a jsonable dict/list down to only the requested (optionally dotted) paths.""" + if isinstance(data, list): + return [_select_from_item(item, paths) if isinstance(item, dict) else item for item in data] + if isinstance(data, dict): + return _select_from_item(data, paths) + return data + + +def _select_from_item(item: dict[str, Any], paths: list[str]) -> dict[str, Any]: + """Apply select_fields' path rules to a single dict: group by top-level key, + then recurse for a purely-dotted group or keep the bare value if any path + targeting that key was bare (bare beats dotted on a collision).""" + groups: dict[str, list[str]] = {} + for path in paths: + top, _, rest = path.partition(".") + groups.setdefault(top, []).append(rest) + result: dict[str, Any] = {} + for top, rests in groups.items(): + if top not in item: + continue + value = item[top] + if isinstance(value, (dict, list)) and all(rests): + result[top] = select_fields(value, rests) + else: + result[top] = value + return result + + +_LABEL_KEYS = ("name", "full_name_display") +_EXTRA_INFO_EXTRA_KEEP = {"status_extra_info": ("is_closed",)} +_ADDITIONAL_COLLAPSIBLE_KEYS = frozenset({"invited_by"}) + + +def collapse_extra_info(data: Any) -> Any: + """Shrink every `*_extra_info` block, plus `invited_by`, to its id plus whichever + descriptive label field is present. + + Keeps `name` or `full_name_display` (the only two label fields used across + owner_extra_info, assigned_to_extra_info, project_extra_info and status_extra_info in + this codebase), and additionally `is_closed` for status_extra_info specifically, so a + compact caller can still tell whether an item is closed without expanding the block. + `invited_by` (seen on membership records) is a full nested user block too, but doesn't + end in `_extra_info`, so it needs naming explicitly rather than falling out of the + suffix check. + """ + if isinstance(data, dict): + result: dict[str, Any] = {} + for key, value in data.items(): + if (key.endswith("_extra_info") or key in _ADDITIONAL_COLLAPSIBLE_KEYS) and isinstance(value, dict): + keep = ("id", *_LABEL_KEYS, *_EXTRA_INFO_EXTRA_KEEP.get(key, ())) + result[key] = {k: value[k] for k in keep if k in value} + else: + result[key] = collapse_extra_info(value) + return result + if isinstance(data, list): + return [collapse_extra_info(item) for item in data] + return data + + +MINIMAL_FIELDS: dict[str, list[str]] = { + "milestone": [ + "id", + "name", + "slug", + "project", + "project_extra_info.name", + "project_extra_info.slug", + "estimated_start", + "estimated_finish", + "closed", + ], + "userstory": [ + "id", + "ref", + "subject", + "version", + "milestone", + "milestone_name", + "status_extra_info.name", + "status_extra_info.is_closed", + "is_closed", + "finish_date", + "is_blocked", + "assigned_to_extra_info.full_name_display", + "assigned_users", + "epics.ref", + "assigned_users_extra_info", + ], + "issue": [ + "id", + "ref", + "subject", + "version", + "milestone", + "milestone_name", + "status_extra_info.name", + "status_extra_info.is_closed", + "is_closed", + "finish_date", + "is_blocked", + "assigned_to_extra_info.full_name_display", + "epics.ref", + ], + "epic": ["id", "ref", "subject", "status_extra_info.name", "project"], +} + + +def _merge_paths(projected: dict[str, Any], source: dict[str, Any], paths: list[str]) -> dict[str, Any]: + """Copy each named top-level key's full, unprojected value from `source` into `projected`.""" + result = dict(projected) + for key in paths: + if key in source: + result[key] = source[key] + return result + + +def apply_payload( + data: Any, + entity: str, + *, + payload: str = "full", + fields: list[str] | None = None, + strip_media: bool | None = None, + expand: list[str] | None = None, +) -> Any: + """Apply the payload/fields/strip_media/expand projection to a jsonable dict/list. + + `payload="full"` with `fields`/`expand` both unset and `strip_media` not explicitly + `True` is always identity - this is what guarantees the byte-identical-by-default + compatibility contract. `MINIMAL_FIELDS` only has measured entries for "milestone", + "userstory", "issue" and "epic"; for any other `entity`, `payload="minimal"` falls + back to the same projection as `payload="compact"`. + + `expand` merges each named key's full, unprojected value from `data` before the strip + pass runs, so an expanded block still loses its avatar/logo fields under the same + `strip_media` semantics as the rest of the response - `expand` widens which fields + come back, not whether media is stripped from them. + """ + if payload == "full" and fields is None and expand is None and strip_media in (None, False): + return data + + if isinstance(data, list): + return [ + apply_payload(item, entity, payload=payload, fields=fields, strip_media=strip_media, expand=expand) + for item in data + ] + + if fields is not None: + out = select_fields(data, fields) + elif payload == "minimal": + minimal_paths = MINIMAL_FIELDS.get(entity) + out = select_fields(data, minimal_paths) if minimal_paths is not None else collapse_extra_info(data) + elif payload == "compact": + out = collapse_extra_info(data) + else: + out = data + + if expand: + out = _merge_paths(out, data, expand) + + # `fields` overrides `payload` entirely, so the payload-derived default must not apply + # to an explicit field projection; an explicit `strip_media` still wins either way. + default_strip = fields is None and payload in ("compact", "minimal") + strip = strip_media if strip_media is not None else default_strip + if strip: + out = strip_avatar_fields(out) + + return out diff --git a/taiga/mcp_server/server.py b/taiga/mcp_server/server.py index 974066d..ce00b1a 100644 --- a/taiga/mcp_server/server.py +++ b/taiga/mcp_server/server.py @@ -7,9 +7,10 @@ from typing import Any, Literal from mcp.server.mcpserver import MCPServer +from mcp.server.mcpserver.exceptions import ToolError from .auth import get_client -from .serialize import to_jsonable +from .serialize import apply_payload, to_jsonable mcp = MCPServer( name="taiga", @@ -67,6 +68,41 @@ def _get_by_ref(entity_type: str, project: str | int, ref: int) -> Any: return getattr(proj, _REF_METHOD[entity_type])(ref) +def _represent( + resource: Any, + written: dict[str, Any], + return_representation: str, +) -> dict[str, Any]: + """Project a written resource down per return_representation - "minimal"/"none" only. + + Callers handle "full" themselves: that shape differs between create_* (no re-fetch) + and update_* (re-fetch required), and duplicating that branching here would obscure, + not simplify, either. + """ + base: dict[str, Any] = {"id": resource.id, "version": getattr(resource, "version", None)} + if hasattr(resource, "ref"): + base["ref"] = resource.ref + if return_representation == "minimal": + return {**written, **base} + return {"ok": True, **base} + + +def _patch(resource: Any, fields: dict[str, Any]) -> None: + """Apply `resource.patch()`, surfacing the real failure message to the caller. + + Any exception here (a stale/missing `version`, a resource that vanished after + lookup, ...) would otherwise reach the caller as a bare "Error executing tool + " - the mcp SDK replaces any exception that isn't its own `ToolError` + with a fixed generic message, deliberately, treating it as an unanticipated + crash. Re-raising as `ToolError` preserves the real message, the same detail + `update_work_items` already surfaces per-row for the same underlying failures. + """ + try: + resource.patch(list(fields.keys()), **fields) + except Exception as exc: + raise ToolError(f"{type(exc).__name__}: {exc}") from exc + + DEFAULT_PAGE_SIZE = 100 @@ -92,6 +128,34 @@ def _paginated(query: dict[str, Any]) -> dict[str, Any]: return query +_FILTER_TRAP_KEYS = frozenset( + {"payload", "fields", "strip_media", "expand", "strict_filters", "include_user_stories", "resolve_assigned_users"} +) + + +def _check_strict_filters(filters: dict[str, Any] | None, strict_filters: bool) -> None: + """Raise if `filters` nests one of this server's own parameter names by mistake. + + Taiga's REST backend silently ignores unknown query parameters, so a caller who nests + e.g. `filters={"include_user_stories": False}` instead of passing it top-level gets a + full, unfiltered response with no error - a measured 24x size regression with no + signal either way. Checked against a fixed set of this server's own parameter names, + not Taiga's full (and from this client, unknowable) set of real filterable fields, so + this can never false-positive on a genuine Taiga filter. + + Raises `ToolError`, not a bare `ValueError` - the mcp SDK discards a bare exception's + message and replaces it with a generic "Error executing tool ", which would + defeat the entire point of naming the offending key(s) here. + """ + if not strict_filters or not filters: + return + trapped = _FILTER_TRAP_KEYS & filters.keys() + if trapped: + raise ToolError( + f"filters contains parameter name(s) meant to be passed top-level, not nested: {sorted(trapped)}" + ) + + @mcp.tool() def whoami() -> dict[str, Any]: """Return the Taiga user currently authenticated.""" @@ -99,25 +163,51 @@ def whoami() -> dict[str, Any]: @mcp.tool() -def list_projects(member: int | None = None, filters: dict[str, Any] | None = None) -> list[dict[str, Any]]: +def list_projects( + member: int | None = None, + filters: dict[str, Any] | None = None, + payload: Literal["full", "compact", "minimal"] = "full", + fields: list[str] | None = None, + strip_media: bool | None = None, + expand: list[str] | None = None, + strict_filters: bool = False, +) -> list[dict[str, Any]]: """List projects visible to the authenticated user, optionally filtered by member id. Paginated: defaults to page 1 of up to 100 results. Pass `filters` with `page`/ `page_size` to page further, or `order_by` (e.g. '-created_date') to control order. + + `payload`/`fields`/`strip_media`/`expand` optionally shrink the response - see + docs/mcp.rst for full semantics. `strict_filters=True` raises instead of silently + ignoring a `filters` key that collides with this tool's own parameter names. """ + _check_strict_filters(filters, strict_filters) query = dict(filters or {}) if member is not None: query["member"] = member - return to_jsonable(get_client().projects.list(**_paginated(query))) + result = to_jsonable(get_client().projects.list(**_paginated(query))) + return apply_payload(result, "project", payload=payload, fields=fields, strip_media=strip_media, expand=expand) @mcp.tool() -def get_project(project: str | int) -> dict[str, Any]: - """Get full project detail by numeric id or slug, including statuses/priorities/severities/points.""" +def get_project( + project: str | int, + payload: Literal["full", "compact", "minimal"] = "full", + fields: list[str] | None = None, + strip_media: bool | None = None, + expand: list[str] | None = None, +) -> dict[str, Any]: + """Get full project detail by numeric id or slug, including statuses/priorities/severities/points. + + `payload`/`fields`/`strip_media`/`expand` optionally shrink the response - see + docs/mcp.rst for full semantics. + """ client = get_client() if isinstance(project, int) or str(project).isdigit(): - return to_jsonable(client.projects.get(int(project))) - return to_jsonable(client.projects.get_by_slug(str(project))) + result = to_jsonable(client.projects.get(int(project))) + else: + result = to_jsonable(client.projects.get_by_slug(str(project))) + return apply_payload(result, "project", payload=payload, fields=fields, strip_media=strip_media, expand=expand) @mcp.tool() @@ -136,16 +226,30 @@ def search(project: str | int, text: str = "") -> dict[str, Any]: @mcp.tool() -def list_memberships(project: str | int, filters: dict[str, Any] | None = None) -> list[dict[str, Any]]: +def list_memberships( + project: str | int, + filters: dict[str, Any] | None = None, + payload: Literal["full", "compact", "minimal"] = "full", + fields: list[str] | None = None, + strip_media: bool | None = None, + expand: list[str] | None = None, + strict_filters: bool = False, +) -> list[dict[str, Any]]: """List a project's memberships (username, full_name, user_email, role_name, etc.) - the pool of users assignable as owner/assigned_to/watcher on that project's items. Paginated: defaults to page 1 of up to 100 results. Pass `filters` with `page`/ `page_size` to page further. + + `payload`/`fields`/`strip_media`/`expand` optionally shrink the response - see + docs/mcp.rst for full semantics. `strict_filters=True` raises instead of silently + ignoring a `filters` key that collides with this tool's own parameter names. """ + _check_strict_filters(filters, strict_filters) proj = _resolve_project(project) query = _paginated(dict(filters or {})) - return to_jsonable(proj.list_memberships(**query)) + result = to_jsonable(proj.list_memberships(**query)) + return apply_payload(result, "membership", payload=payload, fields=fields, strip_media=strip_media, expand=expand) @mcp.tool() @@ -206,7 +310,8 @@ def get_history( @mcp.tool() def get_history_by_id( - entity_type: Literal["user_story", "task", "issue", "epic", "wiki"], id: int # noqa: A002 + entity_type: Literal["user_story", "task", "issue", "epic", "wiki"], + id: int, # noqa: A002 ) -> list[dict[str, Any]]: """Get history by database id. @@ -237,7 +342,8 @@ def get_custom_attributes_values( @mcp.tool() def get_custom_attributes_values_by_id( - entity_type: Literal["user_story", "task", "issue", "epic"], id: int # noqa: A002 + entity_type: Literal["user_story", "task", "issue", "epic"], + id: int, # noqa: A002 ) -> dict[str, Any]: """Get custom-attribute values by database id. @@ -287,63 +393,181 @@ def set_custom_attribute_value_by_id( return to_jsonable(resource.set_attribute(attribute_id, value, version=version)) +def _resolve_assigned_users( + data: dict[str, Any] | list[dict[str, Any]], +) -> dict[str, Any] | list[dict[str, Any]]: + """Attach `assigned_users_extra_info` (id + full_name_display) to each item's bare + `assigned_users` id list, resolving names via each distinct project's memberships. + + Membership records use `full_name` for the display name, not the `full_name_display` + field seen on other Taiga user blocks (`owner_extra_info`, `assigned_to_extra_info`, + ...) - confirmed against a live instance. We still key our own output as + `full_name_display`, for consistency with those other blocks; only the source field + read from the membership record differs. Fetches at most one membership page (up to + 100 members) per distinct project id actually referenced. + """ + items = data if isinstance(data, list) else [data] + if not any(item.get("assigned_users") for item in items): + return data + client = get_client() + membership_maps: dict[int, dict[int, str | None]] = {} + for item in items: + assigned_users = item.get("assigned_users") + if not assigned_users: + continue + project_id = item["project"] + if project_id not in membership_maps: + memberships = to_jsonable(client.projects.get(project_id).list_memberships(**_paginated({}))) + membership_maps[project_id] = {m["user"]: m.get("full_name") for m in memberships} + name_map = membership_maps[project_id] + item["assigned_users_extra_info"] = [ + {"id": uid, "full_name_display": name_map.get(uid)} for uid in assigned_users + ] + return data + + # --- User stories ----------------------------------------------------------------- @mcp.tool() -def list_user_stories(project: str | int | None = None, filters: dict[str, Any] | None = None) -> list[dict[str, Any]]: +def list_user_stories( + project: str | int | None = None, + filters: dict[str, Any] | None = None, + payload: Literal["full", "compact", "minimal"] = "full", + fields: list[str] | None = None, + strip_media: bool | None = None, + expand: list[str] | None = None, + resolve_assigned_users: bool = False, + strict_filters: bool = False, +) -> list[dict[str, Any]]: """List user stories, optionally scoped to a project and/or filtered by extra query params. Paginated: defaults to page 1 of up to 100 results. Pass `filters` with `page`/ `page_size` to page further, or `order_by` (e.g. '-created_date') to control order. + + `payload`/`fields`/`strip_media`/`expand` optionally shrink the response - see + docs/mcp.rst for full semantics. `resolve_assigned_users=True` attaches + `assigned_users_extra_info` (costs an extra API call per project) - see docs/mcp.rst. + `strict_filters=True` raises instead of silently ignoring a `filters` key that collides + with this tool's own parameter names. """ + _check_strict_filters(filters, strict_filters) query = dict(filters or {}) if project is not None: query["project"] = _resolve_project_id(project) - return to_jsonable(get_client().user_stories.list(**_paginated(query))) + result = to_jsonable(get_client().user_stories.list(**_paginated(query))) + if resolve_assigned_users: + result = _resolve_assigned_users(result) + return apply_payload(result, "userstory", payload=payload, fields=fields, strip_media=strip_media, expand=expand) @mcp.tool() -def get_user_story(project: str | int, ref: int) -> dict[str, Any]: - """Get a user story by its per-project ref number (the number shown in the Taiga UI/URL).""" - return to_jsonable(_get_by_ref("user_story", project, ref)) +def get_user_story( + project: str | int, + ref: int, + payload: Literal["full", "compact", "minimal"] = "full", + fields: list[str] | None = None, + strip_media: bool | None = None, + expand: list[str] | None = None, + resolve_assigned_users: bool = False, +) -> dict[str, Any]: + """Get a user story by its per-project ref number (the number shown in the Taiga UI/URL). + + `payload`/`fields`/`strip_media`/`expand` optionally shrink the response - see + docs/mcp.rst for full semantics. `resolve_assigned_users=True` attaches + `assigned_users_extra_info` (costs an extra API call per project) - see docs/mcp.rst. + """ + result = to_jsonable(_get_by_ref("user_story", project, ref)) + if resolve_assigned_users: + result = _resolve_assigned_users(result) + return apply_payload(result, "userstory", payload=payload, fields=fields, strip_media=strip_media, expand=expand) @mcp.tool() -def get_user_story_by_id(id: int) -> dict[str, Any]: # noqa: A002 +def get_user_story_by_id( + id: int, # noqa: A002 + payload: Literal["full", "compact", "minimal"] = "full", + fields: list[str] | None = None, + strip_media: bool | None = None, + expand: list[str] | None = None, + resolve_assigned_users: bool = False, +) -> dict[str, Any]: """Get a user story by its database id. Secondary lookup: prefer `get_user_story` with a project + ref. Use this only when you already hold the raw database id, not the ref shown in the Taiga UI/URL. + + `payload`/`fields`/`strip_media`/`expand` optionally shrink the response - see + docs/mcp.rst for full semantics. `resolve_assigned_users=True` attaches + `assigned_users_extra_info` (costs an extra API call per project) - see docs/mcp.rst. """ - return to_jsonable(get_client().user_stories.get(id)) + result = to_jsonable(get_client().user_stories.get(id)) + if resolve_assigned_users: + result = _resolve_assigned_users(result) + return apply_payload(result, "userstory", payload=payload, fields=fields, strip_media=strip_media, expand=expand) @mcp.tool() -def create_user_story(project: str | int, subject: str, fields: dict[str, Any] | None = None) -> dict[str, Any]: - """Create a user story. `fields` may set status, points, milestone, description, tags, etc.""" +def create_user_story( + project: str | int, + subject: str, + fields: dict[str, Any] | None = None, + return_representation: Literal["full", "minimal", "none"] = "full", +) -> dict[str, Any]: + """Create a user story. `fields` may set status, points, milestone, description, tags, etc. + + `return_representation` ("full" default / "minimal" / "none") controls how much of the + created resource comes back - "minimal" returns just id/ref/version plus whatever was + in `fields`; "none" returns only an acknowledgement. See docs/mcp.rst for the exact shape. + """ pid = _resolve_project_id(project) - return to_jsonable(get_client().user_stories.create(pid, subject, **(fields or {}))) + resource = get_client().user_stories.create(pid, subject, **(fields or {})) + if return_representation == "full": + return to_jsonable(resource) + return _represent(resource, fields or {}, return_representation) @mcp.tool() -def update_user_story(project: str | int, ref: int, fields: dict[str, Any]) -> dict[str, Any]: - """Update a user story identified by its per-project ref number. `fields` is a dict of the attributes to change.""" - # InstanceResource.patch() only refreshes `version` on the local object, not the other - # fields the server actually applied, so the result must be re-fetched, not serialized - # from the patched object itself. +def update_user_story( + project: str | int, + ref: int, + fields: dict[str, Any], + return_representation: Literal["full", "minimal", "none"] = "full", +) -> dict[str, Any]: + """Update a user story identified by its per-project ref number. `fields` is a dict of the attributes to change. + + `return_representation` ("full" default / "minimal" / "none") controls how much of the + updated resource comes back. "full" re-fetches the complete resource, exactly as before - + InstanceResource.patch() only refreshes `version` on the local object, not the other + fields the server actually applied, so "full" must re-fetch, not serialize from the + patched object itself. "minimal"/"none" skip that re-fetch entirely - they only need + id/ref/version (already on the patched-in-place resource) and, for "minimal", the + caller's own `fields` (already known). See docs/mcp.rst. + """ resource = _get_by_ref("user_story", project, ref) - resource.patch(list(fields.keys()), **fields) - return to_jsonable(get_client().user_stories.get(resource.id)) + _patch(resource, fields) + if return_representation == "full": + return to_jsonable(get_client().user_stories.get(resource.id)) + return _represent(resource, fields, return_representation) @mcp.tool() -def update_user_story_by_id(id: int, fields: dict[str, Any]) -> dict[str, Any]: # noqa: A002 - """Update a user story by its database id. Secondary lookup - prefer `update_user_story` with a project + ref.""" +def update_user_story_by_id( + id: int, # noqa: A002 + fields: dict[str, Any], + return_representation: Literal["full", "minimal", "none"] = "full", +) -> dict[str, Any]: + """Update a user story by its database id. Secondary lookup - prefer `update_user_story` with a project + ref. + + `return_representation` ("full" default / "minimal" / "none") - see `update_user_story` + for the full explanation; "minimal"/"none" skip the re-fetch this tool otherwise performs. + """ client = get_client() resource = client.user_stories.get(id) - resource.patch(list(fields.keys()), **fields) - return to_jsonable(client.user_stories.get(id)) + _patch(resource, fields) + if return_representation == "full": + return to_jsonable(client.user_stories.get(id)) + return _represent(resource, fields, return_representation) @mcp.tool() @@ -366,60 +590,129 @@ def delete_user_story_by_id(id: int) -> dict[str, str]: # noqa: A002 @mcp.tool() def list_tasks( - project: str | int | None = None, user_story: int | None = None, filters: dict[str, Any] | None = None + project: str | int | None = None, + user_story: int | None = None, + filters: dict[str, Any] | None = None, + payload: Literal["full", "compact", "minimal"] = "full", + fields: list[str] | None = None, + strip_media: bool | None = None, + expand: list[str] | None = None, + strict_filters: bool = False, ) -> list[dict[str, Any]]: """List tasks, optionally scoped to a project and/or a user story. Paginated: defaults to page 1 of up to 100 results. Pass `filters` with `page`/ `page_size` to page further, or `order_by` (e.g. '-created_date') to control order. + + `payload`/`fields`/`strip_media`/`expand` optionally shrink the response - see + docs/mcp.rst for full semantics. `strict_filters=True` raises instead of silently + ignoring a `filters` key that collides with this tool's own parameter names. """ + _check_strict_filters(filters, strict_filters) query = dict(filters or {}) if project is not None: query["project"] = _resolve_project_id(project) if user_story is not None: query["user_story"] = user_story - return to_jsonable(get_client().tasks.list(**_paginated(query))) + result = to_jsonable(get_client().tasks.list(**_paginated(query))) + return apply_payload(result, "task", payload=payload, fields=fields, strip_media=strip_media, expand=expand) @mcp.tool() -def get_task(project: str | int, ref: int) -> dict[str, Any]: - """Get a task by its per-project ref number (the number shown in the Taiga UI/URL).""" - return to_jsonable(_get_by_ref("task", project, ref)) +def get_task( + project: str | int, + ref: int, + payload: Literal["full", "compact", "minimal"] = "full", + fields: list[str] | None = None, + strip_media: bool | None = None, + expand: list[str] | None = None, +) -> dict[str, Any]: + """Get a task by its per-project ref number (the number shown in the Taiga UI/URL). + + `payload`/`fields`/`strip_media`/`expand` optionally shrink the response - see + docs/mcp.rst for full semantics. + """ + result = to_jsonable(_get_by_ref("task", project, ref)) + return apply_payload(result, "task", payload=payload, fields=fields, strip_media=strip_media, expand=expand) @mcp.tool() -def get_task_by_id(id: int) -> dict[str, Any]: # noqa: A002 +def get_task_by_id( + id: int, # noqa: A002 + payload: Literal["full", "compact", "minimal"] = "full", + fields: list[str] | None = None, + strip_media: bool | None = None, + expand: list[str] | None = None, +) -> dict[str, Any]: """Get a task by its database id. Secondary lookup: prefer `get_task` with a project + ref. Use this only when you already hold the raw database id, not the ref shown in the Taiga UI/URL. + + `payload`/`fields`/`strip_media`/`expand` optionally shrink the response - see + docs/mcp.rst for full semantics. """ - return to_jsonable(get_client().tasks.get(id)) + result = to_jsonable(get_client().tasks.get(id)) + return apply_payload(result, "task", payload=payload, fields=fields, strip_media=strip_media, expand=expand) @mcp.tool() -def create_task(project: str | int, subject: str, status: int, fields: dict[str, Any] | None = None) -> dict[str, Any]: - """Create a task. `status` is the numeric task-status id (see get_project). `fields` may set user_story, etc.""" +def create_task( + project: str | int, + subject: str, + status: int, + fields: dict[str, Any] | None = None, + return_representation: Literal["full", "minimal", "none"] = "full", +) -> dict[str, Any]: + """Create a task. `status` is the numeric task-status id (see get_project). `fields` may set user_story, etc. + + `return_representation` ("full" default / "minimal" / "none") controls how much of the + created resource comes back - see docs/mcp.rst. + """ pid = _resolve_project_id(project) - return to_jsonable(get_client().tasks.create(pid, subject, status, **(fields or {}))) + resource = get_client().tasks.create(pid, subject, status, **(fields or {})) + if return_representation == "full": + return to_jsonable(resource) + return _represent(resource, fields or {}, return_representation) @mcp.tool() -def update_task(project: str | int, ref: int, fields: dict[str, Any]) -> dict[str, Any]: - """Update a task identified by its per-project ref number. `fields` is a dict of the attributes to change.""" - # See update_user_story: patch() doesn't refresh the local object, so re-fetch it. +def update_task( + project: str | int, + ref: int, + fields: dict[str, Any], + return_representation: Literal["full", "minimal", "none"] = "full", +) -> dict[str, Any]: + """Update a task identified by its per-project ref number. `fields` is a dict of the attributes to change. + + `return_representation` ("full" default / "minimal" / "none") - see `update_user_story` + for the full explanation; "minimal"/"none" skip the re-fetch this tool otherwise performs. + """ + # See update_user_story: patch() doesn't refresh the local object, so re-fetch it for "full". resource = _get_by_ref("task", project, ref) - resource.patch(list(fields.keys()), **fields) - return to_jsonable(get_client().tasks.get(resource.id)) + _patch(resource, fields) + if return_representation == "full": + return to_jsonable(get_client().tasks.get(resource.id)) + return _represent(resource, fields, return_representation) @mcp.tool() -def update_task_by_id(id: int, fields: dict[str, Any]) -> dict[str, Any]: # noqa: A002 - """Update a task by its database id. Secondary lookup - prefer `update_task` with a project + ref.""" +def update_task_by_id( + id: int, # noqa: A002 + fields: dict[str, Any], + return_representation: Literal["full", "minimal", "none"] = "full", +) -> dict[str, Any]: + """Update a task by its database id. Secondary lookup - prefer `update_task` with a project + ref. + + `return_representation` ("full" default / "minimal" / "none") - see `update_user_story` + for the full explanation; "minimal"/"none" skip the re-fetch this tool otherwise performs. + """ client = get_client() resource = client.tasks.get(id) - resource.patch(list(fields.keys()), **fields) - return to_jsonable(client.tasks.get(id)) + _patch(resource, fields) + if return_representation == "full": + return to_jsonable(client.tasks.get(id)) + return _represent(resource, fields, return_representation) @mcp.tool() @@ -441,32 +734,68 @@ def delete_task_by_id(id: int) -> dict[str, str]: # noqa: A002 @mcp.tool() -def list_issues(project: str | int | None = None, filters: dict[str, Any] | None = None) -> list[dict[str, Any]]: +def list_issues( + project: str | int | None = None, + filters: dict[str, Any] | None = None, + payload: Literal["full", "compact", "minimal"] = "full", + fields: list[str] | None = None, + strip_media: bool | None = None, + expand: list[str] | None = None, + strict_filters: bool = False, +) -> list[dict[str, Any]]: """List issues, optionally scoped to a project. Paginated: defaults to page 1 of up to 100 results. Pass `filters` with `page`/ `page_size` to page further, or `order_by` (e.g. '-created_date') to control order. + + `payload`/`fields`/`strip_media`/`expand` optionally shrink the response - see + docs/mcp.rst for full semantics. `strict_filters=True` raises instead of silently + ignoring a `filters` key that collides with this tool's own parameter names. """ + _check_strict_filters(filters, strict_filters) query = dict(filters or {}) if project is not None: query["project"] = _resolve_project_id(project) - return to_jsonable(get_client().issues.list(**_paginated(query))) + result = to_jsonable(get_client().issues.list(**_paginated(query))) + return apply_payload(result, "issue", payload=payload, fields=fields, strip_media=strip_media, expand=expand) @mcp.tool() -def get_issue(project: str | int, ref: int) -> dict[str, Any]: - """Get an issue by its per-project ref number (the number shown in the Taiga UI/URL, e.g. .../issues/45634).""" - return to_jsonable(_get_by_ref("issue", project, ref)) +def get_issue( + project: str | int, + ref: int, + payload: Literal["full", "compact", "minimal"] = "full", + fields: list[str] | None = None, + strip_media: bool | None = None, + expand: list[str] | None = None, +) -> dict[str, Any]: + """Get an issue by its per-project ref number (the number shown in the Taiga UI/URL, e.g. .../issues/45634). + + `payload`/`fields`/`strip_media`/`expand` optionally shrink the response - see + docs/mcp.rst for full semantics. + """ + result = to_jsonable(_get_by_ref("issue", project, ref)) + return apply_payload(result, "issue", payload=payload, fields=fields, strip_media=strip_media, expand=expand) @mcp.tool() -def get_issue_by_id(id: int) -> dict[str, Any]: # noqa: A002 +def get_issue_by_id( + id: int, # noqa: A002 + payload: Literal["full", "compact", "minimal"] = "full", + fields: list[str] | None = None, + strip_media: bool | None = None, + expand: list[str] | None = None, +) -> dict[str, Any]: """Get an issue by its database id. Secondary lookup: prefer `get_issue` with a project + ref. Use this only when you already hold the raw database id, not the ref shown in the Taiga UI/URL. + + `payload`/`fields`/`strip_media`/`expand` optionally shrink the response - see + docs/mcp.rst for full semantics. """ - return to_jsonable(get_client().issues.get(id)) + result = to_jsonable(get_client().issues.get(id)) + return apply_payload(result, "issue", payload=payload, fields=fields, strip_media=strip_media, expand=expand) @mcp.tool() @@ -478,30 +807,57 @@ def create_issue( issue_type: int, severity: int, fields: dict[str, Any] | None = None, + return_representation: Literal["full", "minimal", "none"] = "full", ) -> dict[str, Any]: - """Create an issue. `priority`/`status`/`issue_type`/`severity` are numeric ids (see get_project).""" + """Create an issue. `priority`/`status`/`issue_type`/`severity` are numeric ids (see get_project). + + `return_representation` ("full" default / "minimal" / "none") controls how much of the + created resource comes back - see docs/mcp.rst. + """ pid = _resolve_project_id(project) - return to_jsonable( - get_client().issues.create(pid, subject, priority, status, issue_type, severity, **(fields or {})) - ) + resource = get_client().issues.create(pid, subject, priority, status, issue_type, severity, **(fields or {})) + if return_representation == "full": + return to_jsonable(resource) + return _represent(resource, fields or {}, return_representation) @mcp.tool() -def update_issue(project: str | int, ref: int, fields: dict[str, Any]) -> dict[str, Any]: - """Update an issue identified by its per-project ref number. `fields` is a dict of the attributes to change.""" - # See update_user_story: patch() doesn't refresh the local object, so re-fetch it. +def update_issue( + project: str | int, + ref: int, + fields: dict[str, Any], + return_representation: Literal["full", "minimal", "none"] = "full", +) -> dict[str, Any]: + """Update an issue identified by its per-project ref number. `fields` is a dict of the attributes to change. + + `return_representation` ("full" default / "minimal" / "none") - see `update_user_story` + for the full explanation; "minimal"/"none" skip the re-fetch this tool otherwise performs. + """ + # See update_user_story: patch() doesn't refresh the local object, so re-fetch it for "full". resource = _get_by_ref("issue", project, ref) - resource.patch(list(fields.keys()), **fields) - return to_jsonable(get_client().issues.get(resource.id)) + _patch(resource, fields) + if return_representation == "full": + return to_jsonable(get_client().issues.get(resource.id)) + return _represent(resource, fields, return_representation) @mcp.tool() -def update_issue_by_id(id: int, fields: dict[str, Any]) -> dict[str, Any]: # noqa: A002 - """Update an issue by its database id. Secondary lookup - prefer `update_issue` with a project + ref.""" +def update_issue_by_id( + id: int, # noqa: A002 + fields: dict[str, Any], + return_representation: Literal["full", "minimal", "none"] = "full", +) -> dict[str, Any]: + """Update an issue by its database id. Secondary lookup - prefer `update_issue` with a project + ref. + + `return_representation` ("full" default / "minimal" / "none") - see `update_user_story` + for the full explanation; "minimal"/"none" skip the re-fetch this tool otherwise performs. + """ client = get_client() resource = client.issues.get(id) - resource.patch(list(fields.keys()), **fields) - return to_jsonable(client.issues.get(id)) + _patch(resource, fields) + if return_representation == "full": + return to_jsonable(client.issues.get(id)) + return _represent(resource, fields, return_representation) @mcp.tool() @@ -523,57 +879,126 @@ def delete_issue_by_id(id: int) -> dict[str, str]: # noqa: A002 @mcp.tool() -def list_epics(project: str | int | None = None, filters: dict[str, Any] | None = None) -> list[dict[str, Any]]: +def list_epics( + project: str | int | None = None, + filters: dict[str, Any] | None = None, + payload: Literal["full", "compact", "minimal"] = "full", + fields: list[str] | None = None, + strip_media: bool | None = None, + expand: list[str] | None = None, + strict_filters: bool = False, +) -> list[dict[str, Any]]: """List epics, optionally scoped to a project. Paginated: defaults to page 1 of up to 100 results. Pass `filters` with `page`/ `page_size` to page further, or `order_by` (e.g. '-created_date') to control order. + + `payload`/`fields`/`strip_media`/`expand` optionally shrink the response - see + docs/mcp.rst for full semantics. `strict_filters=True` raises instead of silently + ignoring a `filters` key that collides with this tool's own parameter names. """ + _check_strict_filters(filters, strict_filters) query = dict(filters or {}) if project is not None: query["project"] = _resolve_project_id(project) - return to_jsonable(get_client().epics.list(**_paginated(query))) + result = to_jsonable(get_client().epics.list(**_paginated(query))) + return apply_payload(result, "epic", payload=payload, fields=fields, strip_media=strip_media, expand=expand) @mcp.tool() -def get_epic(project: str | int, ref: int) -> dict[str, Any]: - """Get an epic by its per-project ref number (the number shown in the Taiga UI/URL).""" - return to_jsonable(_get_by_ref("epic", project, ref)) +def get_epic( + project: str | int, + ref: int, + payload: Literal["full", "compact", "minimal"] = "full", + fields: list[str] | None = None, + strip_media: bool | None = None, + expand: list[str] | None = None, +) -> dict[str, Any]: + """Get an epic by its per-project ref number (the number shown in the Taiga UI/URL). + + `payload`/`fields`/`strip_media`/`expand` optionally shrink the response - see + docs/mcp.rst for full semantics. + """ + result = to_jsonable(_get_by_ref("epic", project, ref)) + return apply_payload(result, "epic", payload=payload, fields=fields, strip_media=strip_media, expand=expand) @mcp.tool() -def get_epic_by_id(id: int) -> dict[str, Any]: # noqa: A002 +def get_epic_by_id( + id: int, # noqa: A002 + payload: Literal["full", "compact", "minimal"] = "full", + fields: list[str] | None = None, + strip_media: bool | None = None, + expand: list[str] | None = None, +) -> dict[str, Any]: """Get an epic by its database id. Secondary lookup: prefer `get_epic` with a project + ref. Use this only when you already hold the raw database id, not the ref shown in the Taiga UI/URL. + + `payload`/`fields`/`strip_media`/`expand` optionally shrink the response - see + docs/mcp.rst for full semantics. """ - return to_jsonable(get_client().epics.get(id)) + result = to_jsonable(get_client().epics.get(id)) + return apply_payload(result, "epic", payload=payload, fields=fields, strip_media=strip_media, expand=expand) @mcp.tool() -def create_epic(project: str | int, subject: str, fields: dict[str, Any] | None = None) -> dict[str, Any]: - """Create an epic.""" +def create_epic( + project: str | int, + subject: str, + fields: dict[str, Any] | None = None, + return_representation: Literal["full", "minimal", "none"] = "full", +) -> dict[str, Any]: + """Create an epic. + + `return_representation` ("full" default / "minimal" / "none") controls how much of the + created resource comes back - see docs/mcp.rst. + """ pid = _resolve_project_id(project) - return to_jsonable(get_client().epics.create(pid, subject, **(fields or {}))) + resource = get_client().epics.create(pid, subject, **(fields or {})) + if return_representation == "full": + return to_jsonable(resource) + return _represent(resource, fields or {}, return_representation) @mcp.tool() -def update_epic(project: str | int, ref: int, fields: dict[str, Any]) -> dict[str, Any]: - """Update an epic identified by its per-project ref number. `fields` is a dict of the attributes to change.""" - # See update_user_story: patch() doesn't refresh the local object, so re-fetch it. +def update_epic( + project: str | int, + ref: int, + fields: dict[str, Any], + return_representation: Literal["full", "minimal", "none"] = "full", +) -> dict[str, Any]: + """Update an epic identified by its per-project ref number. `fields` is a dict of the attributes to change. + + `return_representation` ("full" default / "minimal" / "none") - see `update_user_story` + for the full explanation; "minimal"/"none" skip the re-fetch this tool otherwise performs. + """ + # See update_user_story: patch() doesn't refresh the local object, so re-fetch it for "full". resource = _get_by_ref("epic", project, ref) - resource.patch(list(fields.keys()), **fields) - return to_jsonable(get_client().epics.get(resource.id)) + _patch(resource, fields) + if return_representation == "full": + return to_jsonable(get_client().epics.get(resource.id)) + return _represent(resource, fields, return_representation) @mcp.tool() -def update_epic_by_id(id: int, fields: dict[str, Any]) -> dict[str, Any]: # noqa: A002 - """Update an epic by its database id. Secondary lookup - prefer `update_epic` with a project + ref.""" +def update_epic_by_id( + id: int, # noqa: A002 + fields: dict[str, Any], + return_representation: Literal["full", "minimal", "none"] = "full", +) -> dict[str, Any]: + """Update an epic by its database id. Secondary lookup - prefer `update_epic` with a project + ref. + + `return_representation` ("full" default / "minimal" / "none") - see `update_user_story` + for the full explanation; "minimal"/"none" skip the re-fetch this tool otherwise performs. + """ client = get_client() resource = client.epics.get(id) - resource.patch(list(fields.keys()), **fields) - return to_jsonable(client.epics.get(id)) + _patch(resource, fields) + if return_representation == "full": + return to_jsonable(client.epics.get(id)) + return _represent(resource, fields, return_representation) @mcp.tool() @@ -611,26 +1036,119 @@ def link_epic_user_story_by_id(epic_id: int, user_story_id: int) -> dict[str, An return to_jsonable(epic.add_related_user_story(user_story_id)) +@mcp.tool() +def update_work_items( + project: str | int, + updates: list[dict[str, Any]], + return_representation: Literal["full", "minimal", "none"] = "full", +) -> list[dict[str, Any]]: + """Update a batch of user stories/tasks/issues/epics in one call. + + Each entry in `updates` is `{"entity_type": "user_story"|"task"|"issue"|"epic", "ref": , + "fields": {...}}` - `ref` is the per-project ref number (as in `update_user_story` etc.), and + `fields` is the dict of attributes to change, exactly as a single-item `update_*` call would + take it. If Taiga needs a `version` for optimistic locking, put it inside that item's own + `fields`, same as today's single-item contract - this tool adds no new version handling. + + Not atomic: items are processed in order, each succeeds or fails independently, and a + failure does not roll back or block any other item. Returns one result row per input item, + in the same order, so `updates` and the result can always be zipped. A successful item's row + follows `return_representation` exactly like a single-item `update_*` call. A failed item's + row is `{"status": "error", "entity_type": ..., "ref": ..., "error": ""}`. + + Wiki pages are not supported here - they have no per-project ref number, and aren't part of + the sprint-rollover workflow this tool targets. Use `update_wiki_page` directly. + """ + results: list[dict[str, Any]] = [] + for item in updates: + try: + entity_type = item["entity_type"] + ref = item["ref"] + fields = item["fields"] + resource = _get_by_ref(entity_type, project, ref) + resource.patch(list(fields.keys()), **fields) + if return_representation == "full": + client_attr = getattr(get_client(), _ENTITY_ATTR[entity_type]) + results.append(to_jsonable(client_attr.get(resource.id))) + else: + results.append(_represent(resource, fields, return_representation)) + except Exception as exc: # A single bad item must not abort the rest of the batch. + results.append( + { + "status": "error", + "entity_type": item.get("entity_type"), + "ref": item.get("ref"), + "error": f"{type(exc).__name__}: {exc}", + } + ) + return results + + # --- Milestones (sprints) ----------------------------------------------------------------- +def _strip_user_stories( + data: dict[str, Any] | list[dict[str, Any]], +) -> dict[str, Any] | list[dict[str, Any]]: + """Drop the 'user_stories' key from one or more serialized milestone dicts.""" + for item in data if isinstance(data, list) else [data]: + item.pop("user_stories", None) + return data + + @mcp.tool() -def list_milestones(project: str | int, filters: dict[str, Any] | None = None) -> list[dict[str, Any]]: - """List milestones (sprints) of a project. +def list_milestones( + project: str | int | None = None, + filters: dict[str, Any] | None = None, + include_user_stories: bool = True, + payload: Literal["full", "compact", "minimal"] = "full", + fields: list[str] | None = None, + strip_media: bool | None = None, + expand: list[str] | None = None, + strict_filters: bool = False, +) -> list[dict[str, Any]]: + """List milestones (sprints), optionally scoped to a project. Paginated: defaults to page 1 of up to 100 results. Pass `filters` with `page`/ `page_size` to page further, or `order_by` (e.g. '-created_date') to control order. + Each milestone embeds its full `user_stories`; pass `include_user_stories=False` + to strip that (potentially large) field from every returned milestone. + + `payload`/`fields`/`strip_media`/`expand` optionally shrink the response - see + docs/mcp.rst for full semantics. `strict_filters=True` raises instead of silently + ignoring a `filters` key that collides with this tool's own parameter names. """ - pid = _resolve_project_id(project) + _check_strict_filters(filters, strict_filters) query = dict(filters or {}) - query["project"] = pid - return to_jsonable(get_client().milestones.list(**_paginated(query))) + if project is not None: + query["project"] = _resolve_project_id(project) + result = to_jsonable(get_client().milestones.list(**_paginated(query))) + if not include_user_stories: + result = _strip_user_stories(result) + return apply_payload(result, "milestone", payload=payload, fields=fields, strip_media=strip_media, expand=expand) @mcp.tool() -def get_milestone(id: int) -> dict[str, Any]: # noqa: A002 - """Get a milestone by id.""" - return to_jsonable(get_client().milestones.get(id)) +def get_milestone( + id: int, # noqa: A002 + include_user_stories: bool = True, + payload: Literal["full", "compact", "minimal"] = "full", + fields: list[str] | None = None, + strip_media: bool | None = None, + expand: list[str] | None = None, +) -> dict[str, Any]: + """Get a milestone by id. + + The milestone embeds its full `user_stories`; pass `include_user_stories=False` + to strip that (potentially large) field from the returned milestone. + + `payload`/`fields`/`strip_media`/`expand` optionally shrink the response - see + docs/mcp.rst for full semantics. + """ + result = to_jsonable(get_client().milestones.get(id)) + if not include_user_stories: + result = _strip_user_stories(result) + return apply_payload(result, "milestone", payload=payload, fields=fields, strip_media=strip_media, expand=expand) @mcp.tool() @@ -640,10 +1158,18 @@ def create_milestone( estimated_start: str, estimated_finish: str, fields: dict[str, Any] | None = None, + return_representation: Literal["full", "minimal", "none"] = "full", ) -> dict[str, Any]: - """Create a milestone. Dates are ISO strings ('YYYY-MM-DD').""" + """Create a milestone. Dates are ISO strings ('YYYY-MM-DD'). + + `return_representation` ("full" default / "minimal" / "none") controls how much of the + created resource comes back - see docs/mcp.rst. + """ pid = _resolve_project_id(project) - return to_jsonable(get_client().milestones.create(pid, name, estimated_start, estimated_finish, **(fields or {}))) + resource = get_client().milestones.create(pid, name, estimated_start, estimated_finish, **(fields or {})) + if return_representation == "full": + return to_jsonable(resource) + return _represent(resource, fields or {}, return_representation) @mcp.tool() @@ -657,38 +1183,86 @@ def delete_milestone(id: int) -> dict[str, str]: # noqa: A002 @mcp.tool() -def list_wiki_pages(project: str | int, filters: dict[str, Any] | None = None) -> list[dict[str, Any]]: - """List wiki pages of a project. +def list_wiki_pages( + project: str | int | None = None, + filters: dict[str, Any] | None = None, + payload: Literal["full", "compact", "minimal"] = "full", + fields: list[str] | None = None, + strip_media: bool | None = None, + expand: list[str] | None = None, + strict_filters: bool = False, +) -> list[dict[str, Any]]: + """List wiki pages, optionally scoped to a project. Paginated: defaults to page 1 of up to 100 results. Pass `filters` with `page`/ `page_size` to page further, or `order_by` (e.g. '-created_date') to control order. + + `payload`/`fields`/`strip_media`/`expand` optionally shrink the response - see + docs/mcp.rst for full semantics. `strict_filters=True` raises instead of silently + ignoring a `filters` key that collides with this tool's own parameter names. """ - pid = _resolve_project_id(project) + _check_strict_filters(filters, strict_filters) query = dict(filters or {}) - query["project"] = pid - return to_jsonable(get_client().wikipages.list(**_paginated(query))) + if project is not None: + query["project"] = _resolve_project_id(project) + result = to_jsonable(get_client().wikipages.list(**_paginated(query))) + return apply_payload(result, "wikipage", payload=payload, fields=fields, strip_media=strip_media, expand=expand) @mcp.tool() -def get_wiki_page(id: int) -> dict[str, Any]: # noqa: A002 - """Get a wiki page by id.""" - return to_jsonable(get_client().wikipages.get(id)) +def get_wiki_page( + id: int, # noqa: A002 + payload: Literal["full", "compact", "minimal"] = "full", + fields: list[str] | None = None, + strip_media: bool | None = None, + expand: list[str] | None = None, +) -> dict[str, Any]: + """Get a wiki page by id. + + `payload`/`fields`/`strip_media`/`expand` optionally shrink the response - see + docs/mcp.rst for full semantics. + """ + result = to_jsonable(get_client().wikipages.get(id)) + return apply_payload(result, "wikipage", payload=payload, fields=fields, strip_media=strip_media, expand=expand) @mcp.tool() def create_wiki_page( - project: str | int, slug: str, content: str, fields: dict[str, Any] | None = None + project: str | int, + slug: str, + content: str, + fields: dict[str, Any] | None = None, + return_representation: Literal["full", "minimal", "none"] = "full", ) -> dict[str, Any]: - """Create a wiki page.""" + """Create a wiki page. + + `return_representation` ("full" default / "minimal" / "none") controls how much of the + created resource comes back - see docs/mcp.rst. Wiki pages have no `ref` number, so + "minimal"/"none" here never include a `ref` key. + """ pid = _resolve_project_id(project) - return to_jsonable(get_client().wikipages.create(pid, slug, content, **(fields or {}))) + resource = get_client().wikipages.create(pid, slug, content, **(fields or {})) + if return_representation == "full": + return to_jsonable(resource) + return _represent(resource, fields or {}, return_representation) @mcp.tool() -def update_wiki_page(id: int, fields: dict[str, Any]) -> dict[str, Any]: # noqa: A002 - """Update a wiki page. `fields` is a dict of the attributes to change.""" - # See update_user_story: patch() doesn't refresh the local object, so re-fetch it. +def update_wiki_page( + id: int, # noqa: A002 + fields: dict[str, Any], + return_representation: Literal["full", "minimal", "none"] = "full", +) -> dict[str, Any]: + """Update a wiki page. `fields` is a dict of the attributes to change. + + `return_representation` ("full" default / "minimal" / "none") - see `update_user_story` + for the full explanation; "minimal"/"none" skip the re-fetch this tool otherwise performs. + Wiki pages have no `ref` number, so "minimal"/"none" here never include a `ref` key. + """ + # See update_user_story: patch() doesn't refresh the local object, so re-fetch it for "full". client = get_client() resource = client.wikipages.get(id) - resource.patch(list(fields.keys()), **fields) - return to_jsonable(client.wikipages.get(id)) + _patch(resource, fields) + if return_representation == "full": + return to_jsonable(client.wikipages.get(id)) + return _represent(resource, fields, return_representation) diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 17e767b..238b547 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -3,7 +3,9 @@ from unittest.mock import MagicMock, call, patch import pytest +from mcp.server.mcpserver.exceptions import ToolError +from taiga.exceptions import TaigaRestException from taiga.mcp_server import server _HISTORY_ENTRY = { @@ -532,6 +534,25 @@ def test_update_user_story_by_id(mock_get_client): assert result == {"id": 1, "subject": "Updated"} +@patch("taiga.mcp_server.server.get_client") +def test_update_user_story_surfaces_the_real_patch_failure_message(mock_get_client): + # A bare exception from resource.patch() (e.g. a stale `version`) would otherwise reach + # the caller as a generic "Error executing tool update_user_story" with no detail - the + # mcp SDK replaces any exception that isn't its own ToolError with a fixed message. + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_resource = MagicMock(id=1) + mock_resource.patch.side_effect = TaigaRestException( + "url", 400, '{"version": "The version doesn\'t match with the current one"}' + ) + mock_project.get_userstory_by_ref.return_value = mock_resource + mock_get_client.return_value = mock_client + + with pytest.raises(ToolError, match="version doesn't match"): + server.update_user_story(1, 45634, {"subject": "Updated", "version": 7}) + + @patch("taiga.mcp_server.server.get_client") def test_delete_user_story(mock_get_client): mock_client = MagicMock() @@ -980,8 +1001,33 @@ def test_link_epic_user_story_by_id(mock_get_client): # --- Milestones ------------------------------------------------------------------------ +def test_strip_user_stories_removes_key_from_dict(): + assert server._strip_user_stories({"id": 1, "user_stories": [{"id": 10}]}) == {"id": 1} + + +def test_strip_user_stories_removes_key_from_each_item_in_list(): + data = [{"id": 1, "user_stories": []}, {"id": 2, "user_stories": [{"id": 10}]}] + assert server._strip_user_stories(data) == [{"id": 1}, {"id": 2}] + + +def test_strip_user_stories_no_op_when_key_absent(): + assert server._strip_user_stories({"id": 1}) == {"id": 1} + + @patch("taiga.mcp_server.server.get_client") -def test_list_milestones(mock_get_client): +def test_list_milestones_no_project(mock_get_client): + mock_client = MagicMock() + mock_client.milestones.list.return_value = [{"id": 1}] + mock_get_client.return_value = mock_client + + result = server.list_milestones() + + mock_client.milestones.list.assert_called_once_with(page=1, page_size=100) + assert result == [{"id": 1}] + + +@patch("taiga.mcp_server.server.get_client") +def test_list_milestones_with_project(mock_get_client): mock_client = MagicMock() mock_client.milestones.list.return_value = [{"id": 1}] mock_get_client.return_value = mock_client @@ -992,6 +1038,28 @@ def test_list_milestones(mock_get_client): assert result == [{"id": 1}] +@patch("taiga.mcp_server.server.get_client") +def test_list_milestones_includes_user_stories_by_default(mock_get_client): + mock_client = MagicMock() + mock_client.milestones.list.return_value = [{"id": 1, "user_stories": [{"id": 10}]}] + mock_get_client.return_value = mock_client + + result = server.list_milestones() + + assert result == [{"id": 1, "user_stories": [{"id": 10}]}] + + +@patch("taiga.mcp_server.server.get_client") +def test_list_milestones_excludes_user_stories_when_disabled(mock_get_client): + mock_client = MagicMock() + mock_client.milestones.list.return_value = [{"id": 1, "user_stories": [{"id": 10}]}] + mock_get_client.return_value = mock_client + + result = server.list_milestones(include_user_stories=False) + + assert result == [{"id": 1}] + + @patch("taiga.mcp_server.server.get_client") def test_get_milestone(mock_get_client): mock_client = MagicMock() @@ -1004,6 +1072,28 @@ def test_get_milestone(mock_get_client): assert result == {"id": 1} +@patch("taiga.mcp_server.server.get_client") +def test_get_milestone_includes_user_stories_by_default(mock_get_client): + mock_client = MagicMock() + mock_client.milestones.get.return_value = {"id": 1, "user_stories": [{"id": 10}]} + mock_get_client.return_value = mock_client + + result = server.get_milestone(1) + + assert result == {"id": 1, "user_stories": [{"id": 10}]} + + +@patch("taiga.mcp_server.server.get_client") +def test_get_milestone_excludes_user_stories_when_disabled(mock_get_client): + mock_client = MagicMock() + mock_client.milestones.get.return_value = {"id": 1, "user_stories": [{"id": 10}]} + mock_get_client.return_value = mock_client + + result = server.get_milestone(1, include_user_stories=False) + + assert result == {"id": 1} + + @patch("taiga.mcp_server.server.get_client") def test_create_milestone(mock_get_client): mock_client = MagicMock() @@ -1031,7 +1121,19 @@ def test_delete_milestone(mock_get_client): @patch("taiga.mcp_server.server.get_client") -def test_list_wiki_pages(mock_get_client): +def test_list_wiki_pages_no_project(mock_get_client): + mock_client = MagicMock() + mock_client.wikipages.list.return_value = [{"id": 1}] + mock_get_client.return_value = mock_client + + result = server.list_wiki_pages() + + mock_client.wikipages.list.assert_called_once_with(page=1, page_size=100) + assert result == [{"id": 1}] + + +@patch("taiga.mcp_server.server.get_client") +def test_list_wiki_pages_with_project(mock_get_client): mock_client = MagicMock() mock_client.wikipages.list.return_value = [{"id": 1}] mock_get_client.return_value = mock_client @@ -1078,3 +1180,838 @@ def test_update_wiki_page(mock_get_client): mock_client.wikipages.get.assert_has_calls([call(1), call(1)]) mock_resource.patch.assert_called_once_with(["content"], content="Updated") assert result == {"id": 1, "content": "Updated"} + + +@patch("taiga.mcp_server.server.get_client") +def test_update_wiki_page_minimal_has_no_ref_key(mock_get_client): + mock_client = MagicMock() + mock_resource = MagicMock(spec=["id", "version", "patch"], id=6, version=2) + mock_client.wikipages.get.return_value = mock_resource + mock_get_client.return_value = mock_client + + result = server.update_wiki_page(6, {"content": "Updated"}, return_representation="minimal") + + assert result == {"id": 6, "version": 2, "content": "Updated"} + assert "ref" not in result + mock_resource.patch.assert_called_once_with(["content"], content="Updated") + + +@patch("taiga.mcp_server.server.get_client") +def test_get_project_payload_compact_strips_logo(mock_get_client): + mock_client = MagicMock() + mock_client.projects.get.return_value = { + "id": 1, + "owner_extra_info": {"id": 9, "full_name_display": "Alice", "logo_small_url": "https://x/a.png"}, + } + mock_get_client.return_value = mock_client + + result = server.get_project(1, payload="compact") + + assert result == {"id": 1, "owner_extra_info": {"id": 9, "full_name_display": "Alice"}} + + +@patch("taiga.mcp_server.server.get_client") +def test_list_memberships_fields(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_client.projects.get_by_slug.return_value = mock_project + mock_project.list_memberships.return_value = [ + {"user": 10, "full_name": "Alice", "photo": "https://example.com/a.png"} + ] + mock_get_client.return_value = mock_client + + result = server.list_memberships(1, fields=["user", "full_name"]) + + assert result == [{"user": 10, "full_name": "Alice"}] + + +@patch("taiga.mcp_server.server.get_client") +def test_list_memberships_payload_minimal_falls_back_to_compact(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_client.projects.get_by_slug.return_value = mock_project + mock_project.list_memberships.return_value = [ + {"id": 1, "owner_extra_info": {"id": 9, "full_name_display": "Alice", "photo": "x"}} + ] + mock_get_client.return_value = mock_client + + result = server.list_memberships(1, payload="minimal") + + assert result == [{"id": 1, "owner_extra_info": {"id": 9, "full_name_display": "Alice"}}] + + +@patch("taiga.mcp_server.server.get_client") +def test_list_user_stories_payload_minimal_uses_measured_field_set(mock_get_client): + mock_client = MagicMock() + mock_client.user_stories.list.return_value = [ + { + "id": 1, + "ref": 42, + "subject": "hello", + "version": 3, + "milestone": 7, + "milestone_name": "Sprint 1", + "status": 2, + "status_extra_info": {"id": 2, "name": "Done", "is_closed": True}, + "is_closed": True, + "finish_date": None, + "is_blocked": False, + "assigned_to_extra_info": {"id": 5, "full_name_display": "Bob", "photo": "x"}, + "assigned_users": [10, 11], + "epics": [{"ref": 10, "subject": "Epic A"}], + "owner_extra_info": {"id": 9, "full_name_display": "Alice", "photo": "y"}, + } + ] + mock_get_client.return_value = mock_client + + result = server.list_user_stories(payload="minimal") + + assert result == [ + { + "id": 1, + "ref": 42, + "subject": "hello", + "version": 3, + "milestone": 7, + "milestone_name": "Sprint 1", + "status_extra_info": {"name": "Done", "is_closed": True}, + "is_closed": True, + "finish_date": None, + "is_blocked": False, + "assigned_to_extra_info": {"full_name_display": "Bob"}, + "assigned_users": [10, 11], + "epics": [{"ref": 10}], + } + ] + + +@patch("taiga.mcp_server.server.get_client") +def test_get_user_story_expand_adds_full_block_back(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_project.get_userstory_by_ref.return_value = { + "id": 1, + "ref": 45634, + "subject": "hello", + "version": 1, + "milestone": None, + "milestone_name": None, + "status": 2, + "status_extra_info": {"id": 2, "name": "Done", "is_closed": True}, + "is_closed": True, + "finish_date": None, + "is_blocked": False, + "assigned_to_extra_info": {"id": 5, "full_name_display": "Bob", "photo": "x"}, + "epics": [], + } + mock_get_client.return_value = mock_client + + result = server.get_user_story(1, 45634, payload="minimal", expand=["assigned_to_extra_info"]) + + # expand adds back every field of the block ("id" here, absent from MINIMAL_FIELDS), + # but the block still goes through the same strip pass as the rest of the response - + # "photo" is gone even though it was requested to be expanded (M6). + assert result["assigned_to_extra_info"] == {"id": 5, "full_name_display": "Bob"} + + +@patch("taiga.mcp_server.server.get_client") +def test_list_tasks_payload_minimal_falls_back_to_compact(mock_get_client): + mock_client = MagicMock() + mock_client.tasks.list.return_value = [ + {"id": 1, "subject": "hello", "owner_extra_info": {"id": 9, "full_name_display": "Alice", "photo": "x"}} + ] + mock_get_client.return_value = mock_client + + result = server.list_tasks(payload="minimal") + + assert result == [{"id": 1, "subject": "hello", "owner_extra_info": {"id": 9, "full_name_display": "Alice"}}] + + +@patch("taiga.mcp_server.server.get_client") +def test_get_issue_payload_minimal_uses_measured_field_set(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_project.get_issue_by_ref.return_value = { + "id": 1, + "ref": 45634, + "subject": "hello", + "version": 1, + "milestone": None, + "milestone_name": None, + "status": 2, + "status_extra_info": {"id": 2, "name": "Done", "is_closed": True}, + "is_closed": True, + "finish_date": None, + "is_blocked": False, + "assigned_to_extra_info": {"id": 5, "full_name_display": "Bob", "photo": "x"}, + "epics": [], + } + mock_get_client.return_value = mock_client + + result = server.get_issue(1, 45634, payload="minimal") + + assert result == { + "id": 1, + "ref": 45634, + "subject": "hello", + "version": 1, + "milestone": None, + "milestone_name": None, + "status_extra_info": {"name": "Done", "is_closed": True}, + "is_closed": True, + "finish_date": None, + "is_blocked": False, + "assigned_to_extra_info": {"full_name_display": "Bob"}, + "epics": [], + } + + +@patch("taiga.mcp_server.server.get_client") +def test_list_epics_payload_minimal_uses_measured_field_set(mock_get_client): + mock_client = MagicMock() + mock_client.epics.list.return_value = [ + { + "id": 1, + "ref": 7, + "subject": "hello", + "project": 5, + "status_extra_info": {"id": 2, "name": "Done", "is_closed": True}, + "owner_extra_info": {"id": 9, "full_name_display": "Alice", "photo": "x"}, + } + ] + mock_get_client.return_value = mock_client + + result = server.list_epics(payload="minimal") + + assert result == [{"id": 1, "ref": 7, "subject": "hello", "status_extra_info": {"name": "Done"}, "project": 5}] + + +@patch("taiga.mcp_server.server.get_client") +def test_list_milestones_include_user_stories_false_and_payload_compact_combine(mock_get_client): + mock_client = MagicMock() + mock_client.milestones.list.return_value = [ + { + "id": 1, + "name": "Sprint 1", + "user_stories": [{"id": 10}], + "project_extra_info": {"id": 5, "name": "Demo", "logo_small_url": "https://x/logo.png"}, + } + ] + mock_get_client.return_value = mock_client + + result = server.list_milestones(include_user_stories=False, payload="compact") + + assert result == [{"id": 1, "name": "Sprint 1", "project_extra_info": {"id": 5, "name": "Demo"}}] + + +@patch("taiga.mcp_server.server.get_client") +def test_get_milestone_payload_minimal_uses_measured_field_set(mock_get_client): + mock_client = MagicMock() + mock_client.milestones.get.return_value = { + "id": 1, + "name": "Sprint 1", + "slug": "sprint-1", + "project": 5, + "estimated_start": "2026-09-01", + "estimated_finish": "2026-09-14", + "closed": False, + "user_stories": [{"id": 10}], + "project_extra_info": {"id": 5, "name": "Demo", "slug": "demo", "logo_small_url": "https://x/logo.png"}, + } + mock_get_client.return_value = mock_client + + result = server.get_milestone(1, payload="minimal") + + assert result == { + "id": 1, + "name": "Sprint 1", + "slug": "sprint-1", + "project": 5, + "project_extra_info": {"name": "Demo", "slug": "demo"}, + "estimated_start": "2026-09-01", + "estimated_finish": "2026-09-14", + "closed": False, + } + + +@patch("taiga.mcp_server.server.get_client") +def test_list_wiki_pages_default_unchanged(mock_get_client): + mock_client = MagicMock() + mock_client.wikipages.list.return_value = [{"id": 1, "slug": "home", "content": "hello"}] + mock_get_client.return_value = mock_client + + result = server.list_wiki_pages() + + assert result == [{"id": 1, "slug": "home", "content": "hello"}] + + +@patch("taiga.mcp_server.server.get_client") +def test_get_wiki_page_strip_media_true(mock_get_client): + mock_client = MagicMock() + mock_client.wikipages.get.return_value = { + "id": 1, + "slug": "home", + "owner_extra_info": {"id": 9, "full_name_display": "Alice", "photo": "x"}, + } + mock_get_client.return_value = mock_client + + result = server.get_wiki_page(1, strip_media=True) + + assert result == {"id": 1, "slug": "home", "owner_extra_info": {"id": 9, "full_name_display": "Alice"}} + + +# --- _resolve_assigned_users ----------------------------------------------------------- + + +@patch("taiga.mcp_server.server.get_client") +def test_resolve_assigned_users_attaches_names_from_project_memberships(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_project.list_memberships.return_value = [ + {"user": 10, "full_name": "Alice"}, + {"user": 11, "full_name": "Bob"}, + ] + mock_get_client.return_value = mock_client + + data = [{"id": 1, "project": 5, "assigned_users": [10, 11]}] + + result = server._resolve_assigned_users(data) + + assert result[0]["assigned_users_extra_info"] == [ + {"id": 10, "full_name_display": "Alice"}, + {"id": 11, "full_name_display": "Bob"}, + ] + mock_client.projects.get.assert_called_once_with(5) + mock_project.list_memberships.assert_called_once_with(page=1, page_size=100) + + +@patch("taiga.mcp_server.server.get_client") +def test_resolve_assigned_users_fetches_memberships_once_per_distinct_project(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_project.list_memberships.return_value = [{"user": 10, "full_name": "Alice"}] + mock_get_client.return_value = mock_client + + data = [ + {"id": 1, "project": 5, "assigned_users": [10]}, + {"id": 2, "project": 5, "assigned_users": [10]}, + ] + + server._resolve_assigned_users(data) + + mock_client.projects.get.assert_called_once_with(5) + + +@patch("taiga.mcp_server.server.get_client") +def test_resolve_assigned_users_unmatched_id_gets_none_name(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_project.list_memberships.return_value = [{"user": 10, "full_name": "Alice"}] + mock_get_client.return_value = mock_client + + data = {"id": 1, "project": 5, "assigned_users": [10, 99]} + + result = server._resolve_assigned_users(data) + + assert result["assigned_users_extra_info"] == [ + {"id": 10, "full_name_display": "Alice"}, + {"id": 99, "full_name_display": None}, + ] + + +def test_resolve_assigned_users_no_op_when_assigned_users_absent(): + data = {"id": 1, "project": 5} + + result = server._resolve_assigned_users(data) + + assert "assigned_users_extra_info" not in result + + +def test_resolve_assigned_users_no_op_when_assigned_users_empty(): + data = {"id": 1, "project": 5, "assigned_users": []} + + result = server._resolve_assigned_users(data) + + assert "assigned_users_extra_info" not in result + + +@patch("taiga.mcp_server.server.get_client") +def test_list_user_stories_resolve_assigned_users(mock_get_client): + mock_client = MagicMock() + mock_client.user_stories.list.return_value = [{"id": 1, "project": 5, "assigned_users": [10]}] + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_project.list_memberships.return_value = [{"user": 10, "full_name": "Alice"}] + mock_get_client.return_value = mock_client + + result = server.list_user_stories(resolve_assigned_users=True) + + assert result == [ + { + "id": 1, + "project": 5, + "assigned_users": [10], + "assigned_users_extra_info": [{"id": 10, "full_name_display": "Alice"}], + } + ] + + +@patch("taiga.mcp_server.server.get_client") +def test_list_user_stories_resolve_assigned_users_defaults_false(mock_get_client): + mock_client = MagicMock() + mock_client.user_stories.list.return_value = [{"id": 1, "project": 5, "assigned_users": [10]}] + mock_get_client.return_value = mock_client + + result = server.list_user_stories() + + assert "assigned_users_extra_info" not in result[0] + mock_client.projects.get.assert_not_called() + + +@patch("taiga.mcp_server.server.get_client") +def test_list_user_stories_resolve_assigned_users_survives_payload_minimal(mock_get_client): + mock_client = MagicMock() + mock_client.user_stories.list.return_value = [{"id": 1, "project": 5, "assigned_users": [10]}] + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_project.list_memberships.return_value = [{"user": 10, "full_name": "Alice"}] + mock_get_client.return_value = mock_client + + result = server.list_user_stories(resolve_assigned_users=True, payload="minimal") + + assert result[0]["assigned_users_extra_info"] == [{"id": 10, "full_name_display": "Alice"}] + + +# --- _check_strict_filters -------------------------------------------------------------- + + +def test_check_strict_filters_no_op_when_disabled(): + server._check_strict_filters({"include_user_stories": False}, strict_filters=False) + + +def test_check_strict_filters_no_op_when_filters_none(): + server._check_strict_filters(None, strict_filters=True) + + +def test_check_strict_filters_no_op_when_filters_clean(): + server._check_strict_filters({"closed": False, "project": 1}, strict_filters=True) + + +def test_check_strict_filters_raises_naming_the_trapped_key(): + with pytest.raises(ToolError, match="include_user_stories"): + server._check_strict_filters({"include_user_stories": False}, strict_filters=True) + + +@patch("taiga.mcp_server.server.get_client") +def test_list_milestones_strict_filters_false_ignores_trap_key_silently(mock_get_client): + mock_client = MagicMock() + mock_client.milestones.list.return_value = [{"id": 1}] + mock_get_client.return_value = mock_client + + result = server.list_milestones(filters={"include_user_stories": False}) + + assert result == [{"id": 1}] + + +def test_list_milestones_strict_filters_true_raises(): + with pytest.raises(ToolError, match="include_user_stories"): + server.list_milestones(filters={"include_user_stories": False}, strict_filters=True) + + +# --- _represent ------------------------------------------------------------------------ + + +def test_represent_minimal_includes_id_ref_version_and_written_fields(): + resource = MagicMock(id=1, version=5, ref=42) + + result = server._represent(resource, {"subject": "Updated"}, "minimal") + + assert result == {"id": 1, "version": 5, "ref": 42, "subject": "Updated"} + + +def test_represent_minimal_omits_ref_when_resource_has_none(): + resource = MagicMock(spec=["id", "version"], id=1, version=5) + + result = server._represent(resource, {"content": "Updated"}, "minimal") + + assert result == {"id": 1, "version": 5, "content": "Updated"} + + +def test_represent_minimal_version_from_resource_wins_over_stale_written_version(): + resource = MagicMock(id=1, version=8, ref=42) + + result = server._represent(resource, {"subject": "Updated", "version": 7}, "minimal") + + assert result == {"id": 1, "version": 8, "ref": 42, "subject": "Updated"} + + +def test_represent_none_includes_only_ok_id_ref_version(): + resource = MagicMock(id=1, version=5, ref=42) + + result = server._represent(resource, {"subject": "Updated"}, "none") + + assert result == {"ok": True, "id": 1, "version": 5, "ref": 42} + + +def test_represent_none_omits_ref_when_resource_has_none(): + resource = MagicMock(spec=["id", "version"], id=1, version=5) + + result = server._represent(resource, {}, "none") + + assert result == {"ok": True, "id": 1, "version": 5} + + +@patch("taiga.mcp_server.server.get_client") +def test_create_user_story_return_representation_minimal(mock_get_client): + mock_client = MagicMock() + mock_resource = MagicMock(id=1, version=1, ref=99) + mock_client.user_stories.create.return_value = mock_resource + mock_get_client.return_value = mock_client + + result = server.create_user_story(1, "New story", fields={"points": {"1": 2}}, return_representation="minimal") + + assert result == {"id": 1, "version": 1, "ref": 99, "points": {"1": 2}} + + +@patch("taiga.mcp_server.server.get_client") +def test_update_user_story_minimal_skips_refetch(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_resource = MagicMock(id=1, version=3, ref=45634) + mock_project.get_userstory_by_ref.return_value = mock_resource + mock_get_client.return_value = mock_client + + result = server.update_user_story(1, 45634, {"subject": "Updated"}, return_representation="minimal") + + assert result == {"id": 1, "version": 3, "ref": 45634, "subject": "Updated"} + mock_resource.patch.assert_called_once_with(["subject"], subject="Updated") + mock_client.user_stories.get.assert_not_called() + + +@patch("taiga.mcp_server.server.get_client") +def test_update_user_story_by_id_none(mock_get_client): + mock_client = MagicMock() + mock_resource = MagicMock(id=1, version=4, ref=45634) + mock_client.user_stories.get.return_value = mock_resource + mock_get_client.return_value = mock_client + + result = server.update_user_story_by_id(1, {"subject": "Updated"}, return_representation="none") + + assert result == {"ok": True, "id": 1, "version": 4, "ref": 45634} + mock_resource.patch.assert_called_once_with(["subject"], subject="Updated") + mock_client.user_stories.get.assert_called_once_with(1) + + +@patch("taiga.mcp_server.server.get_client") +def test_create_task_return_representation_minimal(mock_get_client): + mock_client = MagicMock() + mock_resource = MagicMock(id=2, version=1, ref=50) + mock_client.tasks.create.return_value = mock_resource + mock_get_client.return_value = mock_client + + result = server.create_task(1, "New task", 3, fields={"user_story": 10}, return_representation="minimal") + + assert result == {"id": 2, "version": 1, "ref": 50, "user_story": 10} + + +@patch("taiga.mcp_server.server.get_client") +def test_update_issue_return_representation_minimal(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_resource = MagicMock(id=3, version=2, ref=60) + mock_project.get_issue_by_ref.return_value = mock_resource + mock_get_client.return_value = mock_client + + result = server.update_issue(1, 60, {"status": 5}, return_representation="minimal") + + assert result == {"id": 3, "version": 2, "ref": 60, "status": 5} + mock_client.issues.get.assert_not_called() + + +@patch("taiga.mcp_server.server.get_client") +def test_create_epic_return_representation_none(mock_get_client): + mock_client = MagicMock() + mock_resource = MagicMock(id=4, version=1, ref=70) + mock_client.epics.create.return_value = mock_resource + mock_get_client.return_value = mock_client + + result = server.create_epic(1, "New epic", return_representation="none") + + assert result == {"ok": True, "id": 4, "version": 1, "ref": 70} + + +@patch("taiga.mcp_server.server.get_client") +def test_create_milestone_return_representation_minimal(mock_get_client): + mock_client = MagicMock() + mock_resource = MagicMock(spec=["id", "version"]) + mock_resource.id = 5 + mock_resource.version = 1 + mock_client.milestones.create.return_value = mock_resource + mock_get_client.return_value = mock_client + + result = server.create_milestone(1, "Sprint 1", "2026-09-01", "2026-09-14", return_representation="minimal") + + assert result == {"id": 5, "version": 1} + + +# --- update_work_items ------------------------------------------------------------------ + + +def test_update_work_items_empty_list(): + assert server.update_work_items(1, []) == [] + + +@patch("taiga.mcp_server.server.get_client") +def test_update_work_items_all_success(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_us = MagicMock(id=1) + mock_task = MagicMock(id=2) + mock_project.get_userstory_by_ref.return_value = mock_us + mock_project.get_task_by_ref.return_value = mock_task + mock_client.user_stories.get.return_value = {"id": 1, "subject": "US updated"} + mock_client.tasks.get.return_value = {"id": 2, "subject": "Task updated"} + mock_get_client.return_value = mock_client + + updates = [ + {"entity_type": "user_story", "ref": 10, "fields": {"subject": "US updated"}}, + {"entity_type": "task", "ref": 20, "fields": {"subject": "Task updated"}}, + ] + + result = server.update_work_items(1, updates) + + mock_us.patch.assert_called_once_with(["subject"], subject="US updated") + mock_task.patch.assert_called_once_with(["subject"], subject="Task updated") + assert result == [ + {"id": 1, "subject": "US updated"}, + {"id": 2, "subject": "Task updated"}, + ] + + +@patch("taiga.mcp_server.server.get_client") +def test_update_work_items_mixed_success_and_failure(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_us = MagicMock(id=1) + mock_project.get_userstory_by_ref.return_value = mock_us + mock_project.get_task_by_ref.side_effect = Exception("boom") + mock_client.user_stories.get.return_value = {"id": 1, "subject": "US updated"} + mock_get_client.return_value = mock_client + + updates = [ + {"entity_type": "user_story", "ref": 10, "fields": {"subject": "US updated"}}, + {"entity_type": "task", "ref": 20, "fields": {"subject": "Task updated"}}, + ] + + result = server.update_work_items(1, updates) + + assert result[0] == {"id": 1, "subject": "US updated"} + assert result[1] == { + "status": "error", + "entity_type": "task", + "ref": 20, + "error": "Exception: boom", + } + mock_us.patch.assert_called_once_with(["subject"], subject="US updated") + + +@patch("taiga.mcp_server.server.get_client") +def test_update_work_items_return_representation_minimal(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_us = MagicMock(id=1, version=5, ref=10) + mock_project.get_userstory_by_ref.return_value = mock_us + mock_get_client.return_value = mock_client + + updates = [{"entity_type": "user_story", "ref": 10, "fields": {"subject": "Updated"}}] + + result = server.update_work_items(1, updates, return_representation="minimal") + + assert result == [{"id": 1, "version": 5, "ref": 10, "subject": "Updated"}] + mock_client.user_stories.get.assert_not_called() + + +@patch("taiga.mcp_server.server.get_client") +def test_update_work_items_unsupported_entity_type_produces_error_row(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_get_client.return_value = mock_client + + updates = [{"entity_type": "wiki", "ref": 5, "fields": {"content": "x"}}] + + result = server.update_work_items(1, updates) + + assert result == [{"status": "error", "entity_type": "wiki", "ref": 5, "error": "KeyError: 'wiki'"}] + + +@patch("taiga.mcp_server.server.get_client") +def test_update_work_items_malformed_item_does_not_abort_batch(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_us = MagicMock(id=1) + mock_project.get_userstory_by_ref.return_value = mock_us + mock_client.user_stories.get.return_value = {"id": 1, "subject": "US updated"} + mock_get_client.return_value = mock_client + + updates = [ + {"entity_type": "user_story", "ref": 10, "fields": {"subject": "US updated"}}, + {"entity_type": "task", "ref": 20}, # missing "fields" - malformed + ] + + result = server.update_work_items(1, updates) + + assert len(result) == 2 + assert result[0] == {"id": 1, "subject": "US updated"} + assert result[1]["status"] == "error" + assert result[1]["entity_type"] == "task" + assert result[1]["ref"] == 20 + assert isinstance(result[1]["error"], str) and result[1]["error"] + mock_us.patch.assert_called_once_with(["subject"], subject="US updated") + + +# --- return_representation on remaining write tools ------------------------------------- + + +@patch("taiga.mcp_server.server.get_client") +def test_update_task_by_id_return_representation_minimal(mock_get_client): + mock_client = MagicMock() + mock_client.tasks.get.return_value = MagicMock(id=1, version=2, ref=9) + mock_get_client.return_value = mock_client + + result = server.update_task_by_id(1, {"status": 5}, return_representation="minimal") + + assert result == {"id": 1, "version": 2, "ref": 9, "status": 5} + mock_client.tasks.get.assert_called_once_with(1) + + +@patch("taiga.mcp_server.server.get_client") +def test_create_issue_return_representation_minimal(mock_get_client): + mock_client = MagicMock() + mock_client.issues.create.return_value = MagicMock(id=3, version=1, ref=11) + mock_get_client.return_value = mock_client + + result = server.create_issue(1, "New", 2, 3, 4, 5, fields={"description": "d"}, return_representation="minimal") + + assert result == {"description": "d", "id": 3, "version": 1, "ref": 11} + + +@patch("taiga.mcp_server.server.get_client") +def test_update_issue_by_id_return_representation_none(mock_get_client): + mock_client = MagicMock() + mock_client.issues.get.return_value = MagicMock(id=3, version=4, ref=12) + mock_get_client.return_value = mock_client + + result = server.update_issue_by_id(3, {"status": 5}, return_representation="none") + + assert result == {"ok": True, "id": 3, "version": 4, "ref": 12} + mock_client.issues.get.assert_called_once_with(3) + + +@patch("taiga.mcp_server.server.get_client") +def test_update_epic_return_representation_minimal(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_project.get_epic_by_ref.return_value = MagicMock(id=4, version=1, ref=70) + mock_get_client.return_value = mock_client + + result = server.update_epic(1, 70, {"subject": "S"}, return_representation="minimal") + + assert result == {"subject": "S", "id": 4, "version": 1, "ref": 70} + mock_client.epics.get.assert_not_called() + + +@patch("taiga.mcp_server.server.get_client") +def test_update_epic_by_id_return_representation_none(mock_get_client): + mock_client = MagicMock() + mock_client.epics.get.return_value = MagicMock(id=4, version=1, ref=70) + mock_get_client.return_value = mock_client + + result = server.update_epic_by_id(4, {"subject": "S"}, return_representation="none") + + assert result == {"ok": True, "id": 4, "version": 1, "ref": 70} + mock_client.epics.get.assert_called_once_with(4) + + +@patch("taiga.mcp_server.server.get_client") +def test_create_wiki_page_return_representation_minimal(mock_get_client): + mock_client = MagicMock() + mock_client.wikipages.create.return_value = MagicMock(id=8, version=1, spec=["id", "version"]) + mock_get_client.return_value = mock_client + + result = server.create_wiki_page(1, "home", "Welcome", fields={"x": 1}, return_representation="minimal") + + assert result == {"x": 1, "id": 8, "version": 1} + + +# --- resolve_assigned_users on get_user_story* ------------------------------------------- + + +@patch("taiga.mcp_server.server.get_client") +def test_get_user_story_resolves_assigned_users(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_project.get_userstory_by_ref.return_value = {"id": 1, "project": 5, "assigned_users": [10]} + mock_project.list_memberships.return_value = [{"user": 10, "full_name": "Alice"}] + mock_get_client.return_value = mock_client + + result = server.get_user_story(5, 1, resolve_assigned_users=True) + + assert result["assigned_users_extra_info"] == [{"id": 10, "full_name_display": "Alice"}] + + +@patch("taiga.mcp_server.server.get_client") +def test_get_user_story_by_id_resolves_assigned_users(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_client.user_stories.get.return_value = {"id": 1, "project": 5, "assigned_users": [10]} + mock_project.list_memberships.return_value = [{"user": 10, "full_name": "Alice"}] + mock_get_client.return_value = mock_client + + result = server.get_user_story_by_id(1, resolve_assigned_users=True) + + assert result["assigned_users_extra_info"] == [{"id": 10, "full_name_display": "Alice"}] + + +@patch("taiga.mcp_server.server.get_client") +def test_resolve_assigned_users_skips_items_without_assignees(mock_get_client): + mock_client = MagicMock() + mock_client.projects.get.return_value.list_memberships.return_value = [{"user": 10, "full_name": "Alice"}] + mock_get_client.return_value = mock_client + + data = [{"id": 1, "project": 5, "assigned_users": []}, {"id": 2, "project": 5, "assigned_users": [10]}] + + result = server._resolve_assigned_users(data) + + assert "assigned_users_extra_info" not in result[0] + assert result[1]["assigned_users_extra_info"] == [{"id": 10, "full_name_display": "Alice"}] + + +@patch("taiga.mcp_server.server.get_client") +def test_update_task_return_representation_none(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_project.get_task_by_ref.return_value = MagicMock(id=2, version=3, ref=45) + mock_get_client.return_value = mock_client + + result = server.update_task(1, 45, {"status": 5}, return_representation="none") + + assert result == {"ok": True, "id": 2, "version": 3, "ref": 45} + mock_client.tasks.get.assert_not_called() diff --git a/tests/test_mcp_server_serialize.py b/tests/test_mcp_server_serialize.py index fedff0d..e21eca7 100644 --- a/tests/test_mcp_server_serialize.py +++ b/tests/test_mcp_server_serialize.py @@ -3,7 +3,13 @@ import datetime from unittest.mock import MagicMock -from taiga.mcp_server.serialize import to_jsonable +from taiga.mcp_server.serialize import ( + apply_payload, + collapse_extra_info, + select_fields, + strip_avatar_fields, + to_jsonable, +) from taiga.models.base import InstanceResource @@ -73,3 +79,390 @@ def test_to_jsonable_converts_plain_date_and_datetime_values(): "due_date": "2026-01-01", "finished_at": "2026-01-01T12:30:00+00:00", } + + +def test_strip_avatar_fields_removes_known_keys_from_nested_dict(): + data = { + "id": 1, + "owner_extra_info": { + "full_name_display": "Alice", + "photo": "https://example.com/a.png", + "big_photo": "https://example.com/a-big.png", + "gravatar_id": "abc123", + }, + } + + result = strip_avatar_fields(data) + + assert result == {"id": 1, "owner_extra_info": {"full_name_display": "Alice"}} + + +def test_strip_avatar_fields_removes_logo_small_url(): + data = {"project_extra_info": {"name": "Demo", "logo_small_url": "https://example.com/logo.png"}} + + result = strip_avatar_fields(data) + + assert result == {"project_extra_info": {"name": "Demo"}} + + +def test_strip_avatar_fields_recurses_into_list_of_dicts(): + data = [{"photo": "x", "id": 1}, {"photo": "y", "id": 2}] + + result = strip_avatar_fields(data) + + assert result == [{"id": 1}, {"id": 2}] + + +def test_strip_avatar_fields_no_op_when_no_avatar_keys_present(): + assert strip_avatar_fields({"id": 1, "subject": "hello"}) == {"id": 1, "subject": "hello"} + + +def test_strip_avatar_fields_passes_through_non_dict_non_list_values(): + assert strip_avatar_fields("hello") == "hello" + assert strip_avatar_fields(42) == 42 + assert strip_avatar_fields(None) is None + + +def test_select_fields_keeps_only_requested_top_level_keys(): + data = {"id": 1, "subject": "hello", "status": 2} + + result = select_fields(data, ["id", "subject"]) + + assert result == {"id": 1, "subject": "hello"} + + +def test_select_fields_projects_dotted_path_into_nested_dict(): + data = { + "ref": 42, + "status_extra_info": {"id": 3, "name": "In progress", "color": "#000000"}, + } + + result = select_fields(data, ["ref", "status_extra_info.name"]) + + assert result == {"ref": 42, "status_extra_info": {"name": "In progress"}} + + +def test_select_fields_ignores_paths_not_present_in_item(): + data = {"id": 1} + + result = select_fields(data, ["id", "missing", "missing.nested"]) + + assert result == {"id": 1} + + +def test_select_fields_applies_per_item_when_data_is_a_list(): + data = [{"id": 1, "subject": "a"}, {"id": 2, "subject": "b"}] + + result = select_fields(data, ["id"]) + + assert result == [{"id": 1}, {"id": 2}] + + +def test_select_fields_bare_key_wins_over_dotted_path_for_same_top_level_key(): + data = {"status": {"id": 3, "name": "In progress"}} + + result = select_fields(data, ["status", "status.name"]) + + assert result == {"status": {"id": 3, "name": "In progress"}} + + +def test_select_fields_supports_nested_path_deeper_than_two_levels(): + data = {"a": {"b": {"c": 1, "d": 2}}} + + result = select_fields(data, ["a.b.c"]) + + assert result == {"a": {"b": {"c": 1}}} + + +def test_select_fields_projects_into_each_item_of_a_nested_list(): + data = {"id": 1, "epics": [{"ref": 10, "subject": "Epic A"}, {"ref": 11, "subject": "Epic B"}]} + + result = select_fields(data, ["id", "epics.ref"]) + + assert result == {"id": 1, "epics": [{"ref": 10}, {"ref": 11}]} + + +def test_select_fields_nested_list_items_missing_the_path_are_dropped_down_to_empty_dict(): + data = {"epics": [{"ref": 10}, {"subject": "no ref here"}]} + + result = select_fields(data, ["epics.ref"]) + + assert result == {"epics": [{"ref": 10}, {}]} + + +def test_collapse_extra_info_keeps_id_and_name_for_project_like_blocks(): + data = { + "id": 1, + "project_extra_info": {"id": 7, "name": "Demo", "slug": "demo", "logo_small_url": "https://x/y.png"}, + } + + result = collapse_extra_info(data) + + assert result == {"id": 1, "project_extra_info": {"id": 7, "name": "Demo"}} + + +def test_collapse_extra_info_keeps_id_and_full_name_display_for_user_like_blocks(): + data = { + "owner_extra_info": { + "id": 3, + "full_name_display": "Alice", + "username": "alice", + "photo": "https://x/a.png", + } + } + + result = collapse_extra_info(data) + + assert result == {"owner_extra_info": {"id": 3, "full_name_display": "Alice"}} + + +def test_collapse_extra_info_keeps_is_closed_for_status_extra_info_only(): + data = { + "status_extra_info": {"id": 2, "name": "Done", "is_closed": True, "color": "#00ff00"}, + "project_extra_info": {"id": 7, "name": "Demo", "is_closed": True}, + } + + result = collapse_extra_info(data) + + assert result == { + "status_extra_info": {"id": 2, "name": "Done", "is_closed": True}, + "project_extra_info": {"id": 7, "name": "Demo"}, + } + + +def test_collapse_extra_info_recurses_into_list_of_dicts(): + data = [ + {"owner_extra_info": {"id": 1, "full_name_display": "Alice", "photo": "x"}}, + {"owner_extra_info": {"id": 2, "full_name_display": "Bob", "photo": "y"}}, + ] + + result = collapse_extra_info(data) + + assert result == [ + {"owner_extra_info": {"id": 1, "full_name_display": "Alice"}}, + {"owner_extra_info": {"id": 2, "full_name_display": "Bob"}}, + ] + + +def test_collapse_extra_info_leaves_non_extra_info_keys_untouched(): + data = {"id": 1, "subject": "hello", "user_stories": [{"id": 10}]} + + assert collapse_extra_info(data) == data + + +def test_collapse_extra_info_collapses_invited_by_despite_the_name_mismatch(): + data = {"invited_by": {"id": 5, "username": "yakky", "full_name_display": "Iacopo Spalletti", "is_active": True}} + + result = collapse_extra_info(data) + + assert result == {"invited_by": {"id": 5, "full_name_display": "Iacopo Spalletti"}} + + +# --- apply_payload --------------------------------------------------------------------- + + +def test_apply_payload_returns_data_unchanged_by_default(): + data = {"id": 1, "owner_extra_info": {"photo": "x", "full_name_display": "Alice"}} + + assert apply_payload(data, "userstory") == data + assert apply_payload(data, "userstory", payload="full") == data + + +def test_apply_payload_full_with_only_strip_media_strips_without_collapsing(): + # Regression test for the source spec's own pseudocode bug: payload="full" combined + # with an explicit strip_media=True must strip media only, not fall through to + # compact's collapse_extra_info behavior. + data = { + "id": 1, + "subject": "hello", + "owner_extra_info": {"id": 9, "full_name_display": "Alice", "photo": "x", "username": "alice"}, + } + + result = apply_payload(data, "userstory", payload="full", strip_media=True) + + assert result == { + "id": 1, + "subject": "hello", + "owner_extra_info": {"id": 9, "full_name_display": "Alice", "username": "alice"}, + } + + +def test_apply_payload_compact_collapses_extra_info_and_strips_media_by_default(): + data = { + "id": 1, + "owner_extra_info": {"id": 9, "full_name_display": "Alice", "photo": "x"}, + } + + result = apply_payload(data, "userstory", payload="compact") + + assert result == {"id": 1, "owner_extra_info": {"id": 9, "full_name_display": "Alice"}} + + +def test_apply_payload_compact_with_strip_media_false_keeps_top_level_media(): + data = {"id": 1, "photo": "https://x/a.png"} + + result = apply_payload(data, "membership", payload="compact", strip_media=False) + + assert result == {"id": 1, "photo": "https://x/a.png"} + + +def test_apply_payload_minimal_uses_the_entity_measured_field_set(): + data = { + "id": 1, + "ref": 42, + "subject": "hello", + "version": 3, + "milestone": 7, + "milestone_name": "Sprint 1", + "status": 2, + "status_extra_info": {"id": 2, "name": "Done", "is_closed": True}, + "is_closed": True, + "finish_date": None, + "is_blocked": False, + "assigned_to_extra_info": {"id": 5, "full_name_display": "Bob", "photo": "x"}, + "assigned_users": [10, 11], + "epics": [{"ref": 10, "subject": "Epic A"}], + "owner_extra_info": {"id": 9, "full_name_display": "Alice", "photo": "y"}, + } + + result = apply_payload(data, "userstory", payload="minimal") + + assert result == { + "id": 1, + "ref": 42, + "subject": "hello", + "version": 3, + "milestone": 7, + "milestone_name": "Sprint 1", + "status_extra_info": {"name": "Done", "is_closed": True}, + "is_closed": True, + "finish_date": None, + "is_blocked": False, + "assigned_to_extra_info": {"full_name_display": "Bob"}, + "assigned_users": [10, 11], + "epics": [{"ref": 10}], + } + + +def test_apply_payload_minimal_keeps_assigned_users_extra_info_for_userstory(): + data = { + "id": 1, + "ref": 42, + "subject": "hello", + "version": 3, + "milestone": 7, + "milestone_name": "Sprint 1", + "status": 2, + "status_extra_info": {"id": 2, "name": "Done", "is_closed": True}, + "is_closed": True, + "finish_date": None, + "is_blocked": False, + "assigned_to_extra_info": {"id": 5, "full_name_display": "Bob", "photo": "x"}, + "assigned_users": [10, 11], + "epics": [{"ref": 10, "subject": "Epic A"}], + "owner_extra_info": {"id": 9, "full_name_display": "Alice", "photo": "y"}, + "assigned_users_extra_info": [ + {"id": 10, "full_name_display": "Alice"}, + {"id": 11, "full_name_display": "Bob"}, + ], + } + + result = apply_payload(data, "userstory", payload="minimal") + + assert result == { + "id": 1, + "ref": 42, + "subject": "hello", + "version": 3, + "milestone": 7, + "milestone_name": "Sprint 1", + "status_extra_info": {"name": "Done", "is_closed": True}, + "is_closed": True, + "finish_date": None, + "is_blocked": False, + "assigned_to_extra_info": {"full_name_display": "Bob"}, + "assigned_users": [10, 11], + "epics": [{"ref": 10}], + "assigned_users_extra_info": [ + {"id": 10, "full_name_display": "Alice"}, + {"id": 11, "full_name_display": "Bob"}, + ], + } + + +def test_apply_payload_minimal_falls_back_to_compact_for_entity_without_a_measured_set(): + data = {"id": 1, "owner_extra_info": {"id": 9, "full_name_display": "Alice", "photo": "x"}} + + result = apply_payload(data, "task", payload="minimal") + + assert result == apply_payload(data, "task", payload="compact") + assert result == {"id": 1, "owner_extra_info": {"id": 9, "full_name_display": "Alice"}} + + +def test_apply_payload_fields_overrides_payload(): + data = {"id": 1, "subject": "hello", "status": 2} + + result = apply_payload(data, "userstory", payload="minimal", fields=["id", "subject"]) + + assert result == {"id": 1, "subject": "hello"} + + +def test_apply_payload_expand_adds_full_block_back_on_top_of_minimal(): + data = { + "id": 1, + "ref": 42, + "subject": "hello", + "version": 3, + "milestone": 7, + "milestone_name": "Sprint 1", + "status": 2, + "status_extra_info": {"id": 2, "name": "Done", "is_closed": True}, + "is_closed": True, + "finish_date": None, + "is_blocked": False, + "assigned_to_extra_info": {"id": 5, "full_name_display": "Bob", "photo": "x"}, + "epics": [], + } + + result = apply_payload(data, "userstory", payload="minimal", expand=["assigned_to_extra_info"]) + + # expand adds back every field of the block ("id" here, absent from MINIMAL_FIELDS), + # but the block still goes through the same strip pass as the rest of the response - + # "photo" is gone even though it was requested to be expanded (M6). + assert result["assigned_to_extra_info"] == {"id": 5, "full_name_display": "Bob"} + + +def test_apply_payload_applies_per_item_when_data_is_a_list(): + data = [{"id": 1, "subject": "a", "status": 2}, {"id": 2, "subject": "b", "status": 3}] + + result = apply_payload(data, "userstory", fields=["id", "subject"]) + + assert result == [{"id": 1, "subject": "a"}, {"id": 2, "subject": "b"}] + + +def test_to_jsonable_falls_back_to_str_for_unknown_types(): + assert to_jsonable(object) == str(object) + + +def test_select_fields_returns_scalars_unchanged(): + assert select_fields("text", ["a"]) == "text" + + +def test_select_fields_passes_non_dict_list_items_through(): + assert select_fields([1, {"a": 1, "b": 2}], ["a"]) == [1, {"a": 1}] + + +def test_merge_paths_ignores_keys_missing_from_source(): + from taiga.mcp_server.serialize import _merge_paths + + assert _merge_paths({"a": 1}, {"b": 2}, ["b", "missing"]) == {"a": 1, "b": 2} + + +def test_apply_payload_fields_are_not_media_stripped_by_default_from_payload(): + data = {"id": 1, "owner_extra_info": {"id": 2, "photo": "u"}} + + kept = apply_payload(data, "issue", payload="minimal", fields=["owner_extra_info.photo"]) + stripped = apply_payload(data, "issue", payload="minimal", fields=["owner_extra_info.photo"], strip_media=True) + + assert kept == {"owner_extra_info": {"photo": "u"}} + assert stripped == {"owner_extra_info": {}}