zendesk-mcp-server

from typing import Dict, Any, List import json import urllib.request import urllib.parse import base64 from zenpy import Zenpy from zenpy.lib.api_objects import Comment class ZendeskClient: def __init__(self, subdomain: str, email: str, token: str): """ Initialize the Zendesk client using zenpy lib and direct API. """ self.client = Zenpy( subdomain=subdomain, email=email, token=token ) # For direct API calls self.subdomain = subdomain self.email = email self.token = token self.base_url = f"https://{subdomain}.zendesk.com/api/v2" # Create basic auth header credentials = f"{email}/token:{token}" encoded_credentials = base64.b64encode(credentials.encode()).decode('ascii') self.auth_header = f"Basic {encoded_credentials}" def get_ticket(self, ticket_id: int) -> Dict[str, Any]: """ Query a ticket by its ID """ try: ticket = self.client.tickets(id=ticket_id) return { 'id': ticket.id, 'subject': ticket.subject, 'description': ticket.description, 'status': ticket.status, 'priority': ticket.priority, 'created_at': str(ticket.created_at), 'updated_at': str(ticket.updated_at), 'requester_id': ticket.requester_id, 'assignee_id': ticket.assignee_id, 'organization_id': ticket.organization_id } except Exception as e: raise Exception(f"Failed to get ticket {ticket_id}: {str(e)}") def get_ticket_comments(self, ticket_id: int) -> List[Dict[str, Any]]: """ Get all comments for a specific ticket. """ try: comments = self.client.tickets.comments(ticket=ticket_id) return [{ 'id': comment.id, 'author_id': comment.author_id, 'body': comment.body, 'html_body': comment.html_body, 'public': comment.public, 'created_at': str(comment.created_at) } for comment in comments] except Exception as e: raise Exception(f"Failed to get comments for ticket {ticket_id}: {str(e)}") def post_comment(self, ticket_id: int, comment: str, public: bool = True) -> str: """ Post a comment to an existing ticket. """ try: ticket = self.client.tickets(id=ticket_id) ticket.comment = Comment( html_body=comment, public=public ) self.client.tickets.update(ticket) return comment except Exception as e: raise Exception(f"Failed to post comment on ticket {ticket_id}: {str(e)}") def get_tickets(self, page: int = 1, per_page: int = 25, sort_by: str = 'created_at', sort_order: str = 'desc') -> Dict[str, Any]: """ Get the latest tickets with proper pagination support using direct API calls. Args: page: Page number (1-based) per_page: Number of tickets per page (max 100) sort_by: Field to sort by (created_at, updated_at, priority, status) sort_order: Sort order (asc or desc) Returns: Dict containing tickets and pagination info """ try: # Cap at reasonable limit per_page = min(per_page, 100) # Build URL with parameters for offset pagination params = { 'page': str(page), 'per_page': str(per_page), 'sort_by': sort_by, 'sort_order': sort_order } query_string = urllib.parse.urlencode(params) url = f"{self.base_url}/tickets.json?{query_string}" # Create request with auth header req = urllib.request.Request(url) req.add_header('Authorization', self.auth_header) req.add_header('Content-Type', 'application/json') # Make the API request with urllib.request.urlopen(req) as response: data = json.loads(response.read().decode()) tickets_data = data.get('tickets', []) # Process tickets to return only essential fields ticket_list = [] for ticket in tickets_data: ticket_list.append({ 'id': ticket.get('id'), 'subject': ticket.get('subject'), 'status': ticket.get('status'), 'priority': ticket.get('priority'), 'description': ticket.get('description'), 'created_at': ticket.get('created_at'), 'updated_at': ticket.get('updated_at'), 'requester_id': ticket.get('requester_id'), 'assignee_id': ticket.get('assignee_id') }) return { 'tickets': ticket_list, 'page': page, 'per_page': per_page, 'count': len(ticket_list), 'sort_by': sort_by, 'sort_order': sort_order, 'has_more': data.get('next_page') is not None, 'next_page': page + 1 if data.get('next_page') else None, 'previous_page': page - 1 if data.get('previous_page') and page > 1 else None } except urllib.error.HTTPError as e: error_body = e.read().decode() if e.fp else "No response body" raise Exception(f"Failed to get latest tickets: HTTP {e.code} - {e.reason}. {error_body}") except Exception as e: raise Exception(f"Failed to get latest tickets: {str(e)}") def get_all_articles(self) -> Dict[str, Any]: """ Fetch help center articles as knowledge base. Returns a Dict of section -> [article]. """ try: # Get all sections sections = self.client.help_center.sections() # Get articles for each section kb = {} for section in sections: articles = self.client.help_center.sections.articles(section.id) kb[section.name] = { 'section_id': section.id, 'description': section.description, 'articles': [{ 'id': article.id, 'title': article.title, 'body': article.body, 'updated_at': str(article.updated_at), 'url': article.html_url } for article in articles] } return kb except Exception as e: raise Exception(f"Failed to fetch knowledge base: {str(e)}")