Remove redundant CEL text helper (#6528)

This commit removes the redundant CEL text helper, in favor of the
easier interpolation syntax.

docs/docs.json

@@ -375,8 +375,7 @@
                       "edge/en/learn/using-annotations",
                       "edge/en/learn/execution-hooks",
                       "edge/en/learn/llm-hooks",
-                      "edge/en/learn/tool-hooks",
-                      "edge/en/learn/interception-hooks"
+                      "edge/en/learn/tool-hooks"
                     ]
                   },
                   {

docs/edge/en/learn/execution-hooks.mdx

@@ -42,14 +42,6 @@ Control and monitor tool execution:
 
 [View Tool Hooks Documentation →](/learn/tool-hooks)
 
-<Note>
-LLM and tool hooks are two points in a larger catalog. See
-[Interception Hooks](/learn/interception-hooks) for every framework-native
-interception point (execution boundaries, steps, memory, knowledge, flow
-transitions, and more) and the shared payload-in/payload-out contract they all
-follow.
-</Note>
-
 ## Hook Registration Methods
 
 ### 1. Decorator-Based Hooks (Recommended)

docs/edge/en/learn/interception-hooks.mdx

@@ -1,162 +0,0 @@
----
-title: Interception Hooks
-description: The full catalog of framework-native interception points and the payload-in/payload-out contract every hook follows
-mode: "wide"
----
-
-Interception hooks give you a single, uniform way to observe and modify CrewAI's
-runtime at well-defined points — from the moment an execution starts, through
-every model call, tool call, memory read, and flow transition, down to the final
-output. All points share one contract and one registration API.
-
-The four LLM/tool hooks documented in [LLM Hooks](/learn/llm-hooks) and
-[Tool Hooks](/learn/tool-hooks) are the same mechanism. Their existing
-decorators (`@before_llm_call`, `@before_tool_call`, ...) and `return False`
-semantics keep working unchanged; interception hooks generalize the same engine
-to the rest of the framework.
-
-## The contract
-
-Every hook is a **synchronous** callable that receives a single typed context:
-
-```python
-from crewai.hooks import on, HookAborted, InterceptionPoint
-
-@on(InterceptionPoint.INPUT)
-def add_defaults(ctx):
-    # 1. Observe: read anything off the context.
-    # 2. Mutate in place: change ctx.payload or nested fields directly.
-    ctx.payload.setdefault("locale", "en-US")
-    # 3. Or replace: return a new value to swap ctx.payload.
-    # 4. Or abort: raise HookAborted(reason, source) to stop the operation.
-    return None
-```
-
-A hook may do any of four things:
-
-| Action | How | Effect |
-|--------|-----|--------|
-| **Proceed** | `return None` (or nothing) | Operation continues unchanged |
-| **Mutate** | Change `ctx.payload` / fields in place | Change is visible downstream |
-| **Replace** | `return new_payload` | A non-`None` return replaces `ctx.payload` |
-| **Abort** | `raise HookAborted(reason, source)` | Operation is stopped; the reason propagates |
-
-### Composition, ordering, and fail-open
-
-- Multiple hooks on the same point run in **registration order**, global hooks
-  first, then execution-scoped hooks.
-- The (possibly mutated) payload flows from one hook to the next.
-- `HookAborted` **propagates by design** and stops the chain.
-- Any *other* exception raised by a hook is **swallowed** (fail-open) so a single
-  buggy hook can't crash a run — the same protection the legacy hooks provide.
-- When no hook is registered for a point, dispatch is a single dict lookup
-  (no-op fast path), so unused points cost effectively nothing.
-
-## Registering hooks
-
-Use the `@on` decorator for global hooks. It mirrors the legacy decorators'
-ergonomics, including `agents=` / `tools=` filters:
-
-```python
-from crewai.hooks import on, InterceptionPoint, HookAborted
-
-@on(InterceptionPoint.PRE_TOOL_CALL, tools=["delete_file"])
-def guard_deletes(ctx):
-    raise HookAborted(reason="file deletion is not allowed", source="policy")
-```
-
-Applied to a method inside a `@CrewBase` class, `@on` registers a crew-scoped
-hook (active only while that crew runs), matching the existing crew-scoped hook
-behavior.
-
-## Interception point catalog
-
-`payload` is the value a hook may mutate or replace at each point.
-
-### Execution boundaries
-
-| Point | When | `payload` |
-|-------|------|-----------|
-| `EXECUTION_START` | A crew or flow is about to begin | inputs `dict` |
-| `INPUT` | Resolved inputs for the execution | inputs `dict` |
-| `OUTPUT` | Final result is ready | the output object |
-| `EXECUTION_END` | A crew or flow has finished | the output object |
-
-### Model & tool boundaries (legacy-compatible)
-
-| Point | When | `payload` |
-|-------|------|-----------|
-| `PRE_MODEL_CALL` | Before an LLM call | `LLMCallHookContext` |
-| `POST_MODEL_CALL` | After an LLM call | response |
-| `PRE_TOOL_CALL` | Before a tool runs | `ToolCallHookContext` |
-| `POST_TOOL_CALL` | After a tool runs | tool result |
-
-### Step & agent points
-
-| Point | When | `payload` |
-|-------|------|-----------|
-| `PRE_STEP` | Before a task or flow-method step | step input |
-| `POST_STEP` | After a task or flow-method step | step output |
-| `TOOL_SELECTION` | Tools are offered to an agent | list of tools |
-| `PRE_DELEGATION` | An agent is about to delegate | delegation input |
-| `RETRY_ATTEMPT` | An operation is about to be retried | retry input |
-
-`PRE_STEP` / `POST_STEP` carry `ctx.kind` (`"task"` or `"flow_method"`) and
-`ctx.step_name`.
-
-### Subsystem points
-
-| Point | When | `payload` |
-|-------|------|-----------|
-| `MEMORY_WRITE` | A value is about to be stored in memory | value |
-| `MEMORY_READ` | A memory query is issued | query |
-| `KNOWLEDGE_RETRIEVAL` | A knowledge query is issued | query |
-| `PRE_CODE_EXECUTION` | Code is about to run (flow `ScriptAction`) | code string |
-| `MCP_CONNECT` | An MCP client is about to connect | connection params |
-
-### Flow-specific points
-
-| Point | When | `payload` |
-|-------|------|-----------|
-| `FLOW_TRANSITION` | A flow moves to its triggered methods | list of target methods |
-| `ROUTER_DECISION` | A flow router picks a route | route label |
-
-## Aborting an operation
-
-`HookAborted` carries a `reason` and an optional `source`. The `source` defaults
-to the aborting hook when omitted, which is useful for telemetry and failure
-messages:
-
-```python
-@on(InterceptionPoint.EXECUTION_START)
-def enforce_policy(ctx):
-    if not ctx.payload.get("authorized"):
-        raise HookAborted(reason="unauthorized execution", source="access-control")
-```
-
-## Telemetry
-
-Whenever a point actually dispatches to at least one hook, CrewAI emits a
-`HookDispatchedEvent` on the event bus with the point, the outcome
-(`proceeded` / `modified` / `aborted`), the hook count, the duration, and — for
-aborts — the reason and source. The no-op fast path emits nothing.
-
-## Managing hooks in tests
-
-```python
-import pytest
-from crewai.hooks import clear_all_hooks
-
-@pytest.fixture(autouse=True)
-def reset_hooks():
-    clear_all_hooks()
-    yield
-    clear_all_hooks()
-```
-
-## Related documentation
-
-- [Execution Hooks Overview →](/learn/execution-hooks)
-- [LLM Call Hooks →](/learn/llm-hooks)
-- [Tool Call Hooks →](/learn/tool-hooks)
-- [Before and After Kickoff Hooks →](/learn/before-and-after-kickoff-hooks)

lib/cli/src/crewai_cli/templates/declarative_flow/AGENTS.md

@@ -59,9 +59,9 @@ Use these expression forms correctly:
 - Raw CEL: use in `expr`. Do not wrap raw CEL in `${...}`.
 - Use `${...}` inside action mapping strings to read Flow data with CEL. Example value: `Ticket: ${state.ticket_id}`.
 - Use `state` for input data. Use `outputs.step_name` for a completed method result.
+- In action mapping strings, keep literal text outside `${...}` and interpolate each Flow value directly. Write `Ticket: ${state.ticket_id}`; do not assemble the string with CEL `+`.
 - If a value is only one `${...}` expression, the result keeps its type. Use this for numbers, booleans, objects, and lists.
 - If the string has other text, the final value is text. Non-text values become JSON. `null` becomes empty text.
-- Use `text(root, "path", "default")` for values that may be missing or null. The default is optional and is `""`.
 
 Expression examples:
 
@@ -78,12 +78,6 @@ domains: "${state.domains}"
 limit: "${state.limit}"
 ```
 
-Use a default for missing text:
-
-```yaml
-input: "Ticket ${text(state, \"ticket.id\", \"unknown\")}"
-```
-
 - Crew text: use `{name}` placeholders from crew inputs. Example: `Research {topic}`.
 - Crew inputs become prompt text only when agent or task text references matching `{name}` placeholders.
 - Passing an input that is not referenced by any `{name}` placeholder does not ground the crew. If the crew needs a field, put that placeholder in an agent `goal`, task `description`, or task `expected_output`.
@@ -111,6 +105,7 @@ Dynamic value rules:
 - Do not use fields outside the declaration schema.
 - Do not put more than one action under a method's `do`.
 - Do not make `do` a list.
+- Do not use CEL `+` to build text in action mappings. Keep the text literal and insert each dynamic value with `${...}`.
 - Do not reference `outputs.some_method` before `some_method` can run.
 - Do not set a method's `listen` to its own method name.
 - Do not use the same string for an emitted event and a method name unless the user asks for it.
@@ -246,7 +241,7 @@ Shape:
 Fields:
 - `call` (required): must be `crew`. Action discriminator. Use crew to run an inline Crew definition. Example: `crew`
 - `with` (required): inline crew definition. Inline Crew definition to load and execute for this action. Example: `{"agents": {"researcher": {"backstory": "Knows the domain.", "goal": "Research {topic}", "role": "Researcher"}}, "name": "inline_research", "tasks": [{"agent": "researcher", "description": "Research {topic}", "expected_output": "Findings about {topic}", "name": "research_task"}]}`
-- `inputs` (optional): map of string to expression data | null; default `null`. Actual kickoff inputs passed to the Crew. Use `${...}` inside action mapping strings to read Flow data with CEL. Example value: `Ticket: ${state.ticket_id}`. Use `state` for input data. Use `outputs.step_name` for a completed method result. If a value is only one `${...}` expression, the result keeps its type. Use this for numbers, booleans, objects, and lists. If the string has other text, the final value is text. Non-text values become JSON. `null` becomes empty text. Use `text(root, "path", "default")` for values that may be missing or null. The default is optional and is `""`. The evaluated values are available to crew agent and task interpolation as `{name}` placeholders; reference each input the crew needs in agent or task text. Example: `{"topic": "${state.topic}"}`
+- `inputs` (optional): map of string to expression data | null; default `null`. Runtime inputs passed to the Crew. Insert Flow values with `${...}` and reference each input as `{name}` in agent or task text. Example: `{"topic": "${state.topic}"}`
 
 #### Crew Definition (`methods.<name>.do[call=crew].with`)
 
@@ -311,7 +306,7 @@ Fields:
 - `tools` (optional): list[string | map of string to any] | null; default `null`. Tool refs or serialized tool definitions available to this agent. String refs can use CrewAI tool names, `custom:<name>`, or fully qualified `module:Class` references. Example: `["crewai_tools:SerperDevTool", "custom:file_read"]`
 - `apps` (optional): list[string] | null; default `null`. Platform apps available to this agent. Can contain app names such as `gmail` or app/action refs such as `gmail/send_email`. Example: `["gmail", "slack/send_message"]`
 - `mcps` (optional): list[string | map of string to any] | null; default `null`. MCP server refs or serialized MCP server configs available to this agent. String refs can use HTTPS URLs, connected MCP integration slugs, or refs with a `#tool_name` suffix for specific tools. Example: `["https://api.weather.com/mcp#get_current_weather", "snowflake", "stripe#list_invoices", {"cache_tools_list": true, "headers": {"Authorization": "Bearer your_token"}, "streamable": true, "url": "https://api.example.com/mcp"}]`
-- `input` (required): string. Input passed to the individual agent kickoff outside of a crew. Use one string. Use `${...}` inside action mapping strings to read Flow data with CEL. Example value: `Ticket: ${state.ticket_id}`. Use `state` for input data. Use `outputs.step_name` for a completed method result. If a value is only one `${...}` expression, the result keeps its type. Use this for numbers, booleans, objects, and lists. If the string has other text, the final value is text. Non-text values become JSON. `null` becomes empty text. Use `text(root, "path", "default")` for values that may be missing or null. The default is optional and is `""`. When an agent needs multiple fields, write one string with labels and separators, for example `Ticket ID: ${state.ticket_id}; Message: ${state.message}`. Example: `${state.ticket.body}`
+- `input` (required): string. Agent prompt template. Insert Flow values with `${...}`, for example `Ticket: ${state.ticket_id}`. Example: `${state.ticket.body}`
 
 #### LLM Definition
 
@@ -349,4 +344,3 @@ Fields:
 - Crew action-level `inputs` are the Crew kickoff inputs; use CEL-wrapped strings there for runtime values.
 - Crew agent/task interpolation uses `{name}` placeholders from evaluated crew inputs.
 - Agent `with.input` must be text. Use `${outputs.method_name.raw}` or a text field like `${outputs.method_name.json_dict.summary}`.
-

lib/cli/tests/test_create_flow.py

@@ -29,7 +29,9 @@ def test_create_flow_declarative_project_can_run(
     agents_md = (project_root / "AGENTS.md").read_text(encoding="utf-8")
     assert "CrewAI Flow declaration" in agents_md
     assert "schema: crewai.flow/v1" in agents_md
-    assert 'text(root, "path", "default")' in agents_md
+    assert "do not assemble the string with CEL `+`" in agents_md
+    assert "Agent prompt template. Insert Flow values with `${...}`" in agents_md
+    assert "Runtime inputs passed to the Crew" in agents_md
     assert "call: expression" in agents_md
     assert "call: tool" not in agents_md
     assert "call: script" not in agents_md

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

@@ -656,22 +656,6 @@ class Agent(BaseAgent):
 
         return result
 
-    def _dispatch_retry_attempt(self, e: Exception, task: Task) -> None:
-        """Fire the ``retry_attempt`` interception point before re-executing a task."""
-        from crewai.hooks.contexts import RetryAttemptContext
-        from crewai.hooks.dispatch import InterceptionPoint, dispatch
-
-        retry_ctx = RetryAttemptContext(
-            agent=self,
-            agent_role=getattr(self, "role", None),
-            task=task,
-            attempt=self._times_executed,
-            max_attempts=self.max_retry_limit,
-            error=e,
-            payload=e,
-        )
-        dispatch(InterceptionPoint.RETRY_ATTEMPT, retry_ctx)
-
     def _check_execution_error(self, e: Exception, task: Task) -> None:
         """Check if an execution error should be re-raised immediately.
 
@@ -725,7 +709,6 @@ class Agent(BaseAgent):
             Result from retried execution.
         """
         self._check_execution_error(e, task)
-        self._dispatch_retry_attempt(e, task)
         return self.execute_task(task, context, tools)
 
     async def _handle_execution_error_async(
@@ -747,7 +730,6 @@ class Agent(BaseAgent):
             Result from retried execution.
         """
         self._check_execution_error(e, task)
-        self._dispatch_retry_attempt(e, task)
         return await self.aexecute_task(task, context, tools)
 
     def message(self, content: str, **kwargs: Any) -> str:
@@ -1072,21 +1054,6 @@ class Agent(BaseAgent):
             An instance of the CrewAgentExecutor class.
         """
         raw_tools: list[BaseTool] = tools or self.tools or []
-
-        from crewai.hooks.contexts import ToolSelectionContext
-        from crewai.hooks.dispatch import InterceptionPoint, dispatch
-
-        selection_ctx = ToolSelectionContext(
-            agent=self,
-            agent_role=getattr(self, "role", None),
-            task=task,
-            crew=self.crew,
-            tools=raw_tools,
-            payload=raw_tools,
-        )
-        dispatch(InterceptionPoint.TOOL_SELECTION, selection_ctx)
-        raw_tools = selection_ctx.payload
-
         parsed_tools = parse_tools(raw_tools)
 
         prompt, stop_words, rpm_limit_fn = self._build_execution_prompt(raw_tools)

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

@@ -46,8 +46,8 @@ from crewai.hooks.llm_hooks import (
 )
 from crewai.hooks.tool_hooks import (
     ToolCallHookContext,
-    run_after_tool_call_hooks,
-    run_before_tool_call_hooks,
+    get_after_tool_call_hooks,
+    get_before_tool_call_hooks,
 )
 from crewai.types.callback import SerializableCallable
 from crewai.utilities.agent_utils import (
@@ -951,6 +951,7 @@ 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 {},
@@ -959,7 +960,19 @@ class CrewAgentExecutor(BaseAgentExecutor):
             task=self.task,
             crew=self.crew,
         )
-        hook_blocked = run_before_tool_call_hooks(before_hook_context)
+        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",
+                )
 
         if hook_blocked:
             result = f"Tool execution blocked by hook. Tool: {func_name}"
@@ -1020,9 +1033,19 @@ class CrewAgentExecutor(BaseAgentExecutor):
             tool_result=result,
             raw_tool_result=raw_tool_result,
         )
-        modified_result = run_after_tool_call_hooks(after_hook_context)
-        if modified_result is not None:
-            result = modified_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",
+                )
 
         if not error_event_emitted:
             crewai_event_bus.emit(

lib/crewai/src/crewai/crew.py

@@ -1687,9 +1687,6 @@ class Crew(FlowTrackable, BaseModel):
             if files_needing_tool:
                 tools = self._add_file_tools(tools, files_needing_tool)
 
-        # TOOL_SELECTION is dispatched once, in Agent.create_agent_executor,
-        # which every crew task funnels through. Dispatching here as well would
-        # fire the point twice on a crew run (and duplicate additive edits).
         return tools
 
     def _get_agent_to_use(self, task: Task) -> BaseAgent | None:
@@ -1909,30 +1906,6 @@ 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/EXECUTION_END run before the kickoff-completed event (mirroring
-        # the flow OUTPUT-before-FlowFinishedEvent ordering) so a HookAborted
-        # prevents a spurious completed signal and any payload replacement is
-        # honored on the returned output.
-        output_ctx = OutputContext(crew=self, output=crew_output, payload=crew_output)
-        dispatch(InterceptionPoint.OUTPUT, output_ctx)
-        crew_output = output_ctx.payload
-
-        end_ctx = ExecutionEndContext(crew=self, output=crew_output, payload=crew_output)
-        dispatch(InterceptionPoint.EXECUTION_END, end_ctx)
-        crew_output = end_ctx.payload
-
         # Ensure background memory saves finish (and emit their
         # completed/failed events) before the kickoff-completed event below
         # triggers listener teardown/finalization.
@@ -1951,7 +1924,13 @@ class Crew(FlowTrackable, BaseModel):
         # Finalization is handled by trace listener (always initialized)
         # The batch manager checks contextvar to determine if tracing is enabled
 
-        return crew_output
+        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,
+        )
 
     def _process_async_tasks(
         self,

lib/crewai/src/crewai/crews/utils.py

@@ -278,9 +278,6 @@ 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):
@@ -289,30 +286,11 @@ 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

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

@@ -1,19 +0,0 @@
-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

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

@@ -62,8 +62,8 @@ from crewai.hooks.llm_hooks import (
 )
 from crewai.hooks.tool_hooks import (
     ToolCallHookContext,
-    run_after_tool_call_hooks,
-    run_before_tool_call_hooks,
+    get_after_tool_call_hooks,
+    get_before_tool_call_hooks,
 )
 from crewai.hooks.types import (
     AfterLLMCallHookCallable,
@@ -1975,6 +1975,7 @@ 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,
@@ -1983,7 +1984,19 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
             task=self.task,
             crew=self.crew,
         )
-        hook_blocked = run_before_tool_call_hooks(before_hook_context)
+        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",
+                )
 
         if hook_blocked:
             result = f"Tool execution blocked by hook. Tool: {func_name}"
