diff --git a/changelogs/fragments/migrate_ec2_vpc_vgw.yml b/changelogs/fragments/migrate_ec2_vpc_vgw.yml
new file mode 100644
index 0000000000..54bb4eb118
--- /dev/null
+++ b/changelogs/fragments/migrate_ec2_vpc_vgw.yml
@@ -0,0 +1,8 @@
+---
+major_changes:
+  - ec2_vpc_vgw - The module has been migrated from the ``community.aws`` collection.
+    Playbooks using the Fully Qualified Collection Name for this module should be
+    updated to use ``amazon.aws.ec2_vpc_vgw``.
+  - ec2_vpc_vgw_info - The module has been migrated from the ``community.aws`` collection.
+    Playbooks using the Fully Qualified Collection Name for this module should be
+    updated to use ``amazon.aws.ec2_vpc_vgw_info``.
diff --git a/meta/runtime.yml b/meta/runtime.yml
index 94614615f1..283dafde28 100644
--- a/meta/runtime.yml
+++ b/meta/runtime.yml
@@ -68,6 +68,8 @@ action_groups:
     - ec2_vpc_route_table_info
     - ec2_vpc_subnet
     - ec2_vpc_subnet_info
+    - ec2_vpc_vgw
+    - ec2_vpc_vgw_info
     - elb_application_lb
     - elb_application_lb_info
     - elb_classic_lb
