diff --git a/lib/crewai/src/crewai/agents/crew_agent_executor.py b/lib/crewai/src/crewai/agents/crew_agent_executor.py
index 8da511864..56abaae02 100644
--- a/lib/crewai/src/crewai/agents/crew_agent_executor.py
+++ b/lib/crewai/src/crewai/agents/crew_agent_executor.py
@@ -50,6 +50,7 @@ from crewai.utilities.agent_utils import (
     handle_unknown_error,
     has_reached_max_iterations,
     is_context_length_exceeded,
+    parse_tool_call_args,
     process_llm_response,
     track_delegation_if_needed,
 )
@@ -894,18 +895,9 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
             ToolUsageStartedEvent,
         )
 
-        json_parse_error: str | None = None
-        if isinstance(func_args, str):
-            try:
-                args_dict = json.loads(func_args)
-            except json.JSONDecodeError as e:
-                args_dict = {}
-                json_parse_error = (
-                    f"Error: Failed to parse tool arguments as JSON: {e}. "
-                    f"Please provide valid JSON arguments for the '{func_name}' tool."
-                )
-        else:
-            args_dict = func_args
+        args_dict, parse_error = parse_tool_call_args(func_args, func_name, call_id, original_tool)
+        if parse_error is not None:
+            return parse_error
 
         if original_tool is None:
             for tool in self.original_tools or []:
@@ -985,9 +977,7 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
                     color="red",
                 )
 
-        if json_parse_error:
-            result = json_parse_error
-        elif hook_blocked:
+        if hook_blocked:
             result = f"Tool execution blocked by hook. Tool: {func_name}"
         elif max_usage_reached and original_tool:
             result = f"Tool '{func_name}' has reached its usage limit of {original_tool.max_usage_count} times and cannot be used anymore."
diff --git a/lib/crewai/src/crewai/experimental/agent_executor.py b/lib/crewai/src/crewai/experimental/agent_executor.py
index 56c4da030..e568dc0d4 100644
--- a/lib/crewai/src/crewai/experimental/agent_executor.py
+++ b/lib/crewai/src/crewai/experimental/agent_executor.py
@@ -66,6 +66,7 @@ from crewai.utilities.agent_utils import (
     has_reached_max_iterations,
     is_context_length_exceeded,
     is_inside_event_loop,
+    parse_tool_call_args,
     process_llm_response,
     track_delegation_if_needed,
 )
@@ -848,13 +849,9 @@ class AgentExecutor(Flow[AgentReActState], CrewAgentExecutorMixin):
         call_id, func_name, func_args = info
 
         # Parse arguments
-        if isinstance(func_args, str):
-            try:
-                args_dict = json.loads(func_args)
-            except json.JSONDecodeError:
-                args_dict = {}
-        else:
-            args_dict = func_args
+        args_dict, parse_error = parse_tool_call_args(func_args, func_name, call_id)
+        if parse_error is not None:
+            return parse_error
 
         # Get agent_key for event tracking
         agent_key = getattr(self.agent, "key", "unknown") if self.agent else "unknown"
