From d7cf5cd735c6c87ac7ad52ea4d4769c148620a02 Mon Sep 17 00:00:00 2001 From: MacOS Date: Tue, 18 Aug 2026 18:15:48 +0800 Subject: [PATCH 1/2] feat: add AgentFuse tool guardrail plugin --- README.md | 27 +++ pyproject.toml | 7 +- src/praisonai_plugins/guardrails/agentfuse.py | 113 ++++++++++++ tests/test_agentfuse_tool_middleware.py | 165 ++++++++++++++++++ 4 files changed, 311 insertions(+), 1 deletion(-) create mode 100644 src/praisonai_plugins/guardrails/agentfuse.py create mode 100644 tests/test_agentfuse_tool_middleware.py 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..58eb65f --- /dev/null +++ b/src/praisonai_plugins/guardrails/agentfuse.py @@ -0,0 +1,113 @@ +"""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 + self._decisions: dict[str, Any] = {} + wrap_tool_call(self) + + def decision_for(self, tool_call_id: str) -> Any | None: + """Return the completed policy decision for a tool call, if available.""" + return self._decisions.get(tool_call_id) + + 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 + tool_call_id = ( + context.metadata.get("tool_call_id") if context is not None else None + ) + 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, + ) + + self._decisions[tool_call_id] = 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..a30b9b8 --- /dev/null +++ b/tests/test_agentfuse_tool_middleware.py @@ -0,0 +1,165 @@ +"""Tests for the optional AgentFuse tool middleware.""" + +from __future__ import annotations + +import pytest +from dhms_agentfuse import RuntimeGuard +from praisonaiagents import Agent +from praisonaiagents.hooks import InvocationContext, ToolRequest + +from praisonai_plugins.guardrails.agentfuse import AgentFuseToolMiddleware + + +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(): + agent, middleware, calls = _sync_agent( + RuntimeGuard(allow_tools={"protected_write"}) + ) + + result = agent.execute_tool( + "protected_write", {"value": "synthetic-value"}, "praison-allow-001" + ) + + assert result == "write completed" + assert calls == ["synthetic-value"] + assert middleware.decision_for("praison-allow-001").tool_call_id == ( + "praison-allow-001" + ) + + +def test_sync_block_returns_non_execution_without_dispatch(): + agent, middleware, calls = _sync_agent(RuntimeGuard(deny_tools={"protected_write"})) + + 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 middleware.decision_for("praison-block-001").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") + + agent, middleware, calls = _sync_agent(RuntimeGuard(policy=failing_policy)) + + 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 middleware.decision_for("praison-policy-001").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 middleware.decision_for("praison-handler-failure-001").action == "allow" From 6a33d73f442a711421bed7c8d096dcaf7bfbf0b8 Mon Sep 17 00:00:00 2001 From: MacOS Date: Wed, 19 Aug 2026 00:27:51 +0800 Subject: [PATCH 2/2] fix: avoid retaining AgentFuse decisions across calls --- src/praisonai_plugins/guardrails/agentfuse.py | 12 ++---- tests/test_agentfuse_tool_middleware.py | 39 ++++++++++++------- 2 files changed, 28 insertions(+), 23 deletions(-) diff --git a/src/praisonai_plugins/guardrails/agentfuse.py b/src/praisonai_plugins/guardrails/agentfuse.py index 58eb65f..ed48ab7 100644 --- a/src/praisonai_plugins/guardrails/agentfuse.py +++ b/src/praisonai_plugins/guardrails/agentfuse.py @@ -25,13 +25,8 @@ def __init__(self, guard: Any) -> None: self._guard = guard self._request_type = ToolCallRequest - self._decisions: dict[str, Any] = {} wrap_tool_call(self) - def decision_for(self, tool_call_id: str) -> Any | None: - """Return the completed policy decision for a tool call, if available.""" - return self._decisions.get(tool_call_id) - def _not_executed_response( self, request: ToolRequest, @@ -69,9 +64,8 @@ def __call__( call_next: Callable[[ToolRequest], ToolResponse], ) -> ToolResponse: context = request.context - tool_call_id = ( - context.metadata.get("tool_call_id") if context is not None else None - ) + 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, @@ -99,7 +93,7 @@ def __call__( guard_failed=True, ) - self._decisions[tool_call_id] = decision + metadata["agentfuse_decision"] = decision if decision.action == "block": return self._not_executed_response( request, diff --git a/tests/test_agentfuse_tool_middleware.py b/tests/test_agentfuse_tool_middleware.py index a30b9b8..d43de75 100644 --- a/tests/test_agentfuse_tool_middleware.py +++ b/tests/test_agentfuse_tool_middleware.py @@ -3,13 +3,26 @@ from __future__ import annotations import pytest -from dhms_agentfuse import RuntimeGuard +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] = [] @@ -29,9 +42,8 @@ def protected_write(value: str) -> str: def test_sync_allow_dispatches_once_and_preserves_identity(): - agent, middleware, calls = _sync_agent( - RuntimeGuard(allow_tools={"protected_write"}) - ) + guard = RecordingRuntimeGuard(allow_tools={"protected_write"}) + agent, _, calls = _sync_agent(guard) result = agent.execute_tool( "protected_write", {"value": "synthetic-value"}, "praison-allow-001" @@ -39,13 +51,12 @@ def test_sync_allow_dispatches_once_and_preserves_identity(): assert result == "write completed" assert calls == ["synthetic-value"] - assert middleware.decision_for("praison-allow-001").tool_call_id == ( - "praison-allow-001" - ) + assert guard.recorded_decisions[0].tool_call_id == "praison-allow-001" def test_sync_block_returns_non_execution_without_dispatch(): - agent, middleware, calls = _sync_agent(RuntimeGuard(deny_tools={"protected_write"})) + guard = RecordingRuntimeGuard(deny_tools={"protected_write"}) + agent, _, calls = _sync_agent(guard) result = agent.execute_tool( "protected_write", {"value": "synthetic-value"}, "praison-block-001" @@ -59,7 +70,7 @@ def test_sync_block_returns_non_execution_without_dispatch(): "outcome": "not_executed", "handler_started": False, } - assert middleware.decision_for("praison-block-001").action == "block" + assert guard.recorded_decisions[0].action == "block" def test_runtime_guard_policy_error_fails_closed_without_dispatch(): @@ -67,7 +78,8 @@ def failing_policy(tool_call): del tool_call raise RuntimeError("synthetic policy failure") - agent, middleware, calls = _sync_agent(RuntimeGuard(policy=failing_policy)) + guard = RecordingRuntimeGuard(policy=failing_policy) + agent, _, calls = _sync_agent(guard) result = agent.execute_tool( "protected_write", {"value": "synthetic-value"}, "praison-policy-001" @@ -76,9 +88,7 @@ def failing_policy(tool_call): assert calls == [] assert result["reason_code"] == "policy_exception" assert result["host_execution"]["handler_started"] is False - assert middleware.decision_for("praison-policy-001").reason_code == ( - "policy_exception" - ) + assert guard.recorded_decisions[0].reason_code == "policy_exception" def test_adapter_evaluate_exception_fails_closed_without_dispatch(): @@ -162,4 +172,5 @@ def failing_handler(tool_request): with pytest.raises(RuntimeError, match="synthetic handler failure"): middleware(request, failing_handler) - assert middleware.decision_for("praison-handler-failure-001").action == "allow" + assert request.context is not None + assert request.context.metadata["agentfuse_decision"].action == "allow"