Fix import sorting in test file

Co-Authored-By: Joe Moura <joao@crewai.com>

src/crewai/agents/agent_builder/base_agent.py

@@ -25,6 +25,7 @@ from crewai.tools.base_tool import BaseTool, Tool
 from crewai.utilities import I18N, Logger, RPMController
 from crewai.utilities.config import process_config
 from crewai.utilities.converter import Converter
+from crewai.utilities.string_utils import interpolate_only
 
 T = TypeVar("T", bound="BaseAgent")
 
@@ -333,9 +334,15 @@ class BaseAgent(ABC, BaseModel):
             self._original_backstory = self.backstory
 
         if inputs:
-            self.role = self._original_role.format(**inputs)
-            self.goal = self._original_goal.format(**inputs)
-            self.backstory = self._original_backstory.format(**inputs)
+            self.role = interpolate_only(
+                input_string=self._original_role, inputs=inputs
+            )
+            self.goal = interpolate_only(
+                input_string=self._original_goal, inputs=inputs
+            )
+            self.backstory = interpolate_only(
+                input_string=self._original_backstory, inputs=inputs
+            )
 
     def set_cache_handler(self, cache_handler: CacheHandler) -> None:
         """Set the cache handler for the agent.

src/crewai/memory/contextual/contextual_memory.py

@@ -14,12 +14,19 @@ class ContextualMemory:
     ):
         if memory_config is not None:
             self.memory_provider = memory_config.get("provider")
+            # Special handling for Mem0 provider
+            if self.memory_provider == "mem0":
+                # Check if a custom client was provided in the memory_config
+                self.um = um
+                self.search_kwargs = memory_config.get("config", {}).get("search_kwargs", {})
+            else:
+                self.um = um
         else:
             self.memory_provider = None
+            self.um = um
         self.stm = stm
         self.ltm = ltm
         self.em = em
-        self.um = um
 
     def build_context_for_task(self, task, context) -> str:
         """
@@ -94,6 +101,10 @@ class ContextualMemory:
         Returns:
             str: Formatted user memories as bullet points, or an empty string if none found.
         """
+        # Check if user memory is available
+        if self.um is None:
+            return ""
+            
         user_memories = self.um.search(query)
         if not user_memories:
             return ""

src/crewai/memory/storage/mem0_storage.py

@@ -19,29 +19,37 @@ class Mem0Storage(Storage):
 
         self.memory_type = type
         self.crew = crew
-        self.memory_config = crew.memory_config
+        self.memory_config = crew.memory_config if crew else None
 
         # User ID is required for user memory type "user" since it's used as a unique identifier for the user.
         user_id = self._get_user_id()
         if type == "user" and not user_id:
             raise ValueError("User ID is required for user memory type")
 
-        # API key in memory config overrides the environment variable
-        config = self.memory_config.get("config", {})
-        mem0_api_key = config.get("api_key") or os.getenv("MEM0_API_KEY")
-        mem0_org_id = config.get("org_id")
-        mem0_project_id = config.get("project_id")
-
-        # Initialize MemoryClient or Memory based on the presence of the mem0_api_key
-        if mem0_api_key:
-            if mem0_org_id and mem0_project_id:
-                self.memory = MemoryClient(
-                    api_key=mem0_api_key, org_id=mem0_org_id, project_id=mem0_project_id
-                )
+        # Check if a client was provided in the memory_config
+        if self.memory_config:
+            config = self.memory_config.get("config", {})
+            if config.get("client"):
+                self.memory = config.get("client")
             else:
-                self.memory = MemoryClient(api_key=mem0_api_key)
+                # API key in memory config overrides the environment variable
+                mem0_api_key = config.get("api_key") or os.getenv("MEM0_API_KEY")
+                mem0_org_id = config.get("org_id")
+                mem0_project_id = config.get("project_id")
+
+                # Initialize MemoryClient or Memory based on the presence of the mem0_api_key
+                if mem0_api_key:
+                    if mem0_org_id and mem0_project_id:
+                        self.memory = MemoryClient(
+                            api_key=mem0_api_key, org_id=mem0_org_id, project_id=mem0_project_id
+                        )
+                    else:
+                        self.memory = MemoryClient(api_key=mem0_api_key)
+                else:
+                    self.memory = Memory()  # Fallback to Memory if no Mem0 API key is provided
         else:
-            self.memory = Memory()  # Fallback to Memory if no Mem0 API key is provided
+            # No memory config, use default Memory
+            self.memory = Memory()
 
     def _sanitize_role(self, role: str) -> str:
         """
