feat: add AgentFuse tool guardrail plugin

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

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"

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)

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"