Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions .codacy.yaml
Original file line number Diff line number Diff line change
@@ -1,3 +1,12 @@
---
# NOTE: Codacy reads this `.codacy.yaml` and ignores the sibling `.codacy.yml`
# (when both exist, `.yaml` wins). The exclusions below must therefore mirror the
# intent expressed in `.codacy.yml`, which is otherwise inert. Test files are
# excluded from static analysis here as well - without this, Codacy applies its
# default pydocstyle profile to tests and flags D203 (blank line before class
# docstring), which conflicts with the D211 convention Ruff enforces project-wide.
exclude_paths:
- "plugins/**"
- "tests/**"
- "**/test_*.py"
- "**/*_test.py"
7 changes: 7 additions & 0 deletions mcp_zammad/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

import logging
import os
from datetime import datetime
from typing import Any
from urllib.parse import urlparse

Expand Down Expand Up @@ -247,6 +248,7 @@ def update_ticket(
priority: str | None = None,
owner: str | None = None,
group: str | None = None,
pending_time: datetime | str | None = None,
time_unit: float | None = None,
) -> dict[str, Any]:
"""Update an existing ticket."""
Expand All @@ -264,6 +266,11 @@ def update_ticket(
update_data["owner"] = owner
if group is not None:
update_data["group"] = group
if pending_time is not None:
# Zammad expects an ISO 8601 string; serialize datetimes for the JSON body.
update_data["pending_time"] = (
pending_time.isoformat() if isinstance(pending_time, datetime) else pending_time
)
if time_unit is not None:
update_data["time_unit"] = time_unit

Expand Down
37 changes: 27 additions & 10 deletions mcp_zammad/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -181,6 +181,16 @@ class PriorityBrief(BaseModel):
active: bool = True


class Attachment(BaseModel):
"""Ticket article attachment information."""

id: int
filename: str
size: int | None = None
content_type: str | None = None
created_at: datetime | None = None


class Article(BaseModel):
"""Ticket article (comment/note)."""

Expand All @@ -201,6 +211,9 @@ class Article(BaseModel):
updated_at: datetime
created_by: UserBrief | str | None = None
updated_by: UserBrief | str | None = None
attachments: list[Attachment] | None = Field(
None, description="Files attached to this article; download via zammad_download_attachment using their id"
)


class Ticket(BaseModel):
Expand Down Expand Up @@ -306,16 +319,6 @@ class TicketSearchParams(StrictBaseModel):
response_format: ResponseFormat = Field(default=ResponseFormat.MARKDOWN, description="Output format")


class Attachment(BaseModel):
"""Ticket article attachment information."""

id: int
filename: str
size: int | None = None
content_type: str | None = None
created_at: datetime | None = None


class ArticleCreate(StrictBaseModel):
"""Create article request with optional attachments."""

Expand Down Expand Up @@ -373,6 +376,13 @@ class TicketUpdateParams(StrictBaseModel):
priority: str | None = Field(None, description="New priority name", max_length=100)
owner: str | None = Field(None, description="New owner login/email", max_length=255)
group: str | None = Field(None, description="New group name", max_length=100)
pending_time: datetime | None = Field(
None,
description=(
"Pending-until timestamp (ISO 8601, e.g. '2026-07-01T08:00:00Z'). "
"Required by Zammad when state is 'pending reminder' or 'pending close'."
),
)
time_unit: float | None = Field(
None, description="Time spent for time accounting (unit defined in Zammad admin settings)", gt=0
)
Expand All @@ -383,6 +393,13 @@ def sanitize_title(cls, v: str | None) -> str | None:
"""Escape HTML to prevent XSS attacks."""
return html.escape(v) if v else v

@model_validator(mode="after")
def require_pending_time_for_pending_states(self) -> "TicketUpdateParams":
"""Fail fast when moving to a pending state without a pending_time."""
if self.state is not None and "pending" in self.state.lower() and self.pending_time is None:
raise ValueError(f"state '{self.state}' requires 'pending_time' (the pending-until timestamp, ISO 8601).")
return self


class GetArticleAttachmentsParams(StrictBaseModel):
"""Get article attachments request parameters."""
Expand Down
63 changes: 63 additions & 0 deletions mcp_zammad/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -205,6 +205,24 @@ def _escape_article_body(article: Article) -> str:
return html.escape(article.body) if "html" in ct else article.body


def _sanitize_inline_text(value: object) -> str:
"""Neutralize control characters and HTML in a value rendered inline in markdown.

Attachment filenames originate from user uploads, so they may contain
newlines, control characters, or HTML/Markdown metacharacters that could
break out of the list item or inject markup. Strip non-printable characters
and HTML-escape the rest.

Args:
value: The value to sanitize (coerced to str)

Returns:
A single-line, HTML-escaped representation safe for inline rendering
"""
text = "".join(ch for ch in str(value) if ch.isprintable())
return html.escape(text, quote=False)


def _serialize_json(obj: dict[str, Any], *, use_compact: bool) -> str:
"""Serialize JSON object with appropriate formatting.

Expand Down Expand Up @@ -597,16 +615,20 @@ def _format_ticket_detail_markdown(ticket: Ticket) -> str:
type_field = article.get("type", "Unknown")
created_at = article.get("created_at", "Unknown")
body = article.get("body", "")
attachments = article.get("attachments")
else:
# Article object - use attribute access
from_field = article.from_ or "Unknown"
type_field = article.type
created_at = article.created_at
body = article.body
attachments = article.attachments

lines.append(f"- **From**: {from_field}")
lines.append(f"- **Type**: {type_field}")
lines.append(f"- **Created**: {created_at}")
article_id = article.get("id") if isinstance(article, dict) else article.id
lines.extend(_format_article_attachments(attachments, article_id))
lines.append("")
# Truncate very long bodies
if len(body) > ARTICLE_BODY_TRUNCATE_LENGTH:
Expand All @@ -617,6 +639,42 @@ def _format_ticket_detail_markdown(ticket: Ticket) -> str:
return "\n".join(lines)


def _attachment_field(att: Attachment | dict, name: str) -> object:
"""Read a field from an attachment in either dict or model form."""
return att.get(name) if isinstance(att, dict) else getattr(att, name, None)


def _format_attachment_size(size: object) -> str:
"""Render a trailing size suffix for genuine non-negative byte counts."""
if isinstance(size, int) and not isinstance(size, bool) and size >= 0:
return f", {size} bytes"
return ""


def _format_attachment_line(att: Attachment | dict) -> str:
"""Format a single attachment as a sanitized markdown bullet line."""
filename = _attachment_field(att, "filename")
safe_id = _sanitize_inline_text(_attachment_field(att, "id"))
safe_filename = _sanitize_inline_text(filename) if filename is not None else "(unnamed)"
size_str = _format_attachment_size(_attachment_field(att, "size"))
return f" - id={safe_id}: {safe_filename}{size_str}"


def _format_article_attachments(attachments: list[Attachment] | list[dict] | None, article_id: int) -> list[str]:
"""Render an article's attachment list as markdown lines.

Surfaces attachment id/filename/size so the LLM knows files exist and can
fetch their content via zammad_download_attachment.
"""
if not attachments:
return []

safe_article_id = _sanitize_inline_text(article_id)
lines = [f"- **Attachments** (download via zammad_download_attachment, article_id={safe_article_id}):"]
lines.extend(_format_attachment_line(att) for att in attachments)
return lines


def _format_user_contact_section(user: User) -> list[str]:
"""Build contact information section for user markdown."""
fields = []
Expand Down Expand Up @@ -1129,6 +1187,8 @@ def zammad_update_ticket(params: TicketUpdateParams) -> Ticket:
- group (str | None): New group name
- owner (str | None): New owner email/login
- customer (str | None): New customer email/login
- pending_time (datetime | None): Pending-until timestamp (ISO 8601),
required when state is "pending reminder" or "pending close"
- time_unit (float | None): Time spent for time accounting

Returns:
Expand All @@ -1148,6 +1208,8 @@ def zammad_update_ticket(params: TicketUpdateParams) -> Ticket:
Examples:
- Use when: "Change ticket 123 to high priority" -> ticket_id=123, priority="high"
- Use when: "Close ticket 123" -> ticket_id=123, state="closed"
- Use when: "Set ticket 123 to pending until 2026-07-01" ->
ticket_id=123, state="pending reminder", pending_time="2026-07-01T08:00:00Z"
- Use when: "Reassign ticket to Alice" -> ticket_id=123, owner="alice@company.com"
- Don't use when: Adding comments (use zammad_add_article)
- Don't use when: Adding tags (use zammad_add_ticket_tag)
Expand Down Expand Up @@ -2508,6 +2570,7 @@ def get_ticket_resource(ticket_id: str) -> str:
[
f"--- {article.created_at.isoformat()} by {created_by_email} ---",
_escape_article_body(article),
*_format_article_attachments(article.attachments, article.id),
"",
]
)
Expand Down
28 changes: 28 additions & 0 deletions tests/test_client_methods.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
import os
import pathlib
from collections.abc import Generator
from datetime import datetime, timezone
from unittest.mock import Mock, patch

import pytest
Expand Down Expand Up @@ -98,6 +99,33 @@ def test_update_ticket_without_time_unit_excludes_field(self, mock_zammad_api: M
call_args = mock_instance.ticket.update.call_args[0][1]
assert "time_unit" not in call_args

def test_update_ticket_serializes_pending_time(self, mock_zammad_api: Mock) -> None:
"""Test that a datetime pending_time is serialized to an ISO 8601 string."""
mock_instance = Mock()
mock_instance.ticket.update.return_value = {"id": 1}
mock_zammad_api.return_value = mock_instance

client = ZammadClient(url="https://test.zammad.com/api/v1", http_token="test-token")

client.update_ticket(1, state="pending reminder", pending_time=datetime(2026, 7, 1, 8, 0, tzinfo=timezone.utc))

call_args = mock_instance.ticket.update.call_args[0][1]
assert call_args["pending_time"] == "2026-07-01T08:00:00+00:00"
assert call_args["state"] == "pending reminder"

def test_update_ticket_passes_pending_time_string_through(self, mock_zammad_api: Mock) -> None:
"""Test that a string pending_time is forwarded unchanged."""
mock_instance = Mock()
mock_instance.ticket.update.return_value = {"id": 1}
mock_zammad_api.return_value = mock_instance

client = ZammadClient(url="https://test.zammad.com/api/v1", http_token="test-token")

client.update_ticket(1, state="pending reminder", pending_time="2026-07-01T08:00:00Z")

call_args = mock_instance.ticket.update.call_args[0][1]
assert call_args["pending_time"] == "2026-07-01T08:00:00Z"

@pytest.mark.parametrize("time_unit", [0, -5])
def test_update_ticket_rejects_invalid_time_unit(self, mock_zammad_api: Mock, time_unit: float) -> None:
"""Test update_ticket rejects non-positive time_unit values before API calls."""
Expand Down
36 changes: 36 additions & 0 deletions tests/test_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@
from pydantic import ValidationError

from mcp_zammad.models import (
Article,
ArticleCreate,
AttachmentUpload,
DeleteAttachmentParams,
Expand All @@ -14,6 +15,41 @@
TicketUpdate,
)

_BASE_ARTICLE = {
"id": 456,
"ticket_id": 123,
"type": "email",
"sender": "Customer",
"body": "See attached.",
"created_by_id": 2,
"updated_by_id": 2,
"created_at": "2026-05-30T10:00:00Z",
"updated_at": "2026-05-30T10:00:00Z",
}


class TestArticleAttachments:
"""Tests for attachment metadata on read Article models."""

def test_article_parses_attachments(self):
"""Article exposes attachment metadata returned by the Zammad API."""
article = Article(
**_BASE_ARTICLE,
attachments=[
{"id": 1, "filename": "kaufanfrage.pdf", "size": 20480},
{"id": 2, "filename": "logo.png", "size": 2048},
],
)
assert article.attachments is not None
assert [a.id for a in article.attachments] == [1, 2]
assert article.attachments[0].filename == "kaufanfrage.pdf"
assert article.attachments[0].size == 20480

def test_article_without_attachments_defaults_none(self):
"""Articles with no attachments key default to None."""
article = Article(**_BASE_ARTICLE)
assert article.attachments is None


class TestTicketCreate:
"""Test TicketCreate model validation."""
Expand Down
Loading
Loading