diff --git a/dbm-ui/backend/db_periodic_task/local_tasks/db_meta/sync_cluster_stat.py b/dbm-ui/backend/db_periodic_task/local_tasks/db_meta/sync_cluster_stat.py
index eb55e92e58..26c678d048 100644
--- a/dbm-ui/backend/db_periodic_task/local_tasks/db_meta/sync_cluster_stat.py
+++ b/dbm-ui/backend/db_periodic_task/local_tasks/db_meta/sync_cluster_stat.py
@@ -127,6 +127,8 @@ def sync_cluster_stat_by_cluster_type(bk_biz_id, cluster_type):
         f"{CACHE_CLUSTER_STATS}_{bk_biz_id}_{cluster_type}", json.dumps(cluster_stats), timeout=2 * TimeUnit.HOUR
     )
 
+    return cluster_stats
+
 
 @register_periodic_task(run_every=crontab(hour="*/1", minute=0))
 def sync_cluster_stat_from_monitor():
diff --git a/dbm-ui/backend/db_services/dbbase/serializers.py b/dbm-ui/backend/db_services/dbbase/serializers.py
index 9c91ef7550..0ca48a47d7 100644
--- a/dbm-ui/backend/db_services/dbbase/serializers.py
+++ b/dbm-ui/backend/db_services/dbbase/serializers.py
@@ -194,3 +194,13 @@ class ClusterDbTypeSerializer(serializers.Serializer):
 
 class QueryClusterInstanceCountSerializer(serializers.Serializer):
     bk_biz_id = serializers.IntegerField(help_text=_("业务ID"))
+
+
+class QueryClusterCapSerializer(serializers.Serializer):
+    bk_biz_id = serializers.IntegerField(help_text=_("业务ID"))
+    cluster_type = serializers.ChoiceField(help_text=_("集群类型"), choices=ClusterType.get_choices())
+
+
+class QueryClusterCapResponseSerializer(serializers.Serializer):
+    class Meta:
+        swagger_schema_fields = {"example": {"cluster1": {"used": 1, "total": 2, "in_use": 50}}}
diff --git a/dbm-ui/backend/db_services/dbbase/views.py b/dbm-ui/backend/db_services/dbbase/views.py
index 5e54ee291d..5383769e5f 100644
--- a/dbm-ui/backend/db_services/dbbase/views.py
+++ b/dbm-ui/backend/db_services/dbbase/views.py
@@ -40,6 +40,8 @@
     QueryAllTypeClusterSerializer,
     QueryBizClusterAttrsResponseSerializer,
     QueryBizClusterAttrsSerializer,
+    QueryClusterCapResponseSerializer,
+    QueryClusterCapSerializer,
     QueryClusterInstanceCountSerializer,
     ResourceAdministrationSerializer,
     WebConsoleResponseSerializer,
@@ -312,7 +314,6 @@ def get_ips_list(self, request, *args, **kwargs):
     )
     @action(methods=["GET"], detail=False, serializer_class=QueryClusterInstanceCountSerializer)
     def query_cluster_instance_count(self, request, *args, **kwargs):
-
         validate_data = self.params_validate(self.get_serializer_class())
         cluster_queryset = Cluster.objects.filter(**validate_data)
         storage_instance_queryset = StorageInstance.objects.filter(**validate_data)
@@ -353,11 +354,11 @@ def list_to_dict(type_counts):
 
         # 构建最终输出格式，包含所有 ClusterType 成员
         cluster_type_count_map = {}
-        for cluster_type in ClusterType:
-            if cluster_type.value not in redis_cluster_types:
-                cluster_count = cluster_type_dict.get(cluster_type.value, 0)
-                instance_count = instance_count_dict.get(cluster_type.value, 0)
-                cluster_type_count_map[cluster_type.value] = {
+        for cluster_type in ClusterType.get_values():
+            if cluster_type not in redis_cluster_types:
+                cluster_count = cluster_type_dict.get(cluster_type, 0)
+                instance_count = instance_count_dict.get(cluster_type, 0)
+                cluster_type_count_map[cluster_type] = {
                     "cluster_count": cluster_count,
                     "instance_count": instance_count,
                 }
@@ -367,3 +368,18 @@ def list_to_dict(type_counts):
         }
 
         return Response(cluster_type_count_map)
+
+    @common_swagger_auto_schema(
+        operation_summary=_("查询集群容量"),
+        auto_schema=ResponseSwaggerAutoSchema,
+        query_serializer=QueryClusterCapSerializer(),
+        responses={status.HTTP_200_OK: QueryClusterCapResponseSerializer()},
+        tags=[SWAGGER_TAG],
+    )
+    @action(methods=["GET"], detail=False, serializer_class=QueryClusterCapSerializer, pagination_class=None)
+    def query_cluster_stat(self, request, *args, **kwargs):
+        from backend.db_periodic_task.local_tasks.db_meta.sync_cluster_stat import sync_cluster_stat_by_cluster_type
+
+        data = self.params_validate(self.get_serializer_class())
+        cluster_stat_map = sync_cluster_stat_by_cluster_type(data["bk_biz_id"], data["cluster_type"])
+        return Response(cluster_stat_map)