diff --git a/apps/backend/components/collections/agent_new/base.py b/apps/backend/components/collections/agent_new/base.py
index 2aed95485..48c01ca02 100644
--- a/apps/backend/components/collections/agent_new/base.py
+++ b/apps/backend/components/collections/agent_new/base.py
@@ -34,6 +34,18 @@
 from ..base import BaseService, CommonData
 
 
+@dataclass
+class AgentCommonData(CommonData):
+    # 默认接入点
+    default_ap: models.AccessPoint
+    # 主机ID - 接入点 映射关系
+    host_id__ap_map: Dict[int, models.AccessPoint]
+    # AgentStep 适配器
+    agent_step_adapter: AgentStepAdapter
+    # 注入AP_ID
+    injected_ap_id: int
+
+
 class AgentBaseService(BaseService, metaclass=abc.ABCMeta):
     """
     AGENT安装基类
@@ -305,17 +317,13 @@ def maintain_agent_proc_status_uniqueness(self, bk_host_ids: Set[int]) -> None:
             proc_statuses_to_be_created.append(models.ProcessStatus(bk_host_id=host_id, **self.agent_proc_common_data))
         models.ProcessStatus.objects.bulk_create(proc_statuses_to_be_created, batch_size=self.batch_size)
 
-
-@dataclass
-class AgentCommonData(CommonData):
-    # 默认接入点
-    default_ap: models.AccessPoint
-    # 主机ID - 接入点 映射关系
-    host_id__ap_map: Dict[int, models.AccessPoint]
-    # AgentStep 适配器
-    agent_step_adapter: AgentStepAdapter
-    # 注入AP_ID
-    injected_ap_id: int
+    def get_host_ap(self, common_data: AgentCommonData, host: models.Host) -> models.AccessPoint:
+        # 优先使用注入的AP ID
+        if common_data.injected_ap_id:
+            host_ap: models.AccessPoint = common_data.ap_id_obj_map[common_data.injected_ap_id]
+        else:
+            host_ap: models.AccessPoint = common_data.host_id__ap_map[host.bk_host_id]
+        return host_ap
 
 
 class RetryHandler:
diff --git a/apps/backend/components/collections/agent_new/check_agent_ability.py b/apps/backend/components/collections/agent_new/check_agent_ability.py
index 27f729e71..a0f54153e 100644
--- a/apps/backend/components/collections/agent_new/check_agent_ability.py
+++ b/apps/backend/components/collections/agent_new/check_agent_ability.py
@@ -31,7 +31,8 @@ def get_script_content(self, data, common_data: AgentCommonData, host: models.Ho
         path_handler: PathHandler = PathHandler(host.os_type)
         ctl_exe_name: str = ("gse_agent", "gse_agent.exe")[host.os_type == constants.OsType.WINDOWS]
         general_node_type: str = self.get_general_node_type(host.node_type)
-        setup_path: str = common_data.host_id__ap_map[host.bk_host_id].get_agent_config(host.os_type)["setup_path"]
+        host_ap: models.AccessPoint = self.get_host_ap(common_data=common_data, host=host)
+        setup_path: str = host_ap.get_agent_config(host.os_type)["setup_path"]
         agent_path: str = path_handler.join(setup_path, general_node_type, "bin", ctl_exe_name)
 
         return f"{agent_path} --version"
diff --git a/apps/backend/components/collections/agent_new/push_upgrade_package.py b/apps/backend/components/collections/agent_new/push_upgrade_package.py
index 72c9d0cb2..a4f33512b 100644
--- a/apps/backend/components/collections/agent_new/push_upgrade_package.py
+++ b/apps/backend/components/collections/agent_new/push_upgrade_package.py
@@ -21,14 +21,16 @@
 
 class PushUpgradeFileService(AgentTransferFileService):
     def get_file_target_path(self, data, common_data: AgentCommonData, host: models.Host) -> str:
-        return common_data.host_id__ap_map[host.bk_host_id].get_agent_config(host.os_type)["temp_path"]
+        host_ap = self.get_host_ap(common_data=common_data, host=host)
+        return host_ap.get_agent_config(host.os_type)["temp_path"]
 
     def get_upgrade_package_source_path(self, common_data: AgentCommonData, host: models.Host) -> Tuple[str, str]:
         """
         获取升级包源路径
         """
+        host_ap = self.get_host_ap(common_data=common_data, host=host)
         # 1.x 升级到 1.x，使用老到路径，升级包直接放在 download 目录下
-        agent_path = root_path = common_data.host_id__ap_map[host.bk_host_id].nginx_path or settings.DOWNLOAD_PATH
+        agent_path = root_path = host_ap.nginx_path or settings.DOWNLOAD_PATH
         if not common_data.agent_step_adapter.is_legacy:
             # 2.x 升级到 2.x，根据操作系统、CPU 架构等组合路径
             agent_path = os.path.join(root_path, "agent", host.os_type.lower(), host.cpu_arch.lower())
diff --git a/apps/backend/components/collections/agent_new/reload_agent_config.py b/apps/backend/components/collections/agent_new/reload_agent_config.py
index 61b4a4441..fa9a694af 100644
--- a/apps/backend/components/collections/agent_new/reload_agent_config.py
+++ b/apps/backend/components/collections/agent_new/reload_agent_config.py
@@ -37,7 +37,8 @@ def get_script_content(self, data, common_data: AgentCommonData, host: models.Ho
         # 路径处理器
         path_handler = PathHandler(host.os_type)
         general_node_type = self.get_general_node_type(host.node_type)
-        setup_path = common_data.host_id__ap_map[host.bk_host_id].get_agent_config(host.os_type)["setup_path"]
+        host_ap: models.AccessPoint = self.get_host_ap(common_data=common_data, host=host)
+        setup_path = host_ap.get_agent_config(host.os_type)["setup_path"]
         agent_path = path_handler.join(setup_path, general_node_type, "bin")
         if common_data.agent_step_adapter.is_legacy:
             return f"cd {agent_path} && ./gse_agent --reload"
diff --git a/apps/backend/components/collections/agent_new/render_and_push_gse_config.py b/apps/backend/components/collections/agent_new/render_and_push_gse_config.py
index 0db4f0657..25ffaaf94 100644
--- a/apps/backend/components/collections/agent_new/render_and_push_gse_config.py
+++ b/apps/backend/components/collections/agent_new/render_and_push_gse_config.py
@@ -21,13 +21,15 @@ class RenderAndPushGseConfigService(AgentPushConfigService):
     def get_config_info_list(self, data, common_data: AgentCommonData, host: models.Host) -> List[Dict[str, Any]]:
         file_name = common_data.agent_step_adapter.get_main_config_filename()
         general_node_type = self.get_general_node_type(host.node_type)
+        host_ap: models.AccessPoint = self.get_host_ap(common_data=common_data, host=host)
         content = common_data.agent_step_adapter.get_config(
-            host=host, filename=file_name, node_type=general_node_type, ap=common_data.host_id__ap_map[host.bk_host_id]
+            host=host, filename=file_name, node_type=general_node_type, ap=host_ap
         )
         return [{"file_name": file_name, "content": content}]
 
     def get_file_target_path(self, data, common_data: AgentCommonData, host: models.Host) -> str:
         general_node_type = self.get_general_node_type(host.node_type)
         path_handler = PathHandler(host.os_type)
-        setup_path = common_data.host_id__ap_map[host.bk_host_id].get_agent_config(host.os_type)["setup_path"]
+        host_ap: models.AccessPoint = self.get_host_ap(common_data=common_data, host=host)
+        setup_path = host_ap.get_agent_config(host.os_type)["setup_path"]
         return path_handler.join(setup_path, general_node_type, "etc")
diff --git a/apps/backend/components/collections/agent_new/restart.py b/apps/backend/components/collections/agent_new/restart.py
index ea01b3e59..bc23e2761 100644
--- a/apps/backend/components/collections/agent_new/restart.py
+++ b/apps/backend/components/collections/agent_new/restart.py
@@ -33,7 +33,8 @@ def get_script_content(self, data, common_data: AgentCommonData, host: models.Ho
         ctl_exe_name = ("gsectl", "gsectl.bat")[host.os_type == constants.OsType.WINDOWS]
         cmd_suffix = ("restart >/dev/null 2>&1", "restart")[host.os_type == constants.OsType.WINDOWS]
         general_node_type = self.get_general_node_type(host.node_type)
-        setup_path = common_data.host_id__ap_map[host.bk_host_id].get_agent_config(host.os_type)["setup_path"]
+        host_ap: models.AccessPoint = self.get_host_ap(common_data=common_data, host=host)
+        setup_path = host_ap.get_agent_config(host.os_type)["setup_path"]
         agent_path = path_handler.join(setup_path, general_node_type, "bin", ctl_exe_name)
 
         return f"{agent_path} {cmd_suffix}"
diff --git a/apps/backend/components/collections/agent_new/run_upgrade_command.py b/apps/backend/components/collections/agent_new/run_upgrade_command.py
index 0366e90c5..c447db999 100644
--- a/apps/backend/components/collections/agent_new/run_upgrade_command.py
+++ b/apps/backend/components/collections/agent_new/run_upgrade_command.py
@@ -78,7 +78,8 @@ def script_name(self):
     def get_script_content(self, data, common_data: AgentCommonData, host: models.Host) -> str:
         agent_upgrade_pkg_name = self.get_agent_pkg_name(common_data, host, is_upgrade=True)
         general_node_type = self.get_general_node_type(host.node_type)
-        agent_config = common_data.host_id__ap_map[host.bk_host_id].get_agent_config(host.os_type)
+        host_ap: models.AccessPoint = self.get_host_ap(common_data=common_data, host=host)
+        agent_config = host_ap.get_agent_config(host.os_type)
 
         if host.os_type == constants.OsType.WINDOWS:
             scripts = WINDOWS_UPGRADE_CMD_TEMPLATE.format(
diff --git a/apps/backend/components/collections/agent_new/update_install_info.py b/apps/backend/components/collections/agent_new/update_install_info.py
index f47678298..199cd81a3 100644
--- a/apps/backend/components/collections/agent_new/update_install_info.py
+++ b/apps/backend/components/collections/agent_new/update_install_info.py
@@ -36,6 +36,7 @@ def _execute(self, data, parent_data, common_data: AgentCommonData):
             added_extra_data = filter_values(added_extra_data)
             # 同名配置覆盖优先级：added_extra_data（新增配置）> host_obj.extra_data（已有配置）> default_extra_data（默认配置）
             host_obj.extra_data = dict(ChainMap(added_extra_data, host_obj.extra_data, default_extra_data))
-            host_obj.ap_id = host_info["ap_id"]
+            if common_data.injected_ap_id is None:
+                host_obj.ap_id = host_info["ap_id"]
             hosts_to_be_updated.append(host_obj)
         models.Host.objects.bulk_update(hosts_to_be_updated, fields=["extra_data", "ap_id"], batch_size=self.batch_size)
diff --git a/apps/backend/subscription/tools.py b/apps/backend/subscription/tools.py
index 745c65baa..a1c104af1 100644
--- a/apps/backend/subscription/tools.py
+++ b/apps/backend/subscription/tools.py
@@ -34,6 +34,7 @@
 )
 from apps.backend.utils.data_renderer import nested_render_data
 from apps.component.esbclient import client_v2
+from apps.core.ipchooser.tools.base import HostQuerySqlHelper
 from apps.exceptions import ComponentCallError
 from apps.node_man import constants, models
 from apps.node_man import tools as node_man_tools
@@ -41,7 +42,6 @@
 from apps.utils.batch_request import batch_request, request_multi_thread
 from apps.utils.cache import func_cache_decorator
 from apps.utils.time_handler import strftime_local
-from apps.core.ipchooser.tools.base import HostQuerySqlHelper
 
 logger = logging.getLogger("app")
 
@@ -681,7 +681,7 @@ def wrapper(scope: Dict[str, Union[Dict, Any]], *args, **kwargs) -> Dict[str, Di
                     "object_type": scope["object_type"],
                     "node_type": scope["node_type"],
                     "nodes": list(nodes),
-                    "instance_selector": scope.get("instance_selector")
+                    "instance_selector": scope.get("instance_selector"),
                 },
                 **kwargs,
             }
@@ -806,10 +806,14 @@ def get_instances_by_scope(scope: Dict[str, Union[Dict, int, Any]]) -> Dict[str,
 
     if not need_register:
         # 补充必要的主机或实例相关信息
+
         add_host_info_to_instances(bk_biz_id, scope, instances)
         add_scope_info_to_instances(nodes, scope, instances, module_to_topo)
         add_process_info_to_instances(bk_biz_id, scope, instances)
 
+        # 回填原始参数
+        add_meta_info_to_instance(scope, instances)
+
     instances_dict = {}
     data = {
         "object_type": scope["object_type"],
@@ -831,7 +835,7 @@ def get_instances_by_scope(scope: Dict[str, Union[Dict, int, Any]]) -> Dict[str,
         instance_selector_host_ids = HostQuerySqlHelper.multiple_cond_sql(
             params={"bk_host_id": bk_host_ids, "conditions": instance_selector},
             biz_scope=[bk_biz_id],
-            return_all_node_type=True
+            return_all_node_type=True,
         ).values_list("bk_host_id", flat=True)
 
         selector_instances_dict = {}
@@ -847,6 +851,13 @@ def get_instances_by_scope(scope: Dict[str, Union[Dict, int, Any]]) -> Dict[str,
     return instances_dict
 
 
+def add_meta_info_to_instance(scope: Dict, instances: Dict):
+    if scope["object_type"] == models.Subscription.ObjectType.HOST:
+        host_dict = {host["bk_host_id"]: host for host in scope["nodes"] if host.get("bk_host_id") is not None}
+        for instance in instances:
+            instance["host"].update(host_dict.get(instance["host"]["bk_host_id"], {}))
+
+
 def add_host_info_to_instances(bk_biz_id: int, scope: Dict, instances: Dict):
     """
     补全实例的主机信息
