"""MCP tool for adding tasks to the todo list. [Task]: T013, T031 [From]: specs/004-ai-chatbot/tasks.md, specs/007-intermediate-todo-features/tasks.md (US2) This tool allows the AI agent to create tasks on behalf of users through natural language conversations. Now supports tag extraction from natural language patterns. """ from typing import Optional, Any, List from uuid import UUID, uuid4 from datetime import datetime, timedelta from models.task import Task from core.database import engine from sqlmodel import Session # Import tag extraction service [T029, T031] import sys import os sys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(__file__)))) from services.nlp_service import extract_tags_from_task_data, normalize_tag_name # Tool metadata for MCP registration tool_metadata = { "name": "add_task", "description": """Create a new task in the user's todo list. Use this tool when the user wants to create, add, or remind themselves about a task. The task will be associated with their user account and persist across conversations. Parameters: - title (required): Brief task title (max 255 characters) - description (optional): Detailed task description (max 2000 characters) - due_date (optional): When the task is due (ISO 8601 date string or relative like 'tomorrow', 'next week') - priority (optional): Task priority - 'low', 'medium', or 'high' (default: 'medium') - tags (optional): List of tag names for categorization (e.g., ["work", "urgent"]) Natural Language Tag Support [T031]: - "tagged with X" or "tags X" → extracts tag X - "add tag X" or "with tag X" → extracts tag X - "#tagname" → extracts hashtag as tag - "labeled X" → extracts tag X Returns: Created task details including ID, title, and confirmation. """, "inputSchema": { "type": "object", "properties": { "user_id": { "type": "string", "description": "User ID (UUID) who owns this task" }, "title": { "type": "string", "description": "Task title (brief description)", "maxLength": 255 }, "description": { "type": "string", "description": "Detailed task description", "maxLength": 2000 }, "due_date": { "type": "string", "description": "Due date in ISO 8601 format (e.g., '2025-01-15') or relative terms" }, "priority": { "type": "string", "enum": ["low", "medium", "high"], "description": "Task priority level" }, "tags": { "type": "array", "items": {"type": "string"}, "description": "List of tag names for categorization" } }, "required": ["user_id", "title"] } } async def add_task( user_id: str, title: str, description: Optional[str] = None, due_date: Optional[str] = None, priority: Optional[str] = None, tags: Optional[List[str]] = None ) -> dict[str, Any]: """Create a new task for the user. [From]: specs/004-ai-chatbot/spec.md - US1 [Task]: T031 - Integrate tag extraction for natural language Args: user_id: User ID (UUID string) who owns this task title: Brief task title description: Optional detailed description due_date: Optional due date (ISO 8601 or relative) priority: Optional priority level (low/medium/high) tags: Optional list of tag names Returns: Dictionary with created task details Raises: ValueError: If validation fails ValidationError: If task constraints violated """ from core.validators import validate_task_title, validate_task_description # Validate inputs validated_title = validate_task_title(title) validated_description = validate_task_description(description) if description else None # Parse and validate due date if provided parsed_due_date = None if due_date: parsed_due_date = _parse_due_date(due_date) # Normalize priority normalized_priority = _normalize_priority(priority) # [T031] Extract tags from natural language in title and description extracted_tags = extract_tags_from_task_data(validated_title, validated_description) # Normalize extracted tags normalized_extracted_tags = [normalize_tag_name(tag) for tag in extracted_tags] # Combine provided tags with extracted tags, removing duplicates all_tags = set(normalized_extracted_tags) if tags: # Normalize provided tags normalized_provided_tags = [normalize_tag_name(tag) for tag in tags] all_tags.update(normalized_provided_tags) final_tags = sorted(list(all_tags)) if all_tags else [] # Get database session (synchronous) with Session(engine) as db: try: # Create task instance task = Task( id=uuid4(), user_id=UUID(user_id), title=validated_title, description=validated_description, due_date=parsed_due_date, priority=normalized_priority, tags=final_tags, completed=False, created_at=datetime.utcnow(), updated_at=datetime.utcnow() ) # Save to database db.add(task) db.commit() db.refresh(task) # Return success response return { "success": True, "task": { "id": str(task.id), "title": task.title, "description": task.description, "due_date": task.due_date.isoformat() if task.due_date else None, "priority": task.priority, "tags": task.tags, "completed": task.completed, "created_at": task.created_at.isoformat() }, "message": f"✅ Task created: {task.title}" + (f" (tags: {', '.join(final_tags)})" if final_tags else "") } except Exception as e: db.rollback() raise ValueError(f"Failed to create task: {str(e)}") def _parse_due_date(due_date_str: str) -> Optional[datetime]: """Parse due date from ISO 8601 or natural language. [From]: specs/004-ai-chatbot/plan.md - Natural Language Processing Supports: - ISO 8601: "2025-01-15", "2025-01-15T10:00:00Z" - Relative: "today", "tomorrow", "next week", "in 3 days" Args: due_date_str: Date string to parse Returns: Parsed datetime or None if parsing fails Raises: ValueError: If date format is invalid """ from datetime import datetime import re # Try ISO 8601 format first try: # Handle YYYY-MM-DD format if re.match(r"^\d{4}-\d{2}-\d{2}$", due_date_str): return datetime.fromisoformat(due_date_str) # Handle full ISO 8601 with time if "T" in due_date_str: return datetime.fromisoformat(due_date_str.replace("Z", "+00:00")) except ValueError: pass # Fall through to natural language parsing # Natural language parsing (simplified) due_date_str = due_date_str.lower().strip() today = datetime.utcnow().replace(hour=23, minute=59, second=59, microsecond=999999) if due_date_str == "today": return today elif due_date_str == "tomorrow": return today + timedelta(days=1) elif due_date_str == "next week": return today + timedelta(weeks=1) elif due_date_str.startswith("in "): # Parse "in X days/weeks" match = re.match(r"in (\d+) (day|days|week|weeks)", due_date_str) if match: amount = int(match.group(1)) unit = match.group(2) if unit.startswith("day"): return today + timedelta(days=amount) elif unit.startswith("week"): return today + timedelta(weeks=amount) # If parsing fails, return None and let AI agent ask for clarification return None def _normalize_priority(priority: Optional[str]) -> str: """Normalize priority string to valid values. [From]: models/task.py - Task model [Task]: T009-T011 - Priority extraction from natural language Args: priority: Priority string to normalize Returns: Normalized priority: "low", "medium", or "high" Raises: ValueError: If priority is invalid """ if not priority: return "medium" # Default priority priority_normalized = priority.lower().strip() # Direct matches if priority_normalized in ["low", "medium", "high"]: return priority_normalized # Enhanced priority mapping from natural language patterns # [Task]: T011 - Integrate priority extraction in MCP tools priority_map_high = { # Explicit high priority keywords "urgent", "asap", "important", "critical", "emergency", "immediate", "high", "priority", "top", "now", "today", "deadline", "crucial", # Numeric mappings "3", "high priority", "very important", "must do" } priority_map_low = { # Explicit low priority keywords "low", "later", "whenever", "optional", "nice to have", "someday", "eventually", "routine", "normal", "regular", "backlog", # Numeric mappings "1", "low priority", "no rush", "can wait" } priority_map_medium = { "2", "medium", "normal", "standard", "default", "moderate" } # Check high priority patterns if priority_normalized in priority_map_high or any( keyword in priority_normalized for keyword in ["urgent", "asap", "critical", "deadline", "today"] ): return "high" # Check low priority patterns if priority_normalized in priority_map_low or any( keyword in priority_normalized for keyword in ["whenever", "later", "optional", "someday"] ): return "low" # Default to medium return "medium" # Register tool with MCP server def register_tool(mcp_server: Any) -> None: """Register this tool with the MCP server. [From]: backend/mcp_server/server.py Args: mcp_server: MCP server instance """ mcp_server.tool( name=tool_metadata["name"], description=tool_metadata["description"] )(add_task)