diff --git a/README.md b/README.md index 6e878ae..5de59de 100644 --- a/README.md +++ b/README.md @@ -120,6 +120,33 @@ result = manager.run_code("print('hello')", language="python") | `praisonai.plugins` | Lifecycle hooks, guardrails, policies | `simple_logger`, `pii_guardrail` | | `praisonai.sandbox` | Optional sandbox backends | `capsule` | +## AgentFuse Tool Guardrail + +The optional AgentFuse middleware evaluates a tool-call policy through +PraisonAI's public `wrap_tool_call` boundary before the protected handler starts. + +```bash +pip install -e ".[agentfuse]" +``` + +```python +from dhms_agentfuse import RuntimeGuard +from praisonaiagents import Agent +from praisonai_plugins.guardrails.agentfuse import AgentFuseToolMiddleware + +guard = RuntimeGuard(deny_tools={"protected_write"}) +agent = Agent( + name="guarded-agent", + instructions="Use tools when needed.", + hooks=[AgentFuseToolMiddleware(guard)], +) +``` + +An allowed call continues through PraisonAI's existing handler chain. A blocked +call, missing tool-call identity, or unexpected guard-evaluation exception returns +a host-native `ToolResponse` with `outcome=not_executed`; it does not claim that a +handler failed after execution began. + --- ## The PraisonAI Protocol Philosophy diff --git a/pyproject.toml b/pyproject.toml index d385e9f..1bf3391 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,7 +17,12 @@ dependencies = [ [project.optional-dependencies] capsule = ["capsule>=0.1.0"] -dev = ["pytest>=8.0.0", "pytest-asyncio>=0.23.0"] +agentfuse = ["dhms-agentfuse==3.7.3"] +dev = [ + "pytest>=8.0.0", + "pytest-asyncio>=0.23.0", + "dhms-agentfuse==3.7.3", +] [project.entry-points."praisonai.plugins"] simple_logger = "praisonai_plugins.hooks.simple_logger:SimpleLoggerPlugin" diff --git a/src/praisonai_plugins/guardrails/agentfuse.py b/src/praisonai_plugins/guardrails/agentfuse.py new file mode 100644 index 0000000..ed48ab7 --- /dev/null +++ b/src/praisonai_plugins/guardrails/agentfuse.py @@ -0,0 +1,107 @@ +"""Optional AgentFuse middleware for pre-dispatch tool decisions.""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import Any + +from praisonaiagents.hooks import ToolRequest, ToolResponse, wrap_tool_call + + +class AgentFuseToolMiddleware: + """Apply an AgentFuse decision at PraisonAI's public tool boundary.""" + + def __init__(self, guard: Any) -> None: + try: + from dhms_agentfuse import RuntimeGuard, ToolCallRequest + except ImportError as exc: + raise ImportError( + "AgentFuse middleware requires the 'agentfuse' optional dependency: " + "pip install 'praisonai-plugins[agentfuse]'" + ) from exc + + if not isinstance(guard, RuntimeGuard): + raise TypeError("guard must be a dhms_agentfuse.RuntimeGuard") + + self._guard = guard + self._request_type = ToolCallRequest + wrap_tool_call(self) + + def _not_executed_response( + self, + request: ToolRequest, + *, + tool_call_id: str | None, + reason_code: str, + policy_denied: bool, + guard_failed: bool, + decision: Any | None = None, + ) -> ToolResponse: + result = { + "status": "blocked", + "policy_denied": policy_denied, + "guard_failed": guard_failed, + "tool_failure": False, + "reason_code": reason_code, + "tool_call_id": tool_call_id, + "host_execution": { + "outcome": "not_executed", + "handler_started": False, + }, + } + if decision is not None: + result["agentfuse_decision"] = decision.to_safe_dict() + + return ToolResponse( + tool_name=request.tool_name, + result=result, + context=request.context, + ) + + def __call__( + self, + request: ToolRequest, + call_next: Callable[[ToolRequest], ToolResponse], + ) -> ToolResponse: + context = request.context + metadata = context.metadata if context is not None else {} + tool_call_id = metadata.get("tool_call_id") + if not tool_call_id: + return self._not_executed_response( + request, + tool_call_id=None, + reason_code="missing_tool_call_id", + policy_denied=False, + guard_failed=False, + ) + + try: + decision = self._guard.evaluate( + self._request_type( + tool_call_id=tool_call_id, + tool_name=request.tool_name, + arguments=request.arguments, + safe_metadata={"integration": "praisonai-plugins"}, + ) + ) + except Exception: # noqa: BLE001 - the adapter boundary must fail closed + return self._not_executed_response( + request, + tool_call_id=tool_call_id, + reason_code="guard_evaluation_exception", + policy_denied=False, + guard_failed=True, + ) + + metadata["agentfuse_decision"] = decision + if decision.action == "block": + return self._not_executed_response( + request, + tool_call_id=tool_call_id, + reason_code=decision.reason_code, + policy_denied=True, + guard_failed=False, + decision=decision, + ) + + return call_next(request) diff --git a/tests/test_agentfuse_tool_middleware.py b/tests/test_agentfuse_tool_middleware.py new file mode 100644 index 0000000..d43de75 --- /dev/null +++ b/tests/test_agentfuse_tool_middleware.py @@ -0,0 +1,176 @@ +"""Tests for the optional AgentFuse tool middleware.""" + +from __future__ import annotations + +import pytest +from dhms_agentfuse import RuntimeGuard, RuntimeGuardDecision, ToolCallRequest +from praisonaiagents import Agent +from praisonaiagents.hooks import InvocationContext, ToolRequest + +from praisonai_plugins.guardrails.agentfuse import AgentFuseToolMiddleware + + +class RecordingRuntimeGuard(RuntimeGuard): + """Keep completed decisions inside one test instead of production state.""" + + def __init__(self, *args, **kwargs) -> None: + super().__init__(*args, **kwargs) + self.recorded_decisions: list[RuntimeGuardDecision] = [] + + def evaluate(self, tool_call: ToolCallRequest) -> RuntimeGuardDecision: + decision = super().evaluate(tool_call) + self.recorded_decisions.append(decision) + return decision + + +def _sync_agent(guard: RuntimeGuard): + handler_calls: list[str] = [] + + def protected_write(value: str) -> str: + handler_calls.append(value) + return "write completed" + + middleware = AgentFuseToolMiddleware(guard) + agent = Agent( + name="agentfuse-sync-test", + instructions="Exercise one inert test tool.", + tools=[protected_write], + hooks=[middleware], + approval=True, + ) + return agent, middleware, handler_calls + + +def test_sync_allow_dispatches_once_and_preserves_identity(): + guard = RecordingRuntimeGuard(allow_tools={"protected_write"}) + agent, _, calls = _sync_agent(guard) + + result = agent.execute_tool( + "protected_write", {"value": "synthetic-value"}, "praison-allow-001" + ) + + assert result == "write completed" + assert calls == ["synthetic-value"] + assert guard.recorded_decisions[0].tool_call_id == "praison-allow-001" + + +def test_sync_block_returns_non_execution_without_dispatch(): + guard = RecordingRuntimeGuard(deny_tools={"protected_write"}) + agent, _, calls = _sync_agent(guard) + + result = agent.execute_tool( + "protected_write", {"value": "synthetic-value"}, "praison-block-001" + ) + + assert calls == [] + assert result["status"] == "blocked" + assert result["tool_failure"] is False + assert result["tool_call_id"] == "praison-block-001" + assert result["host_execution"] == { + "outcome": "not_executed", + "handler_started": False, + } + assert guard.recorded_decisions[0].action == "block" + + +def test_runtime_guard_policy_error_fails_closed_without_dispatch(): + def failing_policy(tool_call): + del tool_call + raise RuntimeError("synthetic policy failure") + + guard = RecordingRuntimeGuard(policy=failing_policy) + agent, _, calls = _sync_agent(guard) + + result = agent.execute_tool( + "protected_write", {"value": "synthetic-value"}, "praison-policy-001" + ) + + assert calls == [] + assert result["reason_code"] == "policy_exception" + assert result["host_execution"]["handler_started"] is False + assert guard.recorded_decisions[0].reason_code == "policy_exception" + + +def test_adapter_evaluate_exception_fails_closed_without_dispatch(): + class RaisingGuard(RuntimeGuard): + def evaluate(self, tool_call): + del tool_call + raise RuntimeError("synthetic adapter-boundary failure") + + agent, _, calls = _sync_agent(RaisingGuard()) + + result = agent.execute_tool( + "protected_write", {"value": "synthetic-value"}, "praison-raise-001" + ) + + assert calls == [] + assert result["reason_code"] == "guard_evaluation_exception" + assert result["guard_failed"] is True + assert result["policy_denied"] is False + assert result["tool_call_id"] == "praison-raise-001" + assert result["host_execution"] == { + "outcome": "not_executed", + "handler_started": False, + } + + +@pytest.mark.asyncio +async def test_async_block_uses_same_guard_without_dispatch(): + handler_calls: list[str] = [] + + async def protected_write(value: str) -> str: + handler_calls.append(value) + return "write completed" + + middleware = AgentFuseToolMiddleware(RuntimeGuard(deny_tools={"protected_write"})) + agent = Agent( + name="agentfuse-async-test", + instructions="Exercise one inert test tool.", + tools=[protected_write], + hooks=[middleware], + approval=True, + ) + + result = await agent.execute_tool_async( + "protected_write", {"value": "synthetic-value"}, "praison-async-001" + ) + + assert handler_calls == [] + assert result["tool_call_id"] == "praison-async-001" + assert result["host_execution"]["outcome"] == "not_executed" + + +def test_missing_identity_fails_closed_without_fabricating_id(): + agent, _, calls = _sync_agent(RuntimeGuard(allow_tools={"protected_write"})) + + result = agent.execute_tool("protected_write", {"value": "synthetic-value"}) + + assert calls == [] + assert result["reason_code"] == "missing_tool_call_id" + assert result["tool_call_id"] is None + assert result["host_execution"]["handler_started"] is False + + +def test_handler_failure_after_allow_is_not_converted_to_policy_block(): + middleware = AgentFuseToolMiddleware(RuntimeGuard(allow_tools={"protected_write"})) + request = ToolRequest( + tool_name="protected_write", + arguments={"value": "synthetic-value"}, + context=InvocationContext( + agent_id="agentfuse-handler-failure", + run_id="run-001", + session_id="session-001", + tool_name="protected_write", + metadata={"tool_call_id": "praison-handler-failure-001"}, + ), + ) + + def failing_handler(tool_request): + del tool_request + raise RuntimeError("synthetic handler failure") + + with pytest.raises(RuntimeError, match="synthetic handler failure"): + middleware(request, failing_handler) + + assert request.context is not None + assert request.context.metadata["agentfuse_decision"].action == "allow"