refactor: drop knowledge, code-execution and mcp interception points

Removes `knowledge_retrieval`, `pre_code_execution`, and `mcp_connect`
from the catalog — no concrete use case justifies them yet, and the mcp
seam had an observability gap (an abort fires before the connection
events, leaving no native record). The catalog now adds only `pre_step`
and `post_step` on top of the model/tool and execution-boundary points.

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 knowledge lookup, down to the final output.
-All points share one contract and one registration API.
+every model call, tool call, and task or flow-method step, 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
@@ -101,14 +101,6 @@ behavior.
 `PRE_STEP` / `POST_STEP` carry `ctx.kind` (`"task"` or `"flow_method"`) and
 `ctx.step_name`.
 
-### Subsystem points
-
-| Point | When | `payload` |
-|-------|------|-----------|
-| `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 |
-
 ## Aborting an operation
 
 `HookAborted` carries a `reason` and an optional `source`. The `source` defaults

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

@@ -69,26 +69,3 @@ class StepContext(InterceptionContext):
     kind: str | None = None
     step_name: str | None = None
     output: Any = 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

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

@@ -60,11 +60,6 @@ class InterceptionPoint(str, Enum):
     PRE_STEP = "pre_step"
     POST_STEP = "post_step"
 
-    # Subsystem points
-    KNOWLEDGE_RETRIEVAL = "knowledge_retrieval"
-    PRE_CODE_EXECUTION = "pre_code_execution"
-    MCP_CONNECT = "mcp_connect"
-
 
 class HookAborted(Exception):  # noqa: N818 - public contract name from OSS-86
     """Raised by a hook (or a legacy adapter) to abort the intercepted operation.

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/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,