diff --git a/.gitignore b/.gitignore
index d512c0ee65..82bf40915d 100644
--- a/.gitignore
+++ b/.gitignore
@@ -24,8 +24,8 @@ tests/files/org/openml/test/datasets/1/
 tests/files/org/openml/test/datasets/2/features.xml.pkl
 tests/files/org/openml/test/datasets/2/qualities.xml.pkl
 tests/files/org/openml/test/locks/
-tests/files/org/openml/test/tasks/1/datasplits.pkl.py3
-tests/files/org/openml/test/tasks/1882/datasplits.pkl.py3
+tests/files/org/openml/test/api_splits/get/1/Task_1_splits.arff/body.pkl.py3
+tests/files/org/openml/test/api_splits/get/1882/Task_1882_splits.arff/body.pkl.py3
 
 # Distribution / packaging
 
diff --git a/openml/_api/resources/base/resources.py b/openml/_api/resources/base/resources.py
index 45b3f910b5..bb499434b8 100644
--- a/openml/_api/resources/base/resources.py
+++ b/openml/_api/resources/base/resources.py
@@ -17,11 +17,11 @@
     from openml.datasets.data_feature import OpenMLDataFeature
     from openml.datasets.dataset import OpenMLDataset
     from openml.estimation_procedures import OpenMLEstimationProcedure
-    from openml.evaluations import OpenMLEvaluation
+    from openml.evaluations.evaluation import OpenMLEvaluation
     from openml.flows.flow import OpenMLFlow
     from openml.runs.run import OpenMLRun
     from openml.setups.setup import OpenMLSetup
-    from openml.tasks.task import TaskType
+    from openml.tasks.task import OpenMLTask, TaskType
 
 
 class DatasetAPI(ResourceAPI):
@@ -146,6 +146,51 @@ class TaskAPI(ResourceAPI):
 
     resource_type: ResourceType = ResourceType.TASK
 
+    @abstractmethod
+    def get(
+        self,
+        task_id: int,
+    ) -> OpenMLTask:
+        """
+        API v1:
+            GET /task/{task_id}
+
+        API v2:
+            GET /tasks/{task_id}
+        """
+        ...
+
+    @abstractmethod
+    def download_split(self, url: str, *, refresh_cache: bool = False) -> Path:
+        """Download a task's split file and return its path in the cache."""
+        ...
+
+    # Task listing (V1 only)
+    @abstractmethod
+    def list(
+        self,
+        limit: int,
+        offset: int,
+        task_type: TaskType | int | None = None,
+        *,
+        estimation_procedures: builtins.list[dict[str, Any]],
+        **kwargs: Any,
+    ) -> pd.DataFrame:
+        """
+        List tasks with filters.
+
+        API v1:
+            GET /task/list
+
+        API v2:
+            Not available.
+
+        Returns
+        -------
+        pandas.DataFrame
+        """
+        ...
+
 
 class EvaluationMeasureAPI(ResourceAPI):
     """Abstract API interface for evaluation measure resources."""
diff --git a/openml/_api/resources/task.py b/openml/_api/resources/task.py
index 1f62aa3f30..7a3372564d 100644
--- a/openml/_api/resources/task.py
+++ b/openml/_api/resources/task.py
@@ -1,11 +1,396 @@
 from __future__ import annotations
 
+import warnings
+from typing import TYPE_CHECKING, Any
+
+import pandas as pd
+import xmltodict
+
+from openml.tasks.task import (
+    OpenMLClassificationTask,
+    OpenMLClusteringTask,
+    OpenMLLearningCurveTask,
+    OpenMLRegressionTask,
+    OpenMLTask,
+    TaskType,
+)
+
 from .base import ResourceV1API, ResourceV2API, TaskAPI
 
+if TYPE_CHECKING:
+    import builtins
+    from pathlib import Path
+
 
 class TaskV1API(ResourceV1API, TaskAPI):
