airbyte_ops_mcp.mcp.zendesk_ops

  1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
  2"""MCP tools for Zendesk Support operations.
  3
  4This module exposes tools for pulling Zendesk tickets (and their comments) and
  5for posting internal (private, agent-only) notes to a ticket, so support
  6workflows can inspect and annotate ticket context through the hosted Ops MCP
  7server.
  8
  9## MCP reference
 10
 11.. include:: ../../../docs/mcp-generated/zendesk_ops.md
 12    :start-line: 2
 13"""
 14
 15# NOTE: We intentionally do NOT use `from __future__ import annotations` here.
 16# FastMCP has issues resolving forward references when PEP 563 deferred
 17# annotations are used. See: https://github.com/jlowin/fastmcp/issues/905
 18
 19__all__: list[str] = []
 20
 21from typing import Annotated, Any
 22
 23from fastmcp import FastMCP
 24from fastmcp_extensions import mcp_tool, register_mcp_tools
 25from pydantic import BaseModel, Field
 26
 27from airbyte_ops_mcp.zendesk_api import (
 28    ZendeskAPIError,
 29    add_internal_note,
 30    add_ticket_tags,
 31    get_ticket,
 32    get_ticket_comments,
 33)
 34
 35
 36class ZendeskAttachment(BaseModel):
 37    """An attachment on a Zendesk ticket comment."""
 38
 39    id: int | None = Field(default=None, description="Attachment ID.")
 40    file_name: str | None = Field(default=None, description="Original file name.")
 41    content_type: str | None = Field(default=None, description="MIME content type.")
 42    content_url: str | None = Field(
 43        default=None, description="URL to download the attachment content."
 44    )
 45    size: int | None = Field(default=None, description="File size in bytes.")
 46
 47
 48class ZendeskCustomField(BaseModel):
 49    """A single custom field value on a Zendesk ticket."""
 50
 51    id: int | None = Field(default=None, description="Custom field ID.")
 52    value: str | int | float | bool | None = Field(
 53        default=None, description="Custom field value (may be null when unset)."
 54    )
 55
 56
 57class ZendeskComment(BaseModel):
 58    """A single comment on a Zendesk ticket."""
 59
 60    id: int | None = Field(default=None, description="Comment ID.")
 61    author_id: int | None = Field(default=None, description="Author user ID.")
 62    public: bool | None = Field(
 63        default=None,
 64        description="`True` for public replies, `False` for internal notes.",
 65    )
 66    body: str = Field(default="", description="Plain-text comment body.")
 67    created_at: str | None = Field(
 68        default=None,
 69        description="ISO-8601 timestamp when the comment was created.",
 70    )
 71    attachments: list[ZendeskAttachment] = Field(
 72        default_factory=list,
 73        description="Files attached to the comment (metadata only).",
 74    )
 75
 76
 77class ZendeskTicketResponse(BaseModel):
 78    """Response from the `get_zendesk_ticket` tool."""
 79
 80    success: bool = Field(description="Whether the ticket was retrieved.")
 81    message: str = Field(description="Human-readable status message.")
 82    ticket_id: int | None = Field(default=None, description="Zendesk ticket ID.")
 83    subject: str | None = Field(default=None, description="Ticket subject line.")
 84    status: str | None = Field(
 85        default=None,
 86        description="Ticket status (e.g. `open`, `pending`, `solved`, `closed`).",
 87    )
 88    description: str | None = Field(
 89        default=None,
 90        description="The ticket's first comment / description text.",
 91    )
 92    priority: str | None = Field(default=None, description="Ticket priority, if set.")
 93    tags: list[str] = Field(default_factory=list, description="Ticket tags.")
 94    requester_id: int | None = Field(
 95        default=None, description="User ID of the ticket requester."
 96    )
 97    organization_id: int | None = Field(
 98        default=None, description="Organization ID associated with the ticket."
 99    )