diff --git a/lib/crewai/src/crewai/utilities/agent_utils.py b/lib/crewai/src/crewai/utilities/agent_utils.py
index 80c80dbb6..7cad2ad67 100644
--- a/lib/crewai/src/crewai/utilities/agent_utils.py
+++ b/lib/crewai/src/crewai/utilities/agent_utils.py
@@ -1146,6 +1146,36 @@ def extract_tool_call_info(
     return None
 
 
+def parse_tool_call_args(
+    func_args: dict[str, Any] | str,
+    func_name: str,
+    call_id: str,
+    original_tool: Any = None,
+) -> tuple[dict[str, Any], None] | tuple[None, dict[str, Any]]:
+    """Parse tool call arguments from a JSON string or dict.
+
+    Returns:
+        ``(args_dict, None)`` on success, or ``(None, error_result)`` on
+        JSON parse failure where ``error_result`` is a ready-to-return dict
+        with the same shape as ``_execute_single_native_tool_call`` return values.
+    """
+    if isinstance(func_args, str):
+        try:
+            return json.loads(func_args), None
+        except json.JSONDecodeError as e:
+            return None, {
+                "call_id": call_id,
+                "func_name": func_name,
+                "result": (
+                    f"Error: Failed to parse tool arguments as JSON: {e}. "
+                    f"Please provide valid JSON arguments for the '{func_name}' tool."
+                ),
+                "from_cache": False,
+                "original_tool": original_tool,
+            }
+    return func_args, None
+
+
 def _setup_before_llm_call_hooks(
     executor_context: CrewAgentExecutor | AgentExecutor | LiteAgent | None,
     printer: Printer,
diff --git a/lib/crewai/tests/utilities/test_agent_utils.py b/lib/crewai/tests/utilities/test_agent_utils.py
index 31d7b9705..8e3093219 100644
--- a/lib/crewai/tests/utilities/test_agent_utils.py
+++ b/lib/crewai/tests/utilities/test_agent_utils.py
@@ -17,6 +17,7 @@ from crewai.utilities.agent_utils import (
     _format_messages_for_summary,
     _split_messages_into_chunks,
     convert_tools_to_openai_schema,
+    parse_tool_call_args,
     summarize_messages,
 )
 
@@ -922,3 +923,56 @@ class TestParallelSummarizationVCR:
         assert summary_msg["role"] == "user"
         assert "files" in summary_msg
         assert "report.pdf" in summary_msg["files"]
+
+
+class TestParseToolCallArgs:
+    """Unit tests for parse_tool_call_args."""
+
+    def test_valid_json_string_returns_dict(self) -> None:
+        args_dict, error = parse_tool_call_args('{"code": "print(1)"}', "run_code", "call_1")
+        assert error is None
+        assert args_dict == {"code": "print(1)"}
+
+    def test_malformed_json_returns_error_dict(self) -> None:
+        args_dict, error = parse_tool_call_args('{"code": "print("hi")"}', "run_code", "call_1")
+        assert args_dict is None
+        assert error is not None
+        assert error["call_id"] == "call_1"
+        assert error["func_name"] == "run_code"
+        assert error["from_cache"] is False
+        assert "Failed to parse tool arguments as JSON" in error["result"]
+        assert "run_code" in error["result"]
+
+    def test_malformed_json_preserves_original_tool(self) -> None:
+        mock_tool = object()
+        _, error = parse_tool_call_args("{bad}", "my_tool", "call_2", original_tool=mock_tool)
+        assert error is not None
+        assert error["original_tool"] is mock_tool
+
+    def test_malformed_json_original_tool_defaults_to_none(self) -> None:
+        _, error = parse_tool_call_args("{bad}", "my_tool", "call_3")
+        assert error is not None
+        assert error["original_tool"] is None
+
+    def test_dict_input_returned_directly(self) -> None:
+        func_args = {"code": "x = 42"}
+        args_dict, error = parse_tool_call_args(func_args, "run_code", "call_4")
+        assert error is None
+        assert args_dict == {"code": "x = 42"}
+
+    def test_empty_dict_input_returned_directly(self) -> None:
+        args_dict, error = parse_tool_call_args({}, "run_code", "call_5")
+        assert error is None
+        assert args_dict == {}
+
+    def test_valid_json_with_nested_values(self) -> None:
+        args_dict, error = parse_tool_call_args(
+            '{"query": "hello", "options": {"limit": 10}}', "search", "call_6"
+        )
+        assert error is None
+        assert args_dict == {"query": "hello", "options": {"limit": 10}}
+
+    def test_error_result_has_correct_keys(self) -> None:
+        _, error = parse_tool_call_args("{bad json}", "tool", "call_7")
+        assert error is not None
+        assert set(error.keys()) == {"call_id", "func_name", "result", "from_cache", "original_tool"}