From 3eb9e6c2724c8d5db9a4623c93ef8c96f1ba5917 Mon Sep 17 00:00:00 2001
From: Dawid Rycerz <dawid@rycerz.xyz>
Date: Fri, 19 Dec 2025 17:39:33 +0100
Subject: feat: implement bulk task addition and deletion in TaskWarrior server

- Added `add_tasks` method to allow adding multiple tasks in a single call, enhancing efficiency and usability.
- Updated `delete_task` method to accept a list of UUIDs for deleting multiple tasks, improving flexibility.
- Implemented validation to ensure that at least one UUID is provided for deletion.
- Enhanced unit tests to cover scenarios for adding multiple tasks, including edge cases for empty lists and single task addition.
- Improved docstrings for clarity, detailing parameters, return values, and usage examples.
---
 .../src/mcp_server_taskwarrior/server.py           | 81 +++++++++++++++++++---
 1 file changed, 70 insertions(+), 11 deletions(-)

(limited to 'servers/taskwarrior/src')

diff --git a/servers/taskwarrior/src/mcp_server_taskwarrior/server.py b/servers/taskwarrior/src/mcp_server_taskwarrior/server.py
index c561d2a..324311f 100644
--- a/servers/taskwarrior/src/mcp_server_taskwarrior/server.py
+++ b/servers/taskwarrior/src/mcp_server_taskwarrior/server.py
@@ -195,6 +195,35 @@ class TaskWarriorServer:
             "output": stdout.strip(),
         }
 
+    async def add_tasks(
+        self,
+        tasks: list[dict[str, Any]],
+    ) -> list[dict[str, Any]]:
+        """Add multiple tasks to TaskWarrior.
+
+        Args:
+            tasks: List of task definitions, each with:
+                - description (required): Task description
+                - project (optional): Project name
+                - priority (optional): H, M, or L
+                - due (optional): Due date
+                - tags (optional): List of tags
+
+        Returns:
+            List of dictionaries with task information including UUIDs
+        """
+        results = []
+        for task in tasks:
+            result = await self.add_task(
+                description=task["description"],
+                project=task.get("project"),
+                priority=task.get("priority"),
+                due=task.get("due"),
+                tags=task.get("tags"),
+            )
+            results.append(result)
+        return results
+
     async def done_task(self, uuids: list[str]) -> dict[str, Any]:
         """Mark one or more tasks as completed.
 
@@ -218,19 +247,25 @@ class TaskWarriorServer:
             "output": stdout.strip(),
         }
 
-    async def delete_task(self, uuid: str) -> dict[str, Any]:
-        """Delete a task.
+    async def delete_task(self, uuids: list[str]) -> dict[str, Any]:
+        """Delete one or more tasks.
 
         Args:
-            uuid: Task UUID (stable identifier)
+            uuids: List of task UUIDs (stable identifiers) to delete
 
         Returns:
-            Dictionary with deletion information
+            Dictionary with deletion information including all UUIDs
+
+        Raises:
+            ValueError: If the uuids list is empty
         """
-        stdout, stderr, _ = await self._run_task_command(uuid, "delete")
+        if not uuids:
+            raise ValueError("At least one UUID is required")
+
+        stdout, stderr, _ = await self._run_task_command(*uuids, "delete")
 
         return {
-            "uuid": uuid,
+            "uuids": uuids,
             "status": "deleted",
             "output": stdout.strip(),
         }
@@ -459,6 +494,30 @@ def create_server(host: str = "127.0.0.1", port: int = 8080) -> FastMCP:
             logger.error(f"Error adding task: {e}")
             return json.dumps({"error": str(e)})
 
+    @mcp.tool()
+    async def add_tasks(
+        tasks: list[dict[str, Any]],
+    ) -> str:
+        """Add multiple tasks to TaskWarrior in one call.
+
+        Args:
+            tasks: List of task definitions. Each task should have:
+                - description (required): Task description
+                - project (optional): Project name
+                - priority (optional): Priority (H, M, or L)
+                - due (optional): Due date
+                - tags (optional): List of tags
+
+        Returns:
+            JSON string with list of created task information
+        """
+        try:
+            results = await taskwarrior.add_tasks(tasks)
+            return json.dumps(results, indent=2)
+        except Exception as e:
+            logger.error(f"Error adding tasks: {e}")
+            return json.dumps({"error": str(e)})
+
     @mcp.tool()
     async def done_task(uuids: list[str]) -> str:
         """Mark one or more tasks as completed.
@@ -477,17 +536,17 @@ def create_server(host: str = "127.0.0.1", port: int = 8080) -> FastMCP:
             return json.dumps({"error": str(e)})
 
     @mcp.tool()
-    async def delete_task(uuid: str) -> str:
-        """Delete a task.
+    async def delete_task(uuids: list[str]) -> str:
+        """Delete one or more tasks.
 
         Args:
-            uuid: Task UUID (stable identifier, not the numeric ID)
+            uuids: List of task UUIDs (stable identifiers, not numeric IDs)
 
         Returns:
-            JSON string with deletion information
+            JSON string with deletion information for all tasks
         """
         try:
-            result = await taskwarrior.delete_task(uuid)
+            result = await taskwarrior.delete_task(uuids)
             return json.dumps(result, indent=2)
         except Exception as e:
             logger.error(f"Error deleting task: {e}")
-- 
cgit v1.2.3