diff --git a/prowler/providers/aws/services/firehose/__init__.py b/prowler/providers/aws/services/firehose/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/prowler/providers/aws/services/firehose/firehose_client.py b/prowler/providers/aws/services/firehose/firehose_client.py
new file mode 100644
index 00000000000..46d3bdeab7e
--- /dev/null
+++ b/prowler/providers/aws/services/firehose/firehose_client.py
@@ -0,0 +1,4 @@
+from prowler.providers.aws.services.firehose.firehose_service import Firehose
+from prowler.providers.common.provider import Provider
+
+firehose_client = Firehose(Provider.get_global_provider())
diff --git a/prowler/providers/aws/services/firehose/firehose_service.py b/prowler/providers/aws/services/firehose/firehose_service.py
new file mode 100644
index 00000000000..fafd87cd3c9
--- /dev/null
+++ b/prowler/providers/aws/services/firehose/firehose_service.py
@@ -0,0 +1,58 @@
+from typing import Dict, List, Optional
+
+from botocore.client import ClientError
+from pydantic import BaseModel, Field
+
+from prowler.lib.logger import logger
+from prowler.lib.scan_filters.scan_filters import is_resource_filtered
+from prowler.providers.aws.lib.service.service import AWSService
+
+
+class Firehose(AWSService):
+    def __init__(self, provider):
+        # Call AWSService's __init__
+        super().__init__(__class__.__name__, provider)
+        self.delivery_streams = {}
+        self.__threading_call__(self._list_delivery_streams)
+        self.__threading_call__(
+            self._list_tags_for_delivery_stream, self.delivery_streams.values()
+        )
+
+    def _list_delivery_streams(self, regional_client):
+        logger.info("Firehose - Listing delivery streams...")
+        try:
+            for stream_name in regional_client.list_delivery_streams()[
+                "DeliveryStreamNames"
+            ]:
+                stream_arn = f"arn:{self.audited_partition}:firehose:{regional_client.region}:{self.audited_account}:deliverystream/{stream_name}"
+                if not self.audit_resources or (
+                    is_resource_filtered(stream_arn, self.audit_resources)
+                ):
+                    self.delivery_streams[stream_arn] = DeliveryStream(
+                        arn=stream_arn,
+                        name=stream_name,
+                        region=regional_client.region,
+                    )
+        except ClientError as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+    def _list_tags_for_delivery_stream(self, stream):
+        try:
+            stream.tags = (
+                self.regional_clients[stream.region]
+                .list_tags_for_delivery_stream(DeliveryStreamName=stream.name)
+                .get("Tags", [])
+            )
+        except ClientError as error:
+            logger.error(
+                f"{stream.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+
+class DeliveryStream(BaseModel):
+    arn: str
+    name: str
+    region: str
+    tags: Optional[List[Dict[str, str]]] = Field(default_factory=list)
diff --git a/prowler/providers/aws/services/mq/mq_service.py b/prowler/providers/aws/services/mq/mq_service.py
index c66c05a4858..c4b5f99f001 100644
--- a/prowler/providers/aws/services/mq/mq_service.py
+++ b/prowler/providers/aws/services/mq/mq_service.py
@@ -11,7 +11,7 @@
 class MQ(AWSService):
     def __init__(self, provider):
         # Call AWSService's __init__
-        super().__init__("mq", provider)
+        super().__init__(__class__.__name__, provider)
         self.brokers = {}
         self.__threading_call__(self._list_brokers)
         self.__threading_call__(self._describe_broker, self.brokers.values())
diff --git a/tests/providers/aws/services/firehose/firehose_service_test.py b/tests/providers/aws/services/firehose/firehose_service_test.py
new file mode 100644
index 00000000000..46d6c8c5226
--- /dev/null
+++ b/tests/providers/aws/services/firehose/firehose_service_test.py
@@ -0,0 +1,69 @@
+from boto3 import client
+from moto import mock_aws
+
+from prowler.providers.aws.services.firehose.firehose_service import Firehose
+from tests.providers.aws.utils import AWS_REGION_EU_WEST_1, set_mocked_aws_provider
+
+
+class Test_Firehose_Service:
+    # Test Firehose Service
+    @mock_aws
+    def test_service(self):
+        # Firehose client for this test class
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+        firehose = Firehose(aws_provider)
+        assert firehose.service == "firehose"
+
+    # Test Firehose Client
+    @mock_aws
+    def test_client(self):
+        # Firehose client for this test class
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+        firehose = Firehose(aws_provider)
+        for regional_client in firehose.regional_clients.values():
+            assert regional_client.__class__.__name__ == "Firehose"
+
+    # Test Firehose Session
+    @mock_aws
+    def test__get_session__(self):
+        # Firehose client for this test class
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+        firehose = Firehose(aws_provider)
+        assert firehose.session.__class__.__name__ == "Session"
+
+    # Test Firehose List Delivery Streams
+    @mock_aws
+    def test_list_delivery_streams(self):
+        # Generate S3 client
+        s3_client = client("s3", region_name=AWS_REGION_EU_WEST_1)
+        s3_client.create_bucket(
+            Bucket="test-bucket",
+            CreateBucketConfiguration={"LocationConstraint": AWS_REGION_EU_WEST_1},
+        )
+
+        # Generate Firehose client
+        firehose_client = client("firehose", region_name=AWS_REGION_EU_WEST_1)
+        delivery_stream = firehose_client.create_delivery_stream(
+            DeliveryStreamName="test-delivery-stream",
+            DeliveryStreamType="DirectPut",
+            S3DestinationConfiguration={
+                "RoleARN": "arn:aws:iam::012345678901:role/firehose-role",
+                "BucketARN": "arn:aws:s3:::test-bucket",
+                "Prefix": "",
+                "BufferingHints": {"IntervalInSeconds": 300, "SizeInMBs": 5},
+                "CompressionFormat": "UNCOMPRESSED",
+            },
+            Tags=[{"Key": "key", "Value": "value"}],
+        )
+        arn = delivery_stream["DeliveryStreamARN"]
+        delivery_stream_name = arn.split("/")[-1]
+
+        # Firehose Client for this test class
+        aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
+        firehose = Firehose(aws_provider)
+
+        assert len(firehose.delivery_streams) == 1
+        assert firehose.delivery_streams[arn].arn == arn
+        assert firehose.delivery_streams[arn].name == delivery_stream_name
+        assert firehose.delivery_streams[arn].region == AWS_REGION_EU_WEST_1
+        assert firehose.delivery_streams[arn].tags == [{"Key": "key", "Value": "value"}]