refactor: defer tool_selection and memory interception points

Pulls `tool_selection`, `memory_write`, and `memory_read` out of the
catalog. The tool seam needs a defined matching contract first (platform
and MCP tools expose different name shapes and attributes), and the
memory seams need real skip semantics instead of surfacing a policy
abort as an error log. Both are tracked as follow-up cards; the points
will return once those contracts are settled.

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

@@ -6,8 +6,8 @@ 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, and memory read, down to the final output. All
-points share one contract and one registration API.
+every model call, tool call, and knowledge lookup, 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
@@ -91,13 +91,12 @@ behavior.
 | `PRE_TOOL_CALL` | Before a tool runs | `ToolCallHookContext` |
 | `POST_TOOL_CALL` | After a tool runs | tool result |
 
-### Step & agent points
+### Step 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_STEP` / `POST_STEP` carry `ctx.kind` (`"task"` or `"flow_method"`) and
 `ctx.step_name`.
@@ -106,8 +105,6 @@ behavior.
 
 | 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 |

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

@@ -1054,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)
@@ -1433,22 +1418,6 @@ class Agent(BaseAgent):
                 if sanitize_tool_name(mt.name) not in existing_names
             )
 
-        from crewai.hooks.contexts import ToolSelectionContext
-        from crewai.hooks.dispatch import InterceptionPoint, dispatch
-
-        # Same policy seam as create_agent_executor: standalone kickoff must
-        # also let tool_selection hooks see (and filter) the final tool list,
-        # or the point can't be trusted to gate what an agent can use.
-        selection_ctx = ToolSelectionContext(
-            agent=self,
-            agent_role=getattr(self, "role", None),
-            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)
 
         agent_info = {

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:

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

@@ -71,29 +71,6 @@ class StepContext(InterceptionContext):
     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 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."""

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

@@ -56,14 +56,11 @@ class InterceptionPoint(str, Enum):
     PRE_TOOL_CALL = "pre_tool_call"
     POST_TOOL_CALL = "post_tool_call"
 
-    # Step & agent points
+    # Step points
     PRE_STEP = "pre_step"
     POST_STEP = "post_step"
-    TOOL_SELECTION = "tool_selection"
 
     # Subsystem points
-    MEMORY_WRITE = "memory_write"
-    MEMORY_READ = "memory_read"
     KNOWLEDGE_RETRIEVAL = "knowledge_retrieval"
     PRE_CODE_EXECUTION = "pre_code_execution"
     MCP_CONNECT = "mcp_connect"

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/tests/hooks/test_interception_conformance.py

@@ -117,60 +117,3 @@ class TestFlowStepPoints:
             return None
 
         assert _SimpleFlow().kickoff() == "rewritten"
-
-
-class TestToolSelection:
-    """tool_selection must gate every path that hands tools to an executor."""
-
-    @staticmethod
-    def _make_agent():
-        from crewai import Agent
-
-        return Agent(
-            role="Selector",
-            goal="g",
-            backstory="b",
-            llm="gpt-4o-mini",
-        )
-
-    @staticmethod
-    def _make_tools():
-        from crewai.tools import tool
-
-        @tool("allowed_tool")
-        def allowed_tool() -> str:
-            """A tool the policy permits."""
-            return "ok"
-
-        @tool("blocked_tool")
-        def blocked_tool() -> str:
-            """A tool the policy strips."""
-            return "nope"
-
-        return [allowed_tool, blocked_tool]
-
-    def test_hook_filters_tools_on_executor_creation(self):
-        @on(InterceptionPoint.TOOL_SELECTION)
-        def drop_blocked(ctx):
-            return [t for t in ctx.payload if t.name != "blocked_tool"]
-
-        agent = self._make_agent()
-        agent.create_agent_executor(tools=self._make_tools())
-
-        names = [t.name for t in agent.agent_executor.original_tools]
-        assert "allowed_tool" in names
-        assert "blocked_tool" not in names
-
-    def test_hook_filters_tools_on_standalone_kickoff_prepare(self):
-        @on(InterceptionPoint.TOOL_SELECTION)
-        def drop_blocked(ctx):
-            return [t for t in ctx.payload if t.name != "blocked_tool"]
-
-        agent = self._make_agent()
-        agent.tools = self._make_tools()
-        _executor, _inputs, agent_info, parsed_tools = agent._prepare_kickoff("hi")
-
-        names = [t.name for t in agent_info["tools"]]
-        assert "allowed_tool" in names
-        assert "blocked_tool" not in names
-        assert all(t.name != "blocked_tool" for t in parsed_tools)