Add TickTick integration, MCP server, and dev project tracking
New integration with TickTick API for project/task management: - OAuth 2.0 auth flow (mcp-servers/ticktick/ticktick_auth.py) - MCP server with 9 tools for Claude Code (ticktick_mcp.py) - FastAPI service with SOPS vault credentials (api/services/ticktick_service.py) - JWT-protected REST router at /api/ticktick/ (api/routers/ticktick.py) - Credentials stored in SOPS vault (services/ticktick.sops.yaml) Dev project tracking (hybrid TickTick + DB): - New dev_projects table migration (14 columns, status index) - TickTick "Dev Projects" list for mobile visibility - First project seeded: TickTick Integration (linked both sides) Security: .tokens.json gitignored, token file permissions restricted, HTML-escaped OAuth callback, SOPS vault (not env vars) for secrets. Also: Installed Tailscale on ACG-5070 for office network access. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
313
mcp-servers/ticktick/ticktick_auth.py
Normal file
313
mcp-servers/ticktick/ticktick_auth.py
Normal file
@@ -0,0 +1,313 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
TickTick OAuth 2.0 Authentication Script
|
||||
|
||||
Performs the one-time OAuth flow to obtain access and refresh tokens from TickTick.
|
||||
Reads client credentials from the SOPS vault, opens a browser for user authorization,
|
||||
captures the callback on a local HTTP server, exchanges the code for tokens, and
|
||||
saves them to an encrypted local file.
|
||||
|
||||
Usage:
|
||||
python ticktick_auth.py
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import secrets
|
||||
import stat
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
import webbrowser
|
||||
from datetime import datetime, timezone
|
||||
from http.server import HTTPServer, BaseHTTPRequestHandler
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlencode, urlparse, parse_qs
|
||||
from urllib.request import Request, urlopen
|
||||
from html import escape as html_escape
|
||||
from urllib.error import URLError, HTTPError
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Configuration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
VAULT_SCRIPT = "D:/vault/scripts/vault.sh"
|
||||
VAULT_ENTRY = "services/ticktick.sops.yaml"
|
||||
|
||||
AUTH_URL = "https://ticktick.com/oauth/authorize"
|
||||
TOKEN_URL = "https://ticktick.com/oauth/token"
|
||||
REDIRECT_URI = "http://localhost:9876/callback"
|
||||
SCOPES = "tasks:read tasks:write"
|
||||
CALLBACK_PORT = 9876
|
||||
CALLBACK_TIMEOUT_SECONDS = 60
|
||||
|
||||
TOKEN_FILE = Path(__file__).resolve().parent / ".tokens.json"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Vault credential retrieval
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
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,
|
||||
)
|
||||
except FileNotFoundError:
|
||||
print(f"[ERROR] Could not find bash or vault script at {VAULT_SCRIPT}")
|
||||
sys.exit(1)
|
||||
except subprocess.TimeoutExpired:
|
||||
print(f"[ERROR] Vault command timed out while retrieving {field}")
|
||||
sys.exit(1)
|
||||
|
||||
if result.returncode != 0:
|
||||
stderr = result.stderr.strip()
|
||||
print(f"[ERROR] Vault returned non-zero exit code for field '{field}'")
|
||||
if stderr:
|
||||
print(f" {stderr}")
|
||||
sys.exit(1)
|
||||
|
||||
value = result.stdout.strip()
|
||||
if not value:
|
||||
print(f"[ERROR] Vault returned empty value for field '{field}'")
|
||||
sys.exit(1)
|
||||
|
||||
return value
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Callback HTTP server
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class _CallbackState:
|
||||
"""Shared mutable state between the HTTP handler and the main thread."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.authorization_code: str | None = None
|
||||
self.error: str | None = None
|
||||
self.received = threading.Event()
|
||||
|
||||
|
||||
class _CallbackHandler(BaseHTTPRequestHandler):
|
||||
"""Handles the OAuth redirect callback from TickTick."""
|
||||
|
||||
state: _CallbackState # set on the class before the server starts
|
||||
expected_csrf: str # set on the class before the server starts
|
||||
|
||||
def do_GET(self) -> None: # noqa: N802 – required method name
|
||||
parsed = urlparse(self.path)
|
||||
if parsed.path != "/callback":
|
||||
self._respond(404, "Not found")
|
||||
return
|
||||
|
||||
params = parse_qs(parsed.query)
|
||||
|
||||
# Check for error response from provider
|
||||
if "error" in params:
|
||||
error_msg = params["error"][0]
|
||||
description = params.get("error_description", [""])[0]
|
||||
self.state.error = f"{error_msg}: {description}" if description else error_msg
|
||||
self._respond(400, f"Authorization failed: {self.state.error}")
|
||||
self.state.received.set()
|
||||
return
|
||||
|
||||
# Validate CSRF state parameter
|
||||
returned_state = params.get("state", [None])[0]
|
||||
if returned_state != self.expected_csrf:
|
||||
self.state.error = "CSRF state mismatch -- possible request forgery"
|
||||
self._respond(400, self.state.error)
|
||||
self.state.received.set()
|
||||
return
|
||||
|
||||
# Extract authorization code
|
||||
code = params.get("code", [None])[0]
|
||||
if not code:
|
||||
self.state.error = "No authorization code in callback"
|
||||
self._respond(400, self.state.error)
|
||||
self.state.received.set()
|
||||
return
|
||||
|
||||
self.state.authorization_code = code
|
||||
self._respond(
|
||||
200,
|
||||
"Authorization successful! You can close this tab and return to the terminal.",
|
||||
)
|
||||
self.state.received.set()
|
||||
|
||||
def _respond(self, status: int, body: str) -> None:
|
||||
self.send_response(status)
|
||||
self.send_header("Content-Type", "text/html; charset=utf-8")
|
||||
self.end_headers()
|
||||
html = (
|
||||
"<!DOCTYPE html><html><head><title>TickTick Auth</title></head>"
|
||||
f"<body><h2>{html_escape(body)}</h2></body></html>"
|
||||
)
|
||||
self.wfile.write(html.encode("utf-8"))
|
||||
|
||||
# Silence default request logging
|
||||
def log_message(self, format: str, *args: object) -> None: # noqa: A002
|
||||
pass
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Token exchange
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def exchange_code_for_tokens(
|
||||
code: str,
|
||||
client_id: str,
|
||||
client_secret: str,
|
||||
) -> dict:
|
||||
"""Exchange an authorization code for access and refresh tokens."""
|
||||
body = urlencode({
|
||||
"grant_type": "authorization_code",
|
||||
"code": code,
|
||||
"redirect_uri": REDIRECT_URI,
|
||||
"client_id": client_id,
|
||||
"client_secret": client_secret,
|
||||
"scope": SCOPES,
|
||||
}).encode("utf-8")
|
||||
|
||||
request = Request(
|
||||
TOKEN_URL,
|
||||
data=body,
|
||||
headers={
|
||||
"Content-Type": "application/x-www-form-urlencoded",
|
||||
"Accept": "application/json",
|
||||
},
|
||||
method="POST",
|
||||
)
|
||||
|
||||
try:
|
||||
with urlopen(request, timeout=15) as response:
|
||||
data = json.loads(response.read().decode("utf-8"))
|
||||
except HTTPError as exc:
|
||||
error_body = exc.read().decode("utf-8", errors="replace")
|
||||
print(f"[ERROR] Token exchange failed (HTTP {exc.code})")
|
||||
print(f" Response: {error_body}")
|
||||
sys.exit(1)
|
||||
except URLError as exc:
|
||||
print(f"[ERROR] Could not reach token endpoint: {exc.reason}")
|
||||
sys.exit(1)
|
||||
except json.JSONDecodeError:
|
||||
print("[ERROR] Token endpoint returned invalid JSON")
|
||||
sys.exit(1)
|
||||
|
||||
if "access_token" not in data:
|
||||
print("[ERROR] Token response missing 'access_token'")
|
||||
print(f" Full response: {json.dumps(data, indent=2)}")
|
||||
sys.exit(1)
|
||||
|
||||
return data
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Token persistence
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def save_tokens(token_data: dict) -> None:
|
||||
"""Persist tokens to a local JSON file."""
|
||||
payload = {
|
||||
"access_token": token_data["access_token"],
|
||||
"refresh_token": token_data.get("refresh_token", ""),
|
||||
"token_type": token_data.get("token_type", "bearer"),
|
||||
"obtained_at": datetime.now(timezone.utc).isoformat(),
|
||||
}
|
||||
|
||||
TOKEN_FILE.parent.mkdir(parents=True, exist_ok=True)
|
||||
TOKEN_FILE.write_text(json.dumps(payload, indent=2), encoding="utf-8")
|
||||
# Restrict file permissions (owner read/write only)
|
||||
try:
|
||||
TOKEN_FILE.chmod(stat.S_IRUSR | stat.S_IWUSR)
|
||||
except OSError:
|
||||
pass # Windows may not support POSIX permissions
|
||||
print(f"[OK] Tokens saved to {TOKEN_FILE}")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Main flow
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def main() -> None:
|
||||
print("[INFO] TickTick OAuth 2.0 Authentication")
|
||||
print("=" * 50)
|
||||
|
||||
# -- 1. Read credentials from SOPS vault ----------------------------------
|
||||
print("[INFO] Reading credentials from SOPS vault ...")
|
||||
client_id = vault_get_field("credentials.client_id")
|
||||
client_secret = vault_get_field("credentials.client_secret")
|
||||
print(f"[OK] Client ID retrieved (ends ...{client_id[-4:]})")
|
||||
|
||||
# -- 2. Prepare CSRF state and authorization URL --------------------------
|
||||
csrf_state = secrets.token_urlsafe(32)
|
||||
|
||||
auth_params = urlencode({
|
||||
"client_id": client_id,
|
||||
"redirect_uri": REDIRECT_URI,
|
||||
"response_type": "code",
|
||||
"scope": SCOPES,
|
||||
"state": csrf_state,
|
||||
})
|
||||
full_auth_url = f"{AUTH_URL}?{auth_params}"
|
||||
|
||||
# -- 3. Start local callback server ---------------------------------------
|
||||
callback_state = _CallbackState()
|
||||
_CallbackHandler.state = callback_state
|
||||
_CallbackHandler.expected_csrf = csrf_state
|
||||
|
||||
try:
|
||||
server = HTTPServer(("127.0.0.1", CALLBACK_PORT), _CallbackHandler)
|
||||
except OSError as exc:
|
||||
print(f"[ERROR] Could not start callback server on port {CALLBACK_PORT}: {exc}")
|
||||
sys.exit(1)
|
||||
|
||||
server_thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||||
server_thread.start()
|
||||
print(f"[OK] Callback server listening on http://127.0.0.1:{CALLBACK_PORT}/callback")
|
||||
|
||||
# -- 4. Open browser for authorization ------------------------------------
|
||||
print("[INFO] Opening browser for TickTick authorization ...")
|
||||
print(f"[INFO] If the browser does not open, visit this URL manually:")
|
||||
print(f" {full_auth_url}")
|
||||
webbrowser.open(full_auth_url)
|
||||
|
||||
# -- 5. Wait for callback -------------------------------------------------
|
||||
print(f"[INFO] Waiting up to {CALLBACK_TIMEOUT_SECONDS}s for authorization callback ...")
|
||||
received = callback_state.received.wait(timeout=CALLBACK_TIMEOUT_SECONDS)
|
||||
server.shutdown()
|
||||
|
||||
if not received:
|
||||
print(f"[ERROR] Timed out after {CALLBACK_TIMEOUT_SECONDS}s waiting for callback")
|
||||
print(" Make sure you completed the authorization in your browser.")
|
||||
sys.exit(1)
|
||||
|
||||
if callback_state.error:
|
||||
print(f"[ERROR] Authorization failed: {callback_state.error}")
|
||||
sys.exit(1)
|
||||
|
||||
code = callback_state.authorization_code
|
||||
if not code:
|
||||
print("[ERROR] No authorization code received (unknown error)")
|
||||
sys.exit(1)
|
||||
|
||||
print("[OK] Authorization code received")
|
||||
|
||||
# -- 6. Exchange code for tokens ------------------------------------------
|
||||
print("[INFO] Exchanging authorization code for tokens ...")
|
||||
token_data = exchange_code_for_tokens(code, client_id, client_secret)
|
||||
print("[OK] Token exchange successful")
|
||||
|
||||
# -- 7. Save tokens -------------------------------------------------------
|
||||
save_tokens(token_data)
|
||||
|
||||
print("=" * 50)
|
||||
print("[OK] TickTick authentication complete")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
595
mcp-servers/ticktick/ticktick_mcp.py
Normal file
595
mcp-servers/ticktick/ticktick_mcp.py
Normal file
@@ -0,0 +1,595 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
TickTick MCP Server
|
||||
Provides Claude Code with direct tools to manage TickTick projects and tasks.
|
||||
|
||||
Requires:
|
||||
- pip install mcp httpx
|
||||
- Token file at .tokens.json (run ticktick_auth.py first)
|
||||
- Vault credentials at services/ticktick.sops.yaml
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional
|
||||
|
||||
import httpx
|
||||
|
||||
try:
|
||||
from mcp.server import Server
|
||||
from mcp.types import Tool, TextContent
|
||||
except ImportError:
|
||||
print("[ERROR] MCP package not installed. Run: pip install mcp", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Configuration
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
TICKTICK_API_BASE = "https://api.ticktick.com/open/v1"
|
||||
TICKTICK_OAUTH_TOKEN_URL = "https://ticktick.com/oauth/token"
|
||||
SCRIPT_DIR = Path(__file__).parent
|
||||
TOKENS_PATH = SCRIPT_DIR / ".tokens.json"
|
||||
VAULT_SCRIPT = "D:/vault/scripts/vault.sh"
|
||||
VAULT_ENTRY = "services/ticktick.sops.yaml"
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Credential & token helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_vault_cache: dict[str, str] = {}
|
||||
|
||||
|
||||
def _vault_get_field(field: str) -> str:
|
||||
"""Retrieve a field from the SOPS vault, caching results in memory."""
|
||||
if field in _vault_cache:
|
||||
return _vault_cache[field]
|
||||
result = subprocess.run(
|
||||
["bash", VAULT_SCRIPT, "get-field", VAULT_ENTRY, field],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=15,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
raise RuntimeError(
|
||||
f"[ERROR] Vault lookup failed for {field}: {result.stderr.strip()}"
|
||||
)
|
||||
value = result.stdout.strip()
|
||||
_vault_cache[field] = value
|
||||
return value
|
||||
|
||||
|
||||
def _load_tokens() -> dict[str, str]:
|
||||
"""Load tokens from disk. Raises FileNotFoundError if missing."""
|
||||
if not TOKENS_PATH.exists():
|
||||
raise FileNotFoundError(
|
||||
f"[ERROR] Token file not found at {TOKENS_PATH}. "
|
||||
"Run 'python ticktick_auth.py' in the ticktick directory first "
|
||||
"to complete the OAuth flow and generate .tokens.json."
|
||||
)
|
||||
with open(TOKENS_PATH, "r", encoding="utf-8") as fh:
|
||||
return json.load(fh)
|
||||
|
||||
|
||||
def _save_tokens(tokens: dict[str, str]) -> None:
|
||||
"""Persist tokens to disk."""
|
||||
with open(TOKENS_PATH, "w", encoding="utf-8") as fh:
|
||||
json.dump(tokens, fh, indent=2)
|
||||
|
||||
|
||||
async def _refresh_access_token(refresh_token: str) -> dict[str, str]:
|
||||
"""Exchange a refresh token for a new access token."""
|
||||
client_id = _vault_get_field("credentials.client_id")
|
||||
client_secret = _vault_get_field("credentials.client_secret")
|
||||
|
||||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||||
resp = await client.post(
|
||||
TICKTICK_OAUTH_TOKEN_URL,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
data={
|
||||
"client_id": client_id,
|
||||
"client_secret": client_secret,
|
||||
"refresh_token": refresh_token,
|
||||
"grant_type": "refresh_token",
|
||||
},
|
||||
)
|
||||
if resp.status_code != 200:
|
||||
raise RuntimeError(
|
||||
f"[ERROR] Token refresh failed ({resp.status_code}): {resp.text}"
|
||||
)
|
||||
new_data = resp.json()
|
||||
|
||||
tokens = {
|
||||
"access_token": new_data["access_token"],
|
||||
"refresh_token": new_data.get("refresh_token", refresh_token),
|
||||
"token_type": new_data.get("token_type", "bearer"),
|
||||
"obtained_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
||||
}
|
||||
_save_tokens(tokens)
|
||||
return tokens
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# HTTP helper with automatic 401 retry
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def _ticktick_request(
|
||||
method: str,
|
||||
path: str,
|
||||
*,
|
||||
json_body: Optional[dict] = None,
|
||||
) -> httpx.Response:
|
||||
"""
|
||||
Make an authenticated request to the TickTick API.
|
||||
|
||||
On a 401 response, automatically refreshes the access token and retries
|
||||
the request exactly once.
|
||||
"""
|
||||
tokens = _load_tokens()
|
||||
|
||||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||||
url = f"{TICKTICK_API_BASE}{path}"
|
||||
headers = {"Authorization": f"Bearer {tokens['access_token']}"}
|
||||
kwargs: dict[str, Any] = {"headers": headers}
|
||||
if json_body is not None:
|
||||
kwargs["json"] = json_body
|
||||
|
||||
resp = await client.request(method, url, **kwargs)
|
||||
|
||||
if resp.status_code == 401:
|
||||
# Attempt token refresh and retry once
|
||||
tokens = await _refresh_access_token(tokens["refresh_token"])
|
||||
headers = {"Authorization": f"Bearer {tokens['access_token']}"}
|
||||
kwargs["headers"] = headers
|
||||
resp = await client.request(method, url, **kwargs)
|
||||
|
||||
return resp
|
||||
|
||||
|
||||
def _format_response(data: Any) -> str:
|
||||
"""Serialize a response payload to pretty JSON text."""
|
||||
if isinstance(data, (dict, list)):
|
||||
return json.dumps(data, indent=2, ensure_ascii=False)
|
||||
return str(data)
|
||||
|
||||
|
||||
def _error_text(msg: str) -> list[TextContent]:
|
||||
return [TextContent(type="text", text=msg)]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# MCP Server
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
app = Server("ticktick")
|
||||
|
||||
|
||||
@app.list_tools()
|
||||
async def list_tools() -> list[Tool]:
|
||||
"""Enumerate all TickTick tools."""
|
||||
return [
|
||||
# ----- Projects -----
|
||||
Tool(
|
||||
name="ticktick_list_projects",
|
||||
description="List all TickTick projects. Returns an array of projects with id, name, color, viewMode, and kind.",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {},
|
||||
},
|
||||
),
|
||||
Tool(
|
||||
name="ticktick_get_project",
|
||||
description="Get a TickTick project and all its tasks by project ID.",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"project_id": {
|
||||
"type": "string",
|
||||
"description": "The project ID to retrieve",
|
||||
},
|
||||
},
|
||||
"required": ["project_id"],
|
||||
},
|
||||
),
|
||||
Tool(
|
||||
name="ticktick_create_project",
|
||||
description="Create a new TickTick project.",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"name": {
|
||||
"type": "string",
|
||||
"description": "Project name (required)",
|
||||
},
|
||||
"color": {
|
||||
"type": "string",
|
||||
"description": "Hex color code (e.g. '#ff6347')",
|
||||
},
|
||||
"viewMode": {
|
||||
"type": "string",
|
||||
"enum": ["list", "kanban", "timeline"],
|
||||
"description": "View mode for the project",
|
||||
},
|
||||
"kind": {
|
||||
"type": "string",
|
||||
"enum": ["TASK", "NOTE"],
|
||||
"description": "Project kind (default TASK)",
|
||||
},
|
||||
},
|
||||
"required": ["name"],
|
||||
},
|
||||
),
|
||||
Tool(
|
||||
name="ticktick_update_project",
|
||||
description="Update an existing TickTick project's name, color, or viewMode.",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"project_id": {
|
||||
"type": "string",
|
||||
"description": "The project ID to update",
|
||||
},
|
||||
"name": {
|
||||
"type": "string",
|
||||
"description": "New project name",
|
||||
},
|
||||
"color": {
|
||||
"type": "string",
|
||||
"description": "New hex color code",
|
||||
},
|
||||
"viewMode": {
|
||||
"type": "string",
|
||||
"enum": ["list", "kanban", "timeline"],
|
||||
"description": "New view mode",
|
||||
},
|
||||
},
|
||||
"required": ["project_id"],
|
||||
},
|
||||
),
|
||||
Tool(
|
||||
name="ticktick_delete_project",
|
||||
description="Delete a TickTick project by ID. This is irreversible.",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"project_id": {
|
||||
"type": "string",
|
||||
"description": "The project ID to delete",
|
||||
},
|
||||
},
|
||||
"required": ["project_id"],
|
||||
},
|
||||
),
|
||||
# ----- Tasks -----
|
||||
Tool(
|
||||
name="ticktick_create_task",
|
||||
description="Create a new task in a TickTick project.",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"title": {
|
||||
"type": "string",
|
||||
"description": "Task title (required)",
|
||||
},
|
||||
"project_id": {
|
||||
"type": "string",
|
||||
"description": "Project ID to create the task in (required)",
|
||||
},
|
||||
"content": {
|
||||
"type": "string",
|
||||
"description": "Task description / notes",
|
||||
},
|
||||
"priority": {
|
||||
"type": "integer",
|
||||
"enum": [0, 1, 3, 5],
|
||||
"description": "Priority: 0=none, 1=low, 3=medium, 5=high",
|
||||
},
|
||||
"due_date": {
|
||||
"type": "string",
|
||||
"description": "Due date in ISO 8601 format (e.g. 2026-04-01T12:00:00+0000)",
|
||||
},
|
||||
"tags": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
"description": "List of tag names to attach",
|
||||
},
|
||||
},
|
||||
"required": ["title", "project_id"],
|
||||
},
|
||||
),
|
||||
Tool(
|
||||
name="ticktick_update_task",
|
||||
description="Update an existing task's title, content, priority, due date, or tags.",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"task_id": {
|
||||
"type": "string",
|
||||
"description": "The task ID to update",
|
||||
},
|
||||
"project_id": {
|
||||
"type": "string",
|
||||
"description": "The project ID the task belongs to",
|
||||
},
|
||||
"title": {
|
||||
"type": "string",
|
||||
"description": "New task title",
|
||||
},
|
||||
"content": {
|
||||
"type": "string",
|
||||
"description": "New task description / notes",
|
||||
},
|
||||
"priority": {
|
||||
"type": "integer",
|
||||
"enum": [0, 1, 3, 5],
|
||||
"description": "New priority: 0=none, 1=low, 3=medium, 5=high",
|
||||
},
|
||||
"due_date": {
|
||||
"type": "string",
|
||||
"description": "New due date in ISO 8601 format",
|
||||
},
|
||||
"tags": {
|
||||
"type": "array",
|
||||
"items": {"type": "string"},
|
||||
"description": "Replacement list of tag names",
|
||||
},
|
||||
},
|
||||
"required": ["task_id", "project_id"],
|
||||
},
|
||||
),
|
||||
Tool(
|
||||
name="ticktick_complete_task",
|
||||
description="Mark a TickTick task as completed.",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"task_id": {
|
||||
"type": "string",
|
||||
"description": "The task ID to complete",
|
||||
},
|
||||
"project_id": {
|
||||
"type": "string",
|
||||
"description": "The project ID the task belongs to",
|
||||
},
|
||||
},
|
||||
"required": ["task_id", "project_id"],
|
||||
},
|
||||
),
|
||||
Tool(
|
||||
name="ticktick_delete_task",
|
||||
description="Delete a TickTick task. This is irreversible.",
|
||||
inputSchema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"task_id": {
|
||||
"type": "string",
|
||||
"description": "The task ID to delete",
|
||||
},
|
||||
"project_id": {
|
||||
"type": "string",
|
||||
"description": "The project ID the task belongs to",
|
||||
},
|
||||
},
|
||||
"required": ["task_id", "project_id"],
|
||||
},
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tool dispatch
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@app.call_tool()
|
||||
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
|
||||
"""Route tool calls to the appropriate handler."""
|
||||
try:
|
||||
if name == "ticktick_list_projects":
|
||||
return await _handle_list_projects()
|
||||
elif name == "ticktick_get_project":
|
||||
return await _handle_get_project(arguments)
|
||||
elif name == "ticktick_create_project":
|
||||
return await _handle_create_project(arguments)
|
||||
elif name == "ticktick_update_project":
|
||||
return await _handle_update_project(arguments)
|
||||
elif name == "ticktick_delete_project":
|
||||
return await _handle_delete_project(arguments)
|
||||
elif name == "ticktick_create_task":
|
||||
return await _handle_create_task(arguments)
|
||||
elif name == "ticktick_update_task":
|
||||
return await _handle_update_task(arguments)
|
||||
elif name == "ticktick_complete_task":
|
||||
return await _handle_complete_task(arguments)
|
||||
elif name == "ticktick_delete_task":
|
||||
return await _handle_delete_task(arguments)
|
||||
else:
|
||||
return _error_text(f"[ERROR] Unknown tool: {name}")
|
||||
except FileNotFoundError as exc:
|
||||
return _error_text(str(exc))
|
||||
except RuntimeError as exc:
|
||||
return _error_text(str(exc))
|
||||
except Exception as exc:
|
||||
return _error_text(f"[ERROR] Unexpected failure in {name}: {exc}")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Handler implementations
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def _handle_list_projects() -> list[TextContent]:
|
||||
resp = await _ticktick_request("GET", "/project")
|
||||
if resp.status_code != 200:
|
||||
return _error_text(
|
||||
f"[ERROR] Failed to list projects ({resp.status_code}): {resp.text}"
|
||||
)
|
||||
projects = resp.json()
|
||||
return [TextContent(type="text", text=f"[OK] {len(projects)} projects found\n\n{_format_response(projects)}")]
|
||||
|
||||
|
||||
async def _handle_get_project(args: dict) -> list[TextContent]:
|
||||
project_id = args["project_id"]
|
||||
resp = await _ticktick_request("GET", f"/project/{project_id}/data")
|
||||
if resp.status_code != 200:
|
||||
return _error_text(
|
||||
f"[ERROR] Failed to get project {project_id} ({resp.status_code}): {resp.text}"
|
||||
)
|
||||
data = resp.json()
|
||||
task_count = len(data.get("tasks", []))
|
||||
return [TextContent(type="text", text=f"[OK] Project retrieved ({task_count} tasks)\n\n{_format_response(data)}")]
|
||||
|
||||
|
||||
async def _handle_create_project(args: dict) -> list[TextContent]:
|
||||
body: dict[str, Any] = {"name": args["name"]}
|
||||
for key in ("color", "viewMode", "kind"):
|
||||
if key in args:
|
||||
body[key] = args[key]
|
||||
|
||||
resp = await _ticktick_request("POST", "/project", json_body=body)
|
||||
if resp.status_code not in (200, 201):
|
||||
return _error_text(
|
||||
f"[ERROR] Failed to create project ({resp.status_code}): {resp.text}"
|
||||
)
|
||||
project = resp.json()
|
||||
return [TextContent(type="text", text=f"[OK] Project created\n\n{_format_response(project)}")]
|
||||
|
||||
|
||||
async def _handle_update_project(args: dict) -> list[TextContent]:
|
||||
project_id = args["project_id"]
|
||||
body: dict[str, Any] = {}
|
||||
for key in ("name", "color", "viewMode"):
|
||||
if key in args:
|
||||
body[key] = args[key]
|
||||
|
||||
if not body:
|
||||
return _error_text("[WARNING] No update fields provided. Supply at least one of: name, color, viewMode.")
|
||||
|
||||
# TickTick uses POST for project updates in some API versions; fall back to PUT.
|
||||
resp = await _ticktick_request("POST", f"/project/{project_id}", json_body=body)
|
||||
if resp.status_code in (404, 405):
|
||||
resp = await _ticktick_request("PUT", f"/project/{project_id}", json_body=body)
|
||||
|
||||
if resp.status_code not in (200, 201):
|
||||
return _error_text(
|
||||
f"[ERROR] Failed to update project {project_id} ({resp.status_code}): {resp.text}"
|
||||
)
|
||||
project = resp.json()
|
||||
return [TextContent(type="text", text=f"[OK] Project updated\n\n{_format_response(project)}")]
|
||||
|
||||
|
||||
async def _handle_delete_project(args: dict) -> list[TextContent]:
|
||||
project_id = args["project_id"]
|
||||
resp = await _ticktick_request("DELETE", f"/project/{project_id}")
|
||||
if resp.status_code not in (200, 204):
|
||||
return _error_text(
|
||||
f"[ERROR] Failed to delete project {project_id} ({resp.status_code}): {resp.text}"
|
||||
)
|
||||
return [TextContent(type="text", text=f"[OK] Project {project_id} deleted successfully.")]
|
||||
|
||||
|
||||
async def _handle_create_task(args: dict) -> list[TextContent]:
|
||||
body: dict[str, Any] = {
|
||||
"title": args["title"],
|
||||
"projectId": args["project_id"],
|
||||
}
|
||||
if "content" in args:
|
||||
body["content"] = args["content"]
|
||||
if "priority" in args:
|
||||
body["priority"] = args["priority"]
|
||||
if "due_date" in args:
|
||||
body["dueDate"] = args["due_date"]
|
||||
if "tags" in args:
|
||||
body["tags"] = args["tags"]
|
||||
|
||||
resp = await _ticktick_request("POST", "/task", json_body=body)
|
||||
if resp.status_code not in (200, 201):
|
||||
return _error_text(
|
||||
f"[ERROR] Failed to create task ({resp.status_code}): {resp.text}"
|
||||
)
|
||||
task = resp.json()
|
||||
return [TextContent(type="text", text=f"[OK] Task created\n\n{_format_response(task)}")]
|
||||
|
||||
|
||||
async def _handle_update_task(args: dict) -> list[TextContent]:
|
||||
task_id = args["task_id"]
|
||||
project_id = args["project_id"]
|
||||
|
||||
body: dict[str, Any] = {
|
||||
"taskId": task_id,
|
||||
"projectId": project_id,
|
||||
}
|
||||
if "title" in args:
|
||||
body["title"] = args["title"]
|
||||
if "content" in args:
|
||||
body["content"] = args["content"]
|
||||
if "priority" in args:
|
||||
body["priority"] = args["priority"]
|
||||
if "due_date" in args:
|
||||
body["dueDate"] = args["due_date"]
|
||||
if "tags" in args:
|
||||
body["tags"] = args["tags"]
|
||||
|
||||
resp = await _ticktick_request("POST", f"/task/{task_id}", json_body=body)
|
||||
if resp.status_code not in (200, 201):
|
||||
return _error_text(
|
||||
f"[ERROR] Failed to update task {task_id} ({resp.status_code}): {resp.text}"
|
||||
)
|
||||
task = resp.json()
|
||||
return [TextContent(type="text", text=f"[OK] Task updated\n\n{_format_response(task)}")]
|
||||
|
||||
|
||||
async def _handle_complete_task(args: dict) -> list[TextContent]:
|
||||
task_id = args["task_id"]
|
||||
project_id = args["project_id"]
|
||||
resp = await _ticktick_request(
|
||||
"POST", f"/project/{project_id}/task/{task_id}/complete"
|
||||
)
|
||||
if resp.status_code not in (200, 204):
|
||||
return _error_text(
|
||||
f"[ERROR] Failed to complete task {task_id} ({resp.status_code}): {resp.text}"
|
||||
)
|
||||
return [TextContent(type="text", text=f"[OK] Task {task_id} marked as completed.")]
|
||||
|
||||
|
||||
async def _handle_delete_task(args: dict) -> list[TextContent]:
|
||||
task_id = args["task_id"]
|
||||
project_id = args["project_id"]
|
||||
resp = await _ticktick_request(
|
||||
"DELETE", f"/project/{project_id}/task/{task_id}"
|
||||
)
|
||||
if resp.status_code not in (200, 204):
|
||||
return _error_text(
|
||||
f"[ERROR] Failed to delete task {task_id} ({resp.status_code}): {resp.text}"
|
||||
)
|
||||
return [TextContent(type="text", text=f"[OK] Task {task_id} deleted successfully.")]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Entry point
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
async def main() -> None:
|
||||
"""Run the TickTick MCP server over stdio transport."""
|
||||
try:
|
||||
from mcp.server.stdio import stdio_server
|
||||
|
||||
async with stdio_server() as (read_stream, write_stream):
|
||||
await app.run(
|
||||
read_stream,
|
||||
write_stream,
|
||||
app.create_initialization_options(),
|
||||
)
|
||||
except Exception as exc:
|
||||
print(f"[ERROR] MCP server failed: {exc}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(main())
|
||||
Reference in New Issue
Block a user