sync_tasks
Sync tasks from Obsidian markdown files into SQLite database for centralized task management.
Instructions
Sync tasks from Obsidian markdown into SQLite.
Tool call template
This tool declares no required parameters. Add optional parameters from the input schema as needed.
{
"id": 1,
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"arguments": {},
"name": "sync_tasks"
}
}Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Output 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)}"}