Feat/improve yaml extraction (#2428)

* Support wildcard handling in `emit()`

Change `emit()` to call handlers registered for parent classes using
`isinstance()`. Ensures that base event handlers receive derived
events.

* Fix failing test

* Remove unused variable

* update interpolation to work with example response types in yaml docs

* make tests

* fix circular deps

* Fixing interpolation imports

* Improve test

---------

Co-authored-by: Vinicius Brasil <vini@hey.com>
Co-authored-by: Lorenze Jay <63378463+lorenzejay@users.noreply.github.com>

tests/task_test.py

@@ -15,6 +15,7 @@ from crewai import Agent, Crew, Process, Task
 from crewai.tasks.conditional_task import ConditionalTask
 from crewai.tasks.task_output import TaskOutput
 from crewai.utilities.converter import Converter
+from crewai.utilities.string_utils import interpolate_only
 
 
 def test_task_tool_reflect_agent_tools():
@@ -822,7 +823,7 @@ def test_interpolate_only():
 
     # Test JSON structure preservation
     json_string = '{"info": "Look at {placeholder}", "nested": {"val": "{nestedVal}"}}'
-    result = task.interpolate_only(
+    result = interpolate_only(
         input_string=json_string,
         inputs={"placeholder": "the data", "nestedVal": "something else"},
     )
@@ -833,20 +834,18 @@ def test_interpolate_only():
 
     # Test normal string interpolation
     normal_string = "Hello {name}, welcome to {place}!"
-    result = task.interpolate_only(
+    result = interpolate_only(
         input_string=normal_string, inputs={"name": "John", "place": "CrewAI"}
     )
     assert result == "Hello John, welcome to CrewAI!"
 
     # Test empty string
-    result = task.interpolate_only(input_string="", inputs={"unused": "value"})
+    result = interpolate_only(input_string="", inputs={"unused": "value"})
     assert result == ""
 
     # Test string with no placeholders
     no_placeholders = "Hello, this is a test"
-    result = task.interpolate_only(
-        input_string=no_placeholders, inputs={"unused": "value"}
-    )
+    result = interpolate_only(input_string=no_placeholders, inputs={"unused": "value"})
     assert result == no_placeholders
 
 
@@ -858,7 +857,7 @@ def test_interpolate_only_with_dict_inside_expected_output():
     )
 
     json_string = '{"questions": {"main_question": "What is the user\'s name?", "secondary_question": "What is the user\'s age?"}}'
-    result = task.interpolate_only(
+    result = interpolate_only(
         input_string=json_string,
         inputs={
             "questions": {
@@ -872,18 +871,16 @@ def test_interpolate_only_with_dict_inside_expected_output():
     assert result == json_string
 
     normal_string = "Hello {name}, welcome to {place}!"
-    result = task.interpolate_only(
+    result = interpolate_only(
         input_string=normal_string, inputs={"name": "John", "place": "CrewAI"}
     )
     assert result == "Hello John, welcome to CrewAI!"
 
-    result = task.interpolate_only(input_string="", inputs={"unused": "value"})
+    result = interpolate_only(input_string="", inputs={"unused": "value"})
     assert result == ""
 
     no_placeholders = "Hello, this is a test"
-    result = task.interpolate_only(
-        input_string=no_placeholders, inputs={"unused": "value"}
-    )
+    result = interpolate_only(input_string=no_placeholders, inputs={"unused": "value"})
     assert result == no_placeholders
 
 
@@ -1085,12 +1082,12 @@ def test_interpolate_with_list_of_strings():
     # Test simple list of strings
     input_str = "Available items: {items}"
     inputs = {"items": ["apple", "banana", "cherry"]}
-    result = task.interpolate_only(input_str, inputs)
+    result = interpolate_only(input_str, inputs)
     assert result == f"Available items: {inputs['items']}"
 
     # Test empty list
     empty_list_input = {"items": []}
-    result = task.interpolate_only(input_str, empty_list_input)
+    result = interpolate_only(input_str, empty_list_input)
     assert result == "Available items: []"
 
 
@@ -1106,7 +1103,7 @@ def test_interpolate_with_list_of_dicts():
             {"name": "Bob", "age": 25, "skills": ["Java", "Cloud"]},
         ]
     }
-    result = task.interpolate_only("{people}", input_data)
+    result = interpolate_only("{people}", input_data)
 
     parsed_result = eval(result)
     assert isinstance(parsed_result, list)
@@ -1138,7 +1135,7 @@ def test_interpolate_with_nested_structures():
             ],
         }
     }
-    result = task.interpolate_only("{company}", input_data)
+    result = interpolate_only("{company}", input_data)
     parsed = eval(result)
 
     assert parsed["name"] == "TechCorp"
@@ -1161,7 +1158,7 @@ def test_interpolate_with_special_characters():
             "empty": "",
         }
     }
-    result = task.interpolate_only("{special_data}", input_data)
+    result = interpolate_only("{special_data}", input_data)
     parsed = eval(result)
 
     assert parsed["quotes"] == """This has "double" and 'single' quotes"""
@@ -1188,7 +1185,7 @@ def test_interpolate_mixed_types():
             },
         }
     }
