# 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:
--------------------------------------------------------------------------------
```
3.12
```
--------------------------------------------------------------------------------
/.env.example:
--------------------------------------------------------------------------------
```
ZENDESK_SUBDOMAIN=xxx
ZENDESK_EMAIL=xxx
ZENDESK_API_KEY=xxx
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info
# Virtual environments
.venv
# Intellij Idea files
.idea/
.env
.env.prod
.env.dev
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Zendesk MCP Server
[](https://opensource.org/licenses/Apache-2.0)
A Model Context Protocol server for Zendesk.
This server provides a comprehensive integration with Zendesk. It offers:
- Tools for retrieving and managing Zendesk tickets and comments
- Specialized prompts for ticket analysis and response drafting
- Full access to the Zendesk Help Center articles as knowledge base
## Setup
- build: `uv venv && uv pip install -e .` or `uv build` in short.
- setup zendesk credentials in `.env` file, refer to [.env.example](.env.example).
- configure in Claude desktop:
```json
{
"mcpServers": {
"zendesk": {
"command": "uv",
"args": [
"--directory",
"/path/to/zendesk-mcp-server",
"run",
"zendesk"
]
}
}
}
```
### Docker
You can containerize the server if you prefer an isolated runtime:
1. Copy `.env.example` to `.env` and fill in your Zendesk credentials. Keep this file outside version control.
2. Build the image:
```bash
docker build -t zendesk-mcp-server .
```
3. Run the server, providing the environment file:
```bash
docker run --rm --env-file /path/to/.env zendesk-mcp-server
```
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`.
The image installs dependencies from `requirements.lock`, drops privileges to a non-root user, and expects configuration exclusively via environment variables.
#### Claude MCP Integration
To use the Dockerized server from Claude Code/Desktop, add an entry to Claude Code's `settings.json` similar to:
```json
{
"mcpServers": {
"zendesk": {
"command": "/usr/local/bin/docker",
"args": [
"run",
"--rm",
"-i",
"--env-file",
"/path/to/zendesk-mcp-server/.env",
"zendesk-mcp-server"
]
}
}
}
```
Adjust the paths to match your environment. After saving the file, restart Claude for the new MCP server to be detected.
## Resources
- zendesk://knowledge-base, get access to the whole help center articles.
## Prompts
### analyze-ticket
Analyze a Zendesk ticket and provide a detailed analysis of the ticket.
### draft-ticket-response
Draft a response to a Zendesk ticket.
## Tools
### get_tickets
Fetch the latest tickets with pagination support
- Input:
- `page` (integer, optional): Page number (defaults to 1)
- `per_page` (integer, optional): Number of tickets per page, max 100 (defaults to 25)
- `sort_by` (string, optional): Field to sort by - created_at, updated_at, priority, or status (defaults to created_at)
- `sort_order` (string, optional): Sort order - asc or desc (defaults to desc)
- Output: Returns a list of tickets with essential fields including id, subject, status, priority, description, timestamps, and assignee information, along with pagination metadata
### get_ticket
Retrieve a Zendesk ticket by its ID
- Input:
- `ticket_id` (integer): The ID of the ticket to retrieve
### get_ticket_comments
Retrieve all comments for a Zendesk ticket by its ID
- Input:
- `ticket_id` (integer): The ID of the ticket to get comments for
### create_ticket_comment
Create a new comment on an existing Zendesk ticket
- Input:
- `ticket_id` (integer): The ID of the ticket to comment on
- `comment` (string): The comment text/content to add
- `public` (boolean, optional): Whether the comment should be public (defaults to true)
### create_ticket
Create a new Zendesk ticket
- Input:
- `subject` (string): Ticket subject
- `description` (string): Ticket description
- `requester_id` (integer, optional)
- `assignee_id` (integer, optional)
- `priority` (string, optional): one of `low`, `normal`, `high`, `urgent`
- `type` (string, optional): one of `problem`, `incident`, `question`, `task`
- `tags` (array[string], optional)
- `custom_fields` (array[object], optional)
### update_ticket
Update fields on an existing Zendesk ticket (e.g., status, priority, assignee)
- Input:
- `ticket_id` (integer): The ID of the ticket to update
- `subject` (string, optional)
- `status` (string, optional): one of `new`, `open`, `pending`, `on-hold`, `solved`, `closed`
- `priority` (string, optional): one of `low`, `normal`, `high`, `urgent`
- `type` (string, optional)
- `assignee_id` (integer, optional)
- `requester_id` (integer, optional)
- `tags` (array[string], optional)
- `custom_fields` (array[object], optional)
- `due_at` (string, optional): ISO8601 datetime
```
--------------------------------------------------------------------------------
/src/zendesk_mcp_server/__init__.py:
--------------------------------------------------------------------------------
```python
import asyncio
from . import server
def main():
asyncio.run(server.main())
__all__ = ["main", "server"]
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "zendesk-mcp-server"
version = "0.1.0"
description = "A simple Zendesk MCP server"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
"mcp>=1.1.2",
"python-dotenv>=1.0.1",
"zenpy>=2.0.56",
]
[build-system]
requires = [ "hatchling",]
build-backend = "hatchling.build"
[project.scripts]
zendesk = "zendesk_mcp_server:main"
```
--------------------------------------------------------------------------------
/.github/workflows/ci.yml:
--------------------------------------------------------------------------------
```yaml
name: CI
on:
push:
branches: [ main ]
pull_request:
branches: [ main ]
jobs:
build:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Install uv
uses: astral-sh/setup-uv@v4
with:
enable-cache: true
cache-dependency-glob: "uv.lock"
- name: Set up Python
run: uv python install
- name: Build the project
run: uv build
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
FROM python:3.12-slim AS runtime
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PIP_DISABLE_PIP_VERSION_CHECK=1 \
PIP_ROOT_USER_ACTION=ignore
WORKDIR /app
# Install minimal OS dependencies and create an unprivileged user
RUN apt-get update \
&& apt-get install --no-install-recommends -y ca-certificates \
&& rm -rf /var/lib/apt/lists/* \
&& groupadd --system appuser \
&& useradd --system --gid appuser --shell /usr/sbin/nologin appuser
COPY pyproject.toml requirements.lock README.md /app/
RUN pip install --no-cache-dir --upgrade pip \
&& pip install --no-cache-dir -r requirements.lock \
&& rm -rf /root/.cache
COPY src /app/src
RUN pip install --no-cache-dir --no-deps .
# Drop privileges for the runtime container
USER appuser
# Default command – expects Zendesk credentials via environment variables or an --env-file
CMD ["zendesk"]
```
--------------------------------------------------------------------------------
/src/zendesk_mcp_server/zendesk_client.py:
--------------------------------------------------------------------------------
```python
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
from zenpy.lib.api_objects import Ticket as ZenpyTicket
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)}")
def create_ticket(
self,
subject: str,
description: str,
requester_id: int | None = None,
assignee_id: int | None = None,
priority: str | None = None,
type: str | None = None,
tags: List[str] | None = None,
custom_fields: List[Dict[str, Any]] | None = None,
) -> Dict[str, Any]:
"""
Create a new Zendesk ticket using Zenpy and return essential fields.
Args:
subject: Ticket subject
description: Ticket description (plain text). Will also be used as initial comment.
requester_id: Optional requester user ID
assignee_id: Optional assignee user ID
priority: Optional priority (low, normal, high, urgent)
type: Optional ticket type (problem, incident, question, task)
tags: Optional list of tags
custom_fields: Optional list of dicts: {id: int, value: Any}
"""
try:
ticket = ZenpyTicket(
subject=subject,
description=description,
requester_id=requester_id,
assignee_id=assignee_id,
priority=priority,
type=type,
tags=tags,
custom_fields=custom_fields,
)
created_audit = self.client.tickets.create(ticket)
# Fetch created ticket id from audit
created_ticket_id = getattr(getattr(created_audit, 'ticket', None), 'id', None)
if created_ticket_id is None:
# Fallback: try to read id from audit events
created_ticket_id = getattr(created_audit, 'id', None)
# Fetch full ticket to return consistent data
created = self.client.tickets(id=created_ticket_id) if created_ticket_id else None
return {
'id': getattr(created, 'id', created_ticket_id),
'subject': getattr(created, 'subject', subject),
'description': getattr(created, 'description', description),
'status': getattr(created, 'status', 'new'),
'priority': getattr(created, 'priority', priority),
'type': getattr(created, 'type', type),
'created_at': str(getattr(created, 'created_at', '')),
'updated_at': str(getattr(created, 'updated_at', '')),
'requester_id': getattr(created, 'requester_id', requester_id),
'assignee_id': getattr(created, 'assignee_id', assignee_id),
'organization_id': getattr(created, 'organization_id', None),
'tags': list(getattr(created, 'tags', tags or []) or []),
}
except Exception as e:
raise Exception(f"Failed to create ticket: {str(e)}")
def update_ticket(self, ticket_id: int, **fields: Any) -> Dict[str, Any]:
"""
Update a Zendesk ticket with provided fields using Zenpy.
Supported fields include common ticket attributes like:
subject, status, priority, type, assignee_id, requester_id,
tags (list[str]), custom_fields (list[dict]), due_at, etc.
"""
try:
# Load the ticket, mutate fields directly, and update
ticket = self.client.tickets(id=ticket_id)
for key, value in fields.items():
if value is None:
continue
setattr(ticket, key, value)
# This call returns a TicketAudit (not a Ticket). Don't read attrs from it.
self.client.tickets.update(ticket)
# Fetch the fresh ticket to return consistent data
refreshed = self.client.tickets(id=ticket_id)
return {
'id': refreshed.id,
'subject': refreshed.subject,
'description': refreshed.description,
'status': refreshed.status,
'priority': refreshed.priority,
'type': getattr(refreshed, 'type', None),
'created_at': str(refreshed.created_at),
'updated_at': str(refreshed.updated_at),
'requester_id': refreshed.requester_id,
'assignee_id': refreshed.assignee_id,
'organization_id': refreshed.organization_id,
'tags': list(getattr(refreshed, 'tags', []) or []),
}
except Exception as e:
raise Exception(f"Failed to update ticket {ticket_id}: {str(e)}")
```
--------------------------------------------------------------------------------
/src/zendesk_mcp_server/server.py:
--------------------------------------------------------------------------------
```python
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="create_ticket",
description="Create a new Zendesk ticket",
inputSchema={
"type": "object",
"properties": {
"subject": {"type": "string", "description": "Ticket subject"},
"description": {"type": "string", "description": "Ticket description"},
"requester_id": {"type": "integer"},
"assignee_id": {"type": "integer"},
"priority": {"type": "string", "description": "low, normal, high, urgent"},
"type": {"type": "string", "description": "problem, incident, question, task"},
"tags": {"type": "array", "items": {"type": "string"}},
"custom_fields": {"type": "array", "items": {"type": "object"}},
},
"required": ["subject", "description"],
}
),
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"]
}
),
types.Tool(
name="update_ticket",
description="Update fields on an existing Zendesk ticket (e.g., status, priority, assignee_id)",
inputSchema={
"type": "object",
"properties": {
"ticket_id": {"type": "integer", "description": "The ID of the ticket to update"},
"subject": {"type": "string"},
"status": {"type": "string", "description": "new, open, pending, on-hold, solved, closed"},
"priority": {"type": "string", "description": "low, normal, high, urgent"},
"type": {"type": "string"},
"assignee_id": {"type": "integer"},
"requester_id": {"type": "integer"},
"tags": {"type": "array", "items": {"type": "string"}},
"custom_fields": {"type": "array", "items": {"type": "object"}},
"due_at": {"type": "string", "description": "ISO8601 datetime"}
},
"required": ["ticket_id"]
}
)
]
@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 == "create_ticket":
if not arguments:
raise ValueError("Missing arguments")
created = zendesk_client.create_ticket(
subject=arguments.get("subject"),
description=arguments.get("description"),
requester_id=arguments.get("requester_id"),
assignee_id=arguments.get("assignee_id"),
priority=arguments.get("priority"),
type=arguments.get("type"),
tags=arguments.get("tags"),
custom_fields=arguments.get("custom_fields"),
)
return [types.TextContent(
type="text",
text=json.dumps({"message": "Ticket created successfully", "ticket": created}, indent=2)
)]
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}"
)]
elif name == "update_ticket":
if not arguments:
raise ValueError("Missing arguments")
ticket_id = arguments.get("ticket_id")
if ticket_id is None:
raise ValueError("ticket_id is required")
update_fields = {k: v for k, v in arguments.items() if k != "ticket_id"}
updated = zendesk_client.update_ticket(ticket_id=int(ticket_id), **update_fields)
return [types.TextContent(
type="text",
text=json.dumps({"message": "Ticket updated successfully", "ticket": updated}, indent=2)
)]
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())
```