#!/usr/bin/env python3
import asyncio
import json
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from dotenv import load_dotenv
from .zendesk_client import ZendeskClient
load_dotenv()
server = Server("zendesk-mcp-server")
TOOLS = [
Tool(
name="get_tickets",
description="Get a list of Zendesk tickets with optional filtering",
inputSchema={
"type": "object",
"properties": {
"status": {"type": "string", "description": "Filter by status: new, open, pending, hold, solved, closed"},
"priority": {"type": "string", "description": "Filter by priority: low, normal, high, urgent"},
"sort_by": {"type": "string", "description": "Sort field: created_at, updated_at, priority, status"},
"sort_order": {"type": "string", "description": "Sort order: asc or desc"},
"per_page": {"type": "integer", "description": "Results per page (max 100)"},
"page": {"type": "integer", "description": "Page number"}
}
}
),
Tool(
name="get_ticket",
description="Get detailed information about a specific ticket",
inputSchema={
"type": "object",
"properties": {
"ticket_id": {"type": "integer", "description": "The ticket ID"}
},
"required": ["ticket_id"]
}
),
Tool(
name="get_ticket_comments",
description="Get all comments on a ticket",
inputSchema={
"type": "object",
"properties": {
"ticket_id": {"type": "integer", "description": "The ticket ID"}
},
"required": ["ticket_id"]
}
),
Tool(
name="create_ticket",
description="Create a new support ticket",
inputSchema={
"type": "object",
"properties": {
"subject": {"type": "string", "description": "Ticket subject"},
"description": {"type": "string", "description": "Ticket description/initial comment"},
"priority": {"type": "string", "description": "Priority: low, normal, high, urgent"},
"ticket_type": {"type": "string", "description": "Type: question, incident, problem, task"},
"tags": {"type": "array", "items": {"type": "string"}, "description": "Tags to apply"},
"requester_email": {"type": "string", "description": "Email of the requester"}
},
"required": ["subject", "description"]
}
),
Tool(
name="create_ticket_comment",
description="Add a comment to an existing ticket",
inputSchema={
"type": "object",
"properties": {
"ticket_id": {"type": "integer", "description": "The ticket ID"},
"body": {"type": "string", "description": "Comment text"},
"public": {"type": "boolean", "description": "True for public reply, false for internal note"}
},
"required": ["ticket_id", "body"]
}
),
Tool(
name="update_ticket",
description="Update ticket properties",
inputSchema={
"type": "object",
"properties": {
"ticket_id": {"type": "integer", "description": "The ticket ID"},
"status": {"type": "string", "description": "New status"},
"priority": {"type": "string", "description": "New priority"},
"assignee_id": {"type": "integer", "description": "User ID to assign to"},
"tags": {"type": "array", "items": {"type": "string"}, "description": "Replace tags"},
"add_tags": {"type": "array", "items": {"type": "string"}, "description": "Add tags"}
},
"required": ["ticket_id"]
}
)
]
@server.list_tools()
async def list_tools():
return TOOLS
@server.call_tool()
async def call_tool(name: str, arguments: dict):
try:
client = ZendeskClient()
if name == "get_tickets":
result = await client.get_tickets(
status=arguments.get("status"),
priority=arguments.get("priority"),
sort_by=arguments.get("sort_by", "created_at"),
sort_order=arguments.get("sort_order", "desc"),
per_page=arguments.get("per_page", 25),
page=arguments.get("page", 1)
)
elif name == "get_ticket":
result = await client.get_ticket(arguments["ticket_id"])
elif name == "get_ticket_comments":
result = await client.get_ticket_comments(arguments["ticket_id"])
elif name == "create_ticket":
result = await client.create_ticket(
subject=arguments["subject"],
description=arguments["description"],
priority=arguments.get("priority"),
ticket_type=arguments.get("ticket_type"),
tags=arguments.get("tags"),
requester_email=arguments.get("requester_email")
)
elif name == "create_ticket_comment":
result = await client.create_ticket_comment(
ticket_id=arguments["ticket_id"],
body=arguments["body"],
public=arguments.get("public", True)
)
elif name == "update_ticket":
result = await client.update_ticket(
ticket_id=arguments["ticket_id"],
status=arguments.get("status"),
priority=arguments.get("priority"),
assignee_id=arguments.get("assignee_id"),
tags=arguments.get("tags"),
add_tags=arguments.get("add_tags")
)
else:
return [TextContent(type="text", text=f"Unknown tool: {name}")]
return [TextContent(type="text", text=json.dumps(result, indent=2))]
except Exception as e:
return [TextContent(type="text", text=f"Error: {str(e)}")]
async def main():
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream, server.create_initialization_options())
if __name__ == "__main__":
asyncio.run(main())
