diff --git a/api/controllers/console/tag/tags.py b/api/controllers/console/tag/tags.py
index e828d54ff4..bc0776f658 100644
--- a/api/controllers/console/tag/tags.py
+++ b/api/controllers/console/tag/tags.py
@@ -1,14 +1,27 @@
 from typing import Literal
-from uuid import UUID
 
+from flask import request
+from flask_restx import Namespace, Resource, fields, marshal_with
 from pydantic import BaseModel, Field
 from werkzeug.exceptions import Forbidden
 
+from controllers.common.schema import register_schema_models
+from controllers.console import console_ns
 from controllers.console.wraps import account_initialization_required, edit_permission_required, setup_required
-from controllers.fastopenapi import console_router
 from libs.login import current_account_with_tenant, login_required
 from services.tag_service import TagService
 
+dataset_tag_fields = {
+    "id": fields.String,
+    "name": fields.String,
+    "type": fields.String,
+    "binding_count": fields.String,
+}
+
+
+def build_dataset_tag_fields(api_or_ns: Namespace):
+    return api_or_ns.model("DataSetTag", dataset_tag_fields)
+
 
 class TagBasePayload(BaseModel):
     name: str = Field(description="Tag name", min_length=1, max_length=50)
@@ -32,129 +45,115 @@ class TagListQueryParam(BaseModel):
     keyword: str | None = Field(None, description="Search keyword")
 
 