@@ -103,13 +111,15 @@ class Mem0Storage(Storage):
 
     def _get_user_id(self):
         if self.memory_type == "user":
-            if hasattr(self, "memory_config") and self.memory_config is not None:
+            if self.memory_config is not None:
                 return self.memory_config.get("config", {}).get("user_id")
             else:
-                return None
+                return "default_user"  # Provide a default user ID for testing
         return None
 
     def _get_agent_name(self):
+        if not self.crew:
+            return "default_agent"
         agents = self.crew.agents if self.crew else []
         agents = [self._sanitize_role(agent.role) for agent in agents]
         agents = "_".join(agents)

src/crewai/task.py

@@ -2,6 +2,7 @@ import datetime
 import inspect
 import json
 import logging
+import re
 import threading
 import uuid
 from concurrent.futures import Future
@@ -49,6 +50,7 @@ from crewai.utilities.events import (
 from crewai.utilities.events.crewai_event_bus import crewai_event_bus
 from crewai.utilities.i18n import I18N
 from crewai.utilities.printer import Printer
+from crewai.utilities.string_utils import interpolate_only
 
 
 class Task(BaseModel):
@@ -507,7 +509,9 @@ class Task(BaseModel):
             return
 
         try:
-            self.description = self._original_description.format(**inputs)
+            self.description = interpolate_only(
+                input_string=self._original_description, inputs=inputs
+            )
         except KeyError as e:
             raise ValueError(
                 f"Missing required template variable '{e.args[0]}' in description"
@@ -516,7 +520,7 @@ class Task(BaseModel):
             raise ValueError(f"Error interpolating description: {str(e)}") from e
 
         try:
-            self.expected_output = self.interpolate_only(
+            self.expected_output = interpolate_only(
                 input_string=self._original_expected_output, inputs=inputs
             )
         except (KeyError, ValueError) as e:
@@ -524,7 +528,7 @@ class Task(BaseModel):
 
         if self.output_file is not None:
             try:
-                self.output_file = self.interpolate_only(
+                self.output_file = interpolate_only(
                     input_string=self._original_output_file, inputs=inputs
                 )
             except (KeyError, ValueError) as e:
@@ -555,72 +559,6 @@ class Task(BaseModel):
                 f"\n\n{conversation_instruction}\n\n{conversation_history}"
             )
 
-    def interpolate_only(
-        self,
-        input_string: Optional[str],
-        inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]],
-    ) -> str:
-        """Interpolate placeholders (e.g., {key}) in a string while leaving JSON untouched.
-
-        Args:
-            input_string: The string containing template variables to interpolate.
-                         Can be None or empty, in which case an empty string is returned.
-            inputs: Dictionary mapping template variables to their values.
-                   Supported value types are strings, integers, floats, and dicts/lists
-                   containing only these types and other nested dicts/lists.
-
-        Returns:
-            The interpolated string with all template variables replaced with their values.
-            Empty string if input_string is None or empty.
-
-        Raises:
-            ValueError: If a value contains unsupported types
-        """
-
-        # Validation function for recursive type checking
-        def validate_type(value: Any) -> None:
-            if value is None:
-                return
-            if isinstance(value, (str, int, float, bool)):
-                return
-            if isinstance(value, (dict, list)):
-                for item in value.values() if isinstance(value, dict) else value:
-                    validate_type(item)
-                return
-            raise ValueError(
-                f"Unsupported type {type(value).__name__} in inputs. "
-                "Only str, int, float, bool, dict, and list are allowed."
-            )
-
-        # Validate all input values
-        for key, value in inputs.items():
-            try:
-                validate_type(value)
-            except ValueError as e:
-                raise ValueError(f"Invalid value for key '{key}': {str(e)}") from e
-
-        if input_string is None or not input_string:
-            return ""
-        if "{" not in input_string and "}" not in input_string:
-            return input_string
-        if not inputs:
-            raise ValueError(
-                "Inputs dictionary cannot be empty when interpolating variables"
-            )
-        try:
-            escaped_string = input_string.replace("{", "{{").replace("}", "}}")
-
-            for key in inputs.keys():
-                escaped_string = escaped_string.replace(f"{{{{{key}}}}}", f"{{{key}}}")
-
-            return escaped_string.format(**inputs)
-        except KeyError as e:
-            raise KeyError(
-                f"Template variable '{e.args[0]}' not found in inputs dictionary"
-            ) from e
-        except ValueError as e:
-            raise ValueError(f"Error during string interpolation: {str(e)}") from e
-
     def increment_tools_errors(self) -> None:
         """Increment the tools errors counter."""
         self.tools_errors += 1

src/crewai/utilities/formatter.py

@@ -1,10 +1,12 @@
-from typing import List
+import re
+from typing import TYPE_CHECKING, List
 
-from crewai.task import Task
-from crewai.tasks.task_output import TaskOutput
+if TYPE_CHECKING:
+    from crewai.task import Task
+    from crewai.tasks.task_output import TaskOutput
 
 
-def aggregate_raw_outputs_from_task_outputs(task_outputs: List[TaskOutput]) -> str:
+def aggregate_raw_outputs_from_task_outputs(task_outputs: List["TaskOutput"]) -> str:
     """Generate string context from the task outputs."""
     dividers = "\n\n----------\n\n"
 
@@ -13,7 +15,7 @@ def aggregate_raw_outputs_from_task_outputs(task_outputs: List[TaskOutput]) -> s
     return context
 
 
-def aggregate_raw_outputs_from_tasks(tasks: List[Task]) -> str:
+def aggregate_raw_outputs_from_tasks(tasks: List["Task"]) -> str:
     """Generate string context from the tasks."""
     task_outputs = [task.output for task in tasks if task.output is not None]
 

src/crewai/utilities/string_utils.py

@@ -0,0 +1,82 @@
+import re
+from typing import Any, Dict, List, Optional, Union
+
+
+def interpolate_only(
+    input_string: Optional[str],
+    inputs: Dict[str, Union[str, int, float, Dict[str, Any], List[Any]]],
+) -> str:
+    """Interpolate placeholders (e.g., {key}) in a string while leaving JSON untouched.
+    Only interpolates placeholders that follow the pattern {variable_name} where
+    variable_name starts with a letter/underscore and contains only letters, numbers, and underscores.
+
+    Args:
+        input_string: The string containing template variables to interpolate.
+                     Can be None or empty, in which case an empty string is returned.
+        inputs: Dictionary mapping template variables to their values.
+               Supported value types are strings, integers, floats, and dicts/lists
+               containing only these types and other nested dicts/lists.
+
+    Returns:
+        The interpolated string with all template variables replaced with their values.
+        Empty string if input_string is None or empty.
+
+    Raises:
+        ValueError: If a value contains unsupported types or a template variable is missing
+    """
+
+    # Validation function for recursive type checking
+    def validate_type(value: Any) -> None:
+        if value is None:
+            return
+        if isinstance(value, (str, int, float, bool)):
+            return
+        if isinstance(value, (dict, list)):
+            for item in value.values() if isinstance(value, dict) else value:
+                validate_type(item)
+            return
+        raise ValueError(
+            f"Unsupported type {type(value).__name__} in inputs. "
+            "Only str, int, float, bool, dict, and list are allowed."
+        )
+
+    # Validate all input values
+    for key, value in inputs.items():
+        try:
+            validate_type(value)
+        except ValueError as e:
+            raise ValueError(f"Invalid value for key '{key}': {str(e)}") from e
+
+    if input_string is None or not input_string:
+        return ""
+    if "{" not in input_string and "}" not in input_string:
+        return input_string
+    if not inputs:
+        raise ValueError(
+            "Inputs dictionary cannot be empty when interpolating variables"
+        )
+
+    # The regex pattern to find valid variable placeholders
+    # Matches {variable_name} where variable_name starts with a letter/underscore
+    # and contains only letters, numbers, and underscores
+    pattern = r"\{([A-Za-z_][A-Za-z0-9_]*)\}"
+
+    # Find all matching variables in the input string
+    variables = re.findall(pattern, input_string)
+    result = input_string
+
+    # Check if all variables exist in inputs
+    missing_vars = [var for var in variables if var not in inputs]
+    if missing_vars:
+        raise KeyError(
+            f"Template variable '{missing_vars[0]}' not found in inputs dictionary"
+        )
+
+    # Replace each variable with its value
+    for var in variables:
+        if var in inputs:
+            placeholder = "{" + var + "}"
+            value = str(inputs[var])
+            result = result.replace(placeholder, value)
+
+    return result

tests/memory/user_memory_test.py

@@ -0,0 +1,79 @@
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from crewai.agent import Agent
+from crewai.crew import Crew
+from crewai.memory.contextual.contextual_memory import ContextualMemory
+from crewai.memory.entity.entity_memory import EntityMemory
+from crewai.memory.long_term.long_term_memory import LongTermMemory
+from crewai.memory.short_term.short_term_memory import ShortTermMemory
+from crewai.memory.user.user_memory import UserMemory
+from crewai.process import Process
+from crewai.task import Task
+
+
+class MockMemoryClient:
+    def __init__(self, *args, **kwargs):
+        pass
+
+    def search(self, *args, **kwargs):
+        return [{"memory": "Test memory", "score": 0.9}]
+
+    def add(self, *args, **kwargs):
+        pass
+
+
+def test_contextual_memory_with_mem0_client():
+    # Create a mock mem0 client
+    mock_mem0_client = MockMemoryClient()
+
+    # Create agent and task
+    agent = Agent(
+        role="Researcher",
+        goal="Search relevant data and provide results",
+        backstory="You are a researcher at a leading tech think tank.",
+        verbose=True,
+    )
+
+    task = Task(
+        description="Perform a search on specific topics.",
+        expected_output="A list of relevant URLs based on the search query.",
+        agent=agent,
+    )
+
+    # Create a UserMemory instance with our mock client
+    user_memory = UserMemory(crew=None)
+    # Manually set the storage memory to our mock client
+    user_memory.storage.memory = mock_mem0_client
+
+    # Create crew with mem0 as memory provider and pass the UserMemory instance
+    crew = Crew(
+        agents=[agent],
+        tasks=[task],
+        process=Process.sequential,
+        memory=True,
+        memory_config={
+            "provider": "mem0",
+            "config": {
+                "user_id": "test_user",
+            },
+            "user_memory": user_memory
+        },
+    )
+
+    # Create contextual memory manually with the crew's memory components
+    contextual_memory = ContextualMemory(
+        memory_config=crew.memory_config,
+        stm=crew._short_term_memory,
+        ltm=crew._long_term_memory,
+        em=crew._entity_memory,
+        um=crew._user_memory,
+    )
+
+    # Test _fetch_user_context
+    result = contextual_memory._fetch_user_context("test query")
+    
+    # Should return formatted memories from the mock client
+    assert "User memories/preferences" in result
+    assert "- Test memory" in result

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()