diff --git a/meta/runtime.yml b/meta/runtime.yml
index 09b32abd6..68c870b98 100644
--- a/meta/runtime.yml
+++ b/meta/runtime.yml
@@ -357,3 +357,5 @@ action_groups:
     - azure.azcollection.azure_rm_afdruleset_info
     - azure.azcollection.azure_rm_afdrules
     - azure.azcollection.azure_rm_afdrules_info
+    - azure.azcollection.azure_rm_tags
+    - azure.azcollection.azure_rm_tags_info
diff --git a/plugins/modules/azure_rm_tags.py b/plugins/modules/azure_rm_tags.py
new file mode 100644
index 000000000..b272f1660
--- /dev/null
+++ b/plugins/modules/azure_rm_tags.py
@@ -0,0 +1,346 @@
+#!/usr/bin/python
+#
+# Copyright (c) 2025 xuzhang3 (@xuzhang3), Fred-sun (@Fred-sun)
+#
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+
+DOCUMENTATION = '''
+---
+module: azure_rm_tags
+version_added: "3.4.0"
+short_description: Manage tags
+description:
+    - Create, update ,delete the tags.
+options:
+    tag_name:
+        description:
+            - The name of the tag.
+        type: str
+    tag_value:
+        description:
+            - The value of the tag.
+        type: str
+    scope:
+        description:
+            - The resource scope.
+        type: str
+    operation:
+        description:
+            - The operation type for the patch API.
+            - The default value is I(operation=Merge) and use to add tags.
+        type: str
+        default: Merge
+        choices:
+            - Delete
+            - Replace
+            - Merge
+    state:
+        description:
+            - State of the SSH Public Key. Use C(present) to create or update and C(absent) to delete.
+        default: present
+        type: str
+        choices:
+            - absent
+            - present
+
+extends_documentation_fragment:
+    - azure.azcollection.azure
+    - azure.azcollection.azure_tags
+
+author:
+    - xuzhang3 (@xuzhang3)
+    - Fred-sun (@Fred-sun)
+
+'''
+
+EXAMPLES = '''
+- name: Create a tag with tag_name
+  azure_rm_tags:
+    tag_name: testkey
+
+- name: Create a tag with tag_value
+  azure_rm_tags:
+    tag_name: testkey
+    tag_value: testvalue
+
+- name: Create a new tags with scope
+  azure_rm_tags:
+    scope: "/subscriptions/xxxxxxxxxxxxxxxxxxxxxxxxxx/resourceGroups/v-xisuRG02"
+    tags:
+      key4: value4
+
+- name: Update the tags with scope
+  azure_rm_tags:
+    scope: "/subscriptions/xxxxxxxxxxxxxxxxxxxxxxxxxx/resourceGroups/v-xisuRG02"
+    operation: Delete
+    tags:
+      key5: value5
+
+- name: Delete the tags by scope
+  azure_rm_tags:
+    scope: "/subscriptions/xxxxxxxxxxxxxxxxxxxxxxxxxx/resourceGroups/v-xisuRG02"
+    state: absent
+
+- name: Delete the tag with tag_name
+  azure_rm_tags:
+    tag_name: testkey
+    state: absent
+
+- name: Delete the tag with tag_value
+  azure_rm_tags:
+    tag_name: testkey
+    tag_value: testvalue
+    state: absent
+'''
+RETURN = '''
+tag_info:
+    description:
+        - The tag info.
+    returned: when I(scope=None)
+    type: complex
+    contains:
+        id:
+            description:
+                - The ID of the tags wrapper resource.
+            returned: always
+            type: str
+            sample: "/subscriptions/xxx-xxx/resourceGroups/testRG/providers/Microsoft.Resources/tags/default"
+        name:
+            description:
+                - The name of the tags wrapper resource.
+            returned: always
+            type: str
+            sample: default
+        type:
+            description:
+                - The type of the tags wrapper resource.
+            returned: always
+            type: str
+            sample: Microsoft.Resources/tags
+        properties:
+            description:
+                - The set of tags.
+            returned: always
+            type: dict
+            sample: { 'tags': {'key1': 'value1', 'key2': 'value2'}}
+'''
+
+
+try:
+    from ansible_collections.azure.azcollection.plugins.module_utils.azure_rm_common import AzureRMModuleBase
+    import copy
+    from azure.core.polling import LROPoller
+except ImportError:
+    # This is handled in azure_rm_common
+    pass
+
+
+class AzureRMTags(AzureRMModuleBase):
+
+    def __init__(self):
+
+        self.module_arg_spec = dict(
+            tag_name=dict(type='str'),
+            tag_value=dict(type='str'),
+            scope=dict(type='str'),
+            operation=dict(type='str', choices=['Replace', 'Merge', 'Delete'], default='Merge'),
+            state=dict(type='str', default='present', choices=['present', 'absent']),
+        )
+
+        self.tag_name = None
+        self.tag_value = None
+        self.scope = None
+        self.operation = None
+        self.tags = None
+        self.state = None
+
+        self.results = dict(
+            changed=False,
+            tag_info=dict()
+        )
+
+        super(AzureRMTags, self).__init__(self.module_arg_spec,
+                                          supports_tags=True,
+                                          supports_check_mode=True)
+
+    def exec_module(self, **kwargs):
+
+        for key in list(self.module_arg_spec.keys()) + ['tags']:
+            setattr(self, key, kwargs[key])
+
+        changed = False
+        if self.state == 'present':
+            if self.scope is not None:
+                response = self.get_at_scope()
+                if response is not None and response['properties'].get('tags'):
+                    if self.tags is not None:
+                        update_tags = self.tags_update(response['properties']['tags'], self.tags)
+                        if update_tags:
+                            changed = True
+                            response = self.begin_update_at_scope(self.tags, self.operation)
+                        elif self.operation == 'Delete':
+                            changed = True
+                            response = self.begin_update_at_scope(self.tags, self.operation)
+                else:
+                    if self.tags is not None:
+                        changed = True
+                        response = self.begin_create_or_update_at_scope(self.tags)
+            else:
+                response = self.get_by_tag_name(self.tag_name)
+                if self.tag_name is not None and self.tag_value is not None:
+                    if response is not None:
+                        key_value = [item['tag_value'] for item in response['values']]
+                        if self.tag_value not in key_value:
+                            changed = True
+                            response = self.create_or_update_value()
+                    else:
+                        self.fail("The tag_name {0} not exist, Please makesure the tag_name exist".format(self.tag_name))
+                elif self.tag_name is not None:
+                    if response is None:
+                        changed = True
+                        response = self.create_or_update()
+                else:
+                    self.fail("If I(scope!=None), The tag_name, or tag_name and tag_value must be configured")
+        else:
+            if self.scope is not None:
+                response = self.get_at_scope()
+                if response['properties']['tags']:
+                    changed = True
+                    response = self.delete_at_scope()
+            elif self.tag_name is not None and self.tag_value is not None:
+                response = self.get_by_tag_name(self.tag_name)
+                if response is not None:
+                    key_value = [item['tag_value'] for item in response['values']]
+                    if self.tag_value in key_value:
+                        changed = True
+                        response = self.delete_value()
+            elif self.tag_name is not None:
+                response = self.get_by_tag_name(self.tag_name)
+                if response is not None:
+                    changed = True
+                    response = self.delete_tags()
+            else:
+                self.fail("When I(state=absent), scope, tag_name, or tag_name and tag_value must be configured")
+
+        self.results['changed'] = changed
+        self.results['tag_info'] = response
+
+        return self.results
+
+    def begin_create_or_update_at_scope(self, tags):
+        self.log('Creates or updates the entire set of tags on a resource or subscription.')
+        try:
+            response = self.rm_client.tags.begin_create_or_update_at_scope(self.scope,
+                                                                           dict(properties=dict(tags=tags)))
+            if isinstance(response, LROPoller):
+                response = self.get_poller_result(response)
+        except Exception as exc:
+            self.fail('Creates or updates the entire set of tags on a resource or subscription got Exception as as {0}'.format(exc.message or str(exc)))
+        return self.format_tags(response)
+
+    def begin_update_at_scope(self, tags, operation):
+        self.log('Selectively updates the set of tags on a resource or subscription.')
+        try:
+            response = self.rm_client.tags.begin_update_at_scope(self.scope,
+                                                                 dict(operation=operation, properties=dict(tags=tags)))
+            if isinstance(response, LROPoller):
+                response = self.get_poller_result(response)
+        except Exception as exc:
+            self.fail('Selectively updates the set of tags on a resource or subscription got Excption as {0}'.format(exc.message or str(exc)))
+
+        return self.format_tags(response)
+
+    def create_or_update_value(self):
+        self.log('Creates a predefined value for a predefined tag name.')
+        try:
+            results = self.rm_client.tags.create_or_update_value(self.tag_name, self.tag_value)
+        except Exception as exc:
+            self.fail('Creates a predefined value for a predefined tag name got Excetion as {0}'.format(exc.message or str(exc)))
+        return results.as_dict()
+
+    def create_or_update(self):
+        self.log('Creates a predefined tag name.')
+        try:
+            results = self.rm_client.tags.create_or_update(self.tag_name)
+        except Exception as exc:
+            self.fail('Creates a predefined tag name got Excetion as {0}'.format(exc.message or str(exc)))
+
+        return results.as_dict()
+
+    def delete_at_scope(self):
+        self.log('Deletes the entire set of tags on a resource or subscription.')
+        try:
+            self.rm_client.tags.begin_delete_at_scope(self.scope)
+        except Exception as exc:
+            self.fail('Delete the entire set of tag got Excetion as {0}'.format(exc.message or str(exc)))
+
+    def delete_value(self):
+        self.log('Deletes a predefined tag value for a predefined tag name {0} and value {1}'.format(self.tag_name, self.tag_value))
+        try:
+            self.rm_client.tags.delete_value(self.tag_name, self.tag_value)
+        except Exception as exc:
+            self.fail('Delete the predefined tag value got Excetion as {0}'.format(exc.message or str(exc)))
+
+    def delete_tags(self):
+        self.log('Delete a predefined tag name {0}'.format(self.tag_name))
+        try:
+            self.rm_client.tags.delete(self.tag_name)
+        except Exception as exc:
+            self.fail('Delete the predefined tag name got Excetion as {0}'.format(exc.message or str(exc)))
+
+    def get_at_scope(self):
+        self.log('Get properties for {0}'.format(self.scope))
+        try:
+            response = self.rm_client.tags.get_at_scope(self.scope)
+            if response is not None:
+                return self.format_tags(response)
+        except Exception as exc:
+            self.fail('Error when get the tags info under specified scope got Excetion as {0}'.format(exc.message or str(exc)))
+
+    def get_by_tag_name(self, tag_name):
+        try:
+            response = self.rm_client.tags.list()
+            while True:
+                item = response.next().as_dict()
+                if item['tag_name'] == tag_name:
+                    return item
+        except StopIteration:
+            return None
+        except Exception as exc:
+            self.fail('Error when listing all tags under subscription got Excetion as {0}'.format(exc.message or str(exc)))
+
+    def format_tags(self, tags):
+        results = dict(
+            id=tags.id,
+            name=tags.name,
+            type=tags.type,
+            properties=dict()
+        )
+        if tags.properties is not None:
+            results['properties'] = tags.properties.as_dict()
+
+        return results
+
+    def tags_update(self, old, new):
+        tags = old or dict()
+        new_tags = copy.copy(tags) if isinstance(tags, dict) else dict()
+        param_tags = new if isinstance(new, dict) else dict()
+        changed = False
+        # check add or update
+        for key, value in param_tags.items():
+            if not new_tags.get(key) or new_tags[key] != value:
+                changed = True
+                new_tags[key] = value
+        return changed
+
+
+def main():
+    AzureRMTags()
+
+
+if __name__ == '__main__':
+    main()
diff --git a/plugins/modules/azure_rm_tags_info.py b/plugins/modules/azure_rm_tags_info.py
new file mode 100644
index 000000000..bca1c8c71
--- /dev/null
+++ b/plugins/modules/azure_rm_tags_info.py
@@ -0,0 +1,209 @@
+#!/usr/bin/python
+#
+# Copyright (c) 2025 xuhang3 (@xuzhang3), Fred-sun (@Fred-sun)
+#
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import absolute_import, division, print_function
+__metaclass__ = type
+
+
+DOCUMENTATION = '''
+---
+module: azure_rm_tags_info
+
+version_added: "3.4.0"
+
+short_description: List tags facts
+
+description:
+    - List all tag details under subscription_id.
+    - Get the tag info at the specified scope.
+
+options:
+    scope:
+        description:
+            - The resource scope.
+        type: str
+
+extends_documentation_fragment:
+    - azure.azcollection.azure
+
+author:
+    - xuzhang3 (@xuzhang3)
+    - Fred-sun (@Fred-sun)
+'''
+
+EXAMPLES = '''
+- name: Get the tag info at thespecified resource
+  azure_rm_tags_info:
+    scope: scope_str
+
+- name: List all tag details under subscription
+  azure_rm_tags_info:
+'''
+RETURN = '''
+tag_details:
+    description:
+        - List tag details.
+    returned: when I(scope!=None)
+    type: complex
+    contains:
+        id:
+            description:
+                - The tag name ID.
+            returned: always
+            type: str
+            sample: "/subscriptions/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx/tagNames/k1"
+        tag_name:
+            description:
+                - The tag name.
+            returned: always
+            type: str
+            sample: K1
+        count:
+            description:
+                - The total number of resources that use the resource tag.
+                -  When a tag is initially created and has no associated resources, the value is 0.
+            returned: always
+            type: dict
+            sample: { 'type': 'Total', 'value': 1}
+        values:
+            description:
+                - The list of tag values.
+            returned: always
+            type: complex
+            contains:
+                id:
+                    description:
+                        - The tag value ID.
+                    returned: always
+                    type: str
+                    sample: "/subscriptions/xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx/tagNames/key2/tagValues/v1"
+                tag_value:
+                    description:
+                        - The tag value.
+                    returned: always
+                    type: str
+                    sample: V1
+                count:
+                    description:
+                        - The tag value count
+                    returned: always
+                    type: dict
+                    sample: {'type': 'totoal', 'value': 1}
+tag_info:
+    description:
+        - The tag info.
+    returned: when I(scope=None)
+    type: complex
+    contains:
+        id:
+            description:
+                - The ID of the tags wrapper resource.
+            returned: always
+            type: str
+            sample: "/subscriptions/xxx-xxx/resourceGroups/testRG/providers/Microsoft.Resources/tags/default"
+        name:
+            description:
+                - The name of the tags wrapper resource.
+            returned: always
+            type: str
+            sample: default
+        type:
+            description:
+                - The type of the tags wrapper resource.
+            returned: always
+            type: str
+            sample: Microsoft.Resources/tags
+        properties:
+            description:
+                - The set of tags.
+            returned: always
+            type: dict
+            sample: { 'tags': {'key1': 'value1', 'key2': 'value2'}}
+'''
+
+from ansible_collections.azure.azcollection.plugins.module_utils.azure_rm_common import AzureRMModuleBase
+
+
+AZURE_OBJECT_CLASS = 'Tags'
+
+
+class AzureRMTagsInfo(AzureRMModuleBase):
+
+    def __init__(self):
+
+        self.module_arg_spec = dict(
+            scope=dict(type='str'),
+        )
+
+        self.results = dict(
+            changed=False,
+            tags_detail=[],
+            tag_info=None
+        )
+
+        self.scope = None
+
+        super(AzureRMTagsInfo, self).__init__(self.module_arg_spec,
+                                              supports_check_mode=True,
+                                              supports_tags=False,
+                                              facts_module=True)
+
+    def exec_module(self, **kwargs):
+        for key in self.module_arg_spec:
+            setattr(self, key, kwargs[key])
+
+        if self.scope is not None:
+            self.results['tag_info'] = self.get_at_scope(self.scope)
+        else:
+            self.results['tag_details'] = self.list_all()
+
+        return self.results
+
+    def get_at_scope(self, scope):
+        self.log('Get properties for {0}'.format(self.scope))
+        results = []
+        try:
+            response = self.rm_client.tags.get_at_scope(scope)
+            if response is not None:
+                results.append(self.format_tags(response))
+        except StopIteration:
+            pass
+        except Exception as exc:
+            self.fail('Error when get the tags info under specified scope got Excetion as {0}'.format(exc.message or str(exc)))
+        return results
+
+    def list_all(self):
+        self.log('List resources under resource group')
+        results = []
+        try:
+            response = self.rm_client.tags.list()
+            while True:
+                results.append(response.next().as_dict())
+        except StopIteration:
+            pass
+        except Exception as exc:
+            self.fail('Error when listing all tags under subscription got Excetion as {0}'.format(exc.message or str(exc)))
+        return results
+
+    def format_tags(self, tags):
+        results = dict(
+            id=tags.id,
+            name=tags.name,
+            type=tags.type,
+            properties=dict()
+        )
+        if tags.properties is not None:
+            results['properties'] = tags.properties.as_dict()
+
+        return results
+
+
+def main():
+    AzureRMTagsInfo()
+
+
+if __name__ == '__main__':
+    main()
diff --git a/pr-pipelines.yml b/pr-pipelines.yml
index 08e62f58a..b50189f12 100644
--- a/pr-pipelines.yml
+++ b/pr-pipelines.yml
@@ -163,6 +163,7 @@ parameters:
       - "azure_rm_afdorigin"
       - "azure_rm_afdruleset"
       - "azure_rm_afdrules"
