todoist_mcp.py•52.3 kB
#!/usr/bin/env python3
"""
Todoist MCP Server
This server provides tools to interact with Todoist API, including task management,
project organization, and productivity features.
"""
from typing import Optional, List, Dict, Any
from enum import Enum
import httpx
from pydantic import BaseModel, Field, field_validator, ConfigDict
from mcp.server.fastmcp import FastMCP, Context
import json
from datetime import datetime
import os
# Initialize the MCP server
mcp = FastMCP("todoist_mcp")
# Constants
API_BASE_URL = "https://api.todoist.com/rest/v2"
CHARACTER_LIMIT = 25000 # Maximum response size in characters
# Enums
class ResponseFormat(str, Enum):
"""Output format for tool responses."""
MARKDOWN = "markdown"
JSON = "json"
class TaskPriority(int, Enum):
"""Task priority levels in Todoist."""
NORMAL = 1
MEDIUM = 2
HIGH = 3
URGENT = 4
# ==================== Pydantic Models for Input Validation ====================
class ListProjectsInput(BaseModel):
"""Input model for listing projects."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
response_format: ResponseFormat = Field(
default=ResponseFormat.MARKDOWN,
description="Output format: 'markdown' for human-readable or 'json' for machine-readable"
)
class CreateProjectInput(BaseModel):
"""Input model for creating a project."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
name: str = Field(..., description="Name of the project (e.g., 'Work Tasks', 'Personal Goals')", min_length=1, max_length=200)
parent_id: Optional[str] = Field(default=None, description="Parent project ID for sub-projects")
color: Optional[str] = Field(default="charcoal", description="Project color (e.g., 'red', 'blue', 'green')")
is_favorite: Optional[bool] = Field(default=False, description="Whether to mark as favorite")
view_style: Optional[str] = Field(default="list", description="Display style: 'list' or 'board'", pattern="^(list|board)$")
@field_validator('name')
@classmethod
def validate_name(cls, v: str) -> str:
if not v.strip():
raise ValueError("Project name cannot be empty")
return v.strip()
class UpdateProjectInput(BaseModel):
"""Input model for updating a project."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
project_id: str = Field(..., description="Project ID to update", min_length=1)
name: Optional[str] = Field(default=None, description="New project name", min_length=1, max_length=200)
color: Optional[str] = Field(default=None, description="New project color")
is_favorite: Optional[bool] = Field(default=None, description="Update favorite status")
view_style: Optional[str] = Field(default=None, description="Display style: 'list' or 'board'", pattern="^(list|board)$")
class ListTasksInput(BaseModel):
"""Input model for listing tasks."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
project_id: Optional[str] = Field(default=None, description="Filter by project ID")
section_id: Optional[str] = Field(default=None, description="Filter by section ID")
label: Optional[str] = Field(default=None, description="Filter by label name")
filter: Optional[str] = Field(default=None, description="Todoist filter query (e.g., 'today', 'overdue', 'p1')")
response_format: ResponseFormat = Field(
default=ResponseFormat.MARKDOWN,
description="Output format: 'markdown' for human-readable or 'json' for machine-readable"
)
class CreateTaskInput(BaseModel):
"""Input model for creating a task."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
content: str = Field(..., description="Task content/title (e.g., 'Buy milk', 'Finish report')", min_length=1, max_length=500)
description: Optional[str] = Field(default="", description="Task description with more details")
project_id: Optional[str] = Field(default=None, description="Project ID to add task to")
section_id: Optional[str] = Field(default=None, description="Section ID within the project")
parent_id: Optional[str] = Field(default=None, description="Parent task ID for subtasks")
labels: Optional[List[str]] = Field(default_factory=list, description="List of label names to apply", max_items=10)
priority: Optional[TaskPriority] = Field(default=TaskPriority.NORMAL, description="Priority (1=normal, 2=medium, 3=high, 4=urgent)")
due_string: Optional[str] = Field(default=None, description="Natural language due date (e.g., 'tomorrow', 'next Monday at 2pm')")
due_date: Optional[str] = Field(default=None, description="Due date in YYYY-MM-DD format")
assignee_id: Optional[str] = Field(default=None, description="User ID to assign task to (for shared projects)")
@field_validator('content')
@classmethod
def validate_content(cls, v: str) -> str:
if not v.strip():
raise ValueError("Task content cannot be empty")
return v.strip()
class UpdateTaskInput(BaseModel):
"""Input model for updating a task."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
task_id: str = Field(..., description="Task ID to update", min_length=1)
content: Optional[str] = Field(default=None, description="New task content", min_length=1, max_length=500)
description: Optional[str] = Field(default=None, description="New task description")
labels: Optional[List[str]] = Field(default=None, description="New list of labels", max_items=10)
priority: Optional[TaskPriority] = Field(default=None, description="New priority level")
due_string: Optional[str] = Field(default=None, description="New due date in natural language")
due_date: Optional[str] = Field(default=None, description="New due date in YYYY-MM-DD format")
assignee_id: Optional[str] = Field(default=None, description="New assignee ID")
class CompleteTaskInput(BaseModel):
"""Input model for completing a task."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
task_id: str = Field(..., description="Task ID to complete", min_length=1)
class QuickAddTaskInput(BaseModel):
"""Input model for quick add task using natural language."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
text: str = Field(..., description="Natural language task (e.g., 'Buy milk tomorrow at 2pm #Shopping p1')", min_length=1, max_length=500)
@field_validator('text')
@classmethod
def validate_text(cls, v: str) -> str:
if not v.strip():
raise ValueError("Task text cannot be empty")
return v.strip()
class ListSectionsInput(BaseModel):
"""Input model for listing sections."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
project_id: Optional[str] = Field(default=None, description="Filter sections by project ID")
response_format: ResponseFormat = Field(
default=ResponseFormat.MARKDOWN,
description="Output format: 'markdown' for human-readable or 'json' for machine-readable"
)
class CreateSectionInput(BaseModel):
"""Input model for creating a section."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
name: str = Field(..., description="Section name (e.g., 'In Progress', 'Waiting')", min_length=1, max_length=200)
project_id: str = Field(..., description="Project ID to add section to", min_length=1)
order: Optional[int] = Field(default=None, description="Section order/position", ge=0, le=1000)
class ListLabelsInput(BaseModel):
"""Input model for listing labels."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
response_format: ResponseFormat = Field(
default=ResponseFormat.MARKDOWN,
description="Output format: 'markdown' for human-readable or 'json' for machine-readable"
)
class CreateLabelInput(BaseModel):
"""Input model for creating a label."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
name: str = Field(..., description="Label name (e.g., 'urgent', 'work', 'personal')", min_length=1, max_length=100)
color: Optional[str] = Field(default="charcoal", description="Label color")
is_favorite: Optional[bool] = Field(default=False, description="Whether to mark as favorite")
class GetCommentsInput(BaseModel):
"""Input model for getting comments."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
task_id: Optional[str] = Field(default=None, description="Task ID to get comments for")
project_id: Optional[str] = Field(default=None, description="Project ID to get comments for")
response_format: ResponseFormat = Field(
default=ResponseFormat.MARKDOWN,
description="Output format: 'markdown' for human-readable or 'json' for machine-readable"
)
@field_validator('task_id')
@classmethod
def validate_ids(cls, v: Optional[str], info) -> Optional[str]:
values = info.data
if not v and not values.get('project_id'):
raise ValueError("Either task_id or project_id must be provided")
return v
class AddCommentInput(BaseModel):
"""Input model for adding a comment."""
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
extra='forbid'
)
content: str = Field(..., description="Comment text", min_length=1, max_length=1000)
task_id: Optional[str] = Field(default=None, description="Task ID to comment on")
project_id: Optional[str] = Field(default=None, description="Project ID to comment on")
@field_validator('content')
@classmethod
def validate_content(cls, v: str) -> str:
if not v.strip():
raise ValueError("Comment content cannot be empty")
return v.strip()
@field_validator('task_id')
@classmethod
def validate_ids(cls, v: Optional[str], info) -> Optional[str]:
values = info.data
if not v and not values.get('project_id'):
raise ValueError("Either task_id or project_id must be provided")
return v
# ==================== Shared Utility Functions ====================
def _get_api_token() -> Optional[str]:
"""Get API token from environment variable."""
return os.environ.get('TODOIST_API_TOKEN')
async def _make_api_request(
endpoint: str,
token: str,
method: str = "GET",
json_data: Optional[dict] = None,
params: Optional[dict] = None
) -> dict:
"""Reusable function for all API calls."""
headers = {"Authorization": f"Bearer {token}"}
if json_data:
headers["Content-Type"] = "application/json"
async with httpx.AsyncClient() as client:
response = await client.request(
method,
f"{API_BASE_URL}/{endpoint}",
headers=headers,
json=json_data,
params=params,
timeout=30.0
)
# Handle 204 No Content responses
if response.status_code == 204:
return {"success": True}
response.raise_for_status()
# Some endpoints return arrays, not objects
data = response.json()
if isinstance(data, list):
return {"items": data}
return data
def _handle_api_error(e: Exception) -> str:
"""Consistent error formatting across all tools."""
if isinstance(e, httpx.HTTPStatusError):
if e.response.status_code == 401:
return "Error: Invalid or missing API token. Please check your Todoist API token."
elif e.response.status_code == 403:
return "Error: Permission denied. You don't have access to this resource."
elif e.response.status_code == 404:
return "Error: Resource not found. Please check the ID is correct."
elif e.response.status_code == 429:
return "Error: Rate limit exceeded (max 1000 requests per 15 minutes). Please wait before making more requests."
return f"Error: API request failed with status {e.response.status_code}: {e.response.text}"
elif isinstance(e, httpx.TimeoutException):
return "Error: Request timed out. Please try again."
return f"Error: Unexpected error occurred: {type(e).__name__}: {str(e)}"
def _format_datetime(dt_str: str) -> str:
"""Convert ISO datetime to human-readable format."""
if not dt_str:
return "No date"
try:
dt = datetime.fromisoformat(dt_str.replace("Z", "+00:00"))
return dt.strftime("%Y-%m-%d %H:%M:%S UTC")
except:
return dt_str
def _format_due_date(due: Optional[dict]) -> str:
"""Format due date information for display."""
if not due:
return "No due date"
parts = []
if due.get('date'):
parts.append(due['date'])
if due.get('datetime'):
parts.append(f"at {_format_datetime(due['datetime'])}")
if due.get('string'):
parts.append(f"({due['string']})")
if due.get('is_recurring'):
parts.append("🔄 Recurring")
return " ".join(parts) if parts else "No due date"
def _priority_to_emoji(priority: int) -> str:
"""Convert priority number to emoji."""
return {
1: "⚪", # Normal
2: "🔵", # Medium
3: "🟡", # High
4: "🔴" # Urgent
}.get(priority, "⚪")
# ==================== Tool Definitions ====================
# ---------- Project Tools ----------
@mcp.tool(
name="todoist_list_projects",
annotations={
"title": "List Todoist Projects",
"readOnlyHint": True,
"destructiveHint": False,
"idempotentHint": True,
"openWorldHint": True
}
)
async def todoist_list_projects(params: ListProjectsInput, ctx: Context) -> str:
"""List all projects in Todoist account.
This tool retrieves all projects from the user's Todoist account, including
inbox, personal projects, and shared projects. Projects are displayed with
their hierarchy, colors, and metadata.
Args:
params (ListProjectsInput): Validated input parameters containing:
- response_format (ResponseFormat): Output format preference
ctx (Context): MCP context with access to environment variables
Returns:
str: Formatted list of projects in requested format
Markdown format includes:
- Hierarchical project structure with indentation
- Project colors and favorite status
- Comment counts and sharing status
JSON format includes full project objects with all metadata
Examples:
- Use when: "Show me all my Todoist projects"
- Use when: "What projects do I have?"
- Don't use when: You need to create or modify projects
"""
try:
# Get API token from environment
token = _get_api_token()
if not token:
return "Error: TODOIST_API_TOKEN not configured. Please set your Todoist API token in the environment variables."
data = await _make_api_request("projects", token)
projects = data.get("items", [])
if not projects:
return "No projects found in your Todoist account."
if params.response_format == ResponseFormat.MARKDOWN:
lines = ["# 📁 Todoist Projects", ""]
# Sort projects by parent_id and order for hierarchy
root_projects = [p for p in projects if not p.get('parent_id')]
for project in sorted(root_projects, key=lambda x: x.get('order', 0)):
# Format root project
icon = "📥" if project.get('is_inbox_project') else "⭐" if project.get('is_favorite') else "📁"
shared = " 👥" if project.get('is_shared') else ""
lines.append(f"## {icon} {project['name']} ({project['id']}){shared}")
lines.append(f"- **Color**: {project.get('color', 'default')}")
lines.append(f"- **View**: {project.get('view_style', 'list')}")
if project.get('comment_count', 0) > 0:
lines.append(f"- **Comments**: {project['comment_count']}")
# Find child projects
children = [p for p in projects if p.get('parent_id') == project['id']]
for child in sorted(children, key=lambda x: x.get('order', 0)):
lines.append(f" ### ↳ {child['name']} ({child['id']})")
lines.append(f" - Color: {child.get('color', 'default')}")
lines.append("")
result = "\n".join(lines)
else:
result = json.dumps(projects, indent=2)
# Check character limit
if len(result) > CHARACTER_LIMIT:
if params.response_format == ResponseFormat.MARKDOWN:
return f"Response truncated (too many projects). Found {len(projects)} projects. Consider filtering by specific project."
else:
truncated = projects[:len(projects)//2]
result = json.dumps({
"projects": truncated,
"truncated": True,
"total_count": len(projects),
"shown_count": len(truncated)
}, indent=2)
return result
except Exception as e:
return _handle_api_error(e)
@mcp.tool(
name="todoist_create_project",
annotations={
"title": "Create Todoist Project",
"readOnlyHint": False,
"destructiveHint": False,
"idempotentHint": False,
"openWorldHint": True
}
)
async def todoist_create_project(params: CreateProjectInput, ctx: Context) -> str:
"""Create a new project in Todoist.
This tool creates a new project with specified attributes. Projects can be
organized hierarchically with parent-child relationships, customized with
colors, and set as favorites for quick access.
Args:
params (CreateProjectInput): Validated input parameters containing:
- name (str): Project name
- parent_id (Optional[str]): Parent project for nesting
- color (Optional[str]): Project color
- is_favorite (Optional[bool]): Mark as favorite
- view_style (Optional[str]): 'list' or 'board' view
ctx (Context): MCP context with access to environment variables
Returns:
str: Confirmation message with new project details
Examples:
- Use when: "Create a new project called 'Q1 Goals'"
- Use when: "Add a Work project as a favorite"
- Don't use when: Updating existing projects
"""
try:
# Get API token from environment
token = _get_api_token()
if not token:
return "Error: TODOIST_API_TOKEN not configured. Please set your Todoist API token in the environment variables."
json_data = {"name": params.name}
if params.parent_id:
json_data["parent_id"] = params.parent_id
if params.color:
json_data["color"] = params.color
if params.is_favorite is not None:
json_data["is_favorite"] = params.is_favorite
if params.view_style:
json_data["view_style"] = params.view_style
data = await _make_api_request("projects", token, method="POST", json_data=json_data)
return f"""✅ Project created successfully!
**Name**: {data['name']}
**ID**: {data['id']}
**Color**: {data.get('color', 'default')}
**View Style**: {data.get('view_style', 'list')}
**Favorite**: {'⭐ Yes' if data.get('is_favorite') else 'No'}
**URL**: {data.get('url', 'N/A')}"""
except Exception as e:
return _handle_api_error(e)
@mcp.tool(
name="todoist_update_project",
annotations={
"title": "Update Todoist Project",
"readOnlyHint": False,
"destructiveHint": False,
"idempotentHint": True,
"openWorldHint": True
}
)
async def todoist_update_project(params: UpdateProjectInput, ctx: Context) -> str:
"""Update an existing Todoist project.
This tool modifies project properties like name, color, favorite status,
and view style. Only provided fields will be updated.
Args:
params (UpdateProjectInput): Validated input parameters containing:
- project_id (str): ID of project to update
- name (Optional[str]): New project name
- color (Optional[str]): New color
- is_favorite (Optional[bool]): Update favorite status
- view_style (Optional[str]): Change between 'list' and 'board'
ctx (Context): MCP context with access to environment variables
Returns:
str: Confirmation message with updated project details
"""
try:
# Get API token from environment
token = _get_api_token()
if not token:
return "Error: TODOIST_API_TOKEN not configured. Please set your Todoist API token in the environment variables."
json_data = {}
if params.name:
json_data["name"] = params.name
if params.color:
json_data["color"] = params.color
if params.is_favorite is not None:
json_data["is_favorite"] = params.is_favorite
if params.view_style:
json_data["view_style"] = params.view_style
if not json_data:
return "No updates provided. Please specify at least one field to update."
data = await _make_api_request(f"projects/{params.project_id}", token, method="POST", json_data=json_data)
return f"""✅ Project updated successfully!
**Name**: {data['name']}
**ID**: {data['id']}
**Color**: {data.get('color', 'default')}
**View Style**: {data.get('view_style', 'list')}
**Favorite**: {'⭐ Yes' if data.get('is_favorite') else 'No'}"""
except Exception as e:
return _handle_api_error(e)
# ---------- Task Tools ----------
@mcp.tool(
name="todoist_list_tasks",
annotations={
"title": "List Todoist Tasks",
"readOnlyHint": True,
"destructiveHint": False,
"idempotentHint": True,
"openWorldHint": True
}
)
async def todoist_list_tasks(params: ListTasksInput, ctx: Context) -> str:
"""List tasks from Todoist with optional filters.
This tool retrieves active tasks from Todoist with support for filtering
by project, section, label, or using Todoist's filter syntax. Tasks are
displayed with their full details including due dates, priorities, and labels.
Args:
params (ListTasksInput): Validated input parameters containing:
- project_id (Optional[str]): Filter by specific project
- section_id (Optional[str]): Filter by specific section
- label (Optional[str]): Filter by label name
- filter (Optional[str]): Todoist filter query (e.g., 'today', 'p1')
- response_format (ResponseFormat): Output format preference
ctx (Context): MCP context with access to environment variables
Returns:
str: Formatted list of tasks in requested format
Markdown format shows tasks organized by project with:
- Priority indicators and completion status
- Due dates and assignees
- Labels and subtask relationships
JSON format returns complete task objects
Examples:
- Use when: "Show me all my tasks"
- Use when: "What tasks are due today?" (with filter='today')
- Use when: "List tasks in my Work project"
"""
try:
# Get API token from environment
token = _get_api_token()
if not token:
return "Error: TODOIST_API_TOKEN not configured. Please set your Todoist API token in the environment variables."
params_dict = {}
if params.project_id:
params_dict["project_id"] = params.project_id
if params.section_id:
params_dict["section_id"] = params.section_id
if params.label:
params_dict["label"] = params.label
if params.filter:
params_dict["filter"] = params.filter
data = await _make_api_request("tasks", token, params=params_dict)
tasks = data.get("items", [])
if not tasks:
return "No tasks found matching the criteria."
if params.response_format == ResponseFormat.MARKDOWN:
lines = ["# 📋 Todoist Tasks", ""]
if params.filter:
lines.append(f"**Filter**: {params.filter}")
lines.append("")
# Group tasks by project
projects_tasks = {}
for task in tasks:
project_id = task.get('project_id', 'no_project')
if project_id not in projects_tasks:
projects_tasks[project_id] = []
projects_tasks[project_id].append(task)
for project_id, project_tasks in projects_tasks.items():
lines.append(f"## Project: {project_id}")
lines.append("")
for task in sorted(project_tasks, key=lambda x: x.get('order', 0)):
# Format task
priority_emoji = _priority_to_emoji(task.get('priority', 1))
completed = "☑️" if task.get('is_completed') else "⬜"
lines.append(f"### {completed} {priority_emoji} {task['content']}")
lines.append(f"- **ID**: {task['id']}")
if task.get('description'):
lines.append(f"- **Description**: {task['description']}")
due = task.get('due')
if due:
lines.append(f"- **Due**: {_format_due_date(due)}")
if task.get('labels'):
lines.append(f"- **Labels**: {', '.join([f'🏷️ {label}' for label in task['labels']])}")
if task.get('assignee_id'):
lines.append(f"- **Assignee**: User {task['assignee_id']}")
if task.get('parent_id'):
lines.append(f"- **Parent Task**: {task['parent_id']}")
lines.append("")
result = "\n".join(lines)
else:
result = json.dumps(tasks, indent=2)
# Check character limit
if len(result) > CHARACTER_LIMIT:
if params.response_format == ResponseFormat.MARKDOWN:
return f"Response truncated (too many tasks). Found {len(tasks)} tasks. Consider using filters to narrow results."
else:
truncated = tasks[:50] # Limit to 50 tasks
result = json.dumps({
"tasks": truncated,
"truncated": True,
"total_count": len(tasks),
"shown_count": len(truncated),
"message": "Use filters to see specific tasks"
}, indent=2)
return result
except Exception as e:
return _handle_api_error(e)
@mcp.tool(
name="todoist_create_task",
annotations={
"title": "Create Todoist Task",
"readOnlyHint": False,
"destructiveHint": False,
"idempotentHint": False,
"openWorldHint": True
}
)
async def todoist_create_task(params: CreateTaskInput, ctx: Context) -> str:
"""Create a new task in Todoist.
This tool creates a task with full control over all properties including
content, project, due dates, priority, labels, and assignees. Supports
natural language due dates and subtask creation.
Args:
params (CreateTaskInput): Validated input parameters containing:
- content (str): Task title/content
- description (Optional[str]): Detailed task description
- project_id (Optional[str]): Target project ID
- section_id (Optional[str]): Target section ID
- parent_id (Optional[str]): Parent task ID for subtasks
- labels (Optional[List[str]]): Labels to apply
- priority (Optional[TaskPriority]): Priority level (1-4)
- due_string (Optional[str]): Natural language due date
- due_date (Optional[str]): Specific date YYYY-MM-DD
- assignee_id (Optional[str]): User to assign to
ctx (Context): MCP context with access to environment variables
Returns:
str: Confirmation message with created task details
Examples:
- Use when: "Create a task 'Review Q1 report' due tomorrow"
- Use when: "Add a high priority task to my Work project"
- Don't use when: You want to use natural language syntax (use quick_add instead)
"""
try:
# Get API token from environment
token = _get_api_token()
if not token:
return "Error: TODOIST_API_TOKEN not configured. Please set your Todoist API token in the environment variables."
json_data = {"content": params.content}
if params.description is not None:
json_data["description"] = params.description
if params.project_id:
json_data["project_id"] = params.project_id
if params.section_id:
json_data["section_id"] = params.section_id
if params.parent_id:
json_data["parent_id"] = params.parent_id
if params.labels:
json_data["labels"] = params.labels
if params.priority:
json_data["priority"] = params.priority
if params.due_string:
json_data["due_string"] = params.due_string
elif params.due_date:
json_data["due_date"] = params.due_date
if params.assignee_id:
json_data["assignee_id"] = params.assignee_id
data = await _make_api_request("tasks", token, method="POST", json_data=json_data)
priority_emoji = _priority_to_emoji(data.get('priority', 1))
result = f"""✅ Task created successfully!
**Content**: {data['content']}
**ID**: {data['id']}
**Priority**: {priority_emoji} {data.get('priority', 1)}"""
if data.get('description'):
result += f"\n**Description**: {data['description']}"
if data.get('due'):
result += f"\n**Due**: {_format_due_date(data['due'])}"
if data.get('labels'):
result += f"\n**Labels**: {', '.join([f'🏷️ {label}' for label in data['labels']])}"
result += f"\n**URL**: {data.get('url', 'N/A')}"
return result
except Exception as e:
return _handle_api_error(e)
@mcp.tool(
name="todoist_quick_add_task",
annotations={
"title": "Quick Add Todoist Task",
"readOnlyHint": False,
"destructiveHint": False,
"idempotentHint": False,
"openWorldHint": True
}
)
async def todoist_quick_add_task(params: QuickAddTaskInput, ctx: Context) -> str:
"""Quick add a task using Todoist's natural language processing.
This tool uses Todoist's Quick Add feature which understands natural language
for dates, times, projects, labels, and priorities. Simply type as you would
naturally and Todoist will parse the components.
Natural language examples:
- "Buy milk tomorrow at 2pm #Shopping p1"
- "Call John next Monday @Work !!"
- "Review report every Friday #Work"
Syntax guide:
- Projects: @ProjectName or #ProjectName
- Labels: @LabelName
- Priority: p1, p2, p3, p4 or !, !!, !!!
- Due dates: today, tomorrow, next Monday, Jan 15, etc.
Args:
params (QuickAddTaskInput): Validated input parameters containing:
- text (str): Natural language task description
ctx (Context): MCP context with access to environment variables
Returns:
str: Confirmation message with parsed task details
Examples:
- Use when: "Add task 'Meeting tomorrow at 3pm #Work p2'"
- Use when: User wants to quickly add tasks with natural language
- Preferred over create_task when using natural language
"""
try:
# Get API token from environment
token = _get_api_token()
if not token:
return "Error: TODOIST_API_TOKEN not configured. Please set your Todoist API token in the environment variables."
# Note: The Quick Add endpoint is not in REST API v2
# Using the Sync API quick/add endpoint
sync_url = "https://api.todoist.com/sync/v9/quick/add"
headers = {"Authorization": f"Bearer {token}"}
json_data = {"text": params.text}
async with httpx.AsyncClient() as client:
response = await client.post(
sync_url,
headers=headers,
json=json_data,
timeout=30.0
)
response.raise_for_status()
data = response.json()
if not data:
return "Task added successfully using Quick Add!"
priority_emoji = _priority_to_emoji(data.get('priority', 1))
result = f"""✅ Task added successfully using Quick Add!
**Content**: {data.get('content', params.text)}
**ID**: {data.get('id', 'N/A')}
**Priority**: {priority_emoji} {data.get('priority', 1)}"""
if data.get('due'):
result += f"\n**Due**: {_format_due_date(data['due'])}"
if data.get('project_id'):
result += f"\n**Project ID**: {data['project_id']}"
return result
except Exception as e:
return _handle_api_error(e)
@mcp.tool(
name="todoist_update_task",
annotations={
"title": "Update Todoist Task",
"readOnlyHint": False,
"destructiveHint": False,
"idempotentHint": True,
"openWorldHint": True
}
)
async def todoist_update_task(params: UpdateTaskInput, ctx: Context) -> str:
"""Update an existing task in Todoist.
This tool modifies task properties. Only fields that are provided will be
updated, others remain unchanged. Use 'no date' or 'no due date' as due_string
to remove due dates.
Args:
params (UpdateTaskInput): Validated input parameters containing:
- task_id (str): ID of task to update
- content (Optional[str]): New task content
- description (Optional[str]): New description
- labels (Optional[List[str]]): Replace labels
- priority (Optional[TaskPriority]): New priority
- due_string (Optional[str]): New due date (natural language)
- due_date (Optional[str]): New due date (YYYY-MM-DD)
- assignee_id (Optional[str]): New assignee
ctx (Context): MCP context with access to environment variables
Returns:
str: Confirmation message with updated task details
"""
try:
# Get API token from environment
token = _get_api_token()
if not token:
return "Error: TODOIST_API_TOKEN not configured. Please set your Todoist API token in the environment variables."
json_data = {}
if params.content:
json_data["content"] = params.content
if params.description is not None:
json_data["description"] = params.description
if params.labels is not None:
json_data["labels"] = params.labels
if params.priority:
json_data["priority"] = params.priority
if params.due_string:
json_data["due_string"] = params.due_string
elif params.due_date:
json_data["due_date"] = params.due_date
if params.assignee_id is not None:
json_data["assignee_id"] = params.assignee_id
if not json_data:
return "No updates provided. Please specify at least one field to update."
data = await _make_api_request(f"tasks/{params.task_id}", token, method="POST", json_data=json_data)
priority_emoji = _priority_to_emoji(data.get('priority', 1))
result = f"""✅ Task updated successfully!
**Content**: {data['content']}
**ID**: {data['id']}
**Priority**: {priority_emoji} {data.get('priority', 1)}"""
if data.get('description'):
result += f"\n**Description**: {data['description']}"
if data.get('due'):
result += f"\n**Due**: {_format_due_date(data['due'])}"
if data.get('labels'):
result += f"\n**Labels**: {', '.join([f'🏷️ {label}' for label in data['labels']])}"
return result
except Exception as e:
return _handle_api_error(e)
@mcp.tool(
name="todoist_complete_task",
annotations={
"title": "Complete Todoist Task",
"readOnlyHint": False,
"destructiveHint": False,
"idempotentHint": True,
"openWorldHint": True
}
)
async def todoist_complete_task(params: CompleteTaskInput, ctx: Context) -> str:
"""Mark a task as complete in Todoist.
This tool completes a task, moving it to history. For recurring tasks,
this will schedule the next occurrence. Subtasks are also completed.
Args:
params (CompleteTaskInput): Validated input parameters containing:
- task_id (str): ID of task to complete
ctx (Context): MCP context with access to environment variables
Returns:
str: Confirmation message
Examples:
- Use when: "Mark task 123456 as done"
- Use when: "Complete my Buy Milk task"
- Note: For recurring tasks, creates next occurrence automatically
"""
try:
# Get API token from environment
token = _get_api_token()
if not token:
return "Error: TODOIST_API_TOKEN not configured. Please set your Todoist API token in the environment variables."
data = await _make_api_request(f"tasks/{params.task_id}/close", token, method="POST")
return f"""✅ Task completed successfully!
**Task ID**: {params.task_id}
The task has been marked as complete and moved to history.
For recurring tasks, the next occurrence has been scheduled."""
except Exception as e:
return _handle_api_error(e)
# ---------- Section Tools ----------
@mcp.tool(
name="todoist_list_sections",
annotations={
"title": "List Todoist Sections",
"readOnlyHint": True,
"destructiveHint": False,
"idempotentHint": True,
"openWorldHint": True
}
)
async def todoist_list_sections(params: ListSectionsInput, ctx: Context) -> str:
"""List sections in Todoist projects.
This tool retrieves sections which are used to organize tasks within projects.
Sections appear as headers in list view and as columns in board view.
Args:
params (ListSectionsInput): Validated input parameters containing:
- project_id (Optional[str]): Filter by specific project
- response_format (ResponseFormat): Output format preference
ctx (Context): MCP context with access to environment variables
Returns:
str: Formatted list of sections
"""
try:
# Get API token from environment
token = _get_api_token()
if not token:
return "Error: TODOIST_API_TOKEN not configured. Please set your Todoist API token in the environment variables."
params_dict = {}
if params.project_id:
params_dict["project_id"] = params.project_id
data = await _make_api_request("sections", token, params=params_dict)
sections = data.get("items", [])
if not sections:
return "No sections found."
if params.response_format == ResponseFormat.MARKDOWN:
lines = ["# 📑 Todoist Sections", ""]
# Group sections by project
projects_sections = {}
for section in sections:
project_id = section.get('project_id', 'unknown')
if project_id not in projects_sections:
projects_sections[project_id] = []
projects_sections[project_id].append(section)
for project_id, project_sections in projects_sections.items():
lines.append(f"## Project: {project_id}")
for section in sorted(project_sections, key=lambda x: x.get('order', 0)):
lines.append(f"- **{section['name']}** (ID: {section['id']})")
lines.append("")
return "\n".join(lines)
else:
return json.dumps(sections, indent=2)
except Exception as e:
return _handle_api_error(e)
@mcp.tool(
name="todoist_create_section",
annotations={
"title": "Create Todoist Section",
"readOnlyHint": False,
"destructiveHint": False,
"idempotentHint": False,
"openWorldHint": True
}
)
async def todoist_create_section(params: CreateSectionInput, ctx: Context) -> str:
"""Create a new section in a Todoist project.
Sections help organize tasks within projects. They appear as headers in
list view and columns in board view.
Args:
params (CreateSectionInput): Validated input parameters containing:
- name (str): Section name
- project_id (str): Project to add section to
- order (Optional[int]): Position/order in project
ctx (Context): MCP context with access to environment variables
Returns:
str: Confirmation message with section details
"""
try:
# Get API token from environment
token = _get_api_token()
if not token:
return "Error: TODOIST_API_TOKEN not configured. Please set your Todoist API token in the environment variables."
json_data = {
"name": params.name,
"project_id": params.project_id
}
if params.order is not None:
json_data["order"] = params.order
data = await _make_api_request("sections", token, method="POST", json_data=json_data)
return f"""✅ Section created successfully!
**Name**: {data['name']}
**ID**: {data['id']}
**Project ID**: {data['project_id']}
**Order**: {data.get('order', 0)}"""
except Exception as e:
return _handle_api_error(e)
# ---------- Label Tools ----------
@mcp.tool(
name="todoist_list_labels",
annotations={
"title": "List Todoist Labels",
"readOnlyHint": True,
"destructiveHint": False,
"idempotentHint": True,
"openWorldHint": True
}
)
async def todoist_list_labels(params: ListLabelsInput, ctx: Context) -> str:
"""List all personal labels in Todoist.
This tool retrieves personal labels that can be applied to tasks for
categorization and filtering. Does not include shared labels from
collaborators.
Args:
params (ListLabelsInput): Validated input parameters containing:
- response_format (ResponseFormat): Output format preference
ctx (Context): MCP context with access to environment variables
Returns:
str: Formatted list of labels
"""
try:
# Get API token from environment
token = _get_api_token()
if not token:
return "Error: TODOIST_API_TOKEN not configured. Please set your Todoist API token in the environment variables."
data = await _make_api_request("labels", token)
labels = data.get("items", [])
if not labels:
return "No labels found in your account."
if params.response_format == ResponseFormat.MARKDOWN:
lines = ["# 🏷️ Todoist Labels", ""]
# Sort labels: favorites first, then by order
labels_sorted = sorted(labels, key=lambda x: (not x.get('is_favorite', False), x.get('order', 0)))
for label in labels_sorted:
favorite = "⭐" if label.get('is_favorite') else ""
lines.append(f"- **{label['name']}** {favorite}")
lines.append(f" - ID: {label['id']}")
lines.append(f" - Color: {label.get('color', 'default')}")
return "\n".join(lines)
else:
return json.dumps(labels, indent=2)
except Exception as e:
return _handle_api_error(e)
@mcp.tool(
name="todoist_create_label",
annotations={
"title": "Create Todoist Label",
"readOnlyHint": False,
"destructiveHint": False,
"idempotentHint": False,
"openWorldHint": True
}
)
async def todoist_create_label(params: CreateLabelInput, ctx: Context) -> str:
"""Create a new personal label in Todoist.
Labels are used to categorize and filter tasks across projects.
Args:
params (CreateLabelInput): Validated input parameters containing:
- name (str): Label name
- color (Optional[str]): Label color
- is_favorite (Optional[bool]): Mark as favorite
ctx (Context): MCP context with access to environment variables
Returns:
str: Confirmation message with label details
"""
try:
# Get API token from environment
token = _get_api_token()
if not token:
return "Error: TODOIST_API_TOKEN not configured. Please set your Todoist API token in the environment variables."
json_data = {"name": params.name}
if params.color:
json_data["color"] = params.color
if params.is_favorite is not None:
json_data["is_favorite"] = params.is_favorite
data = await _make_api_request("labels", token, method="POST", json_data=json_data)
return f"""✅ Label created successfully!
**Name**: {data['name']}
**ID**: {data['id']}
**Color**: {data.get('color', 'default')}
**Favorite**: {'⭐ Yes' if data.get('is_favorite') else 'No'}"""
except Exception as e:
return _handle_api_error(e)
# ---------- Comment Tools ----------
@mcp.tool(
name="todoist_get_comments",
annotations={
"title": "Get Todoist Comments",
"readOnlyHint": True,
"destructiveHint": False,
"idempotentHint": True,
"openWorldHint": True
}
)
async def todoist_get_comments(params: GetCommentsInput, ctx: Context) -> str:
"""Get comments for a task or project.
This tool retrieves all comments attached to a specific task or project,
useful for viewing discussion history and notes.
Args:
params (GetCommentsInput): Validated input parameters containing:
- task_id (Optional[str]): Task to get comments for
- project_id (Optional[str]): Project to get comments for
- response_format (ResponseFormat): Output format preference
ctx (Context): MCP context with access to environment variables
Returns:
str: Formatted list of comments with timestamps
"""
try:
# Get API token from environment
token = _get_api_token()
if not token:
return "Error: TODOIST_API_TOKEN not configured. Please set your Todoist API token in the environment variables."
params_dict = {}
if params.task_id:
params_dict["task_id"] = params.task_id
elif params.project_id:
params_dict["project_id"] = params.project_id
data = await _make_api_request("comments", token, params=params_dict)
comments = data.get("items", [])
if not comments:
return "No comments found."
if params.response_format == ResponseFormat.MARKDOWN:
lines = ["# 💬 Comments", ""]
for comment in comments:
posted = _format_datetime(comment.get('posted_at', ''))
lines.append(f"## Comment {comment['id']}")
lines.append(f"**Posted**: {posted}")
lines.append(f"**Content**: {comment['content']}")
if comment.get('attachment'):
att = comment['attachment']
lines.append(f"**Attachment**: {att.get('file_name', 'File')} ({att.get('resource_type', 'unknown')})")
lines.append("")
return "\n".join(lines)
else:
return json.dumps(comments, indent=2)
except Exception as e:
return _handle_api_error(e)
@mcp.tool(
name="todoist_add_comment",
annotations={
"title": "Add Todoist Comment",
"readOnlyHint": False,
"destructiveHint": False,
"idempotentHint": False,
"openWorldHint": True
}
)
async def todoist_add_comment(params: AddCommentInput, ctx: Context) -> str:
"""Add a comment to a task or project.
Comments are useful for adding notes, updates, or discussion to tasks
and projects. Supports markdown formatting.
Args:
params (AddCommentInput): Validated input parameters containing:
- content (str): Comment text
- task_id (Optional[str]): Task to comment on
- project_id (Optional[str]): Project to comment on
ctx (Context): MCP context with access to environment variables
Returns:
str: Confirmation message with comment details
"""
try:
# Get API token from environment
token = _get_api_token()
if not token:
return "Error: TODOIST_API_TOKEN not configured. Please set your Todoist API token in the environment variables."
json_data = {"content": params.content}
if params.task_id:
json_data["task_id"] = params.task_id
elif params.project_id:
json_data["project_id"] = params.project_id
data = await _make_api_request("comments", token, method="POST", json_data=json_data)
return f"""✅ Comment added successfully!
**Content**: {data['content']}
**ID**: {data['id']}
**Posted**: {_format_datetime(data.get('posted_at', ''))}
**On**: {'Task ' + str(data.get('task_id')) if data.get('task_id') else 'Project ' + str(data.get('project_id'))}"""
except Exception as e:
return _handle_api_error(e)
if __name__ == "__main__":
mcp.run()