sync_tasks
Sync tasks from Obsidian markdown files into SQLite database for centralized task management.
Instructions
Sync tasks from Obsidian markdown into SQLite.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- server.py:59-62 (handler)The main handler function for the 'sync_tasks' MCP tool. It is registered via the @mcp.tool() decorator and delegates the sync logic to TaskService.sync_from_obsidian().
@mcp.tool() async def sync_tasks() -> dict[str, Any]: """Sync tasks from Obsidian markdown into SQLite.""" return service.sync_from_obsidian() - The core helper method in TaskService that implements the synchronization logic from Obsidian markdown files to the SQLite database, handling updates, additions, and error tracking.
def sync_from_obsidian(self) -> Dict[str, Any]: """Sync changes from Obsidian markdown files back to the database.""" logger.info("Starting sync from Obsidian to database") try: # Get all tasks from markdown files markdown_tasks = self.markdown_manager.sync_from_markdown() if not markdown_tasks: logger.info("No markdown tasks found for syncing") return {"message": "No tasks found for syncing", "count": 0} # Track statistics updated_count = 0 error_count = 0 # Process each task for task_data in markdown_tasks: try: task_id = task_data["id"] # Check if task exists existing_task = self.repository.get_todo_by_id(task_id) if existing_task: # Update existing task success = self.repository.update_todo(task_id, task_data) if success: logger.info(f"Updated task {task_id} from markdown") updated_count += 1 else: logger.error(f"Failed to update task {task_id} from markdown") error_count += 1 else: # Add new task self.repository.add_todo(task_data) logger.info(f"Added new task {task_id} from markdown") updated_count += 1 except Exception as e: logger.error(f"Error syncing task {task_data.get('id', 'unknown')}: {e}", exc_info=True) error_count += 1 # Update all views to reflect changes try: self._update_all_views() except Exception as e: logger.error(f"Error updating views after sync: {e}", exc_info=True) result = { "message": "Sync completed", "updated": updated_count, "errors": error_count, "total": len(markdown_tasks) } logger.info(f"Sync completed: {result}") return result except Exception as e: logger.error(f"Error in Obsidian sync process: {e}", exc_info=True) return {"error": f"Sync failed: {str(e)}"}