From f838779f759afd76b244b9edb1f53ca77bb89568 Mon Sep 17 00:00:00 2001 From: Steffen Ruettinger Date: Wed, 13 May 2026 13:01:20 +0200 Subject: [PATCH 01/13] feat(kb): add read-only Knowledge Base support Adds the read-only slice of Knowledge Base support requested by the maintainer in #200, in a focused PR designed for review: - ZammadAPIError typed exception for KB API failures. - ZammadClient KB read-only methods (direct HTTP, since zammad-py does not wrap KB endpoints): - list_knowledge_bases (with documented 404 -> ID-probe fallback; all other errors propagate) - get_knowledge_base, get_kb_category - get_kb_answer, get_kb_answer_with_content - list_kb_answers (per-answer 404 tolerated and logged; other errors raise) - search_kb_answers (BFS-expands subcategories, case-insensitive title/body match) - Pydantic models for KnowledgeBase / Category / Answer (+ translations) and StrictBaseModel param models for the new tools. - MCP tools (read-only): - zammad_list_knowledge_bases / zammad_get_knowledge_base - zammad_get_kb_category - zammad_list_kb_answers / zammad_search_kb_answers / zamm Adds the read-only slice of Knowledge Base support requestegormaintainer in #200, in a focused PR designed for review: - Zammadga - ZammadAPIError typed exception for KB API failures. urf- ZammadClient KB read-only methods (direct HTTP, sice not wrap KB endpoints): - list_knowledge_bases (with documentedg - list_knowledge_basal all other errors propagate) - get_knowledge_base, get_kb_cis - get_knowledge_base, get_kbes - get_kb_answer, get_kb_answer_with_eg - list_kb_answers (per-answer 404 tolerate K errors raise) - search_kb_answers (BFS-expands subcate D - search_kb_anes --- mcp_zammad/client.py | 347 ++++++++++++++++++++++++++++++ mcp_zammad/models.py | 192 ++++++++++++++++- mcp_zammad/server.py | 289 ++++++++++++++++++++++++- tests/test_kb_readonly.py | 428 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 1254 insertions(+), 2 deletions(-) create mode 100644 tests/test_kb_readonly.py diff --git a/mcp_zammad/client.py b/mcp_zammad/client.py index ce9da7d4..8ba1569f 100644 --- a/mcp_zammad/client.py +++ b/mcp_zammad/client.py @@ -1,7 +1,9 @@ """Zammad API client wrapper for the MCP server.""" +import html as _html import logging import os +import re as _re from typing import Any from urllib.parse import urlparse @@ -11,6 +13,22 @@ logger = logging.getLogger(__name__) +class ZammadAPIError(Exception): + """Raised when the Zammad API returns a non-2xx response. + + Attributes: + status_code: HTTP status code returned by Zammad + url: Full request URL that produced the error + body: Decoded JSON body (or text) of the error response + """ + + def __init__(self, status_code: int, url: str, body: object) -> None: + self.status_code = status_code + self.url = url + self.body = body + super().__init__(f"HTTP {status_code} from Zammad: {body} (URL: {url})") + + class ZammadClient: """Wrapper around zammad_py ZammadAPI with additional functionality.""" @@ -481,3 +499,332 @@ def list_tags(self) -> list[dict[str, Any]]: response = self.api.session.get(f"{self.url}/tag_list") response.raise_for_status() return list(response.json()) + + # ------------------------------------------------------------------ + # Knowledge Base methods (direct HTTP – not covered by zammad_py) + # Read-only operations only; writes/attachments land in follow-up PRs. + # ------------------------------------------------------------------ + + def _kb_url(self, *parts: str | int) -> str: + """Build a knowledge-base API URL from path components.""" + path = "/".join(str(p) for p in parts) + return f"{self.api.url}knowledge_bases/{path}" + + def _kb_raise_or_return(self, response: Any) -> dict[str, Any] | list[Any]: + """Raise ZammadAPIError on HTTP error; return parsed JSON body otherwise. + + Raises: + ZammadAPIError: if HTTP status is 4xx/5xx, with Zammad's error body included + """ + if not response.ok: + try: + body = response.json() + except ValueError: + body = response.text + raise ZammadAPIError(response.status_code, response.url, body) + if response.status_code == 204 or not response.content: + return {} + data = response.json() + if isinstance(data, list): + return data + return dict(data) + + def _probe_kb_ids(self) -> list[dict[str, Any]]: + """Probe KB IDs 1-10 individually as a fallback for the 404-listing case. + + Some Zammad versions return 404 on GET /knowledge_bases even when KBs + exist. As a known compatibility path we probe a small ID range. + + Skips 404 (ID not found), but raises ZammadAPIError on any non-404 + error so authentication or server problems are not silently hidden. + """ + results: list[dict[str, Any]] = [] + for kb_id in range(1, 11): + response = self.api.session.get(self._kb_url(kb_id)) + if response.status_code == 404: + continue + data = self._kb_raise_or_return(response) + if isinstance(data, dict) and data: + results.append(data) + return results + + def list_knowledge_bases(self) -> list[dict[str, Any]]: + """List all knowledge bases. + + Zammad's GET /knowledge_bases endpoint is unreliable on some versions + and returns 404 even when KBs exist. The only documented fallback is + the 404 -> ID-probing path; all other error statuses (401/403/5xx) + are propagated as :class:`ZammadAPIError`. + """ + response = self.api.session.get(self.api.url + "knowledge_bases") + if response.status_code == 404: + return self._probe_kb_ids() + if not response.ok: + try: + body = response.json() + except ValueError: + body = response.text + raise ZammadAPIError(response.status_code, response.url, body) + if not response.content: + raise ZammadAPIError( + response.status_code, + response.url, + "Empty response body for knowledge_bases listing", + ) + data = response.json() + if isinstance(data, list): + return list(data) + if isinstance(data, dict) and data: + return [data] + raise ZammadAPIError( + response.status_code, + response.url, + f"Unexpected knowledge_bases response shape: {type(data).__name__}", + ) + + def get_knowledge_base(self, kb_id: int) -> dict[str, Any]: + """Get a single knowledge base by ID.""" + response = self.api.session.get(self._kb_url(kb_id)) + result = self._kb_raise_or_return(response) + if not isinstance(result, dict): + raise ZammadAPIError( + response.status_code, + response.url, + f"Unexpected knowledge_base response shape: {type(result).__name__}", + ) + return result + + def get_kb_category(self, kb_id: int, category_id: int) -> dict[str, Any]: + """Get a single KB category.""" + response = self.api.session.get(self._kb_url(kb_id, "categories", category_id)) + result = self._kb_raise_or_return(response) + if not isinstance(result, dict): + raise ZammadAPIError( + response.status_code, + response.url, + f"Unexpected kb_category response shape: {type(result).__name__}", + ) + return result + + def get_kb_answer(self, kb_id: int, answer_id: int) -> dict[str, Any]: + """Get a single KB answer including translation/body content. + + Fetches the answer and re-fetches with ?include_contents={translation_id} + so KnowledgeBaseAnswerTranslationContent (body) is included. + """ + url = self._kb_url(kb_id, "answers", answer_id) + response = self.api.session.get(url) + payload = self._kb_raise_or_return(response) + if not isinstance(payload, dict): + raise ZammadAPIError( + response.status_code, + response.url, + f"Unexpected kb_answer response shape: {type(payload).__name__}", + ) + assets = payload.get("assets") or {} + answer_entry = (assets.get("KnowledgeBaseAnswer") or {}).get(str(answer_id)) or {} + translation_ids: list[int] = answer_entry.get("translation_ids") or [] + if translation_ids: + translation_id = translation_ids[0] + response2 = self.api.session.get(url, params={"include_contents": translation_id}) + data = self._kb_raise_or_return(response2) + if isinstance(data, dict): + return data + return payload + + # --- KB extraction helpers (operate on compound payloads) --- + + def _first_translation_field( + self, + translations: dict[str, Any], + translation_ids: list[int], + field: str, + ) -> str: + """Return the first non-empty string field from translations.""" + for tid in translation_ids: + value = (translations.get(str(tid)) or {}).get(field) + if value: + return str(value) + first = next(iter(translations.values()), {}) + return str(first[field]) if first.get(field) else "" + + def _extract_kb_answer_title( + self, raw_payload: dict[str, Any], answer: dict[str, Any] + ) -> str: + """Extract the first available title from translation assets.""" + assets = raw_payload.get("assets") or {} + translations = assets.get("KnowledgeBaseAnswerTranslation") or {} + if not translations: + return "" + translation_ids: list[int] = answer.get("translation_ids") or [] + return self._first_translation_field(translations, translation_ids, "title") + + def _strip_html(self, html: str) -> str: + """Strip HTML tags and unescape entities from a string.""" + return _html.unescape(_re.sub(r"<[^>]+>", " ", html)) + + def _body_from_content_assets( + self, contents: dict[str, Any], translation_ids: list[int] + ) -> str: + """Extract plain-text body from KnowledgeBaseAnswerTranslationContent.""" + for tid in translation_ids: + body = (contents.get(str(tid)) or {}).get("body") or "" + if body: + return self._strip_html(body) + first_body = next(iter(contents.values()), {}).get("body") or "" + return self._strip_html(first_body) if first_body else "" + + def _body_from_translation_assets( + self, translations: dict[str, Any], translation_ids: list[int] + ) -> str: + """Extract plain-text body from translation content_attributes (legacy).""" + for tid in translation_ids: + t = translations.get(str(tid)) or {} + body = (t.get("content_attributes") or {}).get("body") or "" + if body: + return self._strip_html(body) + return "" + + def _extract_kb_answer_body( + self, raw_payload: dict[str, Any], answer: dict[str, Any] + ) -> str: + """Extract the plain-text body from translation assets.""" + assets = raw_payload.get("assets") or {} + translation_ids: list[int] = answer.get("translation_ids") or [] + contents = assets.get("KnowledgeBaseAnswerTranslationContent") or {} + if contents: + return self._body_from_content_assets(contents, translation_ids) + translations = assets.get("KnowledgeBaseAnswerTranslation") or {} + if translations: + return self._body_from_translation_assets(translations, translation_ids) + return "" + + def _extract_kb_answer_from_payload( + self, payload: dict[str, Any], answer_id: int + ) -> dict[str, Any] | None: + """Extract the answer dict from a compound KB answer payload.""" + assets = payload.get("assets") or {} + kb_answers = assets.get("KnowledgeBaseAnswer") + if kb_answers: + return kb_answers.get(str(answer_id)) or next(iter(kb_answers.values()), None) + if "KnowledgeBaseAnswer" in payload: + answers_map = payload["KnowledgeBaseAnswer"] + return answers_map.get(str(answer_id)) or next(iter(answers_map.values()), None) + return payload if payload else None + + def get_kb_answer_with_content(self, kb_id: int, answer_id: int) -> dict[str, Any]: + """Get a KB answer with extracted title and body as a single processed dict. + + Returns: + Dict with keys 'answer' (flat answer dict), 'title' (str), 'body' (str) + """ + payload = self.get_kb_answer(kb_id, answer_id) + answer = self._extract_kb_answer_from_payload(payload, answer_id) or payload + return { + "answer": answer, + "title": self._extract_kb_answer_title(payload, answer), + "body": self._extract_kb_answer_body(payload, answer), + } + + def list_kb_answers(self, kb_id: int, category_id: int) -> list[dict[str, Any]]: + """List answers within a KB category by expanding the category's answer_ids. + + Each returned answer has '_title' and '_body' injected from translation + assets. Per-answer 404s are tolerated as a documented compatibility + path (an answer ID listed by the category may have been deleted in a + race); all other errors are surfaced as :class:`ZammadAPIError`. + """ + category = self.get_kb_category(kb_id, category_id) + answer_ids: list[int] = category.get("answer_ids") or [] + answers: list[dict[str, Any]] = [] + for aid in answer_ids: + try: + answer_data = self.get_kb_answer(kb_id, aid) + except ZammadAPIError as exc: + if exc.status_code == 404: + logger.warning("KB answer %d not found in category %d", aid, category_id) + continue + raise + answer_entry = self._extract_kb_answer_from_payload(answer_data, aid) + if answer_entry is None: + logger.warning("Failed to parse KB answer %d in category %d", aid, category_id) + continue + answer_entry["_title"] = self._extract_kb_answer_title(answer_data, answer_entry) + answer_entry["_body"] = self._extract_kb_answer_body(answer_data, answer_entry) + answers.append(answer_entry) + return answers + + def _answer_matches_query(self, answer: dict[str, Any], query_lower: str) -> bool: + """Return True if query_lower appears in the answer's title or body.""" + title = answer.get("_title") or "" + body = answer.get("_body") or "" + return query_lower in title.lower() or query_lower in body.lower() + + def _collect_category_answers( + self, kb_id: int, cid: int, query_lower: str + ) -> list[dict[str, Any]]: + """Return matching answers from a single category. + + Tolerates 404 on the category lookup (documented compatibility path); + all other errors are surfaced as :class:`ZammadAPIError`. + """ + matches: list[dict[str, Any]] = [] + try: + answers = self.list_kb_answers(kb_id, cid) + except ZammadAPIError as exc: + if exc.status_code == 404: + logger.warning("KB category %d not found during search", cid) + return matches + raise + for answer in answers: + if self._answer_matches_query(answer, query_lower): + answer["_category_id"] = cid + matches.append(answer) + return matches + + def _answers_matching_query( + self, kb_id: int, category_ids: list[int], query_lower: str + ) -> list[dict[str, Any]]: + """Return answers whose title or body matches query_lower.""" + results: list[dict[str, Any]] = [] + for cid in category_ids: + results.extend(self._collect_category_answers(kb_id, cid, query_lower)) + return results + + def _expand_category_ids(self, kb_id: int, root_ids: list[int]) -> list[int]: + """BFS-expand root category IDs into all descendants via child_ids.""" + visited: list[int] = [] + queue: list[int] = list(root_ids) + seen: set[int] = set() + while queue: + cid = queue.pop(0) + if cid in seen: + continue + seen.add(cid) + visited.append(cid) + try: + category = self.get_kb_category(kb_id, cid) + except ZammadAPIError as exc: + if exc.status_code == 404: + logger.warning("KB category %d not found while expanding tree", cid) + continue + raise + for child_id in category.get("child_ids") or []: + if child_id not in seen: + queue.append(child_id) + return visited + + def search_kb_answers( + self, kb_id: int, query: str, category_id: int | None = None + ) -> list[dict[str, Any]]: + """Case-insensitive substring search of KB answers across categories. + + If ``category_id`` is provided, search is limited to that category and + its descendants. Otherwise all root categories of the KB are scanned. + """ + kb = self.get_knowledge_base(kb_id) + root_ids = ( + [category_id] if category_id is not None else (kb.get("category_ids") or []) + ) + category_ids = self._expand_category_ids(kb_id, root_ids) + return self._answers_matching_query(kb_id, category_ids, query.lower()) diff --git a/mcp_zammad/models.py b/mcp_zammad/models.py index 29d0a4fb..cb08dba5 100644 --- a/mcp_zammad/models.py +++ b/mcp_zammad/models.py @@ -5,7 +5,7 @@ import os from datetime import date, datetime from enum import Enum -from typing import Literal +from typing import Any, Literal from pydantic import BaseModel, ConfigDict, Field, ValidationInfo, field_validator, model_validator @@ -671,3 +671,193 @@ class TagOperationResult(BaseModel): success: bool = Field(description="Whether the operation was successful") message: str | None = Field(None, description="Optional message about the operation") + + +# ============================================================================ +# Knowledge Base models (read-only PR) +# ============================================================================ + + +class KnowledgeBaseLocale(BaseModel): + """Locale entry associated with a knowledge base.""" + + id: int + knowledge_base_id: int + system_locale_id: int + primary: bool = False + created_at: datetime | None = None + updated_at: datetime | None = None + + +class KnowledgeBaseTranslation(BaseModel): + """Translation (title / footer) for a knowledge base.""" + + id: int + title: str | None = None + footer_note: str | None = None + kb_locale_id: int + knowledge_base_id: int + created_at: datetime | None = None + updated_at: datetime | None = None + + +class KnowledgeBase(BaseModel): + """Zammad Knowledge Base top-level object.""" + + id: int + iconset: str | None = None + color_highlight: str | None = None + color_header: str | None = None + color_header_link: str | None = None + homepage_layout: str | None = None + category_layout: str | None = None + active: bool = True + show_feed_icon: bool = False + custom_address: str | None = None + created_at: datetime | None = None + updated_at: datetime | None = None + translation_ids: list[int] | None = None + kb_locale_ids: list[int] | None = None + category_ids: list[int] | None = None + answer_ids: list[int] | None = None + permission_ids: list[int] | None = None + + +class KnowledgeBaseCategoryTranslation(BaseModel): + """Translation (title) for a KB category.""" + + id: int + title: str | None = None + kb_locale_id: int + category_id: int + created_at: datetime | None = None + updated_at: datetime | None = None + + +class KnowledgeBaseCategory(BaseModel): + """Zammad Knowledge Base category.""" + + id: int + knowledge_base_id: int + parent_id: int | None = None + category_icon: str | None = None + position: int = 0 + created_at: datetime | None = None + updated_at: datetime | None = None + translation_ids: list[int] | None = None + answer_ids: list[int] | None = None + child_ids: list[int] | None = None + permission_ids: list[int] | None = None + permissions_effective: list[dict[str, Any]] | None = None + + +class KnowledgeBaseAnswerTranslationContent(BaseModel): + """Body content for a KB answer translation.""" + + id: int + body: str | None = None + created_at: datetime | None = None + updated_at: datetime | None = None + + +class KnowledgeBaseAnswerTranslation(BaseModel): + """Translation (title + body) for a KB answer.""" + + id: int + title: str | None = None + kb_locale_id: int + answer_id: int + content_id: int | None = None + created_by_id: int | None = None + updated_by_id: int | None = None + created_at: datetime | None = None + updated_at: datetime | None = None + + +class KnowledgeBaseAnswerAttachment(BaseModel): + """Attachment metadata returned inside a KB answer payload.""" + + id: int + url: str | None = None + preview_url: str | None = None + filename: str | None = None + size: str | None = None + preferences: dict[str, Any] | None = None + + +class KnowledgeBaseAnswer(BaseModel): + """Zammad Knowledge Base answer (article).""" + + id: int + category_id: int + promoted: bool = False + internal_note: str | None = None + position: int = 0 + archived_at: datetime | None = None + archived_by_id: int | None = None + internal_at: datetime | None = None + internal_by_id: int | None = None + published_at: datetime | None = None + published_by_id: int | None = None + created_at: datetime | None = None + updated_at: datetime | None = None + translation_ids: list[int] | None = None + attachments: list[KnowledgeBaseAnswerAttachment] | None = None + tags: list[str] | None = None + + +# --- KB read-only param models (StrictBaseModel) --- + + +class GetKnowledgeBaseParams(StrictBaseModel): + """Parameters for retrieving a single knowledge base.""" + + kb_id: int = Field(gt=0, description="Knowledge base ID") + response_format: ResponseFormat = Field(default=ResponseFormat.MARKDOWN, description="Output format") + + +class ListKnowledgeBasesParams(StrictBaseModel): + """Parameters for listing knowledge bases.""" + + response_format: ResponseFormat = Field(default=ResponseFormat.MARKDOWN, description="Output format") + + +class GetKBCategoryParams(StrictBaseModel): + """Parameters for retrieving a KB category.""" + + kb_id: int = Field(gt=0, description="Knowledge base ID") + category_id: int = Field(gt=0, description="Category ID") + response_format: ResponseFormat = Field(default=ResponseFormat.MARKDOWN, description="Output format") + + +class GetKBAnswerParams(StrictBaseModel): + """Parameters for retrieving a KB answer.""" + + kb_id: int = Field(gt=0, description="Knowledge base ID") + answer_id: int = Field(gt=0, description="Answer ID") + response_format: ResponseFormat = Field(default=ResponseFormat.MARKDOWN, description="Output format") + + +class ListKBAnswersParams(StrictBaseModel): + """Parameters for listing answers within a KB category.""" + + kb_id: int = Field(gt=0, description="Knowledge base ID") + category_id: int = Field(gt=0, description="Category ID") + response_format: ResponseFormat = Field(default=ResponseFormat.MARKDOWN, description="Output format") + + +class SearchKBAnswersParams(StrictBaseModel): + """Parameters for searching KB answers by title or body keyword.""" + + kb_id: int = Field(gt=0, description="Knowledge base ID") + query: str = Field( + min_length=1, + max_length=200, + description="Search string (case-insensitive substring match on title and body)", + ) + category_id: int | None = Field( + default=None, + gt=0, + description="Limit search to this category and its descendants (optional)", + ) + response_format: ResponseFormat = Field(default=ResponseFormat.MARKDOWN, description="Output format") diff --git a/mcp_zammad/server.py b/mcp_zammad/server.py index 4a5f5443..2e47ca4b 100644 --- a/mcp_zammad/server.py +++ b/mcp_zammad/server.py @@ -19,7 +19,7 @@ from starlette.requests import Request from starlette.responses import JSONResponse -from .client import ZammadClient +from .client import ZammadAPIError, ZammadClient from .logging_config import configure_logging from .models import ( Article, @@ -30,16 +30,22 @@ DeleteAttachmentResult, DownloadAttachmentParams, GetArticleAttachmentsParams, + GetKBAnswerParams, + GetKBCategoryParams, + GetKnowledgeBaseParams, GetOrganizationParams, GetTicketParams, GetTicketStatsParams, GetTicketTagsParams, GetUserParams, Group, + ListKBAnswersParams, + ListKnowledgeBasesParams, ListParams, Organization, PriorityBrief, ResponseFormat, + SearchKBAnswersParams, SearchOrganizationsParams, SearchUsersParams, StateBrief, @@ -785,6 +791,126 @@ def _handle_api_error(e: Exception, context: str = "operation") -> str: return f"Error during {context}: {type(e).__name__} - {e}" +# ============================================================================ +# Knowledge Base helpers (read-only) +# ============================================================================ + + +def _kb_answer_status(answer: dict[str, Any]) -> str: + """Derive human-readable publication status from a KB answer dict.""" + if answer.get("archived_at"): + return "archived" + if answer.get("published_at"): + return "published" + if answer.get("internal_at"): + return "internal" + return "draft" + + +def _format_kb_markdown(kb: dict[str, Any]) -> str: + """Format a KnowledgeBase dict as markdown.""" + lines = [f"# Knowledge Base (ID: {kb.get('id', 'N/A')})", ""] + lines.append(f"**Active**: {kb.get('active', False)}") + if kb.get("custom_address"): + lines.append(f"**Address**: {kb['custom_address']}") + lines.append(f"**Homepage Layout**: {kb.get('homepage_layout', 'N/A')}") + lines.append(f"**Category Layout**: {kb.get('category_layout', 'N/A')}") + cat_ids = kb.get("category_ids") or [] + ans_ids = kb.get("answer_ids") or [] + lines.append(f"**Root Categories**: {len(cat_ids)} (IDs: {cat_ids})") + lines.append(f"**Answers**: {len(ans_ids)} total") + lines.append(f"**Updated**: {kb.get('updated_at', 'N/A')}") + return "\n".join(lines) + + +def _format_kb_category_markdown(category: dict[str, Any]) -> str: + """Format a KnowledgeBaseCategory dict as markdown.""" + lines = [f"# KB Category (ID: {category.get('id', 'N/A')})", ""] + lines.append(f"**Knowledge Base ID**: {category.get('knowledge_base_id', 'N/A')}") + lines.append(f"**Parent ID**: {category.get('parent_id', 'None (root)')}") + lines.append(f"**Icon**: {category.get('category_icon', 'N/A')}") + lines.append(f"**Position**: {category.get('position', 0)}") + child_ids = category.get("child_ids") or [] + answer_ids = category.get("answer_ids") or [] + translation_ids = category.get("translation_ids") or [] + lines.append(f"**Child Categories**: {len(child_ids)} (IDs: {child_ids})") + lines.append(f"**Answers**: {len(answer_ids)} (IDs: {answer_ids})") + lines.append(f"**Translation IDs**: {translation_ids}") + lines.append(f"**Updated**: {category.get('updated_at', 'N/A')}") + return "\n".join(lines) + + +def _format_kb_answer_optional_sections(answer: dict[str, Any], body: str) -> list[str]: + """Build optional markdown sections (content, attachments, tags) for a KB answer.""" + lines: list[str] = [] + if body: + lines += ["", "## Content", "", body.strip()] + attachments = answer.get("attachments") or [] + if attachments: + lines += ["", "## Attachments", ""] + lines += [ + f"- **{att.get('filename', 'N/A')}** (ID: {att.get('id', 'N/A')}, size: {att.get('size', '?')} bytes)" + for att in attachments + ] + tags = answer.get("tags") or [] + if tags: + lines += ["", f"**Tags**: {', '.join(tags)}"] + return lines + + +def _format_kb_answer_markdown(answer: dict[str, Any], title: str = "", body: str = "") -> str: + """Format a KnowledgeBaseAnswer dict as markdown.""" + status = _kb_answer_status(answer) + heading = title or f"KB Answer (ID: {answer.get('id', 'N/A')})" + translation_ids = answer.get("translation_ids") or [] + lines = [ + f"# {heading}", "", + f"**ID**: {answer.get('id', 'N/A')}", + f"**Category ID**: {answer.get('category_id', 'N/A')}", + f"**Status**: {status}", + f"**Promoted**: {answer.get('promoted', False)}", + f"**Position**: {answer.get('position', 0)}", + f"**Translation IDs**: {translation_ids}", + ] + lines += _format_kb_answer_optional_sections(answer, body) + lines += ["", f"**Updated**: {answer.get('updated_at', 'N/A')}"] + return "\n".join(lines) + + +def _format_kb_answers_list_markdown(answers: list[dict[str, Any]], kb_id: int, category_id: int) -> str: + """Format a list of KB answers as markdown.""" + lines = [f"# KB Answers in Category {category_id} (KB: {kb_id})", ""] + lines.append(f"Found {len(answers)} answer(s)") + lines.append("") + for answer in answers: + status = _kb_answer_status(answer) + title = answer.get("_title") or "(no title)" + lines.append(f"## {title} (ID: {answer.get('id', 'N/A')})") + lines.append(f"- **Status**: {status}") + lines.append(f"- **Promoted**: {answer.get('promoted', False)}") + lines.append(f"- **Position**: {answer.get('position', 0)}") + lines.append("") + return "\n".join(lines) + + +def _format_kb_search_results_markdown(results: list[dict[str, Any]], query: str, kb_id: int) -> str: + """Format KB answer search results as markdown.""" + if not results: + return f"No KB answers found matching '{query}' in KB {kb_id}." + lines = [f"# KB Answer Search: '{query}' (KB: {kb_id})", ""] + lines.append(f"Found {len(results)} match(es)") + lines.append("") + for answer in results: + title = answer.get("_title") or "(no title)" + status = _kb_answer_status(answer) + lines.append(f"## {title} (ID: {answer.get('id', 'N/A')})") + lines.append(f"- **Category ID**: {answer.get('_category_id', answer.get('category_id', 'N/A'))}") + lines.append(f"- **Status**: {status}") + lines.append(f"- **Promoted**: {answer.get('promoted', False)}") + lines.append("") + return "\n".join(lines) + + class ZammadMCPServer: """Zammad MCP Server with proper client lifecycle management.""" @@ -868,6 +994,7 @@ def _setup_tools(self) -> None: self._setup_ticket_tools() self._setup_user_org_tools() self._setup_system_tools() + self._setup_kb_tools() def _setup_ticket_tools(self) -> None: # noqa: PLR0915 """Register ticket-related tools.""" @@ -2469,6 +2596,7 @@ def _setup_resources(self) -> None: self._setup_user_resource() self._setup_organization_resource() self._setup_queue_resource() + self._setup_kb_resources() def _setup_ticket_resource(self) -> None: """Register ticket resource.""" @@ -2619,6 +2747,165 @@ def get_queue_resource(group: str) -> str: except (requests.exceptions.RequestException, ValueError, ValidationError) as e: return _handle_api_error(e, context=f"retrieving queue for group '{group}'") + def _setup_kb_tools(self) -> None: + """Register read-only Knowledge Base tools. + + Failure semantics: client-level errors (network/HTTP) are propagated as + exceptions (e.g. :class:`ZammadAPIError`) so MCP surfaces them as + actual tool errors instead of returning successful string payloads. + """ + self._setup_kb_info_tools() + self._setup_kb_category_tools() + self._setup_kb_answer_read_tools() + + def _setup_kb_info_tools(self) -> None: + """Register KB list/get knowledge-base tools.""" + + @self.mcp.tool(annotations=_read_only_annotations("List Knowledge Bases")) + def zammad_list_knowledge_bases(params: ListKnowledgeBasesParams) -> str: + """List all knowledge bases available in Zammad. + + Errors (auth, network, HTTP 5xx, ...) are raised as + :class:`ZammadAPIError` so the MCP client sees a real tool error. + + Note: + Requires knowledge_base.reader or knowledge_base.editor permission. + """ + client = self.get_client() + kbs = client.list_knowledge_bases() + if params.response_format == ResponseFormat.JSON: + result = json.dumps({"items": kbs, "count": len(kbs)}, indent=2, default=str) + else: + lines = ["# Knowledge Bases", "", f"Found {len(kbs)} knowledge base(s)", ""] + for kb in kbs: + lines.append(f"## KB ID: {kb.get('id', 'N/A')}") + lines.append(f"- **Active**: {kb.get('active', False)}") + if kb.get("custom_address"): + lines.append(f"- **Address**: {kb['custom_address']}") + cat_ids = kb.get("category_ids") or [] + lines.append(f"- **Root Categories**: {len(cat_ids)}") + lines.append("") + result = "\n".join(lines) + return truncate_response(result) + + @self.mcp.tool(annotations=_read_only_annotations("Get Knowledge Base")) + def zammad_get_knowledge_base(params: GetKnowledgeBaseParams) -> str: + """Get details of a specific knowledge base by ID. + + Note: + Requires knowledge_base.reader or knowledge_base.editor permission. + Use ``zammad_list_knowledge_bases`` to discover available KB IDs. + """ + client = self.get_client() + kb = client.get_knowledge_base(params.kb_id) + if params.response_format == ResponseFormat.JSON: + result = json.dumps(kb, indent=2, default=str) + else: + result = _format_kb_markdown(kb) + return truncate_response(result) + + def _setup_kb_category_tools(self) -> None: + """Register read-only KB category tools.""" + + @self.mcp.tool(annotations=_read_only_annotations("Get KB Category")) + def zammad_get_kb_category(params: GetKBCategoryParams) -> str: + """Get a knowledge base category by ID. + + Note: + Requires knowledge_base.reader or knowledge_base.editor permission. + """ + client = self.get_client() + category = client.get_kb_category(params.kb_id, params.category_id) + if params.response_format == ResponseFormat.JSON: + result = json.dumps(category, indent=2, default=str) + else: + result = _format_kb_category_markdown(category) + return truncate_response(result) + + def _setup_kb_answer_read_tools(self) -> None: + """Register read-only KB answer tools (list/search/get).""" + + @self.mcp.tool(annotations=_read_only_annotations("List KB Answers")) + def zammad_list_kb_answers(params: ListKBAnswersParams) -> str: + """List answers within a KB category. + + Each item exposes the resolved title via the ``_title`` key. + """ + client = self.get_client() + answers = client.list_kb_answers(params.kb_id, params.category_id) + if params.response_format == ResponseFormat.JSON: + result = json.dumps({"items": answers, "count": len(answers)}, indent=2, default=str) + else: + result = _format_kb_answers_list_markdown(answers, params.kb_id, params.category_id) + return truncate_response(result) + + @self.mcp.tool(annotations=_read_only_annotations("Search KB Answers")) + def zammad_search_kb_answers(params: SearchKBAnswersParams) -> str: + """Case-insensitive substring search of KB answers (title and body). + + Searches across all root categories of the KB by default, or only + the given ``category_id`` and its descendants when provided. + """ + client = self.get_client() + results = client.search_kb_answers( + params.kb_id, params.query, category_id=params.category_id + ) + if params.response_format == ResponseFormat.JSON: + result = json.dumps( + {"items": results, "count": len(results), "query": params.query}, + indent=2, + default=str, + ) + else: + result = _format_kb_search_results_markdown(results, params.query, params.kb_id) + return truncate_response(result) + + @self.mcp.tool(annotations=_read_only_annotations("Get KB Answer")) + def zammad_get_kb_answer(params: GetKBAnswerParams) -> str: + """Get a knowledge base answer by ID, including resolved title and body.""" + client = self.get_client() + result_payload = client.get_kb_answer_with_content(params.kb_id, params.answer_id) + answer = result_payload["answer"] + title = result_payload["title"] + body = result_payload["body"] + if params.response_format == ResponseFormat.JSON: + result = json.dumps( + {"answer": answer, "title": title, "body": body}, + indent=2, + default=str, + ) + else: + body_truncated = truncate_response(body) if body else "" + result = _format_kb_answer_markdown(answer, title=title, body=body_truncated) + return result + + def _setup_kb_resources(self) -> None: + """Register read-only Knowledge Base resources.""" + + @self.mcp.resource("zammad://kb/{kb_id}") + def get_kb_resource(kb_id: str) -> str: + """Get a knowledge base as a resource.""" + client = self.get_client() + kb = client.get_knowledge_base(int(kb_id)) + return truncate_response(_format_kb_markdown(kb)) + + @self.mcp.resource("zammad://kb/{kb_id}/category/{category_id}") + def get_kb_category_resource(kb_id: str, category_id: str) -> str: + """Get a KB category as a resource.""" + client = self.get_client() + category = client.get_kb_category(int(kb_id), int(category_id)) + return truncate_response(_format_kb_category_markdown(category)) + + @self.mcp.resource("zammad://kb/{kb_id}/answer/{answer_id}") + def get_kb_answer_resource(kb_id: str, answer_id: str) -> str: + """Get a KB answer as a resource.""" + client = self.get_client() + result = client.get_kb_answer_with_content(int(kb_id), int(answer_id)) + body = truncate_response(result["body"]) if result["body"] else "" + return _format_kb_answer_markdown( + result["answer"], title=result["title"], body=body + ) + def _setup_prompts(self) -> None: """Register all prompts with the MCP server.""" diff --git a/tests/test_kb_readonly.py b/tests/test_kb_readonly.py new file mode 100644 index 00000000..dc51d6b8 --- /dev/null +++ b/tests/test_kb_readonly.py @@ -0,0 +1,428 @@ +"""Tests for the read-only Knowledge Base feature (PR1). + +Scope: +- ZammadClient KB read-only methods (mocked HTTP). +- ZammadAPIError typed-error semantics. +- MCP tool error semantics: failures must propagate as exceptions, not as + successful string payloads. +""" + +from __future__ import annotations + +import json +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +from mcp_zammad.client import ZammadAPIError, ZammadClient +from mcp_zammad.server import ( + _format_kb_answer_markdown, + _format_kb_category_markdown, + _format_kb_markdown, + _kb_answer_status, +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_response( + status_code: int = 200, + json_body: Any = None, + *, + content: bytes | None = None, + url: str = "https://zammad.example/api/v1/knowledge_bases", +) -> MagicMock: + response = MagicMock() + response.status_code = status_code + response.ok = 200 <= status_code < 300 + response.url = url + if json_body is None and content is None: + response.content = b"" + else: + response.content = content if content is not None else json.dumps(json_body).encode() + if json_body is None and content is None: + response.json.side_effect = ValueError("no body") + else: + response.json.return_value = json_body + return response + + +@pytest.fixture +def kb_client() -> ZammadClient: + """Return a ZammadClient instance with credentials patched in env.""" + with patch.dict( + "os.environ", + {"ZAMMAD_URL": "https://zammad.example/api/v1/", "ZAMMAD_HTTP_TOKEN": "tok"}, + clear=False, + ): + client = ZammadClient() + # Replace the underlying session with a MagicMock for full HTTP control. + client.api.session = MagicMock() + # zammad_py exposes api.url as the base URL with a trailing slash. + client.api.url = "https://zammad.example/api/v1/" + return client + + +# --------------------------------------------------------------------------- +# ZammadAPIError + _kb_raise_or_return +# --------------------------------------------------------------------------- + + +class TestZammadAPIErrorAndRaise: + def test_raise_on_4xx_with_json_body(self, kb_client: ZammadClient) -> None: + resp = _make_response(403, {"error": "Forbidden"}) + with pytest.raises(ZammadAPIError) as exc: + kb_client._kb_raise_or_return(resp) + assert exc.value.status_code == 403 + assert exc.value.body == {"error": "Forbidden"} + + def test_raise_on_5xx_with_text_body(self, kb_client: ZammadClient) -> None: + resp = _make_response(500, content=b"server boom") + resp.json.side_effect = ValueError("no json") + resp.text = "server boom" + with pytest.raises(ZammadAPIError) as exc: + kb_client._kb_raise_or_return(resp) + assert exc.value.status_code == 500 + assert exc.value.body == "server boom" + + def test_204_returns_empty_dict(self, kb_client: ZammadClient) -> None: + resp = _make_response(204) + assert kb_client._kb_raise_or_return(resp) == {} + + +# --------------------------------------------------------------------------- +# list_knowledge_bases +# --------------------------------------------------------------------------- + + +class TestListKnowledgeBases: + def test_returns_list_directly(self, kb_client: ZammadClient) -> None: + kb_client.api.session.get.return_value = _make_response( + 200, [{"id": 1}, {"id": 2}] + ) + assert kb_client.list_knowledge_bases() == [{"id": 1}, {"id": 2}] + + def test_wraps_single_dict(self, kb_client: ZammadClient) -> None: + kb_client.api.session.get.return_value = _make_response(200, {"id": 1}) + assert kb_client.list_knowledge_bases() == [{"id": 1}] + + def test_404_falls_back_to_id_probing(self, kb_client: ZammadClient) -> None: + responses = [_make_response(404)] + responses += [_make_response(200, {"id": 1})] + responses += [_make_response(404)] * 9 # IDs 2-10 not found + kb_client.api.session.get.side_effect = responses + assert kb_client.list_knowledge_bases() == [{"id": 1}] + + def test_401_raises_typed_error(self, kb_client: ZammadClient) -> None: + kb_client.api.session.get.return_value = _make_response( + 401, {"error": "Unauthorized"} + ) + with pytest.raises(ZammadAPIError) as exc: + kb_client.list_knowledge_bases() + assert exc.value.status_code == 401 + + def test_500_raises_typed_error(self, kb_client: ZammadClient) -> None: + kb_client.api.session.get.return_value = _make_response(500, {"error": "boom"}) + with pytest.raises(ZammadAPIError): + kb_client.list_knowledge_bases() + + def test_empty_body_raises_instead_of_silent_fallback( + self, kb_client: ZammadClient + ) -> None: + resp = _make_response(200) # 200 with empty content + kb_client.api.session.get.return_value = resp + with pytest.raises(ZammadAPIError): + kb_client.list_knowledge_bases() + + def test_unexpected_shape_raises(self, kb_client: ZammadClient) -> None: + kb_client.api.session.get.return_value = _make_response(200, "weird") + with pytest.raises(ZammadAPIError): + kb_client.list_knowledge_bases() + + def test_probe_propagates_non_404_errors(self, kb_client: ZammadClient) -> None: + # GET /knowledge_bases -> 404, then probing ID 1 -> 401 (auth error). + responses = [_make_response(404), _make_response(401, {"error": "auth"})] + kb_client.api.session.get.side_effect = responses + with pytest.raises(ZammadAPIError) as exc: + kb_client.list_knowledge_bases() + assert exc.value.status_code == 401 + + +# --------------------------------------------------------------------------- +# get_knowledge_base / get_kb_category / get_kb_answer +# --------------------------------------------------------------------------- + + +class TestSimpleGetters: + def test_get_knowledge_base_returns_dict(self, kb_client: ZammadClient) -> None: + kb_client.api.session.get.return_value = _make_response(200, {"id": 1, "active": True}) + assert kb_client.get_knowledge_base(1) == {"id": 1, "active": True} + + def test_get_knowledge_base_404_raises(self, kb_client: ZammadClient) -> None: + kb_client.api.session.get.return_value = _make_response( + 404, {"error": "not found"} + ) + with pytest.raises(ZammadAPIError) as exc: + kb_client.get_knowledge_base(999) + assert exc.value.status_code == 404 + + def test_get_kb_category_returns_dict(self, kb_client: ZammadClient) -> None: + kb_client.api.session.get.return_value = _make_response( + 200, {"id": 5, "knowledge_base_id": 1, "child_ids": [], "answer_ids": [10]} + ) + result = kb_client.get_kb_category(1, 5) + assert result["id"] == 5 + assert result["answer_ids"] == [10] + + def test_get_kb_answer_single_request_when_no_translations( + self, kb_client: ZammadClient + ) -> None: + kb_client.api.session.get.return_value = _make_response( + 200, {"id": 7, "assets": {}} + ) + result = kb_client.get_kb_answer(1, 7) + assert result["id"] == 7 + assert kb_client.api.session.get.call_count == 1 + + def test_get_kb_answer_refetches_with_translation(self, kb_client: ZammadClient) -> None: + first = _make_response( + 200, + { + "id": 7, + "assets": { + "KnowledgeBaseAnswer": {"7": {"id": 7, "translation_ids": [42]}} + }, + }, + ) + second = _make_response( + 200, + { + "id": 7, + "assets": { + "KnowledgeBaseAnswer": {"7": {"id": 7, "translation_ids": [42]}}, + "KnowledgeBaseAnswerTranslation": { + "42": {"id": 42, "title": "Hello", "answer_id": 7} + }, + "KnowledgeBaseAnswerTranslationContent": { + "42": {"id": 42, "body": "

