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.
This commit is contained in:
Lucas Gomide
2026-07-13 15:15:56 -03:00
parent 0471ad0233
commit 8fa8dc9571
6 changed files with 209 additions and 4 deletions

View File

@@ -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]],

View File

@@ -278,6 +278,9 @@ def prepare_kickoff(
reset_emission_counter()
reset_last_event_id()
from crewai.hooks.contexts import ExecutionStartContext, InputContext
from crewai.hooks.dispatch import InterceptionPoint, dispatch
normalized: dict[str, Any] | None = None
if inputs is not None:
if not isinstance(inputs, Mapping):
@@ -286,11 +289,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

View File

@@ -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

View File

@@ -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

View File

@@ -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"

View File

@@ -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"