-    """Version 1 API implementation for task resources."""
+    def get(self, task_id: int) -> OpenMLTask:
+        """Download OpenML task for a given task ID.
+
+        Downloads the task representation.
+
+        Parameters
+        ----------
+        task_id : int
+            The OpenML task id of the task to download.
+
+        Returns
+        -------
+        task: OpenMLTask
+        """
+        response = self._http.get(f"task/{task_id}", enable_cache=True)
+        return self._create_task_from_xml(response.text)
+
+    def download_split(self, url: str, *, refresh_cache: bool = False) -> Path:
+        """Download a task's split file and return its path in the cache.
+
+        Parameters
+        ----------
+        url : str
+            URL of the split file.
+        refresh_cache : bool, optional
+            Whether to replace the file if it is already cached.
+
+        Returns
+        -------
+        Path
+        """
+        self._http.get(url, enable_cache=True, refresh_cache=refresh_cache)
+        return self._http.cache_path_from_url(url)
+
+    def list(
+        self,
+        limit: int,
+        offset: int,
+        task_type: TaskType | int | None = None,
+        *,
+        estimation_procedures: builtins.list[dict[str, Any]],
+        **kwargs: Any,
+    ) -> pd.DataFrame:
+        """
+        Perform the api call to return a number of tasks having the given filters.
+
+        Parameters
+        ----------
+        Filter task_type is separated from the other filters because
+        it is used as task_type in the task description, but it is named
+        type when used as a filter in list tasks call.
+        limit: int
+        offset: int
+        task_type : TaskType, optional
+            Refers to the type of task.
+        estimation_procedures : list of dict
+            All estimation procedures on OpenML, used to resolve the procedure
+            ids returned by the listing into names.
+        kwargs: dict, optional
+            Legal filter operators: tag, task_id (list), data_tag, status, limit,
+            offset, data_id, data_name, number_instances, number_features,
+            number_classes, number_missing_values.
+
+        Returns
+        -------
+        dataframe
+        """
+        api_call = self._build_url(limit, offset, task_type, kwargs)
+        return self._parse_list_xml(api_call, estimation_procedures)
+
+    def _parse_list_xml(  # noqa: C901, PLR0912
+        self,
+        api_call: str,
+        estimation_procedures: builtins.list[dict[str, Any]],
+    ) -> pd.DataFrame:
+        """Returns a Pandas DataFrame with information about OpenML tasks.
+
+        Parameters
+        ----------
+        api_call : str
+            The API call specifying which tasks to return.
+        estimation_procedures : list of dict
+            All estimation procedures on OpenML.
+
+        Returns
+        -------
+            A Pandas DataFrame with information about OpenML tasks.
+
+        Raises
+        ------
+        ValueError
+            If the XML returned by the OpenML API does not contain 'oml:tasks', '@xmlns:oml',
+            or has an incorrect value for '@xmlns:oml'.
+        KeyError
+            If an invalid key is found in the XML for a task.
+        """
+        xml_string = self._http.get(api_call).text
+
+        tasks_dict = xmltodict.parse(xml_string, force_list=("oml:task", "oml:input"))
+        # Minimalistic check if the XML is useful
+        if "oml:tasks" not in tasks_dict:
+            raise ValueError(f'Error in return XML, does not contain "oml:tasks": {tasks_dict}')
+
+        if "@xmlns:oml" not in tasks_dict["oml:tasks"]:
+            raise ValueError(
+                f'Error in return XML, does not contain "oml:tasks"/@xmlns:oml: {tasks_dict}'
+            )
+
+        if tasks_dict["oml:tasks"]["@xmlns:oml"] != "http://openml.org/openml":
+            raise ValueError(
+                "Error in return XML, value of "
+                '"oml:tasks"/@xmlns:oml is not '
+                f'"http://openml.org/openml": {tasks_dict!s}',
+            )
+
+        assert isinstance(tasks_dict["oml:tasks"]["oml:task"], list), type(tasks_dict["oml:tasks"])
+
+        tasks = {}
+        proc_dict = {x["id"]: x for x in estimation_procedures}
+
+        for task_ in tasks_dict["oml:tasks"]["oml:task"]:
+            tid = None
+            try:
+                tid = int(task_["oml:task_id"])
+                task_type_int = int(task_["oml:task_type_id"])
+                try:
+                    task_type_id = TaskType(task_type_int)
+                except ValueError as e:
+                    warnings.warn(
+                        f"Could not create task type id for {task_type_int} due to error {e}",
+                        RuntimeWarning,
+                        stacklevel=2,
+                    )
+                    continue
+
+                task = {
+                    "tid": tid,
+                    "ttid": task_type_id,
+                    "did": int(task_["oml:did"]),
+                    "name": task_["oml:name"],
+                    "task_type": task_["oml:task_type"],
+                    "status": task_["oml:status"],
+                }
+
+                # Other task inputs
+                for _input in task_.get("oml:input", []):
+                    if _input["@name"] == "estimation_procedure":
+                        task[_input["@name"]] = proc_dict[int(_input["#text"])]["name"]
+                    else:
+                        value = _input.get("#text")
+                        task[_input["@name"]] = value
+
+                # The number of qualities can range from 0 to infinity
+                for quality in task_.get("oml:quality", []):
+                    if "#text" not in quality:
+                        quality_value = 0.0
+                    else:
+                        quality["#text"] = float(quality["#text"])
+                        if abs(int(quality["#text"]) - quality["#text"]) < 0.0000001:
+                            quality["#text"] = int(quality["#text"])
+                        quality_value = quality["#text"]
+                    task[quality["@name"]] = quality_value
+                tasks[tid] = task
+            except KeyError as e:
+                if tid is not None:
+                    warnings.warn(
+                        f"Invalid xml for task {tid}: {e}\nFrom {task_}",
+                        RuntimeWarning,
+                        stacklevel=2,
+                    )
+                else:
+                    warnings.warn(
+                        f"Could not find key {e} in {task_}!", RuntimeWarning, stacklevel=2
+                    )
+
+        return pd.DataFrame.from_dict(tasks, orient="index")
+
+    @staticmethod
+    def _create_task_from_xml(xml: str) -> OpenMLTask:
+        """Create a task given a xml string.
+
+        Parameters
+        ----------
+        xml : string
+            Task xml representation.
+
+        Returns
+        -------
+        OpenMLTask
+        """
+        dic = xmltodict.parse(xml)["oml:task"]
+        estimation_parameters = {}
+        inputs = {}
+        # Due to the unordered structure we obtain, we first have to extract
+        # the possible keys of oml:input; dic["oml:input"] is a list of
+        # OrderedDicts
+
+        # Check if there is a list of inputs
+        if isinstance(dic["oml:input"], list):
+            for input_ in dic["oml:input"]:
+                name = input_["@name"]
+                inputs[name] = input_
+        # Single input case
+        elif isinstance(dic["oml:input"], dict):
+            name = dic["oml:input"]["@name"]
+            inputs[name] = dic["oml:input"]
+
+        evaluation_measures = None
+        if "evaluation_measures" in inputs:
+            evaluation_measures = inputs["evaluation_measures"]["oml:evaluation_measures"][
+                "oml:evaluation_measure"
+            ]
+
+        task_type = TaskType(int(dic["oml:task_type_id"]))
+        common_kwargs = {
+            "task_id": dic["oml:task_id"],
+            "task_type": dic["oml:task_type"],
+            "task_type_id": task_type,
+            "data_set_id": inputs["source_data"]["oml:data_set"]["oml:data_set_id"],
+            "evaluation_measure": evaluation_measures,
+        }
+        # TODO: add OpenMLClusteringTask?
+        if task_type in (
+            TaskType.SUPERVISED_CLASSIFICATION,
+            TaskType.SUPERVISED_REGRESSION,
+            TaskType.LEARNING_CURVE,
+        ):
+            # Convert some more parameters
+            for parameter in inputs["estimation_procedure"]["oml:estimation_procedure"][
+                "oml:parameter"
+            ]:
+                name = parameter["@name"]
+                text = parameter.get("#text", "")
+                estimation_parameters[name] = text
+
+            common_kwargs["estimation_procedure_type"] = inputs["estimation_procedure"][
+                "oml:estimation_procedure"
+            ]["oml:type"]
+            common_kwargs["estimation_procedure_id"] = int(
+                inputs["estimation_procedure"]["oml:estimation_procedure"]["oml:id"]
+            )
+
+            common_kwargs["estimation_parameters"] = estimation_parameters
+            common_kwargs["target_name"] = inputs["source_data"]["oml:data_set"][
+                "oml:target_feature"
+            ]
+            common_kwargs["data_splits_url"] = inputs["estimation_procedure"][
+                "oml:estimation_procedure"
+            ]["oml:data_splits_url"]
+
+        cls = {
+            TaskType.SUPERVISED_CLASSIFICATION: OpenMLClassificationTask,
+            TaskType.SUPERVISED_REGRESSION: OpenMLRegressionTask,
+            TaskType.CLUSTERING: OpenMLClusteringTask,
+            TaskType.LEARNING_CURVE: OpenMLLearningCurveTask,
+        }.get(task_type)
+        if cls is None:
+            raise NotImplementedError(f"Task type {common_kwargs['task_type']} not supported.")
+        return cls(**common_kwargs)  # type: ignore
+
+    @staticmethod
+    def _build_url(
+        limit: int, offset: int, task_type: TaskType | int | None, kwargs: dict[str, Any]
+    ) -> str:
+        api_call = "task/list"
+        if limit is not None:
+            api_call += f"/limit/{limit}"
+        if offset is not None:
+            api_call += f"/offset/{offset}"
+        if task_type is not None:
+            tvalue = task_type.value if isinstance(task_type, TaskType) else task_type
+            api_call += f"/type/{tvalue}"
+        if kwargs is not None:
+            for operator, value in kwargs.items():
+                if value is not None:
+                    if operator == "task_id":
+                        value = ",".join([str(int(i)) for i in value])  # noqa: PLW2901
+                    api_call += f"/{operator}/{value}"
+        return api_call
 
 
 class TaskV2API(ResourceV2API, TaskAPI):
-    """Version 2 API implementation for task resources."""
+    def get(self, task_id: int) -> OpenMLTask:
+        """Download OpenML task for a given task ID.
+
+        Downloads the task representation.
+
+        Parameters
+        ----------
+        task_id : int
+            The OpenML task id of the task to download.
+
+        Returns
+        -------
+        task: OpenMLTask
+        """
+        response = self._http.get(f"tasks/{task_id}", enable_cache=True)
+        return self._create_task_from_json(response.json())
+
+    def list(
+        self,
+        limit: int,  # noqa: ARG002
+        offset: int,  # noqa: ARG002
+        task_type: TaskType | int | None = None,  # noqa: ARG002
+        *,
+        estimation_procedures: builtins.list[dict[str, Any]],  # noqa: ARG002
+        **kwargs: Any,  # noqa: ARG002
+    ) -> pd.DataFrame:
+        raise self._not_supported(method="list")
+
+    def download_split(self, url: str, *, refresh_cache: bool = False) -> Path:
+        """Download a task's split file and return its path in the cache.
+
+        Parameters
+        ----------
+        url : str
+            URL of the split file.
+        refresh_cache : bool, optional
+            Whether to replace the file if it is already cached.
+
+        Returns
+        -------
+        Path
+        """
+        self._http.get(url, enable_cache=True, refresh_cache=refresh_cache)
+        return self._http.cache_path_from_url(url)
+
+    @staticmethod
+    def _create_task_from_json(task_json: dict) -> OpenMLTask:
+        task_type_id = TaskType(int(task_json["task_type_id"]))
+
+        inputs = {i["name"]: i for i in task_json.get("input", [])}
+
+        source = inputs["source_data"]["data_set"]
+        evaluation_measures = inputs.get("evaluation_measures", {}).get("evaluation_measures", {})
+
+        common_kwargs = {
+            "task_id": int(task_json["id"]),
+            "task_type": task_json["task_type"],
+            "task_type_id": task_type_id,
+            "data_set_id": int(source["data_set_id"]),
+            "evaluation_measure": evaluation_measures.get("evaluation_measure") or None,
+        }
+
+        if task_type_id in (
+            TaskType.SUPERVISED_CLASSIFICATION,
+            TaskType.SUPERVISED_REGRESSION,
+            TaskType.LEARNING_CURVE,
+        ):
+            est = inputs.get("estimation_procedure", {}).get("estimation_procedure")
+
+            if est:
+                common_kwargs["estimation_procedure_id"] = int(est["id"])
+                common_kwargs["estimation_procedure_type"] = est["type"]
+                common_kwargs["estimation_parameters"] = {
+                    p["name"]: "" if p.get("value") in (None, []) else str(p["value"])
+                    for p in est.get("parameter", [])
+                }
+                common_kwargs["data_splits_url"] = est["data_splits_url"]
+
+            common_kwargs["target_name"] = source.get("target_feature")
+
+        cls = {
+            TaskType.SUPERVISED_CLASSIFICATION: OpenMLClassificationTask,
+            TaskType.SUPERVISED_REGRESSION: OpenMLRegressionTask,
+            TaskType.CLUSTERING: OpenMLClusteringTask,
+            TaskType.LEARNING_CURVE: OpenMLLearningCurveTask,
+        }.get(task_type_id)
+        if cls is None:
+            raise NotImplementedError(f"Task type {common_kwargs['task_type']} not supported.")
+        return cls(**common_kwargs)  # type: ignore
diff --git a/openml/tasks/functions.py b/openml/tasks/functions.py
index 0b60b9d43d..b1609fb556 100644
--- a/openml/tasks/functions.py
+++ b/openml/tasks/functions.py
@@ -1,19 +1,14 @@
 # License: BSD 3-Clause
 from __future__ import annotations
 