-    result = task.interpolate_only("{data}", input_data)
+    result = interpolate_only("{data}", input_data)
     parsed = eval(result)
 
     assert parsed["name"] == "Test Dataset"
@@ -1216,7 +1213,7 @@ def test_interpolate_complex_combination():
             },
         ]
     }
-    result = task.interpolate_only("{report}", input_data)
+    result = interpolate_only("{report}", input_data)
     parsed = eval(result)
 
     assert len(parsed) == 2
@@ -1233,7 +1230,7 @@ def test_interpolate_invalid_type_validation():
 
     # Test with invalid top-level type
     with pytest.raises(ValueError) as excinfo:
-        task.interpolate_only("{data}", {"data": set()})  # type: ignore we are purposely testing this failure
+        interpolate_only("{data}", {"data": set()})  # type: ignore we are purposely testing this failure
 
     assert "Unsupported type set" in str(excinfo.value)
 
@@ -1246,7 +1243,7 @@ def test_interpolate_invalid_type_validation():
         }
     }
     with pytest.raises(ValueError) as excinfo:
-        task.interpolate_only("{data}", {"data": invalid_nested})
+        interpolate_only("{data}", {"data": invalid_nested})
     assert "Unsupported type set" in str(excinfo.value)
 
 
@@ -1265,24 +1262,22 @@ def test_interpolate_custom_object_validation():
 
     # Test with custom object at top level
     with pytest.raises(ValueError) as excinfo:
-        task.interpolate_only("{obj}", {"obj": CustomObject(5)})  # type: ignore we are purposely testing this failure
+        interpolate_only("{obj}", {"obj": CustomObject(5)})  # type: ignore we are purposely testing this failure
     assert "Unsupported type CustomObject" in str(excinfo.value)
 
     # Test with nested custom object in dictionary
     with pytest.raises(ValueError) as excinfo:
-        task.interpolate_only(
-            "{data}", {"data": {"valid": 1, "invalid": CustomObject(5)}}
-        )
+        interpolate_only("{data}", {"data": {"valid": 1, "invalid": CustomObject(5)}})
     assert "Unsupported type CustomObject" in str(excinfo.value)
 
     # Test with nested custom object in list
     with pytest.raises(ValueError) as excinfo:
-        task.interpolate_only("{data}", {"data": [1, "valid", CustomObject(5)]})
+        interpolate_only("{data}", {"data": [1, "valid", CustomObject(5)]})
     assert "Unsupported type CustomObject" in str(excinfo.value)
 
     # Test with deeply nested custom object
     with pytest.raises(ValueError) as excinfo:
-        task.interpolate_only(
+        interpolate_only(
             "{data}", {"data": {"level1": {"level2": [{"level3": CustomObject(5)}]}}}
         )
     assert "Unsupported type CustomObject" in str(excinfo.value)
@@ -1306,7 +1301,7 @@ def test_interpolate_valid_complex_types():
     }
 
     # Should not raise any errors
-    result = task.interpolate_only("{data}", {"data": valid_data})
+    result = interpolate_only("{data}", {"data": valid_data})
     parsed = eval(result)
     assert parsed["name"] == "Valid Dataset"
     assert parsed["stats"]["nested"]["deeper"]["b"] == 2.5
@@ -1319,16 +1314,16 @@ def test_interpolate_edge_cases():
     )
 
     # Test empty dict and list
-    assert task.interpolate_only("{}", {"data": {}}) == "{}"
-    assert task.interpolate_only("[]", {"data": []}) == "[]"
+    assert interpolate_only("{}", {"data": {}}) == "{}"
+    assert interpolate_only("[]", {"data": []}) == "[]"
 
     # Test numeric types
-    assert task.interpolate_only("{num}", {"num": 42}) == "42"
-    assert task.interpolate_only("{num}", {"num": 3.14}) == "3.14"
+    assert interpolate_only("{num}", {"num": 42}) == "42"
+    assert interpolate_only("{num}", {"num": 3.14}) == "3.14"
 
     # Test boolean values (valid JSON types)
-    assert task.interpolate_only("{flag}", {"flag": True}) == "True"
-    assert task.interpolate_only("{flag}", {"flag": False}) == "False"
+    assert interpolate_only("{flag}", {"flag": True}) == "True"
+    assert interpolate_only("{flag}", {"flag": False}) == "False"
 
 
 def test_interpolate_valid_types():
@@ -1346,7 +1341,7 @@ def test_interpolate_valid_types():
         "nested": {"flag": True, "empty": None},
     }
 
-    result = task.interpolate_only("{data}", {"data": valid_data})
+    result = interpolate_only("{data}", {"data": valid_data})
     parsed = eval(result)
 
     assert parsed["active"] is True

tests/utilities/test_string_utils.py

