feat: add pre-tool GuardrailProvider adapter

lib/crewai/src/crewai/hooks/__init__.py

@@ -19,10 +19,14 @@
     unregister_before_llm_call_hook,
 )
 from crewai.hooks.tool_hooks import (
+    GuardrailDecision,
+    GuardrailProvider,
+    GuardrailRequest,
     ToolCallHookContext,
     clear_after_tool_call_hooks,
     clear_all_tool_call_hooks,
     clear_before_tool_call_hooks,
+    enable_guardrail,
     get_after_tool_call_hooks,
     get_before_tool_call_hooks,
     register_after_tool_call_hook,
@@ -74,6 +78,9 @@ def clear_all_global_hooks() -> dict[str, tuple[int, int]]:
 
 
 __all__ = [
+    "GuardrailDecision",
+    "GuardrailProvider",
+    "GuardrailRequest",
     "LLMCallHookContext",
     "ToolCallHookContext",
     "after_llm_call",
@@ -87,6 +94,7 @@ def clear_all_global_hooks() -> dict[str, tuple[int, int]]:
     "clear_all_tool_call_hooks",
     "clear_before_llm_call_hooks",
     "clear_before_tool_call_hooks",
+    "enable_guardrail",
     "get_after_llm_call_hooks",
     "get_after_tool_call_hooks",
     "get_before_llm_call_hooks",

lib/crewai/src/crewai/hooks/tool_hooks.py

@@ -1,6 +1,10 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Any
+import logging
+from collections.abc import Mapping
+from copy import deepcopy
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Any, Protocol
 
 from crewai_core.printer import PRINTER
 
@@ -13,6 +17,9 @@
 )
 
 
+logger = logging.getLogger(__name__)
+
+
 if TYPE_CHECKING:
     from crewai.agent import Agent
     from crewai.agents.agent_builder.base_agent import BaseAgent
@@ -316,3 +323,93 @@ def clear_all_tool_call_hooks() -> tuple[int, int]:
     before_count = clear_before_tool_call_hooks()
     after_count = clear_after_tool_call_hooks()
     return (before_count, after_count)
+
+
+@dataclass(frozen=True, slots=True)
+class GuardrailRequest:
+    """Detached context for one provider decision before a tool call."""
+
+    tool_name: str
+    tool_alias: str
+    tool_input: Mapping[str, Any]
+    agent_id: str | None = None
+    agent_role: str | None = None
+
+
+@dataclass(frozen=True, slots=True)
+class GuardrailDecision:
+    """Provider verdict for one proposed tool call."""
+
+    allow: bool
+    reason: str | None = None
+    metadata: Mapping[str, Any] = field(default_factory=dict)
+
+
+class GuardrailProvider(Protocol):
+    """Contract for pluggable pre-tool-call guardrail providers."""
+
+    def evaluate(self, request: GuardrailRequest) -> GuardrailDecision:
+        """Evaluate whether the requested tool call may proceed."""
+        ...
+
+
+def enable_guardrail(
+    provider: GuardrailProvider,
+    *,
+    fail_closed: bool = True,
+) -> BeforeToolCallHookType:
+    """Register a provider through the existing before-tool-call registry.
+
+    Provider failures and invalid results block execution by default. Set
+    ``fail_closed=False`` only when availability is intentionally preferred.
+    The returned hook can be removed with ``unregister_before_tool_call_hook``.
+    """
+
+    def _hook(context: ToolCallHookContext) -> bool | None:
+        try:
+            request = _build_guardrail_request(context)
+            decision = provider.evaluate(request)
+            if not isinstance(decision, GuardrailDecision):
+                raise TypeError(
+                    "GuardrailProvider.evaluate() must return GuardrailDecision"
+                )
+            if not isinstance(decision.allow, bool):
+                raise TypeError("GuardrailDecision.allow must be a bool")
+        except Exception:
+            logger.exception("GuardrailProvider evaluation failed")
+            return False if fail_closed else None
+
+        return None if decision.allow else False
+
+    register_before_tool_call_hook(_hook)
+    return _hook
+
+
+def _build_guardrail_request(context: ToolCallHookContext) -> GuardrailRequest:
+    """Snapshot the mutable tool-call context before provider evaluation."""
+
+    original_tool_name = _optional_text(getattr(context, "tool", None), "name")
+    return GuardrailRequest(
+        tool_name=original_tool_name or context.tool_name,
+        tool_alias=context.tool_name,
+        tool_input=deepcopy(context.tool_input),
+        agent_id=_optional_identifier(context.agent, "id"),
+        agent_role=_optional_text(context.agent, "role"),
+    )
+
+
+def _optional_text(source: Any, attribute: str) -> str | None:
+    if source is None:
+        return None
+    value = getattr(source, attribute, None)
+    return value if isinstance(value, str) and value else None
+
+
+def _optional_identifier(source: Any, attribute: str) -> str | None:
+    if source is None:
+        return None
+    value = getattr(source, attribute, None)
+    if value is None:
+        return None
+    rendered = str(value)
+    return rendered if rendered else None

lib/crewai/tests/hooks/test_decorators.py

@@ -5,10 +5,13 @@
 from unittest.mock import Mock
 
 from crewai.hooks import (
+    GuardrailDecision,
+    GuardrailRequest,
     after_llm_call,
     after_tool_call,
     before_llm_call,
     before_tool_call,
+    enable_guardrail,
     get_after_llm_call_hooks,
     get_after_tool_call_hooks,
     get_before_llm_call_hooks,
@@ -348,3 +351,105 @@ def manual_hook(context):
         hooks = get_before_tool_call_hooks()
 
         assert len(hooks) == 2
+
+
+class RecordingGuardrailProvider:
+    """Small provider fixture that records the latest detached request."""
+
+    def __init__(
+        self,
+        decision: GuardrailDecision | None = None,
+        error: Exception | None = None,
+    ) -> None:
+        self.decision = (
+            decision if decision is not None else GuardrailDecision(allow=True)
+        )
+        self.error = error
+        self.request: GuardrailRequest | None = None
+
+    def evaluate(self, request: GuardrailRequest) -> GuardrailDecision:
+        self.request = request
+        if self.error is not None:
+            raise self.error
+        return self.decision
+
+
+class TestGuardrailProviderAdapter:
+    """Test the provider adapter built on the existing before-tool hook registry."""
+
+    @staticmethod
+    def context() -> ToolCallHookContext:
+        tool = Mock()
+        tool.name = "Send Email"
+        agent = Mock()
+        agent.id = "agent-123"
+        agent.role = "Support Agent"
+        return ToolCallHookContext(
+            tool_name="send_email",
+            tool_input={
+                "recipient": "qa@example.com",
+                "body": "Hello",
+                "headers": {"x-trace": "original"},
+            },
+            tool=tool,
+            agent=agent,
+        )
+
+    def test_registers_provider_and_allows(self):
+        provider = RecordingGuardrailProvider()
+        hook = enable_guardrail(provider)
+
+        assert get_before_tool_call_hooks() == [hook]
+        assert hook(self.context()) is None
+        assert provider.request is not None
+        assert provider.request.tool_name == "Send Email"
+        assert provider.request.tool_alias == "send_email"
+        assert provider.request.agent_id == "agent-123"
+        assert provider.request.agent_role == "Support Agent"
+
+    def test_deny_maps_to_false(self):
+        provider = RecordingGuardrailProvider(
+            GuardrailDecision(allow=False, reason="policy")
+        )
+
+        assert enable_guardrail(provider)(self.context()) is False
+
+    def test_provider_receives_deeply_detached_tool_input(self):
+        context = self.context()
+        provider = RecordingGuardrailProvider()
+
+        assert enable_guardrail(provider)(context) is None
+        assert provider.request is not None
+        provider.request.tool_input["recipient"] = "other@example.com"
+        provider.request.tool_input["headers"]["x-trace"] = "changed"
+
+        assert context.tool_input["recipient"] == "qa@example.com"
+        assert context.tool_input["headers"]["x-trace"] == "original"
+
+    def test_provider_failure_is_fail_closed_by_default(self):
+        provider = RecordingGuardrailProvider(error=RuntimeError("unavailable"))
+
+        assert enable_guardrail(provider)(self.context()) is False
+
+    def test_provider_failure_can_fail_open(self):
+        provider = RecordingGuardrailProvider(error=RuntimeError("unavailable"))
+
+        assert enable_guardrail(provider, fail_closed=False)(self.context()) is None
+
+    @pytest.mark.parametrize("fail_closed, expected", [(True, False), (False, None)])
+    def test_invalid_provider_result_follows_failure_policy(
+        self,
+        fail_closed: bool,
+        expected: bool | None,
+    ):
+        provider = RecordingGuardrailProvider()
+        provider.decision = object()  # type: ignore[assignment]
+
+        assert enable_guardrail(provider, fail_closed=fail_closed)(self.context()) is expected
+
+    def test_non_boolean_allow_is_rejected(self):
+        decision = GuardrailDecision(allow=True)
+        object.__setattr__(decision, "allow", 1)
+        provider = RecordingGuardrailProvider(decision)
+
+        assert enable_guardrail(provider)(self.context()) is False

lib/crewai/tests/hooks/test_guardrail_provider.py

@@ -0,0 +1,44 @@
+"""Focused contract tests for the GuardrailProvider adapter."""
+
+from __future__ import annotations
+
+import pytest
+
+from crewai.hooks import (
+    GuardrailDecision,
+    GuardrailRequest,
+    clear_before_tool_call_hooks,
+    enable_guardrail,
+    get_before_tool_call_hooks,
+    register_before_tool_call_hook,
+    unregister_before_tool_call_hook,
+)
+
+
+class AllowProvider:
+    """Minimal provider fixture that always allows the tool call."""
+
+    def evaluate(self, request: GuardrailRequest) -> GuardrailDecision:
+        return GuardrailDecision(allow=True)
+
+
+@pytest.fixture(autouse=True)
+def restore_before_tool_hooks():
+    """Keep the global hook registry unchanged outside this test module."""
+
+    original_hooks = get_before_tool_call_hooks()
+    clear_before_tool_call_hooks()
+    yield
+    clear_before_tool_call_hooks()
+    for hook in original_hooks:
+        register_before_tool_call_hook(hook)
+
+
+def test_enable_guardrail_returns_unregisterable_hook():
+    """The adapter hook can be removed through the existing registry API."""
+
+    hook = enable_guardrail(AllowProvider())
+
+    assert get_before_tool_call_hooks() == [hook]
+    assert unregister_before_tool_call_hook(hook) is True
+    assert get_before_tool_call_hooks() == []