diff --git a/lib/galaxy/managers/tags.py b/lib/galaxy/managers/tags.py
new file mode 100644
index 00000000000..a166d2578ba
--- /dev/null
+++ b/lib/galaxy/managers/tags.py
@@ -0,0 +1,66 @@
+from enum import Enum
+from typing import (
+    List,
+    Optional,
+)
+
+from pydantic import (
+    BaseModel,
+    Field,
+)
+
+from galaxy.managers.context import ProvidesUserContext
+from galaxy.model import ItemTagAssociation
+from galaxy.schema.fields import EncodedDatabaseIdField
+
+taggable_item_names = {item: item for item in ItemTagAssociation.associated_item_names}
+# This Enum is generated dynamically and mypy can not statically infer it's real type
+# so it should be ignored. See:https://github.com/python/mypy/issues/4865#issuecomment-592560696
+TaggableItemClass = Enum('TaggableItemClass', taggable_item_names)  # type: ignore
+
+
+class ItemTagsPayload(BaseModel):
+    item_id: EncodedDatabaseIdField = Field(
+        ...,  # This field is required
+        title="Item ID",
+        description="The `encoded identifier` of the item whose tags will be updated.",
+    )
+    item_class: TaggableItemClass = Field(
+        ...,  # This field is required
+        title="Item class",
+        description="The name of the class of the item that will be tagged.",
+    )
+    item_tags: Optional[List[str]] = Field(
+        default=None,
+        title="Item tags",
+        description="The list of tags that will replace the current tags associated with the item.",
+    )
+
+    class Config:
+        use_enum_values = True
+
+
+class TagsManager:
+    """Interface/service object shared by controllers for interacting with tags."""
+
+    def update(self, trans: ProvidesUserContext, payload: ItemTagsPayload) -> None:
+        """Apply a new set of tags to an item; previous tags are deleted."""
+        tag_handler = trans.app.tag_handler
+        new_tags: Optional[str] = None
+        if payload.item_tags and len(payload.item_tags) > 0:
+            new_tags = ",".join(payload.item_tags)
+        item = self._get_item(trans, payload)
+        user = trans.user
+        tag_handler.delete_item_tags(user, item)
+        tag_handler.apply_item_tags(user, item, new_tags)
+        trans.sa_session.flush()
+
+    def _get_item(self, trans: ProvidesUserContext, payload: ItemTagsPayload):
+        """
+        Get an item based on type and id.
+        """
+        tag_handler = trans.app.tag_handler
+        id = trans.security.decode_id(payload.item_id)
+        item_class = tag_handler.item_tag_assoc_info[payload.item_class].item_class
+        item = trans.sa_session.query(item_class).filter(item_class.id == id).first()
+        return item
diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py
index 643aa8628c3..1692884bef5 100644
--- a/lib/galaxy/model/__init__.py
+++ b/lib/galaxy/model/__init__.py
@@ -19,7 +19,7 @@ from collections import defaultdict
 from datetime import datetime, timedelta
 from enum import Enum
 from string import Template
-from typing import Optional, TYPE_CHECKING
+from typing import List, Optional, TYPE_CHECKING
 from uuid import UUID, uuid4
 
 from boltons.iterutils import remap
@@ -6350,6 +6350,11 @@ class Tag(RepresentById):
 class ItemTagAssociation(Dictifiable):
     dict_collection_visible_keys = ['id', 'user_tname', 'user_value']
     dict_element_visible_keys = dict_collection_visible_keys
+    associated_item_names: List[str] = []
+
+    def __init_subclass__(cls, **kwargs):
+        super().__init_subclass__(**kwargs)
+        cls.associated_item_names.append(cls.__name__.replace("TagAssociation", ""))
 
     def __init__(self, id=None, user=None, item_id=None, tag_id=None, user_tname=None, value=None):
         self.id = id
diff --git a/lib/galaxy/webapps/base/controller.py b/lib/galaxy/webapps/base/controller.py
index eac3e994319..2be1a97d171 100644
--- a/lib/galaxy/webapps/base/controller.py
+++ b/lib/galaxy/webapps/base/controller.py
@@ -1515,7 +1515,7 @@ class UsesQuotaMixin:
 
 class UsesTagsMixin(SharableItemSecurityMixin):
 
-    def get_tag_handler(self, trans):
+    def get_tag_handler(self, trans) -> tags.GalaxyTagHandler:
         return trans.app.tag_handler
 
     def _get_user_tags(self, trans, item_class_name, id):