diff --git a/apps/node_man/handlers/job.py b/apps/node_man/handlers/job.py
index 2643e2a17..dfaf347a6 100644
--- a/apps/node_man/handlers/job.py
+++ b/apps/node_man/handlers/job.py
@@ -16,8 +16,8 @@
 
 from django.conf import settings
 from django.core.paginator import Paginator
-from django.db.models import Q
 from django.db import transaction
+from django.db.models import Q
 from django.utils import timezone
 from django.utils.translation import get_language
 from django.utils.translation import ugettext_lazy as _
@@ -187,9 +187,7 @@ def list(self, params: dict, username: str):
             biz_scope_query_q = Q()
         else:
             biz_scope_query_q = reduce(
-                operator.or_,
-                [Q(bk_biz_scope__contains=bk_biz_id) for bk_biz_id in biz_scope],
-                Q()
+                operator.or_, [Q(bk_biz_scope__contains=bk_biz_id) for bk_biz_id in biz_scope], Q()
             )
             # 仅查询所有业务时，自身创建的 job 可见
             if not search_biz_ids:
@@ -616,15 +614,13 @@ def update_host(accept_list: list, ip_filter_list: list, is_manual: bool = False
 
         return update_data_info["subscription_host_ids"], ip_filter_list
 
-    def operate(self, job_type, bk_host_ids, bk_biz_scope, extra_params, extra_config):
+    def operate(self, job_type, hosts, bk_biz_scope, extra_params, extra_config):
         """
         用于只有bk_host_id参数的下线、重启等操作
         """
         # 校验器进行校验
 
-        subscription = self.create_subscription(
-            job_type, bk_host_ids, extra_params=extra_params, extra_config=extra_config
-        )
+        subscription = self.create_subscription(job_type, hosts, extra_params=extra_params, extra_config=extra_config)
 
         return tools.JobTools.create_job(
             job_type=job_type,
@@ -634,9 +630,9 @@ def operate(self, job_type, bk_host_ids, bk_biz_scope, extra_params, extra_confi
             statistics={
                 "success_count": 0,
                 "failed_count": 0,
-                "pending_count": len(bk_host_ids),
+                "pending_count": len(hosts),
                 "running_count": 0,
-                "total_count": len(bk_host_ids),
+                "total_count": len(hosts),
             },
         )
 
diff --git a/apps/node_man/handlers/plugin.py b/apps/node_man/handlers/plugin.py
index 40aff2358..53271bb74 100644
--- a/apps/node_man/handlers/plugin.py
+++ b/apps/node_man/handlers/plugin.py
@@ -331,14 +331,14 @@ def operate(params: dict, username: str):
             ).values("bk_host_id", "bk_biz_id", "bk_cloud_id", "inner_ip", "node_type", "os_type")
 
         # 校验器进行校验
-        db_host_ids, host_biz_scope = operate_validator(list(db_host_sql))
+        db_hosts, host_biz_scope = operate_validator(list(db_host_sql))
 
         plugin_name__job_id__map = {}
         for plugin_params in params["plugin_params_list"]:
             plugin_name, plugin_version = plugin_params["name"], plugin_params["version"]
             subscription_create_result = PluginHandler.create_subscription(
                 job_type=params["job_type"],
-                nodes=db_host_ids,
+                nodes=db_hosts,
                 name=plugin_name,
                 version=plugin_version,
                 keep_config=plugin_params.get("keep_config"),
diff --git a/apps/node_man/handlers/validator.py b/apps/node_man/handlers/validator.py
index 58a371ef3..7b31a5a00 100644
--- a/apps/node_man/handlers/validator.py
+++ b/apps/node_man/handlers/validator.py
@@ -155,12 +155,10 @@ def bulk_update_validate(
     for host in accept_list:
         # 系统变更/接入点变更/DHT变更需要更新主机
         host_extra_data = host_info[host["bk_host_id"]]["extra_data"] or {}
-        if host.get("is_need_inject_ap_id"):
-            host["ap_id"] = host_info[host["bk_host_id"]]["ap_id"]
 
         if (
             host.get("os_type") != host_info[host["bk_host_id"]]["os_type"]
-            or host.get("ap_id") != host_info[host["bk_host_id"]]["ap_id"]
+            or (host.get("ap_id") != host_info[host["bk_host_id"]]["ap_id"] and not host.get("is_need_inject_ap_id"))
             or host.get("bt_speed_limit") != host_extra_data.get("bt_speed_limit")
             or host.get("peer_exchange_switch_for_agent") != host_extra_data.get("peer_exchange_switch_for_agent")
             or host.get("enable_compression") != host_extra_data.get("enable_compression")
@@ -183,6 +181,7 @@ def bulk_update_validate(
             {
                 "bk_host_id": host["bk_host_id"],
                 "bk_cloud_id": host["bk_cloud_id"],
+                "ap_id": host["ap_id"],
                 "install_channel_id": host.get("install_channel_id"),
                 "is_need_inject_ap_id": host.get("is_need_inject_ap_id"),
             }
@@ -567,7 +566,7 @@ def install_validate(
     return ip_filter_list, accept_list, proxy_not_alive
 
 
-def operate_validator(db_host_sql):
+def operate_validator(db_host_sql, host_info: typing.Dict[int, typing.Dict[str, typing.Any]] = {}):
     """
     用于operate任务的校验
     :param db_host_sql: 用户操作主机的详细信息
@@ -595,6 +594,11 @@ def operate_validator(db_host_sql):
     # 获得业务ID
     host_biz_scope = list({host["bk_biz_id"] for host in db_host_sql})
 
-    db_host_ids = [{"bk_host_id": host_id} for host_id in permission_host_ids]
+    db_hosts: typing.List[typing.Dict[str, typing.Any]] = []
 
-    return db_host_ids, host_biz_scope
+    for host_id in permission_host_ids:
+        _host = {"bk_host_id": host_id}
+        _host.update(host_info.get(host_id, {}))
+        db_hosts.append(_host)
+
+    return db_hosts, host_biz_scope
diff --git a/apps/node_man/serializers/job.py b/apps/node_man/serializers/job.py
index f355a3fe7..ce05ce23c 100644
--- a/apps/node_man/serializers/job.py
+++ b/apps/node_man/serializers/job.py
@@ -12,6 +12,7 @@
 from collections import defaultdict
 
 from django.conf import settings
+from django.db.models import Q
 from django.utils.translation import ugettext_lazy as _
 from rest_framework import serializers
 
@@ -83,7 +84,75 @@ class ListSerializer(serializers.Serializer):
     end_time = serializers.DateTimeField(label=_("终止时间"), required=False)
 
 
-class HostSerializer(serializers.Serializer):
+class InstallBaseSerializer(serializers.Serializer):
+
+    op_not_need_identity = [
+        constants.OpType.REINSTALL,
+        constants.OpType.RESTART,
+        constants.OpType.UPGRADE,
+        constants.OpType.UNINSTALL,
+        constants.OpType.RELOAD,
+    ]
+
+    def backfill_bk_host_id(self, hosts):
+        query_params = Q()
+        query_params.connector = "OR"
+        cloud_ip_host_info_map = {}
+        for _host in hosts:
+            if _host.get("bk_host_id") is None:
+
+                sub_query = Q()
+                sub_query.connector = "AND"
+                sub_query.children.append(("bk_cloud_id", _host["bk_cloud_id"]))
+                if _host.get("inner_ip"):
+                    sub_query.children.append(("inner_ip", _host["inner_ip"]))
+                    ip_key = _host["inner_ip"]
+                else:
+                    sub_query.children.append(("inner_ipv6", _host["inner_ipv6"]))
+                    ip_key = _host["inner_ipv6"]
+
+                cloud_ip_host_info_map[f"{_host['bk_cloud_id']}:{ip_key}"] = _host
+                query_params.children.append(sub_query)
+
+        query_hosts: typing.List[typing.Dict] = models.Host.objects.filter(query_params).values(
+            "bk_host_id", "bk_cloud_id", "inner_ip", "inner_ipv6"
+        )
+
+        cloup_ip_host_id_map = {}
+        for _query_host in query_hosts:
+            if _query_host["inner_ip"]:
+                cloup_ip_host_id_map[f"{_query_host['bk_cloud_id']}:{_query_host['inner_ip']}"] = _query_host[
+                    "bk_host_id"
+                ]
+            if _query_host["inner_ipv6"]:
+                cloup_ip_host_id_map[f"{_query_host['bk_cloud_id']}:{_query_host['inner_ipv6']}"] = _query_host[
+                    "bk_host_id"
+                ]
+
+        for cloud_ip, host_info in cloud_ip_host_info_map.items():
+            host_info["bk_host_id"] = cloup_ip_host_id_map[cloud_ip]
+
+    def backfill_ap_id(self, hosts):
+        use_ap_map_host_ids: typing.List[int] = [host["bk_host_id"] for host in hosts if host["is_use_ap_map"]]
+        if use_ap_map_host_ids:
+            gray_ap_map: typing.Dict[int, int] = GrayHandler.get_gray_ap_map()
+            host_queryset = models.Host.objects.filter(bk_host_id__in=use_ap_map_host_ids).values("bk_host_id", "ap_id")
+            host_id_ap_map: typing.Dict[int, int] = {_host["bk_host_id"]: _host["ap_id"] for _host in host_queryset}
+            for host in hosts:
+                if not host["is_use_ap_map"]:
+                    continue
+
+                try:
+                    host["ap_id"] = gray_ap_map[host_id_ap_map[host["bk_host_id"]]]
+                except KeyError:
+                    raise ValidationError(
+                        _("缺少与主机ID: {bk_host_id} AP ID: {ap_id} 对应的接入点映射，请联系管理员配置").format(
+                            bk_host_id=host["bk_host_id"], ap_id=host_id_ap_map[host["bk_host_id"]]
+                        )
+                    )
+
+
+class HostSerializer(InstallBaseSerializer):
     bk_biz_id = serializers.IntegerField(label=_("业务ID"))
     bk_cloud_id = serializers.IntegerField(label=_("管控区域ID"))
     bk_host_id = serializers.IntegerField(label=_("主机ID"), required=False)
@@ -113,8 +182,9 @@ class HostSerializer(serializers.Serializer):
     peer_exchange_switch_for_agent = serializers.IntegerField(label=_("加速设置"), required=False, default=0)
     bt_speed_limit = serializers.IntegerField(label=_("传输限速"), required=False)
     data_path = serializers.CharField(label=_("数据文件路径"), required=False, allow_blank=True)
-    is_need_inject_ap_id = serializers.IntegerField(label=_("是否需要注入ap_id到meta"), required=False, default=False)
+    is_need_inject_ap_id = serializers.BooleanField(label=_("是否需要注入ap_id到meta"), required=False, default=False)
     enable_compression = serializers.BooleanField(label=_("数据压缩开关"), required=False, default=False)
+    is_use_ap_map = serializers.BooleanField(label=_("是否使用映射接入点"), required=False, default=False)
 
     def validate(self, attrs):
         # 获取任务类型，如果是除安装以外的操作，则密码和秘钥可以为空
@@ -129,16 +199,9 @@ def validate(self, attrs):
 
         basic.ipv6_formatter(data=attrs, ipv6_field_names=["inner_ipv6", "outer_ipv6", "login_ip", "data_ip"])
 
-        op_not_need_identity = [
-            constants.OpType.REINSTALL,
-            constants.OpType.RESTART,
-            constants.OpType.UPGRADE,
-            constants.OpType.UNINSTALL,
-            constants.OpType.RELOAD,
-        ]
+        if attrs["is_use_ap_map"]:
+            attrs["is_need_inject_ap_id"] = True
 
-        if op_type in op_not_need_identity and not attrs.get("bk_host_id"):
-            raise ValidationError(_("{op_type} 操作必须传入 bk_host_id 参数").format(op_type=op_type))
         if (
             not attrs.get("is_manual")
             and not attrs.get("auth_type")
@@ -146,8 +209,22 @@ def validate(self, attrs):
         ):
             raise ValidationError(_("{op_type} 操作必须填写认证类型").format(op_type=op_type))
 
+        if op_type in self.op_not_need_identity:
+            if all(
+                [
+                    attrs.get("bk_host_id") is None,
+                    all(
+                        [
+                            attrs.get("bk_cloud_id") is None,
+                            attrs.get("inner_ip") is None or attrs.get("inner_ipv6") is None,
+                        ]
+                    ),
+                ]
+            ):
+                raise ValidationError(_("bk_host_id 或者 管控区域加IP组合必须选择一种"))
+
         # identity校验
-        if op_type not in op_not_need_identity and not attrs.get("is_manual"):
+        if op_type not in self.op_not_need_identity and not attrs.get("is_manual"):
             if not attrs.get("password") and attrs["auth_type"] == constants.AuthType.PASSWORD:
                 raise ValidationError(_("密码认证方式必须填写密码"))
             if not attrs.get("key") and attrs["auth_type"] == constants.AuthType.KEY:
@@ -157,8 +234,8 @@ def validate(self, attrs):
 
         # 直连区域必须填写Ap_id
         ap_id = attrs.get("ap_id")
-        if attrs["bk_cloud_id"] == int(constants.DEFAULT_CLOUD) and ap_id is None:
-            raise ValidationError(_("直连区域必须填写Ap_id"))
+        if attrs["bk_cloud_id"] == int(constants.DEFAULT_CLOUD) and ap_id is None and not attrs["is_use_ap_map"]:
+            raise ValidationError(_("直连区域必须填写Ap_id或使用映射"))
 
         # 去除空值
         if attrs.get("key") == "":
@@ -179,7 +256,7 @@ class ScriptHook(serializers.Serializer):
     name = serializers.CharField(label=_("脚本名称"), min_length=1)
 
 
-class InstallSerializer(serializers.Serializer):
+class InstallSerializer(InstallBaseSerializer):
     agent_setup_info = AgentSetupInfoSerializer(label=_("Agent 设置信息"), required=False)
     job_type = serializers.ChoiceField(label=_("任务类型"), choices=list(constants.JOB_TYPE_DICT))
     hosts = HostSerializer(label=_("主机信息"), many=True)
@@ -202,6 +279,12 @@ def validate(self, attrs):
         if attrs["job_type"] == constants.JobType.REPLACE_PROXY and not attrs.get("replace_host_id"):
             raise ValidationError(_("替换PROXY必须填写replace_host_id"))
 
+        if attrs["op_type"] in self.op_not_need_identity:
+            # 回填bk_host_id
+            self.backfill_bk_host_id(attrs["hosts"])
+            # 回填使用映射的主机的ap id
+            self.backfill_ap_id(attrs["hosts"])
+
         bk_biz_ids = set()
         expected_bk_host_ids_gby_bk_biz_id: typing.Dict[int, typing.List[int]] = defaultdict(list)
         cipher = tools.HostTools.get_asymmetric_cipher()
@@ -218,8 +301,6 @@ def validate(self, attrs):
                 )
 
             if attrs["op_type"] not in [constants.OpType.INSTALL, constants.OpType.REPLACE]:
-                if "bk_host_id" not in host:
-                    raise ValidationError(_("主机信息缺少主机ID（bk_host_id）"))
                 if "bk_biz_id" not in host:
                     raise ValidationError(_("主机信息缺少业务ID（bk_biz_id）"))
                 expected_bk_host_ids_gby_bk_biz_id[host["bk_biz_id"]].append(host["bk_host_id"])
@@ -273,7 +354,44 @@ def validate(self, attrs):
         return attrs
 
 
-class OperateSerializer(serializers.Serializer):
+class OperateHostSerializer(serializers.Serializer):
+    """
+    操作类任务主机序列化器
+    """
+
+    bk_host_id = serializers.IntegerField(label=_("主机ID"), required=False)
+    ap_id = serializers.IntegerField(label=_("接入点ID"), required=False)
+    bk_cloud_id = serializers.IntegerField(label=_("管控区域ID"), required=False)
+    inner_ip = serializers.IPAddressField(label=_("内网IP"), required=False, allow_blank=True, protocol="ipv4")
+    inner_ipv6 = serializers.IPAddressField(label=_("内网IPv6"), required=False, allow_blank=True, protocol="ipv6")
+    is_use_ap_map = serializers.BooleanField(label=_("是否使用映射接入点"), required=False, default=False)
+
+    # 以下参数不需要用户传入
+    is_need_inject_ap_id = serializers.BooleanField(label=_("是否需要注入ap_id到meta"), required=False, default=False)
+
+    def validate(self, attrs):
+        # 计算is_need_inject_ap_id参数
+        basic.ipv6_formatter(data=attrs, ipv6_field_names=["inner_ipv6"])
+
+        if attrs.get("ap_id") is not None or attrs["is_use_ap_map"]:
+            attrs["is_need_inject_ap_id"] = True
+        if all(
+            [
+                attrs.get("bk_host_id") is None,
+                all(
+                    [
+                        attrs.get("bk_cloud_id") is None,
+                        attrs.get("inner_ip") is None or attrs.get("inner_ipv6") is None,
+                    ]
+                ),
+            ]
+        ):
+            raise ValidationError(_("bk_host_id 或者 管控区域加IP组合必须选择一种"))
+
+        return attrs
+
+
+class OperateSerializer(InstallBaseSerializer):
     agent_setup_info = AgentSetupInfoSerializer(label=_("Agent 设置信息"), required=False)
     job_type = serializers.ChoiceField(label=_("任务类型"), choices=list(constants.JOB_TYPE_DICT))
     bk_biz_id = serializers.ListField(label=_("业务ID"), required=False)
@@ -281,6 +399,7 @@ class OperateSerializer(serializers.Serializer):
     conditions = serializers.ListField(label=_("搜索条件"), required=False, child=serializers.DictField())
     bk_host_id = serializers.ListField(label=_("主机ID"), required=False, child=serializers.IntegerField())
     exclude_hosts = serializers.ListField(label=_("跨页全选排除主机"), required=False, child=serializers.IntegerField())
+    hosts = OperateHostSerializer(label=_("主机信息"), required=False, many=True)
     is_install_latest_plugins = serializers.BooleanField(label=_("是否安装最新版本插件"), required=False, default=True)
     is_install_other_agent = serializers.BooleanField(label=_("是否为安装额外Agent"), required=False, default=False)
 
@@ -302,9 +421,21 @@ def validate(self, attrs):
 
         if attrs.get("exclude_hosts") is not None and attrs.get("bk_host_id") is not None:
             raise ValidationError(_("跨页全选模式下不允许传bk_host_id参数"))
-        if attrs.get("exclude_hosts") is None and attrs.get("bk_host_id") is None:
+        if all(
+            [
+                attrs.get("exclude_hosts") is None,
+                attrs.get("bk_host_id") is None,
+                attrs.get("hosts") is None,
+            ]
+        ):
             raise ValidationError(_("必须选择一种模式(【是否跨页全选】)"))
 
+        if attrs.get("hosts", []):
+            # 回填bk_host_id
+            self.backfill_bk_host_id(attrs["hosts"])
+            # 回填使用映射的主机的ap id
+            self.backfill_ap_id(attrs["hosts"])
+
         if attrs["node_type"] == constants.NodeType.PROXY:
             # 是否为针对代理的操作，用户有权限获取的业务
             # 格式 { bk_biz_id: bk_biz_name , ...}
@@ -318,6 +449,10 @@ def validate(self, attrs):
             filter_node_types = [constants.NodeType.AGENT, constants.NodeType.PAGENT]
             is_proxy = False
 
+        host_info: typing.Dict[int, typing.Dict[str, typing.Any]] = {
+            _host["bk_host_id"]: _host for _host in attrs.get("hosts", [])
+        }
+
         if attrs.get("exclude_hosts") is not None:
             # 跨页全选
             db_host_sql = (
@@ -328,12 +463,13 @@ def validate(self, attrs):
 
         else:
             # 不是跨页全选
+            input_bk_host_ids: typing.List[int] = attrs.get("bk_host_id", []) or host_info.keys()
             db_host_sql = models.Host.objects.filter(
-                bk_host_id__in=attrs["bk_host_id"], node_type__in=filter_node_types
+                bk_host_id__in=input_bk_host_ids, node_type__in=filter_node_types
             ).values("bk_host_id", "bk_biz_id", "bk_cloud_id", "inner_ip", "node_type", "os_type")
 
-        bk_host_ids, bk_biz_scope = validator.operate_validator(list(db_host_sql))
-        attrs["bk_host_ids"] = bk_host_ids
+        db_hosts, bk_biz_scope = validator.operate_validator(list(db_host_sql), host_info=host_info)
+        attrs["hosts"] = db_hosts
         attrs["bk_biz_scope"] = bk_biz_scope
 
         set_agent_setup_info_to_attrs(attrs)
@@ -345,7 +481,7 @@ def validate(self, attrs):
             # 没有 V2 接入点或者全部为 V2 接入点时，无需进行重定向处理
             return attrs
 
-        host_ids: typing.List[int] = [host_info["bk_host_id"] for host_info in bk_host_ids]
+        host_ids: typing.List[int] = [host_info["bk_host_id"] for host_info in db_hosts]
 
         # 进入灰度的管控区域，所属管控区域主机接入点重定向到 V2
         gse_v2_cloud_ids: typing.Set[int] = set(
@@ -384,6 +520,7 @@ def validate(self, attrs):
                 GrayHandler.activate(host_nodes=update_result["host_nodes"], rollback=False, only_status=True)
             except ApiError:
                 pass
+
         return attrs
 
 
diff --git a/apps/node_man/tests/test_views/test_job_views.py b/apps/node_man/tests/test_views/test_job_views.py
index 5636893bf..330185da3 100644
--- a/apps/node_man/tests/test_views/test_job_views.py
+++ b/apps/node_man/tests/test_views/test_job_views.py
@@ -73,12 +73,12 @@ def generate_install_job_request_params():
         return data
 
 
-class TestBkHostIdNeededError(TestJobValidationError):
-    ERROR_MSG_KEYWORD = "操作必须传入 bk_host_id 参数"
-
-    @staticmethod
-    def generate_install_job_request_params():
-        data = copy.deepcopy(job.JOB_INSTALL_REQUEST_PARAMS)
-        data["job_type"] = constants.JobType.REINSTALL_AGENT
-        data["hosts"][0].pop("bk_host_id", "")
-        return data
+# class TestBkHostIdNeededError(TestJobValidationError):
+#     ERROR_MSG_KEYWORD = "操作必须传入 bk_host_id 参数"
+
+#     @staticmethod
+#     def generate_install_job_request_params():
+#         data = copy.deepcopy(job.JOB_INSTALL_REQUEST_PARAMS)
+#         data["job_type"] = constants.JobType.REINSTALL_AGENT
+#         data["hosts"][0].pop("bk_host_id", "")
+#         return data
diff --git a/apps/node_man/views/job.py b/apps/node_man/views/job.py
index 47ec78835..00fefb537 100644
--- a/apps/node_man/views/job.py
+++ b/apps/node_man/views/job.py
@@ -149,14 +149,14 @@ def operate(self, request):
         """
         validated_data = self.validated_data
         job_type = validated_data["job_type"]
-        bk_host_ids = validated_data["bk_host_ids"]
+        hosts = validated_data["hosts"]
         bk_biz_scope = validated_data["bk_biz_scope"]
         extra_params = {
             "is_install_latest_plugins": validated_data["is_install_latest_plugins"],
             "is_install_other_agent": validated_data["is_install_other_agent"],
         }
         extra_config = validated_data.get("agent_setup_info") or {}
-        return Response(JobHandler().operate(job_type, bk_host_ids, bk_biz_scope, extra_params, extra_config))
+        return Response(JobHandler().operate(job_type, hosts, bk_biz_scope, extra_params, extra_config))
 
     @swagger_auto_schema(
         operation_summary="重试任务",
diff --git a/noahchi.py b/noahchi.py
new file mode 100644
index 000000000..d887295c6
--- /dev/null
+++ b/noahchi.py
@@ -0,0 +1,26 @@
+coding=utf-8
+import os
+import django
+
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "settings")
+django.setup()
+
+
+from apps.node_man.handlers.security_group import YunTiSecurityGroupFactory
+
+
+# yunti = YunTiSecurityGroupFactory()
+# print(yunti.describe_security_group_address())
+
+# hosts =
+
+
+# yunti.add_ips_to_security_group(hosts, creator="noahchi")
+
+
+st = '5393176:577,5378517:553,5378518:553,5369160:570,5346573:575,5346572:575,5333494:574,5332466:428,5332467:428,5332464:426,5332465:426,5332215:425,5332216:425,4838330:544,5295634:221,5295633:221,5295255:572,5295254:572,5279819:570,5268397:414,1563145:256,2555794:367,2601897:367,5247429:420,5243257:420,5235691:568,3699221:488,5187060:566,4632691:522,4633081:522,5144961:560,5144962:560,5142387:564,5142386:564,5142492:561,5142491:561,5142333:562,5142332:562,5142485:563,5142486:563,5157629:256,5144857:565,5144858:565,5141608:354,5141609:354,5130648:555,5112282:557,5100597:248,5101057:248,5090678:556,4998269:555,1149017:145,733784:145,4974188:551,4974187:551,4975232:549,4975231:549,4949157:550,4949158:550,4948386:353,4948387:353,4945364:550,4945363:550,4941101:434,4941102:434,4902847:537,4902846:537,4892120:548,4892119:548,4890425:433,4889743:414,4870639:546,4863582:359,4863653:359,4863712:288,4863859:288,4863694:356,4863910:356,4842729:538,4842955:545,4842740:545,4786190:542,4729852:336,4729853:336,4766077:540,4766078:540,1882256:384,1882288:384,4723623:539,4712545:487,4684406:433,4659800:529,4659801:529,4651060:533,4648145:531,4636587:530,855037:118,4613665:393,4584612:517,4583971:245,4583102:527,4583101:527,4578324:526,4573713:526,4573908:440,4573718:440,4566799:227,4566354:262,4566345:244,4566340:247,4566343:214,4566414:209,4566344:243,4566355:292,4566339:212,4566074:232,4566065:236,4566073:231,4566062:241,4566066:233,4566063:237,4565141:223,4565091:219,4565159:238,4565092:220,4565142:216,4565090:229,4545690:525,4545689:525,4540273:524,4533546:227,4531287:517,4530609:238,4521514:244,4501558:262,4501377:243,4501364:247,4521487:231,4501312:233,4501311:237,4501310:216,4519738:209,4501313:212,4501023:220,4447431:249,4446608:249,4347298:522,3707750:489,3707751:489,4280473:518,4280474:518,4049745:512,4023850:510,3929828:487,3924042:507,1170466:170,3916141:505,3916140:505,3915822:504,3915823:504,3793662:500,3787844:500,3753629:498,3753630:498,3747762:301,3747763:301,3710536:448,3163190:448,3083560:447,3083559:447,3698376:488,3666977:474,3666978:474,3658757:475,3609569:483,3537375:479,3537457:479,3508579:251,3501144:471,3508254:251,3459880:467,2658714:431,2658713:431,3396452:132,713480:132,3259445:459,3259187:458,3259186:458,3259446:459,3280856:419,3280857:419,3280021:427,3280022:427,3266555:461,3252075:450,3252074:450,3247015:457,3241702:456,3196971:449,3196972:449,2580516:333,2004151:319,2837441:445,2818529:444,2818530:444,2802576:443,2802575:443,2603592:424,2600007:415,2600008:415,2586293:422,2586294:422,2580515:333,2580710:421,2580709:421,2566986:416,2566987:416,2548923:400,2243955:400,1608811:301,2500830:413,2457660:409,2457661:409,2453281:408,2453280:408,2441452:407,2441453:407,2417300:405,2334831:403,2334832:403,2273937:389,2273938:389,2243884:400,2138811:396,2138810:396,2099259:395,1996028:393,1980421:392,1944351:390,1797139:350,1797138:350,1791669:352,1791498:351,1738667:346,1738666:346,1721523:342,1699479:337,1699194:225,1699193:225,1690109:335,1678918:328,1674963:330,1652716:323,1652715:323,1630891:210,1503144:257,1630892:210,1635162:320,1635163:320,1634099:318,1631401:242,1631400:242,1628889:312,1628890:312,1627191:311,1627190:311,1625689:223,1624333:219,1623731:301,1616095:260,1616094:260,1615824:214,1613654:235,1613653:235,1612476:307,1612475:307,1611762:302,1611763:302,1608810:301,1602615:224,1602498:228,1594108:222,1594107:222,1594000:232,1580890:292,1574182:236,1570167:297,1562821:215,1562820:215,1562553:229,1556380:245,1553470:295,1503145:257,69599:94,1492264:259,1477019:255,1476056:254,1476055:254,1472813:205,1467673:207,1466573:207,1463938:205,1440520:198,1440532:197,1440531:196,1374179:194,1363406:193,1363407:193,1327933:191,1327932:191,1256121:185,1256122:185,1299984:78,30738:78,1278957:163,1278958:163,1256262:187,1256261:187,1256100:184,1256099:184,1255965:183,1255966:183,1255963:182,1255964:182,1255961:181,1255962:181,1255933:180,1255932:180,1249450:161,1249451:161,1249446:160,1249447:160,702790:132,702797:132,715231:135,715232:135,1170189:169,1242944:164,1242945:164,1228327:176,1223160:172,1223114:174,1069390:107,1175577:171,1170188:169,1145081:8,1139564:168,1132879:165,1132878:165,1112712:159,1116238:159,1102107:158,1064857:157,1021987:154,1021986:154,1011136:152,1011140:153,1011141:153,1011135:152,733827:148,937806:150,733745:146,733746:146,729721:144,728181:143,727246:142,719977:139,665462:129,665461:129,605344:125,598565:124,32905:86,44702:86,417879:120,69595:102,69669:103,69670:103,69672:104,29976:77,379320:110,379118:109,29324:74,376926:108,18856:62,44424:89,32886:84,19301:68,5330581:332,5330580:441,5330341:332,5330340:441,5326126:432,5325769:432,5318509:246,5318500:374,5318448:246,5317947:374,5290316:371,5289262:371,5286025:218,5285693:218,5087824:241,4868265:360,4868263:286,4868101:286,4868099:360,4868035:337,4868034:424,1466761:206,1466760:206'
+for i in st.split(','):
+    print(i)
+
+
+