ticktick-mcp-server
by jacepark12
- ticktick_mcp
- src
import os
import json
import base64
import requests
import logging
from pathlib import Path
from dotenv import load_dotenv
from typing import Dict, List, Any, Optional, Tuple
# Set up logging
logger = logging.getLogger(__name__)
class TickTickClient:
"""
Client for the TickTick API using OAuth2 authentication.
"""
def __init__(self):
load_dotenv()
self.client_id = os.getenv("TICKTICK_CLIENT_ID")
self.client_secret = os.getenv("TICKTICK_CLIENT_SECRET")
self.access_token = os.getenv("TICKTICK_ACCESS_TOKEN")
self.refresh_token = os.getenv("TICKTICK_REFRESH_TOKEN")
if not self.access_token:
raise ValueError("TICKTICK_ACCESS_TOKEN environment variable is not set. "
"Please run 'uv run -m ticktick_mcp.authenticate' to set up your credentials.")
self.base_url = "https://api.ticktick.com/open/v1"
self.token_url = "https://ticktick.com/oauth/token"
self.headers = {
"Authorization": f"Bearer {self.access_token}",
"Content-Type": "application/json"
}
def _refresh_access_token(self) -> bool:
"""
Refresh the access token using the refresh token.
Returns:
True if successful, False otherwise
"""
if not self.refresh_token:
logger.warning("No refresh token available. Cannot refresh access token.")
return False
if not self.client_id or not self.client_secret:
logger.warning("Client ID or Client Secret missing. Cannot refresh access token.")
return False
# Prepare the token request
token_data = {
"grant_type": "refresh_token",
"refresh_token": self.refresh_token
}
# Prepare Basic Auth credentials
auth_str = f"{self.client_id}:{self.client_secret}"
auth_bytes = auth_str.encode('ascii')
auth_b64 = base64.b64encode(auth_bytes).decode('ascii')
headers = {
"Authorization": f"Basic {auth_b64}",
"Content-Type": "application/x-www-form-urlencoded"
}
try:
# Send the token request
response = requests.post(self.token_url, data=token_data, headers=headers)
response.raise_for_status()
# Parse the response
tokens = response.json()
# Update the tokens
self.access_token = tokens.get('access_token')
if 'refresh_token' in tokens:
self.refresh_token = tokens.get('refresh_token')
# Update the headers
self.headers["Authorization"] = f"Bearer {self.access_token}"
# Save the tokens to the .env file
self._save_tokens_to_env(tokens)
logger.info("Access token refreshed successfully.")
return True
except requests.exceptions.RequestException as e:
logger.error(f"Error refreshing access token: {e}")
return False
def _save_tokens_to_env(self, tokens: Dict[str, str]) -> None:
"""
Save the tokens to the .env file.
Args:
tokens: A dictionary containing the access_token and optionally refresh_token
"""
# Load existing .env file content
env_path = Path('.env')
env_content = {}
if env_path.exists():
with open(env_path, 'r') as f:
for line in f:
line = line.strip()
if line and not line.startswith('#') and '=' in line:
key, value = line.split('=', 1)
env_content[key] = value
# Update with new tokens
env_content["TICKTICK_ACCESS_TOKEN"] = tokens.get('access_token', '')
if 'refresh_token' in tokens:
env_content["TICKTICK_REFRESH_TOKEN"] = tokens.get('refresh_token', '')
# Make sure client credentials are saved as well
if self.client_id and "TICKTICK_CLIENT_ID" not in env_content:
env_content["TICKTICK_CLIENT_ID"] = self.client_id
if self.client_secret and "TICKTICK_CLIENT_SECRET" not in env_content:
env_content["TICKTICK_CLIENT_SECRET"] = self.client_secret
# Write back to .env file
with open(env_path, 'w') as f:
for key, value in env_content.items():
f.write(f"{key}={value}\n")
logger.debug("Tokens saved to .env file")
def _make_request(self, method: str, endpoint: str, data=None) -> Dict:
"""
Makes a request to the TickTick API.
Args:
method: HTTP method (GET, POST, PUT, DELETE)
endpoint: API endpoint (without base URL)
data: Request data (for POST, PUT)
Returns:
API response as a dictionary
"""
url = f"{self.base_url}{endpoint}"
try:
# Make the request
if method == "GET":
response = requests.get(url, headers=self.headers)
elif method == "POST":
response = requests.post(url, headers=self.headers, json=data)
elif method == "DELETE":
response = requests.delete(url, headers=self.headers)
else:
raise ValueError(f"Unsupported HTTP method: {method}")
# Check if the request was unauthorized (401)
if response.status_code == 401:
logger.info("Access token expired. Attempting to refresh...")
# Try to refresh the access token
if self._refresh_access_token():
# Retry the request with the new token
if method == "GET":
response = requests.get(url, headers=self.headers)
elif method == "POST":
response = requests.post(url, headers=self.headers, json=data)
elif method == "DELETE":
response = requests.delete(url, headers=self.headers)
# Raise an exception for 4xx/5xx status codes
response.raise_for_status()
# Return empty dict for 204 No Content
if response.status_code == 204:
return {}
return response.json()
except requests.exceptions.RequestException as e:
logger.error(f"API request failed: {e}")
return {"error": str(e)}
# Project methods
def get_projects(self) -> List[Dict]:
"""Gets all projects for the user."""
return self._make_request("GET", "/project")
def get_project(self, project_id: str) -> Dict:
"""Gets a specific project by ID."""
return self._make_request("GET", f"/project/{project_id}")
def get_project_with_data(self, project_id: str) -> Dict:
"""Gets project with tasks and columns."""
return self._make_request("GET", f"/project/{project_id}/data")
def create_project(self, name: str, color: str = "#F18181", view_mode: str = "list", kind: str = "TASK") -> Dict:
"""Creates a new project."""
data = {
"name": name,
"color": color,
"viewMode": view_mode,
"kind": kind
}
return self._make_request("POST", "/project", data)
def update_project(self, project_id: str, name: str = None, color: str = None,
view_mode: str = None, kind: str = None) -> Dict:
"""Updates an existing project."""
data = {}
if name:
data["name"] = name
if color:
data["color"] = color
if view_mode:
data["viewMode"] = view_mode
if kind:
data["kind"] = kind
return self._make_request("POST", f"/project/{project_id}", data)
def delete_project(self, project_id: str) -> Dict:
"""Deletes a project."""
return self._make_request("DELETE", f"/project/{project_id}")
# Task methods
def get_task(self, project_id: str, task_id: str) -> Dict:
"""Gets a specific task by project ID and task ID."""
return self._make_request("GET", f"/project/{project_id}/task/{task_id}")
def create_task(self, title: str, project_id: str, content: str = None,
start_date: str = None, due_date: str = None,
priority: int = 0, is_all_day: bool = False) -> Dict:
"""Creates a new task."""
data = {
"title": title,
"projectId": project_id
}
if content:
data["content"] = content
if start_date:
data["startDate"] = start_date
if due_date:
data["dueDate"] = due_date
if priority is not None:
data["priority"] = priority
if is_all_day is not None:
data["isAllDay"] = is_all_day
return self._make_request("POST", "/task", data)
def update_task(self, task_id: str, project_id: str, title: str = None,
content: str = None, priority: int = None,
start_date: str = None, due_date: str = None) -> Dict:
"""Updates an existing task."""
data = {
"id": task_id,
"projectId": project_id
}
if title:
data["title"] = title
if content:
data["content"] = content
if priority is not None:
data["priority"] = priority
if start_date:
data["startDate"] = start_date
if due_date:
data["dueDate"] = due_date
return self._make_request("POST", f"/task/{task_id}", data)
def complete_task(self, project_id: str, task_id: str) -> Dict:
"""Marks a task as complete."""
return self._make_request("POST", f"/project/{project_id}/task/{task_id}/complete")
def delete_task(self, project_id: str, task_id: str) -> Dict:
"""Deletes a task."""
return self._make_request("DELETE", f"/project/{project_id}/task/{task_id}")