fix(declarative): stop coercing static scalar strings in request_body_json

airbyte_cdk/sources/declarative/interpolation/interpolated_nested_mapping.py

@@ -39,6 +39,13 @@ def _eval(
     ) -> Any:
         # Recursively interpolate dictionaries and lists
         if isinstance(value, str):
+            if "{{" not in value and "{%" not in value:
+                # Preserve structured static literals for request bodies, but avoid coercing scalar
+                # values such as "478" or "None" through literal evaluation.
+                evaluated = self._interpolation.eval(
+                    value, config, parameters=self._parameters, **kwargs
+                )
+                return evaluated if isinstance(evaluated, (dict, list)) else value
             return self._interpolation.eval(value, config, parameters=self._parameters, **kwargs)
         elif isinstance(value, dict):
             interpolated_dict = {

unit_tests/sources/declarative/interpolation/test_interpolated_nested_mapping.py

@@ -52,3 +52,76 @@ def test(test_name, path, expected_value):
     interpolated = mapping.eval(config, **{"kwargs": kwargs})
 
     assert dpath.get(interpolated, path) == expected_value
+
+
+@pytest.mark.parametrize(
+    "mapping, path, expected_value",
+    [
+        pytest.param(
+            {"clientId": "478"},
+            "clientId",
+            "478",
+            id="static_digit_only_string",
+        ),
+        pytest.param(
+            {"value": "0012"},
+            "value",
+            "0012",
+            id="static_leading_zero_string",
+        ),
+        pytest.param(
+            {"value": "None"},
+            "value",
+            "None",
+            id="static_none_string",
+        ),
+        pytest.param(
+            {"value": '{"field": "updated_at", "order": "ascending"}'},
+            "value",
+            {"field": "updated_at", "order": "ascending"},
+            id="static_structured_dict",
+        ),
+        pytest.param(
+            {"value": "[1, 2, 3]"},
+            "value",
+            [1, 2, 3],
+            id="static_structured_list",
+        ),
+        pytest.param(
+            {"sort": '{"field": "updated_at", "order": "ascending"}'},
+            "sort",
+            {"field": "updated_at", "order": "ascending"},
+            id="static_intercom_sort_json",
+        ),
+        pytest.param(
+            {"478": "value"},
+            "478",
+            "value",
+            id="static_digit_only_key",
+        ),
+        pytest.param(
+            {"nested": {"clientId": "478"}},
+            "nested/clientId",
+            "478",
+            id="static_string_nested_in_dict",
+        ),
+        pytest.param(
+            {"nested": {"values": ["478", "0012", "None"]}},
+            "nested/values",
+            ["478", "0012", "None"],
+            id="static_strings_nested_in_list",
+        ),
+        pytest.param(
+            {"value": "{{ 1 + 1 }}"},
+            "value",
+            2,
+            id="jinja_value_still_coerces",
+        ),
+    ],
+)
+def test_static_strings_are_preserved_and_jinja_values_are_interpolated(
+    mapping, path, expected_value
+):
+    interpolated = InterpolatedNestedMapping(mapping=mapping, parameters={}).eval({})
+
+    assert dpath.get(interpolated, path) == expected_value

unit_tests/sources/declarative/requesters/request_options/test_interpolated_request_options_provider.py

@@ -141,6 +141,15 @@ def test_interpolated_request_json(test_name, input_request_json, expected_reque
 @pytest.mark.parametrize(
     "test_name, input_request_json, expected_request_json",
     [
+        pytest.param(
+            "test_static_digit_only_string",
+            RequestBodyJsonObject(
+                type="RequestBodyJsonObject",
+                value={"clientId": "478"},
+            ),
+            {"clientId": "478"},
+            id="static_digit_only_string",
+        ),
         (
             "test_static_json",
             RequestBodyJsonObject(