@@ -2047,9 +2060,19 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
             tool_result=result,
             raw_tool_result=raw_tool_result,
         )
-        modified_result = run_after_tool_call_hooks(after_hook_context)
-        if modified_result is not None:
-            result = modified_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",
+                )
 
         if not error_event_emitted:
             crewai_event_bus.emit(

lib/crewai/src/crewai/flow/expressions.py

@@ -11,9 +11,6 @@ from crewai.utilities.serialization import to_serializable
 
 
 if TYPE_CHECKING:
-    from celpy.celtypes import StringType
-    from celpy.evaluation import CELFunction
-
     from crewai.flow.runtime import Flow
 else:
     from typing_extensions import TypeAliasType
@@ -24,29 +21,6 @@ _CEL_MACROS_WITH_LOCAL_BINDINGS = frozenset(
 )
 
 
-def _handle_text_custom_expression(
-    root: Any, path: Any, default: Any = ""
-) -> StringType:
-    from celpy.celtypes import StringType
-
-    fallback = StringType("" if default is None else str(default))
-    current = root
-    for part in str(path).split("."):
-        if current is None:
-            return fallback
-        try:
-            if isinstance(current, list):
-                current = current[int(part)]
-            else:
-                current = current[StringType(part)]
-        except (KeyError, IndexError, TypeError, ValueError):
-            return fallback
-
-    if current is None:
-        return fallback
-    return StringType(_stringify_cel_value(current))
-
-
 def _stringify_cel_value(value: Any) -> str:
     from celpy.adapter import CELJSONEncoder
 
@@ -98,21 +72,18 @@ def _parse_template_segments(value: str) -> tuple[str | _ExpressionSegment, ...]
     return tuple(segments)
 
 
-_EXPRESSION_FUNCTIONS: dict[str, CELFunction] = {
-    "text": _handle_text_custom_expression,
-}
-
 FLOW_TEMPLATE_EXPRESSION_RULES: tuple[str, ...] = (
     "Use `${...}` inside action mapping strings to read Flow data with CEL. "
     "Example value: `Ticket: ${state.ticket_id}`.",
     "Use `state` for input data. Use `outputs.step_name` for a completed "
     "method result.",
+    "In action mapping strings, keep literal text outside `${...}` and "
+    "interpolate each Flow value directly. Write `Ticket: ${state.ticket_id}`; "
+    "do not assemble the string with CEL `+`.",
     "If a value is only one `${...}` expression, the result keeps its type. "
     "Use this for numbers, booleans, objects, and lists.",
     "If the string has other text, the final value is text. Non-text values "
     "become JSON. `null` becomes empty text.",
-    'Use `text(root, "path", "default")` for values that may be missing '
-    'or null. The default is optional and is `""`.',
 )
 FLOW_TEMPLATE_EXPRESSION_CONTRACT = " ".join(FLOW_TEMPLATE_EXPRESSION_RULES)
 FLOW_TEMPLATE_EXPRESSION_EXAMPLES: dict[str, tuple[dict[str, str], ...]] = {
@@ -125,10 +96,6 @@ FLOW_TEMPLATE_EXPRESSION_EXAMPLES: dict[str, tuple[dict[str, str], ...]] = {
             "title": "Keep a list or number type",
             "code": 'domains: "${state.domains}"\nlimit: "${state.limit}"',
         },
-        {
-            "title": "Use a default for missing text",
-            "code": 'input: "Ticket ${text(state, \\"ticket.id\\", \\"unknown\\")}"',
-        },
     ),
     "json": (
         {
@@ -141,14 +108,6 @@ FLOW_TEMPLATE_EXPRESSION_EXAMPLES: dict[str, tuple[dict[str, str], ...]] = {
                 '{\n  "domains": "${state.domains}",\n  "limit": "${state.limit}"\n}'
             ),
         },
-        {
-            "title": "Use a default for missing text",
-            "code": (
-                "{\n"
-                '  "input": "Ticket ${text(state, \\"ticket.id\\", \\"unknown\\")}"\n'
-                "}"
-            ),
-        },
     ),
 }
 
