diff --git a/lib/crewai/src/crewai/hooks/__init__.py b/lib/crewai/src/crewai/hooks/__init__.py index 5b9fbf7bd6..483d429889 100644 --- a/lib/crewai/src/crewai/hooks/__init__.py +++ b/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", diff --git a/lib/crewai/src/crewai/hooks/tool_hooks.py b/lib/crewai/src/crewai/hooks/tool_hooks.py index a860c01d45..9126675cdb 100644 --- a/lib/crewai/src/crewai/hooks/tool_hooks.py +++ b/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 diff --git a/lib/crewai/tests/hooks/test_decorators.py b/lib/crewai/tests/hooks/test_decorators.py index 1d60cf21da..111ee1ef90 100644 --- a/lib/crewai/tests/hooks/test_decorators.py +++ b/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 diff --git a/lib/crewai/tests/hooks/test_guardrail_provider.py b/lib/crewai/tests/hooks/test_guardrail_provider.py new file mode 100644 index 0000000000..b7736e3a5d --- /dev/null +++ b/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() == []