andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-07-25 16:55:11 +00:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

Merge origin/main, keep click floor alongside setuptools override

Browse Source
This commit is contained in:
Joao Moura
2026-07-16 21:40:29 -07:00
parent 2fc4bc8a80 df2e68fe0a
commit 536a02bcab
1141 changed files with 247739 additions and 1583 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
lib/cli/pyproject.toml
View File

@@ -8,7 +8,7 @@ authors = [
]
requires-python = ">=3.10, <3.14"
dependencies = [
"crewai-core==1.15.2",
"crewai-core==1.15.3",
"click>=8.1.7,<9",
"pydantic>=2.11.9,<2.13",
"pydantic-settings~=2.10.1",

2
lib/cli/src/crewai_cli/__init__.py
View File

@@ -1 +1 @@
__version__ = "1.15.2"
__version__ = "1.15.3"

2
lib/crewai-core/src/crewai_core/__init__.py
View File

@@ -1 +1 @@
__version__ = "1.15.2"
__version__ = "1.15.3"

17
lib/crewai-core/src/crewai_core/plus_api.py
View File

@@ -149,7 +149,13 @@ class PlusAPI:
EPHEMERAL_TRACING_RESOURCE: Final = "/crewai_plus/api/v1/tracing/ephemeral"
INTEGRATIONS_RESOURCE: Final = "/crewai_plus/api/v1/integrations"
def __init__(self, api_key: str | None = None) -> None:
def __init__(
self,
api_key: str | None = None,
*,
base_url: str | None = None,
organization_id: str | None = None,
) -> None:
version = get_crewai_version()
self.api_key = api_key
self.headers: Headers = {
@@ -161,12 +167,13 @@ class PlusAPI:
self.headers["Authorization"] = f"Bearer {api_key}"
settings = Settings()
if settings.org_uuid:
self.headers["X-Crewai-Organization-Id"] = settings.org_uuid
if organization_id := organization_id or settings.org_uuid:
self.headers["X-Crewai-Organization-Id"] = organization_id
self.base_url = (
os.getenv("CREWAI_PLUS_URL")
or str(settings.enterprise_base_url)
base_url
or os.getenv("CREWAI_PLUS_URL")
or settings.enterprise_base_url
or DEFAULT_CREWAI_ENTERPRISE_URL
)

2
lib/crewai-files/src/crewai_files/__init__.py
View File

@@ -152,4 +152,4 @@ __all__ = [
"wrap_file_source",
]
__version__ = "1.15.2"
__version__ = "1.15.3"

2
lib/crewai-tools/pyproject.toml
View File

@@ -10,7 +10,7 @@ requires-python = ">=3.10, <3.14"
dependencies = [
"pytube~=15.0.0",
"requests>=2.33.0,<3",
"crewai==1.15.2",
"crewai==1.15.3",
"tiktoken>=0.8.0,<0.13",
"beautifulsoup4~=4.13.4",
"python-docx~=1.2.0",

2
lib/crewai-tools/src/crewai_tools/__init__.py
View File

@@ -330,4 +330,4 @@ __all__ = [
"ZapierActionTools",
]
__version__ = "1.15.2"
__version__ = "1.15.3"

6
lib/crewai/pyproject.toml
View File

@@ -8,8 +8,8 @@ authors = [
]
requires-python = ">=3.10, <3.14"
dependencies = [
"crewai-core==1.15.2",
"crewai-cli==1.15.2",
"crewai-core==1.15.3",
"crewai-cli==1.15.3",
# Core Dependencies
"pydantic>=2.11.9,<2.13",
"openai>=2.30.0,<3",
@@ -55,7 +55,7 @@ Repository = "https://github.com/crewAIInc/crewAI"
[project.optional-dependencies]
tools = [
"crewai-tools==1.15.2",
"crewai-tools==1.15.3",
]
embeddings = [
"tiktoken>=0.8.0,<0.13"

2
lib/crewai/src/crewai/__init__.py
View File

@@ -48,7 +48,7 @@ def _suppress_pydantic_deprecation_warnings() -> None:
_suppress_pydantic_deprecation_warnings()
__version__ = "1.15.2"
__version__ = "1.15.3"
_LAZY_IMPORTS: dict[str, tuple[str, str]] = {
"Memory": ("crewai.memory.unified_memory", "Memory"),

35
lib/crewai/src/crewai/agents/crew_agent_executor.py
View File

@@ -46,8 +46,8 @@ from crewai.hooks.llm_hooks import (
)
from crewai.hooks.tool_hooks import (
ToolCallHookContext,
get_after_tool_call_hooks,
get_before_tool_call_hooks,
run_after_tool_call_hooks,
run_before_tool_call_hooks,
)
from crewai.types.callback import SerializableCallable
from crewai.utilities.agent_utils import (
@@ -951,7 +951,6 @@ class CrewAgentExecutor(BaseAgentExecutor):
track_delegation_if_needed(func_name, args_dict or {}, self.task)
hook_blocked = False
before_hook_context = ToolCallHookContext(
tool_name=func_name,
tool_input=args_dict or {},
@@ -960,19 +959,7 @@ class CrewAgentExecutor(BaseAgentExecutor):
task=self.task,
crew=self.crew,
)
before_hooks = get_before_tool_call_hooks()
try:
for hook in before_hooks:
hook_result = hook(before_hook_context)
if hook_result is False:
hook_blocked = True
break
except Exception as hook_error:
if self.agent.verbose:
PRINTER.print(
content=f"Error in before_tool_call hook: {hook_error}",
color="red",
)
hook_blocked = run_before_tool_call_hooks(before_hook_context)
if hook_blocked:
result = f"Tool execution blocked by hook. Tool: {func_name}"
@@ -1033,19 +1020,9 @@ class CrewAgentExecutor(BaseAgentExecutor):
tool_result=result,
raw_tool_result=raw_tool_result,
)
after_hooks = get_after_tool_call_hooks()
try:
for after_hook in after_hooks:
after_hook_result = after_hook(after_hook_context)
if after_hook_result is not None:
result = after_hook_result
after_hook_context.tool_result = result
except Exception as hook_error:
if self.agent.verbose:
PRINTER.print(
content=f"Error in after_tool_call hook: {hook_error}",
color="red",
)
modified_result = run_after_tool_call_hooks(after_hook_context)
if modified_result is not None:
result = modified_result
if not error_event_emitted:
crewai_event_bus.emit(

33
lib/crewai/src/crewai/crew.py
View File

@@ -1906,6 +1906,31 @@ class Crew(FlowTrackable, BaseModel):
final_string_output = final_task_output.raw
self._finish_execution(final_string_output)
self.token_usage = self.calculate_usage_metrics()
from crewai.hooks.contexts import ExecutionEndContext, OutputContext
from crewai.hooks.dispatch import InterceptionPoint, dispatch
crew_output = CrewOutput(
raw=final_task_output.raw,
pydantic=final_task_output.pydantic,
json_dict=final_task_output.json_dict,
tasks_output=task_outputs,
token_usage=self.token_usage,
)
output_ctx = OutputContext(crew=self, output=crew_output, payload=crew_output)
dispatch(InterceptionPoint.OUTPUT, output_ctx)
crew_output = cast(CrewOutput, output_ctx.payload)
end_ctx = ExecutionEndContext(
crew=self, output=crew_output, payload=crew_output
)
dispatch(InterceptionPoint.EXECUTION_END, end_ctx)
crew_output = cast(CrewOutput, end_ctx.payload)
if isinstance(crew_output, CrewOutput):
final_task_output.raw = crew_output.raw
# Ensure background memory saves finish (and emit their
# completed/failed events) before the kickoff-completed event below
# triggers listener teardown/finalization.
@@ -1924,13 +1949,7 @@ class Crew(FlowTrackable, BaseModel):
# Finalization is handled by trace listener (always initialized)
# The batch manager checks contextvar to determine if tracing is enabled
return CrewOutput(
raw=final_task_output.raw,
pydantic=final_task_output.pydantic,
json_dict=final_task_output.json_dict,
tasks_output=task_outputs,
token_usage=self.token_usage,
)
return crew_output
def _process_async_tasks(
self,

22
lib/crewai/src/crewai/crews/utils.py
View File

@@ -278,6 +278,9 @@ def prepare_kickoff(
reset_emission_counter()
reset_last_event_id()
from crewai.hooks.contexts import ExecutionStartContext, InputContext
from crewai.hooks.dispatch import InterceptionPoint, dispatch
normalized: dict[str, Any] | None = None
if inputs is not None:
if not isinstance(inputs, Mapping):
@@ -286,11 +289,30 @@ def prepare_kickoff(
)
normalized = dict(inputs)
# ``inputs`` aliases the same object as ``payload`` (not a fresh ``{}`` from
# ``or``) so in-place edits to either survive read-back, per the context
# contract. ``None`` inputs are preserved rather than coerced to ``{}``.
start_ctx = ExecutionStartContext(
crew=crew,
inputs=normalized if normalized is not None else {},
payload=normalized,
)
dispatch(InterceptionPoint.EXECUTION_START, start_ctx)
normalized = start_ctx.payload
for before_callback in crew.before_kickoff_callbacks:
if normalized is None:
normalized = {}
normalized = before_callback(normalized)
input_ctx = InputContext(
crew=crew,
inputs=normalized if normalized is not None else {},
payload=normalized,
)
dispatch(InterceptionPoint.INPUT, input_ctx)
normalized = input_ctx.payload
if resuming and crew._kickoff_event_id:
if crew.verbose:
from crewai.events.utils.console_formatter import ConsoleFormatter

19
lib/crewai/src/crewai/events/types/hook_events.py Normal file
View File

@@ -0,0 +1,19 @@
from typing import Literal
from crewai.events.base_events import BaseEvent
class HookDispatchedEvent(BaseEvent):
"""Event emitted whenever an interception point dispatches to hooks.
Only emitted when at least one hook is registered for the point, so the
no-op fast path stays free of event overhead.
"""
type: Literal["hook_dispatched"] = "hook_dispatched"
interception_point: str
outcome: Literal["proceeded", "modified", "aborted"]
hook_count: int
duration_ms: float
abort_reason: str | None = None
abort_source: str | None = None

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

@@ -62,8 +62,8 @@ from crewai.hooks.llm_hooks import (
)
from crewai.hooks.tool_hooks import (
ToolCallHookContext,
get_after_tool_call_hooks,
get_before_tool_call_hooks,
run_after_tool_call_hooks,
run_before_tool_call_hooks,
)
from crewai.hooks.types import (
AfterLLMCallHookCallable,
@@ -1975,7 +1975,6 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
track_delegation_if_needed(func_name, args_dict, self.task)
hook_blocked = False
before_hook_context = ToolCallHookContext(
tool_name=func_name,
tool_input=args_dict,
@@ -1984,19 +1983,7 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
task=self.task,
crew=self.crew,
)
before_hooks = get_before_tool_call_hooks()
try:
for hook in before_hooks:
hook_result = hook(before_hook_context)
if hook_result is False:
hook_blocked = True
break
except Exception as hook_error:
if self.agent.verbose:
PRINTER.print(
content=f"Error in before_tool_call hook: {hook_error}",
color="red",
)
hook_blocked = run_before_tool_call_hooks(before_hook_context)
if hook_blocked:
result = f"Tool execution blocked by hook. Tool: {func_name}"
@@ -2060,19 +2047,9 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
tool_result=result,
raw_tool_result=raw_tool_result,
)
after_hooks = get_after_tool_call_hooks()
try:
for after_hook in after_hooks:
after_hook_result = after_hook(after_hook_context)
if after_hook_result is not None:
result = after_hook_result
after_hook_context.tool_result = result
except Exception as hook_error:
if self.agent.verbose:
PRINTER.print(
content=f"Error in after_tool_call hook: {hook_error}",
color="red",
)
modified_result = run_after_tool_call_hooks(after_hook_context)
if modified_result is not None:
result = modified_result
if not error_event_emitted:
crewai_event_bus.emit(

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

@@ -1476,6 +1476,19 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
else (resumed_method_output if emit else result)
)
from crewai.hooks.contexts import ExecutionEndContext, OutputContext
from crewai.hooks.dispatch import InterceptionPoint, dispatch
output_ctx = OutputContext(flow=self, output=final_result, payload=final_result)
dispatch(InterceptionPoint.OUTPUT, output_ctx)
final_result = output_ctx.payload
end_ctx = ExecutionEndContext(
flow=self, output=final_result, payload=final_result
)
dispatch(InterceptionPoint.EXECUTION_END, end_ctx)
final_result = end_ctx.payload
if self._event_futures:
await asyncio.gather(
*[
@@ -2037,6 +2050,9 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
flow_name_token = None
flow_defer_trace_finalization_token = None
request_id_token = None
# Re-published after the INPUT hook so trigger-payload injection reads
# the hook-rewritten inputs rather than the pre-hook baggage above.
flow_inputs_token = None
if current_flow_id.get() is None:
flow_id_token = current_flow_id.set(self.flow_id)
flow_name_token = current_flow_name.set(
@@ -2062,6 +2078,37 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
self._attach_usage_aggregation_listener()
try:
from crewai.hooks.contexts import (
ExecutionEndContext,
ExecutionStartContext,
InputContext,
OutputContext,
)
from crewai.hooks.dispatch import InterceptionPoint, dispatch
# ``inputs`` aliases the same object as ``payload`` (not a fresh
# ``{}`` from ``or``) so in-place edits survive read-back.
start_ctx = ExecutionStartContext(
flow=self,
inputs=inputs if inputs is not None else {},
payload=inputs,
)
dispatch(InterceptionPoint.EXECUTION_START, start_ctx)
inputs = start_ctx.payload
input_ctx = InputContext(
flow=self,
inputs=inputs if inputs is not None else {},
payload=inputs,
)
dispatch(InterceptionPoint.INPUT, input_ctx)
inputs = input_ctx.payload
# Publish the resolved inputs so trigger-payload injection and other
# baggage readers observe hook rewrites (the baggage set before the
# hooks carried the pre-hook inputs).
flow_inputs_token = attach(baggage.set_baggage("flow_inputs", inputs or {}))
# Reset flow state for fresh execution unless restoring from persistence
is_restoring = (
inputs and "id" in inputs and self.persistence is not None
@@ -2297,6 +2344,21 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
method_outputs = self.method_outputs
final_output = method_outputs[-1] if method_outputs else None
output_ctx = OutputContext(
flow=self, output=final_output, payload=final_output
)
dispatch(InterceptionPoint.OUTPUT, output_ctx)
final_output = output_ctx.payload
# EXECUTION_END runs before FlowFinishedEvent so a HookAborted
# prevents a spurious finished signal and payload replacement is
# honored on the emitted result and the returned value.
end_ctx = ExecutionEndContext(
flow=self, output=final_output, payload=final_output
)
dispatch(InterceptionPoint.EXECUTION_END, end_ctx)
final_output = end_ctx.payload
if self._event_futures:
await asyncio.gather(
*[asyncio.wrap_future(f) for f in self._event_futures]
@@ -2370,6 +2432,8 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
current_flow_name.reset(flow_name_token)
if flow_id_token is not None:
current_flow_id.reset(flow_id_token)
if flow_inputs_token is not None:
detach(flow_inputs_token)
detach(flow_token)
crewai_event_bus._exit_runtime_scope(runtime_scope)
@@ -2562,6 +2626,37 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
if future:
self._event_futures.append(future)
from crewai.hooks.contexts import StepContext
from crewai.hooks.dispatch import InterceptionPoint, dispatch
pre_step_ctx = StepContext(
kind="flow_method",
step_name=str(method_name),
flow=self,
payload=dumped_params,
)
dispatch(InterceptionPoint.PRE_STEP, pre_step_ctx)
# Apply hook edits/replacement of the step params back onto the
# call. ``dumped_params`` maps positional args to ``_0, _1, ...``
# keys and keeps kwargs by name, so reverse that mapping here.
updated_params = pre_step_ctx.payload
if isinstance(updated_params, dict):
positional = sorted(
(
k
for k in updated_params
if k.startswith("_") and k[1:].isdigit()
),
key=lambda k: int(k[1:]),
)
args = tuple(updated_params[k] for k in positional)
kwargs = {
k: v
for k, v in updated_params.items()
if not (k.startswith("_") and k[1:].isdigit())
}
# Set method name in context so ask() can read it without
# stack inspection. Must happen before copy_context() so the
# value propagates into the thread pool for sync methods.
@@ -2589,6 +2684,16 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
method_name, method_definition.human_feedback, result
)
post_step_ctx = StepContext(
kind="flow_method",
step_name=str(method_name),
flow=self,
output=result,
payload=result,
)
dispatch(InterceptionPoint.POST_STEP, post_step_ctx)
result = post_step_ctx.payload
self._method_outputs.append({"method": str(method_name), "output": result})
# For @human_feedback methods with emit, the result is the collapsed outcome

20
lib/crewai/src/crewai/hooks/__init__.py
View File

@@ -6,6 +6,17 @@ from crewai.hooks.decorators import (
before_llm_call,
before_tool_call,
)
from crewai.hooks.dispatch import (
HookAborted,
InterceptionPoint,
clear as clear_hooks,
clear_all as clear_all_hooks,
dispatch,
get_hooks,
on,
register as register_hook,
unregister as unregister_hook,
)
from crewai.hooks.llm_hooks import (
LLMCallHookContext,
clear_after_llm_call_hooks,
@@ -74,6 +85,8 @@ def clear_all_global_hooks() -> dict[str, tuple[int, int]]:
__all__ = [
"HookAborted",
"InterceptionPoint",
"LLMCallHookContext",
"ToolCallHookContext",
"after_llm_call",
@@ -83,20 +96,27 @@ __all__ = [
"clear_after_llm_call_hooks",
"clear_after_tool_call_hooks",
"clear_all_global_hooks",
"clear_all_hooks",
"clear_all_llm_call_hooks",
"clear_all_tool_call_hooks",
"clear_before_llm_call_hooks",
"clear_before_tool_call_hooks",
"clear_hooks",
"dispatch",
"get_after_llm_call_hooks",
"get_after_tool_call_hooks",
"get_before_llm_call_hooks",
"get_before_tool_call_hooks",
"get_hooks",
"on",
"register_after_llm_call_hook",
"register_after_tool_call_hook",
"register_before_llm_call_hook",
"register_before_tool_call_hook",
"register_hook",
"unregister_after_llm_call_hook",
"unregister_after_tool_call_hook",
"unregister_before_llm_call_hook",
"unregister_before_tool_call_hook",
"unregister_hook",
]

71
lib/crewai/src/crewai/hooks/contexts.py Normal file
View File

@@ -0,0 +1,71 @@
"""Typed contexts for the interception points wired in phases 2-5.
Each context is a dataclass whose fields are nullable and defaulted, so a field
that is not meaningful for a given runtime (e.g. ``agent_role`` inside a flow)
is simply ``None`` rather than an error. Every context exposes a ``payload``
field: the interceptable value a hook may mutate in place or replace by
returning a new value.
The legacy ``pre/post_model_call`` and ``pre/post_tool_call`` points keep using
:class:`~crewai.hooks.llm_hooks.LLMCallHookContext` and
:class:`~crewai.hooks.tool_hooks.ToolCallHookContext` for backwards
compatibility; they are intentionally not redefined here.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
@dataclass
class InterceptionContext:
"""Base context shared by the framework-native interception points."""
payload: Any = None
agent: Any = None
agent_role: str | None = None
task: Any = None
crew: Any = None
flow: Any = None
@dataclass
class ExecutionStartContext(InterceptionContext):
"""``execution_start``: a crew or flow is about to begin. ``payload`` = inputs."""
inputs: dict[str, Any] = field(default_factory=dict)
@dataclass
class InputContext(InterceptionContext):
"""``input``: resolved inputs for an execution. ``payload`` = inputs."""
inputs: dict[str, Any] = field(default_factory=dict)
@dataclass
class OutputContext(InterceptionContext):
"""``output``: final result of a crew or flow. ``payload`` = the output object."""
output: Any = None
@dataclass
class ExecutionEndContext(InterceptionContext):
"""``execution_end``: a crew or flow has finished. ``payload`` = the output object."""
output: Any = None
@dataclass
class StepContext(InterceptionContext):
"""``pre_step`` / ``post_step``: a task or flow-method step boundary.
``kind`` is ``"task"`` for crew tasks and ``"flow_method"`` for flow methods.
``payload`` is the step input (pre) or step output (post).
"""
kind: str | None = None
step_name: str | None = None
output: Any = None

432
lib/crewai/src/crewai/hooks/dispatch.py Normal file
View File

@@ -0,0 +1,432 @@
"""Generic interception-hook dispatcher.
This module is the single engine behind every CrewAI interception point. A hook
receives a typed context, may mutate it in place and/or return a replacement
payload, and may raise :class:`HookAborted` to stop the intercepted operation
with a reason and source.
The four public hook families (``before/after_llm_call`` and
``before/after_tool_call``) are adapters registered on this dispatcher, so the
legacy dialect (``register_*``/decorators/``return False``) and the new dialect
(``@on(point)`` / ``HookAborted``) share one ordered queue per point.
Design notes:
- Global registration order is preserved; execution-scoped hooks (via
``contextvars``) run after global ones, mirroring
``events/event_bus.py``'s ``_runtime_state_var`` scoping pattern.
- ``dispatch`` has a no-op fast path (a single dict lookup) when no hooks are
registered for a point.
- Hooks are synchronous. They may be invoked from async seams, so they must not
block on heavy I/O (same restriction as the legacy hooks).
- ``HookAborted`` propagates by design. Any other exception raised by a hook is
swallowed (fail-open) to preserve the framework's protection against a buggy
user hook.
"""
from __future__ import annotations
from collections.abc import Callable, Iterator
from contextlib import contextmanager
import contextvars
from enum import Enum
from functools import wraps
import inspect
import time
from typing import Any
from crewai.utilities.string_utils import sanitize_tool_name
class InterceptionPoint(str, Enum):
"""Interception points wired by this layer.
New points are introduced alongside the seams that dispatch them, so the
enum only ever lists points with a live consumer.
"""
# Execution-level boundaries
EXECUTION_START = "execution_start"
INPUT = "input"
OUTPUT = "output"
EXECUTION_END = "execution_end"
# Model / tool boundaries (legacy-compatible)
PRE_MODEL_CALL = "pre_model_call"
POST_MODEL_CALL = "post_model_call"
PRE_TOOL_CALL = "pre_tool_call"
POST_TOOL_CALL = "post_tool_call"
# Step points
PRE_STEP = "pre_step"
POST_STEP = "post_step"
class HookAborted(Exception): # noqa: N818 - public contract name from OSS-86
"""Raised by a hook (or a legacy adapter) to abort the intercepted operation.
Args:
reason: Human-readable explanation of why the operation was aborted.
source: Optional identifier of the aborting hook (callable, string, or
any object). Used for telemetry and failure messages.
"""
def __init__(self, reason: str, source: Any = None) -> None:
super().__init__(reason)
self.reason = reason
self.source = source
HookFn = Callable[[Any], Any]
# (ctx, result) -> modified? A reducer maps a hook's return value onto the
# context using point-specific semantics. It may raise HookAborted.
Reducer = Callable[[Any, Any], bool]
_global_hooks: dict[InterceptionPoint, list[HookFn]] = {
point: [] for point in InterceptionPoint
}
_scoped_hooks_var: contextvars.ContextVar[
dict[InterceptionPoint, list[HookFn]] | None
] = contextvars.ContextVar("crewai_scoped_hooks", default=None)
_TELEMETRY_SOURCE = object()
def get_global_hook_list(point: InterceptionPoint) -> list[HookFn]:
"""Return the live global hook list for a point.
The returned list object is stable for the lifetime of the process, which
lets legacy modules alias their module-level registries to it. Mutate it in
place (append/remove/clear); never rebind it.
"""
return _global_hooks[point]
def register(point: InterceptionPoint, hook: HookFn) -> None:
"""Register a global hook for an interception point."""
_global_hooks[point].append(hook)
def unregister(point: InterceptionPoint, hook: HookFn) -> bool:
"""Unregister a specific global hook. Returns True if it was removed.
When ``hook`` was registered through :func:`on` with ``agents``/``tools``
filters, the stored callable is a wrapper rather than ``hook`` itself. The
wrapper is stashed on ``hook._registered_hook`` at registration time, so it
can be resolved and removed here.
"""
hooks = _global_hooks[point]
target = hook if hook in hooks else getattr(hook, "_registered_hook", hook)
try:
hooks.remove(target)
return True
except ValueError:
return False
def get_hooks(point: InterceptionPoint) -> list[HookFn]:
"""Return a copy of the global hooks registered for a point."""
return _global_hooks[point].copy()
def clear(point: InterceptionPoint) -> int:
"""Clear all global hooks for a point. Returns the number cleared."""
count = len(_global_hooks[point])
_global_hooks[point].clear()
return count
def clear_all() -> None:
"""Clear all global hooks across every interception point."""
for hooks in _global_hooks.values():
hooks.clear()
@contextmanager
def scoped_hooks(
hooks: dict[InterceptionPoint, list[HookFn]] | None = None,
) -> Iterator[dict[InterceptionPoint, list[HookFn]]]:
"""Enter an execution-scoped hook registry.
Hooks registered inside this context (via :func:`register_scoped`) run after
global hooks and are discarded when the context exits. Mirrors the event
bus's scoped-handler pattern.
"""
scope: dict[InterceptionPoint, list[HookFn]] = hooks if hooks is not None else {}
token = _scoped_hooks_var.set(scope)
try:
yield scope
finally:
_scoped_hooks_var.reset(token)
def register_scoped(point: InterceptionPoint, hook: HookFn) -> None:
"""Register a hook scoped to the current :func:`scoped_hooks` context."""
scope = _scoped_hooks_var.get()
if scope is None:
raise RuntimeError(
"register_scoped() called outside of a scoped_hooks() context"
)
scope.setdefault(point, []).append(hook)
def get_scoped_hooks(point: InterceptionPoint) -> list[HookFn]:
"""Return the hooks registered in the current execution scope for a point.
Used by seams that carry a pre-snapshotted hook list (e.g. the agent
executors' per-executor LLM hook lists) so they can merge in
execution-scoped hooks with the same snapshot-then-scoped ordering that
:func:`dispatch` applies to global vs scoped hooks.
"""
scope = _scoped_hooks_var.get()
if not scope:
return []
return list(scope.get(point, []))
def _resolve_hooks(point: InterceptionPoint) -> list[HookFn]:
"""Resolve the ordered hooks for a point: global first, then scoped."""
global_hooks = _global_hooks[point]
scope = _scoped_hooks_var.get()
if scope:
scoped = scope.get(point)
if scoped:
return [*global_hooks, *scoped]
return global_hooks
def _source_name(source: Any) -> str | None:
"""Best-effort readable name for a hook source."""
if source is None:
return None
if isinstance(source, str):
return source
name = getattr(source, "__name__", None)
if isinstance(name, str):
return name
return type(source).__name__
def _emit_telemetry(
point: InterceptionPoint,
outcome: str,
hook_count: int,
duration_ms: float,
abort_reason: str | None,
abort_source: str | None,
) -> None:
"""Emit a HookDispatchedEvent. Never raises."""
try:
from crewai.events.event_bus import crewai_event_bus
from crewai.events.types.hook_events import HookDispatchedEvent
crewai_event_bus.emit(
_TELEMETRY_SOURCE,
event=HookDispatchedEvent(
interception_point=point.value,
outcome=outcome,
hook_count=hook_count,
duration_ms=duration_ms,
abort_reason=abort_reason,
abort_source=abort_source,
),
)
except Exception: # noqa: S110 - telemetry must never break dispatch
pass
def _default_reducer(ctx: Any, result: Any) -> bool:
"""Default payload semantics: a non-None return replaces ``ctx.payload``.
Only reports a modification when the payload was actually applied, so a
context without a ``payload`` attribute does not produce a misleading
``"modified"`` telemetry outcome.
"""
if result is not None and hasattr(ctx, "payload"):
ctx.payload = result
return True
return False
def _invoke_hook(
point: InterceptionPoint,
hook: HookFn,
ctx: Any,
reducer: Reducer,
verbose: bool,
) -> bool:
"""Run a single hook and apply its result via the reducer.
Returns whether the context was modified. Raises :class:`HookAborted` (with
``source`` populated) to abort; any other exception is swallowed (fail-open).
"""
try:
result = hook(ctx)
return reducer(ctx, result)
except HookAborted as aborted:
if aborted.source is None:
aborted.source = hook
raise
except Exception as error:
if verbose:
from crewai_core.printer import PRINTER
PRINTER.print(
content=f"Error in {point.value} hook: {error}",
color="yellow",
)
return False
def run_hooks(
point: InterceptionPoint,
ctx: Any,
hooks: list[HookFn],
*,
reducer: Reducer | None = None,
verbose: bool = True,
) -> Any:
"""Execute an explicit list of hooks against a context.
This is the shared engine used both by :func:`dispatch` (which resolves
global + scoped hooks) and by seams that carry a pre-snapshotted hook list
(e.g. per-executor LLM hook lists).
Args:
point: The interception point being dispatched.
ctx: The typed context passed to each hook (mutated in place).
hooks: The ordered hooks to run.
reducer: Maps each hook's return value onto ``ctx``. Defaults to
:func:`_default_reducer` (payload replacement). May raise
:class:`HookAborted`.
verbose: Whether to print swallowed-hook-error warnings.
Returns:
The (possibly mutated) context.
Raises:
HookAborted: If a hook or the reducer aborts the operation. Telemetry is
still emitted before propagating.
"""
if not hooks:
return ctx
active_reducer = reducer if reducer is not None else _default_reducer
start = time.perf_counter()
outcome = "proceeded"
abort_reason: str | None = None
abort_source: str | None = None
modified = False
try:
for hook in list(hooks):
if _invoke_hook(point, hook, ctx, active_reducer, verbose):
modified = True
outcome = "modified" if modified else "proceeded"
return ctx
except HookAborted as aborted:
outcome = "aborted"
abort_reason = aborted.reason
abort_source = _source_name(aborted.source)
raise
finally:
_emit_telemetry(
point,
outcome,
len(hooks),
(time.perf_counter() - start) * 1000.0,
abort_reason,
abort_source,
)
def dispatch(
point: InterceptionPoint,
ctx: Any,
*,
reducer: Reducer | None = None,
verbose: bool = True,
) -> Any:
"""Dispatch a context to all hooks registered for a point.
Resolves global then scoped hooks and runs them through :func:`run_hooks`.
No-op fast path when nothing is registered.
"""
hooks = _resolve_hooks(point)
if not hooks:
return ctx
return run_hooks(point, ctx, hooks, reducer=reducer, verbose=verbose)
def _wrap_with_filters(
func: HookFn,
agents: list[str] | None,
tools: list[str] | None,
) -> HookFn:
"""Wrap a hook so it only runs for matching agents/tools (context-shape aware)."""
@wraps(func)
def filtered(ctx: Any) -> Any:
if tools:
tool_name = getattr(ctx, "tool_name", None)
if tool_name is not None and tool_name not in tools:
return None
if agents:
agent = getattr(ctx, "agent", None)
role = getattr(agent, "role", None) if agent is not None else None
if role is None:
role = getattr(ctx, "agent_role", None)
if role is not None and role not in agents:
return None
return func(ctx)
return filtered
def on(
point: InterceptionPoint,
*,
agents: list[str] | None = None,
tools: list[str] | None = None,
) -> Callable[[HookFn], HookFn]:
"""Register a function as a hook for an interception point.
Mirrors the legacy decorators' ergonomics: supports ``agents=`` / ``tools=``
filters and, when applied to a method inside a ``@CrewBase`` class, defers
registration to crew initialization (crew-scoped) instead of registering
globally.
Example:
>>> @on(InterceptionPoint.PRE_TOOL_CALL, tools=["delete_file"])
... def guard(ctx):
... raise HookAborted("deletion not allowed")
"""
normalized_tools = [sanitize_tool_name(t) for t in tools] if tools else None
def decorator(func: HookFn) -> HookFn:
func._interception_point = point # type: ignore[attr-defined]
if normalized_tools:
func._filter_tools = normalized_tools # type: ignore[attr-defined]
if agents:
func._filter_agents = agents # type: ignore[attr-defined]
params = list(inspect.signature(func).parameters.keys())
is_method = len(params) >= 2 and params[0] == "self"
if not is_method:
hook = (
_wrap_with_filters(func, agents, normalized_tools)
if (agents or normalized_tools)
else func
)
register(point, hook)
# Remember the actually-registered callable so unregister_hook(func)
# can resolve the filter wrapper.
func._registered_hook = hook # type: ignore[attr-defined]
return func
return decorator

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

@@ -5,6 +5,11 @@ from typing import TYPE_CHECKING, Any, cast
from crewai_core.printer import PRINTER
from crewai.events.event_listener import event_listener
from crewai.hooks.dispatch import (
HookAborted,
InterceptionPoint,
get_global_hook_list,
)
from crewai.hooks.types import (
AfterLLMCallHookCallable,
AfterLLMCallHookType,
@@ -150,8 +155,37 @@ class LLMCallHookContext:
event_listener.formatter.resume_live_updates()
_before_llm_call_hooks: list[BeforeLLMCallHookType | BeforeLLMCallHookCallable] = []
_after_llm_call_hooks: list[AfterLLMCallHookType | AfterLLMCallHookCallable] = []
# The legacy registries are aliased to the generic dispatcher's global hook
# lists for the model-call points, so legacy registrations and new-dialect
# ``@on(InterceptionPoint.PRE_MODEL_CALL)`` hooks share one ordered queue.
_before_llm_call_hooks: list[BeforeLLMCallHookType | BeforeLLMCallHookCallable] = (
get_global_hook_list(InterceptionPoint.PRE_MODEL_CALL)
)
_after_llm_call_hooks: list[AfterLLMCallHookType | AfterLLMCallHookCallable] = (
get_global_hook_list(InterceptionPoint.POST_MODEL_CALL)
)
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.
"""
if result is False:
raise HookAborted(reason="before_llm_call hook returned False")
return False
def after_llm_call_reducer(context: LLMCallHookContext, result: object) -> bool:
"""Legacy calling convention for ``post_model_call`` hooks.
A non-empty string return replaces the response on the context.
"""
if result is not None and isinstance(result, str):
context.response = result
return True
return False
def register_before_llm_call_hook(

83
lib/crewai/src/crewai/hooks/tool_hooks.py
View File

@@ -5,6 +5,12 @@ from typing import TYPE_CHECKING, Any
from crewai_core.printer import PRINTER
from crewai.events.event_listener import event_listener
from crewai.hooks.dispatch import (
HookAborted,
InterceptionPoint,
dispatch,
get_global_hook_list,
)
from crewai.hooks.types import (
AfterToolCallHookCallable,
AfterToolCallHookType,
@@ -121,8 +127,81 @@ class ToolCallHookContext:
event_listener.formatter.resume_live_updates()
_before_tool_call_hooks: list[BeforeToolCallHookType | BeforeToolCallHookCallable] = []
_after_tool_call_hooks: list[AfterToolCallHookType | AfterToolCallHookCallable] = []
# The legacy registries are aliased to the generic dispatcher's global hook
# lists for the tool-call points, so legacy registrations and new-dialect
# ``@on(InterceptionPoint.PRE_TOOL_CALL)`` hooks share one ordered queue.
_before_tool_call_hooks: list[BeforeToolCallHookType | BeforeToolCallHookCallable] = (
get_global_hook_list(InterceptionPoint.PRE_TOOL_CALL)
)
_after_tool_call_hooks: list[AfterToolCallHookType | AfterToolCallHookCallable] = (
get_global_hook_list(InterceptionPoint.POST_TOOL_CALL)
)
def before_tool_call_reducer(context: ToolCallHookContext, result: object) -> bool:
"""Legacy calling convention for ``pre_tool_call`` hooks.
A ``False`` return blocks the call (mapped to :class:`HookAborted`); tool
input is modified in place, so no payload replacement occurs here.
"""
if result is False:
raise HookAborted(reason="before_tool_call hook returned False")
return False
def after_tool_call_reducer(context: ToolCallHookContext, result: object) -> bool:
"""Legacy calling convention for ``post_tool_call`` hooks.
A non-None return replaces the tool result on the context.
"""
if isinstance(result, str):
context.tool_result = result
return True
return False
def _hook_verbose(context: ToolCallHookContext) -> bool:
"""Whether swallowed-hook-error warnings should be printed.
Mirrors the pre-dispatcher behavior where a failing tool hook surfaced a
warning when the executing agent was verbose.
"""
return bool(getattr(context.agent, "verbose", False))
def run_before_tool_call_hooks(context: ToolCallHookContext) -> bool:
"""Run all ``pre_tool_call`` hooks against a context.
Returns:
True if a hook blocked execution (returned False or raised
:class:`HookAborted`), False otherwise. Tool input mutations on the
context persist regardless.
"""
try:
dispatch(
InterceptionPoint.PRE_TOOL_CALL,
context,
reducer=before_tool_call_reducer,
verbose=_hook_verbose(context),
)
return False
except HookAborted:
return True
def run_after_tool_call_hooks(context: ToolCallHookContext) -> str | None:
"""Run all ``post_tool_call`` hooks against a context.
Returns:
The (possibly modified) tool result carried on the context.
"""
dispatch(
InterceptionPoint.POST_TOOL_CALL,
context,
reducer=after_tool_call_reducer,
verbose=_hook_verbose(context),
)
return context.tool_result
def register_before_tool_call_hook(

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

@@ -1007,15 +1007,14 @@ class BaseLLM(BaseModel, ABC):
from crewai_core.printer import PRINTER
from crewai.hooks.dispatch import HookAborted, InterceptionPoint, dispatch
from crewai.hooks.llm_hooks import (
LLMCallHookContext,
get_before_llm_call_hooks,
before_llm_call_reducer,
)
before_hooks = get_before_llm_call_hooks()
if not before_hooks:
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,
@@ -1024,24 +1023,19 @@ class BaseLLM(BaseModel, ABC):
task=None,
crew=None,
)
verbose = getattr(from_agent, "verbose", True) if from_agent else True
try:
for hook in before_hooks:
result = hook(hook_context)
if result is False:
if verbose:
PRINTER.print(
content="LLM call blocked by before_llm_call hook",
color="yellow",
)
return False
except Exception as e:
if verbose:
PRINTER.print(
content=f"Error in before_llm_call hook: {e}",
color="yellow",
)
dispatch(
InterceptionPoint.PRE_MODEL_CALL,
hook_context,
reducer=before_llm_call_reducer,
)
except HookAborted:
PRINTER.print(
content="LLM call blocked by before_llm_call hook",
color="yellow",
)
return False
return True
@@ -1074,17 +1068,14 @@ class BaseLLM(BaseModel, ABC):
if from_agent is not None or not isinstance(response, str):
return response
from crewai_core.printer import PRINTER
from crewai.hooks.dispatch import InterceptionPoint, dispatch
from crewai.hooks.llm_hooks import (
LLMCallHookContext,
get_after_llm_call_hooks,
after_llm_call_reducer,
)
after_hooks = get_after_llm_call_hooks()
if not after_hooks:
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,
@@ -1094,20 +1085,11 @@ class BaseLLM(BaseModel, ABC):
crew=None,
response=response,
)
verbose = getattr(from_agent, "verbose", True) if from_agent else True
modified_response = response
try:
for hook in after_hooks:
result = hook(hook_context)
if result is not None and isinstance(result, str):
modified_response = result
hook_context.response = modified_response
except Exception as e:
if verbose:
PRINTER.print(
content=f"Error in after_llm_call hook: {e}",
color="yellow",
)
dispatch(
InterceptionPoint.POST_MODEL_CALL,
hook_context,
reducer=after_llm_call_reducer,
)
return modified_response
return hook_context.response if hook_context.response is not None else response

29
lib/crewai/src/crewai/project/crew_base.py
View File

@@ -452,7 +452,15 @@ def _register_crew_hooks(instance: CrewInstance, cls: type) -> None:
)
}
if not hook_methods:
# Methods decorated with @on(InterceptionPoint.X) carry ``_interception_point``
# instead of the legacy markers above.
on_methods = {
name: method
for name, method in cls.__dict__.items()
if hasattr(method, "_interception_point")
}
if not hook_methods and not on_methods:
return
from crewai.hooks import (
@@ -588,6 +596,25 @@ def _register_crew_hooks(instance: CrewInstance, cls: type) -> None:
("after_tool_call", after_tool_hook)
)
if on_methods:
from crewai.hooks.dispatch import (
_wrap_with_filters,
register as register_interception_hook,
)
for on_method in on_methods.values():
point = on_method._interception_point
bound_hook = on_method.__get__(instance, cls)
tools_filter = getattr(on_method, "_filter_tools", None)
agents_filter = getattr(on_method, "_filter_agents", None)
hook = (
_wrap_with_filters(bound_hook, agents_filter, tools_filter)
if (tools_filter or agents_filter)
else bound_hook
)
register_interception_hook(point, hook)
instance._registered_hook_functions.append((point.value, hook))
instance._hooks_being_registered = False

72
lib/crewai/src/crewai/task.py
View File

@@ -662,6 +662,21 @@ class Task(BaseModel):
crewai_event_bus.emit(
self, TaskStartedEvent(context=context, task=self)
)
from crewai.hooks.contexts import StepContext
from crewai.hooks.dispatch import InterceptionPoint, dispatch
pre_step_ctx = StepContext(
kind="task",
step_name=self.name or self.description,
agent=agent,
agent_role=getattr(agent, "role", None),
task=self,
payload=context,
)
dispatch(InterceptionPoint.PRE_STEP, pre_step_ctx)
context = pre_step_ctx.payload
result = await agent.aexecute_task(
task=self,
context=context,
@@ -718,6 +733,18 @@ class Task(BaseModel):
guardrail=self._guardrail,
)
post_step_ctx = StepContext(
kind="task",
step_name=self.name or self.description,
agent=agent,
agent_role=getattr(agent, "role", None),
task=self,
output=task_output,
payload=task_output,
)
dispatch(InterceptionPoint.POST_STEP, post_step_ctx)
task_output = cast(TaskOutput, post_step_ctx.payload)
self.output = task_output
self.end_time = datetime.datetime.now()
@@ -739,10 +766,12 @@ class Task(BaseModel):
if self.output_file:
content = (
json_output
if json_output
task_output.json_dict
if task_output.json_dict
else (
pydantic_output.model_dump_json() if pydantic_output else result
task_output.pydantic.model_dump_json()
if task_output.pydantic
else task_output.raw
)
)
self._save_file(content)
@@ -787,6 +816,21 @@ class Task(BaseModel):
crewai_event_bus.emit(
self, TaskStartedEvent(context=context, task=self)
)
from crewai.hooks.contexts import StepContext
from crewai.hooks.dispatch import InterceptionPoint, dispatch
pre_step_ctx = StepContext(
kind="task",
step_name=self.name or self.description,
agent=agent,
agent_role=getattr(agent, "role", None),
task=self,
payload=context,
)
dispatch(InterceptionPoint.PRE_STEP, pre_step_ctx)
context = pre_step_ctx.payload
result = agent.execute_task(
task=self,
context=context,
@@ -843,6 +887,18 @@ class Task(BaseModel):
guardrail=self._guardrail,
)
post_step_ctx = StepContext(
kind="task",
step_name=self.name or self.description,
agent=agent,
agent_role=getattr(agent, "role", None),
task=self,
output=task_output,
payload=task_output,
)
dispatch(InterceptionPoint.POST_STEP, post_step_ctx)
task_output = cast(TaskOutput, post_step_ctx.payload)
self.output = task_output
self.end_time = datetime.datetime.now()
@@ -864,10 +920,12 @@ class Task(BaseModel):
if self.output_file:
content = (
json_output
if json_output
task_output.json_dict
if task_output.json_dict
else (
pydantic_output.model_dump_json() if pydantic_output else result
task_output.pydantic.model_dump_json()
if task_output.pydantic
else task_output.raw
)
)
self._save_file(content)
@@ -1316,7 +1374,6 @@ 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,
@@ -1426,7 +1483,6 @@ 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,

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

@@ -1150,6 +1150,8 @@ def load_agent_from_repository(from_repository: str) -> dict[str, Any]:
agent = response.json()
for key, value in agent.items():
if value is None:
continue
if key == "tools":
attributes[key] = []
for tool in value:
@@ -1453,8 +1455,8 @@ def execute_single_native_tool_call(
)
from crewai.hooks.tool_hooks import (
ToolCallHookContext,
get_after_tool_call_hooks,
get_before_tool_call_hooks,
run_after_tool_call_hooks,
run_before_tool_call_hooks,
)
info = extract_tool_call_info(tool_call)
@@ -1517,7 +1519,6 @@ def execute_single_native_tool_call(
track_delegation_if_needed(func_name, args_dict, task)
hook_blocked = False
before_hook_context = ToolCallHookContext(
tool_name=func_name,
tool_input=args_dict,
@@ -1526,13 +1527,7 @@ def execute_single_native_tool_call(
task=task,
crew=crew,
)
try:
for hook in get_before_tool_call_hooks():
if hook(before_hook_context) is False:
hook_blocked = True
break
except Exception: # noqa: S110
pass
hook_blocked = run_before_tool_call_hooks(before_hook_context)
error_event_emitted = False
if hook_blocked:
@@ -1587,14 +1582,9 @@ def execute_single_native_tool_call(
tool_result=result,
raw_tool_result=raw_tool_result,
)
try:
for after_hook in get_after_tool_call_hooks():
hook_result = after_hook(after_hook_context)
if hook_result is not None:
result = hook_result
after_hook_context.tool_result = result
except Exception: # noqa: S110
pass
modified_result = run_after_tool_call_hooks(after_hook_context)
if modified_result is not None:
result = modified_result
if not error_event_emitted:
crewai_event_bus.emit(
@@ -1690,28 +1680,42 @@ def _setup_before_llm_call_hooks(
Returns:
True if LLM execution should proceed, False if blocked by a hook.
"""
if executor_context and executor_context.before_llm_call_hooks:
from crewai.hooks.llm_hooks import LLMCallHookContext
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
# Executor snapshot first, then execution-scoped hooks — the same
# ordering dispatch() applies to global vs scoped hooks.
hooks: list[Any] = [
*executor_context.before_llm_call_hooks,
*get_scoped_hooks(InterceptionPoint.PRE_MODEL_CALL),
]
if not hooks:
return True
original_messages = executor_context.messages
hook_context = LLMCallHookContext(executor_context)
try:
for hook in executor_context.before_llm_call_hooks:
result = hook(hook_context)
if result is False:
if verbose:
printer.print(
content="LLM call blocked by before_llm_call hook",
color="yellow",
)
return False
except Exception as e:
run_hooks(
InterceptionPoint.PRE_MODEL_CALL,
hook_context,
hooks,
reducer=before_llm_call_reducer,
verbose=verbose,
)
except HookAborted:
if verbose:
printer.print(
content=f"Error in before_llm_call hook: {e}",
content="LLM call blocked by before_llm_call hook",
color="yellow",
)
return False
if not isinstance(executor_context.messages, list):
if verbose:
@@ -1748,8 +1752,24 @@ def _setup_after_llm_call_hooks(
Returns:
The potentially modified response (string or Pydantic model).
"""
if executor_context and executor_context.after_llm_call_hooks:
from crewai.hooks.llm_hooks import LLMCallHookContext
if executor_context:
from crewai.hooks.dispatch import InterceptionPoint, get_scoped_hooks, run_hooks
from crewai.hooks.llm_hooks import LLMCallHookContext, after_llm_call_reducer
# Don't stringify structured tool-call payloads: the executor would
# treat the result as a final answer and skip tool execution (#6529).
# Hooks still run on the follow-up textual response.
if not isinstance(answer, (str, BaseModel)):
return answer
# Executor snapshot first, then execution-scoped hooks — the same
# ordering dispatch() applies to global vs scoped hooks.
hooks: list[Any] = [
*executor_context.after_llm_call_hooks,
*get_scoped_hooks(InterceptionPoint.POST_MODEL_CALL),
]
if not hooks:
return answer
original_messages = executor_context.messages
@@ -1762,18 +1782,15 @@ def _setup_after_llm_call_hooks(
hook_response = str(answer)
hook_context = LLMCallHookContext(executor_context, response=hook_response)
try:
for hook in executor_context.after_llm_call_hooks:
modified_response = hook(hook_context)
if modified_response is not None and isinstance(modified_response, str):
hook_response = modified_response
except Exception as e:
if verbose:
printer.print(
content=f"Error in after_llm_call hook: {e}",
color="yellow",
)
run_hooks(
InterceptionPoint.POST_MODEL_CALL,
hook_context,
hooks,
reducer=after_llm_call_reducer,
verbose=verbose,
)
if hook_context.response is not None:
hook_response = hook_context.response
if not isinstance(executor_context.messages, list):
if verbose:

107
lib/crewai/src/crewai/utilities/tool_utils.py
View File

@@ -6,15 +6,14 @@ from crewai.agents.parser import AgentAction
from crewai.agents.tools_handler import ToolsHandler
from crewai.hooks.tool_hooks import (
ToolCallHookContext,
get_after_tool_call_hooks,
get_before_tool_call_hooks,
run_after_tool_call_hooks,
run_before_tool_call_hooks,
)
from crewai.security.fingerprint import Fingerprint
from crewai.tools.structured_tool import CrewStructuredTool
from crewai.tools.tool_types import ToolResult
from crewai.tools.tool_usage import ToolUsage, ToolUsageError
from crewai.utilities.i18n import I18N_DEFAULT
from crewai.utilities.logger import Logger
from crewai.utilities.string_utils import sanitize_tool_name
@@ -57,11 +56,10 @@ async def aexecute_tool_and_check_finality(
fingerprint_context: Optional context for fingerprinting.
crew: Optional crew instance for hook context.
Returns:
Returns:
ToolResult containing the execution result and whether it should be
treated as a final answer.
"""
logger = Logger(verbose=crew.verbose if crew else False)
tool_name_to_tool_map = {sanitize_tool_name(tool.name): tool for tool in tools}
if agent_key and agent_role and agent:
@@ -102,18 +100,27 @@ async def aexecute_tool_and_check_finality(
crew=crew,
)
before_hooks = get_before_tool_call_hooks()
try:
for hook in before_hooks:
result = hook(hook_context)
if result is False:
blocked_message = (
f"Tool execution blocked by hook. "
f"Tool: {tool_calling.tool_name}"
)
return ToolResult(blocked_message, False)
except Exception as e:
logger.log("error", f"Error in before_tool_call hook: {e}")
if run_before_tool_call_hooks(hook_context):
blocked_message = (
f"Tool execution blocked by hook. Tool: {tool_calling.tool_name}"
)
# Run POST_TOOL_CALL even on a blocked call so monitoring hooks
# still fire, matching the native tool-call paths.
blocked_hook_context = ToolCallHookContext(
tool_name=sanitized_tool_name,
tool_input=tool_input,
tool=tool,
agent=agent,
task=task,
crew=crew,
tool_result=blocked_message,
raw_tool_result=blocked_message,
)
modified_result = run_after_tool_call_hooks(blocked_hook_context)
return ToolResult(
modified_result if modified_result is not None else blocked_message,
False,
)
tool_result = await tool_usage.ause(tool_calling, agent_action.text)
raw_tool_result = tool_usage.get_last_raw_result(tool_result)
@@ -129,18 +136,12 @@ async def aexecute_tool_and_check_finality(
raw_tool_result=raw_tool_result,
)
after_hooks = get_after_tool_call_hooks()
modified_result: str = tool_result
try:
for after_hook in after_hooks:
hook_result = after_hook(after_hook_context)
if hook_result is not None:
modified_result = hook_result
after_hook_context.tool_result = modified_result
except Exception as e:
logger.log("error", f"Error in after_tool_call hook: {e}")
modified_result = run_after_tool_call_hooks(after_hook_context)
return ToolResult(modified_result, tool.result_as_answer)
return ToolResult(
modified_result if modified_result is not None else tool_result,
tool.result_as_answer,
)
tool_result = I18N_DEFAULT.errors("wrong_tool_name").format(
tool=sanitized_tool_name,
@@ -181,7 +182,6 @@ def execute_tool_and_check_finality(
Returns:
ToolResult containing the execution result and whether it should be treated as a final answer
"""
logger = Logger(verbose=crew.verbose if crew else False)
tool_name_to_tool_map = {sanitize_tool_name(tool.name): tool for tool in tools}
if agent_key and agent_role and agent:
@@ -222,18 +222,27 @@ def execute_tool_and_check_finality(
crew=crew,
)
before_hooks = get_before_tool_call_hooks()
try:
for hook in before_hooks:
result = hook(hook_context)
if result is False:
blocked_message = (
f"Tool execution blocked by hook. "
f"Tool: {tool_calling.tool_name}"
)
return ToolResult(blocked_message, False)
except Exception as e:
logger.log("error", f"Error in before_tool_call hook: {e}")
if run_before_tool_call_hooks(hook_context):
blocked_message = (
f"Tool execution blocked by hook. Tool: {tool_calling.tool_name}"
)
# Run POST_TOOL_CALL even on a blocked call so monitoring hooks
# still fire, matching the native tool-call paths.
blocked_hook_context = ToolCallHookContext(
tool_name=sanitized_tool_name,
tool_input=tool_input,
tool=tool,
agent=agent,
task=task,
crew=crew,
tool_result=blocked_message,
raw_tool_result=blocked_message,
)
modified_result = run_after_tool_call_hooks(blocked_hook_context)
return ToolResult(
modified_result if modified_result is not None else blocked_message,
False,
)
tool_result = tool_usage.use(tool_calling, agent_action.text)
raw_tool_result = tool_usage.get_last_raw_result(tool_result)
@@ -249,18 +258,12 @@ def execute_tool_and_check_finality(
raw_tool_result=raw_tool_result,
)
after_hooks = get_after_tool_call_hooks()
modified_result: str = tool_result
try:
for after_hook in after_hooks:
hook_result = after_hook(after_hook_context)
if hook_result is not None:
modified_result = hook_result
after_hook_context.tool_result = modified_result
except Exception as e:
logger.log("error", f"Error in after_tool_call hook: {e}")
modified_result = run_after_tool_call_hooks(after_hook_context)
return ToolResult(modified_result, tool.result_as_answer)
return ToolResult(
modified_result if modified_result is not None else tool_result,
tool.result_as_answer,
)
tool_result = I18N_DEFAULT.errors("wrong_tool_name").format(
tool=sanitized_tool_name,

19
lib/crewai/tests/agents/test_agent.py
View File

@@ -2335,6 +2335,25 @@ def test_agent_from_repository_override_attributes(mock_get_agent, mock_get_auth
assert isinstance(agent.tools[0], SerperDevTool)
@patch("crewai.plus_api.PlusAPI.get_agent")
def test_agent_from_repository_ignores_null_attributes(
mock_get_agent, mock_get_auth_token
):
mock_get_response = MagicMock()
mock_get_response.status_code = 200
mock_get_response.json.return_value = {
"role": "test role",
"goal": "test goal",
"backstory": "test backstory",
"reasoning": None,
}
mock_get_agent.return_value = mock_get_response
agent = Agent(from_repository="test_agent")
assert agent.reasoning is False
@patch("crewai.plus_api.PlusAPI.get_agent")
def test_agent_from_repository_ignores_empty_skills(
mock_get_agent, mock_get_auth_token

92
lib/crewai/tests/cli/test_plus_api.py
View File

@@ -1,7 +1,10 @@
import os
import unittest
from types import SimpleNamespace
from unittest.mock import ANY, MagicMock, patch
import pytest
from crewai.plus_api import PlusAPI
@@ -393,6 +396,45 @@ class TestPlusAPI(unittest.TestCase):
"https://custom-url-from-env.com",
)
@patch.dict(os.environ, {"CREWAI_PLUS_URL": "https://url-from-env.com"})
def test_explicit_target_takes_precedence(self):
api = PlusAPI(
"test_key",
base_url="https://explicit-url.com",
organization_id="explicit-org-id",
)
self.assertEqual(api.base_url, "https://explicit-url.com")
self.assertEqual(
api.headers["X-Crewai-Organization-Id"], "explicit-org-id"
)
@patch("crewai_core.plus_api.Settings")
def test_explicit_base_url_uses_organization_from_settings(
self, mock_settings_class
):
mock_settings_class.return_value.org_uuid = "settings-org-id"
api = PlusAPI("test_key", base_url="https://explicit-url.com")
self.assertEqual(api.base_url, "https://explicit-url.com")
self.assertEqual(api.headers["X-Crewai-Organization-Id"], "settings-org-id")
@patch("crewai_core.plus_api.Settings")
@patch.dict(os.environ, {"CREWAI_PLUS_URL": ""})
def test_explicit_organization_uses_base_url_from_settings(
self, mock_settings_class
):
mock_settings_class.return_value.enterprise_base_url = (
"https://url-from-settings.com"
)
api = PlusAPI("test_key", organization_id="explicit-org-id")
self.assertEqual(api.base_url, "https://url-from-settings.com")
self.assertEqual(
api.headers["X-Crewai-Organization-Id"], "explicit-org-id"
)
@patch("crewai_core.plus_api.PlusAPI._make_request")
def test_get_agent(mock_make_request):
@@ -430,3 +472,53 @@ def test_get_agent_with_org_uuid(mock_settings_class, mock_make_request):
assert "X-Crewai-Organization-Id" in api.headers
assert api.headers["X-Crewai-Organization-Id"] == org_uuid
assert response == mock_response
@pytest.mark.parametrize(
("target", "other_target"),
(
(
("https://first.example.com", "first-org"),
("https://second.example.com", "second-org"),
),
(
("https://second.example.com", "second-org"),
("https://first.example.com", "first-org"),
),
),
)
def test_clients_keep_targets_independent(target, other_target):
base_url, organization_id = target
api = PlusAPI(
"test_key",
base_url=base_url,
organization_id=organization_id,
)
other_base_url, other_organization_id = other_target
PlusAPI(
"test_key",
base_url=other_base_url,
organization_id=other_organization_id,
)
assert (api.base_url, api.headers["X-Crewai-Organization-Id"]) == target
@patch("crewai_core.plus_api.Settings")
@patch.dict(os.environ, {"CREWAI_PLUS_URL": ""})
def test_default_target_uses_single_settings_snapshot(mock_settings_class):
mock_settings_class.side_effect = (
SimpleNamespace(
org_uuid="first-org",
enterprise_base_url="https://first.example.com",
),
SimpleNamespace(
org_uuid="second-org",
enterprise_base_url="https://second.example.com",
),
)
api = PlusAPI("test_key")
assert api.base_url == "https://first.example.com"
assert api.headers["X-Crewai-Organization-Id"] == "first-org"

56
lib/crewai/tests/hooks/test_crew_scoped_hooks.py
View File

@@ -306,6 +306,62 @@ class TestCrewScopedHooks:
assert len(execution_log) == 1
class TestCrewOnDecoratedMethods:
"""@on(InterceptionPoint.X) methods inside @CrewBase must register.
Regression: CrewBase only scanned the legacy ``is_*_hook`` markers, so
methods decorated with the generic ``@on`` decorator (which sets
``_interception_point``) were silently dropped and never ran.
"""
def test_on_decorated_method_registers_and_binds_self(self):
from crewai.hooks import InterceptionPoint, on
from crewai.hooks.dispatch import _resolve_hooks
execution_log = []
@CrewBase
class TestCrew:
def __init__(self):
self.name = "on-crew"
@on(InterceptionPoint.PRE_MODEL_CALL)
def on_pre_model(self, context):
execution_log.append(self.name)
@agent
def researcher(self):
return Agent(role="Researcher", goal="Research", backstory="Expert")
@crew
def crew(self):
return Crew(agents=self.agents, tasks=[], verbose=False)
before = len(_resolve_hooks(InterceptionPoint.PRE_MODEL_CALL))
instance = TestCrew()
hooks = _resolve_hooks(InterceptionPoint.PRE_MODEL_CALL)
assert len(hooks) == before + 1
assert (
InterceptionPoint.PRE_MODEL_CALL.value,
hooks[-1],
) in instance._registered_hook_functions
mock_executor = Mock()
mock_executor.messages = []
mock_executor.agent = Mock(role="Test")
mock_executor.task = Mock()
mock_executor.crew = Mock()
mock_executor.llm = Mock()
mock_executor.iterations = 0
hooks[-1](LLMCallHookContext(executor=mock_executor))
assert execution_log == ["on-crew"]
class TestCrewScopedHookAttributes:
"""Test that crew-scoped hooks have correct attributes set."""

296
lib/crewai/tests/hooks/test_dispatch.py Normal file
View File

@@ -0,0 +1,296 @@
"""Unit tests for the generic interception-hook dispatcher.
These cover the new contract (payload-in/payload-out + HookAborted), the shared
ordered queue between the legacy and new dialects on the four model/tool points,
execution-scoped hooks, fail-open exception handling, telemetry, and the no-op
fast-path overhead budget.
"""
from __future__ import annotations
from dataclasses import dataclass
import time
from crewai.events.event_bus import crewai_event_bus
from crewai.events.types.hook_events import HookDispatchedEvent
from crewai.hooks.dispatch import (
HookAborted,
InterceptionPoint,
clear_all,
dispatch,
get_hooks,
on,
register,
register_scoped,
scoped_hooks,
unregister as unregister_hook,
)
from crewai.hooks.llm_hooks import (
get_before_llm_call_hooks,
register_before_llm_call_hook,
)
import pytest
@dataclass
class _Ctx:
payload: object = None
tool_name: str | None = None
agent: object = None
agent_role: str | None = None
@pytest.fixture(autouse=True)
def clear_dispatch_registry():
"""Ensure every test starts and ends with an empty global registry."""
clear_all()
yield
clear_all()
class TestDispatchContract:
"""The core payload-in/payload-out + HookAborted contract."""
def test_noop_fast_path_returns_context_unchanged(self):
ctx = _Ctx(payload="original")
result = dispatch(InterceptionPoint.PRE_MODEL_CALL, ctx)
assert result is ctx
assert ctx.payload == "original"
def test_return_value_replaces_payload(self):
def double(ctx):
return ctx.payload * 2
register(InterceptionPoint.PRE_MODEL_CALL, double)
ctx = _Ctx(payload="ab")
dispatch(InterceptionPoint.PRE_MODEL_CALL, ctx)
assert ctx.payload == "abab"
def test_in_place_mutation_is_honored(self):
def mutate(ctx):
ctx.payload.append(1)
return None
register(InterceptionPoint.PRE_MODEL_CALL, mutate)
ctx = _Ctx(payload=[])
dispatch(InterceptionPoint.PRE_MODEL_CALL, ctx)
assert ctx.payload == [1]
def test_hooks_run_in_registration_order(self):
order: list[int] = []
register(InterceptionPoint.PRE_MODEL_CALL, lambda ctx: order.append(1))
register(InterceptionPoint.PRE_MODEL_CALL, lambda ctx: order.append(2))
dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx())
assert order == [1, 2]
def test_hook_aborted_propagates_with_reason_and_source(self):
def blocker(ctx):
raise HookAborted(reason="nope", source="policy")
register(InterceptionPoint.PRE_MODEL_CALL, blocker)
with pytest.raises(HookAborted) as exc:
dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx())
assert exc.value.reason == "nope"
assert exc.value.source == "policy"
def test_ordinary_exception_is_swallowed_and_later_hooks_run(self):
ran: list[str] = []
def boom(ctx):
ran.append("boom")
raise ValueError("bug in user hook")
def after(ctx):
ran.append("after")
register(InterceptionPoint.PRE_MODEL_CALL, boom)
register(InterceptionPoint.PRE_MODEL_CALL, after)
dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx(), verbose=False)
assert ran == ["boom", "after"]
class TestOnDecorator:
"""The @on decorator registers and filters like the legacy decorators."""
def test_on_registers_global_hook(self):
@on(InterceptionPoint.POST_TOOL_CALL)
def hook(ctx):
return None
assert hook in get_hooks(InterceptionPoint.POST_TOOL_CALL)
def test_tool_filter_skips_non_matching_tools(self):
seen: list[str] = []
@on(InterceptionPoint.PRE_TOOL_CALL, tools=["allowed_tool"])
def hook(ctx):
seen.append(ctx.tool_name)
dispatch(InterceptionPoint.PRE_TOOL_CALL, _Ctx(tool_name="other_tool"))
dispatch(InterceptionPoint.PRE_TOOL_CALL, _Ctx(tool_name="allowed_tool"))
assert seen == ["allowed_tool"]
def test_agent_filter_skips_non_matching_agents(self):
seen: list[str] = []
class _Agent:
def __init__(self, role):
self.role = role
@on(InterceptionPoint.PRE_MODEL_CALL, agents=["Researcher"])
def hook(ctx):
seen.append(ctx.agent.role)
dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx(agent=_Agent("Writer")))
dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx(agent=_Agent("Researcher")))
assert seen == ["Researcher"]
def test_agent_filter_falls_back_to_agent_role(self):
seen: list[str] = []
@on(InterceptionPoint.PRE_TOOL_CALL, agents=["Researcher"])
def hook(ctx):
seen.append(ctx.agent_role)
# No agent object, only the agent_role string (e.g. flow seams).
dispatch(InterceptionPoint.PRE_TOOL_CALL, _Ctx(agent_role="Writer"))
dispatch(InterceptionPoint.PRE_TOOL_CALL, _Ctx(agent_role="Researcher"))
assert seen == ["Researcher"]
def test_unregister_resolves_filtered_wrapper(self):
@on(InterceptionPoint.PRE_TOOL_CALL, tools=["allowed_tool"])
def hook(ctx):
return None
assert len(get_hooks(InterceptionPoint.PRE_TOOL_CALL)) == 1
assert unregister_hook(InterceptionPoint.PRE_TOOL_CALL, hook) is True
assert get_hooks(InterceptionPoint.PRE_TOOL_CALL) == []
class TestSharedQueueWithLegacyDialect:
"""Legacy registrations and @on hooks compose in one ordered queue."""
def test_on_and_legacy_share_pre_model_call_queue(self):
def legacy(ctx):
return None
@on(InterceptionPoint.PRE_MODEL_CALL)
def modern(ctx):
return None
register_before_llm_call_hook(legacy)
queue = get_before_llm_call_hooks()
assert modern in queue
assert legacy in queue
# registration order preserved: modern registered before legacy
assert queue.index(modern) < queue.index(legacy)
class TestScopedHooks:
"""Execution-scoped hooks run after globals and are discarded on exit."""
def test_scoped_runs_after_global_then_cleared(self):
order: list[str] = []
register(InterceptionPoint.POST_MODEL_CALL, lambda ctx: order.append("global"))
with scoped_hooks():
register_scoped(InterceptionPoint.POST_MODEL_CALL, lambda ctx: order.append("scoped"))
dispatch(InterceptionPoint.POST_MODEL_CALL, _Ctx())
# outside the scope the scoped hook is gone
dispatch(InterceptionPoint.POST_MODEL_CALL, _Ctx())
assert order == ["global", "scoped", "global"]
class TestTelemetry:
"""dispatch emits a HookDispatchedEvent only when hooks ran."""
def test_no_event_on_empty_fast_path(self):
events: list[HookDispatchedEvent] = []
with crewai_event_bus.scoped_handlers():
@crewai_event_bus.on(HookDispatchedEvent)
def _capture(_source, event):
events.append(event)
dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx())
assert events == []
def test_event_reports_outcome(self):
events: list[HookDispatchedEvent] = []
register(InterceptionPoint.PRE_MODEL_CALL, lambda ctx: "changed")
with crewai_event_bus.scoped_handlers():
@crewai_event_bus.on(HookDispatchedEvent)
def _capture(_source, event):
events.append(event)
dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx())
# Telemetry handlers run on the bus's thread pool; flush so the
# assertion doesn't race the emit.
crewai_event_bus.flush()
assert len(events) == 1
assert events[0].interception_point == "pre_model_call"
assert events[0].outcome == "modified"
assert events[0].hook_count == 1
def test_event_reports_abort_outcome(self):
events: list[HookDispatchedEvent] = []
def blocker(ctx):
raise HookAborted(reason="blocked", source="policy")
register(InterceptionPoint.PRE_MODEL_CALL, blocker)
with crewai_event_bus.scoped_handlers():
@crewai_event_bus.on(HookDispatchedEvent)
def _capture(_source, event):
events.append(event)
with pytest.raises(HookAborted):
dispatch(InterceptionPoint.PRE_MODEL_CALL, _Ctx())
crewai_event_bus.flush()
assert len(events) == 1
assert events[0].interception_point == "pre_model_call"
assert events[0].outcome == "aborted"
assert events[0].abort_reason == "blocked"
assert events[0].abort_source == "policy"
class TestNoOpOverhead:
"""The no-op fast path must stay cheap (a single dict lookup)."""
def test_noop_dispatch_overhead_is_bounded(self):
# Relative (not absolute) budget: the no-op fast path is a dict lookup
# plus a guard, so it should stay within a wide multiple of a bare
# function call. This catches accidental O(n) regressions without
# depending on absolute timing on shared CI runners.
ctx = _Ctx()
iterations = 100_000
def _baseline(_c):
return _c
for _ in range(1000): # warm up both paths
dispatch(InterceptionPoint.PRE_MODEL_CALL, ctx)
_baseline(ctx)
start = time.perf_counter()
for _ in range(iterations):
_baseline(ctx)
baseline = time.perf_counter() - start
start = time.perf_counter()
for _ in range(iterations):
dispatch(InterceptionPoint.PRE_MODEL_CALL, ctx)
noop = time.perf_counter() - start
assert noop < baseline * 50 + 5e-3

183
lib/crewai/tests/hooks/test_interception_conformance.py Normal file
View File

@@ -0,0 +1,183 @@
"""Conformance suite for the framework-native interception points.
For each wired point this suite asserts the shared contract: the probe hook
sees a well-shaped payload, an in-place/returned modification is honored, and a
:class:`HookAborted` interrupts the step.
"""
from __future__ import annotations
from unittest.mock import patch
from crewai.agent import Agent
from crewai.crew import Crew
from crewai.events.event_bus import crewai_event_bus
from crewai.events.types.crew_events import CrewKickoffCompletedEvent
from crewai.flow.flow import Flow, listen, start
from crewai.hooks.dispatch import (
HookAborted,
InterceptionPoint,
clear_all,
on,
)
from crewai.task import Task
import pytest
@pytest.fixture(autouse=True)
def clear_dispatch_registry():
clear_all()
yield
clear_all()
class _SimpleFlow(Flow):
@start()
def begin(self):
return "begin"
@listen(begin)
def finish(self, _):
return "flow-result"
class TestFlowExecutionBoundaries:
"""execution_start / input / output / execution_end on a flow."""
def test_all_boundary_points_fire_once(self):
fired: list[str] = []
for point in (
InterceptionPoint.EXECUTION_START,
InterceptionPoint.INPUT,
InterceptionPoint.OUTPUT,
InterceptionPoint.EXECUTION_END,
):
@on(point)
def _probe(ctx, _point=point):
fired.append(_point.value)
_SimpleFlow().kickoff(inputs={"seed": 1})
assert fired == [
"execution_start",
"input",
"output",
"execution_end",
]
def test_output_modification_is_honored(self):
@on(InterceptionPoint.OUTPUT)
def rewrite(ctx):
return "intercepted"
result = _SimpleFlow().kickoff()
assert result == "intercepted"
def test_input_payload_carries_inputs(self):
seen: dict = {}
@on(InterceptionPoint.INPUT)
def capture(ctx):
seen.update(ctx.payload or {})
_SimpleFlow().kickoff(inputs={"seed": 42})
assert seen == {"seed": 42}
def test_abort_at_execution_start_interrupts(self):
@on(InterceptionPoint.EXECUTION_START)
def block(ctx):
raise HookAborted(reason="not allowed", source="policy")
with pytest.raises(HookAborted) as exc:
_SimpleFlow().kickoff()
assert exc.value.reason == "not allowed"
class TestFlowStepPoints:
"""pre_step / post_step for flow methods (kind=flow_method)."""
def test_pre_and_post_step_fire_per_method(self):
kinds: list[tuple[str, str | None]] = []
@on(InterceptionPoint.PRE_STEP)
def pre(ctx):
kinds.append(("pre", ctx.step_name))
@on(InterceptionPoint.POST_STEP)
def post(ctx):
kinds.append(("post", ctx.step_name))
_SimpleFlow().kickoff()
assert ("pre", "begin") in kinds
assert ("post", "begin") in kinds
assert ("pre", "finish") in kinds
assert ("post", "finish") in kinds
def test_post_step_can_rewrite_method_output(self):
@on(InterceptionPoint.POST_STEP)
def rewrite(ctx):
if ctx.step_name == "finish":
return "rewritten"
return None
assert _SimpleFlow().kickoff() == "rewritten"
class TestTaskStepPoints:
"""pre_step / post_step for task execution (kind=task)."""
def test_post_step_rewrite_is_persisted_to_output_file(
self, tmp_path, monkeypatch
):
@on(InterceptionPoint.POST_STEP)
def sanitize(ctx):
return ctx.payload.model_copy(update={"raw": "sanitized output"})
monkeypatch.chdir(tmp_path)
agent = Agent(role="Writer", goal="Write", backstory="Writes things.")
task = Task(
description="Write something",
expected_output="Some text",
output_file="output.txt",
agent=agent,
)
with patch.object(Agent, "execute_task", return_value="original output"):
result = task.execute_sync(agent=agent)
assert result.raw == "sanitized output"
assert (tmp_path / "output.txt").read_text() == "sanitized output"
class TestCrewOutput:
def test_output_modification_reaches_kickoff_completed_event(self):
@on(InterceptionPoint.OUTPUT)
def append_notice(ctx):
if hasattr(ctx.payload, "raw") and isinstance(ctx.payload.raw, str):
ctx.payload.raw += "\nchanged by hook"
return None
completed_raw: list[str] = []
@crewai_event_bus.on(CrewKickoffCompletedEvent)
def capture_completed(_source, event: CrewKickoffCompletedEvent):
completed_raw.append(event.output.raw)
agent = Agent(role="Writer", goal="Write", backstory="Writes things.")
task = Task(
description="Write something",
expected_output="Some text",
agent=agent,
)
crew = Crew(agents=[agent], tasks=[task], verbose=False)
with patch.object(Agent, "execute_task", return_value="original output"):
result = crew.kickoff()
crewai_event_bus.flush()
assert result.raw.endswith("changed by hook")
assert completed_raw
assert completed_raw[-1].endswith("changed by hook")

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

@@ -272,6 +272,40 @@ class TestLLMHooksIntegration:
assert result == "Original [hook1] [hook2]"
def test_after_hooks_do_not_clobber_native_tool_call_responses(
self, mock_executor
):
"""A registered after hook must not break native tool execution.
Regression for crewAIInc/crewAI#6529: `_setup_after_llm_call_hooks`
stringified structured tool-call payloads, so the executor treated the
raw tool call as the final answer and never executed the tool. Non-str,
non-BaseModel responses now pass through untouched; hooks still fire on
textual responses.
"""
from crewai.utilities.agent_utils import _setup_after_llm_call_hooks
observed = []
def observer(context):
observed.append(context.response)
return None
register_after_llm_call_hook(observer)
mock_executor.after_llm_call_hooks = get_after_llm_call_hooks()
tool_calls = [Mock()] # structured native tool-call payload
result = _setup_after_llm_call_hooks(
mock_executor, tool_calls, printer=Mock(), verbose=False
)
assert result is tool_calls
text = _setup_after_llm_call_hooks(
mock_executor, "final answer", printer=Mock(), verbose=False
)
assert text == "final answer"
assert observed == ["final answer"]
def test_unregister_before_hook(self):
"""Test that before hooks can be unregistered."""
def test_hook(context):
@@ -303,6 +337,105 @@ class TestLLMHooksIntegration:
hooks = get_before_llm_call_hooks()
assert len(hooks) == 0
def test_raising_before_hook_does_not_skip_later_hooks(self, mock_executor):
"""Fail-open is per-hook: a crashing hook must not disable its neighbors.
Regression guard for the dispatcher migration: previously the
``except Exception`` wrapped the whole hook loop, so a raising hook
silently skipped every hook registered after it. Now swallowing is
per-hook — later hooks still run and the LLM call still proceeds.
"""
from crewai.utilities.agent_utils import _setup_before_llm_call_hooks
ran: list[str] = []
def crashing_hook(context):
ran.append("crashing")
raise ValueError("bug in user hook")
def later_hook(context):
ran.append("later")
register_before_llm_call_hook(crashing_hook)
register_before_llm_call_hook(later_hook)
mock_executor.before_llm_call_hooks = get_before_llm_call_hooks()
proceed = _setup_before_llm_call_hooks(
mock_executor, printer=Mock(), verbose=False
)
assert ran == ["crashing", "later"]
assert proceed is True
def test_scoped_hooks_fire_on_agent_executor_llm_seams(self, mock_executor):
"""register_scoped hooks must run on the executor model seams.
Regression: `_setup_before/after_llm_call_hooks` only ran the
executor's snapshot lists, so execution-scoped hooks never fired on
PRE/POST_MODEL_CALL during normal agent execution (while tool seams,
which go through `dispatch`, merged them). Scoped hooks run after the
snapshot, matching dispatch's global-then-scoped ordering.
"""
from crewai.hooks import InterceptionPoint
from crewai.hooks.dispatch import register_scoped, scoped_hooks
from crewai.utilities.agent_utils import (
_setup_after_llm_call_hooks,
_setup_before_llm_call_hooks,
)
order: list[str] = []
def snapshot_hook(context):
order.append("snapshot")
mock_executor.before_llm_call_hooks = [snapshot_hook]
mock_executor.after_llm_call_hooks = []
with scoped_hooks():
register_scoped(
InterceptionPoint.PRE_MODEL_CALL,
lambda ctx: order.append("scoped_pre"),
)
register_scoped(
InterceptionPoint.POST_MODEL_CALL,
lambda ctx: order.append("scoped_post"),
)
proceed = _setup_before_llm_call_hooks(
mock_executor, printer=Mock(), verbose=False
)
answer = _setup_after_llm_call_hooks(
mock_executor, "answer", printer=Mock(), verbose=False
)
assert order == ["snapshot", "scoped_pre", "scoped_post"]
assert proceed is True
assert answer == "answer"
def test_intentional_block_still_short_circuits_later_hooks(self, mock_executor):
"""A hook returning False blocks the call and skips later hooks (unchanged)."""
from crewai.utilities.agent_utils import _setup_before_llm_call_hooks
ran: list[str] = []
def blocking_hook(context):
ran.append("blocking")
return False
def later_hook(context):
ran.append("later")
register_before_llm_call_hook(blocking_hook)
register_before_llm_call_hook(later_hook)
mock_executor.before_llm_call_hooks = get_before_llm_call_hooks()
proceed = _setup_before_llm_call_hooks(
mock_executor, printer=Mock(), verbose=False
)
assert ran == ["blocking"]
assert proceed is False
@pytest.mark.vcr()
def test_lite_agent_hooks_integration_with_real_llm(self):
"""Test that LiteAgent executes before/after LLM call hooks and prints messages correctly."""
@@ -463,3 +596,77 @@ class TestLLMHooksIntegration:
finally:
unregister_before_llm_call_hook(before_hook)
unregister_after_llm_call_hook(after_hook)
class TestDirectLLMScopedHooks:
"""Direct (agent-less) LLM calls must honor execution-scoped hooks.
Regression: the direct-call helpers used to short-circuit when the global
hook list was empty, so hooks registered only for the current
``scoped_hooks()`` context never ran on this path.
"""
@staticmethod
def _stub_llm():
from crewai.llms.base_llm import BaseLLM
class _StubLLM(BaseLLM):
def call(self, *args: object, **kwargs: object) -> str:
return ""
return _StubLLM(model="stub")
def test_scoped_before_hook_runs_on_direct_call(self):
from crewai.hooks import InterceptionPoint
from crewai.hooks.dispatch import register_scoped, scoped_hooks
llm = self._stub_llm()
seen: list[int] = []
with scoped_hooks():
register_scoped(
InterceptionPoint.PRE_MODEL_CALL,
lambda ctx: seen.append(len(ctx.messages)),
)
proceed = llm._invoke_before_llm_call_hooks(
[{"role": "user", "content": "hi"}], from_agent=None
)
assert proceed is True
assert seen == [1]
def test_scoped_before_hook_can_block_direct_call(self):
from crewai.hooks import InterceptionPoint
from crewai.hooks.dispatch import HookAborted, register_scoped, scoped_hooks
llm = self._stub_llm()
def block(ctx: LLMCallHookContext) -> None:
raise HookAborted(reason="blocked by scoped hook")
with scoped_hooks():
register_scoped(InterceptionPoint.PRE_MODEL_CALL, block)
proceed = llm._invoke_before_llm_call_hooks(
[{"role": "user", "content": "hi"}], from_agent=None
)
assert proceed is False
def test_scoped_after_hook_modifies_direct_response(self):
from crewai.hooks import InterceptionPoint
from crewai.hooks.dispatch import register_scoped, scoped_hooks
llm = self._stub_llm()
def redact(ctx: LLMCallHookContext) -> str:
return ctx.response.replace("SECRET", "[REDACTED]")
with scoped_hooks():
register_scoped(InterceptionPoint.POST_MODEL_CALL, redact)
result = llm._invoke_after_llm_call_hooks(
[{"role": "user", "content": "hi"}],
"contains SECRET",
from_agent=None,
)
assert result == "contains [REDACTED]"

69
lib/crewai/tests/hooks/test_tool_hooks.py
View File

@@ -576,6 +576,75 @@ class TestToolHooksIntegration:
unregister_after_tool_call_hook(after_tool_call_hook)
class TestPerHookFailOpen:
"""Fail-open is per-hook: a crashing hook must not disable its neighbors.
Regression guards for the dispatcher migration: previously each seam's
``except Exception`` wrapped the whole hook loop, so a raising hook
silently skipped every hook registered after it.
"""
def test_raising_before_hook_does_not_skip_later_hooks_or_block(
self, mock_tool, mock_agent
):
from crewai.hooks.tool_hooks import run_before_tool_call_hooks
mock_agent.verbose = False
ran: list[str] = []
def crashing_hook(context):
ran.append("crashing")
raise ValueError("bug in user hook")
def later_hook(context):
ran.append("later")
register_before_tool_call_hook(crashing_hook)
register_before_tool_call_hook(later_hook)
context = ToolCallHookContext(
tool_name="test_tool",
tool_input={"arg": "value"},
tool=mock_tool,
agent=mock_agent,
)
blocked = run_before_tool_call_hooks(context)
assert ran == ["crashing", "later"]
assert blocked is False
def test_raising_after_hook_does_not_skip_later_result_rewrites(
self, mock_tool, mock_agent
):
from crewai.hooks.tool_hooks import run_after_tool_call_hooks
mock_agent.verbose = False
ran: list[str] = []
def crashing_hook(context):
ran.append("crashing")
raise ValueError("bug in user hook")
def rewriting_hook(context):
ran.append("rewriting")
return f"{context.tool_result} [rewritten]"
register_after_tool_call_hook(crashing_hook)
register_after_tool_call_hook(rewriting_hook)
context = ToolCallHookContext(
tool_name="test_tool",
tool_input={"arg": "value"},
tool=mock_tool,
agent=mock_agent,
tool_result="original",
)
result = run_after_tool_call_hooks(context)
assert ran == ["crashing", "rewriting"]
assert result == "original [rewritten]"
class TestNativeToolCallingHooksIntegration:
"""Integration tests for hooks with native function calling (Agent and Crew)."""

2
lib/devtools/src/crewai_devtools/__init__.py
View File

@@ -1,3 +1,3 @@
"""CrewAI development tools."""
__version__ = "1.15.2"
__version__ = "1.15.3"