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

* 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

* fix(tools): address review round 1 on tool-failure signalling

Five real defects from Bugbot, none of them cosmetic.

Tool-scoped policy never applied (high). `resolve_tool_failure_policy`
read `tool_failure_policy` off the object handed to it, but every
execution path passes the `CrewStructuredTool` wrapper, which never
carried the attribute -- and `BaseTool` never declared it in the first
place. A tool-scoped `raise`/`ignore` was silently ignored while the
docs and a unit test claimed otherwise; the test passed only because it
called the resolver directly with an authored tool. Declared the field on
`BaseTool`, propagated it through `to_structured_tool()` and
`CrewStructuredTool`, and made resolution fall back through
`_original_tool` so either shape works.

A failed call still printed the green "Completed" panel, then the red
one. That is the terminal version of the exact bug this PR is about.
Suppressed the success panel when the call reported failure.

A raised tool printed twice: `ToolUsageErrorEvent` already renders a red
panel, and the new failure panel repeated it. The event is still emitted
-- policy and traces need it -- but the duplicate console output is gone.
Both decisions now live in named predicates on `ConsoleFormatter` rather
than inline in the listener closure, so they are directly testable.

Unknown tools were reported on the ReAct path but silently ignored on all
three native paths, so the same miss was loud or silent depending on
executor style. Native paths now record `UNKNOWN_TOOL` too. This also
surfaced a live `NameError`: ruff had pruned `ToolFailureReason` from
`agent_utils` as unused, so the new branch would have crashed at runtime.

`LiteAgentOutput` had `tool_failures` but not `has_tool_failures`, which
the PR promised on all three output types -- an `AttributeError` for any
caller sharing one check across result types.

Testing: 16 further tests, 45 total. Two console tests were passing
vacuously because `emit()` dispatches sync handlers on a thread pool, so
the assertions raced the handler; they now assert on the predicates
directly, and the native-path test drains the bus with `flush()` and
checks the synchronously-written record. Full suite still matches
baseline exactly at 377 pre-existing failures.

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

* chore: update tool specifications

* fix(tools): address review round 2 and fix CI type failure

CI caught a type error I should have: widening `agent` to accept a
`LiteAgent` (so a standalone LiteAgent resolves its own policy) left the
declared signatures behind. Widened `execute_tool_and_check_finality`,
its async twin, and `ToolCallHookContext` to `Agent | BaseAgent |
LiteAgent | None`, which is what those actually receive now.

Seven CodeRabbit findings, all verified against the code first:

`raise` was being downgraded by three enclosing handlers. With
`max_execution_time` set, `_execute_with_timeout` wrapped every exception
in `RuntimeError`, so `_check_execution_error` no longer recognized the
passthrough and sent the task through the retry loop instead of aborting.
`StepExecutor.execute` turned it into `StepResult(success=False)` and let
the plan continue. `LiteAgent.kickoff` ran it through
`handle_unknown_error` and printed "This is likely a bug - please report
it" for what is a deliberate, configured stop.

Failure records were dropped on two paths. `reset_tool_failures()` only
ran in `_prepare_task_execution`, so `Agent.kickoff()` / `kickoff_async()`
— which enter through `_prepare_kickoff` — accumulated records across
runs. And a guardrail retry calls `execute_task` again, which resets the
agent, so a tool that failed on a blocked attempt vanished from the final
output entirely: a run could report zero failures having demonstrably
failed one. Failures now accumulate across guardrail attempts.

Writing the tests for that surfaced a further miss of my own:
`Agent.kickoff()` builds its `LiteAgentOutput` in `agent/core.py` via
`AgentExecutor`, not through `LiteAgent`, so `tool_failures` was always
empty there regardless of the recording fix. Wired up, and the LiteAgent
path now reads from whichever agent the executor was handed
(`original_agent` under kickoff, `self` standalone) rather than assuming.

`last_tool_failures` returns a copy, so a caller cannot mutate the
agent's record or watch it shift mid-run.

Testing: 7 further tests, 52 total, covering the timeout wrapper, the
retry limit, kickoff reset, the kickoff output path, copy semantics and
guardrail accumulation. Full suite matches baseline exactly at 377
pre-existing failures; mypy clean on every changed file.

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

* fix(tools): make crew-scoped policy real and close the last raise leak

Two findings, and the first was a documented feature that never worked.

`resolve_tool_failure_policy` consulted a crew, and the docs advertised
crew as a scope, but `Crew` had no `tool_failure_policy` field at all --
and even with one it was unreachable, because `BaseAgent` defaulted the
policy to `WARN` rather than `None`, so resolution always stopped at the
agent. Crew-level configuration was silently ignored.

Fixed by making "inherit" the default everywhere instead of baking `warn`
into one layer: `Crew` gains the field, and `BaseAgent`/`LiteAgent`
default to `None` like `Task` and `BaseTool` already did. The resolver
owns the single fallback, so the chain is genuinely
tool > task > agent > crew > warn and the effective default with nothing
configured is still `warn`. Reading `agent.tool_failure_policy` now
returns `None` (meaning "inherit") rather than `WARN`.

The other: `StepExecutor` re-raised `ToolExecutionFailedError` from its
outer handler, but the nested handler around the native-to-text tooling
fallback still caught it and returned `StepResult(success=False)`. An
agent whose LLM lacked native tool calling would therefore not abort
under `raise`. That is the third distinct place this exception was being
downgraded; it now re-raises there too.

Testing: 8 further tests, 60 total, including the full precedence chain
walked one level at a time and crew-scoped `raise`/`ignore` driven
end-to-end through `kickoff()` rather than only through the resolver --
the gap that let the original crew bug pass review. Full suite matches
baseline exactly at 377 pre-existing failures; mypy clean on every
changed file.

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

* docs: trim comments and docstrings on tool-failure signalling

Prose only -- no behavior change. Cut the module docstring, the longer
class and method docstrings, the multi-line inline comments, and the
verbose Field descriptions down to what actually earns its place. Net 87
lines lighter.

Kept the "why" in every case where the reason is non-obvious (why the
event fires before a raise, why the policy reads through the tool wrapper,
why the bus needs draining in tests) and dropped the restatements of what
the code already says.

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

* fix(tools): make ignore truly silent, stop caching failures, close 4 gaps

Six findings from the latest review round, all verified against the code
before touching it.

`ignore` was not silent. `ToolUsageFinishedEvent.failure` was set before
the policy ran, so traces still saw a failed call under a policy documented
as surfacing nothing. Worse, the console then showed *no* panel at all:
green was suppressed because `failure` was present, red was skipped because
`ignore` never emits `ToolFailureDetectedEvent`. New `reportable_failure()`
resolves the policy before the finished event and drops the flag under
`ignore`; wired into all four execution paths.

Failures were being cached. `CacheHandler.add` stored a `ToolFailure` like
any other result, so a transient error became permanent for the rest of the
run and every later hit re-reported a call that never re-ran. The cache now
refuses to store declared failures -- fixed at the single choke point rather
than at each of the four call sites.

A spent `max_usage_count` was invisible on the shared native path.
`BaseTool._claim_usage` returned a bare string that only the executors
recognising that exact message treated as a failure. It now returns a
`ToolFailure` with `USAGE_LIMIT`, so every path records it.

A guardrail returning a whole `TaskOutput` replaced the output without
carrying accumulated failures over, so earlier attempts vanished from
`CrewOutput.tool_failures`. New `merge_tool_failures()` combines and
deduplicates, and the retry-rebuild path uses it too.

A hook-blocked call inherited a cached failure and attributed it to a call
that never ran. Now cleared. Not reachable through the built-in cache once
failures stop being cached, so the test injects a custom cache handler that
does retain them -- verified to fail without the guard.

Also removed a `datetime` import left unused by the earlier console-test
rewrite.

Testing: 13 further tests, 73 total. Full suite matches baseline exactly at
377 pre-existing failures; the usage-limit suites that `_claim_usage`
touches pass unchanged; mypy clean on every changed file.

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

* fix(tools): let raise through the parallel native path, guard all handlers

Chasing down CodeRabbit's note about callers of
execute_single_native_tool_call turned up a fifth place this exception was
being downgraded: the experimental executor's parallel branch wrapped
future.result() in a broad except and folded the abort into a fake tool
result, so the remaining parallel calls carried on. The sequential path and
crew_agent_executor's parallel branch were already fine.

Five separate handlers have swallowed this during review, so added a guard
test asserting the passthrough at every site rather than trusting the next
one gets spotted.

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

* fix(tools): keep a failed tool out of the final answer, finish crew scope

Three more findings, all confirmed against the code.

A failed `result_as_answer` tool still became the task's output. The native
paths already excluded raised errors and hook blocks from short-circuiting,
but not declared failures -- so an error message silently became the answer,
which is the exact shape of bug this PR exists to prevent. Fixed on all
paths, and there were three independent override points, not one:
`ToolResult.result_as_answer` in tool_utils, the `execution_result`
finality checks in both executors, and `process_tool_results()`, which
reads `agent.tools_results` back separately. The first two fixes alone left
the behavior unchanged; only the third made the test pass.

`ToolUsage` never received a crew, so a crew-level `ignore` half-applied:
recording and `ToolFailureDetectedEvent` stayed quiet, but the flag was
still attached to `ToolUsageFinishedEvent`. It now takes and stores `crew`.

`CrewAgentExecutor.invoke`/`ainvoke` routed a deliberate stop through
`handle_unknown_error`, printing "An unknown error occurred" on verbose
runs. LiteAgent already special-cased this; both now do.

Testing: 5 further tests, 79 total, including that a *successful*
`result_as_answer` tool still short-circuits. Full suite matches baseline
exactly at 377 pre-existing failures; mypy clean on every changed file.

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

* fix(tools): report malformed tool args, correlate the failure event

Two findings from the latest round.

Malformed native tool arguments returned early with a plain error dict and
never reported a failure, so `ToolFailureReason.INVALID_INPUT` was declared
but unreferenced -- a bad tool call was absent from records, events and
`raise` aborts. `parse_tool_call_args` now carries an INVALID_INPUT failure
on the error dict and both executors report it before returning.

`ToolFailureDetectedEvent` never set `agent_id`, so a trace could not tie it
to a specific agent instance. Fixing that exposed the same gap running the
other way: `ToolUsage`'s own started/finished/error events never set
`agent_id` either, so on the ReAct path the paired finished event had
nothing to correlate against. Both now set it.

Set explicitly rather than via `from_agent`, which would also overwrite
`agent_role` and lose the `_original_role` preference those events already
apply -- a behavior change that has nothing to do with correlation.

Testing: 5 further tests, 84 total, asserting the ids match between the
failure event and its paired finished event. One existing test pinned the
exact key set of the parse-error dict and was updated for the new key. Full
suite matches baseline at 377 pre-existing failures; the one apparent
addition was the known `test_trace_enable_disable` order-flake, confirmed by
re-running rather than assumed.

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

* fix(tools): scope failure accumulation per execution, drop deprecated executor

Two review requests from @lorenzejay.

Accumulation no longer lives as mutable state on the shared agent. A
ContextVar collector is opened around each execution -- task, kickoff, and
each guardrail retry -- and the output reads that collector directly instead
of copying the agent's list. ContextVars are copied per asyncio task and per
thread, so concurrent executions cannot see each other's records, and
nesting is safe for retries. `last_tool_failures` prefers the active
collector and falls back to the last completed execution, so the accessor is
correct during a run too. The per-execution reset that caused the erasure is
gone.

Reproducing this took some digging and the finding is worth recording: crew
tasks *cannot* hit it, because `AgentExecutor` refuses concurrent reuse of
one instance and raises. `agent.kickoff()` has no such guard, and there the
bug reproduces exactly as reported -- two concurrent kickoffs each returned
two records. The regression test forces the overlap with a barrier so it is
deterministic rather than timing-dependent, and I verified it reports [2, 2]
against the old behavior and [1, 1] now.