-import os
-import re
 import warnings
 from functools import partial
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import pandas as pd
-import xmltodict
 
-import openml._api_calls
 import openml.utils
 from openml.datasets import get_dataset
-from openml.exceptions import OpenMLCacheException
 
 from .task import (
     OpenMLClassificationTask,
@@ -21,53 +16,13 @@
     OpenMLLearningCurveTask,
     OpenMLRegressionTask,
     OpenMLSupervisedTask,
-    OpenMLTask,
     TaskType,
 )
 
-TASKS_CACHE_DIR_NAME = "tasks"
-
-
-def _get_cached_tasks() -> dict[int, OpenMLTask]:
-    """Return a dict of all the tasks which are cached locally.
-
-    Returns
-    -------
-    tasks : OrderedDict
-        A dict of all the cached tasks. Each task is an instance of
-        OpenMLTask.
-    """
-    task_cache_dir = openml.utils._create_cache_directory(TASKS_CACHE_DIR_NAME)
-    directory_content = os.listdir(task_cache_dir)  # noqa: PTH208, RUF100
-    directory_content.sort()
-
-    # Find all dataset ids for which we have downloaded the dataset
-    # description
-    tids = (int(did) for did in directory_content if re.match(r"[0-9]*", did))
-    return {tid: _get_cached_task(tid) for tid in tids}
-
-
-def _get_cached_task(tid: int) -> OpenMLTask:
-    """Return a cached task based on the given id.
-
-    Parameters
-    ----------
-    tid : int
-        Id of the task.
-
-    Returns
-    -------
-    OpenMLTask
-    """
-    tid_cache_dir = openml.utils._create_cache_directory_for_id(TASKS_CACHE_DIR_NAME, tid)
-
-    task_xml_path = tid_cache_dir / "task.xml"
-    try:
-        with task_xml_path.open(encoding="utf8") as fh:
-            return _create_task_from_xml(fh.read())
-    except OSError as e:
-        openml.utils._remove_cache_dir_for_id(TASKS_CACHE_DIR_NAME, tid_cache_dir)
-        raise OpenMLCacheException(f"Task file for tid {tid} not cached") from e
+if TYPE_CHECKING:
+    from .task import (
+        OpenMLTask,
+    )
 
 
 def _get_estimation_procedure_list() -> list[dict[str, Any]]:
