feat(tools): surface tool failures instead of reporting them as success

A tool can finish without raising and still fail to do what it was asked.
Slack answers HTTP 200 with `{"ok": false, "error": "channel_not_found"}`;
an MCP server sets `isError`; a CrewAI AMP action returns
`API request failed: ...`. In every case the call "worked", so the error
text reached the agent as an ordinary result, the agent narrated the
problem in prose, and the run was recorded as a success.

Concretely: five failed `slackbot_send_message` calls each rendered as
"Tool Execution Completed", the task passed, and the crew passed -- with
the only evidence being a sentence in the final answer. Nothing
downstream could tell the difference, and an agent that keeps going on a
step that silently did nothing builds the rest of its work on it.

Give that outcome a type and a reaction:

- `ToolFailure` -- what a tool returns instead of an error string. The
  agent still reads prose via `as_agent_message()`, so model behavior is
  unchanged; the framework now knows the call failed.
- `ToolFailurePolicy` -- `ignore` (previous behavior), `warn` (default:
  record + emit, keep going), `raise` (abort with
  `ToolExecutionFailedError`). Resolved most-specific-first: tool, task,
  agent, crew.
- `ToolFailureDetectedEvent` -- emitted before a `raise` aborts, so
  subscribers always observe the failure. `ToolUsageFinishedEvent` also
  carries a `failure` field so a trace UI can mark the call failed
  without correlating two events.
- `tool_failures` on `TaskOutput`, `CrewOutput` and `LiteAgentOutput`,
  plus `has_tool_failures`, so consumers never parse a string.

Detection is strictly declarative -- no string sniffing, so a tool that
legitimately returns text about an error is never misread as failing.
Failures come from a returned `ToolFailure`, a raised exception, MCP
`isError`, a spent `max_usage_count`, or an unknown tool.

Wired into all four tool-execution paths (the ReAct path and the three
native function-calling implementations). Sources updated to report
structurally: `MCPClient.call_tool_result()` preserves `isError` that
`call_tool()` dropped, and `CrewAIPlatformActionTool` returns a
`ToolFailure` for non-2xx and for caught exceptions.

Two latent bugs fixed along the way: `ToolUsage` assumed every agent has
a `fingerprint` (LiteAgent does not), and policy resolution now tolerates
malformed values rather than letting telemetry take down a tool call.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ETacm2dMASfpMAYUiDu5YG

Author: joaomdmoura

docs/edge/en/concepts/tools.mdx