Removed the tool-failure integration from `CrewAgentExecutor` entirely; that
file is back to its state on main. Note the shared ReAct helper it calls
still records failures, since that is common code rather than new behavior in
the deprecated file -- so a `raise` policy will be swallowed by that
executor's generic handler. Flagged on the PR rather than papered over.

Testing: 89 total. Two tests I wrote for this were vacuous on the first
attempt -- they passed against the simulated pre-fix code -- so each
concurrency test was checked against the old behavior before being kept.

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

* fix(tools): report malformed calls everywhere, drop the unused block reason

Four findings.

`execute_single_native_tool_call` swallowed a JSON decode error into an empty
args dict and ran the tool with no input at all -- worse than not reporting
it. It now routes through `parse_tool_call_args` like the executors do, so
the StepExecutor/planning path reports INVALID_INPUT and returns instead of
executing. That also removes a duplicated inline parse.

The ReAct path returned a `ToolUsageError` message as an ordinary result
without reporting it, so a malformed call there was invisible while the
equivalent native failure was recorded. Now reported as INVALID_INPUT too.

`Agent.kickoff` opened a collector but no longer reset the agent-level list,
so `last_tool_failures` grew across kickoffs. Reset restored, matching task
execution.

`ToolFailureReason.BLOCKED_BY_HOOK` was declared and never produced. Rather
than start reporting hook blocks as failures, the member is removed: a block
is a deliberate decision by the hook author, and treating it as a failure
would make `raise` abort on an intentional veto. Added a guard test that every
remaining reason is actually produced somewhere, so a dead member cannot
reappear -- the same smell that flagged INVALID_INPUT last round.

Also switched the deprecation guard test to a single import style.

Testing: 6 further tests, 95 total, including that the tool does not run when
its args fail to parse. Full suite matches baseline at 377 pre-existing
failures; mypy clean.

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

* fix(tools): merge failures across kickoff guardrail retries, cancel siblings

Kickoff guardrail retries discarded the blocked attempt's failures. Each
retry calls `_execute_and_build_output`, which opens a fresh collector and
builds a new output, so a run could report zero failures having demonstrably
failed one -- the same bug already fixed on the task guardrail path, which
merges. Now merged there too. Verified the test fails without the fix.

Under `raise`, one parallel native tool aborting left its siblings running:
the pool waited for them and pending ones still started. It now shuts the
pool down with `cancel_futures=True` so a not-yet-started sibling never runs.
Threads already in flight cannot be interrupted in Python, so a concurrent
tool may still complete before the abort surfaces; that is noted at the call
site rather than left implied.

Also satisfied CodeQL by materialising the enum in the guard test's loop.

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

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: Lorenze Jay <63378463+lorenzejay@users.noreply.github.com>

docs/edge/en/concepts/tools.mdx

