azcomputerguru/claudetools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
claudetools/mcp-servers/ticktick/ticktick_mcp.py
Mike Swanson b26e185a80 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>
2026-03-31 10:08:53 -07:00

596 lines
21 KiB
Python
Raw Permalink Blame History

#!/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 View Git Blame Copy Permalink