From 8fa8dc9571d6e3cdf9f24f6069391827d8bf95fd Mon Sep 17 00:00:00 2001 From: Lucas Gomide Date: Mon, 13 Jul 2026 15:15:56 -0300 Subject: [PATCH] feat: wire execution-boundary interception points Adds the typed interception contexts (`crewai/hooks/contexts.py`) and wires the `execution_start`, `input`, `output`, and `execution_end` points for both crews and flows through the dispatcher. `prepare_kickoff` and `Flow.kickoff_async` fire `execution_start`/`input` so a hook can rewrite resolved inputs before the run, while `Crew._create_crew_output` and the flow tail fire `output`/`execution_end` so the final result can be observed or replaced. Closes the eight critical-path points without touching the legacy hooks. --- lib/crewai/src/crewai/crew.py | 14 ++- lib/crewai/src/crewai/crews/utils.py | 13 +++ .../src/crewai/flow/runtime/__init__.py | 29 ++++++ lib/crewai/src/crewai/hooks/contexts.py | 58 ++++++++++++ lib/crewai/src/crewai/hooks/dispatch.py | 10 ++- .../hooks/test_interception_conformance.py | 89 +++++++++++++++++++ 6 files changed, 209 insertions(+), 4 deletions(-) create mode 100644 lib/crewai/src/crewai/hooks/contexts.py create mode 100644 lib/crewai/tests/hooks/test_interception_conformance.py diff --git a/lib/crewai/src/crewai/crew.py b/lib/crewai/src/crewai/crew.py index 696d831c8..2179458c8 100644 --- a/lib/crewai/src/crewai/crew.py +++ b/lib/crewai/src/crewai/crew.py @@ -1924,7 +1924,10 @@ class Crew(FlowTrackable, BaseModel): # Finalization is handled by trace listener (always initialized) # The batch manager checks contextvar to determine if tracing is enabled - return CrewOutput( + 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, @@ -1932,6 +1935,15 @@ class Crew(FlowTrackable, BaseModel): token_usage=self.token_usage, ) + 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) + + return crew_output + def _process_async_tasks( self, futures: list[tuple[Task, Future[TaskOutput], int]], diff --git a/lib/crewai/src/crewai/crews/utils.py b/lib/crewai/src/crewai/crews/utils.py index 6faa0e8c2..c096a8fd2 100644 --- a/lib/crewai/src/crewai/crews/utils.py +++ b/lib/crewai/src/crewai/crews/utils.py @@ -278,6 +278,9 @@ def prepare_kickoff( reset_emission_counter() reset_last_event_id() + from crewai.hooks.contexts import ExecutionStartContext, InputContext + from crewai.hooks.dispatch import InterceptionPoint, dispatch + normalized: dict[str, Any] | None = None if inputs is not None: if not isinstance(inputs, Mapping): @@ -286,11 +289,21 @@ def prepare_kickoff( ) normalized = dict(inputs) + start_ctx = ExecutionStartContext( + crew=crew, inputs=normalized or {}, 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 or {}, 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 diff --git a/lib/crewai/src/crewai/flow/runtime/__init__.py b/lib/crewai/src/crewai/flow/runtime/__init__.py index a6c73e697..201ec4c04 100644 --- a/lib/crewai/src/crewai/flow/runtime/__init__.py +++ b/lib/crewai/src/crewai/flow/runtime/__init__.py @@ -2062,6 +2062,24 @@ 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 + + start_ctx = ExecutionStartContext( + flow=self, inputs=inputs or {}, payload=inputs + ) + dispatch(InterceptionPoint.EXECUTION_START, start_ctx) + inputs = start_ctx.payload + + input_ctx = InputContext(flow=self, inputs=inputs or {}, payload=inputs) + dispatch(InterceptionPoint.INPUT, input_ctx) + inputs = input_ctx.payload + # Reset flow state for fresh execution unless restoring from persistence is_restoring = ( inputs and "id" in inputs and self.persistence is not None @@ -2297,6 +2315,12 @@ 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 + if self._event_futures: await asyncio.gather( *[asyncio.wrap_future(f) for f in self._event_futures] @@ -2347,6 +2371,11 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): else: trace_listener.batch_manager.finalize_batch() + end_ctx = ExecutionEndContext( + flow=self, output=final_output, payload=final_output + ) + dispatch(InterceptionPoint.EXECUTION_END, end_ctx) + return final_output finally: # Safety net for the exception path; the success path already diff --git a/lib/crewai/src/crewai/hooks/contexts.py b/lib/crewai/src/crewai/hooks/contexts.py new file mode 100644 index 000000000..d718903b1 --- /dev/null +++ b/lib/crewai/src/crewai/hooks/contexts.py @@ -0,0 +1,58 @@ +"""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 diff --git a/lib/crewai/src/crewai/hooks/dispatch.py b/lib/crewai/src/crewai/hooks/dispatch.py index 2822c73dd..095484220 100644 --- a/lib/crewai/src/crewai/hooks/dispatch.py +++ b/lib/crewai/src/crewai/hooks/dispatch.py @@ -41,11 +41,15 @@ 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. This layer ships the - model- and tool-call boundaries, which back the legacy - ``before/after_llm_call`` and ``before/after_tool_call`` hooks. + 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" diff --git a/lib/crewai/tests/hooks/test_interception_conformance.py b/lib/crewai/tests/hooks/test_interception_conformance.py new file mode 100644 index 000000000..3593699d5 --- /dev/null +++ b/lib/crewai/tests/hooks/test_interception_conformance.py @@ -0,0 +1,89 @@ +"""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, 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"