@@ -334,6 +334,126 @@ 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, Crew, 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,
+)
+
+# Or set a baseline once for every agent in the crew.
+crew = Crew(
+    agents=[agent],
+    tasks=[task],
+    tool_failure_policy=ToolFailurePolicy.WARN,
+)
+```
+
+The most specific setting wins: **tool → task → agent → crew → `warn`**. Every
+level defaults to `None`, meaning "inherit from the next one out", so the
+effective default with nothing configured anywhere is `warn`.
+
+### 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 @@ import os
 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 @@ class CrewAIPlatformActionTool(BaseTool):
         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,20 @@ class CrewAIPlatformActionTool(BaseTool):
                         error_message = str(error_info)
                 else:
                     error_message = str(data)
-                return f"API request failed: {error_message}"
+                # A non-2xx here means the upstream app rejected the action
+                # (e.g. Slack's channel_not_found) -- report it, not 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,12 @@ from crewai.skills.loader import load_skills
 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,
+    ToolFailureRecord,
+    merge_tool_failures,
+    tool_failure_collector,
+)
 from crewai.types.callback import SerializableCallable
 from crewai.types.usage_metrics import UsageMetrics
 from crewai.utilities.agent_utils import (
@@ -131,7 +137,9 @@ if TYPE_CHECKING:
     from crewai.utilities.types import LLMMessage
 
 
-_passthrough_exceptions: tuple[type[Exception], ...] = ()
+# Deliberate stops, not transient errors: never swallowed into the
+# max_retry_limit loop.
+_passthrough_exceptions: tuple[type[Exception], ...] = (ToolExecutionFailedError,)
 
 _EXECUTOR_CLASS_MAP: dict[str, type] = {
     "CrewAgentExecutor": CrewAgentExecutor,
@@ -550,6 +558,8 @@ class Agent(BaseAgent):
 
         self._inject_date_to_task(task)
 
+        self.reset_tool_failures()
+
         if self.tools_handler:
             self.tools_handler.last_used_tool = None
 
@@ -914,6 +924,11 @@ class Agent(BaseAgent):
                 raise TimeoutError(
                     f"Task '{task.description}' execution timed out after {timeout} seconds. Consider increasing max_execution_time or optimizing the task."
                 ) from e
+            except _passthrough_exceptions:
+                # Wrapping a deliberate stop in RuntimeError would hide it from
+                # _check_execution_error and trigger the retry loop instead.
+                future.cancel()
+                raise
             except Exception as e:
                 future.cancel()
                 raise RuntimeError(f"Task execution failed: {e!s}") from e
@@ -1454,6 +1469,8 @@ class Agent(BaseAgent):
         Returns:
             Tuple of (executor, inputs, agent_info, parsed_tools) ready for execution.
         """
+        self.reset_tool_failures()
+
         if self.tools_handler:
             self.tools_handler.last_used_tool = None
 
@@ -1786,6 +1803,7 @@ class Agent(BaseAgent):
         executor: AgentExecutor,
         response_format: type[Any] | None = None,
         usage_baseline: UsageMetrics | None = None,
+        kickoff_failures: list[ToolFailureRecord] | None = None,
     ) -> LiteAgentOutput:
         """Build a LiteAgentOutput from an executor result dict.
 
@@ -1866,6 +1884,7 @@ class Agent(BaseAgent):
             todos=todo_results,
             replan_count=executor.state.replan_count,
             last_replan_reason=executor.state.last_replan_reason,
+            tool_failures=list(kickoff_failures or []),
         )
 
     def _execute_and_build_output(
@@ -1876,9 +1895,10 @@ class Agent(BaseAgent):
         usage_baseline: UsageMetrics | None = None,
     ) -> LiteAgentOutput:
         """Execute the agent synchronously and build the output object."""
-        result = cast(dict[str, Any], executor.invoke(inputs))
+        with tool_failure_collector() as kickoff_failures:
+            result = cast(dict[str, Any], executor.invoke(inputs))
         return self._build_output_from_result(
-            result, executor, response_format, usage_baseline
+            result, executor, response_format, usage_baseline, kickoff_failures
         )
 
     async def _execute_and_build_output_async(
@@ -1889,9 +1909,10 @@ class Agent(BaseAgent):
         usage_baseline: UsageMetrics | None = None,
     ) -> LiteAgentOutput:
         """Execute the agent asynchronously and build the output object."""
-        result = await executor.invoke_async(inputs)
+        with tool_failure_collector() as kickoff_failures:
+            result = await executor.invoke_async(inputs)
         return self._build_output_from_result(
-            result, executor, response_format, usage_baseline
+            result, executor, response_format, usage_baseline, kickoff_failures
         )
 
     def _process_kickoff_guardrail(
@@ -1951,9 +1972,15 @@ class Agent(BaseAgent):
                 role="user",
             )
 
-            output = self._execute_and_build_output(
+            retried = self._execute_and_build_output(
                 executor, inputs, response_format, usage_baseline
             )
+            # The retry opens its own collector, so carry the blocked attempt's
+            # failures forward or they vanish from the final output.
+            retried.tool_failures = merge_tool_failures(
+                output.tool_failures, retried.tool_failures
+            )
+            output = retried
 
             return self._process_kickoff_guardrail(
                 output=output,

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

@@ -44,6 +44,11 @@ from crewai.security.security_config import SecurityConfig
 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,
+    collect_tool_failures,
+)
 from crewai.types.callback import SerializableCallable
 from crewai.utilities.config import process_config
 from crewai.utilities.i18n import I18N, get_i18n
@@ -264,6 +269,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 +304,15 @@ 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 | None = Field(
+        default=None,
+        description=(
+            "How to react when a tool completes but reports that it failed. "
+            "'ignore' records nothing; 'warn' records and emits "
+            "ToolFailureDetectedEvent; 'raise' also aborts with "
+            "ToolExecutionFailedError. None inherits from the crew, then 'warn'."
+        ),
+    )
     agent_executor: Annotated[
         SerializeAsAny[BaseAgentExecutor] | None,
         BeforeValidator(_validate_executor_ref),
@@ -652,6 +667,22 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta):
         ]
         return md5("|".join(source).encode(), usedforsecurity=False).hexdigest()
 
+    @property
+    def last_tool_failures(self) -> list[ToolFailureRecord]:
+        """Tool failures recorded during the most recent execution.
+
+        Inside an execution this reports that execution's records, so a
+        shared agent running concurrent tasks does not leak between them.
+        Outside one it reports the most recent execution, like
+        ``last_messages``. Empty when nothing failed or the policy is
+        ``ignore``. Returns a copy.
+        """
+        return collect_tool_failures(self)
+
+    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/cache/cache_handler.py

@@ -23,6 +23,10 @@ class CacheHandler(BaseModel):
     def add(self, tool: str, input: str, output: Any) -> None:
         """Add a tool result to the cache.
 
+        Declared failures are never stored: replaying one would make a
+        transient error permanent for the rest of the run, and every later hit
+        would re-report a call that did not run.
+
         Args:
             tool: Name of the tool.
             input: Input string used for the tool.
@@ -31,6 +35,11 @@ class CacheHandler(BaseModel):
         Notes:
             - TODO: Rename 'input' parameter to avoid shadowing builtin.
         """
+        from crewai.tools.tool_failure import ToolFailure
+
+        if isinstance(output, ToolFailure):
+            return
+
         with self._lock.w_locked():
             self._cache[f"{tool}-{input}"] = output
 

lib/crewai/src/crewai/agents/step_executor.py

@@ -28,6 +28,7 @@ from crewai.events.types.tool_usage_events import (
     ToolUsageFinishedEvent,
     ToolUsageStartedEvent,
 )
+from crewai.tools.tool_failure import ToolExecutionFailedError
 from crewai.utilities.agent_utils import (
     build_text_tool_calling_fallback_message,
     build_tool_calls_assistant_message,
@@ -180,6 +181,11 @@ class StepExecutor:
                 tool_calls_made=tool_calls_made,
                 execution_time=elapsed,
             )
+        except ToolExecutionFailedError:
+            # A deliberate stop: StepResult(success=False) would let the plan
+            # carry on.
+            raise
+
         except Exception as e:
             if self._use_native_tools and is_native_tool_calling_unsupported_error(e):
                 try:
@@ -218,6 +224,11 @@ class StepExecutor:
                         tool_calls_made=tool_calls_made,
                         execution_time=elapsed,
                     )
+                except ToolExecutionFailedError:
+                    # Same as the outer handler, reached via the text-tooling
+                    # fallback.
+                    raise
+
                 except Exception as fallback_error:
                     e = fallback_error
 

lib/crewai/src/crewai/crew.py

@@ -116,6 +116,7 @@ from crewai.tasks.task_output import TaskOutput
 from crewai.tools.agent_tools.agent_tools import AgentTools
 from crewai.tools.agent_tools.read_file_tool import ReadFileTool
 from crewai.tools.base_tool import BaseTool
+from crewai.tools.tool_failure import ToolFailurePolicy
 from crewai.types.callback import SerializableCallable
 from crewai.types.streaming import CrewStreamingOutput
 from crewai.types.usage_metrics import UsageMetrics
@@ -231,6 +232,13 @@ class Crew(FlowTrackable, BaseModel):
             "unless they set a cache_function that prevents caching."
         ),
     )
+    tool_failure_policy: ToolFailurePolicy | None = Field(
+        default=None,
+        description=(
+            "Baseline tool_failure_policy for every agent in this crew. None "
+            "means 'warn'. Agents, tasks and tools may override it."
+        ),
+    )
     tasks: list[Task] = Field(default_factory=list)
     agents: Annotated[
         list[BaseAgent],

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

@@ -7,6 +7,7 @@ from pydantic import BaseModel, Field
 
 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,20 @@ 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 successfully with a non-empty list -- agents narrate a
+        failed step and carry on. Check it 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.

lib/crewai/src/crewai/events/__init__.py

@@ -148,6 +148,7 @@ if TYPE_CHECKING:
     )
     from crewai.events.types.tool_usage_events import (
         ToolExecutionErrorEvent,
+        ToolFailureDetectedEvent,
         ToolSelectionErrorEvent,
         ToolUsageErrorEvent,
         ToolUsageEvent,
@@ -253,6 +254,7 @@ _LAZY_EVENT_MAPPING: dict[str, str] = {
     "TaskFailedEvent": "crewai.events.types.task_events",
     "TaskStartedEvent": "crewai.events.types.task_events",
     "ToolExecutionErrorEvent": "crewai.events.types.tool_usage_events",
+    "ToolFailureDetectedEvent": "crewai.events.types.tool_usage_events",
     "ToolSelectionErrorEvent": "crewai.events.types.tool_usage_events",
     "ToolUsageErrorEvent": "crewai.events.types.tool_usage_events",
     "ToolUsageEvent": "crewai.events.types.tool_usage_events",
@@ -387,6 +389,7 @@ __all__ = [
     "TaskFailedEvent",
     "TaskStartedEvent",
     "ToolExecutionErrorEvent",
+    "ToolFailureDetectedEvent",
     "ToolSelectionErrorEvent",
     "ToolUsageErrorEvent",
     "ToolUsageEvent",

lib/crewai/src/crewai/events/event_listener.py

@@ -114,6 +114,7 @@ from crewai.events.types.task_events import (
     TaskStartedEvent,
 )
 from crewai.events.types.tool_usage_events import (
+    ToolFailureDetectedEvent,
     ToolUsageErrorEvent,
     ToolUsageFinishedEvent,
     ToolUsageStartedEvent,
@@ -424,6 +425,8 @@ class EventListener(BaseEventListener):
 
         @crewai_event_bus.on(ToolUsageFinishedEvent)
         def on_tool_usage_finished(source: Any, event: ToolUsageFinishedEvent) -> None:
+            if not self.formatter.should_render_success_panel(event.failure):
+                return
             if isinstance(source, LLM):
                 self.formatter.handle_llm_tool_usage_finished(
                     event.tool_name,
@@ -449,6 +452,18 @@ class EventListener(BaseEventListener):
                     event.run_attempts,
                 )
 
+        @crewai_event_bus.on(ToolFailureDetectedEvent)
+        def on_tool_failure_detected(
+            source: Any, event: ToolFailureDetectedEvent
+        ) -> None:
+            if not self.formatter.should_render_failure_panel(event.failure):
+                return
+            self.formatter.handle_tool_failure_detected(
+                event.tool_name,
+                event.failure,
+                event.policy,
+            )
+
         @crewai_event_bus.on(LLMCallStartedEvent)
         def on_llm_call_started(_: Any, event: LLMCallStartedEvent) -> None:
             self.text_stream = StringIO()

lib/crewai/src/crewai/events/event_types.py

@@ -118,6 +118,7 @@ from crewai.events.types.task_events import (
     TaskStartedEvent,
 )
 from crewai.events.types.tool_usage_events import (
+    ToolFailureDetectedEvent,
     ToolUsageErrorEvent,
     ToolUsageFinishedEvent,
     ToolUsageStartedEvent,
@@ -178,6 +179,7 @@ EventTypes = (
     | AgentExecutionErrorEvent
     | ToolUsageFinishedEvent
     | ToolUsageErrorEvent
+    | ToolFailureDetectedEvent
     | ToolUsageStartedEvent
     | LLMCallStartedEvent
     | LLMCallCompletedEvent

lib/crewai/src/crewai/events/listeners/tracing/trace_listener.py

@@ -127,6 +127,7 @@ from crewai.events.types.task_events import (
     TaskStartedEvent,
 )
 from crewai.events.types.tool_usage_events import (
+    ToolFailureDetectedEvent,
     ToolUsageErrorEvent,
     ToolUsageFinishedEvent,
     ToolUsageStartedEvent,
@@ -415,6 +416,12 @@ class TraceCollectionListener(BaseEventListener):
         def on_tool_error(source: Any, event: ToolUsageErrorEvent) -> None:
             self._handle_action_event("tool_usage_error", source, event)
 
+        @event_bus.on(ToolFailureDetectedEvent)
+        def on_tool_failure_detected(
+            source: Any, event: ToolFailureDetectedEvent
+        ) -> None:
+            self._handle_action_event("tool_failure_detected", source, event)
+
         @event_bus.on(MemoryQueryStartedEvent)
         def on_memory_query_started(
             source: Any, event: MemoryQueryStartedEvent

lib/crewai/src/crewai/events/types/tool_usage_events.py

@@ -5,6 +5,7 @@ from typing import Any, Literal
 from pydantic import ConfigDict
 
 from crewai.events.base_events import BaseEvent
+from crewai.tools.tool_failure import ToolFailure, ToolFailurePolicy
 
 
 class ToolUsageEvent(BaseEvent):
@@ -66,6 +67,11 @@ class ToolUsageFinishedEvent(ToolUsageEvent):
     finished_at: datetime
     from_cache: bool = False
     output: Any
+    failure: ToolFailure | None = None
+    """Set when the tool ran but reported it did not succeed.
+
+    Lets a trace UI mark the call failed without correlating a second event.
+    """
     type: Literal["tool_usage_finished"] = "tool_usage_finished"
 
 
@@ -76,6 +82,20 @@ class ToolUsageErrorEvent(ToolUsageEvent):
     type: Literal["tool_usage_error"] = "tool_usage_error"
 
 
+class ToolFailureDetectedEvent(ToolUsageEvent):
+    """Event emitted when a tool completed but reported that it failed.
+
+    Distinct from :class:`ToolUsageErrorEvent`, which covers a tool *raising*.
+    This is the quieter case: the call returned normally and says the work was
+    not done. Emitted for every policy except ``IGNORE``, and before a
+    ``RAISE`` aborts, so subscribers see it even on an aborting run.
+    """
+
+    failure: ToolFailure
+    policy: ToolFailurePolicy
+    type: Literal["tool_failure_detected"] = "tool_failure_detected"
+
+
 class ToolValidateInputErrorEvent(ToolUsageEvent):
     """Event emitted when a tool input validation encounters an error"""
 

lib/crewai/src/crewai/events/utils/console_formatter.py

@@ -12,6 +12,7 @@ from rich.live import Live
 from rich.panel import Panel
 from rich.text import Text
 
+from crewai.tools.tool_failure import ToolFailureReason
 from crewai.version import is_current_version_yanked, is_newer_version_available
 
 
@@ -492,6 +493,55 @@ To enable tracing, do any one of these:
             content, f"✅ Tool Execution Completed (#{iteration})", "green"
         )
 
+    @staticmethod
+    def should_render_success_panel(failure: Any) -> bool:
+        """Whether a finished tool call should print the green panel.
+
+        A failed call must not read as successful, so the red panel replaces it.
+        """
+        return failure is None
+
+    @staticmethod
+    def should_render_failure_panel(failure: Any) -> bool:
+        """Whether a reported failure should print its own red panel.
+
+        A tool that *raised* already printed one via ``ToolUsageErrorEvent``,
+        so only the duplicate console output is skipped -- not the event.
+        """
+        return getattr(failure, "reason", None) is not ToolFailureReason.EXCEPTION
+
+    def handle_tool_failure_detected(
+        self,
+        tool_name: str,
+        failure: Any,
+        policy: Any,
+    ) -> None:
+        """Render a tool that ran but reported it did not succeed.
+
+        The case that used to print as a green "Completed" panel.
+        """
+        if not self.verbose:
+            return
+
+        with self._tool_counts_lock:
+            iteration = self.tool_usage_counts.get(tool_name, 1)
+
+        content = Text()
+        content.append("Tool Reported Failure\n", style="red bold")
+        content.append("Tool: ", style="white")
+        content.append(f"{tool_name}\n", style="red bold")
+        content.append("Reason: ", style="white")
+        content.append(f"{getattr(failure, 'reason', 'unknown')}\n", style="red")
+        if getattr(failure, "code", None):
+            content.append("Code: ", style="white")
+            content.append(f"{failure.code}\n", style="red")
+        content.append("Message: ", style="white")
+        content.append(f"{getattr(failure, 'message', failure)}\n", style="red")
+        content.append("Policy: ", style="white")
+        content.append(f"{getattr(policy, 'value', policy)}\n", style="red")
+
+        self.print_panel(content, f"⚠️ Tool Failure (#{iteration})", "red")
+
     def handle_tool_usage_error(
         self,
         tool_name: str,

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

@@ -73,6 +73,15 @@ from crewai.hooks.types import (
 )
 from crewai.tools.base_tool import BaseTool
 from crewai.tools.structured_tool import CrewStructuredTool
+from crewai.tools.tool_failure import (
+    ToolExecutionFailedError,
+    ToolFailure,
+    ToolFailureReason,
+    detect_tool_failure,
+    failure_from_exception,
+    handle_tool_failure,
+    reportable_failure,
+)
 from crewai.utilities.agent_utils import (
     _llm_stop_words_applied,
     build_text_tool_calling_fallback_message,
@@ -1634,6 +1643,11 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
                 function_calling_llm=self.function_calling_llm,
                 crew=self.crew,
             )
+        except ToolExecutionFailedError:
+            # A deliberate stop: the generic handler below would feed it back
+            # to the LLM as a recoverable observation.
+            raise
+
         except Exception as e:
             if self.agent and self.agent.verbose:
                 PRINTER.print(content=f"Error in tool execution: {e}", color="red")
@@ -1753,6 +1767,15 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
                     idx = future_to_idx[future]
                     try:
                         ordered_results[idx] = future.result()
+                    except ToolExecutionFailedError:
+                        # A deliberate stop: folding it into a tool result would
+                        # let the remaining parallel calls carry on. Cancel the
+                        # siblings that have not started so they never run.
+                        # Ones already in flight cannot be interrupted -- Python
+                        # threads are not cancellable -- so a concurrent tool may
+                        # still complete before the abort surfaces.
+                        pool.shutdown(wait=False, cancel_futures=True)
+                        raise
                     except Exception as e:
                         tool_call = runnable_tool_calls[idx]
                         info = extract_tool_call_info(tool_call)
@@ -1799,6 +1822,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
                     original_tool
                     and hasattr(original_tool, "result_as_answer")
                     and original_tool.result_as_answer
+                    # A failed tool must not become the final answer.
+                    and execution_result.get("tool_failure") is None
                 ):
                     self.state.current_answer = AgentFinish(
                         thought="Tool result is the final answer",
@@ -1837,6 +1862,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
                 original_tool
                 and hasattr(original_tool, "result_as_answer")
                 and original_tool.result_as_answer
+                # A failed tool must not become the final answer.
+                and execution_result.get("tool_failure") is None
             ):
                 # Set the result as the final answer
                 self.state.current_answer = AgentFinish(
@@ -1904,6 +1931,14 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
         # Parse arguments
         parsed_args, parse_error = parse_tool_call_args(func_args, func_name, call_id)
         if parse_error is not None:
+            handle_tool_failure(
+                parse_error["tool_failure"],
+                tool_name=func_name,
+                tool_args=func_args,
+                agent=self.agent,
+                task=self.task,
+                crew=self.crew,
+            )
             return parse_error
         args_dict: dict[str, Any] = parsed_args or {}
 
@@ -1949,6 +1984,7 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
         from_cache = False
         result = "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(
@@ -1957,6 +1993,7 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
             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
 
         # Emit tool usage started event
@@ -1988,6 +2025,9 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
         if hook_blocked:
             result = f"Tool execution blocked by hook. Tool: {func_name}"
             raw_tool_result = result
+            # The blocked message replaces any cached result, so a cached
+            # failure must not be attributed to this call.
+            tool_failure = None
         elif not from_cache and not max_usage_reached and output_tool is not None:
             if func_name in self._available_functions:
                 try:
@@ -2010,9 +2050,11 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
                     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()
                     # Emit tool usage error event
@@ -2028,6 +2070,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
                         ),
                     )
                     error_event_emitted = True
+            else:
+                tool_failure = self._unknown_tool_failure(func_name, result)
         elif max_usage_reached:
             # Return error message when max usage limit is reached
             if original_tool:
@@ -2035,6 +2079,11 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
             else:
                 result = f"Tool '{func_name}' has reached its maximum usage limit and cannot be used anymore."
             raw_tool_result = result
+            tool_failure = ToolFailure(
+                message=result, reason=ToolFailureReason.USAGE_LIMIT
+            )
+        elif not from_cache:
+            tool_failure = self._unknown_tool_failure(func_name, result)
 
         # Execute after_tool_call hooks (even if blocked, to allow logging/monitoring)
         after_hook_context = ToolCallHookContext(
@@ -2063,17 +2112,50 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
                     agent_key=agent_key,
                     started_at=started_at,
                     finished_at=datetime.now(),
+                    failure=reportable_failure(
+                        tool_failure,
+                        tool=structured_tool,
+                        agent=self.agent,
+                        task=self.task,
+                        crew=self.crew,
+                    ),
                 ),
             )
 
+        # After the finished event, so subscribers see the full lifecycle even
+        # when the policy aborts.
+        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,
             "result": result,
             "from_cache": from_cache,
             "original_tool": original_tool,
+            "tool_failure": tool_failure,
         }
 
+    @staticmethod
+    def _unknown_tool_failure(func_name: str, result: str) -> ToolFailure:
+        """Build the failure for a tool the model asked for but we lack.
+
+        The ReAct path reports this, so the native path must too.
+        """
+        return ToolFailure(
+            message=result,
+            reason=ToolFailureReason.UNKNOWN_TOOL,
+            code=func_name,
+        )
+
     def _extract_tool_name(self, tool_call: Any) -> str:
         """Extract tool name from various tool call formats."""
         if hasattr(tool_call, "function"):

lib/crewai/src/crewai/hooks/tool_hooks.py

@@ -23,6 +23,7 @@ if TYPE_CHECKING:
     from crewai.agent import Agent
     from crewai.agents.agent_builder.base_agent import BaseAgent
     from crewai.crew import Crew
+    from crewai.lite_agent import LiteAgent
     from crewai.task import Task
     from crewai.tools.structured_tool import CrewStructuredTool
 
@@ -55,7 +56,7 @@ class ToolCallHookContext:
         tool_name: str,
         tool_input: dict[str, Any],
         tool: CrewStructuredTool,
-        agent: Agent | BaseAgent | None = None,
+        agent: Agent | BaseAgent | LiteAgent | None = None,
         task: Task | None = None,
         crew: Crew | None = None,
         tool_result: str | None = None,

lib/crewai/src/crewai/lite_agent.py

@@ -72,6 +72,12 @@ from crewai.llm import LLM
 from crewai.llms.base_llm import BaseLLM
 from crewai.tools.base_tool import BaseTool
 from crewai.tools.structured_tool import CrewStructuredTool
+from crewai.tools.tool_failure import (
+    ToolExecutionFailedError,
+    ToolFailurePolicy,
+    ToolFailureRecord,
+    tool_failure_collector,
+)
 from crewai.utilities.agent_utils import (
     enforce_rpm_limit,
     format_message_for_llm,
@@ -222,6 +228,14 @@ class LiteAgent(FlowTrackable, BaseModel):
     max_iterations: int = Field(
         default=15, description="Maximum number of iterations for tool usage"
     )
+    tool_failure_policy: ToolFailurePolicy | None = Field(
+        default=None,
+        description=(
+            "How to react when a tool runs to completion but reports that it "
+            "failed. None falls back to 'warn'. See "
+            "BaseAgent.tool_failure_policy."
+        ),
+    )
     max_execution_time: int | None = Field(
         default=None, description=". Maximum execution time in seconds"
     )
@@ -289,6 +303,8 @@ class LiteAgent(FlowTrackable, BaseModel):
     _key: str = PrivateAttr(default_factory=lambda: str(uuid.uuid4()))
     _messages: list[LLMMessage] = PrivateAttr(default_factory=list)
     _iterations: int = PrivateAttr(default=0)
+    _tool_failures: list[ToolFailureRecord] = PrivateAttr(default_factory=list)
+    _kickoff_failures: list[ToolFailureRecord] = PrivateAttr(default_factory=list)
     _guardrail: GuardrailCallable | None = PrivateAttr(default=None)
     _guardrail_retry_count: int = PrivateAttr(default=0)
     _callbacks: list[TokenCalcHandler] = PrivateAttr(default_factory=list)
@@ -450,6 +466,14 @@ class LiteAgent(FlowTrackable, BaseModel):
         """Return the original role for compatibility with tool interfaces."""
         return self.role
 
+    @property
+    def last_tool_failures(self) -> list[ToolFailureRecord]:
+        """Tool failures recorded during the most recent kickoff.
+
+        Mirrors ``BaseAgent.last_tool_failures`` so the shared helper works here.
+        """
+        return list(self._tool_failures)
+
     @property
     def before_llm_call_hooks(
         self,
@@ -519,15 +543,34 @@ class LiteAgent(FlowTrackable, BaseModel):
         try:
             self._iterations = 0
             self.tools_results = []
+            self._tool_failures = []
 
             self._messages = self._format_messages(
                 messages, response_format=response_format, input_files=input_files
             )
             self._inject_memory_context()
 
-            return self._execute_core(
-                agent_info=agent_info, response_format=response_format
+            with tool_failure_collector() as kickoff_failures:
+                self._kickoff_failures = kickoff_failures
+                return self._execute_core(
+                    agent_info=agent_info, response_format=response_format
+                )
+
+        except ToolExecutionFailedError as e:
+            # A deliberate stop, not a defect: no bug-report prompt.
+            if self.verbose:
+                PRINTER.print(
+                    content=f"Agent stopped: {e}",
+                    color="red",
+                )
+            crewai_event_bus.emit(
+                self,
+                event=LiteAgentExecutionErrorEvent(
+                    agent_info=agent_info,
+                    error=str(e),
+                ),
             )
+            raise
 
         except Exception as e:
             if self.verbose:
@@ -691,6 +734,9 @@ class LiteAgent(FlowTrackable, BaseModel):
             agent_role=self.role,
             usage_metrics=usage_metrics.model_dump() if usage_metrics else None,
             messages=self._messages,
+            # Read from whichever agent the executor was given, or the records
+            # go missing: original_agent under kickoff, self when standalone.
+            tool_failures=list(self._kickoff_failures),
         )
 
         if self._guardrail is not None:
@@ -916,7 +962,9 @@ class LiteAgent(FlowTrackable, BaseModel):
                             tools=self._parsed_tools,
                             agent_key=self.key,
                             agent_role=self.role,
-                            agent=self.original_agent,
+                            # Fall back to self so a standalone LiteAgent still
+                            # resolves a policy and records failures.
+                            agent=self.original_agent or self,
                             crew=None,
                         )
                     except Exception as e:
@@ -929,6 +977,10 @@ class LiteAgent(FlowTrackable, BaseModel):
                     )
 
                 self._append_message(formatted_answer.text, role="assistant")
+            except ToolExecutionFailedError:
+                # tool_failure_policy="raise" asked for the run to stop.
+                raise
+
             except OutputParserError as e:
                 if self.verbose:
                     PRINTER.print(

lib/crewai/src/crewai/lite_agent_output.py

@@ -6,6 +6,7 @@ from typing import Any
 
 from pydantic import BaseModel, Field
 
+from crewai.tools.tool_failure import ToolFailureRecord
 from crewai.types.usage_metrics import UsageMetrics
 from crewai.utilities.planning_types import TodoItem
 from crewai.utilities.types import LLMMessage
@@ -50,6 +51,17 @@ class LiteAgentOutput(BaseModel):
     messages: list[LLMMessage] = Field(
         description="Messages of the agent", default_factory=list
     )
+    tool_failures: list[ToolFailureRecord] = Field(
+        default_factory=list,
+        description=(
+            "Tools that ran but reported they did not succeed. Empty under 'ignore'."
+        ),
+    )
+
+    @property
+    def has_tool_failures(self) -> bool:
+        """Whether any tool reported a failure while producing this output."""
+        return bool(self.tool_failures)
 
     plan: str | None = Field(
         default=None, description="The execution plan that was generated, if any"

lib/crewai/src/crewai/mcp/client.py

@@ -430,7 +430,27 @@ class MCPClient:
             arguments: Tool arguments.
 
         Returns:
-            Tool execution result.
+            Tool execution result content. The ``isError`` flag is dropped;
+            use :meth:`call_tool_result` when the caller needs it.
+        """
+        return (await self.call_tool_result(tool_name, arguments)).content
+
+    async def call_tool_result(
+        self, tool_name: str, arguments: dict[str, Any] | None = None
+    ) -> _MCPToolResult:
+        """Call a tool and return its content together with the ``isError`` flag.
+
+        MCP servers report a failed tool as a *successful* JSON-RPC response
+        carrying ``isError: true``. Callers that only take the content cannot
+        tell that apart from a normal result, which is how a failed step ends
+        up looking like a successful one.
+
+        Args:
+            tool_name: Name of the tool to call.
+            arguments: Tool arguments.
+
+        Returns:
+            The content string plus whether the server flagged it as an error.
         """
         if not self.connected:
             await self.connect()
@@ -492,7 +512,7 @@ class MCPClient:
                     ),
                 )
 
-            return tool_result.content
+            return tool_result
         except Exception as e:
             failed_at = datetime.now()
             error_type = (

lib/crewai/src/crewai/task.py

@@ -52,6 +52,12 @@ from crewai.security import Fingerprint, SecurityConfig
 from crewai.tasks.output_format import OutputFormat
 from crewai.tasks.task_output import TaskOutput
 from crewai.tools.base_tool import BaseTool
+from crewai.tools.tool_failure import (
+    ToolFailurePolicy,
+    ToolFailureRecord,
+    merge_tool_failures,
+    tool_failure_collector,
+)
 from crewai.utilities.config import process_config
 from crewai.utilities.constants import NOT_SPECIFIED, _NotSpecified
 from crewai.utilities.converter import (
@@ -274,6 +280,13 @@ class Task(BaseModel):
         default=3, description="Maximum number of retries when guardrail fails"
     )
     retry_count: int = Field(default=0, description="Current number of retries")
+    tool_failure_policy: ToolFailurePolicy | None = Field(
+        default=None,
+        description=(
+            "Overrides the agent's tool_failure_policy for this task only. "
+            "None inherits."
+        ),
+    )
     start_time: datetime.datetime | None = Field(
         default=None, description="Start time of the task execution"
     )
@@ -677,11 +690,12 @@ class Task(BaseModel):
             dispatch(InterceptionPoint.PRE_STEP, pre_step_ctx)
             context = pre_step_ctx.payload
 
-            result = await agent.aexecute_task(
-                task=self,
-                context=context,
-                tools=tools,
-            )
+            with tool_failure_collector() as execution_failures:
+                result = await agent.aexecute_task(
+                    task=self,
+                    context=context,
+                    tools=tools,
+                )
 
             self._post_agent_execution(agent)
 
@@ -713,6 +727,7 @@ class Task(BaseModel):
                 agent=agent.role,
                 output_format=self._get_output_format(),
                 messages=agent.last_messages,  # type: ignore[attr-defined]
+                tool_failures=list(execution_failures),
             )
 
             if self._guardrails:
@@ -831,11 +846,12 @@ class Task(BaseModel):
             dispatch(InterceptionPoint.PRE_STEP, pre_step_ctx)
             context = pre_step_ctx.payload
 
-            result = agent.execute_task(
-                task=self,
-                context=context,
-                tools=tools,
-            )
+            with tool_failure_collector() as execution_failures:
+                result = agent.execute_task(
+                    task=self,
+                    context=context,
+                    tools=tools,
+                )
 
             self._post_agent_execution(agent)
 
@@ -867,6 +883,7 @@ class Task(BaseModel):
                 agent=agent.role,
                 output_format=self._get_output_format(),
                 messages=agent.last_messages,  # type: ignore[attr-defined]
+                tool_failures=list(execution_failures),
             )
 
             if self._guardrails:
@@ -1319,6 +1336,10 @@ Follow these guidelines:
 
         max_attempts = self.guardrail_max_retries + 1
 
+        # Each retry resets the agent's failure list, so accumulate to keep
+        # failures from blocked attempts on the final output.
+        accumulated_failures: list[ToolFailureRecord] = list(task_output.tool_failures)
+
         for attempt in range(max_attempts):
             guardrail_result = process_guardrail(
                 output=task_output,
@@ -1343,7 +1364,12 @@ Follow these guidelines:
                     task_output.pydantic = pydantic_output
                     task_output.json_dict = json_output
                 elif isinstance(guardrail_result.result, TaskOutput):
+                    # A guardrail may return a whole new output; carry the
+                    # accumulated failures over or earlier attempts vanish.
                     task_output = guardrail_result.result
+                    task_output.tool_failures = merge_tool_failures(
+                        accumulated_failures, task_output.tool_failures
+                    )
 
                 return task_output
 
@@ -1374,11 +1400,12 @@ Follow these guidelines:
                     content=f"Guardrail {guardrail_index if guardrail_index is not None else ''} blocked (attempt {attempt + 1}/{max_attempts}), retrying due to: {guardrail_result.error}\n",
                     color="yellow",
                 )
-            result = agent.execute_task(
-                task=self,
-                context=context,
-                tools=tools,
-            )
+            with tool_failure_collector() as retry_failures:
+                result = agent.execute_task(
+                    task=self,
+                    context=context,
+                    tools=tools,
+                )
 
             if isinstance(result, BaseModel):
                 raw = result.model_dump_json()
@@ -1405,7 +1432,9 @@ Follow these guidelines:
                 agent=agent.role,
                 output_format=self._get_output_format(),
                 messages=agent.last_messages,  # type: ignore[attr-defined]
+                tool_failures=merge_tool_failures(accumulated_failures, retry_failures),
             )
+            accumulated_failures = list(task_output.tool_failures)
 
         return task_output
 
@@ -1428,6 +1457,10 @@ Follow these guidelines:
 
         max_attempts = self.guardrail_max_retries + 1
 
+        # Each retry resets the agent's failure list, so accumulate to keep
+        # failures from blocked attempts on the final output.
+        accumulated_failures: list[ToolFailureRecord] = list(task_output.tool_failures)
+
         for attempt in range(max_attempts):
             guardrail_result = process_guardrail(
                 output=task_output,
@@ -1452,7 +1485,12 @@ Follow these guidelines:
                     task_output.pydantic = pydantic_output
                     task_output.json_dict = json_output
                 elif isinstance(guardrail_result.result, TaskOutput):
+                    # A guardrail may return a whole new output; carry the
+                    # accumulated failures over or earlier attempts vanish.
                     task_output = guardrail_result.result
+                    task_output.tool_failures = merge_tool_failures(
+                        accumulated_failures, task_output.tool_failures
+                    )
 
                 return task_output
 
@@ -1483,11 +1521,12 @@ Follow these guidelines:
                     content=f"Guardrail {guardrail_index if guardrail_index is not None else ''} blocked (attempt {attempt + 1}/{max_attempts}), retrying due to: {guardrail_result.error}\n",
                     color="yellow",
                 )
-            result = await agent.aexecute_task(
-                task=self,
-                context=context,
-                tools=tools,
-            )
+            with tool_failure_collector() as retry_failures:
+                result = await agent.aexecute_task(
+                    task=self,
+                    context=context,
+                    tools=tools,
+                )
 
             if isinstance(result, BaseModel):
                 raw = result.model_dump_json()
@@ -1514,6 +1553,8 @@ Follow these guidelines:
                 agent=agent.role,
                 output_format=self._get_output_format(),
                 messages=agent.last_messages,  # type: ignore[attr-defined]
+                tool_failures=merge_tool_failures(accumulated_failures, retry_failures),
             )
+            accumulated_failures = list(task_output.tool_failures)
 
         return task_output

lib/crewai/src/crewai/tasks/task_output.py

@@ -8,6 +8,7 @@ from typing import Any
 from pydantic import BaseModel, Field, model_validator
 
 from crewai.tasks.output_format import OutputFormat
+from crewai.tools.tool_failure import ToolFailureRecord
 from crewai.utilities.types import LLMMessage
 
 
@@ -46,6 +47,18 @@ class TaskOutput(BaseModel):
     messages: list[LLMMessage] = Field(
         description="Messages of the task", default_factory=list
     )
+    tool_failures: list[ToolFailureRecord] = Field(
+        default_factory=list,
+        description=(
+            "Tools that ran during this task but reported they did not "
+            "succeed, so 'raw' may be incomplete. Empty under 'ignore'."
+        ),
+    )
+
+    @property
+    def has_tool_failures(self) -> bool:
+        """Whether any tool reported a failure while producing this output."""
+        return bool(self.tool_failures)
 
     @model_validator(mode="after")
     def set_summary(self) -> TaskOutput:

lib/crewai/src/crewai/tools/__init__.py

@@ -1,8 +1,20 @@
 from crewai.tools.base_tool import BaseTool, EnvVar, tool
+from crewai.tools.tool_failure import (
+    ToolExecutionFailedError,
+    ToolFailure,
+    ToolFailurePolicy,
+    ToolFailureReason,
+    ToolFailureRecord,
+)
 
 
 __all__ = [
     "BaseTool",
     "EnvVar",
+    "ToolExecutionFailedError",
+    "ToolFailure",
+    "ToolFailurePolicy",
+    "ToolFailureReason",
+    "ToolFailureRecord",
     "tool",
 ]

lib/crewai/src/crewai/tools/base_tool.py

@@ -38,6 +38,7 @@ from crewai.tools.structured_tool import (
     build_schema_hint,
     format_description_for_llm,
 )
+from crewai.tools.tool_failure import ToolFailure, ToolFailurePolicy, ToolFailureReason
 from crewai.types.callback import SerializableCallable, _resolve_dotted_path
 from crewai.utilities.string_utils import sanitize_tool_name
 
@@ -184,6 +185,13 @@ class BaseTool(BaseModel, ABC):
         default=None,
         description="Maximum number of times this tool can be used. None means unlimited usage.",
     )
+    tool_failure_policy: ToolFailurePolicy | None = Field(
+        default=None,
+        description=(
+            "Overrides the agent's and task's tool_failure_policy for this "
+            "tool only. None inherits."
+        ),
+    )
     current_usage_count: int = Field(
         default=0,
         description="Current number of times this tool has been used.",
@@ -291,21 +299,26 @@ class BaseTool(BaseModel, ABC):
                 ) from e
         return kwargs
 
-    def _claim_usage(self) -> str | None:
+    def _claim_usage(self) -> ToolFailure | None:
         """Atomically check max usage and increment the counter.
 
         Returns:
-            None if usage was claimed successfully, or an error message
-            string if the tool has reached its usage limit.
+            None if usage was claimed, otherwise a :class:`ToolFailure`. A
+            structured result rather than a bare string so every execution
+            path records a spent limit, instead of only the ones that
+            recognise the message.
         """
         with self._usage_lock:
             if (
                 self.max_usage_count is not None
                 and self.current_usage_count >= self.max_usage_count
             ):
-                return (
-                    f"Tool '{self.name}' has reached its usage limit of "
-                    f"{self.max_usage_count} times and cannot be used anymore."
+                return ToolFailure(
+                    message=(
+                        f"Tool '{self.name}' has reached its usage limit of "
+                        f"{self.max_usage_count} times and cannot be used anymore."
+                    ),
+                    reason=ToolFailureReason.USAGE_LIMIT,
                 )
             self.current_usage_count += 1
             return None
@@ -402,6 +415,7 @@ class BaseTool(BaseModel, ABC):
             max_usage_count=self.max_usage_count,
             current_usage_count=self.current_usage_count,
             cache_function=self.cache_function,
+            tool_failure_policy=self.tool_failure_policy,
         )
         structured_tool._original_tool = self
         return structured_tool

lib/crewai/src/crewai/tools/mcp_native_tool.py

@@ -11,6 +11,7 @@ import contextvars
 from typing import Any
 
 from crewai.tools import BaseTool
+from crewai.tools.tool_failure import ToolFailure, ToolFailureReason
 
 
 class MCPNativeTool(BaseTool):
@@ -70,14 +71,15 @@ class MCPNativeTool(BaseTool):
         """Get the server name."""
         return self._server_name
 
-    def _run(self, **kwargs: Any) -> str:
+    def _run(self, **kwargs: Any) -> Any:
         """Execute tool using the MCP client session.
 
         Args:
             **kwargs: Arguments to pass to the MCP tool.
 
         Returns:
-            Result from the MCP tool execution.
+            The tool's text result, or a :class:`ToolFailure` when the server
+            answered with ``isError: true``.
         """
         try:
             try:
@@ -98,7 +100,7 @@ class MCPNativeTool(BaseTool):
                 f"Error executing MCP tool {self.original_tool_name}: {e!s}"
             ) from e
 
-    async def _run_async(self, **kwargs: Any) -> str:
+    async def _run_async(self, **kwargs: Any) -> Any:
         """Async implementation of tool execution.
 
         A fresh ``MCPClient`` is created for every invocation so that
@@ -108,16 +110,36 @@ class MCPNativeTool(BaseTool):
             **kwargs: Arguments to pass to the MCP tool.
 
         Returns:
-            Result from the MCP tool execution.
+            The tool's text result, or a :class:`ToolFailure` when the server
+            answered with ``isError: true``.
         """
         client = self._client_factory()
         await client.connect()
 
         try:
-            result = await client.call_tool(self.original_tool_name, kwargs)
+            tool_result = await client.call_tool_result(self.original_tool_name, kwargs)
         finally:
             await client.disconnect()
 
+        content = self._extract_content(tool_result.content)
+
+        if tool_result.is_error:
+            # isError rides on an otherwise successful response; without this
+            # the agent cannot tell it apart from a real result.
+            return ToolFailure(
+                message=content,
+                reason=ToolFailureReason.MCP_ERROR,
+                details={
+                    "server": self._server_name,
+                    "tool": self._original_tool_name,
+                },
+            )
+
+        return content
+
+    @staticmethod
+    def _extract_content(result: Any) -> str:
+        """Flatten an MCP result payload into the text the agent sees."""
         if isinstance(result, str):
             return result
 

lib/crewai/src/crewai/tools/structured_tool.py

@@ -21,6 +21,7 @@ from pydantic import (
 )
 from typing_extensions import Self
 
+from crewai.tools.tool_failure import ToolFailure, ToolFailurePolicy
 from crewai.utilities.logger import Logger
 from crewai.utilities.pydantic_schema_utils import (
     create_model_from_schema,
@@ -56,6 +57,11 @@ def _infer_result_schema_from_callable(
 
 
 def _format_tool_output_for_agent(tool: Any, raw_result: Any) -> str:
+    # Rendered as prose so the agent sees what an error string would have
+    # given it; the structured object is consumed by the policy machinery.
+    if isinstance(raw_result, ToolFailure):
+        return raw_result.as_agent_message()
+
     original_tool = getattr(tool, "_original_tool", None)
     if original_tool is not None:
         return cast(str, original_tool.format_output_for_agent(raw_result))
@@ -205,6 +211,7 @@ class CrewStructuredTool(BaseModel):
     result_as_answer: bool = Field(default=False)
     max_usage_count: int | None = Field(default=None)
     current_usage_count: int = Field(default=0)
+    tool_failure_policy: ToolFailurePolicy | None = Field(default=None)
     cache_function: Any = Field(default=None, exclude=True)
     _logger: Logger = PrivateAttr(default_factory=Logger)
     _original_tool: Any = PrivateAttr(default=None)

lib/crewai/src/crewai/tools/tool_failure.py

@@ -0,0 +1,384 @@
+"""Structured signalling for tools that run but do not succeed.
+
+A tool can complete without raising and still fail: Slack answers ``HTTP 200``
+with ``{"ok": false, ...}``, an MCP server sets ``isError``. The call
+"worked", so the error used to reach the agent as an ordinary string and the
+run was recorded as a success.
+
+A tool declares failure by returning a :class:`ToolFailure`; the policy
+(:class:`ToolFailurePolicy`) decides the reaction. Detection is strictly
+declarative -- nothing here guesses whether a string "looks like" an error.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Generator
+from contextlib import contextmanager
+from contextvars import ContextVar
+from enum import Enum
+import logging
+from typing import TYPE_CHECKING, Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+logger = logging.getLogger(__name__)
+
+
+if TYPE_CHECKING:
+    from crewai.agents.agent_builder.base_agent import BaseAgent
+    from crewai.crew import Crew
+    from crewai.lite_agent import LiteAgent
+    from crewai.task import Task
+
+
+class ToolFailureReason(str, Enum):
+    """Why a tool call is considered unsuccessful."""
+
+    TOOL_REPORTED = "tool_reported"
+    """The tool itself returned a :class:`ToolFailure`."""
+
+    EXCEPTION = "exception"
+    """The tool raised; the framework caught it and fed the text to the agent."""
+
+    MCP_ERROR = "mcp_error"
+    """An MCP server answered with ``isError: true``."""
+
+    USAGE_LIMIT = "usage_limit"
+    """The tool's ``max_usage_count`` was already spent."""
+
+    UNKNOWN_TOOL = "unknown_tool"
+    """The agent asked for a tool that does not exist."""
+
+    INVALID_INPUT = "invalid_input"
+    """Arguments could not be parsed or validated into the tool's schema."""
+
+
+class ToolFailurePolicy(str, Enum):
+    """How an agent reacts when one of its tools reports a failure."""
+
+    IGNORE = "ignore"
+    """Pre-1.16 behavior: the failure is not recorded, emitted, or acted on."""
+
+    WARN = "warn"
+    """Record the failure, emit an event, and keep going. The default."""
+
+    RAISE = "raise"
+    """Record the failure, emit an event, then abort with
+    :class:`ToolExecutionFailedError`."""
+
+
+class ToolFailure(BaseModel):
+    """A tool's own report that it did not do what it was asked.
+
+    Return one from ``_run``/``_arun`` instead of an error string. The agent
+    still sees text via :meth:`as_agent_message`, so model behavior is
+    unchanged -- but the framework now knows the call failed.
+    """
+
+    model_config = ConfigDict(frozen=True)
+
+    message: str = Field(
+        description="Human and LLM readable explanation of what went wrong."
+    )
+    reason: ToolFailureReason = Field(
+        default=ToolFailureReason.TOOL_REPORTED,
+        description="Category of failure, for grouping and filtering.",
+    )
+    code: str | None = Field(
+        default=None,
+        description=(
+            "Machine-readable identifier from the failing system, "
+            "e.g. 'channel_not_found'."
+        ),
+    )
+    retryable: bool = Field(
+        default=False,
+        description="Whether retrying the same call could plausibly succeed.",
+    )
+    details: dict[str, Any] = Field(
+        default_factory=dict,
+        description="Extra structured context the tool wants to preserve.",
+    )
+
+    def as_agent_message(self) -> str:
+        """Render the text the agent sees for this failure."""
+        if self.code:
+            return f"{self.message} (code: {self.code})"
+        return self.message
+
+
+class ToolFailureRecord(BaseModel):
+    """A :class:`ToolFailure` plus the context of the call that produced it.
+
+    Lands on ``TaskOutput.tool_failures`` and on the event bus, so consumers
+    never parse a string to learn that a step failed.
+    """
+
+    model_config = ConfigDict(frozen=True)
+
+    tool_name: str = Field(description="Name of the tool that failed.")
+    failure: ToolFailure = Field(description="The failure the tool reported.")
+    tool_args: dict[str, Any] | str | None = Field(
+        default=None, description="Arguments the tool was called with."
+    )
+    agent_role: str | None = Field(
+        default=None, description="Role of the agent that made the call."
+    )
+    task_name: str | None = Field(
+        default=None, description="Name or description of the task in flight."
+    )
+    task_id: str | None = Field(default=None, description="Id of the task in flight.")
+
+    @property
+    def message(self) -> str:
+        """Shorthand for the underlying failure message."""
+        return self.failure.message
+
+    def summary(self) -> str:
+        """One-line description suitable for logs and error messages."""
+        where = f" during '{self.task_name}'" if self.task_name else ""
+        return (
+            f"Tool '{self.tool_name}' failed{where}: {self.failure.as_agent_message()}"
+        )
+
+
+class ToolExecutionFailedError(Exception):
+    """Raised when a tool reports failure under :attr:`ToolFailurePolicy.RAISE`."""
+
+    def __init__(self, record: ToolFailureRecord) -> None:
+        self.record = record
+        super().__init__(record.summary())
+
+
+def detect_tool_failure(result: Any) -> ToolFailure | None:
+    """Return the failure a tool declared, if it declared one.
+
+    Only an explicit :class:`ToolFailure` counts, so a tool legitimately
+    returning text about an error is never misread as having failed.
+    """
+    if isinstance(result, ToolFailure):
+        return result
+    return None
+
+
+def failure_from_exception(
+    error: BaseException, *, retryable: bool = False
+) -> ToolFailure:
+    """Build a :class:`ToolFailure` from an exception a tool raised."""
+    return ToolFailure(
+        message=str(error) or error.__class__.__name__,
+        reason=ToolFailureReason.EXCEPTION,
+        code=error.__class__.__name__,
+        retryable=retryable,
+    )
+
+
+def resolve_tool_failure_policy(
+    tool: Any = None,
+    agent: BaseAgent | LiteAgent | None = None,
+    task: Task | None = None,
+    crew: Crew | None = None,
+) -> ToolFailurePolicy:
+    """Resolve the effective policy for one call.
+
+    Most specific wins: tool, task, agent, crew, then
+    :attr:`ToolFailurePolicy.WARN`. Callers pass either a ``BaseTool`` or the
+    ``CrewStructuredTool`` wrapping it, so both are read -- otherwise a
+    tool-scoped policy is ignored on every native function-calling path.
+    """
+    original_tool = getattr(tool, "_original_tool", None) if tool is not None else None
+
+    for source in (tool, original_tool, task, agent, crew):
+        if source is None:
+            continue
+        policy = getattr(source, "tool_failure_policy", None)
+        if policy is None:
+            continue
+        try:
+            return ToolFailurePolicy(policy)
+        except ValueError:
+            # A malformed policy must not take down a tool call.
+            logger.warning(
+                "Ignoring invalid tool_failure_policy %r on %s; expected one of %s.",
+                policy,
+                type(source).__name__,
+                [member.value for member in ToolFailurePolicy],
+            )
+    return ToolFailurePolicy.WARN
+
+
+def merge_tool_failures(
+    *groups: list[ToolFailureRecord],
+) -> list[ToolFailureRecord]:
+    """Concatenate failure lists, dropping records already present.
+
+    Guardrail retries rebuild the output from overlapping sources, so identity
+    is not enough to avoid duplicates.
+    """
+    merged: list[ToolFailureRecord] = []
+    seen: set[tuple[Any, ...]] = set()
+    for group in groups:
+        for record in group:
+            key = (
+                record.tool_name,
+                record.failure.message,
+                record.failure.code,
+                record.task_id,
+                str(record.tool_args),
+            )
+            if key in seen:
+                continue
+            seen.add(key)
+            merged.append(record)
+    return merged
+
+
+def collect_tool_failures(agent: Any) -> list[ToolFailureRecord]:
+    """Failures for the execution in progress, else the agent's last ones.
+
+    Prefers the active collector so a shared agent running concurrent tasks
+    reports only the caller's own records. Tolerates agents that do not expose
+    the attribute at all, since reading telemetry must never raise.
+    """
+    active = active_tool_failures()
+    if active is not None:
+        return list(active)
+
+    records = getattr(agent, "_tool_failures", None)
+    if not isinstance(records, list):
+        return []
+    return [record for record in records if isinstance(record, ToolFailureRecord)]
+
+
+def _agent_id(agent: Any) -> str | None:
+    """Stringified agent id, for correlating events with the call."""
+    agent_id = getattr(agent, "id", None)
+    return str(agent_id) if agent_id is not None else None
+
+
+_active_failures: ContextVar[list[ToolFailureRecord] | None] = ContextVar(
+    "crewai_tool_failures", default=None
+)
+
+
+@contextmanager
+def tool_failure_collector() -> Generator[list[ToolFailureRecord], None, None]:
+    """Collect the failures of one execution, isolated from concurrent ones.
+
+    An agent may be shared by tasks running concurrently, so accumulating on
+    the agent lets one execution erase or inherit another's records. The
+    collector is a ContextVar, which asyncio tasks and threads copy, so each
+    execution sees only its own. Nesting is safe: a guardrail retry can open
+    its own scope inside the outer one.
+    """
+    records: list[ToolFailureRecord] = []
+    token = _active_failures.set(records)
+    try:
+        yield records
+    finally:
+        _active_failures.reset(token)
+
+
+def active_tool_failures() -> list[ToolFailureRecord] | None:
+    """Records for the execution in progress, or None outside a collector."""
+    return _active_failures.get()
+
+
+def _record_failure(agent: Any, record: ToolFailureRecord) -> None:
+    """Store a record on the active collector and on the agent.
+
+    The collector is what outputs read, so it is authoritative. The agent copy
+    only backs ``last_tool_failures``, which reports the most recent execution
+    in the same best-effort way ``last_messages`` does.
+    """
+    records = _active_failures.get()
+    if records is not None:
+        records.append(record)
+
+    failures = getattr(agent, "_tool_failures", None)
+    if isinstance(failures, list):
+        failures.append(record)
+
+
+def reportable_failure(
+    failure: ToolFailure | None,
+    *,
+    tool: Any = None,
+    agent: BaseAgent | LiteAgent | None = None,
+    task: Task | None = None,
+    crew: Crew | None = None,
+) -> ToolFailure | None:
+    """Return the failure to attach to ``ToolUsageFinishedEvent``.
+
+    ``None`` under :attr:`ToolFailurePolicy.IGNORE`, so that policy really does
+    surface nothing -- neither a record, nor an event, nor a flag on the
+    finished event that a trace UI would render as a failure.
+    """
+    if failure is None:
+        return None
+    policy = resolve_tool_failure_policy(tool=tool, agent=agent, task=task, crew=crew)
+    return None if policy is ToolFailurePolicy.IGNORE else failure
+
+
+def handle_tool_failure(
+    failure: ToolFailure,
+    *,
+    tool_name: str,
+    tool_args: dict[str, Any] | str | None = None,
+    tool: Any = None,
+    agent: BaseAgent | LiteAgent | None = None,
+    task: Task | None = None,
+    crew: Crew | None = None,
+) -> ToolFailureRecord | None:
+    """Apply the effective policy to a failure a tool just reported.
+
+    Records it on the agent and emits :class:`ToolFailureDetectedEvent`.
+    Returns the record, or ``None`` under :attr:`ToolFailurePolicy.IGNORE`.
+
+    Raises:
+        ToolExecutionFailedError: Under :attr:`ToolFailurePolicy.RAISE`.
+    """
+    policy = resolve_tool_failure_policy(tool=tool, agent=agent, task=task, crew=crew)
+    if policy is ToolFailurePolicy.IGNORE:
+        return None
+
+    record = ToolFailureRecord(
+        tool_name=tool_name,
+        failure=failure,
+        tool_args=tool_args,
+        agent_role=getattr(agent, "role", None),
+        task_name=(task.name or task.description) if task else None,
+        task_id=str(task.id) if task else None,
+    )
+
+    _record_failure(agent, record)
+
+    # Local import: crewai.events imports tool types back, so a module-level
+    # import would cycle.
+    from crewai.events.event_bus import crewai_event_bus
+    from crewai.events.types.tool_usage_events import ToolFailureDetectedEvent
+
+    crewai_event_bus.emit(
+        agent,
+        ToolFailureDetectedEvent(
+            tool_name=tool_name,
+            tool_args=tool_args if tool_args is not None else {},
+            failure=failure,
+            policy=policy,
+            agent_role=record.agent_role,
+            agent_key=getattr(agent, "key", None),
+            # Set explicitly rather than via from_agent, which would also
+            # overwrite agent_role and lose the _original_role preference that
+            # the paired ToolUsage events use.
+            agent_id=_agent_id(agent),
+            agent=agent,
+            task_name=record.task_name,
+            task_id=record.task_id,
+        ),
+    )
+
+    if policy is ToolFailurePolicy.RAISE:
+        raise ToolExecutionFailedError(record)
+
+    return record

lib/crewai/src/crewai/tools/tool_usage.py

@@ -24,6 +24,13 @@ from crewai.events.types.tool_usage_events import (
 from crewai.telemetry.telemetry import Telemetry
 from crewai.tools.structured_tool import CrewStructuredTool
 from crewai.tools.tool_calling import InstructorToolCalling, ToolCalling
+from crewai.tools.tool_failure import (
+    ToolFailure,
+    ToolFailureReason,
+    detect_tool_failure,
+    failure_from_exception,
+    reportable_failure,
+)
 from crewai.utilities.agent_utils import (
     get_tool_names,
     render_text_description_and_args,
@@ -36,6 +43,7 @@ from crewai.utilities.string_utils import sanitize_tool_name
 if TYPE_CHECKING:
     from crewai.agents.agent_builder.base_agent import BaseAgent
     from crewai.agents.tools_handler import ToolsHandler
+    from crewai.crew import Crew
     from crewai.lite_agent import LiteAgent
     from crewai.llm import LLM
     from crewai.task import Task
@@ -95,6 +103,7 @@ class ToolUsage:
         agent: BaseAgent | LiteAgent | None = None,
         action: Any = None,
         fingerprint_context: dict[str, str] | None = None,
+        crew: Crew | None = None,
     ) -> None:
         self._telemetry: Telemetry = Telemetry()
         self._run_attempts: int = 1
@@ -106,10 +115,17 @@ class ToolUsage:
         self.tools_handler = tools_handler
         self.tools = tools
         self.task = task
+        self.crew = crew
         self.action = action
         self.function_calling_llm = function_calling_llm
         self.fingerprint_context = fingerprint_context or {}
         self.last_raw_result: Any = _RAW_RESULT_UNSET
+        self.last_failure: ToolFailure | None = None
+        """Failure reported by the most recent call, if any.
+
+        Covers both tool-returned failures and framework-generated ones (a
+        stringified exception, a spent usage limit).
+        """
 
         if (
             self.function_calling_llm
@@ -265,8 +281,10 @@ class ToolUsage:
                 "run_attempts": self._run_attempts,
             }
 
-            if self.agent.fingerprint:  # type: ignore
-                event_data.update(self.agent.fingerprint)  # type: ignore
+            # Not every agent type carries a fingerprint (LiteAgent does not).
+            agent_fingerprint = getattr(self.agent, "fingerprint", None)
+            if agent_fingerprint:
+                event_data.update(agent_fingerprint)
             if self.task:
                 event_data["task_name"] = self.task.name or self.task.description
                 event_data["task_id"] = str(self.task.id)
@@ -309,6 +327,10 @@ class ToolUsage:
             if usage_limit_error:
                 result = usage_limit_error
                 self.last_raw_result = result
+                self.last_failure = ToolFailure(
+                    message=usage_limit_error,
+                    reason=ToolFailureReason.USAGE_LIMIT,
+                )
                 self._telemetry.tool_usage_error(llm=self.function_calling_llm)
                 result = self._format_result(result=result)
             elif result is None:
@@ -371,6 +393,7 @@ class ToolUsage:
                         attempts=self._run_attempts,
                     )
                     self.last_raw_result = result
+                    self.last_failure = detect_tool_failure(result)
                     result = self._format_result(
                         result=tool.format_output_for_agent(result)
                     )
@@ -383,6 +406,9 @@ class ToolUsage:
                     if (
                         hasattr(available_tool, "result_as_answer")
                         and available_tool.result_as_answer
+                        # A failed tool must not become the final answer;
+                        # process_tool_results() reads this back independently.
+                        and self.last_failure is None
                     ):
                         result_as_answer = available_tool.result_as_answer
                         data["result_as_answer"] = result_as_answer
@@ -436,6 +462,7 @@ class ToolUsage:
                             f"\n{error_message}.\nMoving on then. {I18N_DEFAULT.slice('format').format(tool_names=self.tools_names)}"
                         ).message
                         self.last_raw_result = result
+                        self.last_failure = failure_from_exception(e)
                         if self.task:
                             self.task.increment_tools_errors()
                         if self.agent and self.agent.verbose:
@@ -446,6 +473,7 @@ class ToolUsage:
                         should_retry = True
             else:
                 self.last_raw_result = result
+                self.last_failure = detect_tool_failure(result)
                 result = self._format_result(
                     result=tool.format_output_for_agent(result)
                 )
@@ -504,9 +532,10 @@ class ToolUsage:
                 "run_attempts": self._run_attempts,
             }
 
-            # TODO: Investigate fingerprint attribute availability on BaseAgent/LiteAgent
-            if self.agent.fingerprint:  # type: ignore
-                event_data.update(self.agent.fingerprint)  # type: ignore
+            # Not every agent type carries a fingerprint (LiteAgent does not).
+            agent_fingerprint = getattr(self.agent, "fingerprint", None)
+            if agent_fingerprint:
+                event_data.update(agent_fingerprint)
             if self.task:
                 event_data["task_name"] = self.task.name or self.task.description
                 event_data["task_id"] = str(self.task.id)
@@ -549,6 +578,10 @@ class ToolUsage:
             if usage_limit_error:
                 result = usage_limit_error
                 self.last_raw_result = result
+                self.last_failure = ToolFailure(
+                    message=usage_limit_error,
+                    reason=ToolFailureReason.USAGE_LIMIT,
+                )
                 self._telemetry.tool_usage_error(llm=self.function_calling_llm)
                 result = self._format_result(result=result)
             elif result is None:
@@ -611,6 +644,7 @@ class ToolUsage:
                         attempts=self._run_attempts,
                     )
                     self.last_raw_result = result
+                    self.last_failure = detect_tool_failure(result)
                     result = self._format_result(
                         result=tool.format_output_for_agent(result)
                     )
@@ -623,6 +657,9 @@ class ToolUsage:
                     if (
                         hasattr(available_tool, "result_as_answer")
                         and available_tool.result_as_answer
+                        # A failed tool must not become the final answer;
+                        # process_tool_results() reads this back independently.
+                        and self.last_failure is None
                     ):
                         result_as_answer = available_tool.result_as_answer
                         data["result_as_answer"] = result_as_answer
@@ -676,6 +713,7 @@ class ToolUsage:
                             f"\n{error_message}.\nMoving on then. {I18N_DEFAULT.slice('format').format(tool_names=self.tools_names)}"
                         ).message
                         self.last_raw_result = result
+                        self.last_failure = failure_from_exception(e)
                         if self.task:
                             self.task.increment_tools_errors()
                         if self.agent and self.agent.verbose:
@@ -686,6 +724,7 @@ class ToolUsage:
                         should_retry = True
             else:
                 self.last_raw_result = result
+                self.last_failure = detect_tool_failure(result)
                 result = self._format_result(
                     result=tool.format_output_for_agent(result)
                 )
@@ -988,6 +1027,13 @@ class ToolUsage:
                 "finished_at": datetime.datetime.fromtimestamp(finished_at),
                 "from_cache": from_cache,
                 "output": result,
+                "failure": reportable_failure(
+                    self.last_failure,
+                    tool=tool,
+                    agent=self.agent,
+                    task=self.task,
+                    crew=self.crew,
+                ),
             }
         )
         if self.task:
@@ -998,9 +1044,13 @@ class ToolUsage:
     def _prepare_event_data(
         self, tool: Any, tool_calling: ToolCalling | InstructorToolCalling
     ) -> dict[str, Any]:
+        agent_id = getattr(self.agent, "id", None) if self.agent else None
         event_data = {
             "run_attempts": self._run_attempts,
             "delegations": self.task.delegations if self.task else 0,
+            # agent_key alone cannot correlate an event with a specific agent
+            # instance; the native paths already carry agent_id via from_agent.
+            "agent_id": str(agent_id) if agent_id is not None else None,
             "tool_name": sanitize_tool_name(tool.name),
             "tool_args": tool_calling.arguments,
             "tool_class": tool.__class__.__name__,

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

@@ -31,6 +31,14 @@ from crewai.tools.structured_tool import (
     CrewStructuredTool,
     strip_composite_description_prefix,
 )
+from crewai.tools.tool_failure import (
+    ToolFailure,
+    ToolFailureReason,
+    detect_tool_failure,
+    failure_from_exception,
+    handle_tool_failure,
+    reportable_failure,
+)
 from crewai.tools.tool_types import ToolResult
 from crewai.utilities.errors import AgentRepositoryError
 from crewai.utilities.exceptions.context_window_exceeding_exception import (
@@ -1532,6 +1540,9 @@ class NativeToolCallResult:
 
 def format_native_tool_output_for_agent(tool: Any, raw_result: Any) -> str:
     """Format native tool output when a tool explicitly defines a formatter."""
+    if isinstance(raw_result, ToolFailure):
+        return raw_result.as_agent_message()
+
     formatter = inspect.getattr_static(tool, "format_output_for_agent", None)
     if formatter is None:
         return str(raw_result)
@@ -1600,13 +1611,30 @@ def execute_single_native_tool_call(
 
     call_id, func_name, func_args = info
 
-    if isinstance(func_args, str):
-        try:
-            args_dict = json.loads(func_args)
-        except json.JSONDecodeError:
-            args_dict = {}
-    else:
-        args_dict = func_args
+    parsed_args, parse_error = parse_tool_call_args(func_args, func_name, call_id)
+    if parse_error is not None:
+        # Previously the decode error was swallowed into empty args and the tool
+        # ran with no input at all.
+        handle_tool_failure(
+            parse_error["tool_failure"],
+            tool_name=func_name,
+            tool_args=func_args,
+            agent=agent,
+            task=task,
+            crew=crew,
+        )
+        return NativeToolCallResult(
+            call_id=call_id,
+            func_name=func_name,
+            result=parse_error["result"],
+            tool_message={
+                "role": "tool",
+                "tool_call_id": call_id,
+                "name": func_name,
+                "content": parse_error["result"],
+            },
+        )
+    args_dict = parsed_args if parsed_args is not None else {}
 
     agent_key = getattr(agent, "key", "unknown") if agent else "unknown"
 
@@ -1628,12 +1656,14 @@ def execute_single_native_tool_call(
     input_str = json.dumps(args_dict) if args_dict else ""
     result = "Tool not found"
     raw_tool_result: Any = result
+    tool_failure: ToolFailure | None = None
 
     if tools_handler and tools_handler.cache and output_tool is not None:
         cached_result = tools_handler.cache.read(tool=func_name, input=input_str)
         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
 
     started_at = datetime.now()
@@ -1666,6 +1696,9 @@ def execute_single_native_tool_call(
     if hook_blocked:
         result = f"Tool execution blocked by hook. Tool: {func_name}"
         raw_tool_result = result
+        # The blocked message replaces any cached result, so a cached failure
+        # must not be attributed to this call.
+        tool_failure = None
     elif not from_cache:
         if func_name in available_functions and output_tool is not None:
             try:
@@ -1685,9 +1718,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 task:
                     task.increment_tools_errors()
                 crewai_event_bus.emit(
@@ -1704,6 +1739,14 @@ def execute_single_native_tool_call(
                     ),
                 )
                 error_event_emitted = True
+        else:
+            # Not cached and not executable: the model asked for a tool we do
+            # not have. The ReAct path reports this, so this one must too.
+            tool_failure = ToolFailure(
+                message=result,
+                reason=ToolFailureReason.UNKNOWN_TOOL,
+                code=func_name,
+            )
 
     after_hook_context = ToolCallHookContext(
         tool_name=func_name,
@@ -1733,9 +1776,29 @@ def execute_single_native_tool_call(
                 plan_step_description=plan_step_description,
                 started_at=started_at,
                 finished_at=datetime.now(),
+                failure=reportable_failure(
+                    tool_failure,
+                    tool=structured_tool,
+                    agent=agent,
+                    task=task,
+                    crew=crew,
+                ),
             ),
         )
 
+    # After the finished event, so subscribers see the full lifecycle even
+    # when the policy aborts.
+    if tool_failure is not None:
+        handle_tool_failure(
+            tool_failure,
+            tool_name=func_name,
+            tool_args=args_dict,
+            tool=structured_tool,
+            agent=agent,
+            task=task,
+            crew=crew,
+        )
+
     tool_message: LLMMessage = {
         "role": "tool",
         "tool_call_id": call_id,
@@ -1756,6 +1819,9 @@ def execute_single_native_tool_call(
         and original_tool.result_as_answer
         and not error_event_emitted
         and not hook_blocked
+        # A declared failure is excluded for the same reason a raised one is:
+        # an error must not silently become the task's answer.
+        and tool_failure is None
     )
 
     return NativeToolCallResult(
@@ -1779,21 +1845,28 @@ def parse_tool_call_args(
     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.
+        with the same shape as ``_execute_single_native_tool_call`` return
+        values, carrying an ``INVALID_INPUT`` failure for the caller to report.
     """
     if isinstance(func_args, str):
         try:
             return json.loads(func_args), None
         except json.JSONDecodeError as e:
+            message = (
+                f"Error: Failed to parse tool arguments as JSON: {e}. "
+                f"Please provide valid JSON arguments for the '{func_name}' tool."
+            )
             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."
-                ),
+                "result": message,
                 "from_cache": False,
                 "original_tool": original_tool,
+                "tool_failure": ToolFailure(
+                    message=message,
+                    reason=ToolFailureReason.INVALID_INPUT,
+                    code="json_decode_error",
+                ),
             }
     return func_args, None
 

lib/crewai/src/crewai/utilities/tool_utils.py

@@ -11,6 +11,11 @@ from crewai.hooks.tool_hooks import (
 )
 from crewai.security.fingerprint import Fingerprint
 from crewai.tools.structured_tool import CrewStructuredTool
+from crewai.tools.tool_failure import (
+    ToolFailure,
+    ToolFailureReason,
+    handle_tool_failure,
+)
 from crewai.tools.tool_types import ToolResult
 from crewai.tools.tool_usage import ToolUsage, ToolUsageError
 from crewai.utilities.i18n import I18N_DEFAULT
@@ -21,6 +26,7 @@ if TYPE_CHECKING:
     from crewai.agent import Agent
     from crewai.agents.agent_builder.base_agent import BaseAgent
     from crewai.crew import Crew
+    from crewai.lite_agent import LiteAgent
     from crewai.llm import LLM
     from crewai.llms.base_llm import BaseLLM
     from crewai.task import Task
@@ -33,7 +39,7 @@ async def aexecute_tool_and_check_finality(
     agent_role: str | None = None,
     tools_handler: ToolsHandler | None = None,
     task: Task | None = None,
-    agent: Agent | BaseAgent | None = None,
+    agent: Agent | BaseAgent | LiteAgent | None = None,
     function_calling_llm: BaseLLM | LLM | None = None,
     fingerprint_context: dict[str, str] | None = None,
     crew: Crew | None = None,
@@ -80,11 +86,25 @@ async def aexecute_tool_and_check_finality(
         task=task,
         agent=agent,
         action=agent_action,
+        crew=crew,
     )
 
     tool_calling = tool_usage.parse_tool_calling(agent_action.text)
 
     if isinstance(tool_calling, ToolUsageError):
+        # Mirrors the native paths, which report a malformed call as
+        # INVALID_INPUT rather than passing the message along silently.
+        handle_tool_failure(
+            ToolFailure(
+                message=tool_calling.message,
+                reason=ToolFailureReason.INVALID_INPUT,
+            ),
+            tool_name=getattr(agent_action, "tool", "") or "unknown",
+            tool_args=getattr(agent_action, "tool_input", None),
+            agent=agent,
+            task=task,
+            crew=crew,
+        )
         return ToolResult(tool_calling.message, False)
 
     sanitized_tool_name = sanitize_tool_name(tool_calling.tool_name)
@@ -138,15 +158,42 @@ async def aexecute_tool_and_check_finality(
 
         modified_result = run_after_tool_call_hooks(after_hook_context)
 
+        # After the hooks, so post_tool_call can still inspect or rewrite the
+        # result before the policy aborts.
+        if tool_usage.last_failure is not None:
+            handle_tool_failure(
+                tool_usage.last_failure,
+                tool_name=sanitized_tool_name,
+                tool_args=tool_input,
+                tool=tool,
+                agent=agent,
+                task=task,
+                crew=crew,
+            )
+
         return ToolResult(
             modified_result if modified_result is not None else tool_result,
-            tool.result_as_answer,
+            # A failed tool must not become the final answer -- the same
+            # exclusion the native paths already apply to raised errors.
+            tool.result_as_answer and tool_usage.last_failure is None,
         )
 
     tool_result = I18N_DEFAULT.errors("wrong_tool_name").format(
         tool=sanitized_tool_name,
         tools=", ".join(tool_name_to_tool_map.keys()),
     )
+    handle_tool_failure(
+        ToolFailure(
+            message=tool_result,
+            reason=ToolFailureReason.UNKNOWN_TOOL,
+            code=sanitized_tool_name,
+        ),
+        tool_name=sanitized_tool_name,
+        tool_args=tool_calling.arguments,
+        agent=agent,
+        task=task,
+        crew=crew,
+    )
     return ToolResult(result=tool_result, result_as_answer=False)
 
 
@@ -157,7 +204,7 @@ def execute_tool_and_check_finality(
     agent_role: str | None = None,
     tools_handler: ToolsHandler | None = None,
     task: Task | None = None,
-    agent: Agent | BaseAgent | None = None,
+    agent: Agent | BaseAgent | LiteAgent | None = None,
     function_calling_llm: BaseLLM | LLM | None = None,
     fingerprint_context: dict[str, str] | None = None,
     crew: Crew | None = None,
@@ -202,11 +249,25 @@ def execute_tool_and_check_finality(
         task=task,
         agent=agent,
         action=agent_action,
+        crew=crew,
     )
 
     tool_calling = tool_usage.parse_tool_calling(agent_action.text)
 
     if isinstance(tool_calling, ToolUsageError):
+        # Mirrors the native paths, which report a malformed call as
+        # INVALID_INPUT rather than passing the message along silently.
+        handle_tool_failure(
+            ToolFailure(
+                message=tool_calling.message,
+                reason=ToolFailureReason.INVALID_INPUT,
+            ),
+            tool_name=getattr(agent_action, "tool", "") or "unknown",
+            tool_args=getattr(agent_action, "tool_input", None),
+            agent=agent,
+            task=task,
+            crew=crew,
+        )
         return ToolResult(tool_calling.message, False)
 
     sanitized_tool_name = sanitize_tool_name(tool_calling.tool_name)
@@ -260,13 +321,40 @@ def execute_tool_and_check_finality(
 
         modified_result = run_after_tool_call_hooks(after_hook_context)
 
+        # After the hooks, so post_tool_call can still inspect or rewrite the
+        # result before the policy aborts.
+        if tool_usage.last_failure is not None:
+            handle_tool_failure(
+                tool_usage.last_failure,
+                tool_name=sanitized_tool_name,
+                tool_args=tool_input,
+                tool=tool,
+                agent=agent,
+                task=task,
+                crew=crew,
+            )
+
         return ToolResult(
             modified_result if modified_result is not None else tool_result,
-            tool.result_as_answer,
+            # A failed tool must not become the final answer -- the same
+            # exclusion the native paths already apply to raised errors.
+            tool.result_as_answer and tool_usage.last_failure is None,
         )
 
     tool_result = I18N_DEFAULT.errors("wrong_tool_name").format(
         tool=sanitized_tool_name,
         tools=", ".join(tool_name_to_tool_map.keys()),
     )
+    handle_tool_failure(
+        ToolFailure(
+            message=tool_result,
+            reason=ToolFailureReason.UNKNOWN_TOOL,
+            code=sanitized_tool_name,
+        ),
+        tool_name=sanitized_tool_name,
+        tool_args=tool_calling.arguments,
+        agent=agent,
+        task=task,
+        crew=crew,
+    )
     return ToolResult(result=tool_result, result_as_answer=False)

lib/crewai/tests/mcp/test_mcp_config.py

@@ -4,6 +4,7 @@ from unittest.mock import AsyncMock, patch
 
 import pytest
 from crewai.agent.core import Agent
+from crewai.mcp.client import _MCPToolResult
 from crewai.mcp.config import MCPServerHTTP, MCPServerSSE, MCPServerStdio
 from crewai.tools.base_tool import BaseTool
 
@@ -39,6 +40,9 @@ def _make_mock_client(tool_definitions):
     client.connect = AsyncMock()
     client.disconnect = AsyncMock()
     client.call_tool = AsyncMock(return_value="test result")
+    client.call_tool_result = AsyncMock(
+        return_value=_MCPToolResult("test result", False)
+    )
     return client
 
 
@@ -227,9 +231,9 @@ def test_parallel_mcp_tool_execution_same_tool(mock_tool_definitions):
         async def _call_tool(name, args):
             call_log.append(name)
             await asyncio.sleep(0.05)
-            return f"result-{name}"
+            return _MCPToolResult(f"result-{name}", False)
 
-        client.call_tool = AsyncMock(side_effect=_call_tool)
+        client.call_tool_result = AsyncMock(side_effect=_call_tool)
         return client
 
     with patch("crewai.mcp.tool_resolver.MCPClient", side_effect=_make_client):
@@ -273,9 +277,9 @@ def test_parallel_mcp_tool_execution_different_tools(mock_tool_definitions):
         async def _call_tool(name, args):
             call_log.append(name)
             await asyncio.sleep(0.05)
-            return f"result-{name}"
+            return _MCPToolResult(f"result-{name}", False)
 
-        client.call_tool = AsyncMock(side_effect=_call_tool)
+        client.call_tool_result = AsyncMock(side_effect=_call_tool)
         return client
 
     with patch("crewai.mcp.tool_resolver.MCPClient", side_effect=_make_client):

lib/crewai/tests/test_task.py

@@ -162,6 +162,7 @@ def test_task_callback_returns_task_output():
             "expected_output": "Bullet point list of 5 interesting ideas.",
             "output_format": OutputFormat.RAW,
             "messages": [],
+            "tool_failures": [],
         }
         assert output_dict == expected_output
 

lib/crewai/tests/utilities/test_agent_utils.py

@@ -1031,7 +1031,14 @@ class TestParseToolCallArgs:
     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"}
+        assert set(error.keys()) == {
+            "call_id",
+            "func_name",
+            "result",
+            "from_cache",
+            "original_tool",
+            "tool_failure",
+        }
 
 
 class TestExecuteSingleNativeToolCall: