""" TickTick API integration service for ClaudeTools. This module handles all interactions with the TickTick Open API for project and task management. Tokens are managed via a local JSON file with automatic refresh on 401 responses. API Documentation: https://developer.ticktick.com/api """ import json import logging import subprocess from dataclasses import dataclass from pathlib import Path from typing import Optional import httpx logger = logging.getLogger(__name__) TICKTICK_API_BASE_URL = "https://api.ticktick.com/open/v1" TICKTICK_TOKEN_URL = "https://ticktick.com/oauth/token" TICKTICK_TOKEN_FILE = Path(__file__).resolve().parents[2] / "mcp-servers" / "ticktick" / ".tokens.json" VAULT_SCRIPT = "D:/vault/scripts/vault.sh" VAULT_ENTRY = "services/ticktick.sops.yaml" TICKTICK_TIMEOUT_SECONDS = 30.0 TICKTICK_CONNECT_TIMEOUT_SECONDS = 10.0 @dataclass class TickTickResult: """Result wrapper for all TickTick API operations.""" success: bool data: Optional[dict] = None error: Optional[str] = None def _vault_get_field(field: str) -> str: """Retrieve a single field from the SOPS vault entry.""" try: result = subprocess.run( ["bash", VAULT_SCRIPT, "get-field", VAULT_ENTRY, field], capture_output=True, text=True, timeout=15, ) if result.returncode == 0 and result.stdout.strip(): return result.stdout.strip() logger.error("[ERROR] Vault returned empty or error for %s", field) return "" except (FileNotFoundError, subprocess.TimeoutExpired) as exc: logger.error("[ERROR] Vault retrieval failed for %s: %s", field, exc) return "" class TickTickService: """ Service for interacting with the TickTick Open API. Handles project and task CRUD operations with automatic OAuth token refresh when the access token expires. Credentials are retrieved from the SOPS vault. """ def __init__( self, api_base_url: str = TICKTICK_API_BASE_URL, token_file: Path = TICKTICK_TOKEN_FILE, timeout: float = TICKTICK_TIMEOUT_SECONDS, connect_timeout: float = TICKTICK_CONNECT_TIMEOUT_SECONDS, ): self.api_base_url = api_base_url.rstrip("/") self.token_file = token_file self.timeout = httpx.Timeout(timeout, connect=connect_timeout) self._access_token: Optional[str] = None self._refresh_token: Optional[str] = None self._client_id: Optional[str] = None self._client_secret: Optional[str] = None self._load_tokens() # ------------------------------------------------------------------ # Token management # ------------------------------------------------------------------ def _load_tokens(self) -> None: """Load access and refresh tokens from the local token file.""" if not self.token_file.exists(): logger.warning( "[WARNING] TickTick token file not found at %s", self.token_file ) return try: data = json.loads(self.token_file.read_text(encoding="utf-8")) self._access_token = data.get("access_token") self._refresh_token = data.get("refresh_token") logger.info("[OK] TickTick tokens loaded from %s", self.token_file) except (json.JSONDecodeError, OSError) as exc: logger.error( "[ERROR] Failed to read TickTick token file: %s", exc ) def _save_tokens(self) -> None: """Persist current tokens back to the token file.""" try: existing: dict = {} if self.token_file.exists(): try: existing = json.loads( self.token_file.read_text(encoding="utf-8") ) except (json.JSONDecodeError, OSError): existing = {} existing["access_token"] = self._access_token existing["refresh_token"] = self._refresh_token self.token_file.write_text( json.dumps(existing, indent=2) + "\n", encoding="utf-8" ) logger.info("[OK] TickTick tokens saved to %s", self.token_file) except OSError as exc: logger.error( "[ERROR] Failed to write TickTick token file: %s", exc ) async def _refresh_access_token(self) -> bool: """ Refresh the OAuth access token using the stored refresh token. Returns: True if the token was refreshed successfully, False otherwise. """ if not self._refresh_token: logger.error("[ERROR] No refresh token available for TickTick") return False # Lazy-load vault credentials for refresh if not self._client_id: self._client_id = _vault_get_field("credentials.client_id") if not self._client_secret: self._client_secret = _vault_get_field("credentials.client_secret") if not self._client_id or not self._client_secret: logger.error( "[ERROR] Could not retrieve TickTick client credentials from SOPS vault" ) return False logger.info("[INFO] Refreshing TickTick access token") try: async with httpx.AsyncClient(timeout=self.timeout) as client: response = await client.post( TICKTICK_TOKEN_URL, headers={ "Content-Type": "application/x-www-form-urlencoded", }, data={ "grant_type": "refresh_token", "refresh_token": self._refresh_token, "client_id": self._client_id, "client_secret": self._client_secret, }, ) if response.status_code != 200: logger.error( "[ERROR] TickTick token refresh failed with status %d: %s", response.status_code, response.text, ) return False token_data = response.json() self._access_token = token_data.get("access_token") if "refresh_token" in token_data: self._refresh_token = token_data["refresh_token"] self._save_tokens() logger.info("[OK] TickTick access token refreshed successfully") return True except httpx.HTTPError as exc: logger.error( "[ERROR] TickTick token refresh request failed: %s", exc ) return False # ------------------------------------------------------------------ # HTTP helpers # ------------------------------------------------------------------ def _get_client(self) -> httpx.AsyncClient: """ Create an async HTTP client with configured settings. Returns: Configured httpx.AsyncClient for TickTick API calls. """ headers = { "Content-Type": "application/json", "Accept": "application/json", } if self._access_token: headers["Authorization"] = f"Bearer {self._access_token}" return httpx.AsyncClient(timeout=self.timeout, headers=headers) async def _request( self, method: str, endpoint: str, json_body: Optional[dict] = None, retry_on_401: bool = True, ) -> TickTickResult: """ Execute an API request with automatic 401 retry after token refresh. Args: method: HTTP method (GET, POST, PUT, DELETE). endpoint: API path relative to the base URL (e.g., '/project'). json_body: Optional JSON payload for POST/PUT requests. retry_on_401: Whether to attempt a token refresh on 401. Returns: TickTickResult with success status and response data or error. """ url = f"{self.api_base_url}{endpoint}" try: async with self._get_client() as client: response = await client.request( method, url, json=json_body ) if response.status_code == 401 and retry_on_401: logger.info( "[INFO] TickTick API returned 401, attempting token refresh" ) refreshed = await self._refresh_access_token() if refreshed: return await self._request( method, endpoint, json_body, retry_on_401=False ) return TickTickResult( success=False, error="Authentication failed and token refresh was unsuccessful", ) if response.status_code == 204: return TickTickResult(success=True, data={}) if response.status_code >= 400: error_text = response.text logger.error( "[ERROR] TickTick API %s %s returned %d: %s", method, endpoint, response.status_code, error_text, ) return TickTickResult( success=False, error=f"API returned {response.status_code}: {error_text}", ) # Some responses may have empty bodies (e.g., 200 with no content) if not response.text.strip(): return TickTickResult(success=True, data={}) return TickTickResult(success=True, data=response.json()) except httpx.HTTPError as exc: logger.error( "[ERROR] TickTick API request failed (%s %s): %s", method, endpoint, exc, ) return TickTickResult( success=False, error=f"Request failed: {exc}" ) except json.JSONDecodeError as exc: logger.error( "[ERROR] TickTick API returned invalid JSON (%s %s): %s", method, endpoint, exc, ) return TickTickResult( success=False, error=f"Invalid JSON in response: {exc}" ) # ------------------------------------------------------------------ # Project operations # ------------------------------------------------------------------ async def list_projects(self) -> TickTickResult: """ List all projects (lists) in the TickTick account. Returns: TickTickResult with data containing a list of project dicts. """ logger.info("[INFO] Fetching TickTick project list") result = await self._request("GET", "/project") if result.success and isinstance(result.data, list): result.data = {"projects": result.data} return result async def get_project(self, project_id: str) -> TickTickResult: """ Get a single project with its task data. Args: project_id: The TickTick project ID. Returns: TickTickResult with the project data including tasks. """ if not project_id: return TickTickResult(success=False, error="project_id is required") logger.info("[INFO] Fetching TickTick project %s", project_id) result = await self._request("GET", f"/project/{project_id}/data") return result async def create_project( self, name: str, color: Optional[str] = None, view_mode: Optional[str] = None, kind: Optional[str] = None, ) -> TickTickResult: """ Create a new project (list) in TickTick. Args: name: Project name. color: Optional hex color string (e.g., '#FF6347'). view_mode: Optional view mode ('list', 'kanban', 'timeline'). kind: Optional project kind ('TASK' or 'NOTE'). Returns: TickTickResult with the created project data. """ if not name: return TickTickResult(success=False, error="name is required") body: dict = {"name": name} if color is not None: body["color"] = color if view_mode is not None: body["viewMode"] = view_mode if kind is not None: body["kind"] = kind logger.info("[INFO] Creating TickTick project: %s", name) return await self._request("POST", "/project", json_body=body) async def update_project( self, project_id: str, name: Optional[str] = None, color: Optional[str] = None, view_mode: Optional[str] = None, ) -> TickTickResult: """ Update an existing project in TickTick. Args: project_id: The TickTick project ID to update. name: Optional new project name. color: Optional new hex color string. view_mode: Optional new view mode. Returns: TickTickResult with the updated project data. """ if not project_id: return TickTickResult(success=False, error="project_id is required") body: dict = {} if name is not None: body["name"] = name if color is not None: body["color"] = color if view_mode is not None: body["viewMode"] = view_mode if not body: return TickTickResult( success=False, error="At least one field to update is required" ) logger.info("[INFO] Updating TickTick project %s", project_id) return await self._request( "POST", f"/project/{project_id}", json_body=body ) async def delete_project(self, project_id: str) -> TickTickResult: """ Delete a project from TickTick. Args: project_id: The TickTick project ID to delete. Returns: TickTickResult with success status. """ if not project_id: return TickTickResult(success=False, error="project_id is required") logger.info("[INFO] Deleting TickTick project %s", project_id) return await self._request("DELETE", f"/project/{project_id}") # ------------------------------------------------------------------ # Task operations # ------------------------------------------------------------------ async def create_task( self, title: str, project_id: str, content: Optional[str] = None, priority: Optional[int] = None, due_date: Optional[str] = None, tags: Optional[list[str]] = None, ) -> TickTickResult: """ Create a new task in a TickTick project. Args: title: Task title. project_id: ID of the project to create the task in. content: Optional task description/content. priority: Optional priority (0=none, 1=low, 3=medium, 5=high). due_date: Optional due date in ISO 8601 format. tags: Optional list of tag strings. Returns: TickTickResult with the created task data. """ if not title: return TickTickResult(success=False, error="title is required") if not project_id: return TickTickResult(success=False, error="project_id is required") body: dict = {"title": title, "projectId": project_id} if content is not None: body["content"] = content if priority is not None: body["priority"] = priority if due_date is not None: body["dueDate"] = due_date if tags is not None: body["tags"] = tags logger.info( "[INFO] Creating TickTick task '%s' in project %s", title, project_id, ) return await self._request("POST", "/task", json_body=body) async def update_task( self, task_id: str, project_id: str, title: Optional[str] = None, content: Optional[str] = None, priority: Optional[int] = None, due_date: Optional[str] = None, tags: Optional[list[str]] = None, ) -> TickTickResult: """ Update an existing task in TickTick. Args: task_id: The task ID to update. project_id: The project ID containing the task. title: Optional new task title. content: Optional new task content. priority: Optional new priority level. due_date: Optional new due date in ISO 8601 format. tags: Optional new list of tags. Returns: TickTickResult with the updated task data. """ if not task_id: return TickTickResult(success=False, error="task_id is required") if not project_id: return TickTickResult(success=False, error="project_id is required") body: dict = {"id": task_id, "projectId": project_id} if title is not None: body["title"] = title if content is not None: body["content"] = content if priority is not None: body["priority"] = priority if due_date is not None: body["dueDate"] = due_date if tags is not None: body["tags"] = tags logger.info( "[INFO] Updating TickTick task %s in project %s", task_id, project_id, ) return await self._request( "POST", f"/task/{task_id}", json_body=body ) async def complete_task( self, task_id: str, project_id: str ) -> TickTickResult: """ Mark a task as complete in TickTick. Args: task_id: The task ID to complete. project_id: The project ID containing the task. Returns: TickTickResult with success status. """ if not task_id: return TickTickResult(success=False, error="task_id is required") if not project_id: return TickTickResult(success=False, error="project_id is required") logger.info( "[INFO] Completing TickTick task %s in project %s", task_id, project_id, ) return await self._request( "POST", f"/project/{project_id}/task/{task_id}/complete" ) async def delete_task( self, task_id: str, project_id: str ) -> TickTickResult: """ Delete a task from TickTick. Args: task_id: The task ID to delete. project_id: The project ID containing the task. Returns: TickTickResult with success status. """ if not task_id: return TickTickResult(success=False, error="task_id is required") if not project_id: return TickTickResult(success=False, error="project_id is required") logger.info( "[INFO] Deleting TickTick task %s from project %s", task_id, project_id, ) return await self._request( "DELETE", f"/project/{project_id}/task/{task_id}" ) # -------------------------------------------------------------------------- # Singleton accessor # -------------------------------------------------------------------------- _ticktick_service: Optional[TickTickService] = None def get_ticktick_service() -> TickTickService: """ Return a singleton TickTickService instance. Creates the service on first call, reuses it thereafter. Returns: The shared TickTickService instance. """ global _ticktick_service if _ticktick_service is None: _ticktick_service = TickTickService() return _ticktick_service