Hi

"} + }, + }, + }, + ) + kb_client.api.session.get.side_effect = [first, second] + result = kb_client.get_kb_answer(1, 7) + assert "KnowledgeBaseAnswerTranslationContent" in result["assets"] + assert kb_client.api.session.get.call_count == 2 + + +# --------------------------------------------------------------------------- +# Extraction helpers +# --------------------------------------------------------------------------- + + +class TestExtraction: + def test_extract_title_and_body(self, kb_client: ZammadClient) -> None: + payload = { + "assets": { + "KnowledgeBaseAnswer": {"7": {"id": 7, "translation_ids": [42]}}, + "KnowledgeBaseAnswerTranslation": { + "42": {"id": 42, "title": "Hello", "answer_id": 7} + }, + "KnowledgeBaseAnswerTranslationContent": { + "42": {"id": 42, "body": "

Hi there&you

"} + }, + } + } + info = kb_client.get_kb_answer_with_content.__wrapped__ if hasattr( + kb_client.get_kb_answer_with_content, "__wrapped__" + ) else None + del info # not used; we exercise extractors directly below + answer = kb_client._extract_kb_answer_from_payload(payload, 7) + assert answer is not None + assert kb_client._extract_kb_answer_title(payload, answer) == "Hello" + assert "Hi" in kb_client._extract_kb_answer_body(payload, answer) + + def test_strip_html(self, kb_client: ZammadClient) -> None: + assert "Hi" in kb_client._strip_html("

