diff --git a/airbyte_cdk/sources/declarative/parsers/manifest_component_transformer.py b/airbyte_cdk/sources/declarative/parsers/manifest_component_transformer.py
index 00b1a18ffb..5fe3e94f58 100644
--- a/airbyte_cdk/sources/declarative/parsers/manifest_component_transformer.py
+++ b/airbyte_cdk/sources/declarative/parsers/manifest_component_transformer.py
@@ -67,6 +67,14 @@
     "DynamicSchemaLoader.retriever": "SimpleRetriever",
     # SchemaTypeIdentifier
     "SchemaTypeIdentifier.types_map": "TypesMap",
+    # StreamGroup
+    "StreamGroup.streams": "DeclarativeStream",
+}
+
+# Mapping for fields that use the JSON Schema additionalProperties pattern (Dict[str, ComponentType]).
+# The transformer uses this to identify dict values as typed components and recurse into each.
+ADDITIONAL_PROPERTIES_TYPES: Mapping[str, str] = {
+    "DeclarativeSource.stream_groups": "StreamGroup",
 }
 
 # We retain a separate registry for custom components to automatically insert the type if it is missing. This is intended to
@@ -159,12 +167,30 @@ def propagate_types_and_parameters(
                 # We exclude propagating a parameter that matches the current field name because that would result in an infinite cycle
                 excluded_parameter = current_parameters.pop(field_name, None)
                 parent_type_field_identifier = f"{propagated_component.get('type')}.{field_name}"
-                propagated_component[field_name] = self.propagate_types_and_parameters(
-                    parent_type_field_identifier,
-                    field_value,
-                    current_parameters,
-                    use_parent_parameters=use_parent_parameters,
+                additional_props_type = ADDITIONAL_PROPERTIES_TYPES.get(
+                    parent_type_field_identifier
                 )
+                if additional_props_type:
+                    # Handle additionalProperties pattern: field_value is a dict of
+                    # component dicts keyed by arbitrary names.
+                    for key, value in field_value.items():
+                        if isinstance(value, dict):
+                            if "type" not in value:
+                                value["type"] = additional_props_type
+                            field_value[key] = self.propagate_types_and_parameters(
+                                parent_type_field_identifier,
+                                value,
+                                current_parameters,
+                                use_parent_parameters=use_parent_parameters,
+                            )
+                    propagated_component[field_name] = field_value
+                else:
+                    propagated_component[field_name] = self.propagate_types_and_parameters(
+                        parent_type_field_identifier,
+                        field_value,
+                        current_parameters,
+                        use_parent_parameters=use_parent_parameters,
+                    )
                 if excluded_parameter:
                     current_parameters[field_name] = excluded_parameter
             elif isinstance(field_value, typing.List):
diff --git a/unit_tests/sources/declarative/parsers/test_manifest_component_transformer.py b/unit_tests/sources/declarative/parsers/test_manifest_component_transformer.py
index c741fa94a1..ee152c688e 100644
--- a/unit_tests/sources/declarative/parsers/test_manifest_component_transformer.py
+++ b/unit_tests/sources/declarative/parsers/test_manifest_component_transformer.py
@@ -743,3 +743,130 @@ def test_dynamic_stream_use_parent_parameters_configuration():
     assert result_false["retriever"]["name"] == "retriever_name"
     # When use_parent_parameters=True, parent's $parameters win
     assert result_true["retriever"]["name"] == "parent_name"
+
+
+def test_stream_groups_propagate_types_and_parameters():
+    """stream_groups uses additionalProperties (Dict[str, StreamGroup]).
+
+    The transformer must recurse into each group value, assign
+    type=StreamGroup, then propagate types and $parameters into the
+    nested DeclarativeStream items.
+    """
+    component = {
+        "type": "DeclarativeSource",
+        "streams": [],
+        "stream_groups": {
+            "group_a": {
+                "streams": [
+                    {
+                        "$parameters": {"name": "users", "primary_key": "id"},
+                        "retriever": {
+                            "type": "SimpleRetriever",
+                            "record_selector": {
+                                "type": "RecordSelector",
+                                "extractor": {"type": "DpathExtractor", "field_path": []},
+                            },
+                            "requester": {
+                                "type": "HttpRequester",
+                                "url_base": "https://api.example.com",
+                                "path": "/users",
+                            },
+                        },
+                    },
+                ],
+                "action": {"type": "BlockSimultaneousSyncsAction"},
+            },
+        },
+    }
+
+    transformer = ManifestComponentTransformer()
+    result = transformer.propagate_types_and_parameters("", component, {})
+
+    group = result["stream_groups"]["group_a"]
+    # StreamGroup type is assigned
+    assert group["type"] == "StreamGroup"
+    # The action component is preserved
+    assert group["action"]["type"] == "BlockSimultaneousSyncsAction"
+
+    stream = group["streams"][0]
+    # DeclarativeStream type is inferred via DEFAULT_MODEL_TYPES
+    assert stream["type"] == "DeclarativeStream"
+    # $parameters are propagated into the stream and its children
+    assert stream["name"] == "users"
+    assert stream["primary_key"] == "id"
+    assert stream["retriever"]["name"] == "users"
+    assert stream["retriever"]["primary_key"] == "id"
+    assert stream["retriever"]["requester"]["name"] == "users"
+    assert stream["retriever"]["requester"]["primary_key"] == "id"
+    assert stream["retriever"]["requester"]["$parameters"] == {
+        "name": "users",
+        "primary_key": "id",
+    }
+
+
+def test_stream_groups_with_multiple_groups():
+    """Multiple groups each get type=StreamGroup and proper recursion."""
+    component = {
+        "type": "DeclarativeSource",
+        "streams": [],
+        "stream_groups": {
+            "api_v1": {
+                "streams": [
+                    {
+                        "$parameters": {"name": "posts"},
+                        "retriever": {"type": "SimpleRetriever"},
+                    },
+                ],
+                "action": {"type": "BlockSimultaneousSyncsAction"},
+            },
+            "api_v2": {
+                "streams": [
+                    {
+                        "$parameters": {"name": "comments"},
+                        "retriever": {"type": "SimpleRetriever"},
+                    },
+                ],
+                "action": {"type": "BlockSimultaneousSyncsAction"},
+            },
+        },
+    }
+
+    transformer = ManifestComponentTransformer()
+    result = transformer.propagate_types_and_parameters("", component, {})
+
+    for group_name in ("api_v1", "api_v2"):
+        group = result["stream_groups"][group_name]
+        assert group["type"] == "StreamGroup"
+
+    assert result["stream_groups"]["api_v1"]["streams"][0]["type"] == "DeclarativeStream"
+    assert result["stream_groups"]["api_v1"]["streams"][0]["name"] == "posts"
+
+    assert result["stream_groups"]["api_v2"]["streams"][0]["type"] == "DeclarativeStream"
+    assert result["stream_groups"]["api_v2"]["streams"][0]["name"] == "comments"
+
+
+def test_stream_groups_preserves_explicit_types():
+    """When a stream inside a group already has an explicit type, it is not overwritten."""
+    component = {
+        "type": "DeclarativeSource",
+        "streams": [],
+        "stream_groups": {
+            "my_group": {
+                "type": "StreamGroup",
+                "streams": [
+                    {
+                        "type": "DeclarativeStream",
+                        "retriever": {"type": "SimpleRetriever"},
+                    },
+                ],
+                "action": {"type": "BlockSimultaneousSyncsAction"},
+            },
+        },
+    }
+
+    transformer = ManifestComponentTransformer()
+    result = transformer.propagate_types_and_parameters("", component, {})
+
+    group = result["stream_groups"]["my_group"]
+    assert group["type"] == "StreamGroup"
+    assert group["streams"][0]["type"] == "DeclarativeStream"