andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-07-22 07:15:10 +00:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

feat: wire execution-boundary interception points (#6517)

Browse Source 
* 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.

* fix: correct execution-boundary hook ordering and input aliasing

Reworks the crew and flow boundary seams flagged in review. `OUTPUT` and
`EXECUTION_END` now run before the completion event (`CrewKickoffCompletedEvent`
and `FlowFinishedEvent`) so a `HookAborted` no longer leaves a spurious
completed signal and a returned payload replacement is honored on the emitted
and returned result. Boundary contexts alias `inputs` to the same object as
`payload` instead of a fresh dict from `or`, so in-place edits survive
read-back. Flows re-publish the resolved inputs into `flow_inputs` baggage
after the `INPUT` hook so trigger-payload injection observes hook rewrites, and
a resumed flow now dispatches `OUTPUT`/`EXECUTION_END` on its completion path.

* chore: drop redundant seam comments from execution-boundary wiring

Removes two inline comments narrating the OUTPUT/EXECUTION_END dispatch
ordering in `crew.py` and the flow runtime, plus a stray sentence about
enterprise adapters in the conformance-suite docstring. Comment-only
cleanup, no behavior change.

* fix: keep crew output typed across boundary hook dispatch

`_create_crew_output` reassigned `crew_output` from the hook contexts'
`payload`, which is typed `Any`, so mypy flagged `no-any-return` at the
function's return. Cast the payload back to `CrewOutput` after each
dispatch and split the `ExecutionEndContext` construction to satisfy
`ruff format`'s line-length limit.
This commit is contained in:
Lucas Gomide
2026-07-14 11:43:18 -03:00
committed by GitHub
parent 7d21283630
commit a194f3867a
6 changed files with 262 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
lib/crewai/src/crewai/crew.py
View File

@@ -1906,6 +1906,28 @@ 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_ctx = OutputContext(crew=self, output=crew_output, payload=crew_output)
dispatch(InterceptionPoint.OUTPUT, output_ctx)
crew_output = cast(CrewOutput, output_ctx.payload)
end_ctx = ExecutionEndContext(
crew=self, output=crew_output, payload=crew_output
)
dispatch(InterceptionPoint.EXECUTION_END, end_ctx)
crew_output = cast(CrewOutput, end_ctx.payload)
# Ensure background memory saves finish (and emit their
# completed/failed events) before the kickoff-completed event below
# triggers listener teardown/finalization.
@@ -1924,13 +1946,7 @@ 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(
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,
)
return crew_output
def _process_async_tasks(
self,

22
lib/crewai/src/crewai/crews/utils.py
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,30 @@ 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

64
lib/crewai/src/crewai/flow/runtime/__init__.py
View File

@@ -1476,6 +1476,19 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
else (resumed_method_output if emit else result)
)
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(
*[
@@ -2037,6 +2050,9 @@ 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(
@@ -2062,6 +2078,37 @@ 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
@@ -2297,6 +2344,21 @@ 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]
@@ -2370,6 +2432,8 @@ 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)

58
lib/crewai/src/crewai/hooks/contexts.py Normal file
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

10
lib/crewai/src/crewai/hooks/dispatch.py
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"

88
lib/crewai/tests/hooks/test_interception_conformance.py Normal file
View File

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