@@ -133,8 +88,9 @@ def list_tasks(  # noqa: PLR0913,PLR0917
         calculated for the associated dataset, some of these are also returned.
     """
     listing_call = partial(
-        _list_tasks,
+        openml._backend.task.list,
         task_type=task_type,
+        estimation_procedures=_get_estimation_procedure_list(),
         tag=tag,
         data_tag=data_tag,
         status=status,
@@ -152,151 +108,6 @@ def list_tasks(  # noqa: PLR0913,PLR0917
     return pd.concat(batches)
 
 
-def _list_tasks(
-    limit: int,
-    offset: int,
-    task_type: TaskType | int | None = None,
-    **kwargs: Any,
-) -> pd.DataFrame:
-    """
-    Perform the api call to return a number of tasks having the given filters.
-
-    Parameters
-    ----------
-    Filter task_type is separated from the other filters because
-    it is used as task_type in the task description, but it is named
-    type when used as a filter in list tasks call.
-    limit: int
-    offset: int
-    task_type : TaskType, optional
-        Refers to the type of task.
-    kwargs: dict, optional
-        Legal filter operators: tag, task_id (list), data_tag, status, limit,
-        offset, data_id, data_name, number_instances, number_features,
-        number_classes, number_missing_values.
-
-    Returns
-    -------
-    dataframe
-    """
-    api_call = "task/list"
-    if limit is not None:
-        api_call += f"/limit/{limit}"
-    if offset is not None:
-        api_call += f"/offset/{offset}"
-    if task_type is not None:
-        tvalue = task_type.value if isinstance(task_type, TaskType) else task_type
-        api_call += f"/type/{tvalue}"
-    if kwargs is not None:
-        for operator, value in kwargs.items():
-            if value is not None:
-                if operator == "task_id":
-                    value = ",".join([str(int(i)) for i in value])  # noqa: PLW2901
-                api_call += f"/{operator}/{value}"
-
-    return __list_tasks(api_call=api_call)
-
-
-def __list_tasks(api_call: str) -> pd.DataFrame:  # noqa: C901, PLR0912
-    """Returns a Pandas DataFrame with information about OpenML tasks.
-
-    Parameters
-    ----------
-    api_call : str
-        The API call specifying which tasks to return.
-
-    Returns
-    -------
-        A Pandas DataFrame with information about OpenML tasks.
-
-    Raises
-    ------
-    ValueError
-        If the XML returned by the OpenML API does not contain 'oml:tasks', '@xmlns:oml',
-        or has an incorrect value for '@xmlns:oml'.
-    KeyError
-        If an invalid key is found in the XML for a task.
-    """
-    xml_string = openml._api_calls._perform_api_call(api_call, "get")
-    tasks_dict = xmltodict.parse(xml_string, force_list=("oml:task", "oml:input"))
-    # Minimalistic check if the XML is useful
-    if "oml:tasks" not in tasks_dict:
-        raise ValueError(f'Error in return XML, does not contain "oml:runs": {tasks_dict}')
-
-    if "@xmlns:oml" not in tasks_dict["oml:tasks"]:
-        raise ValueError(
-            f'Error in return XML, does not contain "oml:runs"/@xmlns:oml: {tasks_dict}'
-        )
-
-    if tasks_dict["oml:tasks"]["@xmlns:oml"] != "http://openml.org/openml":
-        raise ValueError(
-            "Error in return XML, value of  "
-            '"oml:runs"/@xmlns:oml is not '
-            f'"http://openml.org/openml": {tasks_dict!s}',
-        )
-
-    assert isinstance(tasks_dict["oml:tasks"]["oml:task"], list), type(tasks_dict["oml:tasks"])
-
-    tasks = {}
-    procs = _get_estimation_procedure_list()
-    proc_dict = {x["id"]: x for x in procs}
-
-    for task_ in tasks_dict["oml:tasks"]["oml:task"]:
-        tid = None
-        try:
-            tid = int(task_["oml:task_id"])
-            task_type_int = int(task_["oml:task_type_id"])
-            try:
-                task_type_id = TaskType(task_type_int)
-            except ValueError as e:
-                warnings.warn(
-                    f"Could not create task type id for {task_type_int} due to error {e}",
-                    RuntimeWarning,
-                    stacklevel=2,
-                )
-                continue
-
-            task = {
-                "tid": tid,
-                "ttid": task_type_id,
-                "did": int(task_["oml:did"]),
-                "name": task_["oml:name"],
-                "task_type": task_["oml:task_type"],
-                "status": task_["oml:status"],
-            }
-
-            # Other task inputs
-            for _input in task_.get("oml:input", []):
-                if _input["@name"] == "estimation_procedure":
-                    task[_input["@name"]] = proc_dict[int(_input["#text"])]["name"]
-                else:
-                    value = _input.get("#text")
-                    task[_input["@name"]] = value
-
-            # The number of qualities can range from 0 to infinity
-            for quality in task_.get("oml:quality", []):
-                if "#text" not in quality:
-                    quality_value = 0.0
-                else:
-                    quality["#text"] = float(quality["#text"])
-                    if abs(int(quality["#text"]) - quality["#text"]) < 0.0000001:
-                        quality["#text"] = int(quality["#text"])
-                    quality_value = quality["#text"]
-                task[quality["@name"]] = quality_value
-            tasks[tid] = task
-        except KeyError as e:
-            if tid is not None:
-                warnings.warn(
-                    f"Invalid xml for task {tid}: {e}\nFrom {task_}",
-                    RuntimeWarning,
-                    stacklevel=2,
-                )
-            else:
-                warnings.warn(f"Could not find key {e} in {task_}!", RuntimeWarning, stacklevel=2)
-
-    return pd.DataFrame.from_dict(tasks, orient="index")
-
-
 def get_tasks(
     task_ids: list[int],
     download_data: bool | None = None,
@@ -304,7 +115,7 @@ def get_tasks(
 ) -> list[OpenMLTask]:
     """Download tasks.
 
-    This function iterates :meth:`openml.tasks.get_task`.
+    This function iterates :meth:`openml.task.get`.
 
     Parameters
     ----------
@@ -383,133 +194,21 @@ def get_task(
     if not isinstance(task_id, int):
         raise TypeError(f"Task id should be integer, is {type(task_id)}")
 
-    task_cache_directory = openml.utils._create_cache_directory_for_id(
-        TASKS_CACHE_DIR_NAME, task_id
-    )
-    task_cache_directory_existed = task_cache_directory.exists()
-    try:
-        task = _get_task_description(task_id)
-        dataset = get_dataset(task.dataset_id, **get_dataset_kwargs)
-        # List of class labels available in dataset description
-        # Including class labels as part of task meta data handles
-        #   the case where data download was initially disabled
-        if isinstance(task, (OpenMLClassificationTask, OpenMLLearningCurveTask)):
-            assert task.target_name is not None, (
-                "Supervised tasks must define a target feature before retrieving class labels."
-            )
-            task.class_labels = dataset.retrieve_class_labels(task.target_name)
-        # Clustering tasks do not have class labels
-        # and do not offer download_split
-        if download_splits and isinstance(task, OpenMLSupervisedTask):
-            task.download_split()
-    except Exception as e:
-        if not task_cache_directory_existed:
-            openml.utils._remove_cache_dir_for_id(TASKS_CACHE_DIR_NAME, task_cache_directory)
-        raise e
-
-    return task
+    task = openml._backend.task.get(task_id)
+    dataset = get_dataset(task.dataset_id, **get_dataset_kwargs)
 
-
-def _get_task_description(task_id: int) -> OpenMLTask:
-    try:
-        return _get_cached_task(task_id)
-    except OpenMLCacheException:
-        _cache_dir = openml.utils._create_cache_directory_for_id(TASKS_CACHE_DIR_NAME, task_id)
-        xml_file = _cache_dir / "task.xml"
-        task_xml = openml._api_calls._perform_api_call(f"task/{task_id}", "get")
-
-        with xml_file.open("w", encoding="utf8") as fh:
-            fh.write(task_xml)
-        return _create_task_from_xml(task_xml)
-
-
-def _create_task_from_xml(xml: str) -> OpenMLTask:
-    """Create a task given a xml string.
-
-    Parameters
-    ----------
-    xml : string
-        Task xml representation.
-
-    Returns
-    -------
-    OpenMLTask
-    """
-    dic = xmltodict.parse(xml)["oml:task"]
-    estimation_parameters = {}
-    inputs = {}
-    # Due to the unordered structure we obtain, we first have to extract
-    # the possible keys of oml:input; dic["oml:input"] is a list of
-    # OrderedDicts
-
-    # Check if there is a list of inputs
-    if isinstance(dic["oml:input"], list):
-        for input_ in dic["oml:input"]:
-            name = input_["@name"]
-            inputs[name] = input_
-    # Single input case
-    elif isinstance(dic["oml:input"], dict):
-        name = dic["oml:input"]["@name"]
-        inputs[name] = dic["oml:input"]
-
-    evaluation_measures = None
-    if "evaluation_measures" in inputs:
-        evaluation_measures = inputs["evaluation_measures"]["oml:evaluation_measures"][
-            "oml:evaluation_measure"
-        ]
-
-    task_type = TaskType(int(dic["oml:task_type_id"]))
-    common_kwargs = {
-        "task_id": dic["oml:task_id"],
-        "task_type": dic["oml:task_type"],
-        "task_type_id": task_type,
-        "data_set_id": inputs["source_data"]["oml:data_set"]["oml:data_set_id"],
-        "evaluation_measure": evaluation_measures,
-    }
-    # TODO: add OpenMLClusteringTask?
-    if task_type in (
-        TaskType.SUPERVISED_CLASSIFICATION,
-        TaskType.SUPERVISED_REGRESSION,
-        TaskType.LEARNING_CURVE,
+    if (
+        isinstance(task, (OpenMLClassificationTask, OpenMLLearningCurveTask))
+        and task.target_name is not None
     ):
-        # Convert some more parameters
-        for parameter in inputs["estimation_procedure"]["oml:estimation_procedure"][
-            "oml:parameter"
-        ]:
-            name = parameter["@name"]
-            text = parameter.get("#text", "")
-            estimation_parameters[name] = text
-
-        common_kwargs["estimation_procedure_type"] = inputs["estimation_procedure"][
-            "oml:estimation_procedure"
-        ]["oml:type"]
-        common_kwargs["estimation_procedure_id"] = int(
-            inputs["estimation_procedure"]["oml:estimation_procedure"]["oml:id"]
-        )
+        task.class_labels = dataset.retrieve_class_labels(task.target_name)
 
-        common_kwargs["estimation_parameters"] = estimation_parameters
-        common_kwargs["target_name"] = inputs["source_data"]["oml:data_set"]["oml:target_feature"]
-        common_kwargs["data_splits_url"] = inputs["estimation_procedure"][
-            "oml:estimation_procedure"
-        ]["oml:data_splits_url"]
-
-    cls = {
-        TaskType.SUPERVISED_CLASSIFICATION: OpenMLClassificationTask,
-        TaskType.SUPERVISED_REGRESSION: OpenMLRegressionTask,
-        TaskType.CLUSTERING: OpenMLClusteringTask,
-        TaskType.LEARNING_CURVE: OpenMLLearningCurveTask,
-    }.get(task_type)
-    if cls is None:
-        raise NotImplementedError(
-            f"Task type '{common_kwargs['task_type']}' is not supported. "
-            f"Supported task types: SUPERVISED_CLASSIFICATION,"
-            f"SUPERVISED_REGRESSION, CLUSTERING, LEARNING_CURVE."
-            f"Please check the OpenML documentation for available task types."
-        )
-    return cls(**common_kwargs)  # type: ignore
+    if download_splits and isinstance(task, OpenMLSupervisedTask):
+        task.download_split()
+
+    return task
 
 
-# TODO(eddiebergman): overload on `task_type`
 def create_task(
     task_type: TaskType,
     dataset_id: int,
@@ -597,4 +296,4 @@ def delete_task(task_id: int) -> bool:
     bool
         True if the deletion was successful. False otherwise.
     """
-    return openml.utils._delete_entity("task", task_id)
+    return openml._backend.task.delete(task_id)
diff --git a/openml/tasks/task.py b/openml/tasks/task.py
index 2bccd1af99..6875c21b99 100644
--- a/openml/tasks/task.py
+++ b/openml/tasks/task.py
@@ -10,10 +10,9 @@
 
 import arff
 
-import openml._api_calls
+import openml
 from openml import datasets
 from openml.base import OpenMLBase
-from openml.utils import _create_cache_directory_for_id
 
 from .split import OpenMLSplit
 
@@ -187,31 +186,15 @@ def download_split(self) -> OpenMLSplit:
         """Download the OpenML split for a given task."""
         # TODO(eddiebergman): Can this every be `None`?
         assert self.task_id is not None
-        cache_dir = _create_cache_directory_for_id("tasks", self.task_id)
-        cached_split_file = cache_dir / "datasplits.arff"
+        split_url = str(self.estimation_procedure["data_splits_url"])
+        split_file = openml._backend.task.download_split(split_url)
 
         try:
-            split = OpenMLSplit._from_arff_file(cached_split_file)
-            logger.debug("Loaded file from cache: %s", str(cached_split_file))
-        except (OSError, arff.BadDataFormat):
-            logger.info("Failed to load file from cache: %s", str(cached_split_file))
-            if cached_split_file.exists():
-                logger.debug("Cleaning up old file")
-                cached_split_file.unlink()
-            # Next, download and cache the associated split file
-            split_url = self.estimation_procedure["data_splits_url"]
-            openml._api_calls._download_text_file(
-                source=str(split_url),
-                output_path=str(cached_split_file),
-            )
-            if cached_split_file.exists():
-                logger.info("New file created of size %d", cached_split_file.stat().st_size)
-            else:
-                logger.info("Failed to create new file")
-
-            split = OpenMLSplit._from_arff_file(cached_split_file)
-
-        return split
+            return OpenMLSplit._from_arff_file(split_file)
+        except arff.BadDataFormat:
+            logger.info("Failed to parse cached split file: %s", str(split_file))
+            split_file = openml._backend.task.download_split(split_url, refresh_cache=True)
+            return OpenMLSplit._from_arff_file(split_file)
 
     def get_split_dimensions(self) -> tuple[int, int, int]:
         """Get the (repeats, folds, samples) of the split for a given task."""
@@ -242,6 +225,48 @@ def _parse_publish_response(self, xml_response: dict) -> None:
         """Parse the id from the xml_response and assign it to self."""
         self.task_id = int(xml_response["oml:upload_task"]["oml:id"])
 
+    def publish(self) -> OpenMLTask:
+        """Publish this task to OpenML server.
+
+        Returns
+        -------
+        self : OpenMLTask
+        """
+        file_elements = self._get_file_elements()
+        if "description" not in file_elements:
+            file_elements["description"] = self._to_xml()
+        task_id = openml._backend.task.publish(path="task", files=file_elements)
+        self.task_id = task_id
+        return self
+
+    def push_tag(self, tag: str) -> None:
+        """Annotates this task with a tag on the server.
+
+        Parameters
+        ----------
+        tag : str
+            Tag to attach to the task.
+        """
+        if self.task_id is None:
+            raise openml.exceptions.ObjectNotPublishedError(
+                "Please publish the task first before being able to tag it."
+            )
+        openml._backend.task.tag(self.task_id, tag)
+
+    def remove_tag(self, tag: str) -> None:
+        """Removes a tag from this task on the server.
+
+        Parameters
+        ----------
+        tag : str
+            Tag to remove from the task.
+        """
+        if self.task_id is None:
+            raise openml.exceptions.ObjectNotPublishedError(
+                "Please publish the task first before being able to untag it."
+            )
+        openml._backend.task.untag(self.task_id, tag)
+
 
 class OpenMLSupervisedTask(OpenMLTask, ABC):
     """OpenML Supervised Classification object.
diff --git a/pyproject.toml b/pyproject.toml
index 02b31ef2a9..460ae37e82 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -359,6 +359,7 @@ module = [
   "tests.test_api.test_run",
   "tests.test_api.test_setup",
   "tests.test_api.test_study",
+  "tests.test_api.test_task",
   "tests.test_api.test_versions",
   "tests.conftest",
   "tests.test_datasets.test_dataset",
diff --git a/tests/conftest.py b/tests/conftest.py
index 8f98422c7e..c96cbf0915 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -202,17 +202,6 @@ def _expected_static_cache_state(root_dir: Path) -> list[Path]:
     _c_root_dir = root_dir / "org" / "openml" / "test"
     res_paths = [root_dir, _c_root_dir]
 
-    res_paths.append(_c_root_dir / "tasks")
-
-    for _id in ["1", "3", "1882"]:
-        tmp_p = _c_root_dir / "tasks" / _id
-        res_paths.extend(
-            [
-                tmp_p / "datasplits.arff",
-                tmp_p / "task.xml",
-            ]
-        )
-
     res_paths.extend(
         [
             _c_root_dir / "api" / "v1" / "xml" / "setup",
@@ -239,6 +228,17 @@ def _expected_static_cache_state(root_dir: Path) -> list[Path]:
         ]
     )
 
+    res_paths.extend(
+        [
+            _c_root_dir / "api_splits" / "get" / "1882" / "Task_1882_splits.arff" / "body.arff",
+            _c_root_dir / "api_splits" / "get" / "3" / "Task_3_splits.arff" / "body.arff",
+            _c_root_dir / "api_splits" / "get" / "1" / "Task_1_splits.arff" / "body.arff",
+            _c_root_dir / "api" / "v1" / "xml" / "task" / "1882" / "body.xml",
+            _c_root_dir / "api" / "v1" / "xml" / "task" / "3" / "body.xml",
+            _c_root_dir / "api" / "v1" / "xml" / "task" / "1" / "body.xml",
+        ]
+    )
+
     return res_paths
 
 
diff --git a/tests/files/org/openml/test/tasks/1/task.xml b/tests/files/org/openml/test/api/v1/xml/task/1/body.xml
similarity index 100%
rename from tests/files/org/openml/test/tasks/1/task.xml
rename to tests/files/org/openml/test/api/v1/xml/task/1/body.xml
diff --git a/tests/files/org/openml/test/tasks/1882/task.xml b/tests/files/org/openml/test/api/v1/xml/task/1882/body.xml
similarity index 100%
rename from tests/files/org/openml/test/tasks/1882/task.xml
rename to tests/files/org/openml/test/api/v1/xml/task/1882/body.xml
diff --git a/tests/files/org/openml/test/tasks/3/task.xml b/tests/files/org/openml/test/api/v1/xml/task/3/body.xml
similarity index 100%
rename from tests/files/org/openml/test/tasks/3/task.xml
rename to tests/files/org/openml/test/api/v1/xml/task/3/body.xml
diff --git a/tests/files/org/openml/test/tasks/1/datasplits.arff b/tests/files/org/openml/test/api_splits/get/1/Task_1_splits.arff/body.arff
similarity index 100%
rename from tests/files/org/openml/test/tasks/1/datasplits.arff
rename to tests/files/org/openml/test/api_splits/get/1/Task_1_splits.arff/body.arff
diff --git a/tests/files/org/openml/test/tasks/1882/datasplits.arff b/tests/files/org/openml/test/api_splits/get/1882/Task_1882_splits.arff/body.arff
similarity index 100%
rename from tests/files/org/openml/test/tasks/1882/datasplits.arff
rename to tests/files/org/openml/test/api_splits/get/1882/Task_1882_splits.arff/body.arff
diff --git a/tests/files/org/openml/test/tasks/3/datasplits.arff b/tests/files/org/openml/test/api_splits/get/3/Task_3_splits.arff/body.arff
similarity index 100%
rename from tests/files/org/openml/test/tasks/3/datasplits.arff
rename to tests/files/org/openml/test/api_splits/get/3/Task_3_splits.arff/body.arff
diff --git a/tests/test_api/test_task.py b/tests/test_api/test_task.py
new file mode 100644
index 0000000000..4d30e27403
--- /dev/null
+++ b/tests/test_api/test_task.py
@@ -0,0 +1,276 @@
+# License: BSD 3-Clause
+from __future__ import annotations
+
+from unittest.mock import patch
+
+import pandas as pd
+import pytest
+from requests import Response, Session
+
+import openml
+from openml._api import TaskV1API, TaskV2API
+from openml.exceptions import OpenMLNotSupportedError
+from openml.tasks import OpenMLClassificationTask, OpenMLSplit, TaskType
+from openml.tasks.functions import _get_estimation_procedure_list
+
+CLASSIFICATION_TASK_ID = 1
+REGRESSION_TASK_ID = 613
+LEARNING_CURVE_TASK_ID = 763
+DATASTREAM_TASK_ID = 967
+
+
+@pytest.fixture
+def task_v1(http_client_v1, minio_client) -> TaskV1API:
+    return TaskV1API(http=http_client_v1, minio=minio_client)
+
+
+@pytest.fixture
+def task_v2(http_client_v2, minio_client) -> TaskV2API:
+    return TaskV2API(http=http_client_v2, minio=minio_client)
+
+
+def _assert_task_1(task):
+    assert isinstance(task, OpenMLClassificationTask)
+    assert task.task_id == CLASSIFICATION_TASK_ID
+    assert task.task_type_id == TaskType.SUPERVISED_CLASSIFICATION
+    assert task.dataset_id == 1
+    assert task.target_name == "class"
+    assert task.estimation_procedure_id == 1
+    assert task.estimation_procedure["type"] == "crossvalidation"
+    assert task.estimation_procedure["parameters"]["number_folds"] == "10"
+    assert task.estimation_procedure["data_splits_url"].endswith("Task_1_splits.arff")
+
+
+@pytest.mark.test_server
+def test_v1_get(task_v1):
+    _assert_task_1(task_v1.get(CLASSIFICATION_TASK_ID))
+
+
+@pytest.mark.test_server
+def test_v1_list(task_v1):
+    tasks_df = task_v1.list(
+        limit=5,
+        offset=0,
+        task_type=TaskType.SUPERVISED_REGRESSION,
+        estimation_procedures=_get_estimation_procedure_list(),
+    )
+    assert isinstance(tasks_df, pd.DataFrame)
+    assert len(tasks_df) == 5
+    assert (tasks_df["ttid"] == TaskType.SUPERVISED_REGRESSION).all()
+    names = {procedure["name"] for procedure in _get_estimation_procedure_list()}
+    assert set(tasks_df["estimation_procedure"]) <= names
+
+
+@pytest.mark.test_server
+def test_v1_download_split(task_v1):
+    url = task_v1.get(CLASSIFICATION_TASK_ID).estimation_procedure["data_splits_url"]
+    path = task_v1.download_split(url)
+    assert path.parts[-4:] == ("get", "1", "Task_1_splits.arff", "body.arff")
+    assert isinstance(OpenMLSplit._from_arff_file(path), OpenMLSplit)
+
+
+@pytest.mark.test_server
+def test_v1_download_split_refresh_cache(task_v1):
+    url = task_v1.get(CLASSIFICATION_TASK_ID).estimation_procedure["data_splits_url"]
+    path = task_v1.download_split(url)
+    original = path.read_bytes()
+    path.write_bytes(b"@relation stale")
+
+    assert task_v1.download_split(url).read_bytes() == b"@relation stale"
+    assert task_v1.download_split(url, refresh_cache=True).read_bytes() == original
+
+
+@pytest.mark.test_server
+def test_v2_get(task_v2):
+    _assert_task_1(task_v2.get(CLASSIFICATION_TASK_ID))
+
+
+def test_v2_list(task_v2):
+    with pytest.raises(OpenMLNotSupportedError):
+        task_v2.list(limit=5, offset=0, estimation_procedures=[])
+
+
+@pytest.mark.test_server
+def test_v2_download_split(task_v2):
+    url = task_v2.get(CLASSIFICATION_TASK_ID).estimation_procedure["data_splits_url"]
+    path = task_v2.download_split(url)
+    assert path.parts[-4:] == ("get", "1", "Task_1_splits.arff", "body.arff")
+    assert isinstance(OpenMLSplit._from_arff_file(path), OpenMLSplit)
+
+
+@pytest.mark.test_server
+def test_v2_evaluation_measure(task_v2):
+    task_json = task_v2._http.get(f"tasks/{CLASSIFICATION_TASK_ID}").json()
+    inputs = {i["name"]: i for i in task_json["input"]}
+    measures = inputs["evaluation_measures"]["evaluation_measures"]
+    assert measures["evaluation_measure"] == []
+    assert task_v2._create_task_from_json(task_json).evaluation_measure is None
+
+    measures["evaluation_measure"] = "predictive_accuracy"
+    task = task_v2._create_task_from_json(task_json)
+    assert task.evaluation_measure == "predictive_accuracy"
+
+
+@pytest.mark.test_server
+@pytest.mark.parametrize(
+    "task_id",
+    [CLASSIFICATION_TASK_ID, REGRESSION_TASK_ID, LEARNING_CURVE_TASK_ID],
+)
+def test_get_matches(task_v1, task_v2, task_id):
+    assert vars(task_v1.get(task_id)) == vars(task_v2.get(task_id))
+
+
+@pytest.mark.test_server
+def test_get_unsupported_task_type_matches(task_v1, task_v2):
+    match = "Supervised Data Stream Classification not supported"
+    with pytest.raises(NotImplementedError, match=match):
+        task_v1.get(DATASTREAM_TASK_ID)
+    with pytest.raises(NotImplementedError, match=match):
+        task_v2.get(DATASTREAM_TASK_ID)
+
+
+@pytest.mark.test_server
+def test_download_split_matches(task_v1, task_v2):
+    url_v1 = task_v1.get(CLASSIFICATION_TASK_ID).estimation_procedure["data_splits_url"]
+    url_v2 = task_v2.get(CLASSIFICATION_TASK_ID).estimation_procedure["data_splits_url"]
+    assert url_v1 == url_v2
+    assert task_v1.download_split(url_v1) == task_v2.download_split(url_v2)
+
+
+def test_v1_publish(task_v1):
+    resource_name = task_v1.resource_type.value
+    resource_files = {"description": "Resource Description File"}
+    resource_id = 123
+
+    with patch.object(Session, "request") as mock_request:
+        mock_request.return_value = Response()
+        mock_request.return_value.status_code = 200
+        mock_request.return_value._content = (
+            f'<oml:upload_task xmlns:oml="http://openml.org/openml">\n'
+            f"\t<oml:id>{resource_id}</oml:id>\n"
+            f"</oml:upload_task>\n"
+        ).encode()
+
+        published_resource_id = task_v1.publish(
+            resource_name,
+            files=resource_files,
+        )
+
+        assert resource_id == published_resource_id
+
+        mock_request.assert_called_once_with(
+            method="POST",
+            url=openml.config.server + resource_name,
+            params={},
+            data={"api_key": openml.config.apikey},
+            headers=openml.config._HEADERS,
+            files=resource_files,
+        )
+
+
+def test_v1_delete(task_v1):
+    resource_name = task_v1.resource_type.value
+    resource_id = 123
+
+    with patch.object(Session, "request") as mock_request:
+        mock_request.return_value = Response()
+        mock_request.return_value.status_code = 200
+        mock_request.return_value._content = (
+            f'<oml:task_delete xmlns:oml="http://openml.org/openml">\n'
+            f"  <oml:id>{resource_id}</oml:id>\n"
+            f"</oml:task_delete>\n"
+        ).encode()
+
+        task_v1.delete(resource_id)
+
+        mock_request.assert_called_once_with(
+            method="DELETE",
+            url=(openml.config.server + resource_name + "/" + str(resource_id)),
+            params={"api_key": openml.config.apikey},
+            data={},
+            headers=openml.config._HEADERS,
+            files=None,
+        )
+
+
+def test_v1_tag(task_v1):
+    resource_id = 123
+    resource_tag = "TAG"
+
+    with patch.object(Session, "request") as mock_request:
+        mock_request.return_value = Response()
+        mock_request.return_value.status_code = 200
+        mock_request.return_value._content = (
+            f'<oml:task_tag xmlns:oml="http://openml.org/openml">'
+            f"<oml:id>{resource_id}</oml:id>"
+            f"<oml:tag>{resource_tag}</oml:tag>"
+            f"</oml:task_tag>"
+        ).encode()
+
+        tags = task_v1.tag(resource_id, resource_tag)
+
+        assert resource_tag in tags
+
+        mock_request.assert_called_once_with(
+            method="POST",
+            url=(openml.config.server + task_v1.resource_type.value + "/tag"),
+            params={},
+            data={
+                "api_key": openml.config.apikey,
+                "task_id": resource_id,
+                "tag": resource_tag,
+            },
+            headers=openml.config._HEADERS,
+            files=None,
+        )
+
+
+def test_v1_untag(task_v1):
+    resource_id = 123
+    resource_tag = "TAG"
+
+    with patch.object(Session, "request") as mock_request:
+        mock_request.return_value = Response()
+        mock_request.return_value.status_code = 200
+        mock_request.return_value._content = (
+            f'<oml:task_untag xmlns:oml="http://openml.org/openml">'
+            f"<oml:id>{resource_id}</oml:id>"
+            f"</oml:task_untag>"
+        ).encode()
+
+        tags = task_v1.untag(resource_id, resource_tag)
+
+        assert resource_tag not in tags
+
+        mock_request.assert_called_once_with(
+            method="POST",
+            url=(openml.config.server + task_v1.resource_type.value + "/untag"),
+            params={},
+            data={
+                "api_key": openml.config.apikey,
+                "task_id": resource_id,
+                "tag": resource_tag,
+            },
+            headers=openml.config._HEADERS,
+            files=None,
+        )
+
+
+def test_v2_publish(task_v2):
+    with pytest.raises(OpenMLNotSupportedError):
+        task_v2.publish(path=None, files=None)
+
+
+def test_v2_delete(task_v2):
+    with pytest.raises(OpenMLNotSupportedError):
+        task_v2.delete(resource_id=None)
+
+
+def test_v2_tag(task_v2):
+    with pytest.raises(OpenMLNotSupportedError):
+        task_v2.tag(resource_id=None, tag=None)
+
+
+def test_v2_untag(task_v2):
+    with pytest.raises(OpenMLNotSupportedError):
+        task_v2.untag(resource_id=None, tag=None)
diff --git a/tests/test_tasks/test_split.py b/tests/test_tasks/test_split.py
index 373b479cdc..e2ac15cc05 100644
--- a/tests/test_tasks/test_split.py
+++ b/tests/test_tasks/test_split.py
@@ -28,9 +28,11 @@ def setUp(self):
             / "org"
             / "openml"
             / "test"
-            / "tasks"
+            / "api_splits"
+            / "get"
             / "1882"
-            / "datasplits.arff"
+            / "Task_1882_splits.arff"
+            / "body.arff"
         )
         # Use a unique temp directory for each test to avoid race conditions
         # when running tests in parallel (see issue #1641)
diff --git a/tests/test_tasks/test_task_functions.py b/tests/test_tasks/test_task_functions.py
index 639ad58cde..62be9f376a 100644
--- a/tests/test_tasks/test_task_functions.py
+++ b/tests/test_tasks/test_task_functions.py
@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import os
+from pathlib import Path
 from unittest import mock
 
 import pytest
@@ -9,11 +10,25 @@
 
 import openml
 from openml import OpenMLSplit, OpenMLTask
-from openml.exceptions import OpenMLCacheException, OpenMLNotAuthorizedError, OpenMLServerException
+from openml.exceptions import (
+    OpenMLNotAuthorizedError,
+    OpenMLServerException,
+)
 from openml.tasks import TaskType
 from openml.testing import TestBase, create_request_response
 
 
+def _cached_split_path(task_id: int) -> Path:
+    return Path(
+        openml.config.get_cache_directory(),
+        "api_splits",
+        "get",
+        str(task_id),
+        f"Task_{task_id}_splits.arff",
+        "body.arff",
+    )
+
+
 class TestTask(TestBase):
     _multiprocess_can_split_ = True
 
@@ -23,25 +38,6 @@ def setUp(self):
     def tearDown(self):
         super().tearDown()
 
-    @pytest.mark.test_server
-    def test__get_cached_tasks(self):
-        openml.config.set_root_cache_directory(self.static_cache_dir)
-        tasks = openml.tasks.functions._get_cached_tasks()
-        assert isinstance(tasks, dict)
-        assert len(tasks) == 3
-        assert isinstance(next(iter(tasks.values())), OpenMLTask)
-
-    @pytest.mark.test_server
-    def test__get_cached_task(self):
-        openml.config.set_root_cache_directory(self.static_cache_dir)
-        task = openml.tasks.functions._get_cached_task(1)
-        assert isinstance(task, OpenMLTask)
-
-    def test__get_cached_task_not_cached(self):
-        openml.config.set_root_cache_directory(self.static_cache_dir)
-        with pytest.raises(OpenMLCacheException, match="Task file for tid 2 not cached"):
-            openml.tasks.functions._get_cached_task(2)
-
     @pytest.mark.test_server
     def test__get_estimation_procedure_list(self):
         estimation_procedures = openml.tasks.functions._get_estimation_procedure_list()
@@ -134,7 +130,9 @@ def test_list_tasks_per_type_paginate(self):
     @pytest.mark.test_server
     def test__get_task(self):
         openml.config.set_root_cache_directory(self.static_cache_dir)
-        openml.tasks.get_task(1882)
+        with mock.patch("requests.sessions.Session.request") as mock_request:
+            openml.tasks.get_task(1882)
+            mock_request.assert_not_called()
 
     @pytest.mark.skip(
         reason="Please await outcome of discussion: https://github.com/openml/OpenML/issues/776",
@@ -151,12 +149,6 @@ def test_get_task(self):
         task = openml.tasks.get_task(1, download_data=True)  # anneal; crossvalidation
         dataset = openml.datasets.get_dataset(task.dataset_id)
         assert isinstance(task, OpenMLTask)
-        assert os.path.exists(  # noqa: PTH110
-            os.path.join(openml.config.get_cache_directory(), "tasks", "1", "task.xml")  # noqa: PTH118
-        )
-        assert not os.path.exists(  # noqa: PTH110
-            os.path.join(openml.config.get_cache_directory(), "tasks", "1", "datasplits.arff")  # noqa: PTH118
-        )
         assert os.path.exists(  # noqa: PTH110
             openml.config.get_minio_download_path(dataset._parquet_url)
         )
@@ -164,48 +156,29 @@ def test_get_task(self):
     @pytest.mark.test_server
     def test_get_task_lazy(self):
         task = openml.tasks.get_task(2, download_data=False)  # anneal; crossvalidation
+
         assert isinstance(task, OpenMLTask)
-        assert os.path.exists(  # noqa: PTH110
-            os.path.join(openml.config.get_cache_directory(), "tasks", "2", "task.xml")  # noqa: PTH118
-        )
         assert task.class_labels == ["1", "2", "3", "4", "5", "U"]
 
-        assert not os.path.exists(  # noqa: PTH110
-            os.path.join(openml.config.get_cache_directory(), "tasks", "2", "datasplits.arff")  # noqa: PTH118
-        )
         # Since the download_data=False is propagated to get_dataset
         assert not os.path.exists(  # noqa: PTH110
             os.path.join(openml.config.get_cache_directory(), "datasets", "2", "dataset.arff")  # noqa: PTH118
         )
 
+        assert not _cached_split_path(2).exists()
         task.download_split()
-        assert os.path.exists(  # noqa: PTH110
-            os.path.join(openml.config.get_cache_directory(), "tasks", "2", "datasplits.arff")  # noqa: PTH118
-        )
-
-    @mock.patch("openml.tasks.functions.get_dataset")
-    @pytest.mark.test_server
-    def test_removal_upon_download_failure(self, get_dataset):
-        class WeirdException(Exception):  # noqa: N818
-            pass
-
-        def assert_and_raise(*args, **kwargs):  # noqa: ARG001
-            # Make sure that the file was created!
-            assert os.path.join(os.getcwd(), "tasks", "1", "tasks.xml")  # noqa: PTH109, PTH118
-            raise WeirdException()
-
-        get_dataset.side_effect = assert_and_raise
-        try:  # noqa: SIM105
-            openml.tasks.get_task(1)  # anneal; crossvalidation
-        except WeirdException:
-            pass
-        # Now the file should no longer exist
-        assert not os.path.exists(os.path.join(os.getcwd(), "tasks", "1", "tasks.xml"))  # noqa: PTH109, PTH110, PTH118
+        assert _cached_split_path(2).exists()
 
     @pytest.mark.test_server
     def test_get_task_with_cache(self):
         openml.config.set_root_cache_directory(self.static_cache_dir)
-        task = openml.tasks.get_task(1)
+        with (
+            mock.patch("openml.tasks.functions.get_dataset"),
+            mock.patch("requests.sessions.Session.request") as mock_request,
+        ):
+            task = openml.tasks.get_task(1)
+            mock_request.assert_not_called()
+
         assert isinstance(task, OpenMLTask)
 
     @pytest.mark.production_server
@@ -223,29 +196,41 @@ def test_download_split(self):
         task = openml.tasks.get_task(1)  # anneal; crossvalidation
         split = task.download_split()
         assert type(split) == OpenMLSplit  # noqa: E721
-        assert os.path.exists(  # noqa: PTH110
-            os.path.join(openml.config.get_cache_directory(), "tasks", "1", "datasplits.arff")  # noqa: PTH118
-        )
+        assert _cached_split_path(1).exists()
 
-    def test_deletion_of_cache_dir(self):
-        # Simple removal
-        tid_cache_dir = openml.utils._create_cache_directory_for_id(
-            "tasks",
-            1,
+    @pytest.mark.test_server
+    def test_download_split_replaces_corrupted_file(self):
+        task = openml.tasks.get_task(1)  # anneal; crossvalidation
+        expected = task.download_split()
+
+        split_file = _cached_split_path(1)
+        split_file.with_suffix(".pkl.py3").unlink()
+        split_file.write_text(
+            "@relation corrupted\n"
+            "@attribute type {TRAIN,TEST}\n"
+            "@attribute rowid INTEGER\n"
+            "@attribute repeat INTEGER\n"
+            "@attribute fold INTEGER\n"
+            "@data\n"
+            "TRAIN,1\n"
         )
-        assert os.path.exists(tid_cache_dir)  # noqa: PTH110
-        openml.utils._remove_cache_dir_for_id("tasks", tid_cache_dir)
-        assert not os.path.exists(tid_cache_dir)  # noqa: PTH110
 
+        assert task.download_split() == expected
+        assert "corrupted" not in split_file.read_text()
 
-@mock.patch.object(requests.Session, "delete")
-def test_delete_task_not_owned(mock_delete, test_files_directory, test_server_v1, test_apikey_v1):
+    @pytest.mark.test_server
+    def test_get_task_with_download_splits(self):
+        openml.tasks.get_task(1, download_splits=True)
+        assert _cached_split_path(1).exists()
+
+
+@mock.patch.object(requests.Session, "request")
+def test_delete_task_not_owned(mock_request, test_files_directory, test_server_v1, test_apikey_v1):
     content_file = test_files_directory / "mock_responses" / "tasks" / "task_delete_not_owned.xml"
-    mock_delete.return_value = create_request_response(
+    mock_request.return_value = create_request_response(
         status_code=412,
         content_filepath=content_file,
     )
-
     with pytest.raises(
         OpenMLNotAuthorizedError,
         match=r"The task can not be deleted because it was not uploaded by you.",
@@ -253,14 +238,14 @@ def test_delete_task_not_owned(mock_delete, test_files_directory, test_server_v1
         openml.tasks.delete_task(1)
 
     task_url = test_server_v1 + "task/1"
-    assert task_url == mock_delete.call_args.args[0]
-    assert test_apikey_v1 == mock_delete.call_args.kwargs.get("params", {}).get("api_key")
+    assert task_url == mock_request.call_args.kwargs.get("url")
+    assert test_apikey_v1 == mock_request.call_args.kwargs.get("params", {}).get("api_key")
 
 
-@mock.patch.object(requests.Session, "delete")
-def test_delete_task_with_run(mock_delete, test_files_directory, test_server_v1, test_apikey_v1):
+@mock.patch.object(requests.Session, "request")
+def test_delete_task_with_run(mock_request, test_files_directory, test_server_v1, test_apikey_v1):
     content_file = test_files_directory / "mock_responses" / "tasks" / "task_delete_has_runs.xml"
-    mock_delete.return_value = create_request_response(
+    mock_request.return_value = create_request_response(
         status_code=412,
         content_filepath=content_file,
     )
@@ -272,14 +257,14 @@ def test_delete_task_with_run(mock_delete, test_files_directory, test_server_v1,
         openml.tasks.delete_task(3496)
 
     task_url = test_server_v1 + "task/3496"
-    assert task_url == mock_delete.call_args.args[0]
-    assert test_apikey_v1 == mock_delete.call_args.kwargs.get("params", {}).get("api_key")
+    assert task_url == mock_request.call_args.kwargs.get("url")
+    assert test_apikey_v1 == mock_request.call_args.kwargs.get("params", {}).get("api_key")
 
 
-@mock.patch.object(requests.Session, "delete")
-def test_delete_success(mock_delete, test_files_directory, test_server_v1, test_apikey_v1):
+@mock.patch.object(requests.Session, "request")
+def test_delete_success(mock_request, test_files_directory, test_server_v1, test_apikey_v1):
     content_file = test_files_directory / "mock_responses" / "tasks" / "task_delete_successful.xml"
-    mock_delete.return_value = create_request_response(
+    mock_request.return_value = create_request_response(
         status_code=200,
         content_filepath=content_file,
     )
@@ -288,14 +273,14 @@ def test_delete_success(mock_delete, test_files_directory, test_server_v1, test_
     assert success
 
     task_url = test_server_v1 + "task/361323"
-    assert task_url == mock_delete.call_args.args[0]
-    assert test_apikey_v1 == mock_delete.call_args.kwargs.get("params", {}).get("api_key")
+    assert task_url == mock_request.call_args.kwargs.get("url")
+    assert test_apikey_v1 == mock_request.call_args.kwargs.get("params", {}).get("api_key")
 
 
-@mock.patch.object(requests.Session, "delete")
-def test_delete_unknown_task(mock_delete, test_files_directory, test_server_v1, test_apikey_v1):
+@mock.patch.object(requests.Session, "request")
+def test_delete_unknown_task(mock_request, test_files_directory, test_server_v1, test_apikey_v1):
     content_file = test_files_directory / "mock_responses" / "tasks" / "task_delete_not_exist.xml"
-    mock_delete.return_value = create_request_response(
+    mock_request.return_value = create_request_response(
         status_code=412,
         content_filepath=content_file,
     )
@@ -307,5 +292,5 @@ def test_delete_unknown_task(mock_delete, test_files_directory, test_server_v1,
         openml.tasks.delete_task(9_999_999)
 
     task_url = test_server_v1 + "task/9999999"
-    assert task_url == mock_delete.call_args.args[0]
-    assert test_apikey_v1 == mock_delete.call_args.kwargs.get("params", {}).get("api_key")
+    assert task_url == mock_request.call_args.kwargs.get("url")
+    assert test_apikey_v1 == mock_request.call_args.kwargs.get("params", {}).get("api_key")
diff --git a/tests/test_tasks/test_task_methods.py b/tests/test_tasks/test_task_methods.py
index 53e0a5e95d..218401c598 100644
--- a/tests/test_tasks/test_task_methods.py
+++ b/tests/test_tasks/test_task_methods.py
@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 from time import time
+from unittest import mock
 
 import pytest
 
@@ -36,8 +37,10 @@ def test_tagging(self):
     @pytest.mark.test_server
     def test_get_train_and_test_split_indices(self):
         openml.config.set_root_cache_directory(self.static_cache_dir)
-        task = openml.tasks.get_task(1882)
-        train_indices, test_indices = task.get_train_test_split_indices(0, 0)
+        with mock.patch("requests.sessions.Session.request") as mock_request:
+            task = openml.tasks.get_task(1882)
+            train_indices, test_indices = task.get_train_test_split_indices(0, 0)
+            mock_request.assert_not_called()
         assert train_indices[0] == 16
         assert train_indices[-1] == 395
         assert test_indices[0] == 412
diff --git a/tests/test_utils/test_utils.py b/tests/test_utils/test_utils.py
index b252e98d22..ec3a7a0b7e 100644
--- a/tests/test_utils/test_utils.py
+++ b/tests/test_utils/test_utils.py
@@ -2,6 +2,7 @@
 
 import os
 import unittest.mock
+from functools import partial
 
 import pytest
 import requests
@@ -48,7 +49,7 @@ def min_number_evaluations_on_test_server() -> int:
 
 @pytest.mark.test_server
 def test_list_all():
-    openml.utils._list_all(listing_call=openml.tasks.functions._list_tasks)
+    openml.utils._list_all(listing_call=openml.tasks.functions.list_tasks)
 
 
 @pytest.mark.test_server
@@ -63,7 +64,10 @@ def test_list_all_with_multiple_batches(min_number_tasks_on_test_server):
     # batches and at the same time do as few batches (roundtrips) as possible.
     batch_size = min_number_tasks_on_test_server - 1
     batches = openml.utils._list_all(
-        listing_call=openml.tasks.functions._list_tasks,
+        listing_call=partial(
+            openml._backend.task.list,
+            estimation_procedures=openml.tasks.functions._get_estimation_procedure_list(),
+        ),
         batch_size=batch_size,
     )
     assert len(batches) >= 2