import os
import httpx
from typing import Any, Optional
class ZendeskClient:
def __init__(self):
self.subdomain = os.environ.get("ZENDESK_SUBDOMAIN", "")
self.email = os.environ.get("ZENDESK_EMAIL", "")
self.api_token = os.environ.get("ZENDESK_API_TOKEN", "")
if not all([self.subdomain, self.email, self.api_token]):
raise ValueError("ZENDESK_SUBDOMAIN, ZENDESK_EMAIL, and ZENDESK_API_TOKEN are required")
self.base_url = f"https://{self.subdomain}.zendesk.com/api/v2"
self.auth = (f"{self.email}/token", self.api_token)
async def _request(
self,
method: str,
path: str,
json: Optional[dict] = None,
params: Optional[dict] = None
) -> Any:
async with httpx.AsyncClient() as client:
response = await client.request(
method=method,
url=f"{self.base_url}{path}",
auth=self.auth,
json=json,
params=params,
headers={"Content-Type": "application/json"}
)
response.raise_for_status()
if response.content:
return response.json()
return None
async def get_tickets(
self,
status: Optional[str] = None,
priority: Optional[str] = None,
sort_by: str = "created_at",
sort_order: str = "desc",
per_page: int = 25,
page: int = 1
) -> dict:
params = {
"sort_by": sort_by,
"sort_order": sort_order,
"per_page": min(per_page, 100),
"page": page
}
# Build query for search
query_parts = []
if status:
query_parts.append(f"status:{status}")
if priority:
query_parts.append(f"priority:{priority}")
if query_parts:
params["query"] = " ".join(query_parts)
return await self._request("GET", "/search.json", params={"query": " ".join(query_parts), "sort_by": sort_by, "sort_order": sort_order})
return await self._request("GET", "/tickets.json", params=params)
async def get_ticket(self, ticket_id: int) -> dict:
return await self._request("GET", f"/tickets/{ticket_id}.json")
async def get_ticket_comments(self, ticket_id: int) -> dict:
return await self._request("GET", f"/tickets/{ticket_id}/comments.json")
async def create_ticket(
self,
subject: str,
description: str,
priority: Optional[str] = None,
ticket_type: Optional[str] = None,
tags: Optional[list] = None,
requester_email: Optional[str] = None
) -> dict:
ticket_data: dict = {
"subject": subject,
"comment": {"body": description}
}
if priority:
ticket_data["priority"] = priority
if ticket_type:
ticket_data["type"] = ticket_type
if tags:
ticket_data["tags"] = tags
if requester_email:
ticket_data["requester"] = {"email": requester_email}
return await self._request("POST", "/tickets.json", json={"ticket": ticket_data})
async def create_ticket_comment(
self,
ticket_id: int,
body: str,
public: bool = True
) -> dict:
return await self._request(
"PUT",
f"/tickets/{ticket_id}.json",
json={"ticket": {"comment": {"body": body, "public": public}}}
)
async def update_ticket(
self,
ticket_id: int,
status: Optional[str] = None,
priority: Optional[str] = None,
assignee_id: Optional[int] = None,
tags: Optional[list] = None,
add_tags: Optional[list] = None
) -> dict:
ticket_data: dict = {}
if status:
ticket_data["status"] = status
if priority:
ticket_data["priority"] = priority
if assignee_id:
ticket_data["assignee_id"] = assignee_id
if tags is not None:
ticket_data["tags"] = tags
if add_tags:
ticket_data["additional_tags"] = add_tags
return await self._request("PUT", f"/tickets/{ticket_id}.json", json={"ticket": ticket_data})
