Dooray MCP Server

"""Tasks tool for managing Dooray tasks."""

import json
import logging
import os
from typing import Any, Dict, Optional

logger = logging.getLogger(__name__)

class TasksTool:
    """Tool for managing Dooray tasks."""
    
    def __init__(self, dooray_client):
        """Initialize with Dooray client."""
        self.client = dooray_client
    
    async def handle(self, arguments: Dict[str, Any]) -> str:
        """Handle tasks tool requests.
        
        Args:
            arguments: Tool arguments containing action and parameters
            
        Returns:
            JSON string with results
        """
        action = arguments.get("action")
        if not action:
            return json.dumps({"error": "Action parameter is required"})
        
        try:
            if action == "list":
                return await self._list_tasks(arguments)
            elif action == "get":
                return await self._get_task(arguments)
            elif action == "create":
                return await self._create_task(arguments)
            elif action == "update":
                return await self._update_task(arguments)
            elif action == "delete":
                return await self._delete_task(arguments)
            elif action == "change_status":
                return await self._change_status(arguments)
            elif action == "assign":
                return await self._assign_task(arguments)
            else:
                return json.dumps({"error": f"Unknown action: {action}"})
                
        except Exception as e:
            logger.error(f"Error in tasks tool: {e}")
            return json.dumps({"error": str(e)})
    
    def _resolve_project_id(self, arguments: Dict[str, Any]) -> Optional[str]:
        """Resolve project ID from arguments or defaults."""
        project_id = arguments.get("projectId") or self.client.project_id or os.getenv("DOORAY_DEFAULT_PROJECT_ID")
        if project_id:
            return str(project_id)
        return None

    def _ensure_task_id(self, arguments: Dict[str, Any]) -> Optional[str]:
        task_id = arguments.get("taskId")
        if task_id is None:
            return None
        return str(task_id)

    def _ensure_assignee_id(self, arguments: Dict[str, Any]) -> Optional[str]:
        assignee_id = arguments.get("assigneeId")
        if assignee_id is None:
            return None
        return str(assignee_id)

    async def _list_tasks(self, arguments: Dict[str, Any]) -> str:
        """List tasks in a project."""
        project_id = self._resolve_project_id(arguments)
        if not project_id:
            return json.dumps({"error": "projectId is required for list action (set DOORAY_DEFAULT_PROJECT_ID or provide projectId)"})
        
        # Build query parameters
        params = {}
        if arguments.get("status"):
            params["workflowClass"] = arguments["status"]
        assignee_id = self._ensure_assignee_id(arguments)
        if assignee_id:
            params["assigneeId"] = assignee_id
        
        result = await self.client.list_tasks(project_id, **params)
        return json.dumps(result, ensure_ascii=False)
    
    async def _get_task(self, arguments: Dict[str, Any]) -> str:
        """Get a specific task."""
        project_id = self._resolve_project_id(arguments)
        task_id = self._ensure_task_id(arguments)

        if not project_id or not task_id:
            return json.dumps({"error": "projectId and taskId are required for get action"})

        result = await self.client.get_task(project_id, task_id)
        return json.dumps(result, ensure_ascii=False)
    
    async def _create_task(self, arguments: Dict[str, Any]) -> str:
        """Create a new task."""
        project_id = self._resolve_project_id(arguments)
        title = arguments.get("title")

        if not project_id or not title:
            return json.dumps({"error": "projectId and title are required for create action"})
        
        # Build task data
        task_data = {
            "subject": title,
            "body": {
                "mimeType": "text/x-markdown",
                "content": arguments.get("description", "")
            }
        }
        
        assignee_id = self._ensure_assignee_id(arguments)
        if assignee_id:
            task_data["users"] = [{"member": {"id": assignee_id}}]
        
        if arguments.get("priority"):
            task_data["priority"] = arguments["priority"]
        
        if arguments.get("status"):
            task_data["workflowClass"] = arguments["status"]
        
        result = await self.client.create_task(project_id, task_data)
        return json.dumps(result, ensure_ascii=False)
    
    async def _update_task(self, arguments: Dict[str, Any]) -> str:
        """Update an existing task."""
        project_id = self._resolve_project_id(arguments)
        task_id = self._ensure_task_id(arguments)

        if not project_id or not task_id:
            return json.dumps({"error": "projectId and taskId are required for update action"})
        
        # Build update data
        task_data = {}
        
        if arguments.get("title"):
            task_data["subject"] = arguments["title"]
        
        if arguments.get("description"):
            task_data["body"] = {
                "mimeType": "text/x-markdown",
                "content": arguments["description"]
            }
        
        assignee_id = self._ensure_assignee_id(arguments)
        if assignee_id:
            task_data["users"] = [{"member": {"id": assignee_id}}]

        if arguments.get("priority"):
            task_data["priority"] = arguments["priority"]

        if arguments.get("status"):
            task_data["workflowClass"] = arguments["status"]
        
        result = await self.client.update_task(project_id, task_id, task_data)
        return json.dumps(result, ensure_ascii=False)
    
    async def _delete_task(self, arguments: Dict[str, Any]) -> str:
        """Delete a task."""
        project_id = self._resolve_project_id(arguments)
        task_id = self._ensure_task_id(arguments)

        if not project_id or not task_id:
            return json.dumps({"error": "projectId and taskId are required for delete action"})

        result = await self.client.delete_task(project_id, task_id)
        return json.dumps({"success": True, "message": "Task deleted successfully"})
    
    async def _change_status(self, arguments: Dict[str, Any]) -> str:
        """Change task status."""
        project_id = self._resolve_project_id(arguments)
        task_id = self._ensure_task_id(arguments)
        status = arguments.get("status")

        workflow_id_arg = arguments.get("workflowId")

        if not project_id or not task_id or (status is None and workflow_id_arg is None):
            return json.dumps({"error": "projectId, taskId, and either status or workflowId are required for change_status action"})

        workflow_id = str(workflow_id_arg) if workflow_id_arg is not None else None

        if not workflow_id and status is not None:
            workflow_id = await self._resolve_workflow_id(project_id, status)

        if not workflow_id:
            return json.dumps({"error": "Unable to resolve workflow. Provide workflowId or use status/workflow name."}, ensure_ascii=False)

        result = await self.client.set_task_workflow(project_id, task_id, workflow_id)
        return json.dumps(result, ensure_ascii=False)
    
    async def _assign_task(self, arguments: Dict[str, Any]) -> str:
        """Assign task to a member."""
        project_id = self._resolve_project_id(arguments)
        task_id = self._ensure_task_id(arguments)
        assignee_id = self._ensure_assignee_id(arguments)

        if not task_id or not assignee_id:
            return json.dumps({"error": "taskId and assigneeId are required for assign action"})
        
        task_data = {"users": [{"member": {"id": assignee_id}}]}
        result = await self.client.update_task(project_id, task_id, task_data)
        return json.dumps(result, ensure_ascii=False)

    async def _resolve_workflow_id(self, project_id: str, identifier: str) -> Optional[str]:
        if identifier is None:
            return None

        identifier_text = str(identifier).strip()
        if not identifier_text:
            return None

        workflows = await self.client.list_workflows(project_id)
        items = workflows.get("result") or []

        normalized = identifier_text.lower()
        class_matches: list[tuple[int, str]] = []

        for item in items:
            workflow_id = str(item.get("id")) if item.get("id") is not None else None
            if not workflow_id:
                continue

            if workflow_id == identifier_text:
                return workflow_id

            name = item.get("name")
            if isinstance(name, str) and name.lower() == normalized:
                return workflow_id

            for localized in item.get("names", []) or []:
                localized_name = localized.get("name")
                if isinstance(localized_name, str) and localized_name.lower() == normalized:
                    return workflow_id

            workflow_class = item.get("class")
            if isinstance(workflow_class, str) and workflow_class.lower() == normalized:
                order_value = item.get("order")
                try:
                    order = int(order_value)
                except (TypeError, ValueError):
                    order = 0
                class_matches.append((order, workflow_id))

        if class_matches:
            class_matches.sort(key=lambda pair: pair[0])
            return class_matches[0][1]

        return None