diff --git a/plugins/modules/ec2_vpc_vgw.py b/plugins/modules/ec2_vpc_vgw.py
new file mode 100644
index 0000000000..5f66a9324d
--- /dev/null
+++ b/plugins/modules/ec2_vpc_vgw.py
@@ -0,0 +1,535 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+# Copyright: Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+DOCUMENTATION = r"""
+module: ec2_vpc_vgw
+short_description: Create and delete AWS VPN Virtual Gateways
+version_added: 1.0.0
+version_added_collection: community.aws
+description:
+  - Creates AWS VPN Virtual Gateways
+  - Deletes AWS VPN Virtual Gateways
+  - Attaches Virtual Gateways to VPCs
+  - Detaches Virtual Gateways from VPCs
+options:
+  state:
+    description:
+      - C(present) to ensure resource is created.
+      - C(absent) to remove resource.
+    default: present
+    choices: [ "present", "absent"]
+    type: str
+  name:
+    description:
+      - Name of the VGW to be created or deleted.
+    type: str
+  type:
+    description:
+      - Type of the virtual gateway to be created.
+    choices: [ "ipsec.1" ]
+    default: "ipsec.1"
+    type: str
+  vpn_gateway_id:
+    description:
+      - VPN gateway ID of an existing virtual gateway.
+    type: str
+  vpc_id:
+    description:
+      - The ID of a VPC to attach or detach to the VGW.
+    type: str
+  asn:
+    description:
+      - The BGP ASN on the Amazon side.
+    type: int
+  wait_timeout:
+    description:
+      - Number of seconds to wait for status during VPC attach and detach.
+    default: 320
+    type: int
+notes:
+  - Support for I(purge_tags) was added in release 4.0.0.
+author:
+  - Nick Aslanidis (@naslanidis)
+extends_documentation_fragment:
+  - amazon.aws.common.modules
+  - amazon.aws.region.modules
+  - amazon.aws.tags
+  - amazon.aws.boto3
+"""
+
+EXAMPLES = r"""
+- name: Create a new VGW attached to a specific VPC
+  amazon.aws.ec2_vpc_vgw:
+    state: present
+    region: ap-southeast-2
+    profile: personal
+    vpc_id: vpc-12345678
+    name: personal-testing
+    type: ipsec.1
+  register: created_vgw
+
+- name: Create a new unattached VGW
+  amazon.aws.ec2_vpc_vgw:
+    state: present
+    region: ap-southeast-2
+    profile: personal
+    name: personal-testing
+    type: ipsec.1
+    tags:
+      environment: production
+      owner: ABC
+  register: created_vgw
+
+- name: Remove a new VGW using the name
+  amazon.aws.ec2_vpc_vgw:
+    state: absent
+    region: ap-southeast-2
+    profile: personal
+    name: personal-testing
+    type: ipsec.1
+  register: deleted_vgw
+
+- name: Remove a new VGW using the vpn_gateway_id
+  amazon.aws.ec2_vpc_vgw:
+    state: absent
+    region: ap-southeast-2
+    profile: personal
+    vpn_gateway_id: vgw-3a9aa123
+  register: deleted_vgw
+"""
+
+RETURN = r"""
+vgw:
+  description: A description of the VGW
+  returned: success
+  type: dict
+  contains:
+    id:
+      description: The ID of the VGW.
+      type: str
+      returned: success
+      example: "vgw-0123456789abcdef0"
+    state:
+      description: The state of the VGW.
+      type: str
+      returned: success
+      example: "available"
+    tags:
+      description: A dictionary representing the tags attached to the VGW
+      type: dict
+      returned: success
+      example: { "Name": "ansible-test-ec2-vpc-vgw" }
+    type:
+      description: The type of VPN connection the virtual private gateway supports.
+      type: str
+      returned: success
+      example: "ipsec.1"
+    vpc_id:
+      description: The ID of the VPC to which the VGW is attached.
+      type: str
+      returned: success
+      example: vpc-123456789abcdef01
+"""
+
+import time
+
+try:
+    import botocore
+except ImportError:
+    pass  # Handled by AnsibleAWSModule
+
+from ansible_collections.amazon.aws.plugins.module_utils.botocore import is_boto3_error_code
+from ansible_collections.amazon.aws.plugins.module_utils.ec2 import ensure_ec2_tags
+from ansible_collections.amazon.aws.plugins.module_utils.modules import AnsibleAWSModule
+from ansible_collections.amazon.aws.plugins.module_utils.retries import AWSRetry
+from ansible_collections.amazon.aws.plugins.module_utils.tagging import boto3_tag_list_to_ansible_dict
+from ansible_collections.amazon.aws.plugins.module_utils.tagging import boto3_tag_specifications
+from ansible_collections.amazon.aws.plugins.module_utils.waiters import get_waiter
+
+
+# AWS uses VpnGatewayLimitExceeded for both 'Too many VGWs' and 'Too many concurrent changes'
+# we need to look at the mesage to tell the difference.
+class VGWRetry(AWSRetry):
+    @staticmethod
+    def status_code_from_exception(error):
+        return (
+            error.response["Error"]["Code"],
+            error.response["Error"]["Message"],
+        )
+
+    @staticmethod
+    def found(response_code, catch_extra_error_codes=None):
+        retry_on = ["The maximum number of mutating objects has been reached."]
+
+        if catch_extra_error_codes:
+            retry_on.extend(catch_extra_error_codes)
+        if not isinstance(response_code, tuple):
+            response_code = (response_code,)
+
+        for code in response_code:
+            if super(VGWRetry, VGWRetry).found(response_code, catch_extra_error_codes):
+                return True
+
+        return False
+
+
+def get_vgw_info(vgws):
+    if not isinstance(vgws, list):
+        return
+
+    for vgw in vgws:
+        vgw_info = {
+            "id": vgw["VpnGatewayId"],
+            "type": vgw["Type"],
+            "state": vgw["State"],
+            "vpc_id": None,
+            "tags": dict(),
+        }
+
+        if vgw["Tags"]:
+            vgw_info["tags"] = boto3_tag_list_to_ansible_dict(vgw["Tags"])
+
+        if len(vgw["VpcAttachments"]) != 0 and vgw["VpcAttachments"][0]["State"] == "attached":
+            vgw_info["vpc_id"] = vgw["VpcAttachments"][0]["VpcId"]
+
+        return vgw_info
+
+
+def wait_for_status(client, module, vpn_gateway_id, status):
+    polling_increment_secs = 15
+    max_retries = module.params.get("wait_timeout") // polling_increment_secs
+    status_achieved = False
+
+    for x in range(0, max_retries):
+        try:
+            response = find_vgw(client, module, vpn_gateway_id)
+            if response[0]["VpcAttachments"][0]["State"] == status:
+                status_achieved = True
+                break
+            else:
+                time.sleep(polling_increment_secs)
+        except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e:
+            module.fail_json_aws(e, msg="Failure while waiting for status update")
+
+    result = response
+    return status_achieved, result
+
+
+def attach_vgw(client, module, vpn_gateway_id):
+    params = dict()
+    params["VpcId"] = module.params.get("vpc_id")
+
+    try:
+        # Immediately after a detachment, the EC2 API sometimes will report the VpnGateways[0].State
+        # as available several seconds before actually permitting a new attachment.
+        # So we catch and retry that error.  See https://github.com/ansible/ansible/issues/53185
+        response = VGWRetry.jittered_backoff(retries=5, catch_extra_error_codes=["InvalidParameterValue"])(
+            client.attach_vpn_gateway
+        )(VpnGatewayId=vpn_gateway_id, VpcId=params["VpcId"])
+    except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e:
+        module.fail_json_aws(e, msg="Failed to attach VPC")
+
+    status_achieved, vgw = wait_for_status(client, module, [vpn_gateway_id], "attached")
+    if not status_achieved:
+        module.fail_json(msg="Error waiting for vpc to attach to vgw - please check the AWS console")
+
+    result = response
+    return result
+
+
+def detach_vgw(client, module, vpn_gateway_id, vpc_id=None):
+    params = dict()
+    params["VpcId"] = module.params.get("vpc_id")
+
+    try:
+        if vpc_id:
+            response = client.detach_vpn_gateway(VpnGatewayId=vpn_gateway_id, VpcId=vpc_id, aws_retry=True)
+        else:
+            response = client.detach_vpn_gateway(VpnGatewayId=vpn_gateway_id, VpcId=params["VpcId"], aws_retry=True)
+    except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e:
+        module.fail_json_aws(e, "Failed to detach gateway")
+
+    status_achieved, vgw = wait_for_status(client, module, [vpn_gateway_id], "detached")
+    if not status_achieved:
+        module.fail_json(msg="Error waiting for  vpc to detach from vgw - please check the AWS console")
+
+    result = response
+    return result
+
+
+def create_vgw(client, module):
+    params = dict()
+    params["Type"] = module.params.get("type")
+    tags = module.params.get("tags") or {}
+    tags["Name"] = module.params.get("name")
+    params["TagSpecifications"] = boto3_tag_specifications(tags, ["vpn-gateway"])
+    if module.params.get("asn"):
+        params["AmazonSideAsn"] = module.params.get("asn")
+
+    try:
+        response = client.create_vpn_gateway(aws_retry=True, **params)
+        get_waiter(client, "vpn_gateway_exists").wait(VpnGatewayIds=[response["VpnGateway"]["VpnGatewayId"]])
+    except botocore.exceptions.WaiterError as e:
+        module.fail_json_aws(
+            e, msg=f"Failed to wait for Vpn Gateway {response['VpnGateway']['VpnGatewayId']} to be available"
+        )
+    except is_boto3_error_code("VpnGatewayLimitExceeded") as e:
+        module.fail_json_aws(e, msg="Too many VPN gateways exist in this account.")
+    except (
+        botocore.exceptions.ClientError,
+        botocore.exceptions.BotoCoreError,
+    ) as e:  # pylint: disable=duplicate-except
+        module.fail_json_aws(e, msg="Failed to create gateway")
+
+    result = response
+    return result
+
+
+def delete_vgw(client, module, vpn_gateway_id):
+    try:
+        response = client.delete_vpn_gateway(VpnGatewayId=vpn_gateway_id, aws_retry=True)
+    except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e:
+        module.fail_json_aws(e, msg="Failed to delete gateway")
+
+    # return the deleted VpnGatewayId as this is not included in the above response
+    result = vpn_gateway_id
+    return result
+
+
+def find_vpc(client, module):
+    params = dict()
+    params["vpc_id"] = module.params.get("vpc_id")
+
+    if params["vpc_id"]:
+        try:
+            response = client.describe_vpcs(VpcIds=[params["vpc_id"]], aws_retry=True)
+        except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e:
+            module.fail_json_aws(e, msg="Failed to describe VPC")
+
+    result = response
+    return result
+
+
+def find_vgw(client, module, vpn_gateway_id=None):
+    params = dict()
+    if vpn_gateway_id:
+        params["VpnGatewayIds"] = vpn_gateway_id
+    else:
+        params["Filters"] = [
+            {"Name": "type", "Values": [module.params.get("type")]},
+            {"Name": "tag:Name", "Values": [module.params.get("name")]},
+        ]
+        if module.params.get("state") == "present":
+            params["Filters"].append({"Name": "state", "Values": ["pending", "available"]})
+    try:
+        response = client.describe_vpn_gateways(aws_retry=True, **params)
+    except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e:
+        module.fail_json_aws(e, msg="Failed to describe gateway using filters")
+
+    return sorted(response["VpnGateways"], key=lambda k: k["VpnGatewayId"])
+
+
+def ensure_vgw_present(client, module):
+    # If an existing vgw name and type matches our args, then a match is considered to have been
+    # found and we will not create another vgw.
+
+    changed = False
+    params = dict()
+    result = dict()
+    params["Name"] = module.params.get("name")
+    params["VpcId"] = module.params.get("vpc_id")
+    params["Type"] = module.params.get("type")
+    params["Tags"] = module.params.get("tags")
+    params["VpnGatewayIds"] = module.params.get("vpn_gateway_id")
+
+    # check that the vpc_id exists. If not, an exception is thrown
+    if params["VpcId"]:
+        vpc = find_vpc(client, module)
+
+    # check if a gateway matching our module args already exists
+    existing_vgw = find_vgw(client, module)
+
+    if existing_vgw != []:
+        vpn_gateway_id = existing_vgw[0]["VpnGatewayId"]
+        desired_tags = module.params.get("tags")
+        purge_tags = module.params.get("purge_tags")
+        if desired_tags is None:
+            desired_tags = dict()
+            purge_tags = False
+        tags = dict(Name=module.params.get("name"))
+        tags.update(desired_tags)
+        changed = ensure_ec2_tags(
+            client, module, vpn_gateway_id, resource_type="vpn-gateway", tags=tags, purge_tags=purge_tags
+        )
+
+        # if a vpc_id was provided, check if it exists and if it's attached
+        if params["VpcId"]:
+            current_vpc_attachments = existing_vgw[0]["VpcAttachments"]
+
+            if current_vpc_attachments != [] and current_vpc_attachments[0]["State"] == "attached":
+                if (
+                    current_vpc_attachments[0]["VpcId"] != params["VpcId"]
+                    or current_vpc_attachments[0]["State"] != "attached"
+                ):
+                    # detach the existing vpc from the virtual gateway
+                    vpc_to_detach = current_vpc_attachments[0]["VpcId"]
+                    detach_vgw(client, module, vpn_gateway_id, vpc_to_detach)
+                    get_waiter(client, "vpn_gateway_detached").wait(VpnGatewayIds=[vpn_gateway_id])
+                    attached_vgw = attach_vgw(client, module, vpn_gateway_id)
+                    changed = True
+            else:
+                # attach the vgw to the supplied vpc
+                attached_vgw = attach_vgw(client, module, vpn_gateway_id)
+                changed = True
+
+        # if params['VpcId'] is not provided, check the vgw is attached to a vpc. if so, detach it.
+        else:
+            existing_vgw = find_vgw(client, module, [vpn_gateway_id])
+
+            if existing_vgw[0]["VpcAttachments"] != []:
+                if existing_vgw[0]["VpcAttachments"][0]["State"] == "attached":
+                    # detach the vpc from the vgw
+                    vpc_to_detach = existing_vgw[0]["VpcAttachments"][0]["VpcId"]
+                    detach_vgw(client, module, vpn_gateway_id, vpc_to_detach)
+                    changed = True
+
+    else:
+        # create a new vgw
+        new_vgw = create_vgw(client, module)
+        changed = True
+        vpn_gateway_id = new_vgw["VpnGateway"]["VpnGatewayId"]
+
+        # if a vpc-id was supplied, attempt to attach it to the vgw
+        if params["VpcId"]:
+            attached_vgw = attach_vgw(client, module, vpn_gateway_id)
+            changed = True
+
+    # return current state of the vgw
+    vgw = find_vgw(client, module, [vpn_gateway_id])
+    result = get_vgw_info(vgw)
+    return changed, result
+
+
+def ensure_vgw_absent(client, module):
+    # If an existing vgw name and type matches our args, then a match is considered to have been
+    # found and we will take steps to delete it.
+
+    changed = False
+    params = dict()
+    result = dict()
+    params["Name"] = module.params.get("name")
+    params["VpcId"] = module.params.get("vpc_id")
+    params["Type"] = module.params.get("type")
+    params["Tags"] = module.params.get("tags")
+    params["VpnGatewayIds"] = module.params.get("vpn_gateway_id")
+    deleted_vgw = None
+
+    # check if a gateway matching our module args already exists
+    if params["VpnGatewayIds"]:
+        existing_vgw_with_id = find_vgw(client, module, [params["VpnGatewayIds"]])
+        if existing_vgw_with_id != [] and existing_vgw_with_id[0]["State"] != "deleted":
+            existing_vgw = existing_vgw_with_id
+            if existing_vgw[0]["VpcAttachments"] != [] and existing_vgw[0]["VpcAttachments"][0]["State"] == "attached":
+                if params["VpcId"]:
+                    if params["VpcId"] != existing_vgw[0]["VpcAttachments"][0]["VpcId"]:
+                        module.fail_json(
+                            msg="The vpc-id provided does not match the vpc-id currently attached - please check the AWS console"
+                        )
+
+                    else:
+                        # detach the vpc from the vgw
+                        detach_vgw(client, module, params["VpnGatewayIds"], params["VpcId"])
+                        deleted_vgw = delete_vgw(client, module, params["VpnGatewayIds"])
+                        changed = True
+
+                else:
+                    # attempt to detach any attached vpcs
+                    vpc_to_detach = existing_vgw[0]["VpcAttachments"][0]["VpcId"]
+                    detach_vgw(client, module, params["VpnGatewayIds"], vpc_to_detach)
+                    deleted_vgw = delete_vgw(client, module, params["VpnGatewayIds"])
+                    changed = True
+
+            else:
+                # no vpc's are attached so attempt to delete the vgw
+                deleted_vgw = delete_vgw(client, module, params["VpnGatewayIds"])
+                changed = True
+
+        else:
+            changed = False
+            deleted_vgw = "Nothing to do"
+
+    else:
+        # Check that a name and type argument has been supplied if no vgw-id
+        if not module.params.get("name") or not module.params.get("type"):
+            module.fail_json(msg="A name and type is required when no vgw-id and a status of 'absent' is supplied")
+
+        existing_vgw = find_vgw(client, module)
+        if existing_vgw != [] and existing_vgw[0]["State"] != "deleted":
+            vpn_gateway_id = existing_vgw[0]["VpnGatewayId"]
+            if existing_vgw[0]["VpcAttachments"] != [] and existing_vgw[0]["VpcAttachments"][0]["State"] == "attached":
+                if params["VpcId"]:
+                    if params["VpcId"] != existing_vgw[0]["VpcAttachments"][0]["VpcId"]:
+                        module.fail_json(
+                            msg="The vpc-id provided does not match the vpc-id currently attached - please check the AWS console"
+                        )
+
+                    else:
+                        # detach the vpc from the vgw
+                        detach_vgw(client, module, vpn_gateway_id, params["VpcId"])
+
+                        # now that the vpc has been detached, delete the vgw
+                        deleted_vgw = delete_vgw(client, module, vpn_gateway_id)
+                        changed = True
+
+                else:
+                    # attempt to detach any attached vpcs
+                    vpc_to_detach = existing_vgw[0]["VpcAttachments"][0]["VpcId"]
+                    detach_vgw(client, module, vpn_gateway_id, vpc_to_detach)
+                    changed = True
+
+                    # now that the vpc has been detached, delete the vgw
+                    deleted_vgw = delete_vgw(client, module, vpn_gateway_id)
+
+            else:
+                # no vpc's are attached so attempt to delete the vgw
+                deleted_vgw = delete_vgw(client, module, vpn_gateway_id)
+                changed = True
+
+        else:
+            changed = False
+            deleted_vgw = None
+
+    result = deleted_vgw
+    return changed, result
+
+
+def main():
+    argument_spec = dict(
+        state=dict(default="present", choices=["present", "absent"]),
+        name=dict(),
+        vpn_gateway_id=dict(),
+        vpc_id=dict(),
+        asn=dict(type="int"),
+        wait_timeout=dict(type="int", default=320),
+        type=dict(default="ipsec.1", choices=["ipsec.1"]),
+        tags=dict(default=None, required=False, type="dict", aliases=["resource_tags"]),
+        purge_tags=dict(default=True, type="bool"),
+    )
+    module = AnsibleAWSModule(argument_spec=argument_spec, required_if=[["state", "present", ["name"]]])
+
+    state = module.params.get("state").lower()
+
+    client = module.client("ec2", retry_decorator=VGWRetry.jittered_backoff(retries=10))
+
+    if state == "present":
+        (changed, results) = ensure_vgw_present(client, module)
+    else:
+        (changed, results) = ensure_vgw_absent(client, module)
+    module.exit_json(changed=changed, vgw=results)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/plugins/modules/ec2_vpc_vgw_info.py b/plugins/modules/ec2_vpc_vgw_info.py
new file mode 100644
index 0000000000..d392ca683d
--- /dev/null
+++ b/plugins/modules/ec2_vpc_vgw_info.py
@@ -0,0 +1,190 @@
+#!/usr/bin/python
+# -*- coding: utf-8 -*-
+
+# Copyright: Ansible Project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+DOCUMENTATION = r"""
+---
+module: ec2_vpc_vgw_info
+version_added: 1.0.0
+version_added_collection: community.aws
+short_description: Gather information about virtual gateways in AWS
+description:
+  - Gather information about virtual gateways (VGWs) in AWS.
+options:
+  filters:
+    description:
+      - A dict of filters to apply. Each dict item consists of a filter key and a filter value.
+        See U(https://docs.aws.amazon.com/AWSEC2/latest/APIReference/API_DescribeVpnGateways.html) for possible filters.
+    type: dict
+    default: {}
+  vpn_gateway_ids:
+    description:
+      - Get details of a specific Virtual Gateway ID.
+    type: list
+    elements: str
+author:
+  - "Nick Aslanidis (@naslanidis)"
+extends_documentation_fragment:
+  - amazon.aws.common.modules
+  - amazon.aws.region.modules
+  - amazon.aws.boto3
+"""
+
+EXAMPLES = r"""
+# # Note: These examples do not set authentication details, see the AWS Guide for details.
+
+- name: Gather information about all virtual gateways for an account or profile
+  amazon.aws.ec2_vpc_vgw_info:
+    region: ap-southeast-2
+    profile: production
+  register: vgw_info
+
+- name: Gather information about a filtered list of Virtual Gateways
+  amazon.aws.ec2_vpc_vgw_info:
+    region: ap-southeast-2
+    profile: production
+    filters:
+      "tag:Name": "main-virt-gateway"
+  register: vgw_info
+
+- name: Gather information about a specific virtual gateway by VpnGatewayIds
+  amazon.aws.ec2_vpc_vgw_info:
+    region: ap-southeast-2
+    profile: production
+    vpn_gateway_ids: vgw-c432f6a7
+  register: vgw_info
+"""
+
+RETURN = r"""
+virtual_gateways:
+    description: The virtual gateways for the account.
+    returned: always
+    type: list
+    elements: dict
+    contains:
+      vpn_gateway_id:
+        description: The ID of the VGW.
+        type: str
+        returned: success
+        example: "vgw-0123456789abcdef0"
+      state:
+        description: The current state of the VGW.
+        type: str
+        returned: success
+        example: "available"
+      type:
+        description: The type of VPN connection the VGW supports.
+        type: str
+        returned: success
+        example: "ipsec.1"
+      vpc_attachments:
+        description: A description of the attachment of VPCs to the VGW.
+        type: list
+        elements: dict
+        returned: success
+        contains:
+          state:
+            description: The current state of the attachment.
+            type: str
+            returned: success
+            example: available
+          vpc_id:
+            description: The ID of the VPC.
+            type: str
+            returned: success
+            example: vpc-12345678901234567
+      tags:
+        description:
+          - A list of dictionaries representing the tags attached to the VGW.
+          - Represents the same details as I(resource_tags).
+        type: list
+        elements: dict
+        returned: success
+        contains:
+          key:
+            description: The key of the tag.
+            type: str
+            returned: success
+            example: MyKey
+          value:
+            description: The value of the tag.
+            type: str
+            returned: success
+            example: MyValue
+      resource_tags:
+        description:
+          - A dictionary representing the tags attached to the VGW.
+          - Represents the same details as I(tags).
+        type: dict
+        returned: success
+        example: {"MyKey": "MyValue"}
+"""
+
+try:
+    import botocore
+except ImportError:
+    pass  # Handled by AnsibleAWSModule
+
+from ansible.module_utils.common.dict_transformations import camel_dict_to_snake_dict
+
+from ansible_collections.amazon.aws.plugins.module_utils.modules import AnsibleAWSModule
+from ansible_collections.amazon.aws.plugins.module_utils.tagging import boto3_tag_list_to_ansible_dict
+from ansible_collections.amazon.aws.plugins.module_utils.transformation import ansible_dict_to_boto3_filter_list
+
+
+def get_virtual_gateway_info(virtual_gateway):
+    tags = virtual_gateway.get("Tags", [])
+    resource_tags = boto3_tag_list_to_ansible_dict(tags)
+    virtual_gateway_info = dict(
+        VpnGatewayId=virtual_gateway["VpnGatewayId"],
+        State=virtual_gateway["State"],
+        Type=virtual_gateway["Type"],
+        VpcAttachments=virtual_gateway["VpcAttachments"],
+        Tags=tags,
+        ResourceTags=resource_tags,
+    )
+    return virtual_gateway_info
+
+
+def list_virtual_gateways(client, module):
+    params = dict()
+
+    params["Filters"] = ansible_dict_to_boto3_filter_list(module.params.get("filters"))
+
+    if module.params.get("vpn_gateway_ids"):
+        params["VpnGatewayIds"] = module.params.get("vpn_gateway_ids")
+
+    try:
+        all_virtual_gateways = client.describe_vpn_gateways(**params)
+    except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e:
+        module.fail_json_aws(e, msg="Failed to list gateways")
+
+    return [
+        camel_dict_to_snake_dict(get_virtual_gateway_info(vgw), ignore_list=["ResourceTags"])
+        for vgw in all_virtual_gateways["VpnGateways"]
+    ]
+
+
+def main():
+    argument_spec = dict(
+        filters=dict(type="dict", default=dict()),
+        vpn_gateway_ids=dict(type="list", default=None, elements="str"),
+    )
+
+    module = AnsibleAWSModule(argument_spec=argument_spec, supports_check_mode=True)
+
+    try:
+        connection = module.client("ec2")
+    except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e:
+        module.fail_json_aws(e, msg="Failed to connect to AWS")
+
+    # call your function here
+    results = list_virtual_gateways(connection, module)
+
+    module.exit_json(virtual_gateways=results)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/integration/targets/ec2_vpc_vgw/aliases b/tests/integration/targets/ec2_vpc_vgw/aliases
new file mode 100644
index 0000000000..a53885886c
--- /dev/null
+++ b/tests/integration/targets/ec2_vpc_vgw/aliases
@@ -0,0 +1,3 @@
+cloud/aws
+time=3m
+ec2_vpc_vgw_info
diff --git a/tests/integration/targets/ec2_vpc_vgw/defaults/main.yml b/tests/integration/targets/ec2_vpc_vgw/defaults/main.yml
new file mode 100644
index 0000000000..76bb479bbd
--- /dev/null
+++ b/tests/integration/targets/ec2_vpc_vgw/defaults/main.yml
@@ -0,0 +1,11 @@
+vpc_name: '{{ resource_prefix }}-ec2-vpc-vgw'
+vgw_name: '{{ resource_prefix }}-ec2-vpc-vgw'
+subnet_name: '{{ resource_prefix }}-ec2-vpc-vgw'
+vpc_cidr: 10.{{ 256 | random(seed=resource_prefix) }}.0.0/16
+subnet_1: 10.{{ 256 | random(seed=resource_prefix) }}.1.0/24
+subnet_2: 10.{{ 256 | random(seed=resource_prefix) }}.2.0/24
+subnet_3: 10.{{ 256 | random(seed=resource_prefix) }}.3.0/24
+subnet_4: 10.{{ 256 | random(seed=resource_prefix) }}.4.0/24
+
+vpc_ipv6_cidr: 10.{{ 256 | random(seed=resource_prefix) }}.5.0/25
+vpc_ipv6_name: '{{ vpc_name }}-ipv6'
diff --git a/tests/integration/targets/ec2_vpc_vgw/meta/main.yml b/tests/integration/targets/ec2_vpc_vgw/meta/main.yml
new file mode 100644
index 0000000000..32cf5dda7e
--- /dev/null
+++ b/tests/integration/targets/ec2_vpc_vgw/meta/main.yml
@@ -0,0 +1 @@
+dependencies: []
diff --git a/tests/integration/targets/ec2_vpc_vgw/tasks/main.yml b/tests/integration/targets/ec2_vpc_vgw/tasks/main.yml
new file mode 100644
index 0000000000..c776a8d7d7
--- /dev/null
+++ b/tests/integration/targets/ec2_vpc_vgw/tasks/main.yml
@@ -0,0 +1,225 @@
+- name: ec2_vpc_vgw integration tests
+  module_defaults:
+    group/aws:
+      access_key: '{{ aws_access_key }}'
+      secret_key: '{{ aws_secret_key }}'
+      session_token: '{{ security_token | default(omit) }}'
+      region: '{{ aws_region }}'
+  block:
+
+    # ============================================================
+  - debug: msg="Setting up test dependencies"
+
+  - name: create a VPC
+    ec2_vpc_net:
+      name: '{{ vpc_name }}-{{ item }}'
+      state: present
+      cidr_block: '{{ vpc_cidr }}'
+      tags:
+        Description: Created by ansible-test for IGW tests
+    register: vpc_result
+    loop: [1, 2]
+
+  - name: use set fact for vpc ids
+    set_fact:
+      vpc_id_1: '{{ vpc_result.results.0.vpc.id }}'
+      vpc_id_2: '{{ vpc_result.results.1.vpc.id }}'
+
+    # ============================================================
+  - debug: msg="Running tests"
+
+  - name: create vpn gateway and attach it to vpc
+    ec2_vpc_vgw:
+      state: present
+      vpc_id: '{{ vpc_id_1 }}'
+      name: '{{ vgw_name }}'
+    register: vgw
+
+  - name: use set fact for vgw ids
+    set_fact:
+      vgw_id: '{{ vgw.vgw.id }}'
+
+  - assert:
+      that:
+      - vgw.changed
+      - vgw.vgw.vpc_id == vpc_id_1
+      - vgw.vgw.tags.Name == vgw_name
+
+  - name: test idempotence
+    ec2_vpc_vgw:
+      state: present
+      vpc_id: '{{ vpc_id_1 }}'
+      name: '{{ vgw_name }}'
+    register: vgw
+
+  - assert:
+      that:
+      - not vgw.changed
+      - vgw.vgw.id == vgw_id
+
+    # ============================================================
+  - name: attach vpn gateway to the other VPC
+    ec2_vpc_vgw:
+      state: present
+      vpc_id: '{{ vpc_id_2 }}'
+      name: '{{ vgw_name }}'
+    register: vgw
+
+  - assert:
+      that:
+      - vgw.changed
+      - vgw.vgw.id == vgw_id
+      - vgw.vgw.vpc_id == vpc_id_2
+
+    # ============================================================
+
+  - name: get VPC VGW facts by ID (CHECK)
+    ec2_vpc_vgw_info:
+      vpn_gateway_ids: ['{{ vgw_id }}']
+    register: vgw_info
+    check_mode: true
+
+  - name: verify expected facts
+    vars:
+      vgw_details: '{{ vgw_info.virtual_gateways[0] }}'
+      attach_1_description:
+        state: detached
+        vpc_id: '{{ vpc_id_1 }}'
+      attach_2_description:
+        state: attached
+        vpc_id: '{{ vpc_id_2 }}'
+    assert:
+      that:
+      - vgw_info.virtual_gateways | length == 1
+      - '"resource_tags" in vgw_details'
+      - '"state" in vgw_details'
+      - '"tags" in vgw_details'
+      - '"type" in vgw_details'
+      - '"vpc_attachments" in vgw_details'
+      - '"vpn_gateway_id" in vgw_details'
+      - vgw_details.vpn_gateway_id == vgw_id
+      - vgw_details.type == 'ipsec.1'
+      - vgw_details.state == 'available'
+      - '"Name" in vgw_details.resource_tags'
+      - vgw_details.resource_tags.Name == vgw_name
+      - ( attach_1_description in vgw_details.vpc_attachments and vgw_details.vpc_attachments
+        | length == 2 ) or ( vgw_details.vpc_attachments | length == 1 )
+      - attach_2_description in vgw_details.vpc_attachments
+
+  - name: get VPC VGW facts by Tag
+    ec2_vpc_vgw_info:
+      filters:
+        tag:Name: '{{ vgw_name }}'
+    register: vgw_info
+
+  - name: verify expected facts
+    vars:
+      vgw_details: '{{ vgw_info.virtual_gateways[0] }}'
+      attach_1_description:
+        state: detached
+        vpc_id: '{{ vpc_id_1 }}'
+      attach_2_description:
+        state: attached
+        vpc_id: '{{ vpc_id_2 }}'
+    assert:
+      that:
+      - vgw_info.virtual_gateways | length == 1
+      - '"resource_tags" in vgw_details'
+      - '"state" in vgw_details'
+      - '"tags" in vgw_details'
+      - '"type" in vgw_details'
+      - '"vpc_attachments" in vgw_details'
+      - '"vpn_gateway_id" in vgw_details'
+      - vgw_details.vpn_gateway_id == vgw_id
+      - vgw_details.type == 'ipsec.1'
+      - vgw_details.state == 'available'
+      - '"Name" in vgw_details.resource_tags'
+      - vgw_details.resource_tags.Name == vgw_name
+      - ( attach_1_description in vgw_details.vpc_attachments and vgw_details.vpc_attachments
+        | length == 2 ) or ( vgw_details.vpc_attachments | length == 1 )
+      - attach_2_description in vgw_details.vpc_attachments
+
+    # ============================================================
+
+  - name: get all VGWs
+    ec2_vpc_vgw_info:
+    register: vgw_info
+
+  - name: verify test VGW is in the results
+    vars:
+      vgw_id_list: '{{ vgw_info.virtual_gateways | map(attribute="vpn_gateway_id")
+        | list }}'
+    assert:
+      that:
+      - vgw_id in vgw_id_list
+
+    # ============================================================
+
+  - name: detach vpn gateway
+    ec2_vpc_vgw:
+      state: present
+      name: '{{ vgw_name }}'
+    register: vgw
+
+  - assert:
+      that:
+      - vgw.changed
+      - not vgw.vgw.vpc_id
+
+  - name: test idempotence
+    ec2_vpc_vgw:
+      state: present
+      name: '{{ vgw_name }}'
+    register: vgw
+
+  - assert:
+      that:
+      - not vgw.changed
+
+    # ============================================================
+
+  - include_tasks: tags.yml
+
+    # ============================================================
+
+  - name: delete vpn gateway
+    ec2_vpc_vgw:
+      state: absent
+      name: '{{ vgw_name }}'
+    register: vgw
+
+  - assert:
+      that:
+      - vgw.changed
+
+  - name: test idempotence
+    ec2_vpc_vgw:
+      state: absent
+      name: '{{ vgw_name }}'
+    register: vgw
+
+  - assert:
+      that:
+      - not vgw.changed
+
+  always:
+
+  - debug: msg="Removing test dependencies"
+
+  - name: delete vpn gateway
+    ec2_vpc_vgw:
+      state: absent
+      vpn_gateway_id: '{{ vgw.vgw.id | default(vgw_id) }}'
+    ignore_errors: yes
+
+  - name: delete vpc
+    ec2_vpc_net:
+      name: '{{ vpc_name }}-{{ item }}'
+      state: absent
+      cidr_block: '{{ vpc_cidr }}'
+    loop: [1, 2]
+    register: result
+    retries: 10
+    delay: 5
+    until: result is not failed
+    ignore_errors: true
diff --git a/tests/integration/targets/ec2_vpc_vgw/tasks/tags.yml b/tests/integration/targets/ec2_vpc_vgw/tasks/tags.yml
new file mode 100644
index 0000000000..e69de29bb2