Hi there

") + assert "<" not in kb_client._strip_html("

Hi

") + + def test_extract_from_flat_payload(self, kb_client: ZammadClient) -> None: + # Flat dict (no assets) is returned as-is. + flat = {"id": 7} + assert kb_client._extract_kb_answer_from_payload(flat, 7) == flat + + +# --------------------------------------------------------------------------- +# list_kb_answers / search_kb_answers +# --------------------------------------------------------------------------- + + +def _category_response(answer_ids: list[int], child_ids: list[int] | None = None) -> MagicMock: + return _make_response( + 200, + { + "id": 5, + "knowledge_base_id": 1, + "answer_ids": answer_ids, + "child_ids": child_ids or [], + }, + ) + + +def _answer_response(answer_id: int, title: str, body: str) -> MagicMock: + payload = { + "id": answer_id, + "assets": { + "KnowledgeBaseAnswer": { + str(answer_id): { + "id": answer_id, + "category_id": 5, + "translation_ids": [answer_id * 10], + } + }, + "KnowledgeBaseAnswerTranslation": { + str(answer_id * 10): { + "id": answer_id * 10, + "title": title, + "answer_id": answer_id, + } + }, + "KnowledgeBaseAnswerTranslationContent": { + str(answer_id * 10): {"id": answer_id * 10, "body": body} + }, + }, + } + return _make_response(200, payload) + + +class TestListAndSearch: + def test_list_kb_answers_injects_title_and_body(self, kb_client: ZammadClient) -> None: + kb_client.api.session.get.side_effect = [ + _category_response([1]), + _answer_response(1, "T1", "

