import os import json import re from typing import Optional, List, Dict, Any from phi.tools import Toolkit from phi.utils.log import logger try: import requests except ImportError: raise ImportError("`requests` not installed. Please install using `pip install requests`") class ClickUpTools(Toolkit): def __init__( self, api_key: Optional[str] = None, master_space_id: Optional[str] = None, list_tasks: bool = True, create_task: bool = True, get_task: bool = True, update_task: bool = True, delete_task: bool = True, list_spaces: bool = True, list_lists: bool = True, ): super().__init__(name="clickup") self.api_key = api_key or os.getenv("CLICKUP_API_KEY") self.master_space_id = master_space_id or os.getenv("MASTER_SPACE_ID") self.base_url = "https://api.clickup.com/api/v2" self.headers = {"Authorization": self.api_key} if not self.api_key: raise ValueError("CLICKUP_API_KEY not set. Please set the CLICKUP_API_KEY environment variable.") if not self.master_space_id: raise ValueError("MASTER_SPACE_ID not set. Please set the MASTER_SPACE_ID environment variable.") if list_tasks: self.register(self.list_tasks) if create_task: self.register(self.create_task) if get_task: self.register(self.get_task) if update_task: self.register(self.update_task) if delete_task: self.register(self.delete_task) if list_spaces: self.register(self.list_spaces) if list_lists: self.register(self.list_lists) def _make_request( self, method: str, endpoint: str, params: Optional[Dict] = None, data: Optional[Dict] = None ) -> Dict[str, Any]: """Make a request to the ClickUp API.""" url = f"{self.base_url}/{endpoint}" try: response = requests.request(method=method, url=url, headers=self.headers, params=params, json=data) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: logger.error(f"Error making request to {url}: {e}") return {"error": str(e)} def _find_by_name(self, items: List[Dict[str, Any]], name: str) -> Optional[Dict[str, Any]]: """Find an item in a list by name using exact match or regex pattern. Args: items: List of items to search through name: Name to search for Returns: Matching item or None if not found """ if not name: return items[0] if items else None pattern = re.compile(name, re.IGNORECASE) for item in items: # Try exact match first (case-insensitive) if item["name"].lower() == name.lower(): return item # Then try regex pattern match if pattern.search(item["name"]): return item return None def _get_space(self, space_name: str) -> Dict[str, Any]: """Get space information by name.""" spaces = self._make_request("GET", f"team/{self.master_space_id}/space") if "error" in spaces: return spaces spaces_list = spaces.get("spaces", []) if not spaces_list: return {"error": "No spaces found"} space = self._find_by_name(spaces_list, space_name) if not space: return {"error": f"Space '{space_name}' not found"} return space def _get_list(self, space_id: str, list_name: str) -> Dict[str, Any]: """Get list information by name.""" lists = self._make_request("GET", f"space/{space_id}/list") if "error" in lists: return lists lists_data = lists.get("lists", []) if not lists_data: return {"error": "No lists found in space"} list_item = self._find_by_name(lists_data, list_name) if not list_item: return {"error": f"List '{list_name}' not found"} return list_item def _get_tasks(self, list_id: str) -> List[Dict[str, Any]]: """Get tasks in a list, optionally filtered by name.""" tasks = self._make_request("GET", f"list/{list_id}/task") if "error" in tasks: return [] tasks_data = tasks.get("tasks", []) return tasks_data def list_tasks(self, space_name: str) -> str: """List all tasks in a space. Args: space_name (str): Name of the space to list tasks from Returns: str: JSON string containing tasks """ # Get space space = self._get_space(space_name) if "error" in space: return json.dumps(space, indent=2) # Get lists lists = self._make_request("GET", f"space/{space['id']}/list") lists_data = lists.get("lists", []) if not lists_data: return json.dumps({"error": f"No lists found in space '{space_name}'"}, indent=2) # Get tasks from all lists all_tasks = [] for list_info in lists_data: tasks = self._get_tasks(list_info["id"]) for task in tasks: task["list_name"] = list_info["name"] # Add list name for context all_tasks.extend(tasks) return json.dumps({"tasks": all_tasks}, indent=2) def create_task(self, space_name: str, task_name: str, task_description: str) -> str: """Create a new task in a space. Args: space_name (str): Name of the space to create task in task_name (str): Name of the task task_description (str): Description of the task Returns: str: JSON string containing created task details """ # Get space space = self._get_space(space_name) if "error" in space: return json.dumps(space, indent=2) # Get first list in space response = self._make_request("GET", f"space/{space['id']}/list") logger.debug(f"Lists: {response}") lists_data = response.get("lists", []) if not lists_data: return json.dumps({"error": f"No lists found in space '{space_name}'"}, indent=2) list_info = lists_data[0] # Use first list # Create task data = {"name": task_name, "description": task_description} task = self._make_request("POST", f"list/{list_info['id']}/task", data=data) return json.dumps(task, indent=2) def list_spaces(self) -> str: """List all spaces in the workspace. Returns: str: JSON string containing list of spaces """ spaces = self._make_request("GET", f"team/{self.master_space_id}/space") return json.dumps(spaces, indent=2) def list_lists(self, space_name: str) -> str: """List all lists in a space. Args: space_name (str): Name of the space to list lists from Returns: str: JSON string containing list of lists """ # Get space space = self._get_space(space_name) if "error" in space: return json.dumps(space, indent=2) # Get lists lists = self._make_request("GET", f"space/{space['id']}/list") return json.dumps(lists, indent=2) def get_task(self, task_id: str) -> str: """Get details of a specific task. Args: task_id (str): The ID of the task Returns: str: JSON string containing task details """ task = self._make_request("GET", f"task/{task_id}") return json.dumps(task, indent=2) def update_task(self, task_id: str, **kwargs) -> str: """Update a specific task. Args: task_id (str): The ID of the task **kwargs: Task fields to update (name, description, status, priority, etc.) Returns: str: JSON string containing updated task details """ task = self._make_request("PUT", f"task/{task_id}", data=kwargs) return json.dumps(task, indent=2) def delete_task(self, task_id: str) -> str: """Delete a specific task. Args: task_id (str): The ID of the task Returns: str: JSON string containing deletion status """ result = self._make_request("DELETE", f"task/{task_id}") if "error" not in result: result = {"success": True, "message": f"Task {task_id} deleted successfully"} return json.dumps(result, indent=2)