diff options
Diffstat (limited to 'servers/taskwarrior')
| -rw-r--r-- | servers/taskwarrior/src/mcp_server_taskwarrior/server.py | 134 | ||||
| -rw-r--r-- | servers/taskwarrior/tests/test_server.py | 165 |
2 files changed, 299 insertions, 0 deletions
diff --git a/servers/taskwarrior/src/mcp_server_taskwarrior/server.py b/servers/taskwarrior/src/mcp_server_taskwarrior/server.py index 2f66dac..c561d2a 100644 --- a/servers/taskwarrior/src/mcp_server_taskwarrior/server.py +++ b/servers/taskwarrior/src/mcp_server_taskwarrior/server.py @@ -235,6 +235,102 @@ class TaskWarriorServer: "output": stdout.strip(), } + async def modify_task( + self, + filter_expr: str, + project: str | None = None, + priority: str | None = None, + due: str | None = None, + description: str | None = None, + tags: list[str] | None = None, + ) -> dict[str, Any]: + """Modify one or more tasks matching the filter expression. + + Args: + filter_expr: Filter expression to identify tasks to modify + (e.g., "uuid:abc123", "+work", "project:Home status:pending") + project: New project name (use empty string "" to clear project) + priority: New priority (H, M, L, or empty string "" to clear) + due: New due date (ISO format or TaskWarrior format like "tomorrow", + or empty string "" to clear) + description: New description text + tags: List of tags with +/- prefixes (e.g., ["+newtag", "-oldtag"]) + + Returns: + Dictionary with modification information + + Raises: + ValueError: If no modification parameters provided, invalid priority, + or invalid tag format + """ + # Validate that at least one modification parameter is provided + if not any( + [ + project is not None, + priority is not None, + due is not None, + description is not None, + tags, + ] + ): + raise ValueError( + "At least one modification parameter must be provided " + "(project, priority, due, description, or tags)" + ) + + # Validate priority if provided + if priority is not None: + priority_upper = priority.upper() if priority else "" + if priority_upper and priority_upper not in ["H", "M", "L"]: + raise ValueError(f"Priority must be H, M, or L, got: {priority}") + + # Validate tags format if provided + if tags: + for tag in tags: + if not tag.startswith(("+", "-")): + raise ValueError(f"Tags must start with + or -, got: {tag}") + + # Build command: task <filter_expr> modify <modifications> + args = [filter_expr, "modify"] + + if project is not None: + if project == "": + args.append("project:") + else: + args.append(f"project:{project}") + + if priority is not None: + if priority == "": + args.append("priority:") + else: + args.append(f"priority:{priority.upper()}") + + if due is not None: + if due == "": + args.append("due:") + else: + args.append(f"due:{due}") + + if description is not None: + args.append(f"description:{description}") + + if tags: + args.extend(tags) + + stdout, stderr, _ = await self._run_task_command(*args) + + return { + "filter_expr": filter_expr, + "modifications": { + "project": project if project is not None else None, + "priority": priority.upper() if priority and priority != "" else None, + "due": due if due != "" else None, + "description": description, + "tags": tags, + }, + "output": stdout.strip(), + } + async def manage_context( self, action: str, name: str | None = None ) -> dict[str, Any]: @@ -398,6 +494,44 @@ def create_server(host: str = "127.0.0.1", port: int = 8080) -> FastMCP: return json.dumps({"error": str(e)}) @mcp.tool() + async def modify_task( + filter_expr: str, + project: str | None = None, + priority: str | None = None, + due: str | None = None, + description: str | None = None, + tags: list[str] | None = None, + ) -> str: + """Modify one or more tasks matching the filter expression. + + Args: + filter_expr: Filter expression to identify tasks to modify + (e.g., "uuid:abc123", "+work", "project:Home status:pending") + project: New project name (use empty string "" to clear project) + priority: New priority (H, M, L, or empty string "" to clear) + due: New due date (ISO format or TaskWarrior format like "tomorrow", + or empty string "" to clear) + description: New description text + tags: List of tags with +/- prefixes (e.g., ["+newtag", "-oldtag"]) + + Returns: + JSON string with modification information + """ + try: + result = await taskwarrior.modify_task( + filter_expr=filter_expr, + project=project, + priority=priority, + due=due, + description=description, + tags=tags, + ) + return json.dumps(result, indent=2) + except Exception as e: + logger.error(f"Error modifying task: {e}") + return json.dumps({"error": str(e)}) + + @mcp.tool() async def context(action: str, name: str | None = None) -> str: """Manage TaskWarrior contexts. diff --git a/servers/taskwarrior/tests/test_server.py b/servers/taskwarrior/tests/test_server.py index 97f7268..601a535 100644 --- a/servers/taskwarrior/tests/test_server.py +++ b/servers/taskwarrior/tests/test_server.py @@ -212,3 +212,168 @@ class TestTaskWarriorServer: await taskwarrior_server.manage_context("set") assert "Context name is required" in str(exc_info.value) + + @pytest.mark.asyncio + async def test_modify_task_project(self, taskwarrior_server): + """Test modifying a task's project.""" + mock_output = "Modified 1 task." + + with patch.object( + taskwarrior_server, '_run_task_command', + new_callable=AsyncMock, + return_value=(mock_output, '', 0) + ): + result = await taskwarrior_server.modify_task( + "uuid:test-uuid-123", + project="NewProject" + ) + assert result["filter_expr"] == "uuid:test-uuid-123" + assert result["modifications"]["project"] == "NewProject" + + @pytest.mark.asyncio + async def test_modify_task_priority(self, taskwarrior_server): + """Test modifying a task's priority.""" + mock_output = "Modified 1 task." + + with patch.object( + taskwarrior_server, '_run_task_command', + new_callable=AsyncMock, + return_value=(mock_output, '', 0) + ): + result = await taskwarrior_server.modify_task( + "uuid:test-uuid-123", + priority="H" + ) + assert result["modifications"]["priority"] == "H" + + @pytest.mark.asyncio + async def test_modify_task_clear_priority(self, taskwarrior_server): + """Test clearing a task's priority.""" + mock_output = "Modified 1 task." + + with patch.object( + taskwarrior_server, '_run_task_command', + new_callable=AsyncMock, + return_value=(mock_output, '', 0) + ): + result = await taskwarrior_server.modify_task( + "uuid:test-uuid-123", + priority="" + ) + assert result["modifications"]["priority"] is None + + @pytest.mark.asyncio + async def test_modify_task_due(self, taskwarrior_server): + """Test modifying a task's due date.""" + mock_output = "Modified 1 task." + + with patch.object( + taskwarrior_server, '_run_task_command', + new_callable=AsyncMock, + return_value=(mock_output, '', 0) + ): + result = await taskwarrior_server.modify_task( + "uuid:test-uuid-123", + due="tomorrow" + ) + assert result["modifications"]["due"] == "tomorrow" + + @pytest.mark.asyncio + async def test_modify_task_description(self, taskwarrior_server): + """Test modifying a task's description.""" + mock_output = "Modified 1 task." + + with patch.object( + taskwarrior_server, '_run_task_command', + new_callable=AsyncMock, + return_value=(mock_output, '', 0) + ): + result = await taskwarrior_server.modify_task( + "uuid:test-uuid-123", + description="New description" + ) + assert result["modifications"]["description"] == "New description" + + @pytest.mark.asyncio + async def test_modify_task_tags(self, taskwarrior_server): + """Test modifying a task's tags.""" + mock_output = "Modified 1 task." + + with patch.object( + taskwarrior_server, '_run_task_command', + new_callable=AsyncMock, + return_value=(mock_output, '', 0) + ): + result = await taskwarrior_server.modify_task( + "uuid:test-uuid-123", + tags=["+urgent", "-later"] + ) + assert result["modifications"]["tags"] == ["+urgent", "-later"] + + @pytest.mark.asyncio + async def test_modify_task_multiple_fields(self, taskwarrior_server): + """Test modifying multiple fields at once.""" + mock_output = "Modified 1 task." + + with patch.object( + taskwarrior_server, '_run_task_command', + new_callable=AsyncMock, + return_value=(mock_output, '', 0) + ): + result = await taskwarrior_server.modify_task( + "+work", + project="Work", + priority="H", + tags=["+urgent"] + ) + assert result["filter_expr"] == "+work" + assert result["modifications"]["project"] == "Work" + assert result["modifications"]["priority"] == "H" + assert result["modifications"]["tags"] == ["+urgent"] + + @pytest.mark.asyncio + async def test_modify_task_no_modifications(self, taskwarrior_server): + """Test that modifying without any parameters raises ValueError.""" + with pytest.raises(ValueError) as exc_info: + await taskwarrior_server.modify_task("uuid:test-uuid-123") + + assert "At least one modification parameter must be provided" in str(exc_info.value) + + @pytest.mark.asyncio + async def test_modify_task_invalid_priority(self, taskwarrior_server): + """Test that invalid priority raises ValueError.""" + with pytest.raises(ValueError) as exc_info: + await taskwarrior_server.modify_task( + "uuid:test-uuid-123", + priority="X" + ) + + assert "Priority must be H, M, or L" in str(exc_info.value) + + @pytest.mark.asyncio + async def test_modify_task_invalid_tag_format(self, taskwarrior_server): + """Test that tags without +/- prefix raise ValueError.""" + with pytest.raises(ValueError) as exc_info: + await taskwarrior_server.modify_task( + "uuid:test-uuid-123", + tags=["invalidtag"] + ) + + assert "Tags must start with + or -" in str(exc_info.value) + + @pytest.mark.asyncio + async def test_modify_task_with_filter_expression(self, taskwarrior_server): + """Test modifying tasks with complex filter expression.""" + mock_output = "Modified 2 tasks." + + with patch.object( + taskwarrior_server, '_run_task_command', + new_callable=AsyncMock, + return_value=(mock_output, '', 0) + ): + result = await taskwarrior_server.modify_task( + "project:Home status:pending", + priority="M" + ) + assert result["filter_expr"] == "project:Home status:pending" + assert result["modifications"]["priority"] == "M" |