fix(tools): return structured JSON error data from tool exceptions

lib/crewai/src/crewai/agents/crew_agent_executor.py

@@ -80,6 +80,7 @@
     aexecute_tool_and_check_finality,
     execute_tool_and_check_finality,
 )
+from crewai.utilities.tool_errors import format_tool_error
 from crewai.utilities.training_handler import CrewTrainingHandler
 
 
@@ -1006,7 +1007,7 @@ def _execute_single_native_tool_call(
 
                 result = format_native_tool_output_for_agent(output_tool, raw_result)
             except Exception as e:
-                result = f"Error executing tool: {e}"
+                result = format_tool_error(e)
                 raw_tool_result = result
                 if self.task:
                     self.task.increment_tools_errors()

lib/crewai/src/crewai/experimental/agent_executor.py

@@ -108,6 +108,7 @@
 )
 from crewai.utilities.step_execution_context import StepExecutionContext, StepResult
 from crewai.utilities.string_utils import sanitize_tool_name
+from crewai.utilities.tool_errors import format_tool_error
 from crewai.utilities.tool_utils import execute_tool_and_check_finality
 from crewai.utilities.training_handler import CrewTrainingHandler
 from crewai.utilities.types import LLMMessage
@@ -1615,9 +1616,10 @@ def execute_tool_action(self) -> Literal["tool_completed", "tool_result_is_final
             if self.task:
                 self.task.increment_tools_errors()
 
-            error_observation = f"\nObservation: Error executing tool: {e}"
+            structured_error = format_tool_error(e)
+            error_observation = f"\nObservation: {structured_error}"
             action.text += error_observation
-            action.result = str(e)
+            action.result = structured_error
             self._append_message_to_state(action.text)
 
             reasoning_prompt = I18N_DEFAULT.slice("post_tool_reasoning")
@@ -1736,7 +1738,7 @@ def execute_native_tool(
                         ordered_results[idx] = {
                             "call_id": call_id,
                             "func_name": func_name,
-                            "result": f"Error executing tool: {e}",
+                            "result": format_tool_error(e),
                             "from_cache": False,
                             "original_tool": None,
                         }
@@ -1999,7 +2001,7 @@ def _execute_single_native_tool_call(self, tool_call: Any) -> dict[str, Any]:
                         output_tool, raw_result
                     )
                 except Exception as e:
-                    result = f"Error executing tool: {e}"
+                    result = format_tool_error(e)
                     raw_tool_result = result
                     if self.task:
                         self.task.increment_tools_errors()

lib/crewai/src/crewai/llm.py

@@ -52,6 +52,7 @@
 from crewai.utilities.logger_utils import suppress_warnings
 from crewai.utilities.string_utils import sanitize_tool_name
 from crewai.utilities.token_counter_callback import TokenCalcHandler
+from crewai.utilities.tool_errors import format_tool_error
 
 
 try:
@@ -1755,11 +1756,12 @@ def _handle_tool_call(
                 return result
             except Exception as e:
                 fn = available_functions.get(function_name, lambda: None)
+                structured_error = format_tool_error(e)
                 logging.error(f"Error executing function '{function_name}': {e}")
                 crewai_event_bus.emit(
                     self,
                     event=LLMCallFailedEvent(
-                        error=f"Tool execution error: {e!s}",
+                        error=structured_error,
                         from_task=from_task,
                         from_agent=from_agent,
                         call_id=get_current_call_id(),
@@ -1770,7 +1772,7 @@ def _handle_tool_call(
                     event=ToolUsageErrorEvent(
                         tool_name=function_name,
                         tool_args=function_args,
-                        error=f"Tool execution error: {e!s}",
+                        error=structured_error,
                         from_task=from_task,
                         from_agent=from_agent,
                     ),

lib/crewai/src/crewai/utilities/agent_utils.py

@@ -37,6 +37,7 @@
 from crewai.utilities.pydantic_schema_utils import generate_model_description
 from crewai.utilities.string_utils import sanitize_tool_name
 from crewai.utilities.token_counter_callback import TokenCalcHandler
+from crewai.utilities.tool_errors import format_tool_error
 from crewai.utilities.types import LLMMessage
 
 
@@ -1546,7 +1547,7 @@ def execute_single_native_tool_call(
 
                 result = format_native_tool_output_for_agent(output_tool, raw_result)
             except Exception as e:
-                result = f"Error executing tool: {e}"
+                result = format_tool_error(e)
                 raw_tool_result = result
                 if task:
                     task.increment_tools_errors()

lib/crewai/src/crewai/utilities/tool_errors.py

@@ -0,0 +1,29 @@
+"""Structured tool error formatting for agent consumption.
+
+When a tool raises an exception, the agent needs structured information
+to decide whether to retry, fix its input, or skip the tool entirely.
+This module provides a consistent error format across all executors.
+"""
+
+import json
+import traceback
+
+RETRYABLE_EXCEPTIONS = (TimeoutError, ConnectionError, OSError)
+
+
+def format_tool_error(exception: Exception, include_traceback: bool = False) -> str:
+    """Format a tool execution error as structured JSON for the agent.
+
+    Returns a string with the "Error executing tool:" prefix (for backward
+    compatibility with existing parsing) followed by a JSON object containing
+    the exception type, message, and retryability hint.
+    """
+    error_data = {
+        "error": True,
+        "type": type(exception).__name__,
+        "message": str(exception),
+        "retryable": isinstance(exception, RETRYABLE_EXCEPTIONS),
+    }
+    if include_traceback:
+        error_data["traceback"] = traceback.format_exc(limit=3)
+    return f"Error executing tool: {json.dumps(error_data)}"

lib/crewai/tests/test_tool_errors.py

@@ -0,0 +1,121 @@
+"""Tests for structured tool error formatting."""
+
+import json
+
+import pytest
+
+from crewai.utilities.tool_errors import RETRYABLE_EXCEPTIONS, format_tool_error
+
+
+class TestFormatToolError:
+    """Tests for the format_tool_error utility function."""
+
+    def test_returns_string_with_prefix(self):
+        err = ValueError("bad input")
+        result = format_tool_error(err)
+        assert result.startswith("Error executing tool: ")
+
+    def test_contains_valid_json_after_prefix(self):
+        err = ValueError("bad input")
+        result = format_tool_error(err)
+        json_str = result[len("Error executing tool: "):]
+        parsed = json.loads(json_str)
+        assert isinstance(parsed, dict)
+
+    def test_error_flag_is_true(self):
+        err = RuntimeError("something broke")
+        result = format_tool_error(err)
+        parsed = json.loads(result[len("Error executing tool: "):])
+        assert parsed["error"] is True
+
+    def test_preserves_exception_type(self):
+        err = KeyError("missing_key")
+        result = format_tool_error(err)
+        parsed = json.loads(result[len("Error executing tool: "):])
+        assert parsed["type"] == "KeyError"
+
+    def test_preserves_exception_message(self):
+        err = ValueError("count must be positive")
+        result = format_tool_error(err)
+        parsed = json.loads(result[len("Error executing tool: "):])
+        assert parsed["message"] == "count must be positive"
+
+    def test_retryable_true_for_timeout(self):
+        err = TimeoutError("connection timed out")
+        result = format_tool_error(err)
+        parsed = json.loads(result[len("Error executing tool: "):])
+        assert parsed["retryable"] is True
+
+    def test_retryable_true_for_connection_error(self):
+        err = ConnectionError("refused")
+        result = format_tool_error(err)
+        parsed = json.loads(result[len("Error executing tool: "):])
+        assert parsed["retryable"] is True
+
+    def test_retryable_true_for_os_error(self):
+        err = OSError("disk full")
+        result = format_tool_error(err)
+        parsed = json.loads(result[len("Error executing tool: "):])
+        assert parsed["retryable"] is True
+
+    def test_retryable_false_for_value_error(self):
+        err = ValueError("invalid")
+        result = format_tool_error(err)
+        parsed = json.loads(result[len("Error executing tool: "):])
+        assert parsed["retryable"] is False
+
+    def test_retryable_false_for_type_error(self):
+        err = TypeError("wrong type")
+        result = format_tool_error(err)
+        parsed = json.loads(result[len("Error executing tool: "):])
+        assert parsed["retryable"] is False
+
+    def test_retryable_false_for_key_error(self):
+        err = KeyError("not found")
+        result = format_tool_error(err)
+        parsed = json.loads(result[len("Error executing tool: "):])
+        assert parsed["retryable"] is False
+
+    def test_no_traceback_by_default(self):
+        err = ValueError("test")
+        result = format_tool_error(err)
+        parsed = json.loads(result[len("Error executing tool: "):])
+        assert "traceback" not in parsed
+
+    def test_traceback_included_when_requested(self):
+        try:
+            raise ValueError("deliberate error")
+        except ValueError as e:
+            result = format_tool_error(e, include_traceback=True)
+        parsed = json.loads(result[len("Error executing tool: "):])
+        assert "traceback" in parsed
+        assert "ValueError" in parsed["traceback"]
+
+    def test_handles_exception_with_special_characters(self):
+        err = ValueError('path "C:\\Users\\test" not found')
+        result = format_tool_error(err)
+        parsed = json.loads(result[len("Error executing tool: "):])
+        assert 'C:\\Users\\test' in parsed["message"]
+
+    def test_handles_exception_with_empty_message(self):
+        err = RuntimeError()
+        result = format_tool_error(err)
+        parsed = json.loads(result[len("Error executing tool: "):])
+        assert parsed["type"] == "RuntimeError"
+        assert parsed["message"] == ""
+
+    def test_handles_custom_exception(self):
+        class MyToolError(Exception):
+            pass
+
+        err = MyToolError("custom failure")
+        result = format_tool_error(err)
+        parsed = json.loads(result[len("Error executing tool: "):])
+        assert parsed["type"] == "MyToolError"
+        assert parsed["message"] == "custom failure"
+        assert parsed["retryable"] is False
+
+    def test_retryable_exceptions_tuple_contains_expected_types(self):
+        assert TimeoutError in RETRYABLE_EXCEPTIONS
+        assert ConnectionError in RETRYABLE_EXCEPTIONS
+        assert OSError in RETRYABLE_EXCEPTIONS