@@ -334,6 +334,118 @@ writer1 = Agent(
     #...
 ```
 
+## Reporting Tool Failures
+
+A tool can finish without raising and still fail to do what it was asked. Slack
+answers `HTTP 200` with `{"ok": false, "error": "channel_not_found"}`; an MCP
+server sets `isError`; a platform action returns an error payload. The tool call
+"worked", so the error text reaches the agent as an ordinary result — the agent
+narrates the problem in its final answer and the run is recorded as a success.
+
+Return a `ToolFailure` instead of an error string and the framework can tell the
+difference:
+
+```python Code
+from typing import Any
+
+from crewai.tools import BaseTool
+from crewai.tools.tool_failure import ToolFailure
+
+
+class SendSlackMessage(BaseTool):
+    name: str = "send_slack_message"
+    description: str = "Post a message to a Slack channel."
+
+    def _run(self, channel: str, text: str) -> Any:
+        payload = slack.post(channel=channel, text=text)
+        if not payload["ok"]:
+            return ToolFailure(
+                message=f"Slack rejected the message: {payload['error']}",
+                code=payload["error"],
+                retryable=payload["error"] == "rate_limited",
+            )
+        return payload
+```
+
+The agent still reads plain prose — `ToolFailure.as_agent_message()` renders the
+message — so model behavior is unchanged. What changes is that the failure is now
+visible to everything downstream.
+
+Detection is strictly declarative. CrewAI never guesses whether a string "looks
+like" an error, so a tool that legitimately returns text about an error is never
+misread as having failed. Failures are recorded when a tool returns a
+`ToolFailure`, when a tool raises, when an MCP server sets `isError`, when a
+tool's `max_usage_count` is spent, or when the agent calls a tool that does not exist.
+
+### Choosing a Failure Policy
+
+`tool_failure_policy` controls what happens next:
+
+| Policy | Behavior |
+| :-- | :-- |
+| `ignore` | Nothing is recorded, emitted, or acted on. |
+| `warn` *(default)* | Records the failure, emits `ToolFailureDetectedEvent`, and continues. |
+| `raise` | Records and emits, then aborts with `ToolExecutionFailedError`. |
+
+```python Code
+from crewai import Agent, Task
+from crewai.tools.tool_failure import ToolFailurePolicy
+
+agent = Agent(
+    role="Slack Messenger",
+    goal="Post the report to Slack",
+    backstory="...",
+    tools=[SendSlackMessage()],
+    tool_failure_policy=ToolFailurePolicy.WARN,
+)
+
+# Tighten a single high-stakes task without changing the agent.
+task = Task(
+    description="Post the final report to #engineering",
+    expected_output="Confirmation the message was posted",
+    agent=agent,
+    tool_failure_policy=ToolFailurePolicy.RAISE,
+)
+```
+
+The most specific setting wins: tool, then task, then agent, then crew, then the
+`warn` default.
+
+### Inspecting Failures
+
+Recorded failures are structured, so nothing downstream has to parse a string:
+
+```python Code
+result = crew.kickoff()
+
+if result.has_tool_failures:
+    for record in result.tool_failures:
+        print(record.tool_name)          # "send_slack_message"
+        print(record.failure.code)       # "channel_not_found"
+        print(record.failure.reason)     # ToolFailureReason.TOOL_REPORTED
+        print(record.summary())
+```
+
+`tool_failures` is available on `TaskOutput`, `CrewOutput`, and
+`LiteAgentOutput`. A crew can finish successfully with a non-empty list — check
+it before treating `raw` as complete.
+
+To react as failures happen, subscribe to the event:
+
+```python Code
+from crewai.events import ToolFailureDetectedEvent
+from crewai.events.event_bus import crewai_event_bus
+
+
+@crewai_event_bus.on(ToolFailureDetectedEvent)
+def on_tool_failure(source, event):
+    print(f"{event.tool_name} failed: {event.failure.message} ({event.policy})")
+```
+
+The event is emitted before the `raise` policy aborts, so subscribers always
+observe the failure. `ToolUsageFinishedEvent` also carries a `failure` field, letting
+a trace UI mark the call as failed without correlating two events.
+
 ## Conclusion
 
 Tools are pivotal in extending the capabilities of CrewAI agents, enabling them to undertake a broad spectrum of tasks and collaborate effectively.

lib/crewai-tools/src/crewai_tools/tools/crewai_platform_tools/crewai_platform_action_tool.py

@@ -5,6 +5,7 @@
 from typing import Any
 
 from crewai.tools import BaseTool
+from crewai.tools.tool_failure import ToolFailure
 from crewai.utilities.pydantic_schema_utils import create_model_from_schema
 from pydantic import Field, create_model
 import requests
@@ -49,7 +50,7 @@ def __init__(
         self.action_name = action_name
         self.action_schema = action_schema
 
-    def _run(self, **kwargs: Any) -> str:
+    def _run(self, **kwargs: Any) -> Any:
         try:
             cleaned_kwargs = {
                 key: value for key, value in kwargs.items() if value is not None
@@ -85,9 +86,22 @@ def _run(self, **kwargs: Any) -> str:
                         error_message = str(error_info)
                 else:
                     error_message = str(data)
-                return f"API request failed: {error_message}"
+                # The platform returns a non-2xx when the upstream app rejects
+                # the action -- e.g. Slack answering channel_not_found. That is
+                # the single most common way an agent "succeeds" at doing
+                # nothing, so report it as a failure rather than as prose.
+                return ToolFailure(
+                    message=f"API request failed: {error_message}",
+                    code=str(response.status_code),
+                    retryable=response.status_code >= 500,
+                    details={"action": self.action_name},
+                )
 
             return json.dumps(data, indent=2)
 
         except Exception as e:
-            return f"Error executing action {self.action_name}: {e!s}"
+            return ToolFailure(
+                message=f"Error executing action {self.action_name}: {e!s}",
+                code=e.__class__.__name__,
+                details={"action": self.action_name},
+            )

lib/crewai/src/crewai/agent/core.py

@@ -86,6 +86,7 @@
 from crewai.skills.models import INSTRUCTIONS, Skill as SkillModel
 from crewai.state.checkpoint_config import CheckpointConfig, apply_checkpoint
 from crewai.tools.agent_tools.agent_tools import AgentTools
+from crewai.tools.tool_failure import ToolExecutionFailedError
 from crewai.types.callback import SerializableCallable
 from crewai.types.usage_metrics import UsageMetrics
 from crewai.utilities.agent_utils import (
@@ -131,7 +132,10 @@
     from crewai.utilities.types import LLMMessage
 
 
-_passthrough_exceptions: tuple[type[Exception], ...] = ()
+# Exceptions that must not be swallowed into the max_retry_limit loop.
+# A tool_failure_policy="raise" abort is a deliberate stop, not a transient
+# error worth re-running the whole task for.
+_passthrough_exceptions: tuple[type[Exception], ...] = (ToolExecutionFailedError,)
 
 _EXECUTOR_CLASS_MAP: dict[str, type] = {
     "CrewAgentExecutor": CrewAgentExecutor,
@@ -550,6 +554,8 @@ def _prepare_task_execution(
 
         self._inject_date_to_task(task)
 
+        self.reset_tool_failures()
+
         if self.tools_handler:
             self.tools_handler.last_used_tool = None
 

lib/crewai/src/crewai/agents/agent_builder/base_agent.py

@@ -44,6 +44,7 @@
 from crewai.skills.models import Skill
 from crewai.state.checkpoint_config import CheckpointConfig, _coerce_checkpoint
 from crewai.tools.base_tool import BaseTool, Tool
+from crewai.tools.tool_failure import ToolFailurePolicy, ToolFailureRecord
 from crewai.types.callback import SerializableCallable
 from crewai.utilities.config import process_config
 from crewai.utilities.i18n import I18N, get_i18n
@@ -264,6 +265,7 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
     _original_backstory: str | None = PrivateAttr(default=None)
     _token_process: TokenProcess = PrivateAttr(default_factory=TokenProcess)
     _kickoff_event_id: str | None = PrivateAttr(default=None)
+    _tool_failures: list[ToolFailureRecord] = PrivateAttr(default_factory=list)
     id: UUID4 = Field(default_factory=uuid.uuid4, frozen=True)
     role: str = Field(description="Role of the agent")
     goal: str = Field(description="Objective of the agent")
@@ -298,6 +300,18 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
     max_iter: int = Field(
         default=25, description="Maximum iterations for an agent to execute a task"
     )
+    tool_failure_policy: ToolFailurePolicy = Field(
+        default=ToolFailurePolicy.WARN,
+        description=(
+            "How to react when a tool runs to completion but reports that it "
+            "failed (an upstream API rejecting the request, an MCP server "
+            "setting isError, a platform action returning an error payload). "
+            "'ignore' restores pre-1.16 behavior and records nothing; 'warn' "
+            "records the failure, emits ToolFailureDetectedEvent and keeps "
+            "going; 'raise' additionally aborts with ToolExecutionFailedError. "
+            "A Task or a tool may override this for a narrower scope."
+        ),
+    )
     agent_executor: Annotated[
         SerializeAsAny[BaseAgentExecutor] | None,
         BeforeValidator(_validate_executor_ref),
@@ -652,6 +666,20 @@ def key(self) -> str:
         ]
         return md5("|".join(source).encode(), usedforsecurity=False).hexdigest()
 
+    @property
+    def last_tool_failures(self) -> list[ToolFailureRecord]:
+        """Tool failures recorded during the most recent execution.
+
+        Empty when nothing failed, or when ``tool_failure_policy`` is
+        ``ignore``. Reset at the start of each task execution, mirroring
+        ``last_messages``.
+        """
+        return self._tool_failures
+
+    def reset_tool_failures(self) -> None:
+        """Clear recorded tool failures before a new execution begins."""
+        self._tool_failures = []
+
     @abstractmethod
     def execute_task(
         self,

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

@@ -49,6 +49,14 @@
     run_after_tool_call_hooks,
     run_before_tool_call_hooks,
 )
+from crewai.tools.tool_failure import (
+    ToolExecutionFailedError,
+    ToolFailure,
+    ToolFailureReason,
+    detect_tool_failure,
+    failure_from_exception,
+    handle_tool_failure,
+)
 from crewai.types.callback import SerializableCallable
 from crewai.utilities.agent_utils import (
     _llm_stop_words_applied,
@@ -431,6 +439,12 @@ def _invoke_loop_react(self) -> AgentFinish:
                 self._invoke_step_callback(formatted_answer)
                 self._append_message(formatted_answer.text)
 
+            except ToolExecutionFailedError:
+                # tool_failure_policy="raise" asked for the run to stop; the
+                # generic handler below would otherwise feed it back to the
+                # LLM as a recoverable observation.
+                raise
+
             except OutputParserError as e:
                 formatted_answer = handle_output_parser_exception(  # type: ignore[assignment]
                     e=e,
@@ -925,6 +939,7 @@ def _execute_single_native_tool_call(
         from_cache = False
         result: str = "Tool not found"
         raw_tool_result: Any = result
+        tool_failure: ToolFailure | None = None
         input_str = json.dumps(args_dict) if args_dict else ""
         if self.tools_handler and self.tools_handler.cache and output_tool is not None:
             cached_result = self.tools_handler.cache.read(
@@ -933,6 +948,7 @@ def _execute_single_native_tool_call(
             if cached_result is not None:
                 raw_tool_result = cached_result
                 result = format_native_tool_output_for_agent(output_tool, cached_result)
+                tool_failure = detect_tool_failure(cached_result)
                 from_cache = True
 
         agent_key = getattr(self.agent, "key", "unknown") if self.agent else "unknown"
@@ -967,6 +983,9 @@ def _execute_single_native_tool_call(
         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."
             raw_tool_result = result
+            tool_failure = ToolFailure(
+                message=result, reason=ToolFailureReason.USAGE_LIMIT
+            )
         elif (
             not from_cache
             and func_name in available_functions
@@ -992,9 +1011,11 @@ def _execute_single_native_tool_call(
                         )
 
                 result = format_native_tool_output_for_agent(output_tool, raw_result)
+                tool_failure = detect_tool_failure(raw_result)
             except Exception as e:
                 result = f"Error executing tool: {e}"
                 raw_tool_result = result
+                tool_failure = failure_from_exception(e)
                 if self.task:
                     self.task.increment_tools_errors()
                 crewai_event_bus.emit(
@@ -1036,9 +1057,23 @@ def _execute_single_native_tool_call(
                     agent_key=agent_key,
                     started_at=started_at,
                     finished_at=datetime.now(),
+                    failure=tool_failure,
                 ),
             )
 
+        # After the hooks and the finished event, so subscribers see the full
+        # lifecycle even when the policy is about to abort the run.
+        if tool_failure is not None:
+            handle_tool_failure(
+                tool_failure,
+                tool_name=func_name,
+                tool_args=args_dict,
+                tool=structured_tool,
+                agent=self.agent,
+                task=self.task,
+                crew=self.crew,
+            )
+
         return {
             "call_id": call_id,
             "func_name": func_name,
@@ -1246,6 +1281,12 @@ async def _ainvoke_loop_react(self) -> AgentFinish:
                 await self._ainvoke_step_callback(formatted_answer)
                 self._append_message(formatted_answer.text)
 
+            except ToolExecutionFailedError:
+                # tool_failure_policy="raise" asked for the run to stop; the
+                # generic handler below would otherwise feed it back to the
+                # LLM as a recoverable observation.
+                raise
+
             except OutputParserError as e:
                 formatted_answer = handle_output_parser_exception(  # type: ignore[assignment]
                     e=e,

lib/crewai/src/crewai/crews/crew_output.py

@@ -7,6 +7,7 @@
 
 from crewai.tasks.output_format import OutputFormat
 from crewai.tasks.task_output import TaskOutput
+from crewai.tools.tool_failure import ToolFailureRecord
 from crewai.types.usage_metrics import UsageMetrics
 
 
@@ -31,6 +32,21 @@ class CrewOutput(BaseModel):
         default_factory=UsageMetrics,
     )
 
+    @property
+    def tool_failures(self) -> list[ToolFailureRecord]:
+        """Every tool failure recorded across all tasks, in task order.
+
+        A crew can finish with a non-empty list: agents routinely narrate a
+        failed step in prose and carry on, which used to make the run look
+        entirely successful. Check this before treating ``raw`` as complete.
+        """
+        return [failure for task in self.tasks_output for failure in task.tool_failures]
+
+    @property
+    def has_tool_failures(self) -> bool:
+        """Whether any tool reported a failure during this crew run."""
+        return any(task.tool_failures for task in self.tasks_output)
+
     @property
     def usage_metrics(self) -> dict[str, Any]:
         """Token usage as a plain dict.