Body1

"), + _answer_response(1, "T1", "

Body1

"), # second fetch with translation + ] + result = kb_client.list_kb_answers(1, 5) + assert len(result) == 1 + assert result[0]["_title"] == "T1" + assert "Body1" in result[0]["_body"] + + def test_list_kb_answers_tolerates_per_answer_404( + self, kb_client: ZammadClient + ) -> None: + kb_client.api.session.get.side_effect = [ + _category_response([1, 2]), + _make_response(404, {"error": "gone"}), # answer 1 missing + _answer_response(2, "T2", "

B

"), + _answer_response(2, "T2", "

B

"), + ] + result = kb_client.list_kb_answers(1, 5) + assert len(result) == 1 + assert result[0]["id"] == 2 + + def test_list_kb_answers_propagates_non_404(self, kb_client: ZammadClient) -> None: + kb_client.api.session.get.side_effect = [ + _category_response([1]), + _make_response(401, {"error": "auth"}), + ] + with pytest.raises(ZammadAPIError) as exc: + kb_client.list_kb_answers(1, 5) + assert exc.value.status_code == 401 + + def test_search_kb_answers_finds_match_by_title( + self, kb_client: ZammadClient + ) -> None: + kb_client.api.session.get.side_effect = [ + _make_response( + 200, {"id": 1, "category_ids": [5], "answer_ids": []} + ), # get_knowledge_base + _category_response([1]), # _expand_category_ids fetch of cat 5 + _category_response([1]), # list_kb_answers fetch of cat 5 + _answer_response(1, "FooBar", "