100    created_at: str | None = Field(
101        default=None, description="ISO-8601 ticket creation timestamp."
102    )
103    updated_at: str | None = Field(
104        default=None, description="ISO-8601 ticket last-updated timestamp."
105    )
106    url: str | None = Field(
107        default=None,
108        description="Agent-facing Zendesk URL for the ticket.",
109    )
110    via_channel: str | None = Field(
111        default=None,
112        description="Channel the ticket came in through (e.g. `web`, `email`).",
113    )
114    via_source_rel: str | None = Field(
115        default=None,
116        description="Source relationship (e.g. `follow_up` for follow-up tickets).",
117    )
118    follow_up_source_ticket_id: int | None = Field(
119        default=None,
120        description=(
121            "For follow-up tickets, the ID of the original (closed) ticket this "
122            "one follows up on. `None` when the ticket is not a follow-up."
123        ),
124    )
125    custom_fields: list[ZendeskCustomField] = Field(
126        default_factory=list,
127        description="Ticket custom fields (`id`/`value` pairs).",
128    )
129    comments: list[ZendeskComment] = Field(
130        default_factory=list,
131        description="Ticket comments, oldest first. Empty unless requested.",
132    )
133
134
135class ZendeskInternalNoteResponse(BaseModel):
136    """Response from the `post_zendesk_internal_comment` tool."""
137
138    success: bool = Field(description="Whether the internal note was posted.")
139    message: str = Field(description="Human-readable status message.")
140    ticket_id: int | None = Field(default=None, description="Zendesk ticket ID.")
141    comment_id: int | None = Field(
142        default=None,
143        description="ID of the created comment, when Zendesk reports it.",
144    )
145    public: bool = Field(
146        default=False,
147        description="Always `False`: the note is a private, agent-only comment.",
148    )
149
150
151class ZendeskTagsResponse(BaseModel):
152    """Response from the `add_zendesk_ticket_tags` tool."""
153
154    success: bool = Field(description="Whether the tags were added.")
155    message: str = Field(description="Human-readable status message.")
156    ticket_id: int | None = Field(default=None, description="Zendesk ticket ID.")
157    tags: list[str] = Field(
158        default_factory=list,
159        description="The ticket's full tag list after the update.",
160    )
161
162
163def _agent_ticket_url(raw_url: str | None, ticket_id: int | None) -> str | None:
164    """Derive the agent-facing ticket URL from the API `url` field."""
165    if not raw_url or ticket_id is None:
166        return None
167    # API url looks like https://<subdomain>.zendesk.com/api/v2/tickets/123.json
168    if "/api/v2/" not in raw_url:
169        return None
170    host = raw_url.split("/api/v2/", 1)[0]
171    if not host:
172        return None
173    return f"{host}/agent/tickets/{ticket_id}"
174
175
176def _map_attachments(raw_comment: dict[str, Any]) -> list[ZendeskAttachment]:
177    """Map the `attachments` array of a raw comment into typed models."""
178    raw_attachments = raw_comment.get("attachments")
179    if not isinstance(raw_attachments, list):
180        return []
181    return [
182        ZendeskAttachment(
183            id=a.get("id"),
184            file_name=a.get("file_name"),
185            content_type=a.get("content_type"),
186            content_url=a.get("content_url"),
187            size=a.get("size"),
188        )
189        for a in raw_attachments
190        if isinstance(a, dict)
191    ]
192
193
194def _map_custom_fields(ticket: dict[str, Any]) -> list[ZendeskCustomField]:
195    """Map the ticket's `custom_fields` array into typed models."""
196    raw_fields = ticket.get("custom_fields")
197    if not isinstance(raw_fields, list):
198        return []
199    return [
200        ZendeskCustomField(id=f.get("id"), value=f.get("value"))
201        for f in raw_fields
202        if isinstance(f, dict)
203    ]
204
205
206def _as_dict(value: Any) -> dict[str, Any]:
207    """Return `value` when it is a dict, otherwise an empty dict."""
208    return value if isinstance(value, dict) else {}
209
210
211def _follow_up_source_ticket_id(via_source: dict[str, Any]) -> int | None:
212    """Return the original ticket ID when the ticket is a follow-up."""
213    if via_source.get("rel") != "follow_up":
214        return None
215    ticket_id = _as_dict(via_source.get("from")).get("ticket_id")
216    return ticket_id if isinstance(ticket_id, int) else None
217
218
219@mcp_tool(
220    read_only=True,
221    idempotent=True,
222    open_world=True,
223)
224def get_zendesk_ticket(
225    ticket_id: Annotated[
226        int,
227        Field(description="The numeric Zendesk ticket ID to retrieve."),
228    ],
229    include_comments: Annotated[
230        bool,
231        Field(
232            description=(
233                "When `True`, also fetch the ticket's comments (oldest first), "
234                "including any attachment metadata. Only the first page (up to "
235                "100 comments) is returned. Defaults to `False` to keep "
236                "responses small."
237            )
238        ),
239    ] = False,
240) -> ZendeskTicketResponse:
241    """Retrieve a Zendesk Support ticket by its numeric ID.
242
243    Returns the ticket's subject, status, description, tags, and requester/
244    organization identifiers. Set `include_comments` to `True` to also pull the
245    ticket's comment thread (public replies and internal notes), oldest first;
246    only the first page (up to 100 comments) is returned.
247
248    Credentials are read from the server environment (`ZENDESK_SUBDOMAIN`,
249    `ZENDESK_EMAIL`, `ZENDESK_API_TOKEN`); this tool never accepts or logs them.
250    """
251    try:
252        ticket: dict[str, Any] = get_ticket(ticket_id)
253    except ZendeskAPIError as exc:
254        return ZendeskTicketResponse(
255            success=False,
256            message=str(exc),
257            ticket_id=ticket_id,
258        )
259
260    comments: list[ZendeskComment] = []
261    comments_note = ""
262    if include_comments:
263        try:
264            raw_comments = get_ticket_comments(ticket_id)
265            comments = [
266                ZendeskComment(
267                    id=c.get("id"),
268                    author_id=c.get("author_id"),
269                    public=c.get("public"),
270                    body=c.get("plain_body") or c.get("body") or "",
271                    created_at=c.get("created_at"),
272                    attachments=_map_attachments(c),
273                )
274                for c in raw_comments
275                if isinstance(c, dict)
276            ]
277        except ZendeskAPIError as exc:
278            comments_note = f" (comments could not be retrieved: {exc})"
279
280    resolved_id = ticket.get("id", ticket_id)
281    via = _as_dict(ticket.get("via"))
282    via_source = _as_dict(via.get("source"))
283    return ZendeskTicketResponse(
284        success=True,
285        message=f"Retrieved Zendesk ticket {resolved_id}.{comments_note}",
286        ticket_id=resolved_id,
287        subject=ticket.get("subject"),
288        status=ticket.get("status"),
289        description=ticket.get("description"),
290        priority=ticket.get("priority"),
291        tags=ticket.get("tags", []) or [],
292        requester_id=ticket.get("requester_id"),
293        organization_id=ticket.get("organization_id"),
294        created_at=ticket.get("created_at"),
295        updated_at=ticket.get("updated_at"),
296        url=_agent_ticket_url(ticket.get("url"), resolved_id),
297        via_channel=via.get("channel"),
298        via_source_rel=via_source.get("rel"),
299        follow_up_source_ticket_id=_follow_up_source_ticket_id(via_source),
300        custom_fields=_map_custom_fields(ticket),
301        comments=comments,
302    )
303
304
305def _created_comment_id(audit: dict[str, Any]) -> int | None:
306    """Return the created comment's ID from a ticket-update `audit`, if present."""
307    events = audit.get("events")
308    if not isinstance(events, list):
309        return None
310    for event in events:
311        if not isinstance(event, dict):
312            continue
313        if event.get("type") == "Comment" and isinstance(event.get("id"), int):
314            return event["id"]
315    return None
316
317
318@mcp_tool(
319    read_only=False,
320    idempotent=False,
321    open_world=True,
322)
323def post_zendesk_internal_comment(
324    ticket_id: Annotated[
325        int,
326        Field(description="The numeric Zendesk ticket ID to comment on."),
327    ],
328    html_body: Annotated[
329        str,
330        Field(
331            description=(
332                "The internal note as an HTML fragment. Posted as a private "
333                "(non-public) comment visible only to agents \u2014 NOT to the "
334                "ticket requester/end user. Provide HTML: use `<br>`/`<p>` for "
335                "line breaks, `<strong>` for bold, and `<a href>` for links. "
336                "Literal `<`, `>`, and `&` must be HTML-escaped; Zendesk "
337                "sanitizes to a limited HTML subset, so keep markup basic."
338            )
339        ),
340    ],
341) -> ZendeskInternalNoteResponse:
342    """Post an internal (private) HTML note to a Zendesk ticket by its numeric ID.
343
344    The comment is added with `public=False`, so it is an internal agent note
345    and is **not** visible to the ticket requester/end user. Use it to record
346    triage findings, cross-references, or context for other agents. The note
347    is sent as HTML (`html_body`), so bold text and links render as intended.
348
349    This tool only posts the note; it does not modify tags. To route/tag a
350    ticket, call `add_zendesk_ticket_tags` separately.
351
352    The numeric `ticket_id` must already be known; this tool does not search
353    for tickets. Credentials are read from the server environment
354    (`ZENDESK_SUBDOMAIN`, `ZENDESK_EMAIL`, `ZENDESK_API_TOKEN`); this tool
355    never accepts or logs them.
356    """
357    try:
358        result: dict[str, Any] = add_internal_note(ticket_id, html_body)
359    except ZendeskAPIError as exc:
360        return ZendeskInternalNoteResponse(
361            success=False,
362            message=str(exc),
363            ticket_id=ticket_id,
364        )
365
366    updated_ticket = _as_dict(result.get("ticket"))
367    resolved_id = updated_ticket.get("id", ticket_id)
368    comment_id = _created_comment_id(_as_dict(result.get("audit")))
369    return ZendeskInternalNoteResponse(
370        success=True,
371        message=f"Posted internal note to Zendesk ticket {resolved_id}.",
372        ticket_id=resolved_id,
373        comment_id=comment_id,
374    )
375
376
377@mcp_tool(
378    read_only=False,
379    idempotent=True,
380    open_world=True,
381)
382def add_zendesk_ticket_tags(
383    ticket_id: Annotated[
384        int,
385        Field(description="The numeric Zendesk ticket ID to tag."),
386    ],
387    tags: Annotated[
388        list[str],
389        Field(
390            description=(
391                "Tags to add to the ticket. Appended to the ticket's existing "
392                "tags (never clobbered). At least one non-empty tag is required."
393            )
394        ),
395    ],
396) -> ZendeskTagsResponse:
397    """Add tags to a Zendesk ticket by its numeric ID.
398
399    Tags are Zendesk's label mechanism. This reads the ticket's current tags
400    and writes back the union (existing + supplied) via a ticket update, so
401    the ticket's existing tags — and any tagger-backed custom fields — are
402    preserved and never dropped. Adding a tag that is already present is a
403    no-op.
404
405    The numeric `ticket_id` must already be known; this tool does not search
406    for tickets. Credentials are read from the server environment
407    (`ZENDESK_SUBDOMAIN`, `ZENDESK_EMAIL`, `ZENDESK_API_TOKEN`); this tool
408    never accepts or logs them.
409    """
410    try:
411        result_tags = add_ticket_tags(ticket_id, tags)
412    except ZendeskAPIError as exc:
413        return ZendeskTagsResponse(
414            success=False,
415            message=str(exc),
416            ticket_id=ticket_id,
417        )
418
419    return ZendeskTagsResponse(
420        success=True,
421        message=f"Added tags to Zendesk ticket {ticket_id}.",
422        ticket_id=ticket_id,
423        tags=result_tags,
424    )
425
426
427def register_zendesk_ops_tools(app: FastMCP) -> None:
428    """Register zendesk_ops tools with the FastMCP app."""
429    register_mcp_tools(app, mcp_module=__name__)