andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-09-22 19:06:25 +00:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

fix: run model call hooks on every path and propagate a deny (#7111)

Browse Source 
* fix: let a hook deny reach the caller as a deny

A hook that raised `HookAborted` on `pre_model_call` never reached the code
making the call: the LLM layer caught it and returned `False`, which providers
translated into `ValueError("LLM call blocked by before_llm_call hook")`,
dropping the reason and the source and making a policy decision
indistinguishable from a provider outage. Every internal model call then
absorbed that error through the `except Exception` that keeps a provider hiccup
from failing a run, so memory analysis fell back to defaults and the converter
and reasoning handler retried the call that was just denied. The abort now
propagates out of the LLM layer while the boolean convention keeps its
documented `ValueError` via `LegacyHookBlocked`, and the fail-open handlers
around internal model calls re-raise it instead of degrading.

* fix: dispatch model call hooks on the paths that skipped them

A model call was only checked when the executor loop drove it: the
`from_agent is not None` short-circuit in `base_llm` silenced the hooks
for agent planning and step observation, no provider `acall` dispatched
them at all, and `InternalInstructor` bypassed `llm.call` entirely. This
replaces that short-circuit with an explicit
`model_call_hooks_already_dispatched` window so the enclosing caller
claims the dispatch, adds the pre-call dispatch to every provider's
`acall`, and runs the hooks around the Instructor client call. A denial
now emits a denied event instead of being logged and reported as a
provider failure.

* fix: report a boolean-convention deny as a deny, not an outage

A `before_llm_call` hook that blocks by returning `False` reached the five
native providers as a plain `ValueError`, which fell through to their generic
`except Exception` and was logged and emitted as `OpenAI API call failed: ...`
— the same deny raised as `HookAborted` was already labelled correctly, so the
two dialects disagreed on whether a policy decision was a provider outage. The
LLM layer now converts it into `LLMCallBlockedError`, still a `ValueError` so
the fail-open handlers around internal model calls keep absorbing it, but its
own type so a provider can report the decision it is. Since a block is raised
rather than returned, the thirteen callers that turned the return flag into a
raise by hand drop that line, and `_prepare_llm_call` raises the same type.

* fix: keep a denied plan from letting the agent run unplanned

`AgentExecutor.generate_plan` wraps `handle_agent_reasoning()` in a bare
`except Exception`, so guarding the reasoning handler alone still left the
deny absorbed one frame up: the executor logged "Error during planning" and
the agent proceeded with no plan. It now re-raises `HookAborted` like the
other planning boundaries, and the accompanying test also covers the
boolean convention still degrading at a fail-open site.

* fix: stop a denied knowledge query from running the task without knowledge

`handle_knowledge_retrieval` and its async twin wrap the query rewrite in
their own `except Exception`, so guarding `_get_knowledge_search_query`
alone still let `execute_task` continue on the unaugmented prompt after a
deny. Both now emit the terminal `KnowledgeSearchQueryFailedEvent` and
re-raise `HookAborted`, matching the second-frame guard already added to
`AgentExecutor.generate_plan`. Also documents the abort contract on
`PlannerObserver.observe`.

* fix: stop nine callers from re-swallowing a model call deny

CodeRabbit caught the replan path re-swallowing a deny, so an AST sweep of
every caller of a guarded function found the same defeat in nine places:
classic and replan planning, memory recall and memory save on both `Agent`
and `LiteAgent`, the base executor's save, and `LLMGuardrail.__call__`,
which turned a refused call into validation feedback. Each now re-raises
`HookAborted` after emitting whatever terminal event it owes, while every
other failure keeps degrading as before — the knowledge guards move to that
same idiom instead of duplicating their emit.

* fix: pair a denied guardrail with the event it started

Re-raising from `LLMGuardrail` left `process_guardrail` between its started
and completed events, so a denied validation read as one still in flight
rather than a policy decision. It now emits `LLMGuardrailCompletedEvent`
with the deny reason before the abort leaves, matching what every other
guarded site in this change already does.

* fix: stop retrying a task after a hook denied its model call

`Agent.execute_task` funnels every exception into `_handle_execution_error`,
which re-runs the whole task up to `max_retry_limit` times, so a policy deny
read as a transient blip: a crew whose first model call was denied retried and
returned a normal answer. `HookAborted` now joins `_passthrough_exceptions`,
the tuple already reserved for deliberate stops. The new boundary tests drive
the public entry points instead of the frame that makes the call, and count
model calls so a deny that gets retried fails the assertion — ten of the twelve
fail against `main`.

* fix: stop a denied plan step from being reported as a failed step

Making model call hooks reachable on agent-bearing calls put a deny inside
`StepExecutor.execute`, whose broad `except Exception` turned it into
`StepResult(success=False)` and let the plan carry on; `HookAborted` now
joins `ToolExecutionFailedError` in the passthrough handlers there, and
`execute_todos_parallel` re-raises a deny that `return_exceptions=True`
would otherwise record as one failed todo. `_emit_call_denied_event` also
renders the source through the now-public `source_name`, so a hook that
names itself with a callable reads as its name instead of a repr.

---------

Co-authored-by: Vidit Ostwal <110953813+Vidit-Ostwal@users.noreply.github.com>
This commit is contained in:
Lucas Gomide
2026-08-28 13:32:09 -03:00
committed by GitHub
parent cba6c03646
commit e9d4c57b2a
32 changed files with 1443 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
lib/crewai/src/crewai/agent/core.py
View File

@@ -74,6 +74,7 @@ from crewai.events.types.memory_events import (
)
from crewai.events.types.skill_events import SkillUsedEvent
from crewai.experimental.agent_executor import AgentExecutor
from crewai.hooks.dispatch import HookAborted
from crewai.knowledge.knowledge import Knowledge
from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource
from crewai.lite_agent_output import LiteAgentOutput
@@ -139,7 +140,10 @@ if TYPE_CHECKING:
# Deliberate stops, not transient errors: never swallowed into the
# max_retry_limit loop.
_passthrough_exceptions: tuple[type[Exception], ...] = (ToolExecutionFailedError,)
_passthrough_exceptions: tuple[type[Exception], ...] = (
ToolExecutionFailedError,
HookAborted,
)
_EXECUTOR_CLASS_MAP: dict[str, type] = {
"CrewAgentExecutor": CrewAgentExecutor,
@@ -711,6 +715,9 @@ class Agent(BaseAgent):
error=str(e),
),
)
# a deny aborts the task; any other failure degrades to no memory
if isinstance(e, HookAborted):
raise
return task_prompt
@@ -1438,6 +1445,18 @@ class Agent(BaseAgent):
),
)
return rewritten_query
except HookAborted as e:
# A deny still owes the started event above its terminal event; only
# the fallback to no query is skipped.
crewai_event_bus.emit(
self,
event=KnowledgeQueryFailedEvent(
error=str(e),
from_task=task,
from_agent=self,
),
)
raise
except Exception as e:
crewai_event_bus.emit(
self,
@@ -1634,6 +1653,9 @@ class Agent(BaseAgent):
error=str(e),
),
)
# a deny aborts the kickoff; any other failure degrades to no memory
if isinstance(e, HookAborted):
raise
inputs: dict[str, Any] = {
"input": formatted_messages,
@@ -1815,6 +1837,8 @@ class Agent(BaseAgent):
extracted = agent_memory.extract_memories(raw)
if extracted:
agent_memory.remember_many(extracted)
except HookAborted:
raise
except Exception as e:
self._logger.log("error", f"Failed to save kickoff result to memory: {e}")

9
lib/crewai/src/crewai/agent/utils.py
View File

@@ -15,6 +15,7 @@ from crewai.events.types.knowledge_events import (
KnowledgeRetrievalStartedEvent,
KnowledgeSearchQueryFailedEvent,
)
from crewai.hooks.dispatch import HookAborted
from crewai.knowledge.utils.knowledge_utils import extract_knowledge_context
from crewai.utilities.pydantic_schema_utils import generate_model_description
from crewai.utilities.types import LLMMessage
@@ -53,6 +54,8 @@ def handle_reasoning(agent: Agent, task: Task) -> None:
planning_handler.handle_agent_reasoning()
)
task.description += f"\n\nPlanning:\n{planning_output.plan.plan}"
except HookAborted:
raise
except Exception as e:
agent._logger.log("error", f"Error during planning: {e!s}")
@@ -195,6 +198,9 @@ def handle_knowledge_retrieval(
from_agent=agent,
),
)
# a deny aborts the task; any other failure degrades to no knowledge
if isinstance(e, HookAborted):
raise
return task_prompt
@@ -391,4 +397,7 @@ async def ahandle_knowledge_retrieval(
from_agent=agent,
),
)
# a deny aborts the task; any other failure degrades to no knowledge
if isinstance(e, HookAborted):
raise
return task_prompt

4
lib/crewai/src/crewai/agents/agent_builder/base_agent_executor.py
View File

@@ -30,6 +30,8 @@ class BaseAgentExecutor(BaseModel):
def _save_to_memory(self, output: AgentFinish) -> None:
"""Save task result to unified memory (memory or crew._memory)."""
from crewai.hooks.dispatch import HookAborted
if self.agent is None:
return
memory = getattr(self.agent, "memory", None) or (
@@ -61,5 +63,7 @@ class BaseAgentExecutor(BaseModel):
)
else:
memory.remember_many(extracted, agent_role=self.agent.role)
except HookAborted:
raise
except Exception as e:
self.agent._logger.log("error", f"Failed to save to memory: {e}")

21
lib/crewai/src/crewai/agents/planner_observer.py
View File

