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
 21import re
 22from typing import Annotated, Any, Literal
 23
 24from fastmcp import FastMCP
 25from fastmcp_extensions import mcp_tool, register_mcp_tools
 26from pydantic import BaseModel, Field
 27
 28from airbyte_ops_mcp.zendesk_api import (
 29    ZendeskAPIError,
 30    add_internal_note,
 31    add_ticket_tags,
 32    create_ticket,
 33    find_organizations_by_airbyte_org_id,
 34    find_ticket_form_id,
 35    get_current_user,
 36    get_ticket,
 37    get_ticket_comments,
 38    list_tickets_by_external_id,
 39    search,
 40)
 41
 42
 43class ZendeskAttachment(BaseModel):
 44    """An attachment on a Zendesk ticket comment."""
 45
 46    id: int | None = Field(default=None, description="Attachment ID.")
 47    file_name: str | None = Field(default=None, description="Original file name.")
 48    content_type: str | None = Field(default=None, description="MIME content type.")
 49    content_url: str | None = Field(
 50        default=None, description="URL to download the attachment content."
 51    )
 52    size: int | None = Field(default=None, description="File size in bytes.")
 53
 54
 55class ZendeskCustomField(BaseModel):
 56    """A single custom field value on a Zendesk ticket."""
 57
 58    id: int | None = Field(default=None, description="Custom field ID.")
 59    value: str | int | float | bool | None = Field(
 60        default=None, description="Custom field value (may be null when unset)."
 61    )
 62
 63
 64class ZendeskComment(BaseModel):
 65    """A single comment on a Zendesk ticket."""
 66
 67    id: int | None = Field(default=None, description="Comment ID.")
 68    author_id: int | None = Field(default=None, description="Author user ID.")
 69    public: bool | None = Field(
 70        default=None,
 71        description="`True` for public replies, `False` for internal notes.",
 72    )
 73    body: str = Field(default="", description="Plain-text comment body.")
 74    created_at: str | None = Field(
 75        default=None,
 76        description="ISO-8601 timestamp when the comment was created.",
 77    )
 78    attachments: list[ZendeskAttachment] = Field(
 79        default_factory=list,
 80        description="Files attached to the comment (metadata only).",
 81    )
 82
 83
 84class ZendeskTicketResponse(BaseModel):
 85    """Response from the `get_zendesk_ticket` tool."""
 86
 87    success: bool = Field(description="Whether the ticket was retrieved.")
 88    message: str = Field(description="Human-readable status message.")
 89    ticket_id: int | None = Field(default=None, description="Zendesk ticket ID.")
 90    subject: str | None = Field(default=None, description="Ticket subject line.")
 91    status: str | None = Field(
 92        default=None,
 93        description="Ticket status (e.g. `open`, `pending`, `solved`, `closed`).",
 94    )
 95    description: str | None = Field(
 96        default=None,
 97        description="The ticket's first comment / description text.",
 98    )
 99    priority: str | None = Field(default=None, description="Ticket priority, if set.")