@@ -374,8 +333,7 @@ class Expression:
 
             environment = Environment()
             program = environment.program(
-                Expression._compile_cel(expression, environment=environment),
-                functions=_EXPRESSION_FUNCTIONS,
+                Expression._compile_cel(expression, environment=environment)
             )
             result = program.evaluate(cast(Context, json_to_cel(context)))
             return json.loads(json.dumps(result, cls=CELJSONEncoder))

lib/crewai/src/crewai/flow/runtime/__init__.py

@@ -1476,22 +1476,6 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
             else (resumed_method_output if emit else result)
         )
 
-        # A resumed flow completes here rather than in kickoff_async, so the
-        # OUTPUT/EXECUTION_END seams must fire on this path too (before
-        # FlowFinishedEvent) to expose the final result to policy hooks.
-        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(
                 *[
@@ -2053,9 +2037,6 @@ 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(
@@ -2081,37 +2062,6 @@ 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
@@ -2347,21 +2297,6 @@ 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]
@@ -2435,8 +2370,6 @@ 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)
 
@@ -2629,33 +2562,6 @@ 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.
@@ -2683,16 +2589,6 @@ 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
@@ -2889,19 +2785,6 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                     if isinstance(router_result, enum.Enum)
                     else router_result
                 )
-
-                from crewai.hooks.contexts import RouterDecisionContext
-                from crewai.hooks.dispatch import InterceptionPoint, dispatch
-
-                router_ctx = RouterDecisionContext(
-                    flow=self,
-                    router_name=str(router_name),
-                    route=router_result,
-                    payload=router_result,
-                )
-                dispatch(InterceptionPoint.ROUTER_DECISION, router_ctx)
-                router_result = router_ctx.payload
-
                 router_result_str = str(router_result)
                 router_result_event = FlowMethodName(router_result_str)
                 router_results.append(router_result_event)