-class TagResponse(BaseModel):
-    id: str = Field(description="Tag ID")
-    name: str = Field(description="Tag name")
-    type: str = Field(description="Tag type")
-    binding_count: int = Field(description="Number of bindings")
-
-
-class TagBindingResult(BaseModel):
-    result: Literal["success"] = Field(description="Operation result", examples=["success"])
-
-
-@console_router.get(
-    "/tags",
-    response_model=list[TagResponse],
-    tags=["console"],
+register_schema_models(
+    console_ns,
+    TagBasePayload,
+    TagBindingPayload,
+    TagBindingRemovePayload,
+    TagListQueryParam,
 )
-@setup_required
-@login_required
-@account_initialization_required
-def list_tags(query: TagListQueryParam) -> list[TagResponse]:
-    _, current_tenant_id = current_account_with_tenant()
-    tags = TagService.get_tags(query.type, current_tenant_id, query.keyword)
-
-    return [
-        TagResponse(
-            id=tag.id,
-            name=tag.name,
-            type=tag.type,
-            binding_count=int(tag.binding_count),
-        )
-        for tag in tags
-    ]
 
 
-@console_router.post(
-    "/tags",
-    response_model=TagResponse,
-    tags=["console"],
-)
-@setup_required
-@login_required
-@account_initialization_required
-def create_tag(payload: TagBasePayload) -> TagResponse:
-    current_user, _ = current_account_with_tenant()
-    # The role of the current user in the tag table must be admin, owner, or editor
-    if not (current_user.has_edit_permission or current_user.is_dataset_editor):
-        raise Forbidden()
+@console_ns.route("/tags")
+class TagListApi(Resource):
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @console_ns.doc(
+        params={"type": 'Tag type filter. Can be "knowledge" or "app".', "keyword": "Search keyword for tag name."}
+    )
+    @marshal_with(dataset_tag_fields)
+    def get(self):
+        _, current_tenant_id = current_account_with_tenant()
+        raw_args = request.args.to_dict()
+        param = TagListQueryParam.model_validate(raw_args)
+        tags = TagService.get_tags(param.type, current_tenant_id, param.keyword)
 
-    tag = TagService.save_tags(payload.model_dump())
+        return tags, 200
 
-    return TagResponse(id=tag.id, name=tag.name, type=tag.type, binding_count=0)
+    @console_ns.expect(console_ns.models[TagBasePayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        current_user, _ = current_account_with_tenant()
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
+            raise Forbidden()
+
+        payload = TagBasePayload.model_validate(console_ns.payload or {})
+        tag = TagService.save_tags(payload.model_dump())
+
+        response = {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": 0}
+
+        return response, 200
 
 
-@console_router.patch(
-    "/tags/<uuid:tag_id>",
-    response_model=TagResponse,
-    tags=["console"],
-)
-@setup_required
-@login_required
-@account_initialization_required
-def update_tag(tag_id: UUID, payload: TagBasePayload) -> TagResponse:
-    current_user, _ = current_account_with_tenant()
-    tag_id_str = str(tag_id)
-    # The role of the current user in the ta table must be admin, owner, or editor
-    if not (current_user.has_edit_permission or current_user.is_dataset_editor):
-        raise Forbidden()
+@console_ns.route("/tags/<uuid:tag_id>")
+class TagUpdateDeleteApi(Resource):
+    @console_ns.expect(console_ns.models[TagBasePayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def patch(self, tag_id):
+        current_user, _ = current_account_with_tenant()
+        tag_id = str(tag_id)
+        # The role of the current user in the ta table must be admin, owner, or editor
+        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
+            raise Forbidden()
 
-    tag = TagService.update_tags(payload.model_dump(), tag_id_str)
+        payload = TagBasePayload.model_validate(console_ns.payload or {})
+        tag = TagService.update_tags(payload.model_dump(), tag_id)
 
-    binding_count = TagService.get_tag_binding_count(tag_id_str)
+        binding_count = TagService.get_tag_binding_count(tag_id)
 
-    return TagResponse(id=tag.id, name=tag.name, type=tag.type, binding_count=binding_count)
+        response = {"id": tag.id, "name": tag.name, "type": tag.type, "binding_count": binding_count}
+
+        return response, 200
+
+    @setup_required
+    @login_required
+    @account_initialization_required
+    @edit_permission_required
+    def delete(self, tag_id):
+        tag_id = str(tag_id)
+
+        TagService.delete_tag(tag_id)
+
+        return 204
 
 
-@console_router.delete(
-    "/tags/<uuid:tag_id>",
-    tags=["console"],
-    status_code=204,
-)
-@setup_required
-@login_required
-@account_initialization_required
-@edit_permission_required
-def delete_tag(tag_id: UUID) -> None:
-    tag_id_str = str(tag_id)
+@console_ns.route("/tag-bindings/create")
+class TagBindingCreateApi(Resource):
+    @console_ns.expect(console_ns.models[TagBindingPayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        current_user, _ = current_account_with_tenant()
+        # The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
+        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
+            raise Forbidden()
 
-    TagService.delete_tag(tag_id_str)
+        payload = TagBindingPayload.model_validate(console_ns.payload or {})
+        TagService.save_tag_binding(payload.model_dump())
+
+        return {"result": "success"}, 200
 
 
-@console_router.post(
-    "/tag-bindings/create",
-    response_model=TagBindingResult,
-    tags=["console"],
-)
-@setup_required
-@login_required
-@account_initialization_required
-def create_tag_binding(payload: TagBindingPayload) -> TagBindingResult:
-    current_user, _ = current_account_with_tenant()
-    # The role of the current user in the tag table must be admin, owner, editor, or dataset_operator
-    if not (current_user.has_edit_permission or current_user.is_dataset_editor):
-        raise Forbidden()
+@console_ns.route("/tag-bindings/remove")
+class TagBindingDeleteApi(Resource):
+    @console_ns.expect(console_ns.models[TagBindingRemovePayload.__name__])
+    @setup_required
+    @login_required
+    @account_initialization_required
+    def post(self):
+        current_user, _ = current_account_with_tenant()
+        # The role of the current user in the ta table must be admin, owner, editor, or dataset_operator
+        if not (current_user.has_edit_permission or current_user.is_dataset_editor):
+            raise Forbidden()
 
-    TagService.save_tag_binding(payload.model_dump())
+        payload = TagBindingRemovePayload.model_validate(console_ns.payload or {})
+        TagService.delete_tag_binding(payload.model_dump())
 
-    return TagBindingResult(result="success")
-
-
-@console_router.post(
-    "/tag-bindings/remove",
-    response_model=TagBindingResult,
-    tags=["console"],
-)
-@setup_required
-@login_required
-@account_initialization_required
-def delete_tag_binding(payload: TagBindingRemovePayload) -> TagBindingResult:
-    current_user, _ = current_account_with_tenant()
-    # The role of the current user in the tag table must be admin, owner, editor, or dataset_operator
-    if not (current_user.has_edit_permission or current_user.is_dataset_editor):
-        raise Forbidden()
-
-    TagService.delete_tag_binding(payload.model_dump())
-
-    return TagBindingResult(result="success")
+        return {"result": "success"}, 200
diff --git a/api/services/tag_service.py b/api/services/tag_service.py
index 56f4ae9494..bd3585acf4 100644
--- a/api/services/tag_service.py
+++ b/api/services/tag_service.py
@@ -24,7 +24,7 @@ class TagService:
             escaped_keyword = escape_like_pattern(keyword)
             query = query.where(sa.and_(Tag.name.ilike(f"%{escaped_keyword}%", escape="\\")))
         query = query.group_by(Tag.id, Tag.type, Tag.name, Tag.created_at)
-        results = query.order_by(Tag.created_at.desc()).all()
+        results: list = query.order_by(Tag.created_at.desc()).all()
         return results
 
     @staticmethod
diff --git a/api/tests/unit_tests/controllers/console/test_fastopenapi_tags.py b/api/tests/unit_tests/controllers/console/test_fastopenapi_tags.py
deleted file mode 100644
index 62d143f32d..0000000000
--- a/api/tests/unit_tests/controllers/console/test_fastopenapi_tags.py
+++ /dev/null
@@ -1,222 +0,0 @@
-import builtins
-import contextlib
-import importlib
-import sys
-from types import SimpleNamespace
-from unittest.mock import MagicMock, patch
-
-import pytest
-from flask import Flask
-from flask.views import MethodView
-
-from extensions import ext_fastopenapi
-from extensions.ext_database import db
-
-
-@pytest.fixture
-def app():
-    app = Flask(__name__)
-    app.config["TESTING"] = True
-    app.config["SECRET_KEY"] = "test-secret"
-    app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
-
-    db.init_app(app)
-
-    return app
-
-
-@pytest.fixture(autouse=True)
-def fix_method_view_issue(monkeypatch):
-    if not hasattr(builtins, "MethodView"):
-        monkeypatch.setattr(builtins, "MethodView", MethodView, raising=False)
-
-
-def _create_isolated_router():
-    import controllers.fastopenapi
-
-    router_class = type(controllers.fastopenapi.console_router)
-    return router_class()
-
-
-@contextlib.contextmanager
-def _patch_auth_and_router(temp_router):
-    def noop(func):
-        return func
-
-    default_user = MagicMock(has_edit_permission=True, is_dataset_editor=False)
-
-    with (
-        patch("controllers.fastopenapi.console_router", temp_router),
-        patch("extensions.ext_fastopenapi.console_router", temp_router),
-        patch("controllers.console.wraps.setup_required", side_effect=noop),
-        patch("libs.login.login_required", side_effect=noop),
-        patch("controllers.console.wraps.account_initialization_required", side_effect=noop),
-        patch("controllers.console.wraps.edit_permission_required", side_effect=noop),
-        patch("libs.login.current_account_with_tenant", return_value=(default_user, "tenant-id")),
-        patch("configs.dify_config.EDITION", "CLOUD"),
-    ):
-        import extensions.ext_fastopenapi
-
-        importlib.reload(extensions.ext_fastopenapi)
-
-        yield
-
-
-def _force_reload_module(target_module: str, alias_module: str):
-    if target_module in sys.modules:
-        del sys.modules[target_module]
-    if alias_module in sys.modules:
-        del sys.modules[alias_module]
-
-    module = importlib.import_module(target_module)
-    sys.modules[alias_module] = sys.modules[target_module]
-
-    return module
-
-
-def _dedupe_routes(router):
-    seen = set()
-    unique_routes = []
-    for path, method, endpoint in reversed(router.get_routes()):
-        key = (path, method, endpoint.__name__)
-        if key in seen:
-            continue
-        seen.add(key)
-        unique_routes.append((path, method, endpoint))
-    router._routes = list(reversed(unique_routes))
-
-
-def _cleanup_modules(target_module: str, alias_module: str):
-    if target_module in sys.modules:
-        del sys.modules[target_module]
-    if alias_module in sys.modules:
-        del sys.modules[alias_module]
-
-
-@pytest.fixture
-def mock_tags_module_env():
-    target_module = "controllers.console.tag.tags"
-    alias_module = "api.controllers.console.tag.tags"
-    temp_router = _create_isolated_router()
-
-    try:
-        with _patch_auth_and_router(temp_router):
-            tags_module = _force_reload_module(target_module, alias_module)
-            _dedupe_routes(temp_router)
-            yield tags_module
-    finally:
-        _cleanup_modules(target_module, alias_module)
-
-
-def test_list_tags_success(app: Flask, mock_tags_module_env):
-    # Arrange
-    tag = SimpleNamespace(id="tag-1", name="Alpha", type="app", binding_count=2)
-    with patch("controllers.console.tag.tags.TagService.get_tags", return_value=[tag]):
-        ext_fastopenapi.init_app(app)
-        client = app.test_client()
-
-        # Act
-        response = client.get("/console/api/tags?type=app&keyword=Alpha")
-
-    # Assert
-    assert response.status_code == 200
-    assert response.get_json() == [
-        {"id": "tag-1", "name": "Alpha", "type": "app", "binding_count": 2},
-    ]
-
-
-def test_create_tag_success(app: Flask, mock_tags_module_env):
-    # Arrange
-    tag = SimpleNamespace(id="tag-2", name="Beta", type="app")
-    with patch("controllers.console.tag.tags.TagService.save_tags", return_value=tag) as mock_save:
-        ext_fastopenapi.init_app(app)
-        client = app.test_client()
-
-        # Act
-        response = client.post("/console/api/tags", json={"name": "Beta", "type": "app"})
-
-    # Assert
-    assert response.status_code == 200
-    assert response.get_json() == {
-        "id": "tag-2",
-        "name": "Beta",
-        "type": "app",
-        "binding_count": 0,
-    }
-    mock_save.assert_called_once_with({"name": "Beta", "type": "app"})
-
-
-def test_update_tag_success(app: Flask, mock_tags_module_env):
-    # Arrange
-    tag = SimpleNamespace(id="tag-3", name="Gamma", type="app")
-    with (
-        patch("controllers.console.tag.tags.TagService.update_tags", return_value=tag) as mock_update,
-        patch("controllers.console.tag.tags.TagService.get_tag_binding_count", return_value=4),
-    ):
-        ext_fastopenapi.init_app(app)
-        client = app.test_client()
-
-        # Act
-        response = client.patch(
-            "/console/api/tags/11111111-1111-1111-1111-111111111111",
-            json={"name": "Gamma", "type": "app"},
-        )
-
-    # Assert
-    assert response.status_code == 200
-    assert response.get_json() == {
-        "id": "tag-3",
-        "name": "Gamma",
-        "type": "app",
-        "binding_count": 4,
-    }
-    mock_update.assert_called_once_with(
-        {"name": "Gamma", "type": "app"},
-        "11111111-1111-1111-1111-111111111111",
-    )
-
-
-def test_delete_tag_success(app: Flask, mock_tags_module_env):
-    # Arrange
-    with patch("controllers.console.tag.tags.TagService.delete_tag") as mock_delete:
-        ext_fastopenapi.init_app(app)
-        client = app.test_client()
-
-        # Act
-        response = client.delete("/console/api/tags/11111111-1111-1111-1111-111111111111")
-
-    # Assert
-    assert response.status_code == 204
-    mock_delete.assert_called_once_with("11111111-1111-1111-1111-111111111111")
-
-
-def test_create_tag_binding_success(app: Flask, mock_tags_module_env):
-    # Arrange
-    payload = {"tag_ids": ["tag-1", "tag-2"], "target_id": "target-1", "type": "app"}
-    with patch("controllers.console.tag.tags.TagService.save_tag_binding") as mock_bind:
-        ext_fastopenapi.init_app(app)
-        client = app.test_client()
-
-        # Act
-        response = client.post("/console/api/tag-bindings/create", json=payload)
-
-    # Assert
-    assert response.status_code == 200
-    assert response.get_json() == {"result": "success"}
-    mock_bind.assert_called_once_with(payload)
-
-
-def test_delete_tag_binding_success(app: Flask, mock_tags_module_env):
-    # Arrange
-    payload = {"tag_id": "tag-1", "target_id": "target-1", "type": "app"}
-    with patch("controllers.console.tag.tags.TagService.delete_tag_binding") as mock_unbind:
-        ext_fastopenapi.init_app(app)
-        client = app.test_client()
-
-        # Act
-        response = client.post("/console/api/tag-bindings/remove", json=payload)
-
-    # Assert
-    assert response.status_code == 200
-    assert response.get_json() == {"result": "success"}
-    mock_unbind.assert_called_once_with(payload)