@@ -131,7 +131,13 @@ class PlannerObserver:
StepObservation with the Planner's analysis. Any suggested
refinements are structured StepRefinement objects ready for
direct application — no second LLM call needed.
Raises:
HookAborted: A `pre_model_call` hook denied the observation call.
Every other failure degrades to a conservative observation.
"""
from crewai.hooks.dispatch import HookAborted
agent_role = self.agent.role
crewai_event_bus.emit(
@@ -188,6 +194,21 @@ class PlannerObserver:
return observation
except HookAborted as e:
# A deny still owes the started event above its terminal event; only
# the conservative-replan fallback is skipped.
crewai_event_bus.emit(
self.agent,
event=StepObservationFailedEvent(
agent_role=agent_role,
step_number=completed_step.step_number,
step_description=completed_step.description,
error=str(e),
from_task=self.task,
from_agent=self.agent,
),
)
raise
except Exception as e:
logger.warning(
f"Observation LLM call failed: {e}. Defaulting to conservative replan."

5
lib/crewai/src/crewai/agents/step_executor.py
View File

@@ -28,6 +28,7 @@ from crewai.events.types.tool_usage_events import (
ToolUsageFinishedEvent,
ToolUsageStartedEvent,
)
from crewai.hooks.dispatch import HookAborted
from crewai.tools.tool_failure import ToolExecutionFailedError
from crewai.utilities.agent_utils import (
build_text_tool_calling_fallback_message,
@@ -181,7 +182,7 @@ class StepExecutor:
tool_calls_made=tool_calls_made,
execution_time=elapsed,
)
except ToolExecutionFailedError:
except (ToolExecutionFailedError, HookAborted):
# A deliberate stop: StepResult(success=False) would let the plan
# carry on.
raise
@@ -224,7 +225,7 @@ class StepExecutor:
tool_calls_made=tool_calls_made,
execution_time=elapsed,
)
except ToolExecutionFailedError:
except (ToolExecutionFailedError, HookAborted):
# Same as the outer handler, reached via the text-tooling
# fallback.
raise

9
lib/crewai/src/crewai/experimental/agent_executor.py
View File

@@ -57,6 +57,7 @@ from crewai.events.types.tool_usage_events import (
from crewai.flow.flow import Flow, listen, or_, router, start
from crewai.flow.flow_context import current_flow_id
from crewai.flow.types import FlowMethodName
from crewai.hooks.dispatch import HookAborted
from crewai.hooks.llm_hooks import (
get_after_llm_call_hooks,
get_before_llm_call_hooks,
@@ -420,6 +421,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
# Do NOT mutate task.description — it's a shared object that
# accumulates plan text on re-invoke.
except HookAborted:
raise
except Exception as e:
if hasattr(self.agent, "_logger"):
self.agent._logger.log("error", f"Error during planning: {e!s}")
@@ -1295,6 +1298,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
# todo ↔ result (or exception) mapping.
step_results: list[tuple[TodoItem, StepResult]] = []
for todo, item in zip(ready, gathered, strict=True):
if isinstance(item, HookAborted):
raise item
if isinstance(item, BaseException):
error_msg = f"Error: {item!s}"
todo.result = error_msg
@@ -2552,6 +2557,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
)
return
except HookAborted:
raise
except Exception as e:
if self.agent and self.agent.verbose:
PRINTER.print(
@@ -2704,6 +2711,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
color="green",
)
except HookAborted:
raise
except Exception as e:
if hasattr(self.agent, "_logger"):
self.agent._logger.log("error", f"Error during replanning: {e!s}")

4
lib/crewai/src/crewai/experimental/evaluation/evaluation_display.py
View File

@@ -311,6 +311,8 @@ class EvaluationDisplayFormatter:
scores: list[float | None],
strategy: AggregationStrategy,
) -> str:
from crewai.hooks.dispatch import HookAborted
if len(feedbacks) <= 2 and all(len(fb) < 200 for fb in feedbacks):
return "\n\n".join(
[f"Feedback {i + 1}: {fb}" for i, fb in enumerate(feedbacks)]
@@ -372,6 +374,8 @@ class EvaluationDisplayFormatter:
raise ValueError("LLM must be initialized")
return llm.call(prompt)
except HookAborted:
raise
except Exception:
return "Synthesized from multiple tasks: " + "\n\n".join(
[f"- {fb[:500]}..." for fb in feedbacks]

8
lib/crewai/src/crewai/flow/human_feedback.py
View File

@@ -253,6 +253,8 @@ def _pre_review_with_lessons(
learn_source: str,
learn_strict: bool,
) -> Any:
from crewai.hooks.dispatch import HookAborted
try:
mem = flow_instance.memory
if mem is None:
@@ -282,6 +284,8 @@ def _pre_review_with_lessons(
return PreReviewResult.model_validate(response).improved_output
reviewed = llm_inst.call(messages)
return reviewed if isinstance(reviewed, str) else str(reviewed)
except HookAborted:
raise
except Exception:
if learn_strict:
logger.warning(
@@ -308,6 +312,8 @@ def _distill_and_store_lessons(
learn_source: str,
learn_strict: bool,
) -> None:
from crewai.hooks.dispatch import HookAborted
try:
mem = flow_instance.memory
if mem is None:
@@ -344,6 +350,8 @@ def _distill_and_store_lessons(
if lessons:
mem.remember_many(lessons, source=learn_source) # type: ignore[union-attr]
except HookAborted:
raise
except Exception:
if learn_strict:
logger.warning(

5
lib/crewai/src/crewai/flow/runtime/__init__.py
View File

@@ -3847,6 +3847,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
from pydantic import BaseModel, Field
from crewai.hooks.dispatch import HookAborted
from crewai.llm import LLM
from crewai.llms.base_llm import BaseLLM as BaseLLMClass
from crewai.utilities.i18n import I18N_DEFAULT
@@ -3902,6 +3903,8 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
logger.warning(f"Unexpected response type: {type(response)}")
return outcomes[0]
except HookAborted:
raise
except Exception as e:
logger.warning(
f"Structured output failed, falling back to simple prompting: {e}"
@@ -3933,6 +3936,8 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
)
return outcomes[0]
except HookAborted:
raise
except Exception as fallback_err:
logger.warning(
f"Simple prompting also failed: {fallback_err}. "

4
lib/crewai/src/crewai/hooks/dispatch.py
View File

@@ -209,7 +209,7 @@ def _resolve_hooks(point: InterceptionPoint) -> list[HookFn]:
return global_hooks
def _source_name(source: Any) -> str | None:
def source_name(source: Any) -> str | None:
"""Best-effort readable name for a hook source."""
if source is None:
return None
@@ -341,7 +341,7 @@ def run_hooks(
except HookAborted as aborted:
outcome = "aborted"
abort_reason = aborted.reason
abort_source = _source_name(aborted.source)
abort_source = source_name(aborted.source)
raise
finally:
_emit_telemetry(

45
lib/crewai/src/crewai/hooks/llm_hooks.py
View File

@@ -1,5 +1,8 @@
from __future__ import annotations
from collections.abc import Iterator
from contextlib import contextmanager
from contextvars import ContextVar
from typing import TYPE_CHECKING, Any, cast
from crewai_core.printer import PRINTER
@@ -166,14 +169,50 @@ _after_llm_call_hooks: list[AfterLLMCallHookType | AfterLLMCallHookCallable] = (
)
class LegacyHookBlocked(HookAborted):
"""A ``before_llm_call`` hook blocked the call by returning ``False``.
Distinguishes the boolean convention, which the LLM layer keeps surfacing as
the documented ``ValueError``, from a hook that raised :class:`HookAborted`
itself and must reach the caller as the deny it is. Raised by the reducer and
consumed inside the LLM layer, which re-raises it as
:class:`~crewai.llms.base_llm.LLMCallBlockedError`.
"""
_model_call_hooks_dispatched: ContextVar[bool] = ContextVar(
"model_call_hooks_dispatched", default=False
)
@contextmanager
def model_call_hooks_dispatched() -> Iterator[None]:
"""Mark the window where the model-call hooks already ran for a pending call.
The executor dispatches with its own richer context (executor, task, crew)
and only then reaches the LLM. Without this marker the LLM layer would
dispatch a second time for the same call.
"""
token = _model_call_hooks_dispatched.set(True)
try:
yield
finally:
_model_call_hooks_dispatched.reset(token)
def model_call_hooks_already_dispatched() -> bool:
"""Whether an enclosing caller already dispatched hooks for the current call."""
return _model_call_hooks_dispatched.get()
def before_llm_call_reducer(context: LLMCallHookContext, result: object) -> bool:
"""Legacy calling convention for ``pre_model_call`` hooks.
A ``False`` return aborts the call (mapped to :class:`HookAborted`); messages
are modified in place, so no payload replacement occurs here.
A ``False`` return aborts the call (mapped to :class:`LegacyHookBlocked`);
messages are modified in place, so no payload replacement occurs here.
"""
if result is False:
raise HookAborted(reason="before_llm_call hook returned False")
raise LegacyHookBlocked(reason="before_llm_call hook returned False")
return False

9
lib/crewai/src/crewai/lite_agent.py
View File

@@ -598,6 +598,8 @@ class LiteAgent(FlowTrackable, BaseModel):
def _inject_memory_context(self) -> None:
"""Recall relevant memories and append to the system message. No-op if _memory is None."""
from crewai.hooks.dispatch import HookAborted
if self._memory is None:
return
query = self._get_last_user_content()
@@ -641,9 +643,14 @@ class LiteAgent(FlowTrackable, BaseModel):
error=str(e),
),
)
# a deny aborts the run; any other failure degrades to no memory
if isinstance(e, HookAborted):
raise
def _save_to_memory(self, output_text: str) -> None:
"""Extract discrete memories from the run and remember each. No-op if _memory is None or read-only."""
from crewai.hooks.dispatch import HookAborted
if self._memory is None or self._memory.read_only:
return
input_str = self._get_last_user_content() or "User request"
@@ -652,6 +659,8 @@ class LiteAgent(FlowTrackable, BaseModel):
extracted = self._memory.extract_memories(raw)
if extracted:
self._memory.remember_many(extracted, agent_role=self.role)
except HookAborted:
raise
except Exception as e:
if self.verbose:
PRINTER.print(

28
lib/crewai/src/crewai/llm.py
View File

@@ -31,10 +31,12 @@ from crewai.events.types.tool_usage_events import (
ToolUsageFinishedEvent,
ToolUsageStartedEvent,
)
from crewai.hooks.dispatch import HookAborted
from crewai.llms._finish_reason_utils import extract_choices_finish_reason_and_id
from crewai.llms.base_llm import (
BaseLLM,
JsonResponseFormat,
LLMCallBlockedError,
get_current_call_id,
llm_call_context,
)
@@ -1039,12 +1041,15 @@ class LLM(BaseLLM):
if not tool_calls or not available_functions:
if response_model and self.is_litellm:
from crewai.hooks.llm_hooks import model_call_hooks_dispatched
instructor_instance = InternalInstructor(
content=full_response,
model=response_model,
llm=self,
)
result = instructor_instance.to_pydantic()
with model_call_hooks_dispatched():
result = instructor_instance.to_pydantic()
structured_response = result.model_dump_json()
usage_dict = self._usage_to_dict(usage_info)
self._handle_emit_call_events(
@@ -1241,6 +1246,7 @@ class LLM(BaseLLM):
str: The response text
"""
if response_model and self.is_litellm:
from crewai.hooks.llm_hooks import model_call_hooks_dispatched
from crewai.utilities.internal_instructor import InternalInstructor
messages = params.get("messages", [])
@@ -1256,7 +1262,8 @@ class LLM(BaseLLM):
model=response_model,
llm=self,
)
result = instructor_instance.to_pydantic()
with model_call_hooks_dispatched():
result = instructor_instance.to_pydantic()
structured_response = result.model_dump_json()
self._handle_emit_call_events(
response=structured_response,
@@ -1396,6 +1403,7 @@ class LLM(BaseLLM):
str: The response text
"""
if response_model and self.is_litellm:
from crewai.hooks.llm_hooks import model_call_hooks_dispatched
from crewai.utilities.internal_instructor import InternalInstructor
messages = params.get("messages", [])
@@ -1411,7 +1419,8 @@ class LLM(BaseLLM):
model=response_model,
llm=self,
)
result = instructor_instance.to_pydantic()
with model_call_hooks_dispatched():
result = instructor_instance.to_pydantic()
structured_response = result.model_dump_json()
self._handle_emit_call_events(
response=structured_response,
@@ -1874,8 +1883,11 @@ class LLM(BaseLLM):
msg_role: Literal["assistant"] = "assistant"
message["role"] = msg_role
if not self._invoke_before_llm_call_hooks(messages, from_agent):
raise ValueError("LLM call blocked by before_llm_call hook")
try:
self._invoke_before_llm_call_hooks(messages, from_agent)
except (HookAborted, LLMCallBlockedError) as e:
self._emit_call_denied_event(e, from_task, from_agent)
raise
with suppress_warnings():
if callbacks and len(callbacks) > 0:
@@ -2016,6 +2028,12 @@ class LLM(BaseLLM):
msg_role: Literal["assistant"] = "assistant"
message["role"] = msg_role
try:
self._invoke_before_llm_call_hooks(messages, from_agent)
except (HookAborted, LLMCallBlockedError) as e:
self._emit_call_denied_event(e, from_task, from_agent)
raise
with suppress_warnings():
if callbacks and len(callbacks) > 0:
self.set_callbacks(callbacks)

89
lib/crewai/src/crewai/llms/base_llm.py
View File

@@ -72,6 +72,15 @@ class JsonResponseFormat(TypedDict):
type: Literal["json_object"]
class LLMCallBlockedError(ValueError):
"""A ``before_llm_call`` hook blocked the call by returning ``False``.
A ``ValueError`` so the fail-open handlers around internal model calls keep
absorbing it, and its own type so a provider can report it as the decision
it is instead of letting it read as a provider outage.
"""
DEFAULT_CONTEXT_WINDOW_SIZE: Final[int] = 4096
DEFAULT_SUPPORTS_STOP_WORDS: Final[bool] = True
_JSON_EXTRACTION_PATTERN: Final[re.Pattern[str]] = re.compile(r"\{.*}", re.DOTALL)
@@ -660,6 +669,28 @@ class BaseLLM(BaseModel, ABC):
),
)
def _emit_call_denied_event(
self,
denial: Exception,
from_task: Task | None = None,
from_agent: BaseAgent | None = None,
) -> None:
"""Report a hook deny as a deny rather than as a provider failure.
The call still owes its started event a terminal one, so the failed event
is emitted — with a message that names the decision instead of blaming
the provider for an outage that never happened.
"""
from crewai.hooks.dispatch import source_name
source = source_name(getattr(denial, "source", None))
reason = getattr(denial, "reason", str(denial))
message = f"LLM call denied by {source or 'hook'}: {reason}"
logging.warning(message)
self._emit_call_failed_event(
error=message, from_task=from_task, from_agent=from_agent
)
def _emit_stream_chunk_event(
self,
chunk: str,
@@ -990,43 +1021,51 @@ class BaseLLM(BaseModel, ABC):
messages: list[LLMMessage],
from_agent: BaseAgent | None = None,
) -> bool:
"""Invoke before_llm_call hooks for direct LLM calls (no agent context).
"""Invoke before_llm_call hooks for an LLM call reaching the provider.
This method should be called by native provider implementations before
making the actual LLM call when from_agent is None (direct calls).
making the actual LLM call. It no-ops when an enclosing caller — the
executor — already dispatched the hooks for this same call.
Args:
messages: The messages being sent to the LLM
from_agent: The agent making the call (None for direct calls)
from_agent: The agent making the call, when there is one
Returns:
True if LLM call should proceed, False if blocked by hook
True, so a provider may still guard the call with ``if not ...``.
A block is raised, never returned.
Raises:
HookAborted: If a hook raised it. The deny reaches the caller intact
instead of being flattened into a provider-style error.
LLMCallBlockedError: If a legacy hook blocked the call by returning
``False``. A ``ValueError``, so the fail-open handlers around
internal model calls keep absorbing it.
Example:
>>> # In a native provider's call() method:
>>> if from_agent is None and not self._invoke_before_llm_call_hooks(
... messages, from_agent
... ):
... raise ValueError("LLM call blocked by hook")
>>> self._invoke_before_llm_call_hooks(messages, from_agent)
"""
if from_agent is not None:
return True
from crewai_core.printer import PRINTER
from crewai.hooks.dispatch import HookAborted, InterceptionPoint, dispatch
from crewai.hooks.dispatch import InterceptionPoint, dispatch
from crewai.hooks.llm_hooks import (
LLMCallHookContext,
LegacyHookBlocked,
before_llm_call_reducer,
model_call_hooks_already_dispatched,
)
if model_call_hooks_already_dispatched():
return True
# No early global-list guard: dispatch resolves global + execution-scoped
# hooks and has its own no-op fast path, so scoped hooks still run here.
hook_context = LLMCallHookContext(
executor=None,
messages=messages,
llm=self,
agent=None,
agent=from_agent,
task=None,
crew=None,
)
@@ -1037,12 +1076,14 @@ class BaseLLM(BaseModel, ABC):
hook_context,
reducer=before_llm_call_reducer,
)
except HookAborted:
except LegacyHookBlocked as blocked:
PRINTER.print(
content="LLM call blocked by before_llm_call hook",
color="yellow",
)
return False
raise LLMCallBlockedError(
"LLM call blocked by before_llm_call hook"
) from blocked
return True
@@ -1052,42 +1093,44 @@ class BaseLLM(BaseModel, ABC):
response: str,
from_agent: BaseAgent | None = None,
) -> str:
"""Invoke after_llm_call hooks for direct LLM calls (no agent context).
"""Invoke after_llm_call hooks for an LLM call that reached the provider.
This method should be called by native provider implementations after
receiving the LLM response when from_agent is None (direct calls).
receiving the LLM response. It no-ops when an enclosing caller — the
executor — already dispatched the hooks for this same call.
Args:
messages: The messages that were sent to the LLM
response: The response from the LLM
from_agent: The agent that made the call (None for direct calls)
from_agent: The agent that made the call, when there is one
Returns:
The potentially modified response string
Example:
>>> # In a native provider's call() method:
>>> if from_agent is None and isinstance(result, str):
>>> if isinstance(result, str):
... result = self._invoke_after_llm_call_hooks(
... messages, result, from_agent
... )
"""
if from_agent is not None or not isinstance(response, str):
return response
from crewai.hooks.dispatch import InterceptionPoint, dispatch
from crewai.hooks.llm_hooks import (
LLMCallHookContext,
after_llm_call_reducer,
model_call_hooks_already_dispatched,
)
if model_call_hooks_already_dispatched() or not isinstance(response, str):
return response
# No early global-list guard: dispatch resolves global + execution-scoped
# hooks and has its own no-op fast path, so scoped hooks still run here.
hook_context = LLMCallHookContext(
executor=None,
messages=messages,
llm=self,
agent=None,
agent=from_agent,
task=None,
crew=None,
response=response,

21
lib/crewai/src/crewai/llms/providers/anthropic/completion.py
View File

@@ -8,7 +8,13 @@ from typing import Any, Final, Literal, Protocol, TypeGuard, TypedDict, cast
from pydantic import BaseModel, PrivateAttr, model_validator
from crewai.events.types.llm_events import LLMCallType
from crewai.llms.base_llm import BaseLLM, JsonResponseFormat, llm_call_context
from crewai.hooks.dispatch import HookAborted
from crewai.llms.base_llm import (
BaseLLM,
JsonResponseFormat,
LLMCallBlockedError,
llm_call_context,
)
from crewai.llms.hooks.base import BaseInterceptor
from crewai.llms.hooks.transport import AsyncHTTPTransport, HTTPTransport
from crewai.llms.providers.utils.common import safe_tool_conversion
@@ -401,10 +407,7 @@ class AnthropicCompletion(BaseLLM):
self._format_messages_for_anthropic(messages)
)
if not self._invoke_before_llm_call_hooks(
formatted_messages, from_agent
):
raise ValueError("LLM call blocked by before_llm_call hook")
self._invoke_before_llm_call_hooks(formatted_messages, from_agent)
completion_params = self._prepare_completion_params(
formatted_messages, system_message, tools, available_functions
@@ -429,6 +432,9 @@ class AnthropicCompletion(BaseLLM):
effective_response_model,
)
except (HookAborted, LLMCallBlockedError) as e:
self._emit_call_denied_event(e, from_task, from_agent)
raise
except Exception as e:
error_msg = f"Anthropic API call failed: {e!s}"
logging.error(error_msg)
@@ -476,6 +482,8 @@ class AnthropicCompletion(BaseLLM):
self._format_messages_for_anthropic(messages)
)
self._invoke_before_llm_call_hooks(formatted_messages, from_agent)
completion_params = self._prepare_completion_params(
formatted_messages, system_message, tools, available_functions
)
@@ -499,6 +507,9 @@ class AnthropicCompletion(BaseLLM):
effective_response_model,
)
except (HookAborted, LLMCallBlockedError) as e:
self._emit_call_denied_event(e, from_task, from_agent)
raise
except Exception as e:
error_msg = f"Anthropic API call failed: {e!s}"
logging.error(error_msg)

21
lib/crewai/src/crewai/llms/providers/azure/completion.py
View File

@@ -9,6 +9,7 @@ from urllib.parse import urlparse
from pydantic import BaseModel, PrivateAttr, model_validator
from typing_extensions import Self
from crewai.hooks.dispatch import HookAborted
from crewai.llms._finish_reason_utils import extract_choices_finish_reason_and_id
from crewai.llms.hooks.base import BaseInterceptor
from crewai.utilities.agent_utils import is_context_length_exceeded
@@ -42,7 +43,12 @@ try:
)
from crewai.events.types.llm_events import LLMCallType
from crewai.llms.base_llm import BaseLLM, call_stream_override, llm_call_context
from crewai.llms.base_llm import (
BaseLLM,
LLMCallBlockedError,
call_stream_override,
llm_call_context,
)
except ImportError:
raise ImportError(
@@ -521,10 +527,7 @@ class AzureCompletion(BaseLLM):
formatted_messages = self._format_messages_for_azure(messages)
if not self._invoke_before_llm_call_hooks(
formatted_messages, from_agent
):
raise ValueError("LLM call blocked by before_llm_call hook")
self._invoke_before_llm_call_hooks(formatted_messages, from_agent)
completion_params = self._prepare_completion_params(
formatted_messages, tools, effective_response_model
@@ -547,6 +550,9 @@ class AzureCompletion(BaseLLM):
effective_response_model,
)
except (HookAborted, LLMCallBlockedError) as e:
self._emit_call_denied_event(e, from_task, from_agent)
raise
except Exception as e:
return self._handle_api_error(e, from_task, from_agent) # type: ignore[func-returns-value]
@@ -603,6 +609,8 @@ class AzureCompletion(BaseLLM):
formatted_messages = self._format_messages_for_azure(messages)
self._invoke_before_llm_call_hooks(formatted_messages, from_agent)
completion_params = self._prepare_completion_params(
formatted_messages, tools, effective_response_model
)
@@ -624,6 +632,9 @@ class AzureCompletion(BaseLLM):
effective_response_model,
)
except (HookAborted, LLMCallBlockedError) as e:
self._emit_call_denied_event(e, from_task, from_agent)
raise
except Exception as e:
self._handle_api_error(e, from_task, from_agent)

16
lib/crewai/src/crewai/llms/providers/bedrock/completion.py
View File

@@ -11,7 +11,8 @@ from pydantic import BaseModel, PrivateAttr, model_validator
from typing_extensions import Required
from crewai.events.types.llm_events import LLMCallType
from crewai.llms.base_llm import BaseLLM, llm_call_context
from crewai.hooks.dispatch import HookAborted
from crewai.llms.base_llm import BaseLLM, LLMCallBlockedError, llm_call_context
from crewai.llms.providers.utils.common import safe_tool_conversion
from crewai.utilities.agent_utils import is_context_length_exceeded
from crewai.utilities.exceptions.context_window_exceeding_exception import (
@@ -377,10 +378,7 @@ class BedrockCompletion(BaseLLM):
messages
)
if not self._invoke_before_llm_call_hooks(
formatted_messages, from_agent
):
raise ValueError("LLM call blocked by before_llm_call hook")
self._invoke_before_llm_call_hooks(formatted_messages, from_agent)
body: BedrockConverseRequestBody = {
"inferenceConfig": self._get_inference_config(),
@@ -447,6 +445,9 @@ class BedrockCompletion(BaseLLM):
effective_response_model,
)
except (HookAborted, LLMCallBlockedError) as e:
self._emit_call_denied_event(e, from_task, from_agent)
raise
except Exception as e:
if is_context_length_exceeded(e):
logging.error(f"Context window exceeded: {e}")
@@ -510,6 +511,8 @@ class BedrockCompletion(BaseLLM):
messages
)
self._invoke_before_llm_call_hooks(formatted_messages, from_agent)
body: BedrockConverseRequestBody = {
"inferenceConfig": self._get_inference_config(),
}
@@ -575,6 +578,9 @@ class BedrockCompletion(BaseLLM):
effective_response_model,
)
except (HookAborted, LLMCallBlockedError) as e:
self._emit_call_denied_event(e, from_task, from_agent)
raise
except Exception as e:
if is_context_length_exceeded(e):
logging.error(f"Context window exceeded: {e}")

18
lib/crewai/src/crewai/llms/providers/gemini/completion.py
View File

@@ -10,7 +10,8 @@ from typing import Any, Literal, cast
from pydantic import BaseModel, Field, PrivateAttr, model_validator
from crewai.events.types.llm_events import LLMCallType
from crewai.llms.base_llm import BaseLLM, llm_call_context
from crewai.hooks.dispatch import HookAborted
from crewai.llms.base_llm import BaseLLM, LLMCallBlockedError, llm_call_context
from crewai.llms.hooks.base import BaseInterceptor
from crewai.utilities.agent_utils import is_context_length_exceeded
from crewai.utilities.exceptions.context_window_exceeding_exception import (
@@ -313,10 +314,7 @@ class GeminiCompletion(BaseLLM):
messages_for_hooks = self._convert_contents_to_dict(formatted_content)
if not self._invoke_before_llm_call_hooks(
messages_for_hooks, from_agent
):
raise ValueError("LLM call blocked by before_llm_call hook")
self._invoke_before_llm_call_hooks(messages_for_hooks, from_agent)
config = self._prepare_generation_config(
system_instruction, tools, effective_response_model
@@ -341,6 +339,9 @@ class GeminiCompletion(BaseLLM):
effective_response_model,
)
except (HookAborted, LLMCallBlockedError) as e:
self._emit_call_denied_event(e, from_task, from_agent)
raise
except APIError as e:
error_msg = f"Google Gemini API error: {e.code} - {e.message}"
logging.error(error_msg)
@@ -397,6 +398,10 @@ class GeminiCompletion(BaseLLM):
self._format_messages_for_gemini(messages)
)
messages_for_hooks = self._convert_contents_to_dict(formatted_content)
self._invoke_before_llm_call_hooks(messages_for_hooks, from_agent)
config = self._prepare_generation_config(
system_instruction, tools, effective_response_model
)
@@ -420,6 +425,9 @@ class GeminiCompletion(BaseLLM):
effective_response_model,
)
except (HookAborted, LLMCallBlockedError) as e:
self._emit_call_denied_event(e, from_task, from_agent)
raise
except APIError as e:
error_msg = f"Google Gemini API error: {e.code} - {e.message}"
logging.error(error_msg)

21
lib/crewai/src/crewai/llms/providers/openai/completion.py
View File

@@ -36,8 +36,14 @@ from openai.types.responses import (
from pydantic import BaseModel, PrivateAttr, model_validator
from crewai.events.types.llm_events import LLMCallType
from crewai.hooks.dispatch import HookAborted
from crewai.llms._finish_reason_utils import extract_choices_finish_reason_and_id
from crewai.llms.base_llm import BaseLLM, JsonResponseFormat, llm_call_context
from crewai.llms.base_llm import (
BaseLLM,
JsonResponseFormat,
LLMCallBlockedError,
llm_call_context,
)
from crewai.llms.hooks.base import BaseInterceptor
from crewai.llms.hooks.transport import AsyncHTTPTransport, HTTPTransport
from crewai.llms.providers.utils.common import safe_tool_conversion
@@ -462,10 +468,7 @@ class OpenAICompletion(BaseLLM):
formatted_messages = self._format_messages(messages)
if not self._invoke_before_llm_call_hooks(
formatted_messages, from_agent
):
raise ValueError("LLM call blocked by before_llm_call hook")
self._invoke_before_llm_call_hooks(formatted_messages, from_agent)
if self._effective_api() == "responses":
return self._call_responses(
@@ -486,6 +489,9 @@ class OpenAICompletion(BaseLLM):
response_model=response_model,
)
except (HookAborted, LLMCallBlockedError) as e:
self._emit_call_denied_event(e, from_task, from_agent)
raise
except Exception as e:
error_msg = f"OpenAI API call failed: {e!s}"
logging.error(error_msg)
@@ -600,6 +606,8 @@ class OpenAICompletion(BaseLLM):
formatted_messages = self._format_messages(messages)
self._invoke_before_llm_call_hooks(formatted_messages, from_agent)
if self._effective_api() == "responses":
return await self._acall_responses(
messages=formatted_messages,
@@ -619,6 +627,9 @@ class OpenAICompletion(BaseLLM):
response_model=response_model,
)
except (HookAborted, LLMCallBlockedError) as e:
self._emit_call_denied_event(e, from_task, from_agent)
raise
except Exception as e:
error_msg = f"OpenAI API call failed: {e!s}"
logging.error(error_msg)

16
lib/crewai/src/crewai/memory/analyze.py
View File

@@ -168,6 +168,8 @@ def extract_memories_from_content(content: str, llm: Any) -> list[str]:
Returns:
List of short, self-contained memory statements (or [content] on failure).
"""
from crewai.hooks.dispatch import HookAborted
if not (content or "").strip():
return []
user = _get_prompt("extract_memories_user").format(content=content)
@@ -188,6 +190,8 @@ def extract_memories_from_content(content: str, llm: Any) -> list[str]:
data = json.loads(response)
return ExtractedMemories.model_validate(data).memories
return ExtractedMemories.model_validate(response).memories
except HookAborted:
raise
except Exception as e:
_logger.warning(
"Memory extraction failed, storing full content as single memory: %s",
@@ -216,6 +220,8 @@ def analyze_query(
Returns:
QueryAnalysis with keywords, suggested_scopes, complexity, recall_queries, time_filter.
"""
from crewai.hooks.dispatch import HookAborted
scope_desc = ""
if scope_info:
scope_desc = f"Current scope has {scope_info.record_count} records, categories: {scope_info.categories}"
@@ -241,6 +247,8 @@ def analyze_query(
data = json.loads(response)
return QueryAnalysis.model_validate(data)
return QueryAnalysis.model_validate(response)
except HookAborted:
raise
except Exception as e:
_logger.warning(
"Query analysis failed, using defaults (complexity=simple): %s",
@@ -284,6 +292,8 @@ def analyze_for_save(
Returns:
MemoryAnalysis with suggested_scope, categories, importance, extracted_metadata.
"""
from crewai.hooks.dispatch import HookAborted
user = _get_prompt("save_user").format(
content=content,
existing_scopes=existing_scopes or ["/"],
@@ -306,6 +316,8 @@ def analyze_for_save(
data = json.loads(response)
return MemoryAnalysis.model_validate(data)
return MemoryAnalysis.model_validate(response)
except HookAborted:
raise
except Exception as e:
_logger.warning(
"Memory save analysis failed, using defaults: %s",
@@ -336,6 +348,8 @@ def analyze_for_consolidation(
Returns:
ConsolidationPlan with actions per record and whether to insert the new content.
"""
from crewai.hooks.dispatch import HookAborted
if not existing_records:
return ConsolidationPlan(actions=[], insert_new=True)
records_lines: list[str] = []
@@ -366,6 +380,8 @@ def analyze_for_consolidation(
data = json.loads(response)
return ConsolidationPlan.model_validate(data)
return ConsolidationPlan.model_validate(response)
except HookAborted:
raise
except Exception as e:
_logger.warning(
"Consolidation analysis failed, defaulting to insert: %s",

4
lib/crewai/src/crewai/memory/recall_flow.py
View File

@@ -296,6 +296,8 @@ class RecallFlow(Flow[RecallState]):
Decrements the exploration budget so the loop terminates.
"""
from crewai.hooks.dispatch import HookAborted
self.state.exploration_budget -= 1
enhanced = []
@@ -321,6 +323,8 @@ class RecallFlow(Flow[RecallState]):
"results": finding["results"],
}
)
except HookAborted:
raise
except Exception:
enhanced.append(
{

6
lib/crewai/src/crewai/tasks/llm_guardrail.py
View File

@@ -105,7 +105,11 @@ class LLMGuardrail:
Tuple[bool, Any]: A tuple containing:
- bool: True if validation passed, False otherwise
- Any: The validation result or error message
Raises:
HookAborted: A `pre_model_call` hook denied the validation call.
"""
from crewai.hooks.dispatch import HookAborted
try:
result = self._validate_output(task_output)
@@ -115,5 +119,7 @@ class LLMGuardrail:
if result.pydantic.valid:
return True, task_output.raw
return False, result.pydantic.feedback
except HookAborted:
raise
except Exception as e:
return False, f"Error while validating the task output: {e!s}"

87
lib/crewai/src/crewai/utilities/agent_utils.py
View File

@@ -24,7 +24,7 @@ from crewai.agents.parser import (
OutputParserError,
parse,
)
from crewai.llms.base_llm import BaseLLM, call_stop_override
from crewai.llms.base_llm import BaseLLM, LLMCallBlockedError, call_stop_override
from crewai.tools import BaseTool as CrewAITool
from crewai.tools.base_tool import BaseTool
from crewai.tools.structured_tool import (
@@ -481,31 +481,41 @@ def enforce_rpm_limit(
request_within_rpm_limit()
@contextlib.contextmanager
def _prepare_llm_call(
executor_context: CrewAgentExecutor | AgentExecutor | LiteAgent | None,
messages: list[LLMMessage],
printer: Printer,
verbose: bool = True,
) -> list[LLMMessage]:
) -> Iterator[list[LLMMessage]]:
"""Shared pre-call logic: run before hooks and resolve messages.
Yields for the duration of the LLM call so the LLM layer knows the hooks
already ran with this executor's context and does not dispatch them twice.
Args:
executor_context: Optional executor context for hook invocation.
messages: The messages to send to the LLM.
printer: Printer instance for output.
verbose: Whether to print output.
Returns:
Yields:
The resolved messages list (may come from executor_context).
Raises:
ValueError: If a before hook blocks the call.
"""
if executor_context is not None:
if not _setup_before_llm_call_hooks(executor_context, printer, verbose=verbose):
raise ValueError("LLM call blocked by before_llm_call hook")
messages = executor_context.messages
return messages
from crewai.hooks.llm_hooks import model_call_hooks_dispatched
if executor_context is None:
yield messages
return
if not _setup_before_llm_call_hooks(executor_context, printer, verbose=verbose):
raise LLMCallBlockedError("LLM call blocked by before_llm_call hook")
with model_call_hooks_dispatched():
yield executor_context.messages
def _validate_and_finalize_llm_response(
@@ -577,17 +587,18 @@ def get_llm_response(
Exception: If an error occurs.
ValueError: If the response is None or empty.
"""
messages = _prepare_llm_call(executor_context, messages, printer, verbose=verbose)
answer = llm.call(
messages,
tools=tools,
callbacks=callbacks,
available_functions=available_functions,
from_task=from_task,
from_agent=from_agent,
response_model=response_model,
)
with _prepare_llm_call(
executor_context, messages, printer, verbose=verbose
) as prepared_messages:
answer = llm.call(
prepared_messages,
tools=tools,
callbacks=callbacks,
available_functions=available_functions,
from_task=from_task,
from_agent=from_agent,
response_model=response_model,
)
return _validate_and_finalize_llm_response(
answer, executor_context, printer, verbose=verbose
@@ -630,17 +641,18 @@ async def aget_llm_response(
Exception: If an error occurs.
ValueError: If the response is None or empty.
"""
messages = _prepare_llm_call(executor_context, messages, printer, verbose=verbose)
answer = await llm.acall(
messages,
tools=tools,
callbacks=callbacks,
available_functions=available_functions,
from_task=from_task,
from_agent=from_agent,
response_model=response_model,
)
with _prepare_llm_call(
executor_context, messages, printer, verbose=verbose
) as prepared_messages:
answer = await llm.acall(
prepared_messages,
tools=tools,
callbacks=callbacks,
available_functions=available_functions,
from_task=from_task,
from_agent=from_agent,
response_model=response_model,
)
return _validate_and_finalize_llm_response(
answer, executor_context, printer, verbose=verbose
@@ -1953,16 +1965,23 @@ def _setup_before_llm_call_hooks(
verbose: Whether to print output.
Returns:
True if LLM execution should proceed, False if blocked by a hook.
True if LLM execution should proceed, False if a hook blocked it by
returning ``False``.
Raises:
HookAborted: If a hook raised it, so the deny reaches the caller intact.
"""
if executor_context:
from crewai.hooks.dispatch import (
HookAborted,
InterceptionPoint,
get_scoped_hooks,
run_hooks,
)
from crewai.hooks.llm_hooks import LLMCallHookContext, before_llm_call_reducer
from crewai.hooks.llm_hooks import (
LLMCallHookContext,
LegacyHookBlocked,
before_llm_call_reducer,
)
# Executor snapshot first, then execution-scoped hooks — the same
# ordering dispatch() applies to global vs scoped hooks.
@@ -1984,7 +2003,7 @@ def _setup_before_llm_call_hooks(
reducer=before_llm_call_reducer,
verbose=verbose,
)
except HookAborted:
except LegacyHookBlocked:
if verbose:
printer.print(
content="LLM call blocked by before_llm_call hook",

16
lib/crewai/src/crewai/utilities/converter.py
View File

@@ -93,6 +93,8 @@ class Converter(OutputConverter):
Raises:
ConverterError: If conversion fails after maximum attempts.
"""
from crewai.hooks.dispatch import HookAborted
try:
if self.llm.supports_function_calling():
response = self.llm.call(
@@ -108,6 +110,8 @@ class Converter(OutputConverter):
raise ConverterError(
f"Failed to convert text into a Pydantic model due to validation error: {e}"
) from e
except HookAborted:
raise
except Exception as e:
if current_attempt < self.max_attempts:
return self.to_pydantic(current_attempt + 1)
@@ -117,6 +121,8 @@ class Converter(OutputConverter):
async def ato_pydantic(self, current_attempt: int = 1) -> BaseModel:
"""Async equivalent of ``to_pydantic`` — uses ``acall`` so the event loop is not blocked."""
from crewai.hooks.dispatch import HookAborted
try:
if self.llm.supports_function_calling():
response = await self.llm.acall(
@@ -132,6 +138,8 @@ class Converter(OutputConverter):
raise ConverterError(
f"Failed to convert text into a Pydantic model due to validation error: {e}"
) from e
except HookAborted:
raise
except Exception as e:
if current_attempt < self.max_attempts:
return await self.ato_pydantic(current_attempt + 1)
@@ -152,10 +160,14 @@ class Converter(OutputConverter):
ConverterError: If conversion fails after maximum attempts.
"""
from crewai.hooks.dispatch import HookAborted
try:
if self.llm.supports_function_calling():
return self._create_instructor().to_json()
return json.dumps(self.llm.call(self._build_messages()))
except HookAborted:
raise
except Exception as e:
if current_attempt < self.max_attempts:
return self.to_json(current_attempt + 1)
@@ -168,10 +180,14 @@ class Converter(OutputConverter):
sync-only); we run it via ``asyncio.to_thread`` so the event loop stays
free.
"""
from crewai.hooks.dispatch import HookAborted
try:
if self.llm.supports_function_calling():
return await asyncio.to_thread(self._create_instructor().to_json)
return json.dumps(await self.llm.acall(self._build_messages()))
except HookAborted:
raise
except Exception as e:
if current_attempt < self.max_attempts:
return await self.ato_json(current_attempt + 1)

8
lib/crewai/src/crewai/utilities/crew_chat.py
View File

@@ -417,6 +417,8 @@ def generate_input_description_with_ai(
Returns:
A concise description of the input.
"""
from crewai.hooks.dispatch import HookAborted
context_texts = []
placeholder_pattern = re.compile(r"\{(.+?)}")
@@ -460,6 +462,8 @@ def generate_input_description_with_ai(
)
try:
response = chat_llm.call(messages=[{"role": "user", "content": prompt}])
except HookAborted:
raise
except Exception as exc:
click.secho(
f"Warning: failed to generate input description for '{input_name}' "
@@ -480,6 +484,8 @@ def generate_crew_description_with_ai(crew: Crew, chat_llm: LLM | BaseLLM) -> st
Returns:
A concise description of the crew's purpose (15 words or less).
"""
from crewai.hooks.dispatch import HookAborted
context_texts = []
placeholder_pattern = re.compile(r"\{(.+?)}")
@@ -514,6 +520,8 @@ def generate_crew_description_with_ai(crew: Crew, chat_llm: LLM | BaseLLM) -> st
)
try:
response = chat_llm.call(messages=[{"role": "user", "content": prompt}])
except HookAborted:
raise
except Exception as exc:
click.secho(
f"Warning: failed to generate crew description ({exc}); using default.",

23
lib/crewai/src/crewai/utilities/guardrail.py
View File

@@ -144,6 +144,7 @@ def process_guardrail(
Raises:
TypeError: If output is not a TaskOutput or LiteAgentOutput
ValueError: If guardrail is None
HookAborted: A `pre_model_call` hook denied an LLM-backed guardrail.
"""
from crewai.lite_agent_output import LiteAgentOutput
from crewai.tasks.task_output import TaskOutput
@@ -158,6 +159,7 @@ def process_guardrail(
LLMGuardrailCompletedEvent,
LLMGuardrailStartedEvent,
)
from crewai.hooks.dispatch import HookAborted
started_event = LLMGuardrailStartedEvent(
guardrail=guardrail,
@@ -167,7 +169,26 @@ def process_guardrail(
)
crewai_event_bus.emit(event_source, started_event)
result = guardrail(output)
try:
result = guardrail(output)
except HookAborted as e:
# a deny ends the validation, so the started event above still needs a
# terminal one before it leaves
crewai_event_bus.emit(
event_source,
LLMGuardrailCompletedEvent(
success=False,
result=None,
error=str(e),
retry_count=retry_count,
guardrail_type=started_event.guardrail_type,
guardrail_name=started_event.guardrail_name,
from_agent=from_agent,
from_task=from_task,
),
)
raise
guardrail_result = GuardrailResult.from_tuple(result)
crewai_event_bus.emit(

11
lib/crewai/src/crewai/utilities/internal_instructor.py
View File

@@ -143,8 +143,12 @@ class InternalInstructor(Generic[T]):
Instance of the specified Pydantic model with structured data
Raises:
ValueError: If LLM is not provided or invalid
ValueError: If LLM is not provided or invalid, or if a hook blocked
the call by returning ``False``.
HookAborted: If a hook denied the call.
"""
from crewai.llms.base_llm import BaseLLM
messages: list[LLMMessage] = [{"role": "user", "content": self.content}]
if not _is_valid_llm(self.llm):
@@ -157,6 +161,11 @@ class InternalInstructor(Generic[T]):
else:
model_name = self.llm.model
if isinstance(self.llm, BaseLLM):
# This reaches the provider client directly, so the hooks the LLM
# layer would have dispatched are dispatched here instead.
self.llm._invoke_before_llm_call_hooks(messages, self.agent)
return self._client.chat.completions.create( # type: ignore[no-any-return]
model=model_name, response_model=self.model, messages=messages
)

6
lib/crewai/src/crewai/utilities/reasoning_handler.py
View File

@@ -362,6 +362,8 @@ class AgentReasoning:
"""
self.logger.debug(f"Using function calling for {plan_type} planning")
from crewai.hooks.dispatch import HookAborted
try:
system_prompt = self._get_system_prompt()
@@ -412,6 +414,8 @@ class AgentReasoning:
"READY: I am ready to execute the task." in response_str,
)
except HookAborted:
raise
except Exception as e:
self.logger.warning(
f"Error during function calling: {e!s}. Falling back to text parsing."
@@ -435,6 +439,8 @@ class AgentReasoning:
[],
"READY: I am ready to execute the task." in fallback_str,
)
except HookAborted:
raise
except Exception as inner_e:
self.logger.error(f"Error during fallback text parsing: {inner_e!s}")
return (

309
lib/crewai/tests/hooks/test_deny_reaches_the_caller.py Normal file
View File

@@ -0,0 +1,309 @@
"""A deny raised inside a run has to reach whoever started the run.
The sibling propagation tests all call the frame that makes the model call
directly, so they prove a deny escapes *that* function and nothing about what
its callers do with it. Every regression in this area has lived one or more
frames up, in a broad ``except Exception`` that turned the deny into a degraded
result. These tests drive the public entry points instead, and count model calls
so a deny that gets retried reads as a failure rather than as a pass.
"""
from __future__ import annotations
from typing import Any
from crewai.agent import Agent
from crewai.agents.step_executor import StepExecutor
from crewai.crew import Crew
from crewai.experimental.agent_executor import AgentExecutor
from crewai.hooks.dispatch import HookAborted, InterceptionPoint, clear_all, on
from crewai.lite_agent import LiteAgent
from crewai.llms.base_llm import BaseLLM
from crewai.task import Task
from crewai.utilities.planning_types import TodoItem
from crewai.utilities.step_execution_context import StepExecutionContext
from crewai.utilities.types import LLMMessage
import pytest
from ..utils import wait_for_event_handlers
class StubProviderLLM(BaseLLM):
"""Answers without a network, dispatching the before hooks like a provider."""
def __init__(self, fail_first_call: bool = False) -> None:
super().__init__(model="stub")
self.fail_first_call = fail_first_call
self.answered = 0
def call(
self,
messages: str | list[LLMMessage],
tools: list[dict[str, Any]] | None = None,
callbacks: list[Any] | None = None,
available_functions: dict[str, Any] | None = None,
from_task: Any | None = None,
from_agent: Any | None = None,
response_model: Any | None = None,
**kwargs: Any,
) -> str:
formatted: list[LLMMessage] = (
messages
if isinstance(messages, list)
else [{"role": "user", "content": messages}]
)
self._invoke_before_llm_call_hooks(formatted, from_agent)
self.answered += 1
if self.fail_first_call and self.answered == 1:
raise RuntimeError("the provider blipped")
return "Thought: done\nFinal Answer: ok"
def supports_function_calling(self) -> bool:
return False
class DenyingMemory:
"""Stands in for the memory whose own model call a hook denied."""
read_only = False
root_scope = None
def __init__(self, deny_on: str, error: Exception | None = None) -> None:
self.deny_on = deny_on
self.error = error or HookAborted(
reason="memory is off limits", source="policy"
)
self.touched: list[str] = []
def _step(self, name: str) -> None:
self.touched.append(name)
if name == self.deny_on:
raise self.error
def drain_writes(self) -> None:
pass
def recall(self, *args: Any, **kwargs: Any) -> list[Any]:
self._step("recall")
return []
def extract_memories(self, *args: Any, **kwargs: Any) -> list[str]:
self._step("extract_memories")
return ["a memory"]
def remember_many(self, *args: Any, **kwargs: Any) -> None:
self._step("remember_many")
def search(self, *args: Any, **kwargs: Any) -> list[Any]:
return []
class StubKnowledge:
def query(self, *args: Any, **kwargs: Any) -> list[Any]:
return []
class DenyingStepExecutor:
"""Stands in for the per-step executor whose own model call was denied."""
def __init__(self) -> None:
self.executed = 0
def execute(self, *args: Any, **kwargs: Any) -> Any:
self.executed += 1
raise HookAborted(reason="no model calls allowed", source="policy")
@pytest.fixture(autouse=True)
def _clean_hooks():
clear_all()
yield
# A kickoff emits events whose handlers run on a pool; draining them here
# keeps a straggler from firing inside an unrelated test.
wait_for_event_handlers()
clear_all()
def deny_nth_model_call(n: int) -> list[str]:
"""Deny the nth model call of the run, returning the log of attempts."""
attempts: list[str] = []
@on(InterceptionPoint.PRE_MODEL_CALL)
def gate(_ctx: Any) -> None:
attempts.append("attempt")
if len(attempts) == n:
raise HookAborted(reason="no model calls allowed", source="policy")
return attempts
def build_agent(**kwargs: Any) -> Agent:
return Agent(
role="Worker",
goal="Answer",
backstory="You answer.",
llm=StubProviderLLM(),
**kwargs,
)
def build_crew(agent: Agent, **task_kwargs: Any) -> Crew:
task = Task(
description="Say ok", expected_output="ok", agent=agent, **task_kwargs
)
return Crew(agents=[agent], tasks=[task])
def run_agent_kickoff() -> Any:
return build_agent().kickoff("say ok")
def run_agent_kickoff_with_planning() -> Any:
return build_agent(planning=True).kickoff("say ok")
def run_crew_kickoff() -> Any:
return build_crew(build_agent()).kickoff()
def run_crew_kickoff_with_planning() -> Any:
return build_crew(build_agent(planning=True)).kickoff()
def run_crew_kickoff_with_knowledge() -> Any:
agent = build_agent()
agent.knowledge = StubKnowledge()
return build_crew(agent).kickoff()
def run_lite_agent_kickoff() -> Any:
return LiteAgent(
role="Worker", goal="Answer", backstory="You answer.", llm=StubProviderLLM()
).kickoff("say ok")
@pytest.mark.parametrize(
"entry_point",
[
run_agent_kickoff,
run_agent_kickoff_with_planning,
run_crew_kickoff,
run_crew_kickoff_with_planning,
run_crew_kickoff_with_knowledge,
run_lite_agent_kickoff,
],
ids=[
"agent.kickoff",
"agent.kickoff-planning",
"crew.kickoff",
"crew.kickoff-planning",
"crew.kickoff-knowledge",
"lite_agent.kickoff",
],
)
def test_a_denied_model_call_reaches_the_caller(entry_point):
attempts = deny_nth_model_call(1)
with pytest.raises(HookAborted):
entry_point()
assert len(attempts) == 1
def test_a_denied_guardrail_stops_the_crew_instead_of_retrying_the_task():
attempts = deny_nth_model_call(2)
crew = build_crew(build_agent(), guardrail="The answer must be polite")
with pytest.raises(HookAborted):
crew.kickoff()
# the answer, then the denied validation, and nothing after it
assert len(attempts) == 2
@pytest.mark.parametrize("deny_on", ["recall", "extract_memories"])
def test_a_denied_memory_step_reaches_the_caller(deny_on):
memory = DenyingMemory(deny_on)
agent = build_agent()
agent.memory = memory
with pytest.raises(HookAborted):
agent.kickoff("say ok")
assert memory.touched.count(deny_on) == 1
def test_a_denied_memory_save_stops_the_crew_instead_of_retrying_the_task():
memory = DenyingMemory("extract_memories")
agent = build_agent()
agent.memory = memory
with pytest.raises(HookAborted):
build_crew(agent).kickoff()
assert memory.touched.count("extract_memories") == 1
def build_step_executor(llm: StubProviderLLM) -> StepExecutor:
return StepExecutor(llm=llm, tools=[], agent=build_agent())
def a_step() -> tuple[TodoItem, StepExecutionContext]:
return (
TodoItem(step_number=1, description="Say ok"),
StepExecutionContext(task_description="Say ok", task_goal="ok"),
)
def test_a_denied_step_stops_the_plan_instead_of_reporting_a_failed_step():
attempts = deny_nth_model_call(1)
todo, context = a_step()
with pytest.raises(HookAborted):
build_step_executor(StubProviderLLM()).execute(todo, context)
assert len(attempts) == 1
def test_an_ordinary_step_failure_still_reports_a_failed_step():
todo, context = a_step()
result = build_step_executor(StubProviderLLM(fail_first_call=True)).execute(
todo, context
)
assert result.success is False
@pytest.mark.asyncio
async def test_a_denied_parallel_step_reaches_the_caller():
agent = build_agent()
executor = AgentExecutor(agent=agent, llm=agent.llm, task=None)
executor.state.todos.items = [
TodoItem(step_number=1, description="first"),
TodoItem(step_number=2, description="second"),
]
step_executor = DenyingStepExecutor()
object.__setattr__(executor, "_ensure_step_executor", lambda: step_executor)
with pytest.raises(HookAborted):
await executor.execute_todos_parallel()
def test_an_ordinary_model_failure_is_still_retried():
agent = Agent(
role="Worker",
goal="Answer",
backstory="You answer.",
llm=StubProviderLLM(fail_first_call=True),
)
assert "ok" in str(build_crew(agent).kickoff())
def test_an_ordinary_memory_failure_still_degrades():
memory = DenyingMemory("extract_memories", error=RuntimeError("storage is down"))
agent = build_agent()
agent.memory = memory
assert str(agent.kickoff("say ok")) == "ok"

522
lib/crewai/tests/hooks/test_hook_abort_propagation.py Normal file
View File

@@ -0,0 +1,522 @@
"""A model-call deny must reach the caller as a deny.
Two layers used to erase it. The LLM layer caught ``HookAborted`` and returned
``False``, which every provider turned into ``ValueError("LLM call blocked...")``
— losing the reason, the source, and any way to tell a policy decision from a
provider outage. Downstream, every internal model call is wrapped in
``except Exception`` so a provider hiccup degrades instead of failing the run,
and those handlers then absorbed the flattened deny, often retrying the very
call that was just denied.
A deny also owes the started event a terminal one, and owes it an honest label:
it is a decision, not an outage the provider caused.
These use a real ``LLM`` with a real ``pre_model_call`` hook: the deny fires
inside ``dispatch`` before the provider is reached, so nothing here needs a
network or a cassette.
"""
from __future__ import annotations
from typing import Any
from crewai.agent import Agent
from crewai.agent.planning_config import PlanningConfig
from crewai.agent.utils import (
ahandle_knowledge_retrieval,
handle_knowledge_retrieval,
handle_reasoning,
)
from crewai.agents.planner_observer import PlannerObserver
from crewai.events.event_bus import crewai_event_bus
from crewai.events.types.knowledge_events import (
KnowledgeQueryFailedEvent,
KnowledgeQueryStartedEvent,
)
from crewai.events.types.llm_events import LLMCallFailedEvent
from crewai.events.types.llm_guardrail_events import (
LLMGuardrailCompletedEvent,
LLMGuardrailStartedEvent,
)
from crewai.events.types.observation_events import (
StepObservationFailedEvent,
StepObservationStartedEvent,
)
from crewai.experimental.agent_executor import AgentExecutor
from crewai.hooks.dispatch import (
HookAborted,
InterceptionPoint,
clear_all,
on,
)
from crewai.hooks.llm_hooks import register_before_llm_call_hook
from crewai.llm import LLM
from crewai.memory.analyze import (
analyze_for_consolidation,
analyze_for_save,
analyze_query,
extract_memories_from_content,
)
from crewai.memory.types import MemoryRecord
from crewai.task import Task
from crewai.tasks.llm_guardrail import LLMGuardrail
from crewai.tasks.task_output import TaskOutput
from crewai.utilities.converter import Converter
from crewai.utilities.guardrail import process_guardrail
from crewai.utilities.planning_types import TodoItem
import pytest
from pydantic import BaseModel
from ..utils import wait_for_event_handlers
@pytest.fixture(autouse=True)
def _clean_hooks():
clear_all()
yield
# These calls emit LLM events whose handlers run on a pool; draining them
# here keeps a straggler from firing inside an unrelated test.
wait_for_event_handlers()
clear_all()
@pytest.fixture
def denied_calls() -> list[Any]:
return []
@pytest.fixture
def denying_llm(denied_calls: list[Any]) -> LLM:
@on(InterceptionPoint.PRE_MODEL_CALL)
def deny(ctx: Any) -> None:
denied_calls.append(ctx)
raise HookAborted(reason="no model calls allowed", source="policy")
return LLM(model="gpt-4o-mini")
class _Person(BaseModel):
name: str
def test_a_raised_deny_keeps_its_reason_out_of_the_llm_layer(denying_llm):
with pytest.raises(HookAborted) as exc:
denying_llm.call([{"role": "user", "content": "hi"}])
assert exc.value.reason == "no model calls allowed"
assert exc.value.source == "policy"
def test_the_boolean_convention_still_blocks_with_the_documented_error():
register_before_llm_call_hook(lambda _ctx: False)
with pytest.raises(ValueError, match="LLM call blocked by before_llm_call hook"):
LLM(model="gpt-4o-mini").call([{"role": "user", "content": "hi"}])
def test_the_boolean_convention_is_still_absorbed_by_a_fail_open_handler():
# Unlike a raised abort, the boolean deny must keep degrading rather than
# failing the run — otherwise adopting this fix breaks existing hooks.
register_before_llm_call_hook(lambda _ctx: False)
analysis = analyze_query("a query", ["/"], None, LLM(model="gpt-4o-mini"))
assert analysis.recall_queries == ["a query"]
@pytest.mark.parametrize(
"denied_helper",
[
lambda llm: extract_memories_from_content("some content", llm),
lambda llm: analyze_query("a query", ["/"], None, llm),
lambda llm: analyze_for_save("some content", ["/"], [], llm),
lambda llm: analyze_for_consolidation(
"new content",
[MemoryRecord(id="1", content="old content", scope="/")],
llm,
),
],
ids=["extract", "query", "save", "consolidate"],
)
def test_memory_analysis_surfaces_a_deny_instead_of_a_safe_default(
denied_helper, denying_llm, denied_calls
):
with pytest.raises(HookAborted):
denied_helper(denying_llm)
assert denied_calls, "the helper never reached the model"
def test_a_denied_conversion_is_not_retried(denying_llm, denied_calls):
converter = Converter(
text="Name: Ada",
llm=denying_llm,
model=_Person,
instructions="Extract the person",
max_attempts=3,
)
with pytest.raises(HookAborted):
converter.to_pydantic()
assert len(denied_calls) == 1
def test_a_denied_knowledge_query_still_reports_the_failure_it_started(denying_llm):
agent = Agent(
role="Researcher",
goal="Answer questions",
backstory="You look things up.",
llm=denying_llm,
)
task = Task(
description="What is the capital of France?",
expected_output="A city name.",
agent=agent,
)
started: list[KnowledgeQueryStartedEvent] = []
failed: list[KnowledgeQueryFailedEvent] = []
with crewai_event_bus.scoped_handlers():
@crewai_event_bus.on(KnowledgeQueryStartedEvent)
def _on_started(_source: Any, event: KnowledgeQueryStartedEvent) -> None:
started.append(event)
@crewai_event_bus.on(KnowledgeQueryFailedEvent)
def _on_failed(_source: Any, event: KnowledgeQueryFailedEvent) -> None:
failed.append(event)
with pytest.raises(HookAborted):
agent._get_knowledge_search_query(task.description, task)
wait_for_event_handlers()
assert len(started) == 1
assert len(failed) == 1
assert failed[0].error == "no model calls allowed"
def test_a_denied_step_observation_still_reports_the_failure_it_started(denying_llm):
agent = Agent(
role="Planner",
goal="Plan work",
backstory="You plan.",
llm=denying_llm,
)
observer = PlannerObserver(agent=agent, task=None)
step = TodoItem(step_number=1, description="do the thing", result="done")
started: list[StepObservationStartedEvent] = []
failed: list[StepObservationFailedEvent] = []
with crewai_event_bus.scoped_handlers():
@crewai_event_bus.on(StepObservationStartedEvent)
def _on_started(_source: Any, event: StepObservationStartedEvent) -> None:
started.append(event)
@crewai_event_bus.on(StepObservationFailedEvent)
def _on_failed(_source: Any, event: StepObservationFailedEvent) -> None:
failed.append(event)
with pytest.raises(HookAborted):
observer.observe(step, "done", [], [])
wait_for_event_handlers()
assert len(started) == 1
assert len(failed) == 1
assert failed[0].error == "no model calls allowed"
assert failed[0].step_number == 1
def test_a_denied_knowledge_retrieval_does_not_fall_back_to_the_plain_prompt(
denying_llm,
):
# the retrieval helper wraps the query rewrite in its own except Exception,
# so guarding the rewrite alone still let the task run without knowledge
agent = Agent(
role="Researcher",
goal="Answer questions",
backstory="You look things up.",
llm=denying_llm,
)
# only has to be truthy: the deny fires before any knowledge is queried
agent.knowledge = object()
task = Task(
description="What is the capital of France?",
expected_output="A city name.",
agent=agent,
)
with pytest.raises(HookAborted):
handle_knowledge_retrieval(
agent,
task,
"the task prompt",
{},
lambda *_args, **_kwargs: [],
lambda *_args, **_kwargs: [],
)
@pytest.mark.asyncio
async def test_a_denied_async_knowledge_retrieval_reaches_the_caller(denying_llm):
agent = Agent(
role="Researcher",
goal="Answer questions",
backstory="You look things up.",
llm=denying_llm,
)
agent.knowledge = object()
task = Task(
description="What is the capital of France?",
expected_output="A city name.",
agent=agent,
)
with pytest.raises(HookAborted):
await ahandle_knowledge_retrieval(agent, task, "the task prompt", {})
def test_a_denied_plan_stops_the_legacy_planning_path(denying_llm):
agent = Agent(
role="Planner",
goal="Plan work",
backstory="You plan.",
llm=denying_llm,
planning=True,
)
task = Task(description="Do the thing", expected_output="A result.", agent=agent)
with pytest.raises(HookAborted):
handle_reasoning(agent, task)
def test_a_denied_replan_does_not_keep_executing_the_stale_plan(denying_llm):
agent = Agent(
role="Planner",
goal="Plan work",
backstory="You plan.",
llm=denying_llm,
planning_config=PlanningConfig(
reasoning_effort="low",
max_attempts=1,
max_steps=2,
max_replans=1,
max_step_iterations=2,
),
)
executor = AgentExecutor(agent=agent, llm=denying_llm, task=None)
executor._kickoff_input = "do the thing"
with pytest.raises(HookAborted):
executor._trigger_replan("the first plan failed")
class _DenyingMemory:
"""Stands in for unified memory whose model call was denied upstream."""
read_only = False
root_scope = None
def __init__(self, error: Exception):
self._error = error
def drain_writes(self) -> None:
pass
def recall(self, *_args: Any, **_kwargs: Any):
raise self._error
def extract_memories(self, *_args: Any, **_kwargs: Any):
raise self._error
def remember_many(self, *_args: Any, **_kwargs: Any) -> None:
pass
def test_a_denied_memory_recall_does_not_run_the_task_without_memory():
agent = Agent(role="Doer", goal="Do", backstory="You do.")
agent.memory = _DenyingMemory(HookAborted(reason="no", source="policy"))
task = Task(description="Do the thing", expected_output="A result.", agent=agent)
with pytest.raises(HookAborted):
agent._retrieve_memory_context(task, "the task prompt")
def test_an_ordinary_memory_recall_failure_still_degrades():
# the guard must single out a deny: a broken store has always been allowed
# to degrade to no memory, and turning that into a failed run is a break
agent = Agent(role="Doer", goal="Do", backstory="You do.")
agent.memory = _DenyingMemory(ValueError("vector store is down"))
task = Task(description="Do the thing", expected_output="A result.", agent=agent)
assert agent._retrieve_memory_context(task, "the prompt") == "the prompt"
def test_a_denied_memory_save_does_not_report_a_clean_kickoff():
agent = Agent(role="Doer", goal="Do", backstory="You do.")
agent.memory = _DenyingMemory(HookAborted(reason="no", source="policy"))
with pytest.raises(HookAborted):
agent._save_kickoff_to_memory("the input", "the output")
def test_an_ordinary_memory_save_failure_still_degrades():
agent = Agent(role="Doer", goal="Do", backstory="You do.")
agent.memory = _DenyingMemory(ValueError("vector store is down"))
agent._save_kickoff_to_memory("the input", "the output")
def test_a_denied_guardrail_does_not_read_as_a_failed_validation(denying_llm):
# returning (False, "Error while validating...") would feed the agent a
# retry prompt built from a call the policy refused
guardrail = LLMGuardrail(description="Must be polite", llm=denying_llm)
task_output = TaskOutput(
description="Say hi", raw="hi", agent="Doer", expected_output="A greeting."
)
with pytest.raises(HookAborted):
guardrail(task_output)
def test_a_denied_guardrail_still_reports_the_validation_it_started(denying_llm):
guardrail = LLMGuardrail(description="Must be polite", llm=denying_llm)
task_output = TaskOutput(
description="Say hi", raw="hi", agent="Doer", expected_output="A greeting."
)
started: list[LLMGuardrailStartedEvent] = []
completed: list[LLMGuardrailCompletedEvent] = []
with crewai_event_bus.scoped_handlers():
@crewai_event_bus.on(LLMGuardrailStartedEvent)
def _on_started(_source: Any, event: LLMGuardrailStartedEvent) -> None:
started.append(event)
@crewai_event_bus.on(LLMGuardrailCompletedEvent)
def _on_completed(_source: Any, event: LLMGuardrailCompletedEvent) -> None:
completed.append(event)
with pytest.raises(HookAborted):
process_guardrail(output=task_output, guardrail=guardrail, retry_count=0)
wait_for_event_handlers()
assert len(started) == 1
assert len(completed) == 1
assert completed[0].success is False
assert "no model calls allowed" in (completed[0].error or "")
def test_a_denied_plan_stops_the_executor_instead_of_running_unplanned(denying_llm):
# the executor wraps planning in its own except Exception, so guarding the
# reasoning handler alone still left the agent proceeding with no plan
agent = Agent(
role="Planner",
goal="Plan work",
backstory="You plan.",
llm=denying_llm,
planning_config=PlanningConfig(
reasoning_effort="low",
max_attempts=1,
max_steps=2,
max_replans=0,
max_step_iterations=2,
),
)
executor = AgentExecutor(agent=agent, llm=denying_llm, task=None)
executor._kickoff_input = "do the thing"
with pytest.raises(HookAborted):
executor.generate_plan()
@pytest.mark.asyncio
async def test_an_async_call_is_denied_like_a_sync_one(denying_llm):
with pytest.raises(HookAborted) as exc:
await denying_llm.acall([{"role": "user", "content": "hi"}])
assert exc.value.reason == "no model calls allowed"
assert exc.value.source == "policy"
def test_a_denied_structured_conversion_reaches_the_caller(denying_llm):
# The function-calling path goes through Instructor, which reaches the
# provider client without passing through ``llm.call``.
assert denying_llm.supports_function_calling()
converter = Converter(
text="Name: Ada",
llm=denying_llm,
model=_Person,
instructions="Extract the person",
max_attempts=1,
)
with pytest.raises(HookAborted):
converter.to_json()
def test_a_deny_is_not_reported_as_a_provider_failure(denying_llm):
failures: list[LLMCallFailedEvent] = []
with crewai_event_bus.scoped_handlers():
@crewai_event_bus.on(LLMCallFailedEvent)
def _on_failed(_source: Any, event: LLMCallFailedEvent) -> None:
failures.append(event)
with pytest.raises(HookAborted):
denying_llm.call([{"role": "user", "content": "hi"}])
wait_for_event_handlers()
assert len(failures) == 1
assert failures[0].error == "LLM call denied by policy: no model calls allowed"
def test_a_deny_names_the_hook_that_raised_it_rather_than_its_repr():
def gate_on_approved_models(_ctx: Any) -> None:
raise HookAborted(reason="model not approved", source=gate_on_approved_models)
on(InterceptionPoint.PRE_MODEL_CALL)(gate_on_approved_models)
failures: list[LLMCallFailedEvent] = []
with crewai_event_bus.scoped_handlers():
@crewai_event_bus.on(LLMCallFailedEvent)
def _on_failed(_source: Any, event: LLMCallFailedEvent) -> None:
failures.append(event)
with pytest.raises(HookAborted):
LLM(model="gpt-4o-mini").call([{"role": "user", "content": "hi"}])
wait_for_event_handlers()
assert len(failures) == 1
assert failures[0].error == (
"LLM call denied by gate_on_approved_models: model not approved"
)
def test_the_boolean_convention_is_not_reported_as_a_provider_failure():
register_before_llm_call_hook(lambda _ctx: False)
failures: list[LLMCallFailedEvent] = []
with crewai_event_bus.scoped_handlers():
@crewai_event_bus.on(LLMCallFailedEvent)
def _on_failed(_source: Any, event: LLMCallFailedEvent) -> None:
failures.append(event)
with pytest.raises(ValueError):
LLM(model="gpt-4o-mini").call([{"role": "user", "content": "hi"}])
wait_for_event_handlers()
assert len(failures) == 1
assert failures[0].error == (
"LLM call denied by hook: LLM call blocked by before_llm_call hook"
)

24
lib/crewai/tests/hooks/test_llm_hooks.py
View File

@@ -646,11 +646,27 @@ class TestDirectLLMScopedHooks:
with scoped_hooks():
register_scoped(InterceptionPoint.PRE_MODEL_CALL, block)
proceed = llm._invoke_before_llm_call_hooks(
[{"role": "user", "content": "hi"}], from_agent=None
)
with pytest.raises(HookAborted, match="blocked by scoped hook"):
llm._invoke_before_llm_call_hooks(
[{"role": "user", "content": "hi"}], from_agent=None
)
assert proceed is False
def test_a_scoped_hook_returning_false_blocks_the_call(self):
from crewai.hooks import InterceptionPoint
from crewai.hooks.dispatch import register_scoped, scoped_hooks
from crewai.llms.base_llm import LLMCallBlockedError
llm = self._stub_llm()
with scoped_hooks():
register_scoped(InterceptionPoint.PRE_MODEL_CALL, lambda _ctx: False)
with pytest.raises(LLMCallBlockedError) as blocked:
llm._invoke_before_llm_call_hooks(
[{"role": "user", "content": "hi"}], from_agent=None
)
# a ValueError so the fail-open handlers around internal calls still absorb it
assert isinstance(blocked.value, ValueError)
def test_scoped_after_hook_modifies_direct_response(self):
from crewai.hooks import InterceptionPoint

149
lib/crewai/tests/hooks/test_model_call_hook_reach.py Normal file
View File

@@ -0,0 +1,149 @@
"""Which model calls the ``pre_model_call`` hooks can see, and how many times.
The LLM layer used to skip the hooks whenever a call carried an agent, assuming
the executor had already dispatched them. That holds inside the executor loop
and left every other agent-bearing call — step observation, planning, plan
synthesis — invisible. The executor now marks the window where it already
dispatched, which is what keeps a call from being seen twice.
The stub stands in for a native provider: it invokes the before hooks the way
every provider does and answers without a network.
"""
from __future__ import annotations
from types import SimpleNamespace
from typing import Any
from crewai.agent import Agent
from crewai.hooks.dispatch import InterceptionPoint, clear_all, on
from crewai.llms.base_llm import BaseLLM
from crewai.utilities.agent_utils import get_llm_response
from crewai.utilities.types import LLMMessage
from crewai_core.printer import Printer
from pydantic import BaseModel
import pytest
from ..utils import wait_for_event_handlers
class StubProviderLLM(BaseLLM):
def __init__(self) -> None:
super().__init__(model="stub")
def call(
self,
messages: str | list[LLMMessage],
tools: list[dict[str, Any]] | None = None,
callbacks: list[Any] | None = None,
available_functions: dict[str, Any] | None = None,
from_task: Any | None = None,
from_agent: Any | None = None,
response_model: Any | None = None,
**kwargs: Any,
) -> str:
formatted: list[LLMMessage] = (
messages
if isinstance(messages, list)
else [{"role": "user", "content": messages}]
)
self._invoke_before_llm_call_hooks(formatted, from_agent)
return "Thought: done\nFinal Answer: ok"
def supports_function_calling(self) -> bool:
return False
@pytest.fixture(autouse=True)
def _clean_hooks():
clear_all()
yield
# A kickoff emits events whose handlers run on a pool; draining them here
# keeps a straggler from firing inside an unrelated test.
wait_for_event_handlers()
clear_all()
@pytest.fixture
def seen_agents() -> list[str | None]:
roles: list[str | None] = []
@on(InterceptionPoint.PRE_MODEL_CALL)
def record(ctx: Any) -> None:
agent = getattr(ctx, "agent", None)
roles.append(getattr(agent, "role", None))
return roles
def test_the_executor_loop_is_seen_exactly_once(seen_agents):
agent = Agent(
role="Worker",
goal="Answer",
backstory="You answer.",
llm=StubProviderLLM(),
)
assert str(agent.kickoff("say ok")) == "ok"
assert seen_agents == ["Worker"]
def test_a_direct_call_carrying_an_agent_is_seen_once(seen_agents):
agent = Agent(
role="Planner",
goal="Plan",
backstory="You plan.",
llm=StubProviderLLM(),
)
StubProviderLLM().call([{"role": "user", "content": "hi"}], from_agent=agent)
assert seen_agents == ["Planner"]
def test_a_direct_call_without_an_agent_is_seen_once(seen_agents):
StubProviderLLM().call([{"role": "user", "content": "hi"}])
assert seen_agents == [None]
def test_a_structured_litellm_call_is_seen_once(seen_agents, monkeypatch):
pytest.importorskip("litellm")
import instructor
from crewai.llm import LLM
class Answer(BaseModel):
text: str
client = SimpleNamespace(
chat=SimpleNamespace(
completions=SimpleNamespace(create=lambda **_: Answer(text="ok"))
)
)
monkeypatch.setattr(instructor, "from_litellm", lambda *_, **__: client)
llm = LLM(model="openai/gpt-4o-mini", is_litellm=True)
llm.call([{"role": "user", "content": "hi"}], response_model=Answer)
assert seen_agents == [None]
def test_an_agent_call_outside_the_executor_loop_is_seen_once(seen_agents):
agent = Agent(
role="Worker",
goal="Answer",
backstory="You answer.",
llm=StubProviderLLM(),
)
get_llm_response(
llm=StubProviderLLM(),
messages=[{"role": "user", "content": "hi"}],
callbacks=[],
printer=Printer(),
from_agent=agent,
executor_context=None,
)
assert seen_agents == ["Worker"]