diff --git a/.github/workflows/lint.yml b/.github/workflows/lint.yml index eb0fbb0..e05667f 100644 --- a/.github/workflows/lint.yml +++ b/.github/workflows/lint.yml @@ -30,9 +30,9 @@ jobs: uses: actions/cache@v6 with: path: .tox - key: ${{ runner.os }}-lint-${{ matrix.toxenv }}-${{ hashFiles('setup.cfg') }} - restore-keys: | - ${{ runner.os }}-lint-${{ matrix.toxenv }}- + # Include every dependency declaration in the exact cache key so dependency + # changes produce a clean tox environment without relying on partial restores. + key: ${{ runner.os }}-lint-${{ matrix.toxenv }}-${{ hashFiles('setup.cfg', 'tox.ini', 'requirements*.txt') }} - name: Install dependencies run: | python -m pip install --upgrade pip setuptools tox>4 diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index fdd0373..d47fecf 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -26,9 +26,9 @@ jobs: uses: actions/cache@v6 with: path: .tox - key: ${{ runner.os }}-tox-${{ format('{{py{0}}}', matrix.python-version) }}-${{ hashFiles('setup.cfg') }} - restore-keys: | - ${{ runner.os }}-tox-${{ format('{{py{0}}}', matrix.python-version) }}- + # Include every dependency declaration in the exact cache key so dependency + # changes produce a clean tox environment without relying on partial restores. + key: ${{ runner.os }}-tox-${{ format('{{py{0}}}', matrix.python-version) }}-${{ hashFiles('setup.cfg', 'tox.ini', 'requirements*.txt') }} - name: Install dependencies run: | sudo apt-get install gettext diff --git a/.gitignore b/.gitignore index ba6122a..3008ff6 100644 --- a/.gitignore +++ b/.gitignore @@ -44,3 +44,7 @@ debian/files debian/python-taiga* debian/python3-taiga* .ruff_cache +.venv +*.egg-link +.superpowers +artifacts diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..fa58ff2 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,112 @@ +# Agent instructions + +This file gives coding agents (Claude Code and similar) step-by-step +instructions for tasks specific to this repository. Human-facing docs live in +``README.rst`` and ``docs/``. + +## Registering the Taiga MCP server in the user's global Claude config + +This repo ships an MCP server (`taiga/mcp_server/`) that exposes the Taiga +REST API as tools over stdio, via the `taiga-mcp-server` console script +(installed by the `mcp` extra: `pip install -e .[mcp]` from this repo, or +`pip install python-taiga[mcp]` from PyPI). + +When asked to "add the Taiga MCP server to Claude" / "register taiga-mcp +globally" / "add it to my user-wide config", follow this procedure: + +1. **Confirm before acting.** Registering at user scope changes the user's + global Claude Code config (`~/.claude.json`), applying to every project, + not just this repo. Confirm the target Taiga instance and scope with the + user before running the command, unless they've already given explicit + go-ahead in this conversation. + +2. **Get a stable `taiga-mcp-server` binary.** Don't point the MCP config at + a project-local `.venv` — Claude Code launches MCP server commands without + inheriting an activated venv, and the binary disappears if that venv is + ever recreated. Install it somewhere durable instead. There are several + equally valid ways to do this; pick whichever fits the user's toolchain, + asking if it's unclear, and default to `pip install --user` since it needs + nothing beyond a reasonably modern Python: + ```bash + # default: pip install --user (works with any modern Python/pip) + pip install --user "python-taiga[mcp]" # from PyPI + pip install --user -e ".[mcp]" # from this checkout + + # pipx (isolated venv per tool, one binary on PATH) + pipx install "python-taiga[mcp]" # from PyPI + pipx install --editable ".[mcp]" # from this checkout + + # uvx (no persistent install; uv manages an ephemeral/cached env) + # here the *registered command* becomes `uvx --from "python-taiga[mcp]" taiga-mcp-server` + # instead of a resolved path — see the uvx example in step 4. + ``` + After a `pip --user`/`pipx` install, resolve the resulting path and use it + verbatim in step 4: + ```bash + command -v taiga-mcp-server + ``` + +3. **Collect credentials.** Ask the user for: + - `TAIGA_HOST` — the Taiga site root, e.g. `https://my.taiga.com`. + For self-hosted instances this is *not* an `api.` subdomain and has no + `/api` suffix — the client appends `/api/v1` itself. + - Either `TAIGA_TOKEN` (pre-issued API token), or both + `TAIGA_USERNAME` and `TAIGA_PASSWORD`. A token takes precedence if both + are configured. + - Optional: `TAIGA_TOKEN_TYPE` (default `Bearer`), `TAIGA_TLS_VERIFY` + (default `true`). + + Never pass `--token`/`--password` as CLI arguments — they'd be visible in + the process list. Always pass credentials as environment variables. + + **Default to username/password over a token, unless the instance has a + real personal-access-token feature.** Stock Taiga (checked against + `https://my.taiga.com`) has no self-service PAT: the only tokens it + issues are (a) short-lived JWTs from `POST /api/v1/auth` — on that + instance, a 24h access token / 8-day refresh token — and (b) OAuth-style + "Application" tokens, which require an admin-registered app and a + consent/`auth_code` flow (`client.auth_app()`), not something a regular + user can self-serve. This server's `auth.py`/CLI has no refresh-token + support, so a manually-generated `TAIGA_TOKEN` will just silently stop + working after ~24h with no renewal — worse than username/password, which + re-authenticates fresh on every server start. Only reach for `TAIGA_TOKEN` + when the target instance genuinely offers a durable personal token (e.g. + a Taiga Enterprise/hosted deployment with PAT support) — verify that + before recommending it, don't assume it exists. + +4. **Register at user scope** with `claude mcp add`, using `-e` for every + credential env var and the resolved binary (or `uvx` invocation) from + step 2: + ```bash + claude mcp add --scope user taiga \ + -e TAIGA_HOST=https://my.taiga.com \ + -e TAIGA_USERNAME= \ + -e TAIGA_PASSWORD= \ + -- /absolute/path/to/taiga-mcp-server serve + ``` + or, with a token instead of username/password: + ```bash + claude mcp add --scope user taiga \ + -e TAIGA_HOST=https://my.taiga.com \ + -e TAIGA_TOKEN= \ + -- /absolute/path/to/taiga-mcp-server serve + ``` + With `uvx` there's no path to resolve — pass the `uvx` invocation itself + as the command: + ```bash + claude mcp add --scope user taiga \ + -e TAIGA_HOST=https://my.taiga.com \ + -e TAIGA_TOKEN= \ + -- uvx --from "python-taiga[mcp]" taiga-mcp-server serve + ``` + `--scope user` (not `local`/`project`) is what makes it "user-wide" — + available in every project for that user, stored outside this repo. + +5. **Verify** with `claude mcp list` (look for `taiga` ... `✔ Connected`) and + `claude mcp get taiga`. If it fails to connect, re-check the resolved + binary/command from step 2 and that `TAIGA_HOST` is the site root, not an + API subdomain. + +6. **Don't persist secrets in the repo.** Credentials belong only in the + `claude mcp add -e ...` invocation (stored in the user's own + `~/.claude.json`) — never write them into files inside this repository. diff --git a/MANIFEST.in b/MANIFEST.in index ee04217..4c7888c 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -1,8 +1,9 @@ +include AGENTS.md include AUTHORS include LICENSE include README.rst include CONTRIBUTING.rst include HISTORY.rst include requirements.txt -include requirements-tests.txt +include requirements-test.txt recursive-include taiga *.html *.png *.gif *js *jpg *jpeg *svg *py *mo *po diff --git a/changes/267.feature b/changes/267.feature new file mode 100644 index 0000000..19e3984 --- /dev/null +++ b/changes/267.feature @@ -0,0 +1 @@ +Add MCP server exposing Taiga projects, user stories, tasks, issues, epics, milestones and wiki pages as tools for AI agents. `taiga-mcp-server` also gains `list-tools` and `call` subcommands, letting tools be listed and invoked directly from a shell without an MCP client. The server also exposes custom-attribute value read/write, project membership listing, and epic-user-story linking. diff --git a/docs/index.rst b/docs/index.rst index b76c672..04a953f 100644 --- a/docs/index.rst +++ b/docs/index.rst @@ -10,6 +10,7 @@ Welcome to python-taiga's documentation! :maxdepth: 3 usage + mcp api models development diff --git a/docs/mcp.rst b/docs/mcp.rst new file mode 100644 index 0000000..3be3a61 --- /dev/null +++ b/docs/mcp.rst @@ -0,0 +1,268 @@ +.. :mcp: + +========== +MCP Server +========== + +Contents: + +python-taiga ships a `Model Context Protocol `_ +(MCP) server that exposes Taiga projects, user stories, tasks, issues, epics, +milestones and wiki pages as tools an LLM-based assistant (Claude, or any +other MCP-compatible client) can call directly, without you writing any glue +code. + +.. note:: The MCP server wraps the same ``TaigaAPI`` documented in + :doc:`the usage guide ` and :doc:`the API reference ` - + if you need to script against Taiga from Python yourself, use + ``TaigaAPI`` directly instead. + +**************** +Installation +**************** + +The server is an optional extra, since it pulls in the official `MCP Python SDK +`_ (``mcp``) as a dependency: + +.. code:: shell + + pip install "python-taiga[mcp]" + +Any of the following also work, depending on your toolchain: + +.. code:: shell + + pip install --user "python-taiga[mcp]" # no virtualenv management needed + pipx install "python-taiga[mcp]" # isolated venv, one command on PATH + uvx --from "python-taiga[mcp]" taiga-mcp-server --help # no persistent install at all + +Any of these makes a ``taiga-mcp-server`` console script available. + +**************** +Configuration +**************** + +Credentials are read from environment variables, or from equivalent +command-line flags (flags take precedence over the environment): + +.. list-table:: + :header-rows: 1 + :widths: 20 25 55 + + * - Environment variable + - CLI flag + - Meaning + * - ``TAIGA_HOST`` + - ``--host`` + - Taiga instance root, e.g. ``https://taiga.example.com``. Defaults to + ``https://api.taiga.io``. + * - ``TAIGA_TOKEN`` + - ``--token`` + - A pre-issued auth token. Takes precedence over username/password if + both are set. + * - ``TAIGA_TOKEN_TYPE`` + - ``--token-type`` + - Type of the token above. Defaults to ``Bearer``. + * - ``TAIGA_USERNAME`` + - ``--username`` + - Username, used together with the password below. + * - ``TAIGA_PASSWORD`` + - ``--password`` + - Password, exchanged for a session token at startup. + * - ``TAIGA_TLS_VERIFY`` + - ``--tls-verify`` / ``--no-tls-verify`` + - Verify TLS certificates. Defaults to ``true``. + +.. warning:: Prefer the environment variables over the CLI flags for + ``--token``/``--password``: command-line arguments are visible + to other processes on the same machine (e.g. via ``ps``), + environment variables set for the server's own process are not. + +.. note:: Most Taiga instances don't offer a durable personal-access-token + feature - the token obtained from a username/password login is a + short-lived JWT (often expiring within a day), and this server + doesn't refresh it once started. Unless you know your instance + issues long-lived tokens, configure ``TAIGA_USERNAME``/ + ``TAIGA_PASSWORD`` rather than a fixed ``TAIGA_TOKEN`` - the server + re-authenticates fresh every time it starts. + +****************************** +Running the server standalone +****************************** + +.. code:: shell + + TAIGA_HOST=https://taiga.example.com \ + TAIGA_USERNAME=myuser \ + TAIGA_PASSWORD=mypassword \ + taiga-mcp-server serve + +The server speaks MCP over stdio and is meant to be launched by an MCP +client, not used interactively - the command above will sit and wait for a +client to connect over stdin/stdout. + +********************************** +Listing and calling tools directly +********************************** + +Outside of an MCP client, ``taiga-mcp-server`` also exposes its tool set +directly from a shell: + +.. code:: shell + + # list every tool, one per line + taiga-mcp-server list-tools + + # ...with each tool's JSON input schema + taiga-mcp-server list-tools --verbose + + # call a single tool by name, passing its arguments as a JSON object + TAIGA_HOST=https://taiga.example.com \ + TAIGA_USERNAME=myuser \ + TAIGA_PASSWORD=mypassword \ + taiga-mcp-server call whoami --json '{}' + + taiga-mcp-server call get_project --json '{"project": "myproject"}' + +On success, ``call`` prints the tool's JSON result to stdout. On failure +(unknown tool name, invalid arguments, or an error from the underlying +Taiga API call) it prints a message to stderr and exits with a non-zero +status. + +***************************** +Connecting an MCP client +***************************** + +Any MCP client that supports the stdio transport can launch +``taiga-mcp-server`` as a subprocess. For `Claude Code +`_, register it once and it's +available in every project: + +.. code:: shell + + claude mcp add --scope user taiga \ + -e TAIGA_HOST=https://taiga.example.com \ + -e TAIGA_USERNAME=myuser \ + -e TAIGA_PASSWORD=mypassword \ + -- "$(command -v taiga-mcp-server)" serve + +``--scope user`` stores the registration in your own Claude configuration, +not in any particular project. Check it went through with: + +.. code:: shell + + claude mcp get taiga + +**************** +Available tools +**************** + +``whoami`` + Return the Taiga user currently authenticated. + +``list_projects`` / ``get_project`` + List projects visible to the user, or fetch one project's full detail + (numeric id or slug) - including the statuses/priorities/severities/points + ids needed to create or update entities in it. + +``search`` + Search user stories, tasks, issues, epics and wiki pages in a project. + +``list_memberships`` + 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. + +``add_comment`` / ``add_comment_by_id`` + Add a comment to a user story, task, issue or epic, identified by + ``project`` + ``ref`` (primary) or by database ``id`` (secondary, see + below). + +``get_history`` / ``get_history_by_id`` + Get the full change/comment history of a user story, task, issue, epic or + wiki page. Each entry's `comment` field is empty for plain field-change + events and non-empty for an actual comment; `delete_comment_date` is + non-null if that comment was later deleted. Wiki pages have no ref number + in Taiga, so for ``entity_type="wiki"`` pass the page's database id as + ``ref`` and omit ``project``. + +``get_custom_attributes_values`` / ``get_custom_attributes_values_by_id`` + Get the custom-attribute values of a user story, task, issue or epic. + Keys of ``attributes_values`` are attribute ids as strings - see + ``get_project``'s ``*_custom_attributes`` lists for id -> name. + +``set_custom_attribute_value`` / ``set_custom_attribute_value_by_id`` + Set one custom-attribute value on a user story, task, issue or epic. + ``attribute_id`` is the numeric id from ``get_project``'s + ``*_custom_attributes`` list. + +.. important:: The ``version`` returned by ``get_custom_attributes_values`` + (and expected by ``set_custom_attribute_value``) belongs to that + custom-attributes-values resource - a separate version sequence + from the entity's own ``version`` field. Always pass back the + version from a prior ``get_custom_attributes_values`` call (or + ``1`` if never set before), not the entity's own ``version``. + +``list_user_stories``, ``get_user_story``, ``create_user_story``, ``update_user_story``, ``delete_user_story`` + Manage user stories. + +``list_tasks``, ``get_task``, ``create_task``, ``update_task``, ``delete_task`` + Manage tasks, optionally scoped to a project and/or a user story. + +``list_issues``, ``get_issue``, ``create_issue``, ``update_issue``, ``delete_issue`` + Manage issues. + +``list_epics``, ``get_epic``, ``create_epic``, ``update_epic``, ``delete_epic`` + Manage epics. + +``link_epic_user_story`` / ``link_epic_user_story_by_id`` + Link a user story to an epic, identifying both by their per-project ref + numbers (primary) or by database id (secondary, see below). + +.. 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 + shows in its UI and URLs (e.g. the ``45634`` in + ``.../issues/45634``). That ref is **not** the database id used + internally for updates/deletes - it's only unique within a project, + so it must be resolved together with ``project``. This is the + primary, recommended way to address an entity, since numbers a user + pastes from a Taiga URL or mentions in conversation are almost + always refs. + + Each of these tools also has a ``_by_id`` counterpart (e.g. + ``get_issue_by_id``, ``update_task_by_id``, ``delete_epic_by_id``, + ``add_comment_by_id``) that takes the raw database ``id`` instead. + These are a secondary, non-default lookup path - use them only when + 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_wiki_pages``, ``get_wiki_page``, ``create_wiki_page``, ``update_wiki_page`` + Manage wiki pages. + +.. tip:: Call ``get_project`` first when creating or updating an entity - it + returns every status/priority/severity/points id valid for that + project, which the ``create_*``/``update_*`` tools expect. + +.. tip:: Every ``list_*`` tool is paginated and defaults to page 1 of up to + 100 results. Pass ``page``/``page_size`` in ``filters`` to move + through further pages, and ``order_by`` (e.g. ``-created_date``) to + control ordering - for example to fetch the most recent items first. + +**************** +Security notes +**************** + +The MCP server has the same permissions as the account it authenticates +with, and the create/update/delete tools above are destructive: an assistant +with access to this server can create, modify or delete real data in your +Taiga projects. Review what an MCP client proposes to do before approving +write operations, and consider a dedicated Taiga account with restricted +project membership if you want to limit the blast radius. + +``set_custom_attribute_value``/``set_custom_attribute_value_by_id`` and +``link_epic_user_story``/``link_epic_user_story_by_id`` are also writes and +fall under the same destructive-tools framing above. diff --git a/docs/usage.rst b/docs/usage.rst index f56bb85..c0ef662 100644 --- a/docs/usage.rst +++ b/docs/usage.rst @@ -202,6 +202,15 @@ Create an issue description='Bug #5' ) +****************************************************** +Link a user story to an epic +****************************************************** + +.. code:: python + + epic = new_project.add_epic('New Epic') + epic.add_related_user_story(userstory.id) + ****************************************************** Create a custom attribute ****************************************************** diff --git a/requirements.txt b/requirements.txt index d6e1198..5f6ce98 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1 +1 @@ --e . +-e .[mcp] diff --git a/setup.cfg b/setup.cfg index f3cc12c..11af04c 100644 --- a/setup.cfg +++ b/setup.cfg @@ -27,21 +27,33 @@ include_package_data = True install_requires = requests>2.11 python-dateutil>=2.4 -packages = taiga +packages = find: python_requires = >=3.11 setup_requires = setuptools zip_safe = False test_suite = tests +[options.packages.find] +include = + taiga + taiga.* + [options.package_data] * = *.txt, *.rst taiga = *.html *.png *.gif *js *jpg *jpeg *svg *py *mo *po +[options.entry_points] +console_scripts = + taiga-mcp-server = taiga.mcp_server.cli:main + [options.extras_require] docs = sphinx sphinx-rtd-theme +mcp = + mcp~=2.0 + typer>=0.12.0 [sdist] formats = zip diff --git a/taiga/mcp_server/__init__.py b/taiga/mcp_server/__init__.py new file mode 100644 index 0000000..d1fbadf --- /dev/null +++ b/taiga/mcp_server/__init__.py @@ -0,0 +1,7 @@ +# python-taiga +# Copyright 2015 Nephila +# See LICENSE for details. + +""" +MCP server exposing python-taiga as a set of tools for LLM clients. +""" diff --git a/taiga/mcp_server/auth.py b/taiga/mcp_server/auth.py new file mode 100644 index 0000000..d25fe7f --- /dev/null +++ b/taiga/mcp_server/auth.py @@ -0,0 +1,70 @@ +# python-taiga +# Copyright 2015 Nephila +# See LICENSE for details. + +from __future__ import annotations + +from dataclasses import dataclass + +from ..client import TaigaAPI +from ..exceptions import TaigaException + +DEFAULT_HOST = "https://api.taiga.io" +DEFAULT_TOKEN_TYPE = "Bearer" + + +class ConfigError(TaigaException): + """Raised when there isn't enough information to authenticate, or the server wasn't configured.""" + + +@dataclass +class Credentials: + host: str = DEFAULT_HOST + tls_verify: bool = True + token: str | None = None + token_type: str = DEFAULT_TOKEN_TYPE + username: str | None = None + password: str | None = None + + +def build_client(credentials: Credentials) -> TaigaAPI: + """ + Build and authenticate a :class:`TaigaAPI` client from the given credentials. + + A token takes precedence over username/password if both are set. + """ + if credentials.token: + return TaigaAPI( + host=credentials.host, + token=credentials.token, + token_type=credentials.token_type, + tls_verify=credentials.tls_verify, + ) + + if credentials.username and credentials.password: + api = TaigaAPI(host=credentials.host, tls_verify=credentials.tls_verify) + api.auth(credentials.username, credentials.password) + return api + + raise ConfigError("Missing Taiga credentials: provide a token, or both a username and a password.") + + +_credentials: Credentials | None = None +_client: TaigaAPI | None = None + + +def configure(credentials: Credentials) -> None: + """Store the credentials used to lazily build the Taiga client on first use.""" + global _credentials, _client + _credentials = credentials + _client = None + + +def get_client() -> TaigaAPI: + """Return a lazily-built, process-wide :class:`TaigaAPI` client.""" + global _client + if _client is None: + if _credentials is None: + raise ConfigError("The Taiga MCP server has not been configured with any credentials.") + _client = build_client(_credentials) + return _client diff --git a/taiga/mcp_server/cli.py b/taiga/mcp_server/cli.py new file mode 100644 index 0000000..92a88c3 --- /dev/null +++ b/taiga/mcp_server/cli.py @@ -0,0 +1,177 @@ +# python-taiga +# Copyright 2015 Nephila +# See LICENSE for details. + +from __future__ import annotations + +import asyncio +import json +import os + +import typer +from mcp.server.mcpserver.exceptions import ToolError +from mcp.shared.exceptions import MCPError +from pydantic import ValidationError as PydanticValidationError + +from .. import __version__ +from .auth import DEFAULT_HOST, DEFAULT_TOKEN_TYPE, Credentials, configure + +app = typer.Typer( + add_completion=False, + no_args_is_help=True, + help="Taiga MCP server & CLI. Prefer TAIGA_TOKEN/TAIGA_PASSWORD env vars over " + "--token/--password, which can be visible in the process list.", +) + + +def _version_callback(value: bool) -> None: + if value: + typer.echo(f"taiga-mcp-server (python-taiga {__version__})") + raise typer.Exit() + + +@app.callback() +def _main( + version: bool | None = typer.Option( + None, "--version", callback=_version_callback, is_eager=True, help="Show the version and exit." + ), +) -> None: + """Taiga MCP server & CLI.""" + + +def _env_bool(name: str, default: bool) -> bool: + value = os.environ.get(name) + if value is None: + return default + return value.strip().lower() not in ("0", "false", "no", "off") + + +def _resolve_credentials( + host: str | None, + token: str | None, + token_type: str | None, + username: str | None, + password: str | None, + tls_verify: bool | None, +) -> Credentials: + return Credentials( + host=host or os.environ.get("TAIGA_HOST", DEFAULT_HOST), + tls_verify=_env_bool("TAIGA_TLS_VERIFY", True) if tls_verify is None else tls_verify, + token=token or os.environ.get("TAIGA_TOKEN"), + token_type=token_type or os.environ.get("TAIGA_TOKEN_TYPE", DEFAULT_TOKEN_TYPE), + username=username or os.environ.get("TAIGA_USERNAME"), + password=password or os.environ.get("TAIGA_PASSWORD"), + ) + + +HostOption = typer.Option(None, help="Taiga instance host (default: TAIGA_HOST env var, or https://api.taiga.io).") +TokenOption = typer.Option(None, help="Taiga auth token (default: TAIGA_TOKEN env var).") +TokenTypeOption = typer.Option(None, help="Type of the auth token (default: TAIGA_TOKEN_TYPE env var, or Bearer).") +UsernameOption = typer.Option(None, help="Taiga username (default: TAIGA_USERNAME env var).") +PasswordOption = typer.Option(None, help="Taiga password (default: TAIGA_PASSWORD env var).") +TlsVerifyOption = typer.Option( + None, + "--tls-verify/--no-tls-verify", + help="Verify TLS certificates (default: TAIGA_TLS_VERIFY env var, or true).", +) + + +@app.command() +def serve( + host: str | None = HostOption, + token: str | None = TokenOption, + token_type: str | None = TokenTypeOption, + username: str | None = UsernameOption, + password: str | None = PasswordOption, + tls_verify: bool | None = TlsVerifyOption, +) -> None: + """Run the MCP server over stdio. + + Credentials can be passed as flags or read from the TAIGA_HOST/TAIGA_TOKEN + or TAIGA_HOST/TAIGA_USERNAME/TAIGA_PASSWORD environment variables. Passing + --token/--password on the command line can expose them via the process + list; prefer the environment variables where possible. + """ + configure(_resolve_credentials(host, token, token_type, username, password, tls_verify)) + + from .server import mcp + + mcp.run(transport="stdio") + + +@app.command("list-tools") +def list_tools( + host: str | None = HostOption, + token: str | None = TokenOption, + token_type: str | None = TokenTypeOption, + username: str | None = UsernameOption, + password: str | None = PasswordOption, + tls_verify: bool | None = TlsVerifyOption, + verbose: bool = typer.Option(False, "--verbose", "-v", help="Include each tool's JSON input schema."), +) -> None: + """List every tool exposed by the MCP server.""" + configure(_resolve_credentials(host, token, token_type, username, password, tls_verify)) + + from .server import mcp + + tools = asyncio.run(mcp.list_tools()) + for tool in sorted(tools, key=lambda t: t.name): + dumped = tool.model_dump(by_alias=True, exclude_none=True) + typer.echo(f"{dumped['name']}\t{dumped.get('description', '')}") + if verbose: + typer.echo(json.dumps(dumped["inputSchema"], indent=2)) + + +@app.command() +def call( + tool_name: str = typer.Argument(..., help="Tool name, as shown by list-tools."), + arguments: str = typer.Option("{}", "--json", "-j", help="JSON object of arguments for the tool."), + host: str | None = HostOption, + token: str | None = TokenOption, + token_type: str | None = TokenTypeOption, + username: str | None = UsernameOption, + password: str | None = PasswordOption, + tls_verify: bool | None = TlsVerifyOption, +) -> None: + """Call a single tool directly, bypassing an MCP client. + + Prefer the TAIGA_TOKEN/TAIGA_PASSWORD environment variables over + --token/--password, which can be visible in the process list. + """ + try: + parsed_arguments = json.loads(arguments) + except json.JSONDecodeError as exc: + typer.echo(f"Invalid JSON in --json: {exc}", err=True) + raise typer.Exit(1) from exc + + configure(_resolve_credentials(host, token, token_type, username, password, tls_verify)) + + from .server import mcp + + try: + result = asyncio.run(mcp.call_tool(tool_name, parsed_arguments)) + except ToolError as exc: + cause = exc.__cause__ + message = str(exc) + if message.startswith("Unknown tool: "): + typer.echo(message, err=True) + elif isinstance(cause, PydanticValidationError): + typer.echo(f"Invalid arguments for {tool_name}: {cause}", err=True) + else: + typer.echo(f"Error calling {tool_name}: {cause if cause is not None else exc}", err=True) + raise typer.Exit(1) from exc + except MCPError as exc: + typer.echo(f"Error calling {tool_name}: {exc}", err=True) + raise typer.Exit(1) from exc + + payload = result.structured_content if result.structured_content is not None else result.content + typer.echo(json.dumps(payload, indent=2, default=str)) + + +def main() -> None: + """Entry point for the ``taiga-mcp-server`` console script.""" + app() + + +if __name__ == "__main__": + main() diff --git a/taiga/mcp_server/serialize.py b/taiga/mcp_server/serialize.py new file mode 100644 index 0000000..d6c7ca3 --- /dev/null +++ b/taiga/mcp_server/serialize.py @@ -0,0 +1,27 @@ +# python-taiga +# Copyright 2015 Nephila +# See LICENSE for details. + +from __future__ import annotations + +import datetime +from typing import Any + +from ..models.base import InstanceResource + +_SKIPPED_ATTRS = {"requester"} + + +def to_jsonable(value: Any) -> Any: + """Recursively convert python-taiga models into plain JSON-serializable structures.""" + if value is None or isinstance(value, (str, int, float, bool)): + return value + if isinstance(value, (datetime.datetime, datetime.date)): + return value.isoformat() + if isinstance(value, InstanceResource): + return {key: to_jsonable(val) for key, val in vars(value).items() if key not in _SKIPPED_ATTRS} + if isinstance(value, dict): + return {key: to_jsonable(val) for key, val in value.items()} + if isinstance(value, (list, tuple)): + return [to_jsonable(item) for item in value] + return str(value) diff --git a/taiga/mcp_server/server.py b/taiga/mcp_server/server.py new file mode 100644 index 0000000..974066d --- /dev/null +++ b/taiga/mcp_server/server.py @@ -0,0 +1,694 @@ +# python-taiga +# Copyright 2015 Nephila +# See LICENSE for details. + +from __future__ import annotations + +from typing import Any, Literal + +from mcp.server.mcpserver import MCPServer + +from .auth import get_client +from .serialize import to_jsonable + +mcp = MCPServer( + name="taiga", + instructions=( + "Tools to read and manage Taiga projects: user stories, tasks, issues, epics, " + "milestones and wiki pages. Configure credentials via the TAIGA_HOST/TAIGA_TOKEN " + "or TAIGA_HOST/TAIGA_USERNAME/TAIGA_PASSWORD environment variables. " + "`get_project` returns the full set of statuses/priorities/severities/points ids " + "needed to create or update entities in that project." + ), +) + +_ENTITY_ATTR = { + "user_story": "user_stories", + "task": "tasks", + "issue": "issues", + "epic": "epics", +} + +_REF_METHOD = { + "user_story": "get_userstory_by_ref", + "task": "get_task_by_ref", + "issue": "get_issue_by_ref", + "epic": "get_epic_by_ref", +} + + +def _resolve_project_id(project: str | int) -> int: + if isinstance(project, int) or str(project).isdigit(): + return int(project) + client = get_client() + return client.projects.get_by_slug(str(project)).id + + +def _resolve_project(project: str | int) -> Any: + """Fetch the full Project resource. + + Ref-based lookups need the project's id *and* slug, so (unlike + `_resolve_project_id`) this always fetches the project even when given a + numeric id. + """ + client = get_client() + if isinstance(project, int) or str(project).isdigit(): + return client.projects.get(int(project)) + return client.projects.get_by_slug(str(project)) + + +def _get_by_ref(entity_type: str, project: str | int, ref: int) -> Any: + """Resolve a user_story/task/issue/epic to its resource via its per-project ref number. + + `ref` is the sequential number Taiga shows per project - e.g. the 45634 in + `.../issues/45634` - not the database id used internally for update/delete. + """ + proj = _resolve_project(project) + return getattr(proj, _REF_METHOD[entity_type])(ref) + + +DEFAULT_PAGE_SIZE = 100 + + +def _paginated(query: dict[str, Any]) -> dict[str, Any]: + """Default a list query to a single bounded page. + + The underlying client only stops auto-fetching subsequent pages once an explicit + `page` is given — `page_size` alone does not limit it — so a caller that omits + `page` would otherwise silently walk and return the *entire* remote collection, + which for large projects can mean tens of thousands of records in one response. + Pass `page`/`page_size` inside `filters` to move through further pages. + + `filters` is forwarded straight into `ListResource.list()`, so a caller could + otherwise defeat this bound by passing `pagination=False` (a client-control kwarg, + stripped here) or an explicit but falsy `page`/`page_size` (e.g. `None` or `0`, + normalized here rather than left as-is like `dict.setdefault` would). + """ + query.pop("pagination", None) + if not query.get("page"): + query["page"] = 1 + if not query.get("page_size"): + query["page_size"] = DEFAULT_PAGE_SIZE + return query + + +@mcp.tool() +def whoami() -> dict[str, Any]: + """Return the Taiga user currently authenticated.""" + return to_jsonable(get_client().me()) + + +@mcp.tool() +def list_projects(member: int | None = None, filters: dict[str, Any] | None = None) -> 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. + """ + query = dict(filters or {}) + if member is not None: + query["member"] = member + return to_jsonable(get_client().projects.list(**_paginated(query))) + + +@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.""" + 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))) + + +@mcp.tool() +def search(project: str | int, text: str = "") -> dict[str, Any]: + """Search user stories, tasks, issues, epics and wiki pages in a project.""" + client = get_client() + result = client.search(_resolve_project_id(project), text) + return { + "count": result.count, + "user_stories": to_jsonable(result.user_stories), + "tasks": to_jsonable(result.tasks), + "issues": to_jsonable(result.issues), + "epics": to_jsonable(result.epics), + "wikipages": to_jsonable(result.wikipages), + } + + +@mcp.tool() +def list_memberships(project: str | int, filters: dict[str, Any] | None = None) -> 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. + """ + proj = _resolve_project(project) + query = _paginated(dict(filters or {})) + return to_jsonable(proj.list_memberships(**query)) + + +@mcp.tool() +def add_comment( + entity_type: Literal["user_story", "task", "issue", "epic"], project: str | int, ref: int, comment: str +) -> dict[str, Any]: + """Add a comment to a user story, task, issue or epic identified by its per-project ref number.""" + # CommentableResource.add_comment() delegates to update(), which returns the stale + # pre-comment resource with only `version` refreshed - not the comment itself - so it + # must not be serialized as the result; return an explicit acknowledgement instead. + resource = _get_by_ref(entity_type, project, ref) + resource.add_comment(comment) + return {"status": "commented", "ref": str(ref), "comment": comment} + + +@mcp.tool() +def add_comment_by_id( + entity_type: Literal["user_story", "task", "issue", "epic"], id: int, comment: str +) -> dict[str, Any]: # noqa: A002 + """Add a comment by database id. + + Secondary lookup: prefer `add_comment` with a project + ref (the number shown in the + Taiga UI/URL). Use this only when you already hold the raw database id. + """ + client = get_client() + resource = getattr(client, _ENTITY_ATTR[entity_type]).get(id) + resource.add_comment(comment) + return {"status": "commented", "id": str(id), "comment": comment} + + +_HISTORY_ENTITY_TYPES = ("user_story", "task", "issue", "epic", "wiki") + + +@mcp.tool() +def get_history( + entity_type: Literal["user_story", "task", "issue", "epic", "wiki"], + ref: int, + project: str | int | None = None, +) -> list[dict[str, Any]]: + """Get the full change/comment history of a user story, task, issue, epic or wiki page. + + For entity_type in user_story/task/issue/epic, identify the entity by its per-project + `ref` number (the one shown in the Taiga UI/URL) plus `project`. Wiki pages have no ref + number in Taiga - for entity_type="wiki", pass the page's database id as `ref` and omit + `project`. + + Each entry has a `comment` field (empty string for pure field-change events, non-empty + for an actual comment) and `delete_comment_date` (non-null if the comment was deleted). + """ + if entity_type != "wiki" and project is None: + raise ValueError("project is required unless entity_type is 'wiki'") + client = get_client() + if entity_type == "wiki": + return to_jsonable(client.history.wiki.get(ref)) + resource = _get_by_ref(entity_type, project, ref) + return to_jsonable(getattr(client.history, entity_type).get(resource.id)) + + +@mcp.tool() +def get_history_by_id( + entity_type: Literal["user_story", "task", "issue", "epic", "wiki"], id: int # noqa: A002 +) -> list[dict[str, Any]]: + """Get history by database id. + + Secondary lookup: prefer `get_history` with a project + ref (the number shown in the + Taiga UI/URL). Use this only when you already hold the raw database id. + """ + client = get_client() + return to_jsonable(getattr(client.history, entity_type).get(id)) + + +@mcp.tool() +def get_custom_attributes_values( + entity_type: Literal["user_story", "task", "issue", "epic"], + project: str | int, + ref: int, +) -> dict[str, Any]: + """Get the custom-attribute values of a user story, task, issue or epic, + identified by its per-project ref number. Keys of `attributes_values` are + attribute ids as strings - see get_project's `*_custom_attributes` lists + for id -> name. The returned `version` belongs to this custom-attributes- + values resource, a separate version sequence from the entity's own + `version` field - pass it back to `set_custom_attribute_value`, not the + entity's version. + """ + resource = _get_by_ref(entity_type, project, ref) + return to_jsonable(resource.get_attributes()) + + +@mcp.tool() +def get_custom_attributes_values_by_id( + entity_type: Literal["user_story", "task", "issue", "epic"], id: int # noqa: A002 +) -> dict[str, Any]: + """Get custom-attribute values by database id. + + Secondary lookup: prefer `get_custom_attributes_values` with a project + ref. + Use this only when you already hold the raw database id. + """ + client = get_client() + resource = getattr(client, _ENTITY_ATTR[entity_type]).get(id) + return to_jsonable(resource.get_attributes()) + + +@mcp.tool() +def set_custom_attribute_value( + entity_type: Literal["user_story", "task", "issue", "epic"], + project: str | int, + ref: int, + attribute_id: int, + value: Any, + version: int, +) -> dict[str, Any]: + """Set one custom-attribute value on a user story, task, issue or epic, + identified by its per-project ref number. `attribute_id` is the numeric id + from get_project's `*_custom_attributes` list (e.g. the "Code" attribute). + `version` is the custom-attributes-values resource's own version (from a + prior get_custom_attributes_values call, or 1 if never set before) - not + the entity's own `version` field. + """ + resource = _get_by_ref(entity_type, project, ref) + return to_jsonable(resource.set_attribute(attribute_id, value, version=version)) + + +@mcp.tool() +def set_custom_attribute_value_by_id( + entity_type: Literal["user_story", "task", "issue", "epic"], + id: int, # noqa: A002 + attribute_id: int, + value: Any, + version: int, +) -> dict[str, Any]: + """Set a custom-attribute value by database id. + + Secondary lookup: prefer `set_custom_attribute_value` with a project + ref. + Use this only when you already hold the raw database id. + """ + client = get_client() + resource = getattr(client, _ENTITY_ATTR[entity_type]).get(id) + return to_jsonable(resource.set_attribute(attribute_id, value, version=version)) + + +# --- User stories ----------------------------------------------------------------- + + +@mcp.tool() +def list_user_stories(project: str | int | None = None, filters: dict[str, Any] | None = None) -> 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. + """ + 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))) + + +@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)) + + +@mcp.tool() +def get_user_story_by_id(id: int) -> dict[str, Any]: # noqa: A002 + """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. + """ + return to_jsonable(get_client().user_stories.get(id)) + + +@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.""" + pid = _resolve_project_id(project) + return to_jsonable(get_client().user_stories.create(pid, subject, **(fields or {}))) + + +@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. + resource = _get_by_ref("user_story", project, ref) + resource.patch(list(fields.keys()), **fields) + return to_jsonable(get_client().user_stories.get(resource.id)) + + +@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.""" + client = get_client() + resource = client.user_stories.get(id) + resource.patch(list(fields.keys()), **fields) + return to_jsonable(client.user_stories.get(id)) + + +@mcp.tool() +def delete_user_story(project: str | int, ref: int) -> dict[str, str]: + """Delete a user story identified by its per-project ref number.""" + resource = _get_by_ref("user_story", project, ref) + resource.delete() + return {"status": "deleted", "ref": str(ref)} + + +@mcp.tool() +def delete_user_story_by_id(id: int) -> dict[str, str]: # noqa: A002 + """Delete a user story by its database id. Secondary lookup - prefer `delete_user_story` with a project + ref.""" + get_client().user_stories.delete(id) + return {"status": "deleted", "id": str(id)} + + +# --- Tasks -------------------------------------------------------------------------- + + +@mcp.tool() +def list_tasks( + project: str | int | None = None, user_story: int | None = None, filters: dict[str, Any] | None = None +) -> 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. + """ + 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))) + + +@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)) + + +@mcp.tool() +def get_task_by_id(id: int) -> dict[str, Any]: # noqa: A002 + """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. + """ + return to_jsonable(get_client().tasks.get(id)) + + +@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.""" + pid = _resolve_project_id(project) + return to_jsonable(get_client().tasks.create(pid, subject, status, **(fields or {}))) + + +@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. + resource = _get_by_ref("task", project, ref) + resource.patch(list(fields.keys()), **fields) + return to_jsonable(get_client().tasks.get(resource.id)) + + +@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.""" + client = get_client() + resource = client.tasks.get(id) + resource.patch(list(fields.keys()), **fields) + return to_jsonable(client.tasks.get(id)) + + +@mcp.tool() +def delete_task(project: str | int, ref: int) -> dict[str, str]: + """Delete a task identified by its per-project ref number.""" + resource = _get_by_ref("task", project, ref) + resource.delete() + return {"status": "deleted", "ref": str(ref)} + + +@mcp.tool() +def delete_task_by_id(id: int) -> dict[str, str]: # noqa: A002 + """Delete a task by its database id. Secondary lookup - prefer `delete_task` with a project + ref.""" + get_client().tasks.delete(id) + return {"status": "deleted", "id": str(id)} + + +# --- Issues --------------------------------------------------------------------------- + + +@mcp.tool() +def list_issues(project: str | int | None = None, filters: dict[str, Any] | None = None) -> 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. + """ + query = dict(filters or {}) + if project is not None: + query["project"] = _resolve_project_id(project) + return to_jsonable(get_client().issues.list(**_paginated(query))) + + +@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)) + + +@mcp.tool() +def get_issue_by_id(id: int) -> dict[str, Any]: # noqa: A002 + """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. + """ + return to_jsonable(get_client().issues.get(id)) + + +@mcp.tool() +def create_issue( + project: str | int, + subject: str, + priority: int, + status: int, + issue_type: int, + severity: int, + fields: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Create an issue. `priority`/`status`/`issue_type`/`severity` are numeric ids (see get_project).""" + pid = _resolve_project_id(project) + return to_jsonable( + get_client().issues.create(pid, subject, priority, status, issue_type, severity, **(fields or {})) + ) + + +@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. + resource = _get_by_ref("issue", project, ref) + resource.patch(list(fields.keys()), **fields) + return to_jsonable(get_client().issues.get(resource.id)) + + +@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.""" + client = get_client() + resource = client.issues.get(id) + resource.patch(list(fields.keys()), **fields) + return to_jsonable(client.issues.get(id)) + + +@mcp.tool() +def delete_issue(project: str | int, ref: int) -> dict[str, str]: + """Delete an issue identified by its per-project ref number.""" + resource = _get_by_ref("issue", project, ref) + resource.delete() + return {"status": "deleted", "ref": str(ref)} + + +@mcp.tool() +def delete_issue_by_id(id: int) -> dict[str, str]: # noqa: A002 + """Delete an issue by its database id. Secondary lookup - prefer `delete_issue` with a project + ref.""" + get_client().issues.delete(id) + return {"status": "deleted", "id": str(id)} + + +# --- Epics ------------------------------------------------------------------------------ + + +@mcp.tool() +def list_epics(project: str | int | None = None, filters: dict[str, Any] | None = None) -> 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. + """ + query = dict(filters or {}) + if project is not None: + query["project"] = _resolve_project_id(project) + return to_jsonable(get_client().epics.list(**_paginated(query))) + + +@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)) + + +@mcp.tool() +def get_epic_by_id(id: int) -> dict[str, Any]: # noqa: A002 + """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. + """ + return to_jsonable(get_client().epics.get(id)) + + +@mcp.tool() +def create_epic(project: str | int, subject: str, fields: dict[str, Any] | None = None) -> dict[str, Any]: + """Create an epic.""" + pid = _resolve_project_id(project) + return to_jsonable(get_client().epics.create(pid, subject, **(fields or {}))) + + +@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. + resource = _get_by_ref("epic", project, ref) + resource.patch(list(fields.keys()), **fields) + return to_jsonable(get_client().epics.get(resource.id)) + + +@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.""" + client = get_client() + resource = client.epics.get(id) + resource.patch(list(fields.keys()), **fields) + return to_jsonable(client.epics.get(id)) + + +@mcp.tool() +def delete_epic(project: str | int, ref: int) -> dict[str, str]: + """Delete an epic identified by its per-project ref number.""" + resource = _get_by_ref("epic", project, ref) + resource.delete() + return {"status": "deleted", "ref": str(ref)} + + +@mcp.tool() +def delete_epic_by_id(id: int) -> dict[str, str]: # noqa: A002 + """Delete an epic by its database id. Secondary lookup - prefer `delete_epic` with a project + ref.""" + get_client().epics.delete(id) + return {"status": "deleted", "id": str(id)} + + +@mcp.tool() +def link_epic_user_story(project: str | int, epic_ref: int, user_story_ref: int) -> dict[str, Any]: + """Link a user story to an epic, identifying both by their per-project ref numbers.""" + proj = _resolve_project(project) + epic = proj.get_epic_by_ref(epic_ref) + user_story = proj.get_userstory_by_ref(user_story_ref) + return to_jsonable(epic.add_related_user_story(user_story.id)) + + +@mcp.tool() +def link_epic_user_story_by_id(epic_id: int, user_story_id: int) -> dict[str, Any]: + """Link a user story to an epic by their database ids. + + Secondary lookup: prefer `link_epic_user_story` with a project + ref numbers. + """ + client = get_client() + epic = client.epics.get(epic_id) + return to_jsonable(epic.add_related_user_story(user_story_id)) + + +# --- Milestones (sprints) ----------------------------------------------------------------- + + +@mcp.tool() +def list_milestones(project: str | int, filters: dict[str, Any] | None = None) -> list[dict[str, Any]]: + """List milestones (sprints) of 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. + """ + pid = _resolve_project_id(project) + query = dict(filters or {}) + query["project"] = pid + return to_jsonable(get_client().milestones.list(**_paginated(query))) + + +@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)) + + +@mcp.tool() +def create_milestone( + project: str | int, + name: str, + estimated_start: str, + estimated_finish: str, + fields: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Create a milestone. Dates are ISO strings ('YYYY-MM-DD').""" + pid = _resolve_project_id(project) + return to_jsonable(get_client().milestones.create(pid, name, estimated_start, estimated_finish, **(fields or {}))) + + +@mcp.tool() +def delete_milestone(id: int) -> dict[str, str]: # noqa: A002 + """Delete a milestone by id.""" + get_client().milestones.delete(id) + return {"status": "deleted", "id": str(id)} + + +# --- Wiki pages ----------------------------------------------------------------------------- + + +@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. + + 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. + """ + pid = _resolve_project_id(project) + query = dict(filters or {}) + query["project"] = pid + return to_jsonable(get_client().wikipages.list(**_paginated(query))) + + +@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)) + + +@mcp.tool() +def create_wiki_page( + project: str | int, slug: str, content: str, fields: dict[str, Any] | None = None +) -> dict[str, Any]: + """Create a wiki page.""" + pid = _resolve_project_id(project) + return to_jsonable(get_client().wikipages.create(pid, slug, content, **(fields or {}))) + + +@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. + client = get_client() + resource = client.wikipages.get(id) + resource.patch(list(fields.keys()), **fields) + return to_jsonable(client.wikipages.get(id)) diff --git a/taiga/models/models.py b/taiga/models/models.py index 884477e..939df41 100644 --- a/taiga/models/models.py +++ b/taiga/models/models.py @@ -328,6 +328,19 @@ def list_user_stories(self, **queryparams): """ return UserStories(self.requester).list(epic=self.id, **queryparams) + def add_related_user_story(self, user_story_id, **attrs): + """ + Link an existing :class:`UserStory` to this epic. + + :param user_story_id: id of the :class:`UserStory` to link + :param attrs: other optional attributes of the relation + """ + attrs.update({"user_story": user_story_id}) + response = self.requester.post( + "/{endpoint}/{id}/related_userstories", endpoint=self.endpoint, id=self.id, payload=attrs + ) + return response.json() + def list_attachments(self): """ Get a list of :class:`EpicAttachment`. @@ -1356,11 +1369,13 @@ def add_membership(self, email, role, **attrs): """ return Memberships(self.requester).create(self.id, email, role, **attrs) - def list_memberships(self): + def list_memberships(self, **queryparams): """ Get the list of :class:`Membership` resources for the project. + + :param queryparams: optional query parameters (e.g. `page`, `page_size`) """ - return Memberships(self.requester).list(project=self.id) + return Memberships(self.requester).list(project=self.id, **queryparams) def add_user_story(self, subject, **attrs): """ diff --git a/tests/test_epics.py b/tests/test_epics.py index 3221500..acd3e72 100644 --- a/tests/test_epics.py +++ b/tests/test_epics.py @@ -83,3 +83,20 @@ def test_add_comment(self, mock_update): epic = Epic(rm, id=1) epic.add_comment("hola") mock_update.assert_called_with(comment="hola") + + +@patch("taiga.requestmaker.RequestMaker.post") +def test_add_related_user_story(mock_requestmaker_post): + mock_requestmaker_post.return_value = MockResponse(200, '{"id": 5, "epic": 1, "user_story": 10}') + rm = RequestMaker("/api/v1", "fakehost", "faketoken") + epic = Epic(rm, id=1) + + result = epic.add_related_user_story(10) + + mock_requestmaker_post.assert_called_with( + "/{endpoint}/{id}/related_userstories", + endpoint=Epic.endpoint, + id=epic.id, + payload={"user_story": 10}, + ) + assert result == {"id": 5, "epic": 1, "user_story": 10} diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py new file mode 100644 index 0000000..17e767b --- /dev/null +++ b/tests/test_mcp_server.py @@ -0,0 +1,1080 @@ +from __future__ import annotations + +from unittest.mock import MagicMock, call, patch + +import pytest + +from taiga.mcp_server import server + +_HISTORY_ENTRY = { + "user": {"pk": 1, "name": "tester"}, + "created_at": "2026-08-20T10:00:00+0000", + "comment": "hello", + "comment_html": "

