Todoist MCP Server

#!/usr/bin/env python3 """ Simple Todoist MCP Server - Manage your Todoist tasks, projects, and more """ import os import sys import logging from datetime import datetime, timezone import httpx from mcp.server.fastmcp import FastMCP # Configure logging to stderr logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', stream=sys.stderr ) logger = logging.getLogger("todoist-server") # Initialize MCP server mcp = FastMCP("todoist") # Configuration API_TOKEN = os.environ.get("TODOIST_API_TOKEN", "") BASE_URL = "https://api.todoist.com/rest/v2" # === UTILITY FUNCTIONS === def get_headers(): """Get authorization headers for Todoist API.""" return { "Authorization": f"Bearer {API_TOKEN}", "Content-Type": "application/json" } def format_task(task): """Format a task object into a readable string.""" result = f"""📋 Task: {task.get('content', 'Untitled')} ID: {task.get('id', 'N/A')} Project ID: {task.get('project_id', 'N/A')} Priority: {task.get('priority', 1)} Due: {task.get('due', {}).get('string', 'No due date') if task.get('due') else 'No due date'} Completed: {'✅ Yes' if task.get('is_completed') else '❌ No'} Description: {task.get('description', 'No description')} Labels: {', '.join(task.get('labels', [])) if task.get('labels') else 'None'} URL: {task.get('url', 'N/A')}""" return result def format_project(project): """Format a project object into a readable string.""" result = f"""📁 Project: {project.get('name', 'Untitled')} ID: {project.get('id', 'N/A')} Color: {project.get('color', 'N/A')} Favorite: {'⭐ Yes' if project.get('is_favorite') else '❌ No'} View Style: {project.get('view_style', 'list')} URL: {project.get('url', 'N/A')}""" return result # === TASK CRUD OPERATIONS === @mcp.tool() async def create_task(content: str = "", description: str = "", project_id: str = "", due_string: str = "", priority: str = "1", labels: str = "") -> str: """Create a new task in Todoist with content, optional description, project, due date, priority (1-4), and comma-separated labels.""" logger.info(f"Creating task: {content}") if not content.strip(): return "❌ Error: Task content is required" if not API_TOKEN.strip(): return "❌ Error: TODOIST_API_TOKEN not configured" try: data = {"content": content.strip()} if description.strip(): data["description"] = description.strip() if project_id.strip(): data["project_id"] = project_id.strip() if due_string.strip(): data["due_string"] = due_string.strip() if priority.strip() and priority.strip() in ["1", "2", "3", "4"]: data["priority"] = int(priority.strip()) if labels.strip(): label_list = [label.strip() for label in labels.split(",") if label.strip()] if label_list: data["labels"] = label_list async with httpx.AsyncClient() as client: response = await client.post( f"{BASE_URL}/tasks", json=data, headers=get_headers(), timeout=10 ) response.raise_for_status() task = response.json() return f"✅ Task created successfully!\n\n{format_task(task)}" except httpx.HTTPStatusError as e: logger.error(f"HTTP error: {e.response.status_code}") return f"❌ API Error: {e.response.status_code} - {e.response.text}" except Exception as e: logger.error(f"Error creating task: {e}") return f"❌ Error: {str(e)}" @mcp.tool() async def list_tasks(project_id: str = "", filter_query: str = "", label: str = "") -> str: """List all active tasks, optionally filtered by project ID, filter query, or label name.""" logger.info("Listing tasks") if not API_TOKEN.strip(): return "❌ Error: TODOIST_API_TOKEN not configured" try: params = {} if project_id.strip(): params["project_id"] = project_id.strip() if filter_query.strip(): params["filter"] = filter_query.strip() if label.strip(): params["label"] = label.strip() async with httpx.AsyncClient() as client: response = await client.get( f"{BASE_URL}/tasks", params=params, headers=get_headers(), timeout=10 ) response.raise_for_status() tasks = response.json() if not tasks: return "📋 No tasks found" result = f"📋 Found {len(tasks)} task(s):\n\n" for i, task in enumerate(tasks, 1): result += f"{i}. {task.get('content', 'Untitled')} (ID: {task.get('id')})\n" result += f" Priority: {task.get('priority', 1)} | Due: {task.get('due', {}).get('string', 'No due date') if task.get('due') else 'No due date'}\n" return result except httpx.HTTPStatusError as e: logger.error(f"HTTP error: {e.response.status_code}") return f"❌ API Error: {e.response.status_code} - {e.response.text}" except Exception as e: logger.error(f"Error listing tasks: {e}") return f"❌ Error: {str(e)}" @mcp.tool() async def get_task(task_id: str = "") -> str: """Get detailed information about a specific task by its ID.""" logger.info(f"Getting task: {task_id}") if not task_id.strip(): return "❌ Error: Task ID is required" if not API_TOKEN.strip(): return "❌ Error: TODOIST_API_TOKEN not configured" try: async with httpx.AsyncClient() as client: response = await client.get( f"{BASE_URL}/tasks/{task_id.strip()}", headers=get_headers(), timeout=10 ) response.raise_for_status() task = response.json() return f"✅ Task details:\n\n{format_task(task)}" except httpx.HTTPStatusError as e: logger.error(f"HTTP error: {e.response.status_code}") if e.response.status_code == 404: return f"❌ Task not found with ID: {task_id}" return f"❌ API Error: {e.response.status_code} - {e.response.text}" except Exception as e: logger.error(f"Error getting task: {e}") return f"❌ Error: {str(e)}" @mcp.tool() async def update_task(task_id: str = "", content: str = "", description: str = "", due_string: str = "", priority: str = "", labels: str = "") -> str: """Update an existing task by ID with new content, description, due date, priority (1-4), or comma-separated labels.""" logger.info(f"Updating task: {task_id}") if not task_id.strip(): return "❌ Error: Task ID is required" if not API_TOKEN.strip(): return "❌ Error: TODOIST_API_TOKEN not configured" try: data = {} if content.strip(): data["content"] = content.strip() if description.strip(): data["description"] = description.strip() if due_string.strip(): data["due_string"] = due_string.strip() if priority.strip() and priority.strip() in ["1", "2", "3", "4"]: data["priority"] = int(priority.strip()) if labels.strip(): label_list = [label.strip() for label in labels.split(",") if label.strip()] data["labels"] = label_list if not data: return "❌ Error: No update fields provided" async with httpx.AsyncClient() as client: response = await client.post( f"{BASE_URL}/tasks/{task_id.strip()}", json=data, headers=get_headers(), timeout=10 ) response.raise_for_status() task = response.json() return f"✅ Task updated successfully!\n\n{format_task(task)}" except httpx.HTTPStatusError as e: logger.error(f"HTTP error: {e.response.status_code}") if e.response.status_code == 404: return f"❌ Task not found with ID: {task_id}" return f"❌ API Error: {e.response.status_code} - {e.response.text}" except Exception as e: logger.error(f"Error updating task: {e}") return f"❌ Error: {str(e)}" @mcp.tool() async def complete_task(task_id: str = "") -> str: """Mark a task as completed by its ID.""" logger.info(f"Completing task: {task_id}") if not task_id.strip(): return "❌ Error: Task ID is required" if not API_TOKEN.strip(): return "❌ Error: TODOIST_API_TOKEN not configured" try: async with httpx.AsyncClient() as client: response = await client.post( f"{BASE_URL}/tasks/{task_id.strip()}/close", headers=get_headers(), timeout=10 ) response.raise_for_status() return f"✅ Task {task_id} marked as completed!" except httpx.HTTPStatusError as e: logger.error(f"HTTP error: {e.response.status_code}") if e.response.status_code == 404: return f"❌ Task not found with ID: {task_id}" return f"❌ API Error: {e.response.status_code} - {e.response.text}" except Exception as e: logger.error(f"Error completing task: {e}") return f"❌ Error: {str(e)}" @mcp.tool() async def reopen_task(task_id: str = "") -> str: """Reopen a completed task by its ID.""" logger.info(f"Reopening task: {task_id}") if not task_id.strip(): return "❌ Error: Task ID is required" if not API_TOKEN.strip(): return "❌ Error: TODOIST_API_TOKEN not configured" try: async with httpx.AsyncClient() as client: response = await client.post( f"{BASE_URL}/tasks/{task_id.strip()}/reopen", headers=get_headers(), timeout=10 ) response.raise_for_status() return f"✅ Task {task_id} reopened successfully!" except httpx.HTTPStatusError as e: logger.error(f"HTTP error: {e.response.status_code}") if e.response.status_code == 404: return f"❌ Task not found with ID: {task_id}" return f"❌ API Error: {e.response.status_code} - {e.response.text}" except Exception as e: logger.error(f"Error reopening task: {e}") return f"❌ Error: {str(e)}" @mcp.tool() async def delete_task(task_id: str = "") -> str: """Delete a task permanently by its ID.""" logger.info(f"Deleting task: {task_id}") if not task_id.strip(): return "❌ Error: Task ID is required" if not API_TOKEN.strip(): return "❌ Error: TODOIST_API_TOKEN not configured" try: async with httpx.AsyncClient() as client: response = await client.delete( f"{BASE_URL}/tasks/{task_id.strip()}", headers=get_headers(), timeout=10 ) response.raise_for_status() return f"✅ Task {task_id} deleted successfully!" except httpx.HTTPStatusError as e: logger.error(f"HTTP error: {e.response.status_code}") if e.response.status_code == 404: return f"❌ Task not found with ID: {task_id}" return f"❌ API Error: {e.response.status_code} - {e.response.text}" except Exception as e: logger.error(f"Error deleting task: {e}") return f"❌ Error: {str(e)}" # === SYNC API FOR COMPLETED TASKS === @mcp.tool() async def list_completed_tasks(project_id: str = "", limit: str = "30") -> str: """List completed tasks, optionally filtered by project ID with a limit (default 30, max 200).""" logger.info("Listing completed tasks") if not API_TOKEN.strip(): return "❌ Error: TODOIST_API_TOKEN not configured" try: # Todoist Sync API endpoint for completed tasks sync_url = "https://api.todoist.com/sync/v9/completed/get_all" params = {} if project_id.strip(): params["project_id"] = project_id.strip() limit_int = 30 if limit.strip(): try: limit_int = int(limit.strip()) if limit_int > 200: limit_int = 200 elif limit_int < 1: limit_int = 30 except ValueError: pass params["limit"] = limit_int async with httpx.AsyncClient() as client: response = await client.get( sync_url, params=params, headers=get_headers(), timeout=10 ) response.raise_for_status() data = response.json() items = data.get("items", []) if not items: return "✅ No completed tasks found" result = f"✅ Found {len(items)} completed task(s):\n\n" for i, item in enumerate(items, 1): task_data = item.get("task_id") or item.get("id", "Unknown") content = item.get("content", "Untitled") completed_at = item.get("completed_at", item.get("completed_date", "Unknown")) result += f"{i}. {content} (ID: {task_data})\n" result += f" Completed: {completed_at}\n" return result except httpx.HTTPStatusError as e: logger.error(f"HTTP error: {e.response.status_code}") return f"❌ API Error: {e.response.status_code} - {e.response.text}" except Exception as e: logger.error(f"Error listing completed tasks: {e}") return f"❌ Error: {str(e)}" # === PROJECT CRUD OPERATIONS === @mcp.tool() async def create_project(name: str = "", color: str = "", is_favorite: str = "false", view_style: str = "list") -> str: """Create a new project with name, optional color, favorite status (true/false), and view style (list or board).""" logger.info(f"Creating project: {name}") if not name.strip(): return "❌ Error: Project name is required" if not API_TOKEN.strip(): return "❌ Error: TODOIST_API_TOKEN not configured" try: data = {"name": name.strip()} if color.strip(): data["color"] = color.strip() if is_favorite.strip().lower() == "true": data["is_favorite"] = True if view_style.strip() and view_style.strip().lower() in ["list", "board"]: data["view_style"] = view_style.strip().lower() async with httpx.AsyncClient() as client: response = await client.post( f"{BASE_URL}/projects", json=data, headers=get_headers(), timeout=10 ) response.raise_for_status() project = response.json() return f"✅ Project created successfully!\n\n{format_project(project)}" except httpx.HTTPStatusError as e: logger.error(f"HTTP error: {e.response.status_code}") return f"❌ API Error: {e.response.status_code} - {e.response.text}" except Exception as e: logger.error(f"Error creating project: {e}") return f"❌ Error: {str(e)}" @mcp.tool() async def list_projects() -> str: """List all projects in your Todoist account.""" logger.info("Listing projects") if not API_TOKEN.strip(): return "❌ Error: TODOIST_API_TOKEN not configured" try: async with httpx.AsyncClient() as client: response = await client.get( f"{BASE_URL}/projects", headers=get_headers(), timeout=10 ) response.raise_for_status() projects = response.json() if not projects: return "📁 No projects found" result = f"📁 Found {len(projects)} project(s):\n\n" for i, project in enumerate(projects, 1): fav = "⭐" if project.get('is_favorite') else "" result += f"{i}. {fav} {project.get('name', 'Untitled')} (ID: {project.get('id')})\n" result += f" Color: {project.get('color', 'N/A')} | View: {project.get('view_style', 'list')}\n" return result except httpx.HTTPStatusError as e: logger.error(f"HTTP error: {e.response.status_code}") return f"❌ API Error: {e.response.status_code} - {e.response.text}" except Exception as e: logger.error(f"Error listing projects: {e}") return f"❌ Error: {str(e)}" @mcp.tool() async def get_project(project_id: str = "") -> str: """Get detailed information about a specific project by its ID.""" logger.info(f"Getting project: {project_id}") if not project_id.strip(): return "❌ Error: Project ID is required" if not API_TOKEN.strip(): return "❌ Error: TODOIST_API_TOKEN not configured" try: async with httpx.AsyncClient() as client: response = await client.get( f"{BASE_URL}/projects/{project_id.strip()}", headers=get_headers(), timeout=10 ) response.raise_for_status() project = response.json() return f"✅ Project details:\n\n{format_project(project)}" except httpx.HTTPStatusError as e: logger.error(f"HTTP error: {e.response.status_code}") if e.response.status_code == 404: return f"❌ Project not found with ID: {project_id}" return f"❌ API Error: {e.response.status_code} - {e.response.text}" except Exception as e: logger.error(f"Error getting project: {e}") return f"❌ Error: {str(e)}" @mcp.tool() async def update_project(project_id: str = "", name: str = "", color: str = "", is_favorite: str = "", view_style: str = "") -> str: """Update an existing project by ID with new name, color, favorite status (true/false), or view style (list or board).""" logger.info(f"Updating project: {project_id}") if not project_id.strip(): return "❌ Error: Project ID is required" if not API_TOKEN.strip(): return "❌ Error: TODOIST_API_TOKEN not configured" try: data = {} if name.strip(): data["name"] = name.strip() if color.strip(): data["color"] = color.strip() if is_favorite.strip().lower() in ["true", "false"]: data["is_favorite"] = is_favorite.strip().lower() == "true" if view_style.strip() and view_style.strip().lower() in ["list", "board"]: data["view_style"] = view_style.strip().lower() if not data: return "❌ Error: No update fields provided" async with httpx.AsyncClient() as client: response = await client.post( f"{BASE_URL}/projects/{project_id.strip()}", json=data, headers=get_headers(), timeout=10 ) response.raise_for_status() project = response.json() return f"✅ Project updated successfully!\n\n{format_project(project)}" except httpx.HTTPStatusError as e: logger.error(f"HTTP error: {e.response.status_code}") if e.response.status_code == 404: return f"❌ Project not found with ID: {project_id}" return f"❌ API Error: {e.response.status_code} - {e.response.text}" except Exception as e: logger.error(f"Error updating project: {e}") return f"❌ Error: {str(e)}" @mcp.tool() async def delete_project(project_id: str = "") -> str: """Delete a project permanently by its ID.""" logger.info(f"Deleting project: {project_id}") if not project_id.strip(): return "❌ Error: Project ID is required" if not API_TOKEN.strip(): return "❌ Error: TODOIST_API_TOKEN not configured" try: async with httpx.AsyncClient() as client: response = await client.delete( f"{BASE_URL}/projects/{project_id.strip()}", headers=get_headers(), timeout=10 ) response.raise_for_status() return f"✅ Project {project_id} deleted successfully!" except httpx.HTTPStatusError as e: logger.error(f"HTTP error: {e.response.status_code}") if e.response.status_code == 404: return f"❌ Project not found with ID: {project_id}" return f"❌ API Error: {e.response.status_code} - {e.response.text}" except Exception as e: logger.error(f"Error deleting project: {e}") return f"❌ Error: {str(e)}" # === COMMENT OPERATIONS === @mcp.tool() async def create_comment(task_id: str = "", content: str = "") -> str: """Create a comment on a task with the given task ID and comment content.""" logger.info(f"Creating comment on task: {task_id}") if not task_id.strip(): return "❌ Error: Task ID is required" if not content.strip(): return "❌ Error: Comment content is required" if not API_TOKEN.strip(): return "❌ Error: TODOIST_API_TOKEN not configured" try: data = { "task_id": task_id.strip(), "content": content.strip() } async with httpx.AsyncClient() as client: response = await client.post( f"{BASE_URL}/comments", json=data, headers=get_headers(), timeout=10 ) response.raise_for_status() comment = response.json() return f"✅ Comment created successfully!\n\nComment ID: {comment.get('id')}\nContent: {comment.get('content')}" except httpx.HTTPStatusError as e: logger.error(f"HTTP error: {e.response.status_code}") return f"❌ API Error: {e.response.status_code} - {e.response.text}" except Exception as e: logger.error(f"Error creating comment: {e}") return f"❌ Error: {str(e)}" @mcp.tool() async def list_comments(task_id: str = "", project_id: str = "") -> str: """List all comments for a task (provide task_id) or project (provide project_id).""" logger.info(f"Listing comments") if not task_id.strip() and not project_id.strip(): return "❌ Error: Either task_id or project_id is required" if not API_TOKEN.strip(): return "❌ Error: TODOIST_API_TOKEN not configured" try: params = {} if task_id.strip(): params["task_id"] = task_id.strip() elif project_id.strip(): params["project_id"] = project_id.strip() async with httpx.AsyncClient() as client: response = await client.get( f"{BASE_URL}/comments", params=params, headers=get_headers(), timeout=10 ) response.raise_for_status() comments = response.json() if not comments: return "💬 No comments found" result = f"💬 Found {len(comments)} comment(s):\n\n" for i, comment in enumerate(comments, 1): result += f"{i}. {comment.get('content', 'No content')} (ID: {comment.get('id')})\n" result += f" Posted: {comment.get('posted_at', 'Unknown')}\n" return result except httpx.HTTPStatusError as e: logger.error(f"HTTP error: {e.response.status_code}") return f"❌ API Error: {e.response.status_code} - {e.response.text}" except Exception as e: logger.error(f"Error listing comments: {e}") return f"❌ Error: {str(e)}" # === LABEL OPERATIONS === @mcp.tool() async def create_label(name: str = "", color: str = "") -> str: """Create a new label with a name and optional color.""" logger.info(f"Creating label: {name}") if not name.strip(): return "❌ Error: Label name is required" if not API_TOKEN.strip(): return "❌ Error: TODOIST_API_TOKEN not configured" try: data = {"name": name.strip()} if color.strip(): data["color"] = color.strip() async with httpx.AsyncClient() as client: response = await client.post( f"{BASE_URL}/labels", json=data, headers=get_headers(), timeout=10 ) response.raise_for_status() label = response.json() return f"✅ Label created successfully!\n\nLabel ID: {label.get('id')}\nName: {label.get('name')}\nColor: {label.get('color')}" except httpx.HTTPStatusError as e: logger.error(f"HTTP error: {e.response.status_code}") return f"❌ API Error: {e.response.status_code} - {e.response.text}" except Exception as e: logger.error(f"Error creating label: {e}") return f"❌ Error: {str(e)}" @mcp.tool() async def list_labels() -> str: """List all labels in your Todoist account.""" logger.info("Listing labels") if not API_TOKEN.strip(): return "❌ Error: TODOIST_API_TOKEN not configured" try: async with httpx.AsyncClient() as client: response = await client.get( f"{BASE_URL}/labels", headers=get_headers(), timeout=10 ) response.raise_for_status() labels = response.json() if not labels: return "🏷️ No labels found" result = f"🏷️ Found {len(labels)} label(s):\n\n" for i, label in enumerate(labels, 1): result += f"{i}. {label.get('name', 'Untitled')} (ID: {label.get('id')})\n" result += f" Color: {label.get('color', 'N/A')}\n" return result except httpx.HTTPStatusError as e: logger.error(f"HTTP error: {e.response.status_code}") return f"❌ API Error: {e.response.status_code} - {e.response.text}" except Exception as e: logger.error(f"Error listing labels: {e}") return f"❌ Error: {str(e)}" # === SERVER STARTUP === if __name__ == "__main__": logger.info("Starting Todoist MCP server...") if not API_TOKEN: logger.warning("TODOIST_API_TOKEN not set - server will not be functional") try: mcp.run(transport='stdio') except Exception as e: logger.error(f"Server error: {e}", exc_info=True) sys.exit(1)