feat(tools): add ToolSafetyFilter for filter chain integration

Author: DNKYr

tests/tools/safety/test_filter.py

@@ -0,0 +1,118 @@
+# Tencent is pleased to support the open source community by making tRPC-Agent-Python available.
+#
+# Copyright (C) 2026 Tencent. All rights reserved.
+#
+# tRPC-Agent-Python is licensed under Apache-2.0.
+"""Tests for ToolSafetyFilter integration."""
+
+from unittest.mock import AsyncMock
+from unittest.mock import MagicMock
+from unittest.mock import patch
+
+import pytest
+
+from trpc_agent_sdk.abc._filter import FilterResult
+from trpc_agent_sdk.tools.safety._filter import ToolSafetyFilter
+from trpc_agent_sdk.tools.safety._types import Decision
+from trpc_agent_sdk.tools.safety._types import RiskLevel
+from trpc_agent_sdk.tools.safety._types import RiskType
+from trpc_agent_sdk.tools.safety._types import RuleFinding
+from trpc_agent_sdk.tools.safety._types import ScanReport
+
+
+@pytest.fixture
+def mock_scanner():
+    scanner = MagicMock()
+    scanner.scan = AsyncMock()
+    scanner.hash_script = MagicMock(return_value="abc123")
+    return scanner
+
+
+@pytest.fixture
+def mock_audit_logger():
+    logger = MagicMock()
+    return logger
+
+
+@pytest.fixture
+def filter_instance(mock_scanner, mock_audit_logger):
+    return ToolSafetyFilter(scanner=mock_scanner, audit_logger=mock_audit_logger)
+
+
+class TestToolSafetyFilter:
+    async def test_filter_blocks_on_deny(self, filter_instance, mock_scanner):
+        mock_scanner.scan.return_value = ScanReport(
+            decision=Decision.DENY,
+            risk_level=RiskLevel.CRITICAL,
+            findings=[
+                RuleFinding(
+                    rule_id="DANGEROUS_DELETE_001",
+                    risk_type=RiskType.DANGEROUS_FILE_OP,
+                    risk_level=RiskLevel.CRITICAL,
+                    evidence="rm -rf /",
+                    message="dangerous",
+                    recommendation="stop",
+                ),
+            ],
+            scan_duration_ms=10.0,
+        )
+
+        ctx = MagicMock()
+        req = MagicMock()
+        req.args = {"command": "rm -rf /"}
+        req.tool_name = "bash_tool"
+        rsp = FilterResult()
+
+        await filter_instance._before(ctx, req, rsp)
+
+        assert rsp.is_continue is False
+        assert rsp.error is not None
+
+    async def test_filter_passes_on_allow(self, filter_instance, mock_scanner):
+        mock_scanner.scan.return_value = ScanReport(
+            decision=Decision.ALLOW,
+            findings=[],
+            scan_duration_ms=1.0,
+        )
+
+        ctx = MagicMock()
+        req = MagicMock()
+        req.args = {"command": "ls -la"}
+        req.tool_name = "bash_tool"
+        rsp = FilterResult()
+
+        await filter_instance._before(ctx, req, rsp)
+
+        assert rsp.is_continue is True
+        assert rsp.error is None
+
+    async def test_filter_with_no_script_content_passes(self, filter_instance, mock_scanner):
+        mock_scanner.scan = AsyncMock()
+
+        ctx = MagicMock()
+        req = MagicMock()
+        req.args = {}
+        req.tool_name = "todo_tool"
+        rsp = FilterResult()
+
+        await filter_instance._before(ctx, req, rsp)
+
+        assert rsp.is_continue is True
+        mock_scanner.scan.assert_not_called()
+
+    async def test_audit_logger_called_on_scan(self, filter_instance, mock_scanner, mock_audit_logger):
+        mock_scanner.scan.return_value = ScanReport(
+            decision=Decision.ALLOW,
+            findings=[],
+            scan_duration_ms=1.0,
+        )
+
+        ctx = MagicMock()
+        req = MagicMock()
+        req.args = {"script": "print('hello')"}
+        req.tool_name = "python_tool"
+        rsp = FilterResult()
+
+        await filter_instance._before(ctx, req, rsp)
+
+        mock_audit_logger.log.assert_called_once()

trpc_agent_sdk/tools/safety/_filter.py

@@ -0,0 +1,92 @@
+# Tencent is pleased to support the open source community by making tRPC-Agent-Python available.
+#
+# Copyright (C) 2026 Tencent. All rights reserved.
+#
+# tRPC-Agent-Python is licensed under Apache-2.0.
+"""Tool safety filter for the TRPC Agent filter pipeline.
+
+Integrates the ToolSafetyScanner as a BaseFilter that runs in _before()
+to inspect tool arguments and block dangerous scripts before execution.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from typing import Optional
+
+from trpc_agent_sdk.abc import FilterType
+from trpc_agent_sdk.filter import BaseFilter
+from trpc_agent_sdk.filter import FilterResult
+from trpc_agent_sdk.context import AgentContext
+
+from ._audit import SafetyAuditLogger
+from ._scanner import ToolSafetyScanner
+from ._telemetry import set_safety_span_attrs
+from ._types import Decision
+
+
+class ToolSafetyFilter(BaseFilter):
+    """A BaseFilter that scans tool script arguments before execution.
+
+    Plugs into the existing tool filter chain via FilterRunner._run_filters().
+    Blocks execution if the safety scanner returns DENY. Writes audit events
+    and sets OpenTelemetry span attributes on every scan.
+    """
+
+    def __init__(
+        self,
+        *,
+        scanner: ToolSafetyScanner,
+        audit_logger: Optional[SafetyAuditLogger] = None,
+    ):
+        super().__init__()
+        self._type = FilterType.TOOL
+        self._name = "tool_safety"
+        self._scanner = scanner
+        self._audit_logger = audit_logger or SafetyAuditLogger()
+
+    async def _before(self, ctx: AgentContext, req: Any, rsp: FilterResult):
+        script = self._extract_script(req)
+        if not script:
+            return
+
+        tool_name = getattr(req, "tool_name", "unknown")
+        args = getattr(req, "args", None)
+        env_vars = getattr(req, "env_vars", None)
+
+        report = await self._scanner.scan(
+            script=script,
+            tool_name=tool_name,
+            args=args,
+            env_vars=env_vars,
+        )
+
+        set_safety_span_attrs(report)
+
+        script_hash = self._scanner.hash_script(script)
+        self._audit_logger.log(report, tool_name=tool_name, script_hash=script_hash)
+
+        if report.decision == Decision.DENY:
+            findings_text = "; ".join(
+                f"{f.rule_id}: {f.message}" for f in report.findings
+            )
+            rsp.error = Exception(
+                f"Tool execution blocked by safety guard: {findings_text}"
+            )
+            rsp.is_continue = False
+
+    @staticmethod
+    def _extract_script(req: Any) -> Optional[str]:
+        args = getattr(req, "args", None)
+        if not args or not isinstance(args, dict):
+            return None
+
+        for key in ("command", "script", "code", "text", "content"):
+            if key in args and isinstance(args[key], str) and args[key].strip():
+                return args[key]
+
+        str_values = [v for v in args.values() if isinstance(v, str) and len(v) > 10]
+        if str_values:
+            return "\n".join(str_values)
+
+        return None