diff --git a/lib/galaxy/webapps/galaxy/api/tags.py b/lib/galaxy/webapps/galaxy/api/tags.py
index 9f0639f486c..aa36d091389 100644
--- a/lib/galaxy/webapps/galaxy/api/tags.py
+++ b/lib/galaxy/webapps/galaxy/api/tags.py
@@ -3,43 +3,74 @@ API Controller providing Galaxy Tags
 """
 import logging
 
-from galaxy.exceptions import MessageException
+from fastapi import (
+    Body,
+    status,
+)
+# TODO: replace with Router after merging #11219
+from fastapi_utils.cbv import cbv
+from fastapi_utils.inferring_router import InferringRouter as APIRouter
+
+from galaxy.managers.context import ProvidesUserContext
+from galaxy.managers.tags import (
+    ItemTagsPayload,
+    TagsManager,
+)
 from galaxy.web import expose_api
-from galaxy.webapps.base.controller import BaseAPIController, UsesTagsMixin
-from galaxy.webapps.base.webapp import GalaxyWebTransaction
+from galaxy.webapps.base.controller import BaseAPIController
+from . import (
+    Depends,
+    get_trans,
+)
 
 log = logging.getLogger(__name__)
 
+# TODO: This FastAPI router is disabled. Please rename it to `router` when the database session issues are fixed.
+_router = APIRouter(tags=['tags'])
 
-class TagsController(BaseAPIController, UsesTagsMixin):
+
+def get_tags_manager() -> TagsManager:
+    return TagsManager()  # TODO: remove/refactor after merging #11180
+
+
+@cbv(_router)
+class FastAPITags:
+    manager: TagsManager = Depends(get_tags_manager)
+
+    @_router.put(
+        '/api/tags',
+        summary="Apply a new set of tags to an item.",
+        status_code=status.HTTP_204_NO_CONTENT,
+    )
+    def update(
+        self,
+        trans: ProvidesUserContext = Depends(get_trans),
+        payload: ItemTagsPayload = Body(
+            ...,  # Required
+            title="Payload",
+            description="Request body containing the item and the tags to be assigned.",
+        ),
+    ):
+        """Replaces the tags associated with an item with the new ones specified in the payload.
+
+        - The previous tags will be __deleted__.
+        - If no tags are provided in the request body, the currently associated tags will also be __deleted__.
+        """
+        self.manager.update(trans, payload)
+
+
+class TagsController(BaseAPIController):
+
+    def __init__(self, app):
+        super().__init__(app)
+        self.manager = TagsManager()
 
     # Retag an item. All previous tags are deleted and new tags are applied.
     @expose_api
-    def update(self, trans: GalaxyWebTransaction, payload: dict, **kwd):
+    def update(self, trans: ProvidesUserContext, payload: dict, **kwd):
         """
         PUT /api/tags/
 
         Apply a new set of tags to an item; previous tags are deleted.
         """
-        item_id = payload.get("item_id")
-        item_class = payload.get("item_class")
-        item_tags = payload.get("item_tags")
-        if item_id is None:
-            raise MessageException("Please provide the item id (item_id).")
-        if item_class is None:
-            raise MessageException("Please provide the item class (item_class).")
-        if item_tags and len(item_tags) > 0:
-            item_tags = ",".join(item_tags)
-        item = self._get_item(trans, item_class, trans.security.decode_id(item_id))
-        user = trans.user
-        self.get_tag_handler(trans).delete_item_tags(user, item)
-        self.get_tag_handler(trans).apply_item_tags(user, item, item_tags)
-        trans.sa_session.flush()
-
-    def _get_item(self, trans: GalaxyWebTransaction, item_class_name, id):
-        """
-        Get an item based on type and id.
-        """
-        item_class = self.get_tag_handler(trans).item_tag_assoc_info[item_class_name].item_class
-        item = trans.sa_session.query(item_class).filter(item_class.id == id).first()
-        return item
+        self.manager.update(trans, ItemTagsPayload(**payload))
diff --git a/lib/galaxy/webapps/galaxy/fast_app.py b/lib/galaxy/webapps/galaxy/fast_app.py
index 82ab083e740..d5bd24f7130 100644
--- a/lib/galaxy/webapps/galaxy/fast_app.py
+++ b/lib/galaxy/webapps/galaxy/fast_app.py
@@ -28,6 +28,10 @@ api_tags_metadata = [
         "name": "licenses",
         "description": "Operations with [SPDX licenses](https://spdx.org/licenses/).",
     },
+    {
+        "name": "tags",
+        "description": "Operations with tags.",
+    },
     {
         "name": "tool data tables",
         "description": "Operations with tool [Data Tables](https://galaxyproject.org/admin/tools/data-tables/).",
diff --git a/lib/galaxy_test/api/test_datasets.py b/lib/galaxy_test/api/test_datasets.py
index 5e4a98768f7..38755e66ba3 100644
--- a/lib/galaxy_test/api/test_datasets.py
+++ b/lib/galaxy_test/api/test_datasets.py
@@ -1,3 +1,4 @@
+import json
 import textwrap
 
 from galaxy_test.base.populators import (
@@ -122,12 +123,13 @@ class DatasetsApiTestCase(ApiTestCase):
 
     def test_tag_change(self):
         hda_id = self.dataset_populator.new_dataset(self.history_id)['id']
-        payload = {
+        payload = json.dumps({
             'item_id': hda_id,
             'item_class': 'HistoryDatasetAssociation',
             'item_tags': ['cool:tag_a', 'cool:tag_b', 'tag_c', 'name:tag_d', '#tag_e'],
-        }
-        self._put("tags", payload).json()
+        })
+        put_response = self._put("tags", payload)
+        self._assert_status_code_is_ok(put_response)
         updated_hda = self._get(
             f"histories/{self.history_id}/contents/{hda_id}").json()
         assert 'cool:tag_a' in updated_hda['tags']