create_task
Create a new task with title, description, urgency, effort estimates, and tags for organizing work in a task management system.
Instructions
Create a new task.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| title | Yes | ||
| description | No | ||
| urgency | No | ||
| effort | No | ||
| tags | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- server.py:32-44 (handler)The core MCP tool handler for 'create_task'. Registered via @mcp.tool(), defines input schema via type hints and defaults, prepares task data with UUID, and delegates to TaskService.add_task() for persistence and file creation.
@mcp.tool() async def create_task(title: str, description: str = "", urgency: int = 1, effort: int = 1, tags: list[str] = []) -> dict[str, Any]: """Create a new task.""" task_data = { "id": str(uuid.uuid4()), # Generate unique ID "title": title, "description": description, "urgency": urgency, "effort": effort, "tags": tags } result = service.add_task(task_data) return result - TaskService.add_task(): Called by the tool handler. Validates effort (Fibonacci) and urgency (1-5), adds task to SQLite database via repository, creates Obsidian markdown file, and updates views/statistics.
def add_task(self, task_data: Dict[str, Any]) -> Dict[str, Any]: """Add a new task and return its details.""" logger.info(f"Adding new task: {task_data.get('title', 'Untitled')}") try: # Ensure task has an ID task_id = self._ensure_id(task_data) # Validate fibonacci sequence for effort valid_efforts = [1, 2, 3, 5, 8, 13, 21] if 'effort' in task_data and task_data['effort'] not in valid_efforts: error_msg = f"Effort must be a Fibonacci number from {valid_efforts}" logger.error(error_msg) raise ValueError(error_msg) # Validate urgency if 'urgency' in task_data and not 1 <= task_data['urgency'] <= 5: error_msg = "Urgency must be between 1 and 5" logger.error(error_msg) raise ValueError(error_msg) try: # Add to database self.repository.add_todo(task_data) logger.info(f"Task {task_id} added to database") except Exception as e: logger.error(f"Failed to add task to database: {e}", exc_info=True) raise try: # Create markdown file self.markdown_manager.create_task_file(task_data) logger.info(f"Markdown file created for task {task_id}") except Exception as e: logger.error(f"Failed to create markdown file for task {task_id}: {e}", exc_info=True) # Continue even if markdown fails - we already have the data in the database try: # Update views self._update_all_views() logger.info("Task views updated") except Exception as e: logger.error(f"Failed to update task views: {e}", exc_info=True) # Continue even if views update fails return {"id": task_id, "message": "Task created successfully"} except Exception as e: logger.error(f"Error adding task: {e}", exc_info=True) return {"error": str(e)} - ObsidianMarkdownManager.create_task_file(): Creates the markdown file in Obsidian vault with frontmatter (task metadata), description content, subtasks/related sections, handles parent links and tags.
def create_task_file(self, task: Dict[str, Any]) -> None: """Create a markdown file for a task.""" logger.info(f"Creating markdown file for task {task.get('id', 'UNKNOWN')}") try: # Validate required fields if "id" not in task: logger.error("Task ID missing when creating task file") raise ValueError("Task ID is required") if "title" not in task: logger.error(f"Task title missing for task {task['id']}") task["title"] = f"Untitled Task {task['id']}" logger.warning(f"Using default title for task {task['id']}") # Prepare frontmatter metadata = { "id": task["id"], "title": task["title"], "created": task.get("created", datetime.now().isoformat()), "updated": task.get("updated", datetime.now().isoformat()), "urgency": task.get("urgency", 1), "effort": task.get("effort", 1), "completed": task.get("completed", False) } if "deadline" in task and task["deadline"]: metadata["deadline"] = task["deadline"] if "parent_id" in task and task["parent_id"]: metadata["parent"] = task["parent_id"] if "tags" in task and task["tags"]: metadata["tags"] = task["tags"] # Create the markdown content content = task.get("description", "") if task.get("description") else "" # Add links to subtasks section content += "\n\n## Subtasks\n\n" # Add links section for related tasks content += "\n\n## Related\n\n" # Add parent link in related section if this is a subtask if "parent_id" in task and task["parent_id"]: parent_id = task["parent_id"] # Get parent title if available parent_title = self._get_parent_title(parent_id) or f"Parent Task {parent_id}" sanitized_parent_title = self._sanitize_filename(parent_title) content += f"- **Parent:** [[tasks/{parent_id} - {sanitized_parent_title}]]\n" # Create the file with frontmatter post = frontmatter.Post(content, **metadata) # Create a filename using ID and title for better readability in graph view sanitized_title = self._sanitize_filename(task['title']) filename = f"{task['id']} - {sanitized_title}.md" file_path = os.path.join(self.vault_path, "tasks", filename) try: # Ensure the directory exists os.makedirs(os.path.dirname(file_path), exist_ok=True) # Write the file with open(file_path, "wb") as f: frontmatter.dump(post, f) logger.info(f"Successfully created task file for {task['id']}") except IOError as e: logger.error(f"Failed to write task file {file_path}: {e}", exc_info=True) raise # Update parent file if this is a subtask if "parent_id" in task and task["parent_id"]: logger.info(f"Updating parent {task['parent_id']} with subtask {task['id']}") self._update_parent_subtasks(task["parent_id"], task["id"], task["title"]) # Update tag files if "tags" in task and task["tags"]: logger.info(f"Updating {len(task['tags'])} tag files for task {task['id']}") self._update_tag_files(task["tags"], task["id"], task["title"]) except Exception as e: logger.error(f"Error creating task file: {e}", exc_info=True) raise