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 remaining interception points and document the catalog

Browse Source 
Wires the step, agent, subsystem, and flow points onto the dispatcher:
`pre_step`/`post_step`, `tool_selection`, `pre_delegation`, `retry_attempt`,
`memory_write`/`memory_read`, `knowledge_retrieval`, `pre_code_execution`,
`mcp_connect`, `flow_transition`, and `router_decision`, extending the typed-
context module with their contexts. Each seam passes a typed context whose
`payload` a hook may observe, mutate, or replace. Adds the conformance suite
for these points and a new `interception-hooks` doc page with the full
point/payload catalog and the interceptor contract.
This commit is contained in:
Lucas Gomide
2026-07-11 14:14:39 -03:00
parent 73bdfaad56
commit fc906c8233
15 changed files with 664 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
lib/crewai/src/crewai/agent/core.py
View File

@@ -656,6 +656,22 @@ class Agent(BaseAgent):
return result
def _dispatch_retry_attempt(self, e: Exception, task: Task) -> None:
"""Fire the ``retry_attempt`` interception point before re-executing a task."""
from crewai.hooks.contexts import RetryAttemptContext
from crewai.hooks.dispatch import InterceptionPoint, dispatch
retry_ctx = RetryAttemptContext(
agent=self,
agent_role=getattr(self, "role", None),
task=task,
attempt=self._times_executed,
max_attempts=self.max_retry_limit,
error=e,
payload=e,
)
dispatch(InterceptionPoint.RETRY_ATTEMPT, retry_ctx)
def _check_execution_error(self, e: Exception, task: Task) -> None:
"""Check if an execution error should be re-raised immediately.
@@ -709,6 +725,7 @@ class Agent(BaseAgent):
Result from retried execution.
"""
self._check_execution_error(e, task)
self._dispatch_retry_attempt(e, task)
return self.execute_task(task, context, tools)
async def _handle_execution_error_async(
@@ -730,6 +747,7 @@ class Agent(BaseAgent):
Result from retried execution.
"""
self._check_execution_error(e, task)
self._dispatch_retry_attempt(e, task)
return await self.aexecute_task(task, context, tools)
def message(self, content: str, **kwargs: Any) -> str:
@@ -1054,6 +1072,21 @@ 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)

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

@@ -1687,7 +1687,20 @@ class Crew(FlowTrackable, BaseModel):
if files_needing_tool:
tools = self._add_file_tools(tools, files_needing_tool)
return tools
from crewai.hooks.contexts import ToolSelectionContext
from crewai.hooks.dispatch import InterceptionPoint, dispatch
selection_ctx = ToolSelectionContext(
agent=agent,
agent_role=getattr(agent, "role", None),
task=task,
crew=self,
tools=tools,
payload=tools,
)
dispatch(InterceptionPoint.TOOL_SELECTION, selection_ctx)
return selection_ctx.payload
def _get_agent_to_use(self, task: Task) -> BaseAgent | None:
if self.process == Process.hierarchical:

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

@@ -2629,6 +2629,17 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
if future:
self._event_futures.append(future)
from crewai.hooks.contexts import StepContext
from crewai.hooks.dispatch import InterceptionPoint, dispatch
pre_step_ctx = StepContext(
kind="flow_method",
step_name=str(method_name),
flow=self,
payload=dumped_params,
)
dispatch(InterceptionPoint.PRE_STEP, pre_step_ctx)
# Set method name in context so ask() can read it without
# stack inspection. Must happen before copy_context() so the
# value propagates into the thread pool for sync methods.
@@ -2656,6 +2667,16 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
method_name, method_definition.human_feedback, result
)
post_step_ctx = StepContext(
kind="flow_method",
step_name=str(method_name),
flow=self,
output=result,
payload=result,
)
dispatch(InterceptionPoint.POST_STEP, post_step_ctx)
result = post_step_ctx.payload
self._method_outputs.append({"method": str(method_name), "output": result})
# For @human_feedback methods with emit, the result is the collapsed outcome
@@ -2852,6 +2873,19 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
if isinstance(router_result, enum.Enum)
else router_result
)
from crewai.hooks.contexts import RouterDecisionContext
from crewai.hooks.dispatch import InterceptionPoint, dispatch
router_ctx = RouterDecisionContext(
flow=self,
router_name=str(router_name),
route=router_result,
payload=router_result,
)
dispatch(InterceptionPoint.ROUTER_DECISION, router_ctx)
router_result = router_ctx.payload
router_result_str = str(router_result)
router_result_event = FlowMethodName(router_result_str)
router_results.append(router_result_event)
@@ -2880,6 +2914,19 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta):
current_trigger, router_only=False
)
if listeners_triggered:
from crewai.hooks.contexts import FlowTransitionContext
from crewai.hooks.dispatch import InterceptionPoint, dispatch
transition_ctx = FlowTransitionContext(
flow=self,
from_method=str(trigger_method),
to_methods=[str(name) for name in listeners_triggered],
trigger=str(current_trigger),
payload=listeners_triggered,
)
dispatch(InterceptionPoint.FLOW_TRANSITION, transition_ctx)
listeners_triggered = transition_ctx.payload
listener_result = router_result_payloads.get(
str(current_trigger), result
)

12
lib/crewai/src/crewai/flow/runtime/_actions.py
View File

@@ -224,6 +224,18 @@ 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)
return self.handler(
state=self.flow.state,
outputs=outputs_by_name(

108
lib/crewai/src/crewai/hooks/contexts.py
View File

@@ -56,3 +56,111 @@ class ExecutionEndContext(InterceptionContext):
"""``execution_end``: a crew or flow has finished. ``payload`` = the output object."""
output: Any = None
@dataclass
class StepContext(InterceptionContext):
"""``pre_step`` / ``post_step``: a task or flow-method step boundary.
``kind`` is ``"task"`` for crew tasks and ``"flow_method"`` for flow methods.
``payload`` is the step input (pre) or step output (post).
"""
kind: str | None = None
step_name: str | None = None
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 PreDelegationContext(InterceptionContext):
"""``pre_delegation``: an agent is about to delegate work. ``payload`` = delegation input."""
coworker: str | None = None
delegate_to: Any = None
@dataclass
class RetryAttemptContext(InterceptionContext):
"""``retry_attempt``: an operation is about to be retried."""
attempt: int = 0
max_attempts: int | None = None
error: Any = None
@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."""
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
@dataclass
class FileAccessContext(InterceptionContext):
"""``file_access``: reserved. No live consumer seam yet."""
path: str | None = None
mode: str | None = None
@dataclass
class ArtifactOutputContext(InterceptionContext):
"""``artifact_output``: reserved. No live consumer seam yet."""
artifact: Any = None
@dataclass
class FlowTransitionContext(InterceptionContext):
"""``flow_transition``: a flow is moving to triggered methods."""
from_method: str | None = None
to_methods: list[str] = field(default_factory=list)
trigger: str | None = None
@dataclass
class RouterDecisionContext(InterceptionContext):
"""``router_decision``: a flow router is choosing a route. ``payload`` = route label."""
router_name: str | None = None
route: Any = None

14
lib/crewai/src/crewai/knowledge/knowledge.py
View File

@@ -145,6 +145,13 @@ 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,
@@ -183,6 +190,13 @@ 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,

10
lib/crewai/src/crewai/mcp/client.py
View File

@@ -152,6 +152,16 @@ 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)
started_at = datetime.now()
crewai_event_bus.emit(
self,

35
lib/crewai/src/crewai/memory/unified_memory.py
View File

@@ -466,6 +466,18 @@ 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
@@ -561,6 +573,18 @@ 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
@@ -712,6 +736,17 @@ 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

94
lib/crewai/src/crewai/task.py
View File

@@ -662,6 +662,21 @@ class Task(BaseModel):
crewai_event_bus.emit(
self, TaskStartedEvent(context=context, task=self)
)
from crewai.hooks.contexts import StepContext
from crewai.hooks.dispatch import InterceptionPoint, dispatch
pre_step_ctx = StepContext(
kind="task",
step_name=self.name or self.description,
agent=agent,
agent_role=getattr(agent, "role", None),
task=self,
payload=context,
)
dispatch(InterceptionPoint.PRE_STEP, pre_step_ctx)
context = pre_step_ctx.payload
result = await agent.aexecute_task(
task=self,
context=context,
@@ -718,6 +733,18 @@ class Task(BaseModel):
guardrail=self._guardrail,
)
post_step_ctx = StepContext(
kind="task",
step_name=self.name or self.description,
agent=agent,
agent_role=getattr(agent, "role", None),
task=self,
output=task_output,
payload=task_output,
)
dispatch(InterceptionPoint.POST_STEP, post_step_ctx)
task_output = post_step_ctx.payload
self.output = task_output
self.end_time = datetime.datetime.now()
@@ -787,6 +814,21 @@ class Task(BaseModel):
crewai_event_bus.emit(
self, TaskStartedEvent(context=context, task=self)
)
from crewai.hooks.contexts import StepContext
from crewai.hooks.dispatch import InterceptionPoint, dispatch
pre_step_ctx = StepContext(
kind="task",
step_name=self.name or self.description,
agent=agent,
agent_role=getattr(agent, "role", None),
task=self,
payload=context,
)
dispatch(InterceptionPoint.PRE_STEP, pre_step_ctx)
context = pre_step_ctx.payload
result = agent.execute_task(
task=self,
context=context,
@@ -843,6 +885,18 @@ class Task(BaseModel):
guardrail=self._guardrail,
)
post_step_ctx = StepContext(
kind="task",
step_name=self.name or self.description,
agent=agent,
agent_role=getattr(agent, "role", None),
task=self,
output=task_output,
payload=task_output,
)
dispatch(InterceptionPoint.POST_STEP, post_step_ctx)
task_output = post_step_ctx.payload
self.output = task_output
self.end_time = datetime.datetime.now()
@@ -884,6 +938,32 @@ class Task(BaseModel):
clear_task_files(self.id)
reset_current_task_id(task_id_token)
def _dispatch_guardrail_retry_attempt(
self,
agent: BaseAgent | None,
context: str | None,
attempt: int,
error: Any,
) -> str | None:
"""Fire ``retry_attempt`` before re-executing a task after a guardrail failure.
Returns the (possibly hook-modified) retry context.
"""
from crewai.hooks.contexts import RetryAttemptContext
from crewai.hooks.dispatch import InterceptionPoint, dispatch
retry_ctx = RetryAttemptContext(
agent=agent,
agent_role=getattr(agent, "role", None),
task=self,
attempt=attempt,
max_attempts=self.guardrail_max_retries,
error=error,
payload=context,
)
dispatch(InterceptionPoint.RETRY_ATTEMPT, retry_ctx)
return retry_ctx.payload
def _post_agent_execution(self, agent: BaseAgent) -> None:
pass
@@ -1317,6 +1397,13 @@ Follow these guidelines:
color="yellow",
)
context = self._dispatch_guardrail_retry_attempt(
agent=agent,
context=context,
attempt=current_retry_count,
error=guardrail_result.error,
)
result = agent.execute_task(
task=self,
context=context,
@@ -1427,6 +1514,13 @@ Follow these guidelines:
color="yellow",
)
context = self._dispatch_guardrail_retry_attempt(
agent=agent,
context=context,
attempt=current_retry_count,
error=guardrail_result.error,
)
result = await agent.aexecute_task(
task=self,
context=context,

13
lib/crewai/src/crewai/tools/agent_tools/base_agent_tools.py
View File

@@ -109,6 +109,19 @@ class BaseAgentTool(BaseTool):
selected_agent = agent[0]
try:
from crewai.hooks.contexts import PreDelegationContext
from crewai.hooks.dispatch import InterceptionPoint, dispatch
delegation_ctx = PreDelegationContext(
agent=selected_agent,
agent_role=getattr(selected_agent, "role", None),
coworker=sanitized_name,
delegate_to=selected_agent,
payload=task,
)
dispatch(InterceptionPoint.PRE_DELEGATION, delegation_ctx)
task = delegation_ctx.payload
task_with_assigned_agent = Task(
description=task,
agent=selected_agent,

16
lib/crewai/src/crewai/tools/tool_usage.py
View File

@@ -879,6 +879,22 @@ class ToolUsage:
return ToolUsageError(
f"{I18N_DEFAULT.errors('tool_usage_error').format(error=e)}\nMoving on then. {I18N_DEFAULT.slice('format').format(tool_names=self.tools_names)}"
)
from crewai.hooks.contexts import RetryAttemptContext
from crewai.hooks.dispatch import InterceptionPoint, dispatch
retry_ctx = RetryAttemptContext(
agent=self.agent,
agent_role=getattr(self.agent, "role", None),
task=self.task,
attempt=self._run_attempts,
max_attempts=self._max_parsing_attempts,
error=e,
payload=tool_string,
)
dispatch(InterceptionPoint.RETRY_ATTEMPT, retry_ctx)
tool_string = retry_ctx.payload
return self._tool_calling(tool_string)
def _validate_tool_input(self, tool_input: str | None) -> dict[str, Any]:

91
lib/crewai/tests/hooks/test_interception_conformance.py
View File

@@ -8,7 +8,7 @@ against these guarantees.
from __future__ import annotations
from crewai.flow.flow import Flow, listen, start
from crewai.flow.flow import Flow, listen, router, start
from crewai.hooks.dispatch import (
HookAborted,
InterceptionPoint,
@@ -87,3 +87,92 @@ class TestFlowExecutionBoundaries:
with pytest.raises(HookAborted) as exc:
_SimpleFlow().kickoff()
assert exc.value.reason == "not allowed"
class TestFlowStepPoints:
"""pre_step / post_step for flow methods (kind=flow_method)."""
def test_pre_and_post_step_fire_per_method(self):
kinds: list[tuple[str, str | None]] = []
@on(InterceptionPoint.PRE_STEP)
def pre(ctx):
kinds.append(("pre", ctx.step_name))
@on(InterceptionPoint.POST_STEP)
def post(ctx):
kinds.append(("post", ctx.step_name))
_SimpleFlow().kickoff()
assert ("pre", "begin") in kinds
assert ("post", "begin") in kinds
assert ("pre", "finish") in kinds
assert ("post", "finish") in kinds
def test_post_step_can_rewrite_method_output(self):
@on(InterceptionPoint.POST_STEP)
def rewrite(ctx):
if ctx.step_name == "finish":
return "rewritten"
return None
assert _SimpleFlow().kickoff() == "rewritten"
class _RouterFlow(Flow):
@start()
def begin(self):
return "begin"
@router(begin)
def route(self):
return "go_left"
@listen("go_left")
def left(self):
return "left"
@listen("go_right")
def right(self):
return "right"
class TestFlowTransitionAndRouter:
"""flow_transition and router_decision on a routed flow."""
def test_transition_payload_carries_from_and_to(self):
seen: list[tuple[str | None, list[str]]] = []
@on(InterceptionPoint.FLOW_TRANSITION)
def capture(ctx):
seen.append((ctx.from_method, list(ctx.to_methods)))
_RouterFlow().kickoff()
assert any(to == ["left"] for _from, to in seen)
def test_router_decision_fires_with_route(self):
routes: list[object] = []
@on(InterceptionPoint.ROUTER_DECISION)
def capture(ctx):
routes.append(ctx.route)
_RouterFlow().kickoff()
assert "go_left" in routes
def test_router_decision_can_reroute(self):
@on(InterceptionPoint.ROUTER_DECISION)
def reroute(ctx):
return "go_right"
landed: list[str] = []
@on(InterceptionPoint.PRE_STEP)
def track(ctx):
landed.append(ctx.step_name)
_RouterFlow().kickoff()
assert "right" in landed
assert "left" not in landed