diff --git a/airbyte_cdk/sources/declarative/manifest_declarative_source.py b/airbyte_cdk/sources/declarative/manifest_declarative_source.py
index 78aeac23f..efc779464 100644
--- a/airbyte_cdk/sources/declarative/manifest_declarative_source.py
+++ b/airbyte_cdk/sources/declarative/manifest_declarative_source.py
@@ -365,6 +365,11 @@ def _dynamic_stream_configs(
                 # Ensure that each stream is created with a unique name
                 name = dynamic_stream.get("name")
 
+                if not isinstance(name, str):
+                    raise ValueError(
+                        f"Expected stream name {name} to be a string, got {type(name)}."
+                    )
+
                 if name in seen_dynamic_streams:
                     error_message = f"Dynamic streams list contains a duplicate name: {name}. Please contact Airbyte Support."
                     failure_type = FailureType.system_error
diff --git a/unit_tests/sources/declarative/resolvers/test_http_components_resolver.py b/unit_tests/sources/declarative/resolvers/test_http_components_resolver.py
index f09ede0d6..357dcceef 100644
--- a/unit_tests/sources/declarative/resolvers/test_http_components_resolver.py
+++ b/unit_tests/sources/declarative/resolvers/test_http_components_resolver.py
@@ -3,6 +3,7 @@
 #
 
 import json
+from copy import deepcopy
 from unittest.mock import MagicMock
 
 import pytest
@@ -362,6 +363,34 @@ def test_http_components_resolver(
     assert result == expected_result
 
 
+def test_wrong_stream_name_type():
+    with HttpMocker() as http_mocker:
+        http_mocker.get(
+            HttpRequest(url="https://api.test.com/int_items"),
+            HttpResponse(
+                body=json.dumps(
+                    [
+                        {"id": 1, "name": 1},
+                        {"id": 2, "name": 2},
+                    ]
+                )
+            ),
+        )
+
+        manifest = deepcopy(_MANIFEST)
+        manifest["dynamic_streams"][0]["components_resolver"]["retriever"]["requester"]["path"] = (
+            "int_items"
+        )
+
+        source = ConcurrentDeclarativeSource(
+            source_config=manifest, config=_CONFIG, catalog=None, state=None
+        )
+        with pytest.raises(ValueError) as exc_info:
+            source.discover(logger=source.logger, config=_CONFIG)
+
+        assert str(exc_info.value) == "Expected stream name 1 to be a string, got <class 'int'>."
+
+
 @pytest.mark.parametrize(
     "components_mapping, retriever_data, stream_template_config, expected_result",
     [