@@ -0,0 +1,187 @@
+from typing import Any, Dict, List, Union
+
+import pytest
+
+from crewai.utilities.string_utils import interpolate_only
+
+
+class TestInterpolateOnly:
+    """Tests for the interpolate_only function in string_utils.py."""
+
+    def test_basic_variable_interpolation(self):
+        """Test basic variable interpolation works correctly."""
+        template = "Hello, {name}! Welcome to {company}."
+        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
+            "name": "Alice",
+            "company": "CrewAI",
+        }
+
+        result = interpolate_only(template, inputs)
+
+        assert result == "Hello, Alice! Welcome to CrewAI."
+
+    def test_multiple_occurrences_of_same_variable(self):
+        """Test that multiple occurrences of the same variable are replaced."""
+        template = "{name} is using {name}'s account."
+        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
+            "name": "Bob"
+        }
+
+        result = interpolate_only(template, inputs)
+
+        assert result == "Bob is using Bob's account."
+
+    def test_json_structure_preservation(self):
+        """Test that JSON structures are preserved and not interpolated incorrectly."""
+        template = """
+        Instructions for {agent}:
+
+        Please return the following object:
+
+        {"name": "person's name", "age": 25, "skills": ["coding", "testing"]}
+        """
+        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
+            "agent": "DevAgent"
+        }
+
+        result = interpolate_only(template, inputs)
+
+        assert "Instructions for DevAgent:" in result
+        assert (
+            '{"name": "person\'s name", "age": 25, "skills": ["coding", "testing"]}'
+            in result
+        )
+
+    def test_complex_nested_json(self):
+        """Test with complex JSON structures containing curly braces."""
+        template = """
+        {agent} needs to process:
+        {
+          "config": {
+            "nested": {
+              "value": 42
+            },
+            "arrays": [1, 2, {"inner": "value"}]
+          }
+        }
+        """
+        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
+            "agent": "DataProcessor"
+        }
+
+        result = interpolate_only(template, inputs)
+
+        assert "DataProcessor needs to process:" in result
+        assert '"nested": {' in result
+        assert '"value": 42' in result
+        assert '[1, 2, {"inner": "value"}]' in result
+
+    def test_missing_variable(self):
+        """Test that an error is raised when a required variable is missing."""
+        template = "Hello, {name}!"
+        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
+            "not_name": "Alice"
+        }
+
+        with pytest.raises(KeyError) as excinfo:
+            interpolate_only(template, inputs)
+
+        assert "template variable" in str(excinfo.value).lower()
+        assert "name" in str(excinfo.value)
+
+    def test_invalid_input_types(self):
+        """Test that an error is raised with invalid input types."""
+        template = "Hello, {name}!"
+        # Using Any for this test since we're intentionally testing an invalid type
+        inputs: Dict[str, Any] = {"name": object()}  # Object is not a valid input type
+
+        with pytest.raises(ValueError) as excinfo:
+            interpolate_only(template, inputs)
+
+        assert "unsupported type" in str(excinfo.value).lower()
+
+    def test_empty_input_string(self):
+        """Test handling of empty or None input string."""
+        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
+            "name": "Alice"
+        }
+
+        assert interpolate_only("", inputs) == ""
+        assert interpolate_only(None, inputs) == ""
+
+    def test_no_variables_in_template(self):
+        """Test a template with no variables to replace."""
+        template = "This is a static string with no variables."
+        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
+            "name": "Alice"
+        }
+
+        result = interpolate_only(template, inputs)
+
+        assert result == template
+
+    def test_variable_name_starting_with_underscore(self):
+        """Test variables starting with underscore are replaced correctly."""
+        template = "Variable: {_special_var}"
+        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
+            "_special_var": "Special Value"
+        }
+
+        result = interpolate_only(template, inputs)
+
+        assert result == "Variable: Special Value"
+
+    def test_preserves_non_matching_braces(self):
+        """Test that non-matching braces patterns are preserved."""
+        template = (
+            "This {123} and {!var} should not be replaced but {valid_var} should."
+        )
+        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
+            "valid_var": "works"
+        }
+
+        result = interpolate_only(template, inputs)
+
+        assert (
+            result == "This {123} and {!var} should not be replaced but works should."
+        )
+
+    def test_complex_mixed_scenario(self):
+        """Test a complex scenario with both valid variables and JSON structures."""
+        template = """
+        {agent_name} is working on task {task_id}.
+        
+        Instructions:
+        1. Process the data
+        2. Return results as:
+        
+        {
+          "taskId": "{task_id}",
+          "results": {
+            "processed_by": "agent_name",
+            "status": "complete",
+            "values": [1, 2, 3]
+          }
+        }
+        """
+        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]] = {
+            "agent_name": "AnalyticsAgent",
+            "task_id": "T-12345",
+        }
+
+        result = interpolate_only(template, inputs)
+
+        assert "AnalyticsAgent is working on task T-12345" in result
+        assert '"taskId": "T-12345"' in result
+        assert '"processed_by": "agent_name"' in result  # This shouldn't be replaced
+        assert '"values": [1, 2, 3]' in result
+
+    def test_empty_inputs_dictionary(self):
+        """Test that an error is raised with empty inputs dictionary."""
+        template = "Hello, {name}!"
+        inputs: Dict[str, Any] = {}
+
+        with pytest.raises(ValueError) as excinfo:
+            interpolate_only(template, inputs)
+
+        assert "inputs dictionary cannot be empty" in str(excinfo.value).lower()