# Directory Structure
```
├── .env.example
├── .github
│ └── workflows
│ └── ci.yml
├── .gitignore
├── .python-version
├── Dockerfile
├── LICENSE
├── pyproject.toml
├── README.md
├── requirements.lock
├── src
│ └── zendesk_mcp_server
│ ├── __init__.py
│ ├── server.py
│ └── zendesk_client.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
1 | 3.12
2 |
```
--------------------------------------------------------------------------------
/.env.example:
--------------------------------------------------------------------------------
```
1 | ZENDESK_SUBDOMAIN=xxx
2 | ZENDESK_EMAIL=xxx
3 | ZENDESK_API_KEY=xxx
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
1 | # Python-generated files
2 | __pycache__/
3 | *.py[oc]
4 | build/
5 | dist/
6 | wheels/
7 | *.egg-info
8 |
9 | # Virtual environments
10 | .venv
11 |
12 | # Intellij Idea files
13 | .idea/
14 |
15 | .env
16 | .env.prod
17 | .env.dev
18 |
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
1 | # Zendesk MCP Server
2 |
3 | 
4 | [](https://opensource.org/licenses/Apache-2.0)
5 |
6 | A Model Context Protocol server for Zendesk.
7 |
8 | This server provides a comprehensive integration with Zendesk. It offers:
9 |
10 | - Tools for retrieving and managing Zendesk tickets and comments
11 | - Specialized prompts for ticket analysis and response drafting
12 | - Full access to the Zendesk Help Center articles as knowledge base
13 |
14 | 
15 |
16 | ## Setup
17 |
18 | - build: `uv venv && uv pip install -e .` or `uv build` in short.
19 | - setup zendesk credentials in `.env` file, refer to [.env.example](.env.example).
20 | - configure in Claude desktop:
21 |
22 | ```json
23 | {
24 | "mcpServers": {
25 | "zendesk": {
26 | "command": "uv",
27 | "args": [
28 | "--directory",
29 | "/path/to/zendesk-mcp-server",
30 | "run",
31 | "zendesk"
32 | ]
33 | }
34 | }
35 | }
36 | ```
37 |
38 | ### Docker
39 |
40 | You can containerize the server if you prefer an isolated runtime:
41 |
42 | 1. Copy `.env.example` to `.env` and fill in your Zendesk credentials. Keep this file outside version control.
43 | 2. Build the image:
44 |
45 | ```bash
46 | docker build -t zendesk-mcp-server .
47 | ```
48 |
49 | 3. Run the server, providing the environment file:
50 |
51 | ```bash
52 | docker run --rm --env-file /path/to/.env zendesk-mcp-server
53 | ```
54 |
55 | Add `-i` when wiring the container to MCP clients over STDIN/STDOUT (Claude Code uses this mode). For daemonized runs, add `-d --name zendesk-mcp`.
56 |
57 | The image installs dependencies from `requirements.lock`, drops privileges to a non-root user, and expects configuration exclusively via environment variables.
58 |
59 | #### Claude MCP Integration
60 |
61 | To use the Dockerized server from Claude Code/Desktop, add an entry to Claude Code's `settings.json` similar to:
62 |
63 | ```json
64 | {
65 | "mcpServers": {
66 | "zendesk": {
67 | "command": "/usr/local/bin/docker",
68 | "args": [
69 | "run",
70 | "--rm",
71 | "-i",
72 | "--env-file",
73 | "/path/to/zendesk-mcp-server/.env",
74 | "zendesk-mcp-server"
75 | ]
76 | }
77 | }
78 | }
79 | ```
80 |
81 | Adjust the paths to match your environment. After saving the file, restart Claude for the new MCP server to be detected.
82 |
83 | ## Resources
84 |
85 | - zendesk://knowledge-base, get access to the whole help center articles.
86 |
87 | ## Prompts
88 |
89 | ### analyze-ticket
90 |
91 | Analyze a Zendesk ticket and provide a detailed analysis of the ticket.
92 |
93 | ### draft-ticket-response
94 |
95 | Draft a response to a Zendesk ticket.
96 |
97 | ## Tools
98 |
99 | ### get_tickets
100 |
101 | Fetch the latest tickets with pagination support
102 |
103 | - Input:
104 | - `page` (integer, optional): Page number (defaults to 1)
105 | - `per_page` (integer, optional): Number of tickets per page, max 100 (defaults to 25)
106 | - `sort_by` (string, optional): Field to sort by - created_at, updated_at, priority, or status (defaults to created_at)
107 | - `sort_order` (string, optional): Sort order - asc or desc (defaults to desc)
108 |
109 | - Output: Returns a list of tickets with essential fields including id, subject, status, priority, description, timestamps, and assignee information, along with pagination metadata
110 |
111 | ### get_ticket
112 |
113 | Retrieve a Zendesk ticket by its ID
114 |
115 | - Input:
116 | - `ticket_id` (integer): The ID of the ticket to retrieve
117 |
118 | ### get_ticket_comments
119 |
120 | Retrieve all comments for a Zendesk ticket by its ID
121 |
122 | - Input:
123 | - `ticket_id` (integer): The ID of the ticket to get comments for
124 |
125 | ### create_ticket_comment
126 |
127 | Create a new comment on an existing Zendesk ticket
128 |
129 | - Input:
130 | - `ticket_id` (integer): The ID of the ticket to comment on
131 | - `comment` (string): The comment text/content to add
132 | - `public` (boolean, optional): Whether the comment should be public (defaults to true)
133 |
134 | ### create_ticket
135 |
136 | Create a new Zendesk ticket
137 |
138 | - Input:
139 | - `subject` (string): Ticket subject
140 | - `description` (string): Ticket description
141 | - `requester_id` (integer, optional)
142 | - `assignee_id` (integer, optional)
143 | - `priority` (string, optional): one of `low`, `normal`, `high`, `urgent`
144 | - `type` (string, optional): one of `problem`, `incident`, `question`, `task`
145 | - `tags` (array[string], optional)
146 | - `custom_fields` (array[object], optional)
147 |
148 | ### update_ticket
149 |
150 | Update fields on an existing Zendesk ticket (e.g., status, priority, assignee)
151 |
152 | - Input:
153 | - `ticket_id` (integer): The ID of the ticket to update
154 | - `subject` (string, optional)
155 | - `status` (string, optional): one of `new`, `open`, `pending`, `on-hold`, `solved`, `closed`
156 | - `priority` (string, optional): one of `low`, `normal`, `high`, `urgent`
157 | - `type` (string, optional)
158 | - `assignee_id` (integer, optional)
159 | - `requester_id` (integer, optional)
160 | - `tags` (array[string], optional)
161 | - `custom_fields` (array[object], optional)
162 | - `due_at` (string, optional): ISO8601 datetime
163 |
```
--------------------------------------------------------------------------------
/src/zendesk_mcp_server/__init__.py:
--------------------------------------------------------------------------------
```python
1 | import asyncio
2 |
3 | from . import server
4 |
5 |
6 | def main():
7 | asyncio.run(server.main())
8 |
9 |
10 | __all__ = ["main", "server"]
11 |
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
1 | [project]
2 | name = "zendesk-mcp-server"
3 | version = "0.1.0"
4 | description = "A simple Zendesk MCP server"
5 | readme = "README.md"
6 | requires-python = ">=3.12"
7 | dependencies = [
8 | "mcp>=1.1.2",
9 | "python-dotenv>=1.0.1",
10 | "zenpy>=2.0.56",
11 | ]
12 |
13 | [build-system]
14 | requires = [ "hatchling",]
15 | build-backend = "hatchling.build"
16 |
17 | [project.scripts]
18 | zendesk = "zendesk_mcp_server:main"
19 |
```
--------------------------------------------------------------------------------
/.github/workflows/ci.yml:
--------------------------------------------------------------------------------
```yaml
1 | name: CI
2 |
3 | on:
4 | push:
5 | branches: [ main ]
6 | pull_request:
7 | branches: [ main ]
8 |
9 | jobs:
10 | build:
11 | runs-on: ubuntu-latest
12 |
13 | steps:
14 | - uses: actions/checkout@v4
15 |
16 | - name: Install uv
17 | uses: astral-sh/setup-uv@v4
18 | with:
19 | enable-cache: true
20 | cache-dependency-glob: "uv.lock"
21 |
22 | - name: Set up Python
23 | run: uv python install
24 |
25 | - name: Build the project
26 | run: uv build
27 |
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
1 | FROM python:3.12-slim AS runtime
2 |
3 | ENV PYTHONDONTWRITEBYTECODE=1 \
4 | PYTHONUNBUFFERED=1 \
5 | PIP_DISABLE_PIP_VERSION_CHECK=1 \
6 | PIP_ROOT_USER_ACTION=ignore
7 |
8 | WORKDIR /app
9 |
10 | # Install minimal OS dependencies and create an unprivileged user
11 | RUN apt-get update \
12 | && apt-get install --no-install-recommends -y ca-certificates \
13 | && rm -rf /var/lib/apt/lists/* \
14 | && groupadd --system appuser \
15 | && useradd --system --gid appuser --shell /usr/sbin/nologin appuser
16 |
17 | COPY pyproject.toml requirements.lock README.md /app/
18 |
19 | RUN pip install --no-cache-dir --upgrade pip \
20 | && pip install --no-cache-dir -r requirements.lock \
21 | && rm -rf /root/.cache
22 |
23 | COPY src /app/src
24 |
25 | RUN pip install --no-cache-dir --no-deps .
26 |
27 | # Drop privileges for the runtime container
28 | USER appuser
29 |
30 | # Default command – expects Zendesk credentials via environment variables or an --env-file
31 | CMD ["zendesk"]
32 |
```
--------------------------------------------------------------------------------
/src/zendesk_mcp_server/zendesk_client.py:
--------------------------------------------------------------------------------
```python
1 | from typing import Dict, Any, List
2 | import json
3 | import urllib.request
4 | import urllib.parse
5 | import base64
6 |
7 | from zenpy import Zenpy
8 | from zenpy.lib.api_objects import Comment
9 | from zenpy.lib.api_objects import Ticket as ZenpyTicket
10 |
11 |
12 | class ZendeskClient:
13 | def __init__(self, subdomain: str, email: str, token: str):
14 | """
15 | Initialize the Zendesk client using zenpy lib and direct API.
16 | """
17 | self.client = Zenpy(
18 | subdomain=subdomain,
19 | email=email,
20 | token=token
21 | )
22 |
23 | # For direct API calls
24 | self.subdomain = subdomain
25 | self.email = email
26 | self.token = token
27 | self.base_url = f"https://{subdomain}.zendesk.com/api/v2"
28 | # Create basic auth header
29 | credentials = f"{email}/token:{token}"
30 | encoded_credentials = base64.b64encode(credentials.encode()).decode('ascii')
31 | self.auth_header = f"Basic {encoded_credentials}"
32 |
33 | def get_ticket(self, ticket_id: int) -> Dict[str, Any]:
34 | """
35 | Query a ticket by its ID
36 | """
37 | try:
38 | ticket = self.client.tickets(id=ticket_id)
39 | return {
40 | 'id': ticket.id,
41 | 'subject': ticket.subject,
42 | 'description': ticket.description,
43 | 'status': ticket.status,
44 | 'priority': ticket.priority,
45 | 'created_at': str(ticket.created_at),
46 | 'updated_at': str(ticket.updated_at),
47 | 'requester_id': ticket.requester_id,
48 | 'assignee_id': ticket.assignee_id,
49 | 'organization_id': ticket.organization_id
50 | }
51 | except Exception as e:
52 | raise Exception(f"Failed to get ticket {ticket_id}: {str(e)}")
53 |
54 | def get_ticket_comments(self, ticket_id: int) -> List[Dict[str, Any]]:
55 | """
56 | Get all comments for a specific ticket.
57 | """
58 | try:
59 | comments = self.client.tickets.comments(ticket=ticket_id)
60 | return [{
61 | 'id': comment.id,
62 | 'author_id': comment.author_id,
63 | 'body': comment.body,
64 | 'html_body': comment.html_body,
65 | 'public': comment.public,
66 | 'created_at': str(comment.created_at)
67 | } for comment in comments]
68 | except Exception as e:
69 | raise Exception(f"Failed to get comments for ticket {ticket_id}: {str(e)}")
70 |
71 | def post_comment(self, ticket_id: int, comment: str, public: bool = True) -> str:
72 | """
73 | Post a comment to an existing ticket.
74 | """
75 | try:
76 | ticket = self.client.tickets(id=ticket_id)
77 | ticket.comment = Comment(
78 | html_body=comment,
79 | public=public
80 | )
81 | self.client.tickets.update(ticket)
82 | return comment
83 | except Exception as e:
84 | raise Exception(f"Failed to post comment on ticket {ticket_id}: {str(e)}")
85 |
86 | def get_tickets(self, page: int = 1, per_page: int = 25, sort_by: str = 'created_at', sort_order: str = 'desc') -> Dict[str, Any]:
87 | """
88 | Get the latest tickets with proper pagination support using direct API calls.
89 |
90 | Args:
91 | page: Page number (1-based)
92 | per_page: Number of tickets per page (max 100)
93 | sort_by: Field to sort by (created_at, updated_at, priority, status)
94 | sort_order: Sort order (asc or desc)
95 |
96 | Returns:
97 | Dict containing tickets and pagination info
98 | """
99 | try:
100 | # Cap at reasonable limit
101 | per_page = min(per_page, 100)
102 |
103 | # Build URL with parameters for offset pagination
104 | params = {
105 | 'page': str(page),
106 | 'per_page': str(per_page),
107 | 'sort_by': sort_by,
108 | 'sort_order': sort_order
109 | }
110 | query_string = urllib.parse.urlencode(params)
111 | url = f"{self.base_url}/tickets.json?{query_string}"
112 |
113 | # Create request with auth header
114 | req = urllib.request.Request(url)
115 | req.add_header('Authorization', self.auth_header)
116 | req.add_header('Content-Type', 'application/json')
117 |
118 | # Make the API request
119 | with urllib.request.urlopen(req) as response:
120 | data = json.loads(response.read().decode())
121 |
122 | tickets_data = data.get('tickets', [])
123 |
124 | # Process tickets to return only essential fields
125 | ticket_list = []
126 | for ticket in tickets_data:
127 | ticket_list.append({
128 | 'id': ticket.get('id'),
129 | 'subject': ticket.get('subject'),
130 | 'status': ticket.get('status'),
131 | 'priority': ticket.get('priority'),
132 | 'description': ticket.get('description'),
133 | 'created_at': ticket.get('created_at'),
134 | 'updated_at': ticket.get('updated_at'),
135 | 'requester_id': ticket.get('requester_id'),
136 | 'assignee_id': ticket.get('assignee_id')
137 | })
138 |
139 | return {
140 | 'tickets': ticket_list,
141 | 'page': page,
142 | 'per_page': per_page,
143 | 'count': len(ticket_list),
144 | 'sort_by': sort_by,
145 | 'sort_order': sort_order,
146 | 'has_more': data.get('next_page') is not None,
147 | 'next_page': page + 1 if data.get('next_page') else None,
148 | 'previous_page': page - 1 if data.get('previous_page') and page > 1 else None
149 | }
150 | except urllib.error.HTTPError as e:
151 | error_body = e.read().decode() if e.fp else "No response body"
152 | raise Exception(f"Failed to get latest tickets: HTTP {e.code} - {e.reason}. {error_body}")
153 | except Exception as e:
154 | raise Exception(f"Failed to get latest tickets: {str(e)}")
155 |
156 | def get_all_articles(self) -> Dict[str, Any]:
157 | """
158 | Fetch help center articles as knowledge base.
159 | Returns a Dict of section -> [article].
160 | """
161 | try:
162 | # Get all sections
163 | sections = self.client.help_center.sections()
164 |
165 | # Get articles for each section
166 | kb = {}
167 | for section in sections:
168 | articles = self.client.help_center.sections.articles(section.id)
169 | kb[section.name] = {
170 | 'section_id': section.id,
171 | 'description': section.description,
172 | 'articles': [{
173 | 'id': article.id,
174 | 'title': article.title,
175 | 'body': article.body,
176 | 'updated_at': str(article.updated_at),
177 | 'url': article.html_url
178 | } for article in articles]
179 | }
180 |
181 | return kb
182 | except Exception as e:
183 | raise Exception(f"Failed to fetch knowledge base: {str(e)}")
184 |
185 | def create_ticket(
186 | self,
187 | subject: str,
188 | description: str,
189 | requester_id: int | None = None,
190 | assignee_id: int | None = None,
191 | priority: str | None = None,
192 | type: str | None = None,
193 | tags: List[str] | None = None,
194 | custom_fields: List[Dict[str, Any]] | None = None,
195 | ) -> Dict[str, Any]:
196 | """
197 | Create a new Zendesk ticket using Zenpy and return essential fields.
198 |
199 | Args:
200 | subject: Ticket subject
201 | description: Ticket description (plain text). Will also be used as initial comment.
202 | requester_id: Optional requester user ID
203 | assignee_id: Optional assignee user ID
204 | priority: Optional priority (low, normal, high, urgent)
205 | type: Optional ticket type (problem, incident, question, task)
206 | tags: Optional list of tags
207 | custom_fields: Optional list of dicts: {id: int, value: Any}
208 | """
209 | try:
210 | ticket = ZenpyTicket(
211 | subject=subject,
212 | description=description,
213 | requester_id=requester_id,
214 | assignee_id=assignee_id,
215 | priority=priority,
216 | type=type,
217 | tags=tags,
218 | custom_fields=custom_fields,
219 | )
220 | created_audit = self.client.tickets.create(ticket)
221 | # Fetch created ticket id from audit
222 | created_ticket_id = getattr(getattr(created_audit, 'ticket', None), 'id', None)
223 | if created_ticket_id is None:
224 | # Fallback: try to read id from audit events
225 | created_ticket_id = getattr(created_audit, 'id', None)
226 |
227 | # Fetch full ticket to return consistent data
228 | created = self.client.tickets(id=created_ticket_id) if created_ticket_id else None
229 |
230 | return {
231 | 'id': getattr(created, 'id', created_ticket_id),
232 | 'subject': getattr(created, 'subject', subject),
233 | 'description': getattr(created, 'description', description),
234 | 'status': getattr(created, 'status', 'new'),
235 | 'priority': getattr(created, 'priority', priority),
236 | 'type': getattr(created, 'type', type),
237 | 'created_at': str(getattr(created, 'created_at', '')),
238 | 'updated_at': str(getattr(created, 'updated_at', '')),
239 | 'requester_id': getattr(created, 'requester_id', requester_id),
240 | 'assignee_id': getattr(created, 'assignee_id', assignee_id),
241 | 'organization_id': getattr(created, 'organization_id', None),
242 | 'tags': list(getattr(created, 'tags', tags or []) or []),
243 | }
244 | except Exception as e:
245 | raise Exception(f"Failed to create ticket: {str(e)}")
246 |
247 | def update_ticket(self, ticket_id: int, **fields: Any) -> Dict[str, Any]:
248 | """
249 | Update a Zendesk ticket with provided fields using Zenpy.
250 |
251 | Supported fields include common ticket attributes like:
252 | subject, status, priority, type, assignee_id, requester_id,
253 | tags (list[str]), custom_fields (list[dict]), due_at, etc.
254 | """
255 | try:
256 | # Load the ticket, mutate fields directly, and update
257 | ticket = self.client.tickets(id=ticket_id)
258 | for key, value in fields.items():
259 | if value is None:
260 | continue
261 | setattr(ticket, key, value)
262 |
263 | # This call returns a TicketAudit (not a Ticket). Don't read attrs from it.
264 | self.client.tickets.update(ticket)
265 |
266 | # Fetch the fresh ticket to return consistent data
267 | refreshed = self.client.tickets(id=ticket_id)
268 |
269 | return {
270 | 'id': refreshed.id,
271 | 'subject': refreshed.subject,
272 | 'description': refreshed.description,
273 | 'status': refreshed.status,
274 | 'priority': refreshed.priority,
275 | 'type': getattr(refreshed, 'type', None),
276 | 'created_at': str(refreshed.created_at),
277 | 'updated_at': str(refreshed.updated_at),
278 | 'requester_id': refreshed.requester_id,
279 | 'assignee_id': refreshed.assignee_id,
280 | 'organization_id': refreshed.organization_id,
281 | 'tags': list(getattr(refreshed, 'tags', []) or []),
282 | }
283 | except Exception as e:
284 | raise Exception(f"Failed to update ticket {ticket_id}: {str(e)}")
```
--------------------------------------------------------------------------------
/src/zendesk_mcp_server/server.py:
--------------------------------------------------------------------------------
```python
1 | import asyncio
2 | import json
3 | import logging
4 | import os
5 | from typing import Any, Dict
6 |
7 | from cachetools.func import ttl_cache
8 | from dotenv import load_dotenv
9 | from mcp.server import InitializationOptions, NotificationOptions
10 | from mcp.server import Server, types
11 | from mcp.server.stdio import stdio_server
12 | from pydantic import AnyUrl
13 |
14 | from zendesk_mcp_server.zendesk_client import ZendeskClient
15 |
16 | logging.basicConfig(
17 | level=logging.INFO,
18 | format='%(asctime)s - %(levelname)s - %(message)s',
19 | datefmt='%Y-%m-%d %H:%M:%S'
20 | )
21 | logger = logging.getLogger("zendesk-mcp-server")
22 | logger.info("zendesk mcp server started")
23 |
24 | load_dotenv()
25 | zendesk_client = ZendeskClient(
26 | subdomain=os.getenv("ZENDESK_SUBDOMAIN"),
27 | email=os.getenv("ZENDESK_EMAIL"),
28 | token=os.getenv("ZENDESK_API_KEY")
29 | )
30 |
31 | server = Server("Zendesk Server")
32 |
33 | TICKET_ANALYSIS_TEMPLATE = """
34 | You are a helpful Zendesk support analyst. You've been asked to analyze ticket #{ticket_id}.
35 |
36 | Please fetch the ticket info and comments to analyze it and provide:
37 | 1. A summary of the issue
38 | 2. The current status and timeline
39 | 3. Key points of interaction
40 |
41 | Remember to be professional and focus on actionable insights.
42 | """
43 |
44 | COMMENT_DRAFT_TEMPLATE = """
45 | You are a helpful Zendesk support agent. You need to draft a response to ticket #{ticket_id}.
46 |
47 | Please fetch the ticket info, comments and knowledge base to draft a professional and helpful response that:
48 | 1. Acknowledges the customer's concern
49 | 2. Addresses the specific issues raised
50 | 3. Provides clear next steps or ask for specific details need to proceed
51 | 4. Maintains a friendly and professional tone
52 | 5. Ask for confirmation before commenting on the ticket
53 |
54 | The response should be formatted well and ready to be posted as a comment.
55 | """
56 |
57 |
58 | @server.list_prompts()
59 | async def handle_list_prompts() -> list[types.Prompt]:
60 | """List available prompts"""
61 | return [
62 | types.Prompt(
63 | name="analyze-ticket",
64 | description="Analyze a Zendesk ticket and provide insights",
65 | arguments=[
66 | types.PromptArgument(
67 | name="ticket_id",
68 | description="The ID of the ticket to analyze",
69 | required=True,
70 | )
71 | ],
72 | ),
73 | types.Prompt(
74 | name="draft-ticket-response",
75 | description="Draft a professional response to a Zendesk ticket",
76 | arguments=[
77 | types.PromptArgument(
78 | name="ticket_id",
79 | description="The ID of the ticket to respond to",
80 | required=True,
81 | )
82 | ],
83 | )
84 | ]
85 |
86 |
87 | @server.get_prompt()
88 | async def handle_get_prompt(name: str, arguments: Dict[str, str] | None) -> types.GetPromptResult:
89 | """Handle prompt requests"""
90 | if not arguments or "ticket_id" not in arguments:
91 | raise ValueError("Missing required argument: ticket_id")
92 |
93 | ticket_id = int(arguments["ticket_id"])
94 | try:
95 | if name == "analyze-ticket":
96 | prompt = TICKET_ANALYSIS_TEMPLATE.format(
97 | ticket_id=ticket_id
98 | )
99 | description = f"Analysis prompt for ticket #{ticket_id}"
100 |
101 | elif name == "draft-ticket-response":
102 | prompt = COMMENT_DRAFT_TEMPLATE.format(
103 | ticket_id=ticket_id
104 | )
105 | description = f"Response draft prompt for ticket #{ticket_id}"
106 |
107 | else:
108 | raise ValueError(f"Unknown prompt: {name}")
109 |
110 | return types.GetPromptResult(
111 | description=description,
112 | messages=[
113 | types.PromptMessage(
114 | role="user",
115 | content=types.TextContent(type="text", text=prompt.strip()),
116 | )
117 | ],
118 | )
119 |
120 | except Exception as e:
121 | logger.error(f"Error generating prompt: {e}")
122 | raise
123 |
124 |
125 | @server.list_tools()
126 | async def handle_list_tools() -> list[types.Tool]:
127 | """List available Zendesk tools"""
128 | return [
129 | types.Tool(
130 | name="get_ticket",
131 | description="Retrieve a Zendesk ticket by its ID",
132 | inputSchema={
133 | "type": "object",
134 | "properties": {
135 | "ticket_id": {
136 | "type": "integer",
137 | "description": "The ID of the ticket to retrieve"
138 | }
139 | },
140 | "required": ["ticket_id"]
141 | }
142 | ),
143 | types.Tool(
144 | name="create_ticket",
145 | description="Create a new Zendesk ticket",
146 | inputSchema={
147 | "type": "object",
148 | "properties": {
149 | "subject": {"type": "string", "description": "Ticket subject"},
150 | "description": {"type": "string", "description": "Ticket description"},
151 | "requester_id": {"type": "integer"},
152 | "assignee_id": {"type": "integer"},
153 | "priority": {"type": "string", "description": "low, normal, high, urgent"},
154 | "type": {"type": "string", "description": "problem, incident, question, task"},
155 | "tags": {"type": "array", "items": {"type": "string"}},
156 | "custom_fields": {"type": "array", "items": {"type": "object"}},
157 | },
158 | "required": ["subject", "description"],
159 | }
160 | ),
161 | types.Tool(
162 | name="get_tickets",
163 | description="Fetch the latest tickets with pagination support",
164 | inputSchema={
165 | "type": "object",
166 | "properties": {
167 | "page": {
168 | "type": "integer",
169 | "description": "Page number",
170 | "default": 1
171 | },
172 | "per_page": {
173 | "type": "integer",
174 | "description": "Number of tickets per page (max 100)",
175 | "default": 25
176 | },
177 | "sort_by": {
178 | "type": "string",
179 | "description": "Field to sort by (created_at, updated_at, priority, status)",
180 | "default": "created_at"
181 | },
182 | "sort_order": {
183 | "type": "string",
184 | "description": "Sort order (asc or desc)",
185 | "default": "desc"
186 | }
187 | },
188 | "required": []
189 | }
190 | ),
191 | types.Tool(
192 | name="get_ticket_comments",
193 | description="Retrieve all comments for a Zendesk ticket by its ID",
194 | inputSchema={
195 | "type": "object",
196 | "properties": {
197 | "ticket_id": {
198 | "type": "integer",
199 | "description": "The ID of the ticket to get comments for"
200 | }
201 | },
202 | "required": ["ticket_id"]
203 | }
204 | ),
205 | types.Tool(
206 | name="create_ticket_comment",
207 | description="Create a new comment on an existing Zendesk ticket",
208 | inputSchema={
209 | "type": "object",
210 | "properties": {
211 | "ticket_id": {
212 | "type": "integer",
213 | "description": "The ID of the ticket to comment on"
214 | },
215 | "comment": {
216 | "type": "string",
217 | "description": "The comment text/content to add"
218 | },
219 | "public": {
220 | "type": "boolean",
221 | "description": "Whether the comment should be public",
222 | "default": True
223 | }
224 | },
225 | "required": ["ticket_id", "comment"]
226 | }
227 | ),
228 | types.Tool(
229 | name="update_ticket",
230 | description="Update fields on an existing Zendesk ticket (e.g., status, priority, assignee_id)",
231 | inputSchema={
232 | "type": "object",
233 | "properties": {
234 | "ticket_id": {"type": "integer", "description": "The ID of the ticket to update"},
235 | "subject": {"type": "string"},
236 | "status": {"type": "string", "description": "new, open, pending, on-hold, solved, closed"},
237 | "priority": {"type": "string", "description": "low, normal, high, urgent"},
238 | "type": {"type": "string"},
239 | "assignee_id": {"type": "integer"},
240 | "requester_id": {"type": "integer"},
241 | "tags": {"type": "array", "items": {"type": "string"}},
242 | "custom_fields": {"type": "array", "items": {"type": "object"}},
243 | "due_at": {"type": "string", "description": "ISO8601 datetime"}
244 | },
245 | "required": ["ticket_id"]
246 | }
247 | )
248 | ]
249 |
250 |
251 | @server.call_tool()
252 | async def handle_call_tool(
253 | name: str,
254 | arguments: dict[str, Any] | None
255 | ) -> list[types.TextContent]:
256 | """Handle Zendesk tool execution requests"""
257 | try:
258 | if name == "get_ticket":
259 | if not arguments:
260 | raise ValueError("Missing arguments")
261 | ticket = zendesk_client.get_ticket(arguments["ticket_id"])
262 | return [types.TextContent(
263 | type="text",
264 | text=json.dumps(ticket)
265 | )]
266 |
267 | elif name == "create_ticket":
268 | if not arguments:
269 | raise ValueError("Missing arguments")
270 | created = zendesk_client.create_ticket(
271 | subject=arguments.get("subject"),
272 | description=arguments.get("description"),
273 | requester_id=arguments.get("requester_id"),
274 | assignee_id=arguments.get("assignee_id"),
275 | priority=arguments.get("priority"),
276 | type=arguments.get("type"),
277 | tags=arguments.get("tags"),
278 | custom_fields=arguments.get("custom_fields"),
279 | )
280 | return [types.TextContent(
281 | type="text",
282 | text=json.dumps({"message": "Ticket created successfully", "ticket": created}, indent=2)
283 | )]
284 |
285 | elif name == "get_tickets":
286 | page = arguments.get("page", 1) if arguments else 1
287 | per_page = arguments.get("per_page", 25) if arguments else 25
288 | sort_by = arguments.get("sort_by", "created_at") if arguments else "created_at"
289 | sort_order = arguments.get("sort_order", "desc") if arguments else "desc"
290 |
291 | tickets = zendesk_client.get_tickets(
292 | page=page,
293 | per_page=per_page,
294 | sort_by=sort_by,
295 | sort_order=sort_order
296 | )
297 | return [types.TextContent(
298 | type="text",
299 | text=json.dumps(tickets, indent=2)
300 | )]
301 |
302 | elif name == "get_ticket_comments":
303 | if not arguments:
304 | raise ValueError("Missing arguments")
305 | comments = zendesk_client.get_ticket_comments(
306 | arguments["ticket_id"])
307 | return [types.TextContent(
308 | type="text",
309 | text=json.dumps(comments)
310 | )]
311 |
312 | elif name == "create_ticket_comment":
313 | if not arguments:
314 | raise ValueError("Missing arguments")
315 | public = arguments.get("public", True)
316 | result = zendesk_client.post_comment(
317 | ticket_id=arguments["ticket_id"],
318 | comment=arguments["comment"],
319 | public=public
320 | )
321 | return [types.TextContent(
322 | type="text",
323 | text=f"Comment created successfully: {result}"
324 | )]
325 |
326 | elif name == "update_ticket":
327 | if not arguments:
328 | raise ValueError("Missing arguments")
329 | ticket_id = arguments.get("ticket_id")
330 | if ticket_id is None:
331 | raise ValueError("ticket_id is required")
332 | update_fields = {k: v for k, v in arguments.items() if k != "ticket_id"}
333 | updated = zendesk_client.update_ticket(ticket_id=int(ticket_id), **update_fields)
334 | return [types.TextContent(
335 | type="text",
336 | text=json.dumps({"message": "Ticket updated successfully", "ticket": updated}, indent=2)
337 | )]
338 |
339 | else:
340 | raise ValueError(f"Unknown tool: {name}")
341 |
342 | except Exception as e:
343 | return [types.TextContent(
344 | type="text",
345 | text=f"Error: {str(e)}"
346 | )]
347 |
348 |
349 | @server.list_resources()
350 | async def handle_list_resources() -> list[types.Resource]:
351 | logger.debug("Handling list_resources request")
352 | return [
353 | types.Resource(
354 | uri=AnyUrl("zendesk://knowledge-base"),
355 | name="Zendesk Knowledge Base",
356 | description="Access to Zendesk Help Center articles and sections",
357 | mimeType="application/json",
358 | )
359 | ]
360 |
361 |
362 | @ttl_cache(ttl=3600)
363 | def get_cached_kb():
364 | return zendesk_client.get_all_articles()
365 |
366 |
367 | @server.read_resource()
368 | async def handle_read_resource(uri: AnyUrl) -> str:
369 | logger.debug(f"Handling read_resource request for URI: {uri}")
370 | if uri.scheme != "zendesk":
371 | logger.error(f"Unsupported URI scheme: {uri.scheme}")
372 | raise ValueError(f"Unsupported URI scheme: {uri.scheme}")
373 |
374 | path = str(uri).replace("zendesk://", "")
375 | if path != "knowledge-base":
376 | logger.error(f"Unknown resource path: {path}")
377 | raise ValueError(f"Unknown resource path: {path}")
378 |
379 | try:
380 | kb_data = get_cached_kb()
381 | return json.dumps({
382 | "knowledge_base": kb_data,
383 | "metadata": {
384 | "sections": len(kb_data),
385 | "total_articles": sum(len(section['articles']) for section in kb_data.values()),
386 | }
387 | }, indent=2)
388 | except Exception as e:
389 | logger.error(f"Error fetching knowledge base: {e}")
390 | raise
391 |
392 |
393 | async def main():
394 | # Run the server using stdin/stdout streams
395 | async with stdio_server() as (read_stream, write_stream):
396 | await server.run(
397 | read_stream=read_stream,
398 | write_stream=write_stream,
399 | initialization_options=InitializationOptions(
400 | server_name="Zendesk",
401 | server_version="0.1.0",
402 | capabilities=server.get_capabilities(
403 | notification_options=NotificationOptions(),
404 | experimental_capabilities={},
405 | ),
406 | ),
407 | )
408 |
409 |
410 | if __name__ == "__main__":
411 | asyncio.run(main())
412 |
```