clickup_client.py•5.08 kB
import requests
from typing import Dict, List, Optional, Any
from urllib.parse import urljoin
from .utils.auth import AuthManager
from .utils.exceptions import (
ClickUpAPIException,
AuthenticationException,
RateLimitException,
NotFoundException,
PermissionException
)
class ClickUpClient:
"""ClickUp API client for MCP server"""
BASE_URL = "https://api.clickup.com/api/v2/"
BASE_URL_V3 = "https://api.clickup.com/api/v3/"
def __init__(self, api_token: Optional[str] = None):
self.auth = AuthManager(api_token)
self.session = requests.Session()
self.session.headers.update(self.auth.get_headers())
def _make_request(self, method: str, endpoint: str, api_version: str = "v2", **kwargs) -> Dict[str, Any]:
"""Make HTTP request to ClickUp API"""
base_url = self.BASE_URL_V3 if api_version == "v3" else self.BASE_URL
url = urljoin(base_url, endpoint)
try:
response = self.session.request(method, url, **kwargs)
if response.status_code == 401:
raise AuthenticationException("Invalid API token")
elif response.status_code == 403:
raise PermissionException("Insufficient permissions")
elif response.status_code == 404:
raise NotFoundException("Resource not found")
elif response.status_code == 429:
raise RateLimitException("Rate limit exceeded")
elif response.status_code >= 400:
raise ClickUpAPIException(f"API error: {response.status_code} - {response.text}")
return response.json() if response.content else {}
except requests.RequestException as e:
raise ClickUpAPIException(f"Request failed: {str(e)}")
def get_workspaces(self) -> List[Dict[str, Any]]:
"""Get user's workspaces"""
response = self._make_request("GET", "team")
return response.get("teams", [])
def get_workspace_members(self, workspace_id: str) -> List[Dict[str, Any]]:
"""Get workspace members"""
response = self._make_request("GET", f"team/{workspace_id}")
return response.get("team", {}).get("members", [])
def get_spaces(self, workspace_id: str) -> List[Dict[str, Any]]:
"""Get spaces in workspace"""
response = self._make_request("GET", f"team/{workspace_id}/space")
return response.get("spaces", [])
def get_lists(self, space_id: str) -> List[Dict[str, Any]]:
"""Get lists in space"""
response = self._make_request("GET", f"space/{space_id}/list")
return response.get("lists", [])
def get_list(self, list_id: str) -> Dict[str, Any]:
"""Get list details including custom fields"""
return self._make_request("GET", f"list/{list_id}")
def get_custom_fields(self, list_id: str) -> List[Dict[str, Any]]:
"""Get custom fields for a list"""
list_details = self.get_list(list_id)
return list_details.get("custom_fields", [])
def get_tasks(self, list_id: str, **params) -> List[Dict[str, Any]]:
"""Get tasks in list"""
response = self._make_request("GET", f"list/{list_id}/task", params=params)
return response.get("tasks", [])
def get_task(self, task_id: str) -> Dict[str, Any]:
"""Get task by ID"""
return self._make_request("GET", f"task/{task_id}")
def create_task(self, list_id: str, task_data: Dict[str, Any]) -> Dict[str, Any]:
"""Create new task"""
return self._make_request("POST", f"list/{list_id}/task", json=task_data)
def update_task(self, task_id: str, task_data: Dict[str, Any]) -> Dict[str, Any]:
"""Update task"""
return self._make_request("PUT", f"task/{task_id}", json=task_data)
def delete_task(self, task_id: str) -> Dict[str, Any]:
"""Delete task"""
return self._make_request("DELETE", f"task/{task_id}")
def create_list(self, space_id: str, list_data: Dict[str, Any]) -> Dict[str, Any]:
"""Create new list"""
return self._make_request("POST", f"space/{space_id}/list", json=list_data)
def update_list(self, list_id: str, list_data: Dict[str, Any]) -> Dict[str, Any]:
"""Update list"""
return self._make_request("PUT", f"list/{list_id}", json=list_data)
def delete_list(self, list_id: str) -> Dict[str, Any]:
"""Delete list"""
return self._make_request("DELETE", f"list/{list_id}")
def create_workspace_audit_log(self, workspace_id: str, log_data: Dict[str, Any]) -> Dict[str, Any]:
"""Create workspace audit log (Enterprise only)"""
return self._make_request("POST", f"workspaces/{workspace_id}/auditlogs", json=log_data, api_version="v3")
def search_tasks(self, workspace_id: str, query: str, **params) -> List[Dict[str, Any]]:
"""Search tasks in workspace"""
params.update({"query": query})
response = self._make_request("GET", f"team/{workspace_id}/task", params=params)
return response.get("tasks", [])