sync_tasks
Sync tasks between Obsidian markdown files and SQLite database to maintain updated task management records within the DeltaTask MCP Server.
Instructions
Sync tasks from Obsidian markdown into SQLite.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- server.py:59-62 (handler)MCP tool handler for 'sync_tasks': decorated with @mcp.tool(), no input parameters, calls TaskService.sync_from_obsidian() and returns its result.@mcp.tool() async def sync_tasks() -> dict[str, Any]: """Sync tasks from Obsidian markdown into SQLite.""" return service.sync_from_obsidian()
- Core implementation of the sync logic in TaskService.sync_from_obsidian(): reads tasks from markdown via markdown_manager, updates/adds to repository, tracks stats, updates views, returns sync result.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)}"}