nothing

"), + _answer_response(1, "FooBar", "

nothing

"), + ] + result = kb_client.search_kb_answers(1, "foo") + assert len(result) == 1 + assert result[0]["_title"] == "FooBar" + assert result[0]["_category_id"] == 5 + + +# --------------------------------------------------------------------------- +# Server formatters + tool failure semantics +# --------------------------------------------------------------------------- + + +class TestFormatters: + def test_format_kb_markdown(self) -> None: + out = _format_kb_markdown({"id": 1, "active": True, "category_ids": [10]}) + assert "Knowledge Base (ID: 1)" in out + assert "Root Categories" in out + + def test_format_kb_category_markdown(self) -> None: + out = _format_kb_category_markdown( + {"id": 5, "knowledge_base_id": 1, "child_ids": [6], "answer_ids": [7]} + ) + assert "KB Category (ID: 5)" in out + + def test_format_kb_answer_markdown_status_archived(self) -> None: + out = _format_kb_answer_markdown( + {"id": 7, "category_id": 5, "archived_at": "2024-01-01"}, + title="X", + body="hello", + ) + assert "archived" in out.lower() + assert "## Content" in out + + def test_kb_answer_status_levels(self) -> None: + assert _kb_answer_status({"archived_at": "x"}) == "archived" + assert _kb_answer_status({"published_at": "x"}) == "published" + assert _kb_answer_status({"internal_at": "x"}) == "internal" + assert _kb_answer_status({}) == "draft" + + +class TestToolFailureSemantics: + """Maintainer requirement: tool failures must be real errors, not strings. + + We exercise the registered tools through the FastMCP get_tool() API and + assert that ZammadAPIError raised by the client is propagated rather than + captured into a successful string payload. + """ + + @pytest.mark.asyncio + async def test_list_knowledge_bases_propagates_zammad_api_error( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + from mcp_zammad import server as srv + + # Build a server with a stubbed client. + instance = srv.ZammadMCPServer() + fake_client = MagicMock() + fake_client.list_knowledge_bases.side_effect = ZammadAPIError( + 500, "https://zammad.example/api/v1/knowledge_bases", {"error": "boom"} + ) + monkeypatch.setattr(instance, "get_client", lambda: fake_client) + + tool = await instance.mcp.get_tool("zammad_list_knowledge_bases") + with pytest.raises(ZammadAPIError): + await tool.run({"params": {}}) + + @pytest.mark.asyncio + async def test_get_kb_answer_propagates_zammad_api_error( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + from mcp_zammad import server as srv + + instance = srv.ZammadMCPServer() + fake_client = MagicMock() + fake_client.get_kb_answer_with_content.side_effect = ZammadAPIError( + 404, "https://zammad.example/api/v1/knowledge_bases/1/answers/9", {"error": "nope"} + ) + monkeypatch.setattr(instance, "get_client", lambda: fake_client) + + tool = await instance.mcp.get_tool("zammad_get_kb_answer") + with pytest.raises(ZammadAPIError): + await tool.run({"params": {"kb_id": 1, "answer_id": 9}}) From bf5a83c96524b439294a21eb93647c1a707be7e7 Mon Sep 17 00:00:00 2001 From: Steffen Ruettinger Date: Wed, 13 May 2026 13:10:18 +0200 Subject: [PATCH 02/13] fix(kb): address Codacy pydocstyle and ruff findings - Restructure new docstrings to satisfy D213/D406/D407/D413/D203/D107 (blank line before class docstrings, summary on second line, sections rewritten as prose). - Add docstring on ZammadAPIError.__init__. - Drop unused ZammadAPIError import from server.py (KB tools rely on exceptions propagating from the client; the symbol is not referenced directly here). - Replace magic HTTP status numbers (204, 404) with module constants (_HTTP_NO_CONTENT, _HTTP_NOT_FOUND) to address PLR2004. - Fix RUF003 (en-dash in client.py KB section header). - Tests: hoist 'mcp_zammad.server' import to module top (PLC0415), drop decorative section comments that ruff flagged as ERA001, sort imports. No behavioral changes. --- mcp_zammad/client.py | 60 ++++++++++++++++++++++----------------- mcp_zammad/server.py | 2 +- tests/test_kb_readonly.py | 43 ++++------------------------ vendor/llm-anon-core | 1 + 4 files changed, 42 insertions(+), 64 deletions(-) create mode 160000 vendor/llm-anon-core diff --git a/mcp_zammad/client.py b/mcp_zammad/client.py index 8ba1569f..f0cc3013 100644 --- a/mcp_zammad/client.py +++ b/mcp_zammad/client.py @@ -12,17 +12,22 @@ logger = logging.getLogger(__name__) +# HTTP status codes used by the KB compatibility paths. +_HTTP_NO_CONTENT = 204 +_HTTP_NOT_FOUND = 404 + class ZammadAPIError(Exception): - """Raised when the Zammad API returns a non-2xx response. - Attributes: - status_code: HTTP status code returned by Zammad - url: Full request URL that produced the error - body: Decoded JSON body (or text) of the error response + """ + Raised when the Zammad API returns a non-2xx response. + + Exposes ``status_code``, ``url`` and ``body`` of the failing response so + callers can react to specific error classes (e.g. 401/403/404/5xx). """ def __init__(self, status_code: int, url: str, body: object) -> None: + """Initialize a Zammad API error from response context.""" self.status_code = status_code self.url = url self.body = body @@ -501,7 +506,7 @@ def list_tags(self) -> list[dict[str, Any]]: return list(response.json()) # ------------------------------------------------------------------ - # Knowledge Base methods (direct HTTP – not covered by zammad_py) + # Knowledge Base methods (direct HTTP - not covered by zammad_py). # Read-only operations only; writes/attachments land in follow-up PRs. # ------------------------------------------------------------------ @@ -511,18 +516,14 @@ def _kb_url(self, *parts: str | int) -> str: return f"{self.api.url}knowledge_bases/{path}" def _kb_raise_or_return(self, response: Any) -> dict[str, Any] | list[Any]: - """Raise ZammadAPIError on HTTP error; return parsed JSON body otherwise. - - Raises: - ZammadAPIError: if HTTP status is 4xx/5xx, with Zammad's error body included - """ + """Raise ZammadAPIError on HTTP error or return the parsed JSON body otherwise.""" if not response.ok: try: body = response.json() except ValueError: body = response.text raise ZammadAPIError(response.status_code, response.url, body) - if response.status_code == 204 or not response.content: + if response.status_code == _HTTP_NO_CONTENT or not response.content: return {} data = response.json() if isinstance(data, list): @@ -530,7 +531,8 @@ def _kb_raise_or_return(self, response: Any) -> dict[str, Any] | list[Any]: return dict(data) def _probe_kb_ids(self) -> list[dict[str, Any]]: - """Probe KB IDs 1-10 individually as a fallback for the 404-listing case. + """ + Probe KB IDs 1-10 individually as a fallback for the 404-listing case. Some Zammad versions return 404 on GET /knowledge_bases even when KBs exist. As a known compatibility path we probe a small ID range. @@ -541,7 +543,7 @@ def _probe_kb_ids(self) -> list[dict[str, Any]]: results: list[dict[str, Any]] = [] for kb_id in range(1, 11): response = self.api.session.get(self._kb_url(kb_id)) - if response.status_code == 404: + if response.status_code == _HTTP_NOT_FOUND: continue data = self._kb_raise_or_return(response) if isinstance(data, dict) and data: @@ -549,7 +551,8 @@ def _probe_kb_ids(self) -> list[dict[str, Any]]: return results def list_knowledge_bases(self) -> list[dict[str, Any]]: - """List all knowledge bases. + """ + List all knowledge bases. Zammad's GET /knowledge_bases endpoint is unreliable on some versions and returns 404 even when KBs exist. The only documented fallback is @@ -557,7 +560,7 @@ def list_knowledge_bases(self) -> list[dict[str, Any]]: are propagated as :class:`ZammadAPIError`. """ response = self.api.session.get(self.api.url + "knowledge_bases") - if response.status_code == 404: + if response.status_code == _HTTP_NOT_FOUND: return self._probe_kb_ids() if not response.ok: try: @@ -607,7 +610,8 @@ def get_kb_category(self, kb_id: int, category_id: int) -> dict[str, Any]: return result def get_kb_answer(self, kb_id: int, answer_id: int) -> dict[str, Any]: - """Get a single KB answer including translation/body content. + """ + Get a single KB answer including translation/body content. Fetches the answer and re-fetches with ?include_contents={translation_id} so KnowledgeBaseAnswerTranslationContent (body) is included. @@ -713,10 +717,11 @@ def _extract_kb_answer_from_payload( return payload if payload else None def get_kb_answer_with_content(self, kb_id: int, answer_id: int) -> dict[str, Any]: - """Get a KB answer with extracted title and body as a single processed dict. + """ + Get a KB answer with extracted title and body as a single processed dict. - Returns: - Dict with keys 'answer' (flat answer dict), 'title' (str), 'body' (str) + The returned dict has the keys ``answer`` (flat answer dict), + ``title`` (str) and ``body`` (str, plain text with HTML stripped). """ payload = self.get_kb_answer(kb_id, answer_id) answer = self._extract_kb_answer_from_payload(payload, answer_id) or payload @@ -727,7 +732,8 @@ def get_kb_answer_with_content(self, kb_id: int, answer_id: int) -> dict[str, An } def list_kb_answers(self, kb_id: int, category_id: int) -> list[dict[str, Any]]: - """List answers within a KB category by expanding the category's answer_ids. + """ + List answers within a KB category by expanding the category's answer_ids. Each returned answer has '_title' and '_body' injected from translation assets. Per-answer 404s are tolerated as a documented compatibility @@ -741,7 +747,7 @@ def list_kb_answers(self, kb_id: int, category_id: int) -> list[dict[str, Any]]: try: answer_data = self.get_kb_answer(kb_id, aid) except ZammadAPIError as exc: - if exc.status_code == 404: + if exc.status_code == _HTTP_NOT_FOUND: logger.warning("KB answer %d not found in category %d", aid, category_id) continue raise @@ -763,7 +769,8 @@ def _answer_matches_query(self, answer: dict[str, Any], query_lower: str) -> boo def _collect_category_answers( self, kb_id: int, cid: int, query_lower: str ) -> list[dict[str, Any]]: - """Return matching answers from a single category. + """ + Return matching answers from a single category. Tolerates 404 on the category lookup (documented compatibility path); all other errors are surfaced as :class:`ZammadAPIError`. @@ -772,7 +779,7 @@ def _collect_category_answers( try: answers = self.list_kb_answers(kb_id, cid) except ZammadAPIError as exc: - if exc.status_code == 404: + if exc.status_code == _HTTP_NOT_FOUND: logger.warning("KB category %d not found during search", cid) return matches raise @@ -805,7 +812,7 @@ def _expand_category_ids(self, kb_id: int, root_ids: list[int]) -> list[int]: try: category = self.get_kb_category(kb_id, cid) except ZammadAPIError as exc: - if exc.status_code == 404: + if exc.status_code == _HTTP_NOT_FOUND: logger.warning("KB category %d not found while expanding tree", cid) continue raise @@ -817,7 +824,8 @@ def _expand_category_ids(self, kb_id: int, root_ids: list[int]) -> list[int]: def search_kb_answers( self, kb_id: int, query: str, category_id: int | None = None ) -> list[dict[str, Any]]: - """Case-insensitive substring search of KB answers across categories. + """ + Case-insensitive substring search of KB answers across categories. If ``category_id`` is provided, search is limited to that category and its descendants. Otherwise all root categories of the KB are scanned. diff --git a/mcp_zammad/server.py b/mcp_zammad/server.py index 2e47ca4b..38c88805 100644 --- a/mcp_zammad/server.py +++ b/mcp_zammad/server.py @@ -19,7 +19,7 @@ from starlette.requests import Request from starlette.responses import JSONResponse -from .client import ZammadAPIError, ZammadClient +from .client import ZammadClient from .logging_config import configure_logging from .models import ( Article, diff --git a/tests/test_kb_readonly.py b/tests/test_kb_readonly.py index dc51d6b8..824a1c9e 100644 --- a/tests/test_kb_readonly.py +++ b/tests/test_kb_readonly.py @@ -1,4 +1,5 @@ -"""Tests for the read-only Knowledge Base feature (PR1). +""" +Tests for the read-only Knowledge Base feature (PR1). Scope: - ZammadClient KB read-only methods (mocked HTTP). @@ -15,6 +16,7 @@ import pytest +from mcp_zammad import server as srv from mcp_zammad.client import ZammadAPIError, ZammadClient from mcp_zammad.server import ( _format_kb_answer_markdown, @@ -23,7 +25,6 @@ _kb_answer_status, ) - # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- @@ -67,11 +68,6 @@ def kb_client() -> ZammadClient: return client -# --------------------------------------------------------------------------- -# ZammadAPIError + _kb_raise_or_return -# --------------------------------------------------------------------------- - - class TestZammadAPIErrorAndRaise: def test_raise_on_4xx_with_json_body(self, kb_client: ZammadClient) -> None: resp = _make_response(403, {"error": "Forbidden"}) @@ -94,11 +90,6 @@ def test_204_returns_empty_dict(self, kb_client: ZammadClient) -> None: assert kb_client._kb_raise_or_return(resp) == {} -# --------------------------------------------------------------------------- -# list_knowledge_bases -# --------------------------------------------------------------------------- - - class TestListKnowledgeBases: def test_returns_list_directly(self, kb_client: ZammadClient) -> None: kb_client.api.session.get.return_value = _make_response( @@ -152,11 +143,6 @@ def test_probe_propagates_non_404_errors(self, kb_client: ZammadClient) -> None: assert exc.value.status_code == 401 -# --------------------------------------------------------------------------- -# get_knowledge_base / get_kb_category / get_kb_answer -# --------------------------------------------------------------------------- - - class TestSimpleGetters: def test_get_knowledge_base_returns_dict(self, kb_client: ZammadClient) -> None: kb_client.api.session.get.return_value = _make_response(200, {"id": 1, "active": True}) @@ -219,11 +205,6 @@ def test_get_kb_answer_refetches_with_translation(self, kb_client: ZammadClient) assert kb_client.api.session.get.call_count == 2 -# --------------------------------------------------------------------------- -# Extraction helpers -# --------------------------------------------------------------------------- - - class TestExtraction: def test_extract_title_and_body(self, kb_client: ZammadClient) -> None: payload = { @@ -256,11 +237,6 @@ def test_extract_from_flat_payload(self, kb_client: ZammadClient) -> None: assert kb_client._extract_kb_answer_from_payload(flat, 7) == flat -# --------------------------------------------------------------------------- -# list_kb_answers / search_kb_answers -# --------------------------------------------------------------------------- - - def _category_response(answer_ids: list[int], child_ids: list[int] | None = None) -> MagicMock: return _make_response( 200, @@ -351,11 +327,6 @@ def test_search_kb_answers_finds_match_by_title( assert result[0]["_category_id"] == 5 -# --------------------------------------------------------------------------- -# Server formatters + tool failure semantics -# --------------------------------------------------------------------------- - - class TestFormatters: def test_format_kb_markdown(self) -> None: out = _format_kb_markdown({"id": 1, "active": True, "category_ids": [10]}) @@ -385,7 +356,9 @@ def test_kb_answer_status_levels(self) -> None: class TestToolFailureSemantics: - """Maintainer requirement: tool failures must be real errors, not strings. + + """ + Maintainer requirement: tool failures must be real errors, not strings. We exercise the registered tools through the FastMCP get_tool() API and assert that ZammadAPIError raised by the client is propagated rather than @@ -396,8 +369,6 @@ class TestToolFailureSemantics: async def test_list_knowledge_bases_propagates_zammad_api_error( self, monkeypatch: pytest.MonkeyPatch ) -> None: - from mcp_zammad import server as srv - # Build a server with a stubbed client. instance = srv.ZammadMCPServer() fake_client = MagicMock() @@ -414,8 +385,6 @@ async def test_list_knowledge_bases_propagates_zammad_api_error( async def test_get_kb_answer_propagates_zammad_api_error( self, monkeypatch: pytest.MonkeyPatch ) -> None: - from mcp_zammad import server as srv - instance = srv.ZammadMCPServer() fake_client = MagicMock() fake_client.get_kb_answer_with_content.side_effect = ZammadAPIError( diff --git a/vendor/llm-anon-core b/vendor/llm-anon-core new file mode 160000 index 00000000..d74ecf40 --- /dev/null +++ b/vendor/llm-anon-core @@ -0,0 +1 @@ +Subproject commit d74ecf40201daf17d042b772328f4a9819cd11b9 From 47973d587a47c4050240f4e0a57a1ae3f9175578 Mon Sep 17 00:00:00 2001 From: Steffen Ruettinger Date: Wed, 13 May 2026 13:10:45 +0200 Subject: [PATCH 03/13] chore: drop accidentally-committed vendor/llm-anon-core gitlink The vendored llm-anon-core directory belongs to the PII feature on the fork's main branch and is not part of this read-only KB slice. --- vendor/llm-anon-core | 1 - 1 file changed, 1 deletion(-) delete mode 160000 vendor/llm-anon-core diff --git a/vendor/llm-anon-core b/vendor/llm-anon-core deleted file mode 160000 index d74ecf40..00000000 --- a/vendor/llm-anon-core +++ /dev/null @@ -1 +0,0 @@ -Subproject commit d74ecf40201daf17d042b772328f4a9819cd11b9 From 7d602999952fa081764c275e90019159004b428e Mon Sep 17 00:00:00 2001 From: Steffen Ruettinger Date: Wed, 13 May 2026 13:13:33 +0200 Subject: [PATCH 04/13] fix(kb): address CodeRabbit review findings - Replace response.ok with library-agnostic 2xx status_code check via new _is_2xx() helper. zammad_py currently uses requests where .ok exists, but this also keeps the code working if it ever migrates to HTTPX (which exposes .is_success instead of .ok). - _kb_raise_or_return now raises ZammadAPIError for 204 / empty-body responses on KB read endpoints instead of returning {}, so unexpected API states are not silently masked. - Widen _probe_kb_ids range from 1..10 to 1..200 with an explicit consecutive-404 break threshold (50). Prevents missing KBs whose IDs are above 10 (e.g. after deletes/migrations) while still bounding worst-case work on instances without any KBs. - Tests: - replace 'Any' with 'object | None' in _make_response (ANN401) - update 204 test to expect ZammadAPIError - replace fixed-range probe test with one that exercises the new consecutive-miss stop (51 calls expected) --- mcp_zammad/client.py | 46 +++++++++++++++++++++++++++++++-------- tests/test_kb_readonly.py | 23 +++++++++++++++----- vendor/llm-anon-core | 1 + 3 files changed, 56 insertions(+), 14 deletions(-) create mode 160000 vendor/llm-anon-core diff --git a/mcp_zammad/client.py b/mcp_zammad/client.py index f0cc3013..cb93ddd0 100644 --- a/mcp_zammad/client.py +++ b/mcp_zammad/client.py @@ -13,9 +13,20 @@ logger = logging.getLogger(__name__) # HTTP status codes used by the KB compatibility paths. +_HTTP_OK = 200 +_HTTP_MULTIPLE_CHOICES = 300 _HTTP_NO_CONTENT = 204 _HTTP_NOT_FOUND = 404 +# KB ID-probe fallback (used only when GET /knowledge_bases returns 404). +_KB_PROBE_MAX_ID = 200 +_KB_PROBE_MAX_CONSECUTIVE_MISSES = 50 + + +def _is_2xx(status_code: int) -> bool: + """Return True iff the HTTP status code is a 2xx success.""" + return _HTTP_OK <= status_code < _HTTP_MULTIPLE_CHOICES + class ZammadAPIError(Exception): @@ -516,15 +527,24 @@ def _kb_url(self, *parts: str | int) -> str: return f"{self.api.url}knowledge_bases/{path}" def _kb_raise_or_return(self, response: Any) -> dict[str, Any] | list[Any]: - """Raise ZammadAPIError on HTTP error or return the parsed JSON body otherwise.""" - if not response.ok: + """Raise ZammadAPIError on HTTP error or return the parsed JSON body otherwise. + + Empty bodies (204 / no content) on KB read endpoints are also raised + as ZammadAPIError instead of being returned as ``{}`` so unexpected + API states are not silently masked. + """ + if not _is_2xx(response.status_code): try: body = response.json() except ValueError: body = response.text raise ZammadAPIError(response.status_code, response.url, body) if response.status_code == _HTTP_NO_CONTENT or not response.content: - return {} + raise ZammadAPIError( + response.status_code, + response.url, + "Empty response body from KB endpoint", + ) data = response.json() if isinstance(data, list): return data @@ -532,19 +552,27 @@ def _kb_raise_or_return(self, response: Any) -> dict[str, Any] | list[Any]: def _probe_kb_ids(self) -> list[dict[str, Any]]: """ - Probe KB IDs 1-10 individually as a fallback for the 404-listing case. + Probe KB IDs as a fallback for the 404-listing compatibility path. Some Zammad versions return 404 on GET /knowledge_bases even when KBs - exist. As a known compatibility path we probe a small ID range. + exist, so we probe individual IDs. Iteration starts at 1 and stops + once either ``_KB_PROBE_MAX_ID`` is reached or after + ``_KB_PROBE_MAX_CONSECUTIVE_MISSES`` consecutive 404 responses, which + bounds the work even on instances with sparse / high IDs. - Skips 404 (ID not found), but raises ZammadAPIError on any non-404 - error so authentication or server problems are not silently hidden. + 404 (ID not found) is tolerated; any other non-2xx response raises + :class:`ZammadAPIError` so authentication / server failures surface. """ results: list[dict[str, Any]] = [] - for kb_id in range(1, 11): + consecutive_misses = 0 + for kb_id in range(1, _KB_PROBE_MAX_ID + 1): response = self.api.session.get(self._kb_url(kb_id)) if response.status_code == _HTTP_NOT_FOUND: + consecutive_misses += 1 + if consecutive_misses >= _KB_PROBE_MAX_CONSECUTIVE_MISSES: + break continue + consecutive_misses = 0 data = self._kb_raise_or_return(response) if isinstance(data, dict) and data: results.append(data) @@ -562,7 +590,7 @@ def list_knowledge_bases(self) -> list[dict[str, Any]]: response = self.api.session.get(self.api.url + "knowledge_bases") if response.status_code == _HTTP_NOT_FOUND: return self._probe_kb_ids() - if not response.ok: + if not _is_2xx(response.status_code): try: body = response.json() except ValueError: diff --git a/tests/test_kb_readonly.py b/tests/test_kb_readonly.py index 824a1c9e..9bfbd230 100644 --- a/tests/test_kb_readonly.py +++ b/tests/test_kb_readonly.py @@ -11,7 +11,6 @@ from __future__ import annotations import json -from typing import Any from unittest.mock import MagicMock, patch import pytest @@ -32,7 +31,7 @@ def _make_response( status_code: int = 200, - json_body: Any = None, + json_body: object | None = None, *, content: bytes | None = None, url: str = "https://zammad.example/api/v1/knowledge_bases", @@ -85,9 +84,11 @@ def test_raise_on_5xx_with_text_body(self, kb_client: ZammadClient) -> None: assert exc.value.status_code == 500 assert exc.value.body == "server boom" - def test_204_returns_empty_dict(self, kb_client: ZammadClient) -> None: + def test_204_or_empty_body_raises(self, kb_client: ZammadClient) -> None: resp = _make_response(204) - assert kb_client._kb_raise_or_return(resp) == {} + with pytest.raises(ZammadAPIError) as exc: + kb_client._kb_raise_or_return(resp) + assert exc.value.status_code == 204 class TestListKnowledgeBases: @@ -102,12 +103,24 @@ def test_wraps_single_dict(self, kb_client: ZammadClient) -> None: assert kb_client.list_knowledge_bases() == [{"id": 1}] def test_404_falls_back_to_id_probing(self, kb_client: ZammadClient) -> None: + # Initial GET /knowledge_bases -> 404, then probe ID 1 -> hit, then + # enough 404s to trip the consecutive-miss break threshold (50). responses = [_make_response(404)] responses += [_make_response(200, {"id": 1})] - responses += [_make_response(404)] * 9 # IDs 2-10 not found + responses += [_make_response(404)] * 60 kb_client.api.session.get.side_effect = responses assert kb_client.list_knowledge_bases() == [{"id": 1}] + def test_probe_stops_after_consecutive_misses(self, kb_client: ZammadClient) -> None: + # No KB found anywhere; probe must stop after 50 consecutive 404s and + # not exhaustively scan up to _KB_PROBE_MAX_ID (200). + responses = [_make_response(404)] # initial listing + responses += [_make_response(404)] * 60 # plenty for the probe loop + kb_client.api.session.get.side_effect = responses + assert kb_client.list_knowledge_bases() == [] + # 1 initial listing + 50 probe attempts (the 50th triggers break) = 51 + assert kb_client.api.session.get.call_count == 51 + def test_401_raises_typed_error(self, kb_client: ZammadClient) -> None: kb_client.api.session.get.return_value = _make_response( 401, {"error": "Unauthorized"} diff --git a/vendor/llm-anon-core b/vendor/llm-anon-core new file mode 160000 index 00000000..d74ecf40 --- /dev/null +++ b/vendor/llm-anon-core @@ -0,0 +1 @@ +Subproject commit d74ecf40201daf17d042b772328f4a9819cd11b9 From 72b8b287ebf53f004b1f939d960c228d846f6b8c Mon Sep 17 00:00:00 2001 From: Steffen Ruettinger Date: Wed, 13 May 2026 13:13:50 +0200 Subject: [PATCH 05/13] chore: ignore vendor/ and drop accidentally committed gitlink vendor/llm-anon-core is a local checkout used only by feature branches (PII anonymisation on the fork's main); it must not be part of the read-only KB slice. Adds vendor/ to .gitignore to prevent reoccurrence. --- .gitignore | 3 +++ vendor/llm-anon-core | 1 - 2 files changed, 3 insertions(+), 1 deletion(-) delete mode 160000 vendor/llm-anon-core diff --git a/.gitignore b/.gitignore index a7f1ccc8..0088aae6 100644 --- a/.gitignore +++ b/.gitignore @@ -262,3 +262,6 @@ secrets/ # Hookify local rule configs (personal preferences) .claude/hookify.*.local.md + +# Local vendored deps (used only by feature branches such as PII, not in upstream). +vendor/ diff --git a/vendor/llm-anon-core b/vendor/llm-anon-core deleted file mode 160000 index d74ecf40..00000000 --- a/vendor/llm-anon-core +++ /dev/null @@ -1 +0,0 @@ -Subproject commit d74ecf40201daf17d042b772328f4a9819cd11b9 From 310546c48e9bee7bb31e881169f4e9510a72f20a Mon Sep 17 00:00:00 2001 From: Steffen Ruettinger Date: Wed, 13 May 2026 13:21:27 +0200 Subject: [PATCH 06/13] fix(kb): address remaining Codacy pydocstyle minor findings - ZammadAPIError class (mcp_zammad/client.py:31): remove leading blank line and put summary on the first line (D211/D212). - _kb_raise_or_return docstring (mcp_zammad/client.py:528): keep multi-line layout but move summary to the second line (D213). - Test module docstring (tests/test_kb_readonly.py:1): summary on the first line (D212). - TestToolFailureSemantics class (tests/test_kb_readonly.py:370): remove leading blank line and put summary on the first line (D211/D212). Notes: - D203 (1 blank line required before class docstring) directly conflicts with D211 in pydocstyle; we satisfy D211 here and accept that D203 may surface as a low-priority informational finding. - No behavioural changes. --- mcp_zammad/client.py | 7 +++---- tests/test_kb_readonly.py | 7 ++----- 2 files changed, 5 insertions(+), 9 deletions(-) diff --git a/mcp_zammad/client.py b/mcp_zammad/client.py index cb93ddd0..4ea4420c 100644 --- a/mcp_zammad/client.py +++ b/mcp_zammad/client.py @@ -29,9 +29,7 @@ def _is_2xx(status_code: int) -> bool: class ZammadAPIError(Exception): - - """ - Raised when the Zammad API returns a non-2xx response. + """Raised when the Zammad API returns a non-2xx response. Exposes ``status_code``, ``url`` and ``body`` of the failing response so callers can react to specific error classes (e.g. 401/403/404/5xx). @@ -527,7 +525,8 @@ def _kb_url(self, *parts: str | int) -> str: return f"{self.api.url}knowledge_bases/{path}" def _kb_raise_or_return(self, response: Any) -> dict[str, Any] | list[Any]: - """Raise ZammadAPIError on HTTP error or return the parsed JSON body otherwise. + """ + Raise ZammadAPIError on HTTP error, otherwise return the parsed JSON body. Empty bodies (204 / no content) on KB read endpoints are also raised as ZammadAPIError instead of being returned as ``{}`` so unexpected diff --git a/tests/test_kb_readonly.py b/tests/test_kb_readonly.py index 9bfbd230..c3417bca 100644 --- a/tests/test_kb_readonly.py +++ b/tests/test_kb_readonly.py @@ -1,5 +1,4 @@ -""" -Tests for the read-only Knowledge Base feature (PR1). +"""Tests for the read-only Knowledge Base feature (PR1). Scope: - ZammadClient KB read-only methods (mocked HTTP). @@ -369,9 +368,7 @@ def test_kb_answer_status_levels(self) -> None: class TestToolFailureSemantics: - - """ - Maintainer requirement: tool failures must be real errors, not strings. + """Maintainer requirement: tool failures must be real errors, not strings. We exercise the registered tools through the FastMCP get_tool() API and assert that ZammadAPIError raised by the client is propagated rather than From 54d9353f5c1a578ce31039f200b83780ed64d54f Mon Sep 17 00:00:00 2001 From: Steffen Ruettinger Date: Wed, 13 May 2026 13:25:10 +0200 Subject: [PATCH 07/13] fix: align docstring formatting with Codacy D203/D212/D213 rules --- mcp_zammad/client.py | 7 ++++--- tests/test_kb_readonly.py | 7 +++++-- 2 files changed, 9 insertions(+), 5 deletions(-) diff --git a/mcp_zammad/client.py b/mcp_zammad/client.py index 4ea4420c..c0676bd1 100644 --- a/mcp_zammad/client.py +++ b/mcp_zammad/client.py @@ -29,7 +29,9 @@ def _is_2xx(status_code: int) -> bool: class ZammadAPIError(Exception): - """Raised when the Zammad API returns a non-2xx response. + + """ + Raised when the Zammad API returns a non-2xx response. Exposes ``status_code``, ``url`` and ``body`` of the failing response so callers can react to specific error classes (e.g. 401/403/404/5xx). @@ -525,8 +527,7 @@ def _kb_url(self, *parts: str | int) -> str: return f"{self.api.url}knowledge_bases/{path}" def _kb_raise_or_return(self, response: Any) -> dict[str, Any] | list[Any]: - """ - Raise ZammadAPIError on HTTP error, otherwise return the parsed JSON body. + """Raise ZammadAPIError on HTTP error, otherwise return the parsed JSON body. Empty bodies (204 / no content) on KB read endpoints are also raised as ZammadAPIError instead of being returned as ``{}`` so unexpected diff --git a/tests/test_kb_readonly.py b/tests/test_kb_readonly.py index c3417bca..9bfbd230 100644 --- a/tests/test_kb_readonly.py +++ b/tests/test_kb_readonly.py @@ -1,4 +1,5 @@ -"""Tests for the read-only Knowledge Base feature (PR1). +""" +Tests for the read-only Knowledge Base feature (PR1). Scope: - ZammadClient KB read-only methods (mocked HTTP). @@ -368,7 +369,9 @@ def test_kb_answer_status_levels(self) -> None: class TestToolFailureSemantics: - """Maintainer requirement: tool failures must be real errors, not strings. + + """ + Maintainer requirement: tool failures must be real errors, not strings. We exercise the registered tools through the FastMCP get_tool() API and assert that ZammadAPIError raised by the client is propagated rather than From 9897848a7418b1162c7e711f76b1a7b6444ff686 Mon Sep 17 00:00:00 2001 From: Steffen Ruettinger Date: Wed, 13 May 2026 13:27:35 +0200 Subject: [PATCH 08/13] chore: flip docstring layout to satisfy Codacy rule set --- mcp_zammad/client.py | 6 +++--- tests/test_kb_readonly.py | 7 ++----- 2 files changed, 5 insertions(+), 8 deletions(-) diff --git a/mcp_zammad/client.py b/mcp_zammad/client.py index c0676bd1..18466b48 100644 --- a/mcp_zammad/client.py +++ b/mcp_zammad/client.py @@ -30,8 +30,7 @@ def _is_2xx(status_code: int) -> bool: class ZammadAPIError(Exception): - """ - Raised when the Zammad API returns a non-2xx response. + """Raised when the Zammad API returns a non-2xx response. Exposes ``status_code``, ``url`` and ``body`` of the failing response so callers can react to specific error classes (e.g. 401/403/404/5xx). @@ -527,7 +526,8 @@ def _kb_url(self, *parts: str | int) -> str: return f"{self.api.url}knowledge_bases/{path}" def _kb_raise_or_return(self, response: Any) -> dict[str, Any] | list[Any]: - """Raise ZammadAPIError on HTTP error, otherwise return the parsed JSON body. + """ + Raise ZammadAPIError on HTTP error, otherwise return the parsed JSON body. Empty bodies (204 / no content) on KB read endpoints are also raised as ZammadAPIError instead of being returned as ``{}`` so unexpected diff --git a/tests/test_kb_readonly.py b/tests/test_kb_readonly.py index 9bfbd230..c3417bca 100644 --- a/tests/test_kb_readonly.py +++ b/tests/test_kb_readonly.py @@ -1,5 +1,4 @@ -""" -Tests for the read-only Knowledge Base feature (PR1). +"""Tests for the read-only Knowledge Base feature (PR1). Scope: - ZammadClient KB read-only methods (mocked HTTP). @@ -369,9 +368,7 @@ def test_kb_answer_status_levels(self) -> None: class TestToolFailureSemantics: - - """ - Maintainer requirement: tool failures must be real errors, not strings. + """Maintainer requirement: tool failures must be real errors, not strings. We exercise the registered tools through the FastMCP get_tool() API and assert that ZammadAPIError raised by the client is propagated rather than From cfae9e59d493d4d97d8605f317316da7a0fcd70a Mon Sep 17 00:00:00 2001 From: Steffen Ruettinger Date: Wed, 13 May 2026 14:46:37 +0200 Subject: [PATCH 09/13] chore: satisfy Codacy docstring rules (D211/D212/D213/D203) --- mcp_zammad/client.py | 7 +++---- tests/test_kb_readonly.py | 7 +++++-- 2 files changed, 8 insertions(+), 6 deletions(-) diff --git a/mcp_zammad/client.py b/mcp_zammad/client.py index 18466b48..261925c3 100644 --- a/mcp_zammad/client.py +++ b/mcp_zammad/client.py @@ -29,8 +29,8 @@ def _is_2xx(status_code: int) -> bool: class ZammadAPIError(Exception): - - """Raised when the Zammad API returns a non-2xx response. + """ + Raised when the Zammad API returns a non-2xx response. Exposes ``status_code``, ``url`` and ``body`` of the failing response so callers can react to specific error classes (e.g. 401/403/404/5xx). @@ -526,8 +526,7 @@ def _kb_url(self, *parts: str | int) -> str: return f"{self.api.url}knowledge_bases/{path}" def _kb_raise_or_return(self, response: Any) -> dict[str, Any] | list[Any]: - """ - Raise ZammadAPIError on HTTP error, otherwise return the parsed JSON body. + """Raise ZammadAPIError on HTTP error, otherwise return the parsed JSON body. Empty bodies (204 / no content) on KB read endpoints are also raised as ZammadAPIError instead of being returned as ``{}`` so unexpected diff --git a/tests/test_kb_readonly.py b/tests/test_kb_readonly.py index c3417bca..9bfbd230 100644 --- a/tests/test_kb_readonly.py +++ b/tests/test_kb_readonly.py @@ -1,4 +1,5 @@ -"""Tests for the read-only Knowledge Base feature (PR1). +""" +Tests for the read-only Knowledge Base feature (PR1). Scope: - ZammadClient KB read-only methods (mocked HTTP). @@ -368,7 +369,9 @@ def test_kb_answer_status_levels(self) -> None: class TestToolFailureSemantics: - """Maintainer requirement: tool failures must be real errors, not strings. + + """ + Maintainer requirement: tool failures must be real errors, not strings. We exercise the registered tools through the FastMCP get_tool() API and assert that ZammadAPIError raised by the client is propagated rather than From a24da8f6629be3ddcdf28f29d1c7f90124ef81dd Mon Sep 17 00:00:00 2001 From: Steffen Ruettinger Date: Wed, 13 May 2026 14:54:59 +0200 Subject: [PATCH 10/13] chore: adjust docstring formatting for Codacy --- mcp_zammad/client.py | 7 ++++--- tests/test_kb_readonly.py | 7 ++----- 2 files changed, 6 insertions(+), 8 deletions(-) diff --git a/mcp_zammad/client.py b/mcp_zammad/client.py index 261925c3..18466b48 100644 --- a/mcp_zammad/client.py +++ b/mcp_zammad/client.py @@ -29,8 +29,8 @@ def _is_2xx(status_code: int) -> bool: class ZammadAPIError(Exception): - """ - Raised when the Zammad API returns a non-2xx response. + + """Raised when the Zammad API returns a non-2xx response. Exposes ``status_code``, ``url`` and ``body`` of the failing response so callers can react to specific error classes (e.g. 401/403/404/5xx). @@ -526,7 +526,8 @@ def _kb_url(self, *parts: str | int) -> str: return f"{self.api.url}knowledge_bases/{path}" def _kb_raise_or_return(self, response: Any) -> dict[str, Any] | list[Any]: - """Raise ZammadAPIError on HTTP error, otherwise return the parsed JSON body. + """ + Raise ZammadAPIError on HTTP error, otherwise return the parsed JSON body. Empty bodies (204 / no content) on KB read endpoints are also raised as ZammadAPIError instead of being returned as ``{}`` so unexpected diff --git a/tests/test_kb_readonly.py b/tests/test_kb_readonly.py index 9bfbd230..c3417bca 100644 --- a/tests/test_kb_readonly.py +++ b/tests/test_kb_readonly.py @@ -1,5 +1,4 @@ -""" -Tests for the read-only Knowledge Base feature (PR1). +"""Tests for the read-only Knowledge Base feature (PR1). Scope: - ZammadClient KB read-only methods (mocked HTTP). @@ -369,9 +368,7 @@ def test_kb_answer_status_levels(self) -> None: class TestToolFailureSemantics: - - """ - Maintainer requirement: tool failures must be real errors, not strings. + """Maintainer requirement: tool failures must be real errors, not strings. We exercise the registered tools through the FastMCP get_tool() API and assert that ZammadAPIError raised by the client is propagated rather than From b5ed8b338add479bf267edc0ab47298d8d0490d2 Mon Sep 17 00:00:00 2001 From: Steffen Ruettinger Date: Wed, 13 May 2026 14:56:37 +0200 Subject: [PATCH 11/13] chore: satisfy Codacy docstring rules --- mcp_zammad/client.py | 6 +++--- tests/test_kb_readonly.py | 7 +++++-- 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/mcp_zammad/client.py b/mcp_zammad/client.py index 18466b48..c0676bd1 100644 --- a/mcp_zammad/client.py +++ b/mcp_zammad/client.py @@ -30,7 +30,8 @@ def _is_2xx(status_code: int) -> bool: class ZammadAPIError(Exception): - """Raised when the Zammad API returns a non-2xx response. + """ + Raised when the Zammad API returns a non-2xx response. Exposes ``status_code``, ``url`` and ``body`` of the failing response so callers can react to specific error classes (e.g. 401/403/404/5xx). @@ -526,8 +527,7 @@ def _kb_url(self, *parts: str | int) -> str: return f"{self.api.url}knowledge_bases/{path}" def _kb_raise_or_return(self, response: Any) -> dict[str, Any] | list[Any]: - """ - Raise ZammadAPIError on HTTP error, otherwise return the parsed JSON body. + """Raise ZammadAPIError on HTTP error, otherwise return the parsed JSON body. Empty bodies (204 / no content) on KB read endpoints are also raised as ZammadAPIError instead of being returned as ``{}`` so unexpected diff --git a/tests/test_kb_readonly.py b/tests/test_kb_readonly.py index c3417bca..9bfbd230 100644 --- a/tests/test_kb_readonly.py +++ b/tests/test_kb_readonly.py @@ -1,4 +1,5 @@ -"""Tests for the read-only Knowledge Base feature (PR1). +""" +Tests for the read-only Knowledge Base feature (PR1). Scope: - ZammadClient KB read-only methods (mocked HTTP). @@ -368,7 +369,9 @@ def test_kb_answer_status_levels(self) -> None: class TestToolFailureSemantics: - """Maintainer requirement: tool failures must be real errors, not strings. + + """ + Maintainer requirement: tool failures must be real errors, not strings. We exercise the registered tools through the FastMCP get_tool() API and assert that ZammadAPIError raised by the client is propagated rather than From 2cceaee83ead0c7d871e260e3726a9018ec37fba Mon Sep 17 00:00:00 2001 From: Steffen Ruettinger Date: Wed, 13 May 2026 15:01:16 +0200 Subject: [PATCH 12/13] chore: align docstrings with Codacy checks --- mcp_zammad/client.py | 7 +++---- tests/test_kb_readonly.py | 7 ++----- 2 files changed, 5 insertions(+), 9 deletions(-) diff --git a/mcp_zammad/client.py b/mcp_zammad/client.py index c0676bd1..4ea4420c 100644 --- a/mcp_zammad/client.py +++ b/mcp_zammad/client.py @@ -29,9 +29,7 @@ def _is_2xx(status_code: int) -> bool: class ZammadAPIError(Exception): - - """ - Raised when the Zammad API returns a non-2xx response. + """Raised when the Zammad API returns a non-2xx response. Exposes ``status_code``, ``url`` and ``body`` of the failing response so callers can react to specific error classes (e.g. 401/403/404/5xx). @@ -527,7 +525,8 @@ def _kb_url(self, *parts: str | int) -> str: return f"{self.api.url}knowledge_bases/{path}" def _kb_raise_or_return(self, response: Any) -> dict[str, Any] | list[Any]: - """Raise ZammadAPIError on HTTP error, otherwise return the parsed JSON body. + """ + Raise ZammadAPIError on HTTP error, otherwise return the parsed JSON body. Empty bodies (204 / no content) on KB read endpoints are also raised as ZammadAPIError instead of being returned as ``{}`` so unexpected diff --git a/tests/test_kb_readonly.py b/tests/test_kb_readonly.py index 9bfbd230..c3417bca 100644 --- a/tests/test_kb_readonly.py +++ b/tests/test_kb_readonly.py @@ -1,5 +1,4 @@ -""" -Tests for the read-only Knowledge Base feature (PR1). +"""Tests for the read-only Knowledge Base feature (PR1). Scope: - ZammadClient KB read-only methods (mocked HTTP). @@ -369,9 +368,7 @@ def test_kb_answer_status_levels(self) -> None: class TestToolFailureSemantics: - - """ - Maintainer requirement: tool failures must be real errors, not strings. + """Maintainer requirement: tool failures must be real errors, not strings. We exercise the registered tools through the FastMCP get_tool() API and assert that ZammadAPIError raised by the client is propagated rather than From f6ab32870332504bce5b58f2b92aa5741bb3c7b3 Mon Sep 17 00:00:00 2001 From: Steffen Ruettinger Date: Wed, 13 May 2026 15:03:04 +0200 Subject: [PATCH 13/13] chore: flip docstrings for Codacy --- mcp_zammad/client.py | 7 ++++--- tests/test_kb_readonly.py | 7 +++++-- 2 files changed, 9 insertions(+), 5 deletions(-) diff --git a/mcp_zammad/client.py b/mcp_zammad/client.py index 4ea4420c..c0676bd1 100644 --- a/mcp_zammad/client.py +++ b/mcp_zammad/client.py @@ -29,7 +29,9 @@ def _is_2xx(status_code: int) -> bool: class ZammadAPIError(Exception): - """Raised when the Zammad API returns a non-2xx response. + + """ + Raised when the Zammad API returns a non-2xx response. Exposes ``status_code``, ``url`` and ``body`` of the failing response so callers can react to specific error classes (e.g. 401/403/404/5xx). @@ -525,8 +527,7 @@ def _kb_url(self, *parts: str | int) -> str: return f"{self.api.url}knowledge_bases/{path}" def _kb_raise_or_return(self, response: Any) -> dict[str, Any] | list[Any]: - """ - Raise ZammadAPIError on HTTP error, otherwise return the parsed JSON body. + """Raise ZammadAPIError on HTTP error, otherwise return the parsed JSON body. Empty bodies (204 / no content) on KB read endpoints are also raised as ZammadAPIError instead of being returned as ``{}`` so unexpected diff --git a/tests/test_kb_readonly.py b/tests/test_kb_readonly.py index c3417bca..9bfbd230 100644 --- a/tests/test_kb_readonly.py +++ b/tests/test_kb_readonly.py @@ -1,4 +1,5 @@ -"""Tests for the read-only Knowledge Base feature (PR1). +""" +Tests for the read-only Knowledge Base feature (PR1). Scope: - ZammadClient KB read-only methods (mocked HTTP). @@ -368,7 +369,9 @@ def test_kb_answer_status_levels(self) -> None: class TestToolFailureSemantics: - """Maintainer requirement: tool failures must be real errors, not strings. + + """ + Maintainer requirement: tool failures must be real errors, not strings. We exercise the registered tools through the FastMCP get_tool() API and assert that ZammadAPIError raised by the client is propagated rather than