zendesk-mcp-server

import asyncio import json import logging import os from typing import Any, Dict from cachetools.func import ttl_cache from dotenv import load_dotenv from mcp.server import InitializationOptions, NotificationOptions from mcp.server import Server, types from mcp.server.stdio import stdio_server from pydantic import AnyUrl from zendesk_mcp_server.zendesk_client import ZendeskClient logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) logger = logging.getLogger("zendesk-mcp-server") logger.info("zendesk mcp server started") load_dotenv() zendesk_client = ZendeskClient( subdomain=os.getenv("ZENDESK_SUBDOMAIN"), email=os.getenv("ZENDESK_EMAIL"), token=os.getenv("ZENDESK_API_KEY") ) server = Server("Zendesk Server") TICKET_ANALYSIS_TEMPLATE = """ You are a helpful Zendesk support analyst. You've been asked to analyze ticket #{ticket_id}. Please fetch the ticket info and comments to analyze it and provide: 1. A summary of the issue 2. The current status and timeline 3. Key points of interaction Remember to be professional and focus on actionable insights. """ COMMENT_DRAFT_TEMPLATE = """ You are a helpful Zendesk support agent. You need to draft a response to ticket #{ticket_id}. Please fetch the ticket info, comments and knowledge base to draft a professional and helpful response that: 1. Acknowledges the customer's concern 2. Addresses the specific issues raised 3. Provides clear next steps or ask for specific details need to proceed 4. Maintains a friendly and professional tone 5. Ask for confirmation before commenting on the ticket The response should be formatted well and ready to be posted as a comment. """ @server.list_prompts() async def handle_list_prompts() -> list[types.Prompt]: """List available prompts""" return [ types.Prompt( name="analyze-ticket", description="Analyze a Zendesk ticket and provide insights", arguments=[ types.PromptArgument( name="ticket_id", description="The ID of the ticket to analyze", required=True, ) ], ), types.Prompt( name="draft-ticket-response", description="Draft a professional response to a Zendesk ticket", arguments=[ types.PromptArgument( name="ticket_id", description="The ID of the ticket to respond to", required=True, ) ], ) ] @server.get_prompt() async def handle_get_prompt(name: str, arguments: Dict[str, str] | None) -> types.GetPromptResult: """Handle prompt requests""" if not arguments or "ticket_id" not in arguments: raise ValueError("Missing required argument: ticket_id") ticket_id = int(arguments["ticket_id"]) try: if name == "analyze-ticket": prompt = TICKET_ANALYSIS_TEMPLATE.format( ticket_id=ticket_id ) description = f"Analysis prompt for ticket #{ticket_id}" elif name == "draft-ticket-response": prompt = COMMENT_DRAFT_TEMPLATE.format( ticket_id=ticket_id ) description = f"Response draft prompt for ticket #{ticket_id}" else: raise ValueError(f"Unknown prompt: {name}") return types.GetPromptResult( description=description, messages=[ types.PromptMessage( role="user", content=types.TextContent(type="text", text=prompt.strip()), ) ], ) except Exception as e: logger.error(f"Error generating prompt: {e}") raise @server.list_tools() async def handle_list_tools() -> list[types.Tool]: """List available Zendesk tools""" return [ types.Tool( name="get_ticket", description="Retrieve a Zendesk ticket by its ID", inputSchema={ "type": "object", "properties": { "ticket_id": { "type": "integer", "description": "The ID of the ticket to retrieve" } }, "required": ["ticket_id"] } ), types.Tool( name="get_tickets", description="Fetch the latest tickets with pagination support", inputSchema={ "type": "object", "properties": { "page": { "type": "integer", "description": "Page number", "default": 1 }, "per_page": { "type": "integer", "description": "Number of tickets per page (max 100)", "default": 25 }, "sort_by": { "type": "string", "description": "Field to sort by (created_at, updated_at, priority, status)", "default": "created_at" }, "sort_order": { "type": "string", "description": "Sort order (asc or desc)", "default": "desc" } }, "required": [] } ), types.Tool( name="get_ticket_comments", description="Retrieve all comments for a Zendesk ticket by its ID", inputSchema={ "type": "object", "properties": { "ticket_id": { "type": "integer", "description": "The ID of the ticket to get comments for" } }, "required": ["ticket_id"] } ), types.Tool( name="create_ticket_comment", description="Create a new comment on an existing Zendesk ticket", inputSchema={ "type": "object", "properties": { "ticket_id": { "type": "integer", "description": "The ID of the ticket to comment on" }, "comment": { "type": "string", "description": "The comment text/content to add" }, "public": { "type": "boolean", "description": "Whether the comment should be public", "default": True } }, "required": ["ticket_id", "comment"] } ) ] @server.call_tool() async def handle_call_tool( name: str, arguments: dict[str, Any] | None ) -> list[types.TextContent]: """Handle Zendesk tool execution requests""" try: if name == "get_ticket": if not arguments: raise ValueError("Missing arguments") ticket = zendesk_client.get_ticket(arguments["ticket_id"]) return [types.TextContent( type="text", text=json.dumps(ticket) )] elif name == "get_tickets": page = arguments.get("page", 1) if arguments else 1 per_page = arguments.get("per_page", 25) if arguments else 25 sort_by = arguments.get("sort_by", "created_at") if arguments else "created_at" sort_order = arguments.get("sort_order", "desc") if arguments else "desc" tickets = zendesk_client.get_tickets( page=page, per_page=per_page, sort_by=sort_by, sort_order=sort_order ) return [types.TextContent( type="text", text=json.dumps(tickets, indent=2) )] elif name == "get_ticket_comments": if not arguments: raise ValueError("Missing arguments") comments = zendesk_client.get_ticket_comments( arguments["ticket_id"]) return [types.TextContent( type="text", text=json.dumps(comments) )] elif name == "create_ticket_comment": if not arguments: raise ValueError("Missing arguments") public = arguments.get("public", True) result = zendesk_client.post_comment( ticket_id=arguments["ticket_id"], comment=arguments["comment"], public=public ) return [types.TextContent( type="text", text=f"Comment created successfully: {result}" )] else: raise ValueError(f"Unknown tool: {name}") except Exception as e: return [types.TextContent( type="text", text=f"Error: {str(e)}" )] @server.list_resources() async def handle_list_resources() -> list[types.Resource]: logger.debug("Handling list_resources request") return [ types.Resource( uri=AnyUrl("zendesk://knowledge-base"), name="Zendesk Knowledge Base", description="Access to Zendesk Help Center articles and sections", mimeType="application/json", ) ] @ttl_cache(ttl=3600) def get_cached_kb(): return zendesk_client.get_all_articles() @server.read_resource() async def handle_read_resource(uri: AnyUrl) -> str: logger.debug(f"Handling read_resource request for URI: {uri}") if uri.scheme != "zendesk": logger.error(f"Unsupported URI scheme: {uri.scheme}") raise ValueError(f"Unsupported URI scheme: {uri.scheme}") path = str(uri).replace("zendesk://", "") if path != "knowledge-base": logger.error(f"Unknown resource path: {path}") raise ValueError(f"Unknown resource path: {path}") try: kb_data = get_cached_kb() return json.dumps({ "knowledge_base": kb_data, "metadata": { "sections": len(kb_data), "total_articles": sum(len(section['articles']) for section in kb_data.values()), } }, indent=2) except Exception as e: logger.error(f"Error fetching knowledge base: {e}") raise async def main(): # Run the server using stdin/stdout streams async with stdio_server() as (read_stream, write_stream): await server.run( read_stream=read_stream, write_stream=write_stream, initialization_options=InitializationOptions( server_name="Zendesk", server_version="0.1.0", capabilities=server.get_capabilities( notification_options=NotificationOptions(), experimental_capabilities={}, ), ), ) if __name__ == "__main__": asyncio.run(main())