+      - "azure_rm_tags"
       - "inventory_azure"
       - "setup_azure"
 
diff --git a/tests/integration/targets/azure_rm_tags/aliases b/tests/integration/targets/azure_rm_tags/aliases
new file mode 100644
index 000000000..aa77c071a
--- /dev/null
+++ b/tests/integration/targets/azure_rm_tags/aliases
@@ -0,0 +1,3 @@
+cloud/azure
+shippable/azure/group2
+destructive
diff --git a/tests/integration/targets/azure_rm_tags/meta/main.yml b/tests/integration/targets/azure_rm_tags/meta/main.yml
new file mode 100644
index 000000000..95e1952f9
--- /dev/null
+++ b/tests/integration/targets/azure_rm_tags/meta/main.yml
@@ -0,0 +1,2 @@
+dependencies:
+  - setup_azure
diff --git a/tests/integration/targets/azure_rm_tags/tasks/main.yml b/tests/integration/targets/azure_rm_tags/tasks/main.yml
new file mode 100644
index 000000000..4c73371ce
--- /dev/null
+++ b/tests/integration/targets/azure_rm_tags/tasks/main.yml
@@ -0,0 +1,154 @@
+- name: Set a random string
+  ansible.builtin.set_fact:
+    rpfx: "{{ resource_group_secondary | hash('md5') | truncate(8, True, '') }}"
+
+- name: Gather Resource Group info
+  azure_rm_resourcegroup_info:
+    name: "{{ resource_group_secondary }}"
+  register: __rg_info
+
+- name: List all the flags under the scope
+  azure_rm_tags_info:
+    scope: "{{ __rg_info.resourcegroups.0.id }}"
+  register: output
+
+- name: Create a new tags with scope
+  azure_rm_tags:
+    scope: "{{ __rg_info.resourcegroups.0.id }}"
+    tags:
+      key1: value1
+  register: output
+
+- name: Assert the tags created
+  ansible.builtin.assert:
+    that:
+      - output.changed
+
+- name: Pause for 3 mimutes
+  ansible.builtin.command: sleep 180
+  changed_when: true
+
+- name: Create a new tags with scope(Idempotent test)
+  azure_rm_tags:
+    scope: "{{ __rg_info.resourcegroups.0.id }}"
+    tags:
+      key1: value1
+  register: output
+
+- name: Assert idempotent
+  ansible.builtin.assert:
+    that:
+      - not output.changed
+
+- name: Pause for 3 mimutes
+  ansible.builtin.command: sleep 180
+  changed_when: true
+
+- name: Update the tags with scope
+  azure_rm_tags:
+    scope: "{{ __rg_info.resourcegroups.0.id }}"
+    operation: Merge
+    tags:
+      key2: value2
+  register: output
+
+- name: Assert the tags updated
+  ansible.builtin.assert:
+    that:
+      - output.changed
+
+- name: Get the tag info at thespecified resource
+  azure_rm_tags_info:
+    scope: "{{ __rg_info.resourcegroups.0.id }}"
+  register: output
+
+- name: Assert the tag facts
+  ansible.builtin.assert:
+    that:
+      - output.tag_info | length == 1
+      - output.tag_info[0].properties.tags | length == 2
+
+- name: Create a tag with tag_name
+  azure_rm_tags:
+    tag_name: "key{{ rpfx }}"
+  register: output
+
+- name: Assert the tag created
+  ansible.builtin.assert:
+    that:
+      - output.changed
+
+
+- name: Pause for 3 mimutes
+  ansible.builtin.command: sleep 180
+  changed_when: true
+
+- name: Create a tag with tag_name(Idemptent test)
+  azure_rm_tags:
+    tag_name: "key{{ rpfx }}"
+  register: output
+
+- name: Assert idempotent
+  ansible.builtin.assert:
+    that:
+      - not output.changed
+
+- name: Create a tag with tag_value
+  azure_rm_tags:
+    tag_name: "key{{ rpfx }}"
+    tag_value: "value{{ rpfx }}"
+  register: output
+
+- name: Assert the tag value created
+  ansible.builtin.assert:
+    that:
+      - output.changed
+
+- name: Pause for 3 mimutes
+  ansible.builtin.command: sleep 180
+  changed_when: true
+
+- name: Create a tag with tag_value(Idemptent test)
+  azure_rm_tags:
+    tag_name: "key{{ rpfx }}"
+    tag_value: "value{{ rpfx }}"
+  register: output
+
+- name: Assert idempotent
+  ansible.builtin.assert:
+    that:
+      - not output.changed
+
+- name: Delete the tags by scope
+  azure_rm_tags:
+    scope: "{{ __rg_info.resourcegroups.0.id }}"
+    state: absent
+  register: output
+
+- name: Assert the tag deleted
+  ansible.builtin.assert:
+    that:
+      - output.changed
+
+- name: Delete the tag with tag_value
+  azure_rm_tags:
+    tag_name: "key{{ rpfx }}"
+    tag_value: "value{{ rpfx }}"
+    state: absent
+  register: output
+
+- name: Assert the tag value deleted
+  ansible.builtin.assert:
+    that:
+      - output.changed
+
+- name: Delete the tag with tag_name
+  azure_rm_tags:
+    tag_name: "key{{ rpfx }}"
+    state: absent
+  register: output
+
+- name: Assert the tag deleted
+  ansible.builtin.assert:
+    that:
+      - output.changed