@@ -2930,19 +2813,6 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
                     current_trigger, router_only=False
                 )
                 if listeners_triggered:
-                    from crewai.hooks.contexts import FlowTransitionContext
-                    from crewai.hooks.dispatch import InterceptionPoint, dispatch
-
-                    transition_ctx = FlowTransitionContext(
-                        flow=self,
-                        from_method=str(trigger_method),
-                        to_methods=[str(name) for name in listeners_triggered],
-                        trigger=str(current_trigger),
-                        payload=listeners_triggered,
-                    )
-                    dispatch(InterceptionPoint.FLOW_TRANSITION, transition_ctx)
-                    listeners_triggered = transition_ctx.payload
-
                     listener_result = router_result_payloads.get(
                         str(current_trigger), result
                     )

lib/crewai/src/crewai/flow/runtime/_actions.py

@@ -224,34 +224,7 @@ class ScriptAction:
 
     def run(self, *args: Any, **kwargs: Any) -> Any:
         local_context = _pop_local_context(kwargs)
-
-        from crewai.hooks.contexts import PreCodeExecutionContext
-        from crewai.hooks.dispatch import InterceptionPoint, dispatch
-
-        code_ctx = PreCodeExecutionContext(
-            flow=self.flow,
-            code=self.definition.code,
-            language="python",
-            payload=self.definition.code,
-        )
-        dispatch(InterceptionPoint.PRE_CODE_EXECUTION, code_ctx)
-
-        # Honor a hook that rewrites the code, via either a returned payload
-        # replacement or an in-place ``ctx.code`` edit. Recompile only when the
-        # source actually changed so the common no-hook path stays free.
-        effective_code = (
-            code_ctx.payload
-            if isinstance(code_ctx.payload, str)
-            and code_ctx.payload != self.definition.code
-            else code_ctx.code
-        )
-        handler = (
-            self.handler
-            if effective_code == self.definition.code
-            else self._compile_handler(effective_code)
-        )
-
-        return handler(
+        return self.handler(
             state=self.flow.state,
             outputs=outputs_by_name(
                 self.flow._method_outputs,
@@ -261,7 +234,7 @@ class ScriptAction:
             item=local_context.get("item") if local_context else None,
         )
 
-    def _compile_handler(self, code: str | None = None) -> Callable[..., Any]:
+    def _compile_handler(self) -> Callable[..., Any]:
         raw = os.environ.get(_ALLOW_SCRIPT_EXECUTION_ENV_VAR, "")
         if raw.strip().lower() not in _TRUSTED_SCRIPT_EXECUTION_VALUES:
             raise FlowScriptExecutionDisabledError(
@@ -270,9 +243,8 @@ class ScriptAction:
                 "trusted flow definitions."
             )
 
-        source = code if code is not None else self.definition.code
         filename = f"crewai.flow.script.{self.flow._definition.name}"
-        module = ast.parse(source, filename=filename)
+        module = ast.parse(self.definition.code, filename=filename)
         function = ast.FunctionDef(
             name="_flow_script",
             args=ast.arguments(

lib/crewai/src/crewai/flow/skill.py

@@ -11,7 +11,6 @@ from jinja2 import Environment, FileSystemLoader
 import yaml
 
 from crewai.flow.expressions import (
-    FLOW_TEMPLATE_EXPRESSION_CONTRACT,
     FLOW_TEMPLATE_EXPRESSION_EXAMPLES,
     FLOW_TEMPLATE_EXPRESSION_RULES,
 )
@@ -186,7 +185,14 @@ MODEL_SPECS: tuple[ModelSpec, ...] = (
         hidden=True,
     ),
     ModelSpec("FlowScriptActionDefinition", "Action", "methods.<name>.do[call=script]"),
-    ModelSpec("FlowToolActionDefinition", "Action", "methods.<name>.do[call=tool]"),
+    ModelSpec(
+        "FlowToolActionDefinition",
+        "Action",
+        "methods.<name>.do[call=tool]",
+        descriptions={
+            "with": "Tool input arguments. Insert Flow values with `${...}`.",
+        },
+    ),
     ModelSpec(
         "FlowCrewActionDefinition",
         "Action",
@@ -194,7 +200,7 @@ MODEL_SPECS: tuple[ModelSpec, ...] = (
         examples=True,
         descriptions={
             "call": "Action discriminator. Use crew to run an inline Crew definition.",
-            "inputs": f"Actual kickoff inputs passed to the Crew. {FLOW_TEMPLATE_EXPRESSION_CONTRACT} The evaluated values are available to crew agent and task interpolation as `{{name}}` placeholders; reference each input the crew needs in agent or task text.",
+            "inputs": "Runtime inputs passed to the Crew. Insert Flow values with `${...}` and reference each input as `{name}` in agent or task text.",
         },
     ),
     ModelSpec(
@@ -262,7 +268,7 @@ MODEL_SPECS: tuple[ModelSpec, ...] = (
         hidden=True,
         examples=True,
         descriptions={
-            "input": f"Input passed to the individual agent kickoff outside of a crew. Use one string. {FLOW_TEMPLATE_EXPRESSION_CONTRACT} When an agent needs multiple fields, write one string with labels and separators, for example `Ticket ID: ${{state.ticket_id}}; Message: ${{state.message}}`.",
+            "input": "Agent prompt template. Insert Flow values with `${...}`, for example `Ticket: ${state.ticket_id}`.",
             "llm": "Language model that runs this agent. Use an object when setting LLM options such as `max_tokens`.",
             "planning_config": "Agent planning configuration. Set `max_attempts` to limit planning refinement attempts before task execution.",
         },

lib/crewai/src/crewai/flow/templates/flow_definition_skill.md.j2

@@ -46,6 +46,7 @@ Pick the simplest action that does the job.
 - Use `call: tool` for packaged deterministic work: API calls, searches, lookups, scoring, file work, or custom CrewAI tools.
 {% endif %}
 - Use `call: agent` for one AI worker that classifies, decides, summarizes, writes, or drafts. Put `role`, `goal`, `backstory`, and `input` under `with`. Do not add an action-level `inputs` map to an agent.
+- Repository-backed agents may set `from_repository` and omit inline `role`, `goal`, and `backstory`. Explicitly provided fields override repository values.
 - Use `call: crew` for coordinated AI work with multiple agents or tasks. Define the crew under `with`. Pass runtime values with the action-level `inputs` map.
 {% if include_each_action %}
 - Use `call: each` when the same ordered mini-pipeline must run once per item. Give every step a `name`.
@@ -137,6 +138,7 @@ Dynamic value rules:
 - Do not use fields outside the declaration schema{% if include_tool_action %} or tool refs shown here{% endif %}.
 - Do not put more than one action under a method's `do`.
 - Do not make `do` a list.
+- Do not use CEL `+` to build text in action mappings. Keep the text literal and insert each dynamic value with `${...}`.
 - Do not reference `outputs.some_method` before `some_method` can run.
 - Do not set a method's `listen` to its own method name.
 - Do not use the same string for an emitted event and a method name unless the user asks for it.

lib/crewai/src/crewai/hooks/__init__.py

@@ -6,17 +6,6 @@ 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,
@@ -85,8 +74,6 @@ def clear_all_global_hooks() -> dict[str, tuple[int, int]]:
 
 
 __all__ = [
-    "HookAborted",
-    "InterceptionPoint",
     "LLMCallHookContext",
     "ToolCallHookContext",
     "after_llm_call",
@@ -96,27 +83,20 @@ __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",
 ]

lib/crewai/src/crewai/hooks/contexts.py

@@ -1,151 +0,0 @@
-"""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
-
-
-@dataclass
-class ToolSelectionContext(InterceptionContext):
-    """``tool_selection``: the set of tools offered to an agent. ``payload`` = tools list."""
-
-    tools: list[Any] = field(default_factory=list)
-
-
-@dataclass
-class PreDelegationContext(InterceptionContext):
-    """``pre_delegation``: an agent is about to delegate work. ``payload`` = delegation input."""
-
-    coworker: str | None = None
-    delegate_to: Any = None
-
-
-@dataclass
-class RetryAttemptContext(InterceptionContext):
-    """``retry_attempt``: an operation is about to be retried."""
-
-    attempt: int = 0
-    max_attempts: int | None = None
-    error: Any = None
-
-
-@dataclass
-class MemoryWriteContext(InterceptionContext):
-    """``memory_write``: a value is about to be written to memory. ``payload`` = value."""
-
-    memory_type: str | None = None
-    metadata: dict[str, Any] = field(default_factory=dict)
-
-
-@dataclass
-class MemoryReadContext(InterceptionContext):
-    """``memory_read``: a memory query is being issued. ``payload`` = query (pre) / results (post)."""
-
-    memory_type: str | None = None
-    query: str | None = None
-
-
-@dataclass
-class KnowledgeRetrievalContext(InterceptionContext):
-    """``knowledge_retrieval``: a knowledge query. ``payload`` = query / retrieved results."""
-
-    query: Any = None
-
-
-@dataclass
-class PreCodeExecutionContext(InterceptionContext):
-    """``pre_code_execution``: code is about to run. ``payload`` = the code string."""
-
-    code: str | None = None
-    language: str | None = None
-
-
-@dataclass
-class MCPConnectContext(InterceptionContext):
-    """``mcp_connect``: an MCP client is about to connect. ``payload`` = connection params."""
-
-    server_name: str | None = None
-    server_params: Any = None
-
-
-@dataclass
-class FlowTransitionContext(InterceptionContext):
-    """``flow_transition``: a flow is moving to triggered methods."""
-
-    from_method: str | None = None
-    to_methods: list[str] = field(default_factory=list)
-    trigger: str | None = None
-
-
-@dataclass
-class RouterDecisionContext(InterceptionContext):
-    """``router_decision``: a flow router is choosing a route. ``payload`` = route label."""
-
-    router_name: str | None = None
-    route: Any = None

lib/crewai/src/crewai/hooks/dispatch.py

@@ -1,446 +0,0 @@
-"""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 & agent points
-    PRE_STEP = "pre_step"
-    POST_STEP = "post_step"
-    TOOL_SELECTION = "tool_selection"
-    PRE_DELEGATION = "pre_delegation"
-    RETRY_ATTEMPT = "retry_attempt"
-
-    # Subsystem points
-    MEMORY_WRITE = "memory_write"
-    MEMORY_READ = "memory_read"
-    KNOWLEDGE_RETRIEVAL = "knowledge_retrieval"
-    PRE_CODE_EXECUTION = "pre_code_execution"
-    MCP_CONNECT = "mcp_connect"
-
-    # Flow-specific points
-    FLOW_TRANSITION = "flow_transition"
-    ROUTER_DECISION = "router_decision"
-
-
-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

lib/crewai/src/crewai/hooks/llm_hooks.py

@@ -5,11 +5,6 @@ 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,
@@ -155,37 +150,8 @@ class LLMCallHookContext:
             event_listener.formatter.resume_live_updates()
 
 
-# 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
+_before_llm_call_hooks: list[BeforeLLMCallHookType | BeforeLLMCallHookCallable] = []
+_after_llm_call_hooks: list[AfterLLMCallHookType | AfterLLMCallHookCallable] = []
 
 
 def register_before_llm_call_hook(

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

@@ -5,12 +5,6 @@ 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,
@@ -127,81 +121,8 @@ class ToolCallHookContext:
             event_listener.formatter.resume_live_updates()
 
 
-# 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
+_before_tool_call_hooks: list[BeforeToolCallHookType | BeforeToolCallHookCallable] = []
+_after_tool_call_hooks: list[AfterToolCallHookType | AfterToolCallHookCallable] = []
 
 
 def register_before_tool_call_hook(

lib/crewai/src/crewai/knowledge/knowledge.py

@@ -145,13 +145,6 @@ class Knowledge(BaseModel):
         if self.storage is None:
             raise ValueError("Storage is not initialized.")
 
-        from crewai.hooks.contexts import KnowledgeRetrievalContext
-        from crewai.hooks.dispatch import InterceptionPoint, dispatch
-
-        retrieval_ctx = KnowledgeRetrievalContext(query=query, payload=query)
-        dispatch(InterceptionPoint.KNOWLEDGE_RETRIEVAL, retrieval_ctx)
-        query = retrieval_ctx.payload
-
         return self.storage.search(
             query,
             limit=results_limit,
@@ -190,13 +183,6 @@ class Knowledge(BaseModel):
         if self.storage is None:
             raise ValueError("Storage is not initialized.")
 
-        from crewai.hooks.contexts import KnowledgeRetrievalContext
-        from crewai.hooks.dispatch import InterceptionPoint, dispatch
-
-        retrieval_ctx = KnowledgeRetrievalContext(query=query, payload=query)
-        dispatch(InterceptionPoint.KNOWLEDGE_RETRIEVAL, retrieval_ctx)
-        query = retrieval_ctx.payload
-
         return await self.storage.asearch(
             query,
             limit=results_limit,

lib/crewai/src/crewai/llms/base_llm.py

@@ -1007,14 +1007,15 @@ 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,
-            before_llm_call_reducer,
+            get_before_llm_call_hooks,
         )
 
-        # 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.
+        before_hooks = get_before_llm_call_hooks()
+        if not before_hooks:
+            return True
+
         hook_context = LLMCallHookContext(
             executor=None,
             messages=messages,
@@ -1023,19 +1024,24 @@ class BaseLLM(BaseModel, ABC):
             task=None,
             crew=None,
         )
+        verbose = getattr(from_agent, "verbose", True) if from_agent else True
 
         try:
-            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
+            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",
+                )
 
         return True
 
@@ -1068,14 +1074,17 @@ class BaseLLM(BaseModel, ABC):
         if from_agent is not None or not isinstance(response, str):
             return response
 
-        from crewai.hooks.dispatch import InterceptionPoint, dispatch
+        from crewai_core.printer import PRINTER
+
         from crewai.hooks.llm_hooks import (
             LLMCallHookContext,
-            after_llm_call_reducer,
+            get_after_llm_call_hooks,
         )
 
-        # 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.
+        after_hooks = get_after_llm_call_hooks()
+        if not after_hooks:
+            return response
+
         hook_context = LLMCallHookContext(
             executor=None,
             messages=messages,
@@ -1085,11 +1094,20 @@ class BaseLLM(BaseModel, ABC):
             crew=None,
             response=response,
         )
+        verbose = getattr(from_agent, "verbose", True) if from_agent else True
+        modified_response = response
 
-        dispatch(
-            InterceptionPoint.POST_MODEL_CALL,
-            hook_context,
-            reducer=after_llm_call_reducer,
-        )
+        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",
+                )
 
-        return hook_context.response if hook_context.response is not None else response
+        return modified_response

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

@@ -152,20 +152,6 @@ class MCPClient:
         server_name, server_url, transport_type = self._get_server_info()
         is_reconnect = self._was_connected
 
-        from crewai.hooks.contexts import MCPConnectContext
-        from crewai.hooks.dispatch import InterceptionPoint, dispatch
-
-        connect_ctx = MCPConnectContext(
-            server_name=server_name,
-            server_params=self.transport,
-            payload=self.transport,
-        )
-        dispatch(InterceptionPoint.MCP_CONNECT, connect_ctx)
-        # Honor a hook that replaces the connection transport/params so the
-        # connection below actually uses the returned value.
-        if connect_ctx.payload is not None:
-            self.transport = connect_ctx.payload
-
         started_at = datetime.now()
         crewai_event_bus.emit(
             self,

lib/crewai/src/crewai/memory/unified_memory.py

@@ -466,18 +466,6 @@ class Memory(BaseModel):
         if self.read_only:
             return None
 
-        from crewai.hooks.contexts import MemoryWriteContext
-        from crewai.hooks.dispatch import InterceptionPoint, dispatch
-
-        write_ctx = MemoryWriteContext(
-            agent_role=agent_role,
-            memory_type="unified_memory",
-            metadata=metadata or {},
-            payload=content,
-        )
-        dispatch(InterceptionPoint.MEMORY_WRITE, write_ctx)
-        content = write_ctx.payload
-
         # Determine effective root_scope: per-call override takes precedence
         effective_root = root_scope if root_scope is not None else self.root_scope
 
@@ -573,18 +561,6 @@ class Memory(BaseModel):
         if not contents or self.read_only:
             return []
 
-        from crewai.hooks.contexts import MemoryWriteContext
-        from crewai.hooks.dispatch import InterceptionPoint, dispatch
-
-        write_ctx = MemoryWriteContext(
-            agent_role=agent_role,
-            memory_type="unified_memory",
-            metadata=metadata or {},
-            payload=contents,
-        )
-        dispatch(InterceptionPoint.MEMORY_WRITE, write_ctx)
-        contents = write_ctx.payload
-
         # Determine effective root_scope: per-call override takes precedence
         effective_root = root_scope if root_scope is not None else self.root_scope
 
@@ -736,17 +712,6 @@ class Memory(BaseModel):
         # so that the search sees all persisted records.
         self.drain_writes()
 
-        from crewai.hooks.contexts import MemoryReadContext
-        from crewai.hooks.dispatch import InterceptionPoint, dispatch
-
-        read_ctx = MemoryReadContext(
-            memory_type="unified_memory",
-            query=query,
-            payload=query,
-        )
-        dispatch(InterceptionPoint.MEMORY_READ, read_ctx)
-        query = read_ctx.payload
-
         effective_scope = scope
         if effective_scope is None and self.root_scope:
             effective_scope = self.root_scope

lib/crewai/src/crewai/project/crew_base.py

@@ -452,15 +452,7 @@ def _register_crew_hooks(instance: CrewInstance, cls: type) -> None:
         )
     }
 
-    # 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:
+    if not hook_methods:
         return
 
     from crewai.hooks import (
@@ -596,25 +588,6 @@ 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
 
 

lib/crewai/src/crewai/task.py

@@ -662,21 +662,6 @@ 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,
@@ -733,18 +718,6 @@ 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 = post_step_ctx.payload
-
             self.output = task_output
             self.end_time = datetime.datetime.now()
 
@@ -814,21 +787,6 @@ 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,
@@ -885,18 +843,6 @@ 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 = post_step_ctx.payload
-
             self.output = task_output
             self.end_time = datetime.datetime.now()
 
@@ -938,32 +884,6 @@ class Task(BaseModel):
             clear_task_files(self.id)
             reset_current_task_id(task_id_token)
 
-    def _dispatch_guardrail_retry_attempt(
-        self,
-        agent: BaseAgent | None,
-        context: str | None,
-        attempt: int,
-        error: Any,
-    ) -> str | None:
-        """Fire ``retry_attempt`` before re-executing a task after a guardrail failure.
-
-        Returns the (possibly hook-modified) retry context.
-        """
-        from crewai.hooks.contexts import RetryAttemptContext
-        from crewai.hooks.dispatch import InterceptionPoint, dispatch
-
-        retry_ctx = RetryAttemptContext(
-            agent=agent,
-            agent_role=getattr(agent, "role", None),
-            task=self,
-            attempt=attempt,
-            max_attempts=self.guardrail_max_retries,
-            error=error,
-            payload=context,
-        )
-        dispatch(InterceptionPoint.RETRY_ATTEMPT, retry_ctx)
-        return retry_ctx.payload
-
     def _post_agent_execution(self, agent: BaseAgent) -> None:
         pass
 
@@ -1397,13 +1317,6 @@ Follow these guidelines:
                     color="yellow",
                 )
 
-            context = self._dispatch_guardrail_retry_attempt(
-                agent=agent,
-                context=context,
-                attempt=current_retry_count,
-                error=guardrail_result.error,
-            )
-
             result = agent.execute_task(
                 task=self,
                 context=context,
@@ -1514,13 +1427,6 @@ Follow these guidelines:
                     color="yellow",
                 )
 
-            context = self._dispatch_guardrail_retry_attempt(
-                agent=agent,
-                context=context,
-                attempt=current_retry_count,
-                error=guardrail_result.error,
-            )
-
             result = await agent.aexecute_task(
                 task=self,
                 context=context,

lib/crewai/src/crewai/tools/agent_tools/base_agent_tools.py

@@ -108,22 +108,6 @@ class BaseAgentTool(BaseTool):
             )
 
         selected_agent = agent[0]
-
-        from crewai.hooks.contexts import PreDelegationContext
-        from crewai.hooks.dispatch import InterceptionPoint, dispatch
-
-        # Dispatched outside the try/except below so a HookAborted propagates
-        # instead of being swallowed into a tool-error string.
-        delegation_ctx = PreDelegationContext(
-            agent=selected_agent,
-            agent_role=getattr(selected_agent, "role", None),
-            coworker=sanitized_name,
-            delegate_to=selected_agent,
-            payload=task,
-        )
-        dispatch(InterceptionPoint.PRE_DELEGATION, delegation_ctx)
-        task = delegation_ctx.payload
-
         try:
             task_with_assigned_agent = Task(
                 description=task,

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

@@ -879,22 +879,6 @@ class ToolUsage:
                 return ToolUsageError(
                     f"{I18N_DEFAULT.errors('tool_usage_error').format(error=e)}\nMoving on then. {I18N_DEFAULT.slice('format').format(tool_names=self.tools_names)}"
                 )
-
-            from crewai.hooks.contexts import RetryAttemptContext
-            from crewai.hooks.dispatch import InterceptionPoint, dispatch
-
-            retry_ctx = RetryAttemptContext(
-                agent=self.agent,
-                agent_role=getattr(self.agent, "role", None),
-                task=self.task,
-                attempt=self._run_attempts,
-                max_attempts=self._max_parsing_attempts,
-                error=e,
-                payload=tool_string,
-            )
-            dispatch(InterceptionPoint.RETRY_ATTEMPT, retry_ctx)
-            tool_string = retry_ctx.payload
-
             return self._tool_calling(tool_string)
 
     def _validate_tool_input(self, tool_input: str | None) -> dict[str, Any]:

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

@@ -1453,8 +1453,8 @@ def execute_single_native_tool_call(
     )
     from crewai.hooks.tool_hooks import (
         ToolCallHookContext,
-        run_after_tool_call_hooks,
-        run_before_tool_call_hooks,
+        get_after_tool_call_hooks,
+        get_before_tool_call_hooks,
     )
 
     info = extract_tool_call_info(tool_call)
@@ -1517,6 +1517,7 @@ 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,
@@ -1525,7 +1526,13 @@ def execute_single_native_tool_call(
         task=task,
         crew=crew,
     )
-    hook_blocked = run_before_tool_call_hooks(before_hook_context)
+    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
 
     error_event_emitted = False
     if hook_blocked:
@@ -1580,9 +1587,14 @@ def execute_single_native_tool_call(
         tool_result=result,
         raw_tool_result=raw_tool_result,
     )
-    modified_result = run_after_tool_call_hooks(after_hook_context)
-    if modified_result is not None:
-        result = modified_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
 
     if not error_event_emitted:
         crewai_event_bus.emit(
@@ -1678,42 +1690,28 @@ def _setup_before_llm_call_hooks(
     Returns:
         True if LLM execution should proceed, False if blocked by a hook.
     """
-    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
+    if executor_context and executor_context.before_llm_call_hooks:
+        from crewai.hooks.llm_hooks import LLMCallHookContext
 
         original_messages = executor_context.messages
 
         hook_context = LLMCallHookContext(executor_context)
         try:
-            run_hooks(
-                InterceptionPoint.PRE_MODEL_CALL,
-                hook_context,
-                hooks,
-                reducer=before_llm_call_reducer,
-                verbose=verbose,
-            )
-        except HookAborted:
+            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:
             if verbose:
                 printer.print(
-                    content="LLM call blocked by before_llm_call hook",
+                    content=f"Error in before_llm_call hook: {e}",
                     color="yellow",
                 )
-            return False
 
         if not isinstance(executor_context.messages, list):
             if verbose:
@@ -1750,18 +1748,8 @@ def _setup_after_llm_call_hooks(
     Returns:
         The potentially modified response (string or Pydantic model).
     """
-    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
-
-        # 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
+    if executor_context and executor_context.after_llm_call_hooks:
+        from crewai.hooks.llm_hooks import LLMCallHookContext
 
         original_messages = executor_context.messages
 
@@ -1774,15 +1762,18 @@ def _setup_after_llm_call_hooks(
             hook_response = str(answer)
 
         hook_context = LLMCallHookContext(executor_context, response=hook_response)
-        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
+        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",
+                )
 
         if not isinstance(executor_context.messages, list):
             if verbose:

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

@@ -6,14 +6,15 @@ from crewai.agents.parser import AgentAction
 from crewai.agents.tools_handler import ToolsHandler
 from crewai.hooks.tool_hooks import (
     ToolCallHookContext,
-    run_after_tool_call_hooks,
-    run_before_tool_call_hooks,
+    get_after_tool_call_hooks,
+    get_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
 
 
@@ -56,10 +57,11 @@ 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:
@@ -100,27 +102,18 @@ async def aexecute_tool_and_check_finality(
             crew=crew,
         )
 
-        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,
-            )
+        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}")
 
         tool_result = await tool_usage.ause(tool_calling, agent_action.text)
         raw_tool_result = tool_usage.get_last_raw_result(tool_result)
@@ -136,12 +129,18 @@ async def aexecute_tool_and_check_finality(
             raw_tool_result=raw_tool_result,
         )
 
-        modified_result = run_after_tool_call_hooks(after_hook_context)
+        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}")
 
-        return ToolResult(
-            modified_result if modified_result is not None else tool_result,
-            tool.result_as_answer,
-        )
+        return ToolResult(modified_result, tool.result_as_answer)
 
     tool_result = I18N_DEFAULT.errors("wrong_tool_name").format(
         tool=sanitized_tool_name,
@@ -182,6 +181,7 @@ 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,27 +222,18 @@ def execute_tool_and_check_finality(
             crew=crew,
         )
 
-        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,
-            )
+        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}")
 
         tool_result = tool_usage.use(tool_calling, agent_action.text)
         raw_tool_result = tool_usage.get_last_raw_result(tool_result)
@@ -258,12 +249,18 @@ def execute_tool_and_check_finality(
             raw_tool_result=raw_tool_result,
         )
 
-        modified_result = run_after_tool_call_hooks(after_hook_context)
+        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}")
 
-        return ToolResult(
-            modified_result if modified_result is not None else tool_result,
-            tool.result_as_answer,
-        )
+        return ToolResult(modified_result, tool.result_as_answer)
 
     tool_result = I18N_DEFAULT.errors("wrong_tool_name").format(
         tool=sanitized_tool_name,

lib/crewai/tests/hooks/test_crew_scoped_hooks.py

@@ -306,62 +306,6 @@ 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."""
 

lib/crewai/tests/hooks/test_dispatch.py

@@ -1,296 +0,0 @@
-"""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

lib/crewai/tests/hooks/test_interception_conformance.py

@@ -1,178 +0,0 @@
-"""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. Enterprise / ACS adapters build
-against these guarantees.
-"""
-
-from __future__ import annotations
-
-from crewai.flow.flow import Flow, listen, router, start
-from crewai.hooks.dispatch import (
-    HookAborted,
-    InterceptionPoint,
-    clear_all,
-    on,
-)
-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 _RouterFlow(Flow):
-    @start()
-    def begin(self):
-        return "begin"
-
-    @router(begin)
-    def route(self):
-        return "go_left"
-
-    @listen("go_left")
-    def left(self):
-        return "left"
-
-    @listen("go_right")
-    def right(self):
-        return "right"
-
-
-class TestFlowTransitionAndRouter:
-    """flow_transition and router_decision on a routed flow."""
-
-    def test_transition_payload_carries_from_and_to(self):
-        seen: list[tuple[str | None, list[str]]] = []
-
-        @on(InterceptionPoint.FLOW_TRANSITION)
-        def capture(ctx):
-            seen.append((ctx.from_method, list(ctx.to_methods)))
-
-        _RouterFlow().kickoff()
-
-        assert any(to == ["left"] for _from, to in seen)
-
-    def test_router_decision_fires_with_route(self):
-        routes: list[object] = []
-
-        @on(InterceptionPoint.ROUTER_DECISION)
-        def capture(ctx):
-            routes.append(ctx.route)
-
-        _RouterFlow().kickoff()
-        assert "go_left" in routes
-
-    def test_router_decision_can_reroute(self):
-        @on(InterceptionPoint.ROUTER_DECISION)
-        def reroute(ctx):
-            return "go_right"
-
-        landed: list[str] = []
-
-        @on(InterceptionPoint.PRE_STEP)
-        def track(ctx):
-            landed.append(ctx.step_name)
-
-        _RouterFlow().kickoff()
-        assert "right" in landed
-        assert "left" not in landed

lib/crewai/tests/hooks/test_llm_hooks.py

@@ -303,105 +303,6 @@ 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."""
@@ -562,77 +463,3 @@ 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]"

lib/crewai/tests/hooks/test_tool_hooks.py

@@ -576,75 +576,6 @@ 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)."""
 

lib/crewai/tests/test_flow_definition.py

@@ -1348,7 +1348,15 @@ def test_skill_documents_flow_wiring():
     assert "```yaml" in skill
     assert "[Method](#method-methods)" in skill
     assert 'input: "Reviewed research: ${outputs.research_brief.raw}"' in skill
-    assert 'text(root, "path", "default")' in skill
+    assert "do not assemble the string with CEL `+`" in skill
+    assert "Do not use CEL `+` to build text in action mappings" in skill
+    assert "Agent prompt template. Insert Flow values with `${...}`" in skill
+    assert (
+        "Repository-backed agents may set `from_repository` and omit inline "
+        "`role`, `goal`, and `backstory`" in skill
+    )
+    assert "Runtime inputs passed to the Crew" in skill
+    assert "Tool input arguments. Insert Flow values with `${...}`" in skill
     assert "trust CrewAI defaults and omit them" in skill
     assert "#### LLM Definition" in skill
     assert "`max_tokens` (optional): integer | null; default `null`" in skill

lib/crewai/tests/test_flow_from_definition.py

@@ -1102,7 +1102,7 @@ methods:
     )
 
 
-def test_tool_action_renders_text_custom_expression_inputs():
+def test_tool_action_renders_interpolated_inputs():
     yaml_str = f"""
 schema: crewai.flow/v1
 name: ToolFlow
@@ -1112,8 +1112,8 @@ methods:
       call: tool
       ref: {__name__}:StaticSearchTool
       with:
-        search_query: "${{'Ticket ID: ' + text(state, 'ticket.id') + '; Subject: ' + text(state, 'ticket.subject') + '; Priority: ' + text(state, 'priority', 'unknown') + '; Message: ' + text(state, 'messages.0.body')}}"
-        prefix: "${{text(state, 'ticket')}}"
+        search_query: "Ticket ID: ${{state.ticket.id}}; Subject: ${{state.ticket.subject}}; Message: ${{state.messages[0].body}}"
+        prefix: "${{state.prefix}}"
     start: true
 """
 
@@ -1124,9 +1124,10 @@ methods:
             inputs={
                 "ticket": {"id": 123, "subject": None},
                 "messages": [{"body": "Initial report"}],
+                "prefix": "ticket",
             }
         )
-        == '{"id": 123, "subject": null}:Ticket ID: 123; Subject: ; Priority: unknown; Message: Initial report'
+        == "ticket:Ticket ID: 123; Subject: ; Message: Initial report"
     )
 
 
@@ -1319,7 +1320,7 @@ methods:
         role: Analyst
         goal: Answer questions
         backstory: Knows things.
-        input: "Ticket ID: ${text(state, 'ticket.id')}; Subject: ${text(state, 'ticket.subject')}"
+        input: "Ticket ID: ${state.ticket.id}; Subject: ${state.ticket.subject}"
     start: true
 """
 
@@ -2909,37 +2910,6 @@ def test_explicit_cel_fields_accept_expression_markers():
     assert Flow.from_declaration(contents=definition).kickoff(inputs={"score": 90}) == "qualified"
 
 
-def test_expression_action_runs_text_custom_expression():
-    definition = FlowDefinition.from_declaration(contents=
-        {
-            "schema": "crewai.flow/v1",
-            "name": "ExpressionFlow",
-            "methods": {
-                "summarize": {
-                    "start": True,
-                    "do": {
-                        "call": "expression",
-                        "expr": (
-                            "'Ticket ID: ' + text(state, 'ticket.id') + "
-                            "'; Tags: ' + text(state, 'tags')"
-                        ),
-                    },
-                }
-            },
-        }
-    )
-
-    assert (
-        Flow.from_declaration(contents=definition).kickoff(
-            inputs={
-                "ticket": {"id": 123},
-                "tags": ["urgent", "billing"],
-            }
-        )
-        == 'Ticket ID: 123; Tags: ["urgent", "billing"]'
-    )
-
-
 def test_expression_local_context_recurses_into_dataclass_values():
     from crewai.flow.expressions import Expression