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=(
148            "Always `False`: this tool can only create a private, agent-only "
149            "note, never a customer-visible reply."
150        ),
151    )
152
153
154class ZendeskTagsResponse(BaseModel):
155    """Response from the `add_zendesk_ticket_tags` tool."""
156
157    success: bool = Field(description="Whether the tags were added.")
158    message: str = Field(description="Human-readable status message.")
159    ticket_id: int | None = Field(default=None, description="Zendesk ticket ID.")
160    tags: list[str] = Field(
161        default_factory=list,
162        description="The ticket's full tag list after the update.",
163    )
164
165
166def _agent_ticket_url(raw_url: str | None, ticket_id: int | None) -> str | None:
167    """Derive the agent-facing ticket URL from the API `url` field."""
168    if not raw_url or ticket_id is None:
169        return None
170    # API url looks like https://<subdomain>.zendesk.com/api/v2/tickets/123.json
171    if "/api/v2/" not in raw_url:
172        return None
173    host = raw_url.split("/api/v2/", 1)[0]
174    if not host:
175        return None
176    return f"{host}/agent/tickets/{ticket_id}"
177
178
179def _map_attachments(raw_comment: dict[str, Any]) -> list[ZendeskAttachment]:
180    """Map the `attachments` array of a raw comment into typed models."""
181    raw_attachments = raw_comment.get("attachments")
182    if not isinstance(raw_attachments, list):
183        return []
184    return [
185        ZendeskAttachment(
186            id=a.get("id"),
187            file_name=a.get("file_name"),
188            content_type=a.get("content_type"),
189            content_url=a.get("content_url"),
190            size=a.get("size"),
191        )
192        for a in raw_attachments
193        if isinstance(a, dict)
194    ]
195
196
197def _map_custom_fields(ticket: dict[str, Any]) -> list[ZendeskCustomField]:
198    """Map the ticket's `custom_fields` array into typed models."""
199    raw_fields = ticket.get("custom_fields")
200    if not isinstance(raw_fields, list):
201        return []
202    return [
203        ZendeskCustomField(id=f.get("id"), value=f.get("value"))
204        for f in raw_fields
205        if isinstance(f, dict)
206    ]
207
208
209def _as_dict(value: Any) -> dict[str, Any]:
210    """Return `value` when it is a dict, otherwise an empty dict."""
211    return value if isinstance(value, dict) else {}
212
213
214def _follow_up_source_ticket_id(via_source: dict[str, Any]) -> int | None:
215    """Return the original ticket ID when the ticket is a follow-up."""
216    if via_source.get("rel") != "follow_up":
217        return None
218    ticket_id = _as_dict(via_source.get("from")).get("ticket_id")
219    return ticket_id if isinstance(ticket_id, int) else None
220
221
222@mcp_tool(
223    read_only=True,
224    idempotent=True,
225    open_world=True,
226)
227def get_zendesk_ticket(
228    ticket_id: Annotated[
229        int,
230        Field(description="The numeric Zendesk ticket ID to retrieve."),
231    ],
232    include_comments: Annotated[
233        bool,
234        Field(
235            description=(
236                "When `True`, also fetch the ticket's comments (oldest first), "
237                "including any attachment metadata. Only the first page (up to "
238                "100 comments) is returned. Defaults to `False` to keep "
239                "responses small."
240            )
241        ),
242    ] = False,
243) -> ZendeskTicketResponse:
244    """Retrieve a Zendesk Support ticket by its numeric ID.
245
246    Returns the ticket's subject, status, description, tags, and requester/
247    organization identifiers. Set `include_comments` to `True` to also pull the
248    ticket's comment thread (public replies and internal notes), oldest first;
249    only the first page (up to 100 comments) is returned.
250
251    Credentials are read from the server environment (`ZENDESK_SUBDOMAIN`,
252    `ZENDESK_EMAIL`, `ZENDESK_API_TOKEN`); this tool never accepts or logs them.
253    """
254    try:
255        ticket: dict[str, Any] = get_ticket(ticket_id)
256    except ZendeskAPIError as exc:
257        return ZendeskTicketResponse(
258            success=False,
259            message=str(exc),
260            ticket_id=ticket_id,
261        )
262
263    comments: list[ZendeskComment] = []
264    comments_note = ""
265    if include_comments:
266        try:
267            raw_comments = get_ticket_comments(ticket_id)
268            comments = [
269                ZendeskComment(
270                    id=c.get("id"),
271                    author_id=c.get("author_id"),
272                    public=c.get("public"),
273                    body=c.get("plain_body") or c.get("body") or "",
274                    created_at=c.get("created_at"),
275                    attachments=_map_attachments(c),
276                )
277                for c in raw_comments
278                if isinstance(c, dict)
279            ]
280        except ZendeskAPIError as exc:
281            comments_note = f" (comments could not be retrieved: {exc})"
282
283    resolved_id = ticket.get("id", ticket_id)
284    via = _as_dict(ticket.get("via"))
285    via_source = _as_dict(via.get("source"))
286    return ZendeskTicketResponse(
287        success=True,
288        message=f"Retrieved Zendesk ticket {resolved_id}.{comments_note}",
289        ticket_id=resolved_id,
290        subject=ticket.get("subject"),
291        status=ticket.get("status"),
292        description=ticket.get("description"),
293        priority=ticket.get("priority"),
294        tags=ticket.get("tags", []) or [],
295        requester_id=ticket.get("requester_id"),
296        organization_id=ticket.get("organization_id"),
297        created_at=ticket.get("created_at"),
298        updated_at=ticket.get("updated_at"),
299        url=_agent_ticket_url(ticket.get("url"), resolved_id),
300        via_channel=via.get("channel"),
301        via_source_rel=via_source.get("rel"),
302        follow_up_source_ticket_id=_follow_up_source_ticket_id(via_source),
303        custom_fields=_map_custom_fields(ticket),
304        comments=comments,
305    )
306
307
308def _created_comment_id(audit: dict[str, Any]) -> int | None:
309    """Return the created comment's ID from a ticket-update `audit`, if present."""
310    events = audit.get("events")
311    if not isinstance(events, list):
312        return None
313    for event in events:
314        if not isinstance(event, dict):
315            continue
316        if event.get("type") == "Comment" and isinstance(event.get("id"), int):
317            return event["id"]
318    return None
319
320
321@mcp_tool(
322    read_only=False,
323    idempotent=False,
324    open_world=True,
325)
326def post_zendesk_internal_comment(
327    ticket_id: Annotated[
328        int,
329        Field(description="The numeric Zendesk ticket ID to comment on."),
330    ],
331    html_body: Annotated[
332        str,
333        Field(
334            description=(
335                "The internal note as an HTML fragment. Posted as a private "
336                "(non-public) comment visible only to agents \u2014 NOT to the "
337                "ticket requester/end user. This server cannot post a public "
338                "customer reply; if asked to reply, write a clearly labelled "
339                "draft for a human agent to send. Provide HTML: use `<br>`/`<p>` for "
340                "line breaks, `<strong>` for bold, and `<a href>` for links. "
341                "Literal `<`, `>`, and `&` must be HTML-escaped; Zendesk "
342                "sanitizes to a limited HTML subset, so keep markup basic."
343            )
344        ),
345    ],
346) -> ZendeskInternalNoteResponse:
347    """Post an internal (private) HTML note to a Zendesk ticket by its numeric ID.
348
349    The comment is added with `public=False`, so it is an internal agent note
350    and is **not** visible to the ticket requester/end user. Use it to record
351    triage findings, cross-references, or context for other agents. This server
352    has no tool that can post a customer-visible/public reply, so it is
353    impossible to answer the customer through the MCP server. When asked to
354    "reply to the customer", write the proposed reply as an internal note
355    clearly labelled as a draft for a human support agent to send, and tell the
356    requesting human that a support agent still has to post it publicly. The
357    note is sent as HTML (`html_body`), so bold text and links render as intended.
358
359    This tool only posts the note; it does not modify tags. To route/tag a
360    ticket, call `add_zendesk_ticket_tags` separately.
361
362    The numeric `ticket_id` must already be known; this tool does not search
363    for tickets. Credentials are read from the server environment
364    (`ZENDESK_SUBDOMAIN`, `ZENDESK_EMAIL`, `ZENDESK_API_TOKEN`); this tool
365    never accepts or logs them.
366    """
367    try:
368        result: dict[str, Any] = add_internal_note(ticket_id, html_body)
369    except ZendeskAPIError as exc:
370        return ZendeskInternalNoteResponse(
371            success=False,
372            message=str(exc),
373            ticket_id=ticket_id,
374        )
375
376    updated_ticket = _as_dict(result.get("ticket"))
377    resolved_id = updated_ticket.get("id", ticket_id)
378    comment_id = _created_comment_id(_as_dict(result.get("audit")))
379    return ZendeskInternalNoteResponse(
380        success=True,
381        message=(
382            f"Posted an internal-only note to Zendesk ticket {resolved_id}; "
383            "it is not visible to the customer."
384        ),
385        ticket_id=resolved_id,
386        comment_id=comment_id,
387    )
388
389
390@mcp_tool(
391    read_only=False,
392    idempotent=True,
393    open_world=True,
394)
395def add_zendesk_ticket_tags(
396    ticket_id: Annotated[
397        int,
398        Field(description="The numeric Zendesk ticket ID to tag."),
399    ],
400    tags: Annotated[
401        list[str],
402        Field(
403            description=(
404                "Tags to add to the ticket. Appended to the ticket's existing "
405                "tags (never clobbered). At least one non-empty tag is required."
406            )
407        ),
408    ],
409) -> ZendeskTagsResponse:
410    """Add tags to a Zendesk ticket by its numeric ID.
411
412    Tags are Zendesk's label mechanism. This reads the ticket's current tags
413    and writes back the union (existing + supplied) via a ticket update, so
414    the ticket's existing tags — and any tagger-backed custom fields — are
415    preserved and never dropped. Adding a tag that is already present is a
416    no-op.
417
418    The numeric `ticket_id` must already be known; this tool does not search
419    for tickets. Credentials are read from the server environment
420    (`ZENDESK_SUBDOMAIN`, `ZENDESK_EMAIL`, `ZENDESK_API_TOKEN`); this tool
421    never accepts or logs them.
422    """
423    try:
424        result_tags = add_ticket_tags(ticket_id, tags)
425    except ZendeskAPIError as exc:
426        return ZendeskTagsResponse(
427            success=False,
428            message=str(exc),
429            ticket_id=ticket_id,
430        )
431
432    return ZendeskTagsResponse(
433        success=True,
434        message=f"Added tags to Zendesk ticket {ticket_id}.",
435        ticket_id=ticket_id,
436        tags=result_tags,
437    )
438
439
440def register_zendesk_ops_tools(app: FastMCP) -> None:
441    """Register zendesk_ops tools with the FastMCP app."""
442    register_mcp_tools(app, mcp_module=__name__)