DeltaTask MCP Server

Instructions

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()

• deltatask/services/task_service.py:285-344 (helper)

  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)}"}