100    tags: list[str] = Field(default_factory=list, description="Ticket tags.")
101    requester_id: int | None = Field(
102        default=None, description="User ID of the ticket requester."
103    )
104    type: str | None = Field(default=None, description="Zendesk ticket type.")
105    problem_id: int | None = Field(
106        default=None, description="Associated problem ticket ID, if any."
107    )
108    assignee_id: int | None = Field(
109        default=None, description="User ID of the ticket assignee."
110    )
111    submitter_id: int | None = Field(
112        default=None, description="User ID of the ticket submitter."
113    )
114    ticket_form_id: int | None = Field(
115        default=None, description="Ticket form ID used by the ticket."
116    )
117    email_cc_ids: list[int] | None = Field(
118        default=None, description="Zendesk user IDs included as email CCs."
119    )
120    external_id: str | None = Field(
121        default=None, description="External idempotency identifier for the ticket."
122    )
123    organization_id: int | None = Field(
124        default=None, description="Organization ID associated with the ticket."
125    )
126    created_at: str | None = Field(
127        default=None, description="ISO-8601 ticket creation timestamp."
128    )
129    updated_at: str | None = Field(
130        default=None, description="ISO-8601 ticket last-updated timestamp."
131    )
132    url: str | None = Field(
133        default=None,
134        description="Agent-facing Zendesk URL for the ticket.",
135    )
136    via_channel: str | None = Field(
137        default=None,
138        description="Channel the ticket came in through (e.g. `web`, `email`).",
139    )
140    via_source_rel: str | None = Field(
141        default=None,
142        description="Source relationship (e.g. `follow_up` for follow-up tickets).",
143    )
144    follow_up_source_ticket_id: int | None = Field(
145        default=None,
146        description=(
147            "For follow-up tickets, the ID of the original (closed) ticket this "
148            "one follows up on. `None` when the ticket is not a follow-up."
149        ),
150    )
151    custom_fields: list[ZendeskCustomField] = Field(
152        default_factory=list,
153        description="Ticket custom fields (`id`/`value` pairs).",
154    )
155    comments: list[ZendeskComment] = Field(
156        default_factory=list,
157        description="Ticket comments, oldest first. Empty unless requested.",
158    )
159
160
161class ZendeskInternalNoteResponse(BaseModel):
162    """Response from the `post_zendesk_internal_comment` tool."""
163
164    success: bool = Field(description="Whether the internal note was posted.")
165    message: str = Field(description="Human-readable status message.")
166    ticket_id: int | None = Field(default=None, description="Zendesk ticket ID.")
167    comment_id: int | None = Field(
168        default=None,
169        description="ID of the created comment, when Zendesk reports it.",
170    )
171    public: bool = Field(
172        default=False,
173        description=(
174            "Always `False`: this tool can only create a private, agent-only "
175            "note, never a customer-visible reply."
176        ),
177    )
178
179
180class ZendeskTagsResponse(BaseModel):
181    """Response from the `add_zendesk_ticket_tags` tool."""
182
183    success: bool = Field(description="Whether the tags were added.")
184    message: str = Field(description="Human-readable status message.")
185    ticket_id: int | None = Field(default=None, description="Zendesk ticket ID.")
186    tags: list[str] = Field(
187        default_factory=list,
188        description="The ticket's full tag list after the update.",
189    )
190
191
192class ZendeskTicketSummary(BaseModel):
193    """A concise Zendesk ticket search result."""
194
195    id: int = Field(description="Zendesk ticket ID.")
196    subject: str | None = Field(default=None, description="Ticket subject.")
197    status: str | None = Field(default=None, description="Ticket status.")
198    type: str | None = Field(default=None, description="Ticket type.")
199    requester_id: int | None = Field(default=None, description="Requester user ID.")
200    assignee_id: int | None = Field(default=None, description="Assignee user ID.")
201    organization_id: int | None = Field(
202        default=None, description="Associated organization ID."
203    )
204    tags: list[str] = Field(default_factory=list, description="Ticket tags.")
205    external_id: str | None = Field(default=None, description="External ticket ID.")
206    created_at: str | None = Field(default=None, description="Ticket creation time.")
207    updated_at: str | None = Field(default=None, description="Ticket update time.")
208    url: str | None = Field(default=None, description="Agent-facing ticket URL.")
209
210
211class ZendeskTicketSearchResponse(BaseModel):
212    """Response from `search_zendesk_tickets`."""
213
214    success: bool = Field(description="Whether the search succeeded.")
215    message: str = Field(description="Human-readable status message.")
216    count: int = Field(description="Number of tickets returned.")
217    tickets: list[ZendeskTicketSummary] = Field(
218        default_factory=list, description="Matching Zendesk tickets."
219    )
220
221
222class ZendeskOrganizationSummary(BaseModel):
223    """A concise Zendesk organization search result."""
224
225    id: int = Field(description="Zendesk organization ID.")
226    name: str | None = Field(default=None, description="Organization name.")
227    url: str | None = Field(default=None, description="Zendesk API organization URL.")
228
229
230class ZendeskOrganizationSearchResponse(BaseModel):
231    """Response from `find_zendesk_organization_by_airbyte_org_id`."""
232
233    success: bool = Field(description="Whether the search succeeded.")
234    message: str = Field(description="Human-readable status message.")
235    organizations: list[ZendeskOrganizationSummary] = Field(
236        default_factory=list, description="Matching Zendesk organizations."
237    )
238
239
240class OutreachContact(BaseModel):
241    """A Zendesk outreach requester or CC contact."""
242
243    email: str = Field(description="Contact email address.")
244    name: str = Field(description="Contact name.")
245
246
247class ZendeskOutreachTicketResponse(BaseModel):
248    """Response from `create_zendesk_outreach_ticket`."""
249
250    success: bool = Field(description="Whether the ticket was found or created.")
251    message: str = Field(description="Human-readable status message.")
252    created: bool = Field(description="Whether a new ticket was created.")
253    ticket_id: int | None = Field(default=None, description="Zendesk ticket ID.")
254    url: str | None = Field(default=None, description="Agent-facing ticket URL.")
255    status: str | None = Field(default=None, description="Ticket status.")
256    type: str | None = Field(default=None, description="Ticket type.")
257    problem_id: int | None = Field(default=None, description="Problem ticket ID.")
258    requester_id: int | None = Field(default=None, description="Requester user ID.")
259    submitter_id: int | None = Field(default=None, description="Submitter user ID.")
260    assignee_id: int | None = Field(default=None, description="Assignee user ID.")
261    organization_id: int | None = Field(
262        default=None, description="Associated organization ID."
263    )
264    ticket_form_id: int | None = Field(default=None, description="Ticket form ID.")
265    email_cc_ids: list[int] | None = Field(
266        default=None, description="Zendesk user IDs included as email CCs."
267    )
268    tags: list[str] = Field(default_factory=list, description="Ticket tags.")
269    duplicate_ticket_ids: list[int] = Field(
270        default_factory=list,
271        description="Existing ticket IDs when an external ID has duplicates.",
272    )
273
274
275def _agent_ticket_url(raw_url: str | None, ticket_id: int | None) -> str | None:
276    """Derive the agent-facing ticket URL from the API `url` field."""
277    if not raw_url or ticket_id is None:
278        return None
279    # API url looks like https://<subdomain>.zendesk.com/api/v2/tickets/123.json
280    if "/api/v2/" not in raw_url:
281        return None
282    host = raw_url.split("/api/v2/", 1)[0]
283    if not host:
284        return None
285    return f"{host}/agent/tickets/{ticket_id}"
286
287
288def _map_attachments(raw_comment: dict[str, Any]) -> list[ZendeskAttachment]:
289    """Map the `attachments` array of a raw comment into typed models."""
290    raw_attachments = raw_comment.get("attachments")
291    if not isinstance(raw_attachments, list):
292        return []
293    return [
294        ZendeskAttachment(
295            id=a.get("id"),
296            file_name=a.get("file_name"),
297            content_type=a.get("content_type"),
298            content_url=a.get("content_url"),
299            size=a.get("size"),
300        )
301        for a in raw_attachments
302        if isinstance(a, dict)
303    ]
304
305
306def _map_custom_fields(ticket: dict[str, Any]) -> list[ZendeskCustomField]:
307    """Map the ticket's `custom_fields` array into typed models."""
308    raw_fields = ticket.get("custom_fields")
309    if not isinstance(raw_fields, list):
310        return []
311    return [
312        ZendeskCustomField(id=f.get("id"), value=f.get("value"))
313        for f in raw_fields
314        if isinstance(f, dict)
315    ]
316
317
318def _as_dict(value: Any) -> dict[str, Any]:
319    """Return `value` when it is a dict, otherwise an empty dict."""
320    return value if isinstance(value, dict) else {}
321
322
323def _follow_up_source_ticket_id(via_source: dict[str, Any]) -> int | None:
324    """Return the original ticket ID when the ticket is a follow-up."""
325    if via_source.get("rel") != "follow_up":
326        return None
327    ticket_id = _as_dict(via_source.get("from")).get("ticket_id")
328    return ticket_id if isinstance(ticket_id, int) else None
329
330
331@mcp_tool(
332    read_only=True,
333    idempotent=True,
334    open_world=True,
335)
336def get_zendesk_ticket(
337    ticket_id: Annotated[
338        int,
339        Field(description="The numeric Zendesk ticket ID to retrieve."),
340    ],
341    include_comments: Annotated[
342        bool,
343        Field(
344            description=(
345                "When `True`, also fetch the ticket's comments (oldest first), "
346                "including any attachment metadata. Only the first page (up to "
347                "100 comments) is returned. Defaults to `False` to keep "
348                "responses small."
349            )
350        ),
351    ] = False,
352) -> ZendeskTicketResponse:
353    """Retrieve a Zendesk Support ticket by its numeric ID.
354
355    Returns the ticket's subject, status, description, tags, and requester/
356    organization identifiers. Set `include_comments` to `True` to also pull the
357    ticket's comment thread (public replies and internal notes), oldest first;
358    only the first page (up to 100 comments) is returned.
359
360    Credentials are read from the server environment (`ZENDESK_SUBDOMAIN`,
361    `ZENDESK_EMAIL`, `ZENDESK_API_TOKEN`); this tool never accepts or logs them.
362    """
363    try:
364        ticket: dict[str, Any] = get_ticket(ticket_id)
365    except ZendeskAPIError as exc:
366        return ZendeskTicketResponse(
367            success=False,
368            message=str(exc),
369            ticket_id=ticket_id,
370        )
371
372    comments: list[ZendeskComment] = []
373    comments_note = ""
374    if include_comments:
375        try:
376            raw_comments = get_ticket_comments(ticket_id)
377            comments = [
378                ZendeskComment(
379                    id=c.get("id"),
380                    author_id=c.get("author_id"),
381                    public=c.get("public"),
382                    body=c.get("plain_body") or c.get("body") or "",
383                    created_at=c.get("created_at"),
384                    attachments=_map_attachments(c),
385                )
386                for c in raw_comments
387                if isinstance(c, dict)
388            ]
389        except ZendeskAPIError as exc:
390            comments_note = f" (comments could not be retrieved: {exc})"
391
392    resolved_id = ticket.get("id", ticket_id)
393    via = _as_dict(ticket.get("via"))
394    via_source = _as_dict(via.get("source"))
395    return ZendeskTicketResponse(
396        success=True,
397        message=f"Retrieved Zendesk ticket {resolved_id}.{comments_note}",
398        ticket_id=resolved_id,
399        subject=ticket.get("subject"),
400        status=ticket.get("status"),
401        description=ticket.get("description"),
402        priority=ticket.get("priority"),
403        tags=ticket.get("tags", []) or [],
404        requester_id=ticket.get("requester_id"),
405        type=ticket.get("type"),
406        problem_id=ticket.get("problem_id"),
407        assignee_id=ticket.get("assignee_id"),
408        submitter_id=ticket.get("submitter_id"),
409        ticket_form_id=ticket.get("ticket_form_id"),
410        email_cc_ids=ticket.get("email_cc_ids"),
411        external_id=ticket.get("external_id"),
412        organization_id=ticket.get("organization_id"),
413        created_at=ticket.get("created_at"),
414        updated_at=ticket.get("updated_at"),
415        url=_agent_ticket_url(ticket.get("url"), resolved_id),
416        via_channel=via.get("channel"),
417        via_source_rel=via_source.get("rel"),
418        follow_up_source_ticket_id=_follow_up_source_ticket_id(via_source),
419        custom_fields=_map_custom_fields(ticket),
420        comments=comments,
421    )
422
423
424def _ticket_summary(ticket: dict[str, Any]) -> ZendeskTicketSummary:
425    """Map a raw Zendesk ticket into a concise search result."""
426    ticket_id = ticket.get("id")
427    if not isinstance(ticket_id, int):
428        raise ZendeskAPIError("Zendesk search result is missing a valid ticket ID.")
429    return ZendeskTicketSummary(
430        id=ticket_id,
431        subject=ticket.get("subject"),
432        status=ticket.get("status"),
433        type=ticket.get("type"),
434        requester_id=ticket.get("requester_id"),
435        assignee_id=ticket.get("assignee_id"),
436        organization_id=ticket.get("organization_id"),
437        tags=ticket.get("tags", []) or [],
438        external_id=ticket.get("external_id"),
439        created_at=ticket.get("created_at"),
440        updated_at=ticket.get("updated_at"),
441        url=_agent_ticket_url(ticket.get("url"), ticket_id),
442    )
443
444
445def _outreach_ticket_response(
446    ticket: dict[str, Any],
447    *,
448    created: bool,
449    message: str,
450    fallbacks: dict[str, Any] | None = None,
451    success: bool = True,
452    duplicate_ticket_ids: list[int] | None = None,
453) -> ZendeskOutreachTicketResponse:
454    """Map a raw Zendesk ticket into an outreach response."""
455    fallbacks = fallbacks or {}
456    ticket_id = ticket.get("id")
457    return ZendeskOutreachTicketResponse(
458        success=success,
459        message=message,
460        created=created,
461        ticket_id=ticket_id,
462        url=_agent_ticket_url(ticket.get("url"), ticket_id),
463        status=ticket.get("status") or fallbacks.get("status"),
464        type=ticket.get("type") or fallbacks.get("type"),
465        problem_id=ticket.get("problem_id", fallbacks.get("problem_id")),
466        requester_id=ticket.get("requester_id", fallbacks.get("requester_id")),
467        submitter_id=ticket.get("submitter_id", fallbacks.get("submitter_id")),
468        assignee_id=ticket.get("assignee_id"),
469        organization_id=ticket.get("organization_id"),
470        ticket_form_id=ticket.get("ticket_form_id", fallbacks.get("ticket_form_id")),
471        email_cc_ids=ticket.get("email_cc_ids", fallbacks.get("email_cc_ids")),
472        tags=ticket.get("tags", fallbacks.get("tags", [])) or [],
473        duplicate_ticket_ids=duplicate_ticket_ids or [],
474    )
475
476
477@mcp_tool(
478    read_only=True,
479    idempotent=True,
480    open_world=True,
481)
482def search_zendesk_tickets(
483    query: Annotated[
484        str,
485        Field(
486            description="Zendesk search query; ticket searches are scoped to tickets."
487        ),
488    ],
489    sort_by: Annotated[
490        Literal["created_at", "updated_at"] | None,
491        Field(description="Optional ticket field to sort by."),
492    ] = None,
493    sort_order: Annotated[
494        Literal["asc", "desc"],
495        Field(description="Sort direction."),
496    ] = "desc",
497    limit: Annotated[
498        int,
499        Field(description="Maximum number of tickets to return.", ge=1, le=1000),
500    ] = 100,
501) -> ZendeskTicketSearchResponse:
502    """Search Zendesk tickets, with results limited to 1,000.
503
504    If `query` has no positive `type:ticket` filter, `type:ticket` is prepended.
505    Negated type filters and queries that explicitly select another record type
506    are rejected.
507    """
508    if re.search(r"(^|\s)-\s*type\s*:", query, flags=re.IGNORECASE):
509        return ZendeskTicketSearchResponse(
510            success=False,
511            message="Zendesk ticket search does not accept negated type filters.",
512            count=0,
513        )
514    type_filters = re.findall(r"\btype\s*:\s*([^\s]+)", query, flags=re.IGNORECASE)
515    if type_filters and any(value.casefold() != "ticket" for value in type_filters):
516        return ZendeskTicketSearchResponse(
517            success=False,
518            message="Zendesk ticket search only accepts `type:ticket` queries.",
519            count=0,
520        )
521    search_query = query.strip()
522    if not type_filters:
523        search_query = f"type:ticket {search_query}".strip()
524
525    try:
526        results = search(
527            search_query,
528            sort_by=sort_by,
529            sort_order=sort_order,
530            max_results=limit,
531        )
532        tickets = [_ticket_summary(ticket) for ticket in results]
533    except ZendeskAPIError as exc:
534        return ZendeskTicketSearchResponse(
535            success=False,
536            message=str(exc),
537            count=0,
538        )
539
540    return ZendeskTicketSearchResponse(
541        success=True,
542        message=f"Found {len(tickets)} Zendesk ticket(s).",
543        count=len(tickets),
544        tickets=tickets,
545    )
546
547
548@mcp_tool(
549    read_only=True,
550    idempotent=True,
551    open_world=True,
552)
553def find_zendesk_organization_by_airbyte_org_id(
554    airbyte_org_id: Annotated[
555        str,
556        Field(
557            description="Airbyte organization UUID stored on the Zendesk organization."
558        ),
559    ],
560) -> ZendeskOrganizationSearchResponse:
561    """Find Zendesk organizations by their `airbyte_org_id` custom field."""
562    try:
563        organizations = find_organizations_by_airbyte_org_id(airbyte_org_id)
564    except ZendeskAPIError as exc:
565        return ZendeskOrganizationSearchResponse(
566            success=False,
567            message=str(exc),
568        )
569
570    mapped = [
571        ZendeskOrganizationSummary(
572            id=organization["id"],
573            name=organization.get("name"),
574            url=organization.get("url"),
575        )
576        for organization in organizations
577        if isinstance(organization.get("id"), int)
578    ]
579    return ZendeskOrganizationSearchResponse(
580        success=True,
581        message=f"Found {len(mapped)} Zendesk organization(s).",
582        organizations=mapped,
583    )
584
585
586def _outreach_tags(tags: list[str]) -> list[str]:
587    """Trim and deduplicate tags, ensuring the `outreach` tag is present."""
588    cleaned: list[str] = []
589    seen: set[str] = set()
590    has_outreach = False
591    for raw_tag in tags:
592        tag = raw_tag.strip()
593        if not tag:
594            continue
595        normalized = tag.casefold()
596        if normalized in seen:
597            continue
598        seen.add(normalized)
599        if normalized == "outreach":
600            cleaned.append("outreach")
601            has_outreach = True
602        else:
603            cleaned.append(tag)
604    if not has_outreach:
605        cleaned.append("outreach")
606    return cleaned
607
608
609@mcp_tool(
610    read_only=False,
611    idempotent=True,
612    open_world=True,
613)
614def create_zendesk_outreach_ticket(
615    external_id: Annotated[
616        str,
617        Field(description="Idempotency key beginning with `outreach:`."),
618    ],
619    ticket_type: Annotated[
620        Literal["problem", "incident"],
621        Field(description="Zendesk ticket type."),
622    ],
623    subject: Annotated[str, Field(description="Non-empty ticket subject.")],
624    internal_note_html: Annotated[
625        str,
626        Field(description="Non-empty first comment, posted as a private HTML note."),
627    ],
628    problem_id: Annotated[
629        int | None,
630        Field(description="Required parent problem ticket ID for an incident."),
631    ] = None,
632    tags: Annotated[list[str], Field(description="Ticket tags.")] = [],  # noqa: B006
633    ticket_form_name: Annotated[
634        str,
635        Field(description="Active ticket form name."),
636    ] = "Support Outreach",
637    requester: Annotated[
638        OutreachContact | None,
639        Field(
640            description="Requester contact; required for incidents; not allowed for problems."
641        ),
642    ] = None,
643    cc: Annotated[
644        list[OutreachContact],
645        Field(
646            description="Requester contacts to CC; incidents only; Zendesk allows at most 48."
647        ),
648    ] = [],  # noqa: B006
649    assignee_email: Annotated[
650        str | None,
651        Field(description="Optional assignee email address."),
652    ] = None,
653) -> ZendeskOutreachTicketResponse:
654    """Create an idempotent ticket for the approved outreach workflow only.
655
656    This tool cannot send anything public. The first comment is always private
657    and visible only to Zendesk agents.
658    Callers must not run concurrent calls with the same `external_id`; duplicates
659    are detected after creation but not prevented.
660    """
661    try:
662        if (
663            not external_id.startswith("outreach:")
664            or not external_id[len("outreach:") :].strip()
665        ):
666            raise ZendeskAPIError("`external_id` must begin with `outreach:`.")
667
668        existing = list_tickets_by_external_id(external_id)
669        if len(existing) > 1:
670            duplicate_ids = [
671                ticket["id"] for ticket in existing if isinstance(ticket.get("id"), int)
672            ]
673            return ZendeskOutreachTicketResponse(
674                success=False,
675                message=(
676                    "Multiple Zendesk tickets share this external ID: "
677                    f"{', '.join(map(str, duplicate_ids))}."
678                ),
679                created=False,
680                duplicate_ticket_ids=duplicate_ids,
681            )
682        if existing:
683            ticket = existing[0]
684            return _outreach_ticket_response(
685                ticket,
686                created=False,
687                message=f"Found existing Zendesk ticket {ticket.get('id')}; no ticket was created.",
688            )
689
690        if ticket_type == "incident" and problem_id is None:
691            raise ZendeskAPIError("`problem_id` is required for an incident.")
692        if ticket_type == "problem" and problem_id is not None:
693            raise ZendeskAPIError("`problem_id` is not allowed for a problem ticket.")
694        if ticket_type == "incident" and requester is None:
695            raise ZendeskAPIError("`requester` is required for an incident.")
696        if ticket_type == "problem" and requester is not None:
697            raise ZendeskAPIError(
698                "`requester` is not allowed for a problem ticket; MoonBot is the requester."
699            )
700        if ticket_type == "problem" and cc:
701            raise ZendeskAPIError("`cc` is not allowed for a problem ticket.")
702        if not subject.strip():
703            raise ZendeskAPIError("Ticket subject must not be empty.")
704        if not internal_note_html.strip():
705            raise ZendeskAPIError("Internal note body must not be empty.")
706        if requester is not None and not requester.email.strip():
707            raise ZendeskAPIError("Requester email must not be empty.")
708        if assignee_email is not None and not assignee_email.strip():
709            raise ZendeskAPIError("Assignee email must not be empty.")
710
711        current_user = get_current_user()
712        current_user_id = current_user.get("id")
713        if not isinstance(current_user_id, int):
714            raise ZendeskAPIError(
715                "Zendesk current-user response has an invalid user ID."
716            )
717
718        requester_email = (
719            requester.email.strip()
720            if requester is not None
721            else str(current_user.get("email") or "").strip()
722        )
723        cc_contacts: list[dict[str, str]] = []
724        seen_cc_emails: set[str] = set()
725        for contact in cc:
726            email = contact.email.strip()
727            if not email:
728                raise ZendeskAPIError("CC contact email must not be empty.")
729            normalized_email = email.casefold()
730            if normalized_email == requester_email.casefold():
731                continue
732            if normalized_email in seen_cc_emails:
733                continue
734            seen_cc_emails.add(normalized_email)
735            cc_contacts.append(
736                {
737                    "user_email": email,
738                    "user_name": contact.name.strip(),
739                    "action": "put",
740                }
741            )
742        if len(cc_contacts) > 48:
743            raise ZendeskAPIError("At most 48 unique CC contacts are allowed.")
744
745        ticket_form_id = find_ticket_form_id(ticket_form_name)
746        ticket_payload: dict[str, Any] = {
747            "subject": subject.strip(),
748            "comment": {"html_body": internal_note_html.strip(), "public": False},
749            "status": "new",
750            "priority": "normal",
751            "type": ticket_type,
752            "tags": _outreach_tags(tags),
753            "external_id": external_id,
754            "ticket_form_id": ticket_form_id,
755            "submitter_id": current_user_id,
756            "email_ccs": cc_contacts,
757        }
758        if ticket_type == "incident":
759            ticket_payload["problem_id"] = problem_id
760        if requester is None:
761            ticket_payload["requester_id"] = current_user_id
762        else:
763            ticket_payload["requester"] = {
764                "name": requester.name.strip(),
765                "email": requester.email.strip(),
766            }
767        if assignee_email is not None:
768            ticket_payload["assignee_email"] = assignee_email.strip()
769
770        ticket = create_ticket(ticket_payload)
771        try:
772            post_create_tickets = list_tickets_by_external_id(external_id)
773        except ZendeskAPIError as exc:
774            post_create_tickets = None
775            duplicate_check_error = str(exc)
776        else:
777            duplicate_check_error = None
778    except ZendeskAPIError as exc:
779        return ZendeskOutreachTicketResponse(
780            success=False,
781            message=str(exc),
782            created=False,
783        )
784
785    response_fallbacks: dict[str, Any] = {
786        "status": "new",
787        "type": ticket_type,
788        "problem_id": problem_id,
789        "submitter_id": current_user_id,
790        "ticket_form_id": ticket_form_id,
791        "email_cc_ids": [],
792        "tags": ticket_payload["tags"],
793    }
794    if requester is None:
795        response_fallbacks["requester_id"] = current_user_id
796    if duplicate_check_error is not None:
797        return _outreach_ticket_response(
798            ticket,
799            created=True,
800            message=(
801                f"Created ticket {ticket.get('id')} but could not check for "
802                "duplicate tickets for this external ID: "
803                f"{duplicate_check_error}."
804            ),
805            fallbacks=response_fallbacks,
806            success=False,
807        )
808    if post_create_tickets is not None and len(post_create_tickets) > 1:
809        duplicate_ids = [
810            existing_ticket["id"]
811            for existing_ticket in post_create_tickets
812            if isinstance(existing_ticket.get("id"), int)
813        ]
814        return _outreach_ticket_response(
815            ticket,
816            created=True,
817            message=(
818                f"Created ticket {ticket.get('id')} but found duplicate tickets "
819                f"for this external ID: {', '.join(map(str, duplicate_ids))}; "
820                "resolve manually."
821            ),
822            fallbacks=response_fallbacks,
823            success=False,
824            duplicate_ticket_ids=duplicate_ids,
825        )
826    return _outreach_ticket_response(
827        ticket,
828        created=True,
829        message=f"Created private outreach ticket {ticket.get('id')}.",
830        fallbacks=response_fallbacks,
831    )
832
833
834def _created_comment_id(audit: dict[str, Any]) -> int | None:
835    """Return the created comment's ID from a ticket-update `audit`, if present."""
836    events = audit.get("events")
837    if not isinstance(events, list):
838        return None
839    for event in events:
840        if not isinstance(event, dict):
841            continue
842        if event.get("type") == "Comment" and isinstance(event.get("id"), int):
843            return event["id"]
844    return None
845
846
847@mcp_tool(
848    read_only=False,
849    idempotent=False,
850    open_world=True,
851)
852def post_zendesk_internal_comment(
853    ticket_id: Annotated[
854        int,
855        Field(description="The numeric Zendesk ticket ID to comment on."),
856    ],
857    html_body: Annotated[
858        str,
859        Field(
860            description=(
861                "The internal note as an HTML fragment. Posted as a private "
862                "(non-public) comment visible only to agents \u2014 NOT to the "
863                "ticket requester/end user. This server cannot post a public "
864                "customer reply; if asked to reply, write a clearly labelled "
865                "draft for a human agent to send. Provide HTML: use `<br>`/`<p>` for "
866                "line breaks, `<strong>` for bold, and `<a href>` for links. "
867                "Literal `<`, `>`, and `&` must be HTML-escaped; Zendesk "
868                "sanitizes to a limited HTML subset, so keep markup basic."
869            )
870        ),
871    ],
872) -> ZendeskInternalNoteResponse:
873    """Post an internal (private) HTML note to a Zendesk ticket by its numeric ID.
874
875    The comment is added with `public=False`, so it is an internal agent note
876    and is **not** visible to the ticket requester/end user. Use it to record
877    triage findings, cross-references, or context for other agents. This server
878    has no tool that can post a customer-visible/public reply, so it is
879    impossible to answer the customer through the MCP server. When asked to
880    "reply to the customer", write the proposed reply as an internal note
881    clearly labelled as a draft for a human support agent to send, and tell the
882    requesting human that a support agent still has to post it publicly. The
883    note is sent as HTML (`html_body`), so bold text and links render as intended.
884
885    This tool only posts the note; it does not modify tags. To route/tag a
886    ticket, call `add_zendesk_ticket_tags` separately.
887
888    The numeric `ticket_id` must already be known; this tool does not search
889    for tickets. Credentials are read from the server environment
890    (`ZENDESK_SUBDOMAIN`, `ZENDESK_EMAIL`, `ZENDESK_API_TOKEN`); this tool
891    never accepts or logs them.
892    """
893    try:
894        result: dict[str, Any] = add_internal_note(ticket_id, html_body)
895    except ZendeskAPIError as exc:
896        return ZendeskInternalNoteResponse(
897            success=False,
898            message=str(exc),
899            ticket_id=ticket_id,
900        )
901
902    updated_ticket = _as_dict(result.get("ticket"))
903    resolved_id = updated_ticket.get("id", ticket_id)
904    comment_id = _created_comment_id(_as_dict(result.get("audit")))
905    return ZendeskInternalNoteResponse(
906        success=True,
907        message=(
908            f"Posted an internal-only note to Zendesk ticket {resolved_id}; "
909            "it is not visible to the customer."
910        ),
911        ticket_id=resolved_id,
912        comment_id=comment_id,
913    )
914
915
916@mcp_tool(
917    read_only=False,
918    idempotent=True,
919    open_world=True,
920)
921def add_zendesk_ticket_tags(
922    ticket_id: Annotated[
923        int,
924        Field(description="The numeric Zendesk ticket ID to tag."),
925    ],
926    tags: Annotated[
927        list[str],
928        Field(
929            description=(
930                "Tags to add to the ticket. Appended to the ticket's existing "
931                "tags (never clobbered). At least one non-empty tag is required."
932            )
933        ),
934    ],
935) -> ZendeskTagsResponse:
936    """Add tags to a Zendesk ticket by its numeric ID.
937
938    Tags are Zendesk's label mechanism. This reads the ticket's current tags
939    and writes back the union (existing + supplied) via a ticket update, so
940    the ticket's existing tags — and any tagger-backed custom fields — are
941    preserved and never dropped. Adding a tag that is already present is a
942    no-op.
943
944    The numeric `ticket_id` must already be known; this tool does not search
945    for tickets. Credentials are read from the server environment
946    (`ZENDESK_SUBDOMAIN`, `ZENDESK_EMAIL`, `ZENDESK_API_TOKEN`); this tool
947    never accepts or logs them.
948    """
949    try:
950        result_tags = add_ticket_tags(ticket_id, tags)
951    except ZendeskAPIError as exc:
952        return ZendeskTagsResponse(
953            success=False,
954            message=str(exc),
955            ticket_id=ticket_id,
956        )
957
958    return ZendeskTagsResponse(
959        success=True,
960        message=f"Added tags to Zendesk ticket {ticket_id}.",
961        ticket_id=ticket_id,
962        tags=result_tags,
963    )
964
965
966def register_zendesk_ops_tools(app: FastMCP) -> None:
967    """Register zendesk_ops tools with the FastMCP app."""
968    register_mcp_tools(app, mcp_module=__name__)