hello

", + "delete_comment_date": None, + "type": 1, +} + + +# --- _resolve_project_id ----------------------------------------------------------------- + + +def test_resolve_project_id_with_int(): + assert server._resolve_project_id(42) == 42 + + +def test_resolve_project_id_with_numeric_string(): + assert server._resolve_project_id("42") == 42 + + +@patch("taiga.mcp_server.server.get_client") +def test_resolve_project_id_with_slug(mock_get_client): + mock_client = MagicMock() + mock_client.projects.get_by_slug.return_value = MagicMock(id=7) + mock_get_client.return_value = mock_client + + assert server._resolve_project_id("my-project") == 7 + + mock_client.projects.get_by_slug.assert_called_once_with("my-project") + + +# --- _resolve_project --------------------------------------------------------------------- + + +@patch("taiga.mcp_server.server.get_client") +def test_resolve_project_with_int(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock(id=42) + mock_client.projects.get.return_value = mock_project + mock_get_client.return_value = mock_client + + result = server._resolve_project(42) + + mock_client.projects.get.assert_called_once_with(42) + assert result is mock_project + + +@patch("taiga.mcp_server.server.get_client") +def test_resolve_project_with_numeric_string(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock(id=42) + mock_client.projects.get.return_value = mock_project + mock_get_client.return_value = mock_client + + result = server._resolve_project("42") + + mock_client.projects.get.assert_called_once_with(42) + assert result is mock_project + + +@patch("taiga.mcp_server.server.get_client") +def test_resolve_project_with_slug(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock(id=7, slug="my-project") + mock_client.projects.get_by_slug.return_value = mock_project + mock_get_client.return_value = mock_client + + result = server._resolve_project("my-project") + + mock_client.projects.get_by_slug.assert_called_once_with("my-project") + assert result is mock_project + + +# --- _get_by_ref ---------------------------------------------------------------------------- + + +@patch("taiga.mcp_server.server.get_client") +def test_get_by_ref_routes_every_entity_type(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_get_client.return_value = mock_client + + for entity_type, method_name in server._REF_METHOD.items(): + getattr(mock_project, method_name).return_value = {"ref": 45634} + + result = server._get_by_ref(entity_type, 1, 45634) + + getattr(mock_project, method_name).assert_called_once_with(45634) + assert result == {"ref": 45634} + + +# --- _paginated --------------------------------------------------------------------------- + + +def test_paginated_defaults_page_and_page_size(): + assert server._paginated({}) == {"page": 1, "page_size": 100} + + +def test_paginated_preserves_other_keys(): + assert server._paginated({"project": 1}) == {"project": 1, "page": 1, "page_size": 100} + + +def test_paginated_does_not_override_explicit_page(): + assert server._paginated({"page": 3}) == {"page": 3, "page_size": 100} + + +def test_paginated_does_not_override_explicit_page_size(): + assert server._paginated({"page_size": 25}) == {"page": 1, "page_size": 25} + + +def test_paginated_strips_pagination_override(): + # `pagination=False` is a ListResource.list() kwarg that disables the bound entirely - + # a caller must not be able to pass it through `filters`. + assert server._paginated({"pagination": False}) == {"page": 1, "page_size": 100} + + +def test_paginated_normalizes_falsy_page(): + assert server._paginated({"page": None}) == {"page": 1, "page_size": 100} + assert server._paginated({"page": 0}) == {"page": 1, "page_size": 100} + + +def test_paginated_normalizes_falsy_page_size(): + assert server._paginated({"page_size": None}) == {"page": 1, "page_size": 100} + assert server._paginated({"page_size": 0}) == {"page": 1, "page_size": 100} + + +# --- whoami / projects / search ---------------------------------------------------------- + + +@patch("taiga.mcp_server.server.get_client") +def test_whoami(mock_get_client): + mock_client = MagicMock() + mock_client.me.return_value = {"id": 1, "username": "tester"} + mock_get_client.return_value = mock_client + + assert server.whoami() == {"id": 1, "username": "tester"} + + +@patch("taiga.mcp_server.server.get_client") +def test_list_projects_without_member(mock_get_client): + mock_client = MagicMock() + mock_client.projects.list.return_value = [{"id": 1}] + mock_get_client.return_value = mock_client + + result = server.list_projects() + + mock_client.projects.list.assert_called_once_with(page=1, page_size=100) + assert result == [{"id": 1}] + + +@patch("taiga.mcp_server.server.get_client") +def test_list_projects_with_member(mock_get_client): + mock_client = MagicMock() + mock_client.projects.list.return_value = [{"id": 1}] + mock_get_client.return_value = mock_client + + server.list_projects(member=9, filters={"is_backlog_activated": True}) + + mock_client.projects.list.assert_called_once_with(is_backlog_activated=True, member=9, page=1, page_size=100) + + +@patch("taiga.mcp_server.server.get_client") +def test_list_projects_explicit_pagination_not_overridden(mock_get_client): + mock_client = MagicMock() + mock_client.projects.list.return_value = [{"id": 1}] + mock_get_client.return_value = mock_client + + server.list_projects(filters={"page": 3, "page_size": 25, "order_by": "-created_date"}) + + mock_client.projects.list.assert_called_once_with(page=3, page_size=25, order_by="-created_date") + + +@patch("taiga.mcp_server.server.get_client") +def test_get_project_by_id(mock_get_client): + mock_client = MagicMock() + mock_client.projects.get.return_value = {"id": 1} + mock_get_client.return_value = mock_client + + result = server.get_project(1) + + mock_client.projects.get.assert_called_once_with(1) + assert result == {"id": 1} + + +@patch("taiga.mcp_server.server.get_client") +def test_get_project_by_slug(mock_get_client): + mock_client = MagicMock() + mock_client.projects.get_by_slug.return_value = {"id": 1, "slug": "my-project"} + mock_get_client.return_value = mock_client + + result = server.get_project("my-project") + + mock_client.projects.get_by_slug.assert_called_once_with("my-project") + assert result == {"id": 1, "slug": "my-project"} + + +@patch("taiga.mcp_server.server.get_client") +def test_search(mock_get_client): + mock_client = MagicMock() + mock_result = MagicMock() + mock_result.count = 2 + mock_result.user_stories = [{"id": 1}] + mock_result.tasks = [] + mock_result.issues = [] + mock_result.epics = [] + mock_result.wikipages = [{"id": 2}] + mock_client.search.return_value = mock_result + mock_get_client.return_value = mock_client + + result = server.search(1, "keyword") + + mock_client.search.assert_called_once_with(1, "keyword") + assert result == { + "count": 2, + "user_stories": [{"id": 1}], + "tasks": [], + "issues": [], + "epics": [], + "wikipages": [{"id": 2}], + } + + +# --- memberships -------------------------------------------------------------------------- + + +@patch("taiga.mcp_server.server._resolve_project") +def test_list_memberships_no_filters(mock_resolve_project): + mock_project = MagicMock() + mock_project.list_memberships.return_value = [{"username": "yakky", "user_email": "i.spalletti@nephila.digital"}] + mock_resolve_project.return_value = mock_project + + result = server.list_memberships(1) + + mock_project.list_memberships.assert_called_once_with(page=1, page_size=100) + assert result == [{"username": "yakky", "user_email": "i.spalletti@nephila.digital"}] + + +@patch("taiga.mcp_server.server._resolve_project") +def test_list_memberships_with_filters(mock_resolve_project): + mock_project = MagicMock() + mock_project.list_memberships.return_value = [] + mock_resolve_project.return_value = mock_project + + server.list_memberships(1, filters={"page": 2}) + + mock_project.list_memberships.assert_called_once_with(page=2, page_size=100) + + +# --- add_comment --------------------------------------------------------------------------- + + +@patch("taiga.mcp_server.server.get_client") +def test_add_comment_routes_every_entity_type(mock_get_client): + # CommentableResource.add_comment() delegates to update(), which returns the stale + # pre-comment resource (only `version` is refreshed) - not the new comment. The tool + # must not serialize that stale resource; it returns an explicit acknowledgement. + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_get_client.return_value = mock_client + + for entity_type, method_name in server._REF_METHOD.items(): + resource = getattr(mock_project, method_name).return_value + + result = server.add_comment(entity_type, 1, 45634, "hello") + + getattr(mock_project, method_name).assert_called_once_with(45634) + resource.add_comment.assert_called_once_with("hello") + assert result == {"status": "commented", "ref": "45634", "comment": "hello"} + + +@patch("taiga.mcp_server.server.get_client") +def test_add_comment_by_id_routes_every_entity_type(mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + for entity_type, attr in server._ENTITY_ATTR.items(): + resource = getattr(mock_client, attr).get.return_value + + result = server.add_comment_by_id(entity_type, 1, "hello") + + getattr(mock_client, attr).get.assert_called_once_with(1) + resource.add_comment.assert_called_once_with("hello") + assert result == {"status": "commented", "id": "1", "comment": "hello"} + + +# --- get_history ----------------------------------------------------------------------- + + +@patch("taiga.mcp_server.server.get_client") +def test_get_history_resolves_ref_for_non_wiki_types(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + resolved = MagicMock(id=99) + mock_project.get_userstory_by_ref.return_value = resolved + mock_client.history.user_story.get.return_value = [_HISTORY_ENTRY] + mock_get_client.return_value = mock_client + + result = server.get_history("user_story", 45634, 1) + + mock_project.get_userstory_by_ref.assert_called_once_with(45634) + mock_client.history.user_story.get.assert_called_once_with(99) + assert result == [_HISTORY_ENTRY] + + +@patch("taiga.mcp_server.server.get_client") +def test_get_history_routes_every_ref_entity_type(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_get_client.return_value = mock_client + + for entity_type, method_name in server._REF_METHOD.items(): + resolved = MagicMock(id=1) + getattr(mock_project, method_name).return_value = resolved + getattr(mock_client.history, entity_type).get.return_value = [] + + result = server.get_history(entity_type, 45634, 1) + + getattr(mock_project, method_name).assert_called_once_with(45634) + getattr(mock_client.history, entity_type).get.assert_called_once_with(1) + assert result == [] + + +@patch("taiga.mcp_server.server.get_client") +def test_get_history_wiki_uses_literal_id(mock_get_client): + mock_client = MagicMock() + mock_client.history.wiki.get.return_value = [_HISTORY_ENTRY] + mock_get_client.return_value = mock_client + + result = server.get_history("wiki", 1) + + mock_client.history.wiki.get.assert_called_once_with(1) + mock_client.projects.get.assert_not_called() + assert result == [_HISTORY_ENTRY] + + +def test_get_history_requires_project_for_non_wiki(): + with pytest.raises(ValueError, match="project"): + server.get_history("issue", 1) + + +@patch("taiga.mcp_server.server.get_client") +def test_get_history_by_id_routes_every_entity_type(mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + for entity_type in server._HISTORY_ENTITY_TYPES: + getattr(mock_client.history, entity_type).get.return_value = [] + result = server.get_history_by_id(entity_type, 1) + getattr(mock_client.history, entity_type).get.assert_called_once_with(1) + assert result == [] + + +# --- custom attribute values ------------------------------------------------------------- + + +@patch("taiga.mcp_server.server.get_client") +def test_get_custom_attributes_values_routes_every_entity_type(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_get_client.return_value = mock_client + + for entity_type, method_name in server._REF_METHOD.items(): + resource = getattr(mock_project, method_name).return_value + resource.get_attributes.return_value = {"attributes_values": {"1": "x"}, "version": 1} + + result = server.get_custom_attributes_values(entity_type, 1, 45634) + + getattr(mock_project, method_name).assert_called_once_with(45634) + resource.get_attributes.assert_called_once_with() + assert result == {"attributes_values": {"1": "x"}, "version": 1} + + +@patch("taiga.mcp_server.server.get_client") +def test_get_custom_attributes_values_by_id_routes_every_entity_type(mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + for entity_type, attr in server._ENTITY_ATTR.items(): + resource = getattr(mock_client, attr).get.return_value + resource.get_attributes.return_value = {"attributes_values": {"1": "x"}, "version": 1} + + result = server.get_custom_attributes_values_by_id(entity_type, 1) + + getattr(mock_client, attr).get.assert_called_once_with(1) + assert result == {"attributes_values": {"1": "x"}, "version": 1} + + +@patch("taiga.mcp_server.server.get_client") +def test_set_custom_attribute_value_routes_every_entity_type(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_get_client.return_value = mock_client + + for entity_type, method_name in server._REF_METHOD.items(): + resource = getattr(mock_project, method_name).return_value + resource.set_attribute.return_value = {"attributes_values": {"10": "NPH-INT"}, "version": 2} + + result = server.set_custom_attribute_value(entity_type, 1, 45634, 10, "NPH-INT", 1) + + getattr(mock_project, method_name).assert_called_once_with(45634) + resource.set_attribute.assert_called_once_with(10, "NPH-INT", version=1) + assert result == {"attributes_values": {"10": "NPH-INT"}, "version": 2} + + +@patch("taiga.mcp_server.server.get_client") +def test_set_custom_attribute_value_by_id_routes_every_entity_type(mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + for entity_type, attr in server._ENTITY_ATTR.items(): + resource = getattr(mock_client, attr).get.return_value + resource.set_attribute.return_value = {"attributes_values": {"10": "NPH-INT"}, "version": 2} + + result = server.set_custom_attribute_value_by_id(entity_type, 1, 10, "NPH-INT", 1) + + getattr(mock_client, attr).get.assert_called_once_with(1) + resource.set_attribute.assert_called_once_with(10, "NPH-INT", version=1) + assert result == {"attributes_values": {"10": "NPH-INT"}, "version": 2} + + +# --- User stories ----------------------------------------------------------------------- + + +@patch("taiga.mcp_server.server.get_client") +def test_list_user_stories_no_project(mock_get_client): + mock_client = MagicMock() + mock_client.user_stories.list.return_value = [{"id": 1}] + mock_get_client.return_value = mock_client + + result = server.list_user_stories() + + mock_client.user_stories.list.assert_called_once_with(page=1, page_size=100) + assert result == [{"id": 1}] + + +@patch("taiga.mcp_server.server.get_client") +def test_list_user_stories_with_project(mock_get_client): + mock_client = MagicMock() + mock_client.user_stories.list.return_value = [{"id": 1}] + mock_get_client.return_value = mock_client + + server.list_user_stories(project=1, filters={"status": 2}) + + mock_client.user_stories.list.assert_called_once_with(status=2, project=1, page=1, page_size=100) + + +@patch("taiga.mcp_server.server.get_client") +def test_get_user_story(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} + mock_get_client.return_value = mock_client + + result = server.get_user_story(1, 45634) + + mock_client.projects.get.assert_called_once_with(1) + mock_project.get_userstory_by_ref.assert_called_once_with(45634) + assert result == {"id": 1, "ref": 45634} + + +@patch("taiga.mcp_server.server.get_client") +def test_get_user_story_by_id(mock_get_client): + mock_client = MagicMock() + mock_client.user_stories.get.return_value = {"id": 1} + mock_get_client.return_value = mock_client + + result = server.get_user_story_by_id(1) + + mock_client.user_stories.get.assert_called_once_with(1) + assert result == {"id": 1} + + +@patch("taiga.mcp_server.server.get_client") +def test_create_user_story(mock_get_client): + mock_client = MagicMock() + mock_client.user_stories.create.return_value = {"id": 1, "subject": "New story"} + mock_get_client.return_value = mock_client + + result = server.create_user_story(1, "New story", fields={"points": {"1": 2}}) + + mock_client.user_stories.create.assert_called_once_with(1, "New story", points={"1": 2}) + assert result == {"id": 1, "subject": "New story"} + + +@patch("taiga.mcp_server.server.get_client") +def test_update_user_story(mock_get_client): + # InstanceResource.patch() only refreshes `version` on the local object, not the other + # fields the server actually applied - the tool must re-fetch before serializing. + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_resource = MagicMock(id=1) + mock_project.get_userstory_by_ref.return_value = mock_resource + mock_client.user_stories.get.return_value = {"id": 1, "subject": "Updated"} + mock_get_client.return_value = mock_client + + result = server.update_user_story(1, 45634, {"subject": "Updated"}) + + mock_project.get_userstory_by_ref.assert_called_once_with(45634) + mock_resource.patch.assert_called_once_with(["subject"], subject="Updated") + mock_client.user_stories.get.assert_called_once_with(1) + assert result == {"id": 1, "subject": "Updated"} + + +@patch("taiga.mcp_server.server.get_client") +def test_update_user_story_by_id(mock_get_client): + mock_client = MagicMock() + mock_resource = MagicMock(id=1) + mock_client.user_stories.get.side_effect = [mock_resource, {"id": 1, "subject": "Updated"}] + mock_get_client.return_value = mock_client + + result = server.update_user_story_by_id(1, {"subject": "Updated"}) + + mock_resource.patch.assert_called_once_with(["subject"], subject="Updated") + mock_client.user_stories.get.assert_has_calls([call(1), call(1)]) + assert result == {"id": 1, "subject": "Updated"} + + +@patch("taiga.mcp_server.server.get_client") +def test_delete_user_story(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_resource = MagicMock() + mock_project.get_userstory_by_ref.return_value = mock_resource + mock_get_client.return_value = mock_client + + result = server.delete_user_story(1, 45634) + + mock_project.get_userstory_by_ref.assert_called_once_with(45634) + mock_resource.delete.assert_called_once_with() + assert result == {"status": "deleted", "ref": "45634"} + + +@patch("taiga.mcp_server.server.get_client") +def test_delete_user_story_by_id(mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + result = server.delete_user_story_by_id(1) + + mock_client.user_stories.delete.assert_called_once_with(1) + assert result == {"status": "deleted", "id": "1"} + + +# --- Tasks ------------------------------------------------------------------------------ + + +@patch("taiga.mcp_server.server.get_client") +def test_list_tasks_no_filters(mock_get_client): + mock_client = MagicMock() + mock_client.tasks.list.return_value = [{"id": 1}] + mock_get_client.return_value = mock_client + + result = server.list_tasks() + + mock_client.tasks.list.assert_called_once_with(page=1, page_size=100) + assert result == [{"id": 1}] + + +@patch("taiga.mcp_server.server.get_client") +def test_list_tasks_with_project_and_user_story(mock_get_client): + mock_client = MagicMock() + mock_client.tasks.list.return_value = [{"id": 1}] + mock_get_client.return_value = mock_client + + server.list_tasks(project=1, user_story=5) + + mock_client.tasks.list.assert_called_once_with(project=1, user_story=5, page=1, page_size=100) + + +@patch("taiga.mcp_server.server.get_client") +def test_get_task(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 = {"id": 1, "ref": 45634} + mock_get_client.return_value = mock_client + + result = server.get_task(1, 45634) + + mock_project.get_task_by_ref.assert_called_once_with(45634) + assert result == {"id": 1, "ref": 45634} + + +@patch("taiga.mcp_server.server.get_client") +def test_get_task_by_id(mock_get_client): + mock_client = MagicMock() + mock_client.tasks.get.return_value = {"id": 1} + mock_get_client.return_value = mock_client + + result = server.get_task_by_id(1) + + mock_client.tasks.get.assert_called_once_with(1) + assert result == {"id": 1} + + +@patch("taiga.mcp_server.server.get_client") +def test_create_task(mock_get_client): + mock_client = MagicMock() + mock_client.tasks.create.return_value = {"id": 1, "subject": "New task"} + mock_get_client.return_value = mock_client + + result = server.create_task(1, "New task", 3, fields={"user_story": 2}) + + mock_client.tasks.create.assert_called_once_with(1, "New task", 3, user_story=2) + assert result == {"id": 1, "subject": "New task"} + + +@patch("taiga.mcp_server.server.get_client") +def test_update_task(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_resource = MagicMock(id=1) + mock_project.get_task_by_ref.return_value = mock_resource + mock_client.tasks.get.return_value = {"id": 1, "subject": "Updated"} + mock_get_client.return_value = mock_client + + result = server.update_task(1, 45634, {"subject": "Updated"}) + + mock_project.get_task_by_ref.assert_called_once_with(45634) + mock_resource.patch.assert_called_once_with(["subject"], subject="Updated") + mock_client.tasks.get.assert_called_once_with(1) + assert result == {"id": 1, "subject": "Updated"} + + +@patch("taiga.mcp_server.server.get_client") +def test_update_task_by_id(mock_get_client): + mock_client = MagicMock() + mock_resource = MagicMock(id=1) + mock_client.tasks.get.side_effect = [mock_resource, {"id": 1, "subject": "Updated"}] + mock_get_client.return_value = mock_client + + result = server.update_task_by_id(1, {"subject": "Updated"}) + + mock_resource.patch.assert_called_once_with(["subject"], subject="Updated") + mock_client.tasks.get.assert_has_calls([call(1), call(1)]) + assert result == {"id": 1, "subject": "Updated"} + + +@patch("taiga.mcp_server.server.get_client") +def test_delete_task(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_resource = MagicMock() + mock_project.get_task_by_ref.return_value = mock_resource + mock_get_client.return_value = mock_client + + result = server.delete_task(1, 45634) + + mock_project.get_task_by_ref.assert_called_once_with(45634) + mock_resource.delete.assert_called_once_with() + assert result == {"status": "deleted", "ref": "45634"} + + +@patch("taiga.mcp_server.server.get_client") +def test_delete_task_by_id(mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + result = server.delete_task_by_id(1) + + mock_client.tasks.delete.assert_called_once_with(1) + assert result == {"status": "deleted", "id": "1"} + + +# --- Issues ----------------------------------------------------------------------------- + + +@patch("taiga.mcp_server.server.get_client") +def test_list_issues_no_project(mock_get_client): + mock_client = MagicMock() + mock_client.issues.list.return_value = [{"id": 1}] + mock_get_client.return_value = mock_client + + result = server.list_issues() + + mock_client.issues.list.assert_called_once_with(page=1, page_size=100) + assert result == [{"id": 1}] + + +@patch("taiga.mcp_server.server.get_client") +def test_list_issues_with_project(mock_get_client): + mock_client = MagicMock() + mock_client.issues.list.return_value = [{"id": 1}] + mock_get_client.return_value = mock_client + + server.list_issues(project=1) + + mock_client.issues.list.assert_called_once_with(project=1, page=1, page_size=100) + + +@patch("taiga.mcp_server.server.get_client") +def test_list_issues_explicit_pagination_not_overridden(mock_get_client): + mock_client = MagicMock() + mock_client.issues.list.return_value = [{"id": 1}] + mock_get_client.return_value = mock_client + + server.list_issues(project=1, filters={"page": 1, "page_size": 2, "order_by": "-created_date"}) + + mock_client.issues.list.assert_called_once_with(project=1, page=1, page_size=2, order_by="-created_date") + + +@patch("taiga.mcp_server.server.get_client") +def test_get_issue(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} + mock_get_client.return_value = mock_client + + result = server.get_issue(1, 45634) + + mock_project.get_issue_by_ref.assert_called_once_with(45634) + assert result == {"id": 1, "ref": 45634} + + +@patch("taiga.mcp_server.server.get_client") +def test_get_issue_by_id(mock_get_client): + mock_client = MagicMock() + mock_client.issues.get.return_value = {"id": 1} + mock_get_client.return_value = mock_client + + result = server.get_issue_by_id(1) + + mock_client.issues.get.assert_called_once_with(1) + assert result == {"id": 1} + + +@patch("taiga.mcp_server.server.get_client") +def test_create_issue(mock_get_client): + mock_client = MagicMock() + mock_client.issues.create.return_value = {"id": 1, "subject": "New issue"} + mock_get_client.return_value = mock_client + + result = server.create_issue(1, "New issue", 2, 3, 4, 5, fields={"description": "oops"}) + + mock_client.issues.create.assert_called_once_with(1, "New issue", 2, 3, 4, 5, description="oops") + assert result == {"id": 1, "subject": "New issue"} + + +@patch("taiga.mcp_server.server.get_client") +def test_update_issue(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_resource = MagicMock(id=1) + mock_project.get_issue_by_ref.return_value = mock_resource + mock_client.issues.get.return_value = {"id": 1, "subject": "Updated"} + mock_get_client.return_value = mock_client + + result = server.update_issue(1, 45634, {"subject": "Updated"}) + + mock_project.get_issue_by_ref.assert_called_once_with(45634) + mock_resource.patch.assert_called_once_with(["subject"], subject="Updated") + mock_client.issues.get.assert_called_once_with(1) + assert result == {"id": 1, "subject": "Updated"} + + +@patch("taiga.mcp_server.server.get_client") +def test_update_issue_by_id(mock_get_client): + mock_client = MagicMock() + mock_resource = MagicMock(id=1) + mock_client.issues.get.side_effect = [mock_resource, {"id": 1, "subject": "Updated"}] + mock_get_client.return_value = mock_client + + result = server.update_issue_by_id(1, {"subject": "Updated"}) + + mock_resource.patch.assert_called_once_with(["subject"], subject="Updated") + mock_client.issues.get.assert_has_calls([call(1), call(1)]) + assert result == {"id": 1, "subject": "Updated"} + + +@patch("taiga.mcp_server.server.get_client") +def test_delete_issue(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_resource = MagicMock() + mock_project.get_issue_by_ref.return_value = mock_resource + mock_get_client.return_value = mock_client + + result = server.delete_issue(1, 45634) + + mock_project.get_issue_by_ref.assert_called_once_with(45634) + mock_resource.delete.assert_called_once_with() + assert result == {"status": "deleted", "ref": "45634"} + + +@patch("taiga.mcp_server.server.get_client") +def test_delete_issue_by_id(mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + result = server.delete_issue_by_id(1) + + mock_client.issues.delete.assert_called_once_with(1) + assert result == {"status": "deleted", "id": "1"} + + +# --- Epics ------------------------------------------------------------------------------ + + +@patch("taiga.mcp_server.server.get_client") +def test_list_epics_no_project(mock_get_client): + mock_client = MagicMock() + mock_client.epics.list.return_value = [{"id": 1}] + mock_get_client.return_value = mock_client + + result = server.list_epics() + + mock_client.epics.list.assert_called_once_with(page=1, page_size=100) + assert result == [{"id": 1}] + + +@patch("taiga.mcp_server.server.get_client") +def test_list_epics_with_project(mock_get_client): + mock_client = MagicMock() + mock_client.epics.list.return_value = [{"id": 1}] + mock_get_client.return_value = mock_client + + server.list_epics(project=1) + + mock_client.epics.list.assert_called_once_with(project=1, page=1, page_size=100) + + +@patch("taiga.mcp_server.server.get_client") +def test_get_epic(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 = {"id": 1, "ref": 45634} + mock_get_client.return_value = mock_client + + result = server.get_epic(1, 45634) + + mock_project.get_epic_by_ref.assert_called_once_with(45634) + assert result == {"id": 1, "ref": 45634} + + +@patch("taiga.mcp_server.server.get_client") +def test_get_epic_by_id(mock_get_client): + mock_client = MagicMock() + mock_client.epics.get.return_value = {"id": 1} + mock_get_client.return_value = mock_client + + result = server.get_epic_by_id(1) + + mock_client.epics.get.assert_called_once_with(1) + assert result == {"id": 1} + + +@patch("taiga.mcp_server.server.get_client") +def test_create_epic(mock_get_client): + mock_client = MagicMock() + mock_client.epics.create.return_value = {"id": 1, "subject": "New epic"} + mock_get_client.return_value = mock_client + + result = server.create_epic(1, "New epic") + + mock_client.epics.create.assert_called_once_with(1, "New epic") + assert result == {"id": 1, "subject": "New epic"} + + +@patch("taiga.mcp_server.server.get_client") +def test_update_epic(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_resource = MagicMock(id=1) + mock_project.get_epic_by_ref.return_value = mock_resource + mock_client.epics.get.return_value = {"id": 1, "subject": "Updated"} + mock_get_client.return_value = mock_client + + result = server.update_epic(1, 45634, {"subject": "Updated"}) + + mock_project.get_epic_by_ref.assert_called_once_with(45634) + mock_resource.patch.assert_called_once_with(["subject"], subject="Updated") + mock_client.epics.get.assert_called_once_with(1) + assert result == {"id": 1, "subject": "Updated"} + + +@patch("taiga.mcp_server.server.get_client") +def test_update_epic_by_id(mock_get_client): + mock_client = MagicMock() + mock_resource = MagicMock(id=1) + mock_client.epics.get.side_effect = [mock_resource, {"id": 1, "subject": "Updated"}] + mock_get_client.return_value = mock_client + + result = server.update_epic_by_id(1, {"subject": "Updated"}) + + mock_resource.patch.assert_called_once_with(["subject"], subject="Updated") + mock_client.epics.get.assert_has_calls([call(1), call(1)]) + assert result == {"id": 1, "subject": "Updated"} + + +@patch("taiga.mcp_server.server.get_client") +def test_delete_epic(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_resource = MagicMock() + mock_project.get_epic_by_ref.return_value = mock_resource + mock_get_client.return_value = mock_client + + result = server.delete_epic(1, 45634) + + mock_project.get_epic_by_ref.assert_called_once_with(45634) + mock_resource.delete.assert_called_once_with() + assert result == {"status": "deleted", "ref": "45634"} + + +@patch("taiga.mcp_server.server.get_client") +def test_delete_epic_by_id(mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + result = server.delete_epic_by_id(1) + + mock_client.epics.delete.assert_called_once_with(1) + assert result == {"status": "deleted", "id": "1"} + + +# --- epic/user-story linking ----------------------------------------------------------- + + +@patch("taiga.mcp_server.server.get_client") +def test_link_epic_user_story(mock_get_client): + mock_client = MagicMock() + mock_project = MagicMock() + mock_client.projects.get.return_value = mock_project + mock_epic = MagicMock(id=1) + mock_us = MagicMock(id=10) + mock_project.get_epic_by_ref.return_value = mock_epic + mock_project.get_userstory_by_ref.return_value = mock_us + mock_epic.add_related_user_story.return_value = {"id": 5, "epic": 1, "user_story": 10} + mock_get_client.return_value = mock_client + + result = server.link_epic_user_story(1, 42, 45634) + + mock_project.get_epic_by_ref.assert_called_once_with(42) + mock_project.get_userstory_by_ref.assert_called_once_with(45634) + mock_epic.add_related_user_story.assert_called_once_with(10) + assert result == {"id": 5, "epic": 1, "user_story": 10} + + +@patch("taiga.mcp_server.server.get_client") +def test_link_epic_user_story_by_id(mock_get_client): + mock_client = MagicMock() + mock_epic = MagicMock() + mock_client.epics.get.return_value = mock_epic + mock_epic.add_related_user_story.return_value = {"id": 5, "epic": 1, "user_story": 10} + mock_get_client.return_value = mock_client + + result = server.link_epic_user_story_by_id(1, 10) + + mock_client.epics.get.assert_called_once_with(1) + mock_epic.add_related_user_story.assert_called_once_with(10) + assert result == {"id": 5, "epic": 1, "user_story": 10} + + +# --- Milestones ------------------------------------------------------------------------ + + +@patch("taiga.mcp_server.server.get_client") +def test_list_milestones(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(1, filters={"closed": False}) + + mock_client.milestones.list.assert_called_once_with(closed=False, project=1, page=1, page_size=100) + assert result == [{"id": 1}] + + +@patch("taiga.mcp_server.server.get_client") +def test_get_milestone(mock_get_client): + mock_client = MagicMock() + mock_client.milestones.get.return_value = {"id": 1} + mock_get_client.return_value = mock_client + + result = server.get_milestone(1) + + mock_client.milestones.get.assert_called_once_with(1) + assert result == {"id": 1} + + +@patch("taiga.mcp_server.server.get_client") +def test_create_milestone(mock_get_client): + mock_client = MagicMock() + mock_client.milestones.create.return_value = {"id": 1, "name": "Sprint 1"} + mock_get_client.return_value = mock_client + + result = server.create_milestone(1, "Sprint 1", "2026-01-01", "2026-01-15") + + mock_client.milestones.create.assert_called_once_with(1, "Sprint 1", "2026-01-01", "2026-01-15") + assert result == {"id": 1, "name": "Sprint 1"} + + +@patch("taiga.mcp_server.server.get_client") +def test_delete_milestone(mock_get_client): + mock_client = MagicMock() + mock_get_client.return_value = mock_client + + result = server.delete_milestone(1) + + mock_client.milestones.delete.assert_called_once_with(1) + assert result == {"status": "deleted", "id": "1"} + + +# --- Wiki pages ------------------------------------------------------------------------ + + +@patch("taiga.mcp_server.server.get_client") +def test_list_wiki_pages(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(1, filters={"slug": "home"}) + + mock_client.wikipages.list.assert_called_once_with(slug="home", project=1, page=1, page_size=100) + assert result == [{"id": 1}] + + +@patch("taiga.mcp_server.server.get_client") +def test_get_wiki_page(mock_get_client): + mock_client = MagicMock() + mock_client.wikipages.get.return_value = {"id": 1} + mock_get_client.return_value = mock_client + + result = server.get_wiki_page(1) + + mock_client.wikipages.get.assert_called_once_with(1) + assert result == {"id": 1} + + +@patch("taiga.mcp_server.server.get_client") +def test_create_wiki_page(mock_get_client): + mock_client = MagicMock() + mock_client.wikipages.create.return_value = {"id": 1, "slug": "home"} + mock_get_client.return_value = mock_client + + result = server.create_wiki_page(1, "home", "Welcome") + + mock_client.wikipages.create.assert_called_once_with(1, "home", "Welcome") + assert result == {"id": 1, "slug": "home"} + + +@patch("taiga.mcp_server.server.get_client") +def test_update_wiki_page(mock_get_client): + mock_client = MagicMock() + mock_resource = MagicMock(id=1) + mock_client.wikipages.get.side_effect = [mock_resource, {"id": 1, "content": "Updated"}] + mock_get_client.return_value = mock_client + + result = server.update_wiki_page(1, {"content": "Updated"}) + + 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"} diff --git a/tests/test_mcp_server_auth.py b/tests/test_mcp_server_auth.py new file mode 100644 index 0000000..8c22a42 --- /dev/null +++ b/tests/test_mcp_server_auth.py @@ -0,0 +1,99 @@ +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest + +from taiga.mcp_server import auth + +# --- build_client ------------------------------------------------------------------------- + + +@patch("taiga.mcp_server.auth.TaigaAPI") +def test_build_client_with_token(mock_taiga_api): + credentials = auth.Credentials(host="https://example.com", token="tok", token_type="Bearer", tls_verify=False) + + result = auth.build_client(credentials) + + mock_taiga_api.assert_called_once_with( + host="https://example.com", token="tok", token_type="Bearer", tls_verify=False + ) + assert result is mock_taiga_api.return_value + + +@patch("taiga.mcp_server.auth.TaigaAPI") +def test_build_client_prefers_token_over_username_password(mock_taiga_api): + credentials = auth.Credentials(token="tok", username="alice", password="secret") + + auth.build_client(credentials) + + mock_taiga_api.assert_called_once_with( + host=auth.DEFAULT_HOST, token="tok", token_type=auth.DEFAULT_TOKEN_TYPE, tls_verify=True + ) + mock_taiga_api.return_value.auth.assert_not_called() + + +@patch("taiga.mcp_server.auth.TaigaAPI") +def test_build_client_with_username_password(mock_taiga_api): + mock_api = MagicMock() + mock_taiga_api.return_value = mock_api + credentials = auth.Credentials(host="https://example.com", username="alice", password="secret", tls_verify=True) + + result = auth.build_client(credentials) + + mock_taiga_api.assert_called_once_with(host="https://example.com", tls_verify=True) + mock_api.auth.assert_called_once_with("alice", "secret") + assert result is mock_api + + +def test_build_client_without_credentials_raises(): + credentials = auth.Credentials() + + with pytest.raises(auth.ConfigError, match="provide a token"): + auth.build_client(credentials) + + +def test_build_client_with_only_username_raises(): + credentials = auth.Credentials(username="alice") + + with pytest.raises(auth.ConfigError, match="provide a token"): + auth.build_client(credentials) + + +# --- configure ------------------------------------------------------------------------------ + + +@patch("taiga.mcp_server.auth._client", "stale-client") +@patch("taiga.mcp_server.auth._credentials", None) +def test_configure_stores_credentials_and_resets_client(): + credentials = auth.Credentials(token="tok") + + auth.configure(credentials) + + assert auth._credentials is credentials + assert auth._client is None + + +# --- get_client ----------------------------------------------------------------------------- + + +@patch("taiga.mcp_server.auth._client", None) +@patch("taiga.mcp_server.auth._credentials", None) +def test_get_client_without_configuration_raises(): + with pytest.raises(auth.ConfigError, match="not been configured"): + auth.get_client() + + +@patch("taiga.mcp_server.auth.build_client") +@patch("taiga.mcp_server.auth._client", None) +@patch("taiga.mcp_server.auth._credentials") +def test_get_client_builds_once_and_caches(mock_credentials, mock_build_client): + mock_client = MagicMock() + mock_build_client.return_value = mock_client + + first = auth.get_client() + second = auth.get_client() + + assert first is mock_client + assert second is mock_client + mock_build_client.assert_called_once_with(mock_credentials) diff --git a/tests/test_mcp_server_cli.py b/tests/test_mcp_server_cli.py new file mode 100644 index 0000000..f7dd9f9 --- /dev/null +++ b/tests/test_mcp_server_cli.py @@ -0,0 +1,209 @@ +from __future__ import annotations + +import json +import os +from unittest.mock import patch + +from typer.testing import CliRunner + +from taiga.mcp_server import cli + +runner = CliRunner() + +# --- _env_bool ------------------------------------------------------------------------------ + + +def test_env_bool_default_when_unset(): + with patch.dict("os.environ", {}, clear=False): + os.environ.pop("TAIGA_TLS_VERIFY", None) + assert cli._env_bool("TAIGA_TLS_VERIFY", True) is True + assert cli._env_bool("TAIGA_TLS_VERIFY", False) is False + + +def test_env_bool_falsy_values(): + for value in ("0", "false", "No", "OFF", " off "): + with patch.dict("os.environ", {"TAIGA_TLS_VERIFY": value}): + assert cli._env_bool("TAIGA_TLS_VERIFY", True) is False + + +def test_env_bool_truthy_values(): + for value in ("1", "true", "yes", "anything-else"): + with patch.dict("os.environ", {"TAIGA_TLS_VERIFY": value}): + assert cli._env_bool("TAIGA_TLS_VERIFY", False) is True + + +# --- serve ------------------------------------------------------------------------------ + + +@patch("taiga.mcp_server.server.mcp") +@patch("taiga.mcp_server.cli.configure") +def test_serve_configures_from_token_argv(mock_configure, mock_mcp): + result = runner.invoke(cli.app, ["serve", "--host", "https://example.com", "--token", "tok", "--no-tls-verify"]) + + assert result.exit_code == 0 + mock_configure.assert_called_once() + credentials = mock_configure.call_args.args[0] + assert credentials.host == "https://example.com" + assert credentials.token == "tok" + assert credentials.tls_verify is False + mock_mcp.run.assert_called_once_with(transport="stdio") + + +@patch("taiga.mcp_server.server.mcp") +@patch("taiga.mcp_server.cli.configure") +def test_serve_configures_from_username_password_argv(mock_configure, mock_mcp): + runner.invoke(cli.app, ["serve", "--username", "alice", "--password", "secret", "--tls-verify"]) + + credentials = mock_configure.call_args.args[0] + assert credentials.username == "alice" + assert credentials.password == "secret" + assert credentials.token is None + assert credentials.tls_verify is True + + +@patch("taiga.mcp_server.server.mcp") +@patch("taiga.mcp_server.cli.configure") +def test_serve_reads_credentials_from_env(mock_configure, mock_mcp): + env = { + "TAIGA_HOST": "https://env.example.com", + "TAIGA_TOKEN": "env-tok", + "TAIGA_TOKEN_TYPE": "Basic", + } + with patch.dict("os.environ", env): + result = runner.invoke(cli.app, ["serve"]) + + assert result.exit_code == 0 + credentials = mock_configure.call_args.args[0] + assert credentials.host == "https://env.example.com" + assert credentials.token == "env-tok" + assert credentials.token_type == "Basic" + + +@patch("taiga.mcp_server.server.mcp") +@patch("taiga.mcp_server.cli.configure") +def test_serve_falls_back_to_tls_verify_env_var(mock_configure, mock_mcp): + with patch.dict("os.environ", {"TAIGA_TLS_VERIFY": "false"}): + runner.invoke(cli.app, ["serve", "--token", "tok"]) + + assert mock_configure.call_args.args[0].tls_verify is False + + +@patch("taiga.mcp_server.server.mcp") +@patch("taiga.mcp_server.cli.configure") +def test_serve_defaults_tls_verify_true_without_env_or_flag(mock_configure, mock_mcp): + with patch.dict("os.environ", {}, clear=False): + os.environ.pop("TAIGA_TLS_VERIFY", None) + runner.invoke(cli.app, ["serve", "--token", "tok"]) + + assert mock_configure.call_args.args[0].tls_verify is True + + +# --- list-tools --------------------------------------------------------------------------- + + +def test_list_tools_lists_all_tool_names(): + result = runner.invoke(cli.app, ["list-tools"]) + + assert result.exit_code == 0 + assert "whoami" in result.output + assert "list_user_stories" in result.output + assert "create_issue" in result.output + + +def test_list_tools_default_excludes_schema(): + result = runner.invoke(cli.app, ["list-tools"]) + + assert result.exit_code == 0 + assert '"properties"' not in result.output + + +def test_list_tools_verbose_includes_schema(): + result = runner.invoke(cli.app, ["list-tools", "--verbose"]) + + assert result.exit_code == 0 + assert '"properties"' in result.output + + +# --- call: success path -------------------------------------------------------------------- + + +@patch("taiga.mcp_server.auth._client", None) +@patch("taiga.mcp_server.auth._credentials", None) +def test_call_success_prints_structured_json_result(monkeypatch): + import taiga.mcp_server.server as server_mod + + monkeypatch.setattr( + server_mod, "get_client", lambda: type("C", (), {"me": lambda self: {"id": 1, "username": "demo"}})() + ) + + result = runner.invoke(cli.app, ["call", "whoami", "--json", "{}"]) + + assert result.exit_code == 0 + assert json.loads(result.output) == {"id": 1, "username": "demo"} + + +# --- call: error matrix --------------------------------------------------------------------- + + +def test_call_invalid_json_errors(): + result = runner.invoke(cli.app, ["call", "whoami", "--json", "{not valid"]) + + assert result.exit_code == 1 + assert "Invalid JSON in --json" in result.output + + +@patch("taiga.mcp_server.auth._client", None) +@patch("taiga.mcp_server.auth._credentials", None) +def test_call_unknown_tool_errors(): + result = runner.invoke(cli.app, ["call", "this_tool_does_not_exist", "--json", "{}"]) + + assert result.exit_code == 1 + assert "Unknown tool: this_tool_does_not_exist" in result.output + + +@patch("taiga.mcp_server.auth._client", None) +@patch("taiga.mcp_server.auth._credentials", None) +def test_call_missing_required_argument_errors(): + result = runner.invoke(cli.app, ["call", "get_project", "--json", "{}"]) + + assert result.exit_code == 1 + assert "Invalid arguments for get_project" in result.output + + +@patch("taiga.mcp_server.auth._client", None) +@patch("taiga.mcp_server.auth._credentials", None) +def test_call_tool_internal_exception_errors(monkeypatch): + for var in ("TAIGA_TOKEN", "TAIGA_USERNAME", "TAIGA_PASSWORD"): + monkeypatch.delenv(var, raising=False) + + result = runner.invoke(cli.app, ["call", "whoami", "--json", "{}"]) + + assert result.exit_code == 1 + assert "Error calling whoami" in result.output + assert "credentials" in result.output + + +# --- bare invocation (breaking change) --------------------------------------------------- + + +@patch("taiga.mcp_server.server.mcp") +@patch("taiga.mcp_server.cli.configure") +def test_bare_invocation_no_longer_serves(mock_configure, mock_mcp): + result = runner.invoke(cli.app, []) + + assert "serve" in result.output + assert result.exit_code != 0 + mock_configure.assert_not_called() + mock_mcp.run.assert_not_called() + + +# --- --version -------------------------------------------------------------------------- + + +def test_version_flag_prints_version_and_exits(): + from taiga import __version__ + + result = runner.invoke(cli.app, ["--version"]) + + assert result.exit_code == 0 + assert __version__ in result.output diff --git a/tests/test_mcp_server_serialize.py b/tests/test_mcp_server_serialize.py new file mode 100644 index 0000000..fedff0d --- /dev/null +++ b/tests/test_mcp_server_serialize.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +import datetime +from unittest.mock import MagicMock + +from taiga.mcp_server.serialize import to_jsonable +from taiga.models.base import InstanceResource + + +def _make_resource(**params): + """Build a real InstanceResource the way python-taiga parses an API response.""" + return InstanceResource(MagicMock(name="requester"), **params) + + +def test_to_jsonable_converts_instance_resource_to_dict(): + resource = _make_resource(id=1, subject="hello") + + result = to_jsonable(resource) + + assert result == {"id": 1, "subject": "hello"} + + +def test_to_jsonable_skips_requester(): + resource = _make_resource(id=1) + + result = to_jsonable(resource) + + assert "requester" not in result + + +def test_to_jsonable_recurses_into_nested_instance_resource(): + owner = _make_resource(id=7, full_name="Alice") + resource = _make_resource(id=1, owner=owner) + + result = to_jsonable(resource) + + assert result == {"id": 1, "owner": {"id": 7, "full_name": "Alice"}} + + +def test_to_jsonable_recurses_into_list_of_instance_resources(): + members = [_make_resource(id=1), _make_resource(id=2)] + resource = _make_resource(id=99, members=members) + + result = to_jsonable(resource) + + assert result == {"id": 99, "members": [{"id": 1}, {"id": 2}]} + + +def test_to_jsonable_converts_dates_parsed_by_instance_resource(): + # InstanceResource.__init__ parses created_date/modified_date matching this exact + # Taiga API format into real datetime objects - use that format here so the + # attribute is an actual datetime, not a string, when it reaches to_jsonable. + resource = _make_resource(id=1, created_date="2026-08-20T10:00:00+0000") + + assert isinstance(resource.created_date, datetime.datetime) + + result = to_jsonable(resource) + + assert result == {"id": 1, "created_date": resource.created_date.isoformat()} + + +def test_to_jsonable_converts_plain_date_and_datetime_values(): + resource = _make_resource( + id=1, + due_date=datetime.date(2026, 1, 1), + finished_at=datetime.datetime(2026, 1, 1, 12, 30, tzinfo=datetime.UTC), + ) + + result = to_jsonable(resource) + + assert result == { + "id": 1, + "due_date": "2026-01-01", + "finished_at": "2026-01-01T12:30:00+00:00", + } diff --git a/tests/test_projects.py b/tests/test_projects.py index edd3342..12a512d 100644 --- a/tests/test_projects.py +++ b/tests/test_projects.py @@ -508,6 +508,9 @@ def test_list_membership(self, mock_list_memberships): project.list_memberships() mock_list_memberships.assert_called_with(project=1) + project.list_memberships(page=2, page_size=50) + mock_list_memberships.assert_called_with(project=1, page=2, page_size=50) + @patch("taiga.models.Webhooks.create") def test_add_webhook(self, mock_new_webhook): rm = RequestMaker("/api/v1", "fakehost", "faketoken") diff --git a/tox.ini b/tox.ini index 9b31b0a..4a2fb8d 100644 --- a/tox.ini +++ b/tox.ini @@ -27,6 +27,17 @@ deps = ruff~=0.15.22 skip_install = true +[testenv:docs] +commands = + {envpython} -m invoke docbuild +deps = + invoke + setuptools + sphinx + sphinx-rtd-theme + -r{toxinidir}/requirements.txt +skip_install = true + [testenv:isort] commands = {envpython} -m isort -c --df taiga tests @@ -97,6 +108,7 @@ ignore = tasks.py tests/** debian/** + artifacts/** *.mo ignore-bad-ideas = *.mo