Feat/tool script safety guard

examples/tool_safety_guard/README.md

@@ -0,0 +1,94 @@
+# Tool Script Safety Guard
+
+This example shows how to scan Python and Bash tool scripts before execution. It uses a policy file, emits structured reports, appends JSONL audit events, and can be attached as a Tool Filter or CodeExecutor wrapper.
+
+## Quickstart Project
+
+For a project-shaped walkthrough similar to the optimization quickstart examples, start here:
+
+```bash
+python examples/tool_safety_guard/quickstart/run_agent.py
+```
+
+That directory follows the same shape as other quickstart examples: `run_agent.py` is the entry point, `agent/` contains the application logic and dry-run tool service, `policy.yaml` holds the safety knobs, and `scripts/` contains sample tool payloads. It demonstrates the same guard through direct scan, Tool Filter, and CodeExecutor wrapper paths without executing untrusted script bodies.
+
+## Run The Samples
+
+```bash
+python scripts/tool_safety_check.py examples/tool_safety_guard/samples/04_network_external.py \
+  --policy examples/tool_safety_guard/tool_safety_policy.yaml \
+  --report-out examples/tool_safety_guard/out/report.json \
+  --audit-out examples/tool_safety_guard/out/audit.jsonl
+```
+
+To scan every sample:
+
+```bash
+for f in examples/tool_safety_guard/samples/*; do
+  lang=python
+  case "$f" in *.sh) lang=bash ;; esac
+  python scripts/tool_safety_check.py "$f" --language "$lang" \
+    --policy examples/tool_safety_guard/tool_safety_policy.yaml || true
+done
+```
+
+## Policy
+
+`tool_safety_policy.yaml` controls:
+
+- `allowed_domains`: domains that network rules may allow without code changes.
+- `allowed_commands`: commands permitted in command argument scans.
+- `denied_paths`: sensitive paths such as `.env`, `~/.ssh`, cloud credentials, and private keys.
+- `system_write_paths`: protected filesystem roots.
+- `max_timeout_seconds`, `max_output_bytes`, `max_sleep_seconds`, `max_loop_iterations`, and write limits.
+- `deny_risk_level`, `review_risk_level`, and `block_on_review`.
+
+Changing the YAML is enough to alter domain allowlists, denied paths, and allowed commands.
+
+## Decisions And Reports
+
+The scanner returns:
+
+- `allow`: no configured rule found a material risk.
+- `needs_human_review`: medium risk or uncertain behavior, for example dynamic network targets.
+- `deny`: high or critical risk such as credential reads, recursive deletion, non-allowlisted network egress, dependency installation, or secret leakage.
+
+Every finding includes `risk_type`, `rule_id`, `evidence`, `recommendation`, line information when available, and whether evidence was redacted. Reports also include OpenTelemetry-ready attributes such as `tool.safety.decision`, `tool.safety.risk_level`, and `tool.safety.rule_id`.
+
+## Filter Integration
+
+Attach the registered Tool Filter to script-like tools:
+
+```python
+from trpc_agent_sdk.tools import FunctionTool
+from trpc_agent_sdk.tools.safety import ToolSafetyFilter  # registers "tool_safety_guard"
+
+tool = FunctionTool(run_script, filters_name=["tool_safety_guard"])
+```
+
+The filter scans common argument shapes: `command`, `script`, `code`, `source`, and `code_blocks`. For `deny`, and for `needs_human_review` when `block_on_review=true`, it returns a structured `TOOL_SAFETY_GUARD_BLOCKED` response before the tool body runs.
+
+## CodeExecutor Wrapper
+
+```python
+from trpc_agent_sdk.code_executors import UnsafeLocalCodeExecutor
+from trpc_agent_sdk.tools.safety import SafetyGuardedCodeExecutor
+from trpc_agent_sdk.tools.safety import ToolSafetyGuard
+from trpc_agent_sdk.tools.safety import ToolSafetyPolicy
+
+policy = ToolSafetyPolicy.load("examples/tool_safety_guard/tool_safety_policy.yaml")
+executor = SafetyGuardedCodeExecutor(
+    delegate=UnsafeLocalCodeExecutor(timeout=10),
+    guard=ToolSafetyGuard(policy=policy),
+)
+```
+
+## Relationship To Sandbox, Filter, Telemetry, And CodeExecutor
+
+This guard is a pre-execution control. It is designed to run in a Filter or wrapper before a Tool, Skill, MCP Tool, or CodeExecutor starts real execution. It complements sandbox isolation by catching obvious high-risk scripts early and by creating audit records. It also emits fields that can be copied into OpenTelemetry spans.
+
+It cannot replace a sandbox. Static scanning has false positives, false negatives, and bypass risks: obfuscated code, generated strings, encoded payloads, indirect imports, shell expansions, downloaded second-stage scripts, and interpreter-specific behavior can evade static rules. Production deployments should still use least-privilege credentials, network egress controls, filesystem mounts, process and memory limits, timeouts, container or remote sandbox isolation, and post-execution auditing.
+
+## Extending Rules
+
+Add new rules in `trpc_agent_sdk.tools.safety._scanner`. Rules should return `ToolSafetyFinding` with a stable `rule_id`, clear `risk_type`, bounded evidence, and a recommendation. Prefer AST checks for Python and explicit token checks for Bash, then use regex only as a fallback for patterns that are hard to parse.

examples/tool_safety_guard/quickstart/README.md

@@ -0,0 +1,84 @@
+# Tool Safety Guard Quickstart
+
+This quickstart is a small project-shaped example for the tool script safety guard. It mirrors the structure used by other quickstart examples: `run_agent.py` is the entry point, `agent/` contains the reusable application modules, `policy.yaml` contains policy knobs, and `scripts/` contains representative tool payloads.
+
+## Directory
+
+```text
+examples/tool_safety_guard/quickstart/
+|-- run_agent.py
+|-- policy.yaml
+|-- design.md
+|-- agent/
+|   |-- __init__.py
+|   |-- agent.py
+|   |-- config.py
+|   `-- tools.py
+`-- scripts/
+    |-- safe_report.py
+    |-- external_upload.py
+    |-- read_secret.py
+    |-- review_subprocess.py
+    `-- dangerous_cleanup.sh
+```
+
+`scripts/` is not meant to be executed directly. The runner feeds each file into the guard and uses dry-run delegates to show where a real tool or CodeExecutor would be blocked.
+
+## Run
+
+From the repository root:
+
+```bash
+python examples/tool_safety_guard/quickstart/run_agent.py
+```
+
+Or with the Windows launcher used in this workspace:
+
+```bash
+py -3.14 examples/tool_safety_guard/quickstart/run_agent.py
+```
+
+The run writes:
+
+```text
+examples/tool_safety_guard/quickstart/out/
+|-- quickstart_report.json
+`-- audit.jsonl
+```
+
+Expected decisions:
+
+| case | expected decision | why |
+| --- | --- | --- |
+| `safe_report` | `allow` | Local bounded file write under the project output directory. |
+| `external_upload` | `deny` | Network egress to a domain not in `allowed_domains`. |
+| `read_secret` | `deny` | Reads `.env`, which is configured as a denied path. |
+| `review_subprocess` | `needs_human_review` | Starts a subprocess, which is review-worthy but not always unsafe. |
+| `dangerous_cleanup` | `deny` | Recursive delete pattern. |
+
+## What This Demonstrates
+
+The same `ToolSafetyGuard` is used in three places from `agent/agent.py`:
+
+- Direct scan: `guard.scan(ToolSafetyScanRequest(...))` returns a structured report.
+- Tool Filter: `ToolSafetyFilter` blocks script-like tool arguments before the tool body runs.
+- CodeExecutor wrapper: `SafetyGuardedCodeExecutor` scans code blocks before delegating to an executor.
+
+This is the intended production shape: keep execution code simple, put static pre-execution policy in one reusable guard, and write audit events for every decision.
+
+## Policy Knobs
+
+`policy.yaml` controls the main behavior without code changes:
+
+- `allowed_domains`: network egress allowlist.
+- `allowed_commands`: command allowlist for shell-like command scans.
+- `denied_paths`: sensitive path patterns.
+- `system_write_paths`: protected roots.
+- resource limits such as `max_sleep_seconds`, `max_loop_iterations`, and `max_parallel_tasks`.
+- `deny_risk_level`, `review_risk_level`, and `block_on_review`.
+
+Try adding `evil.example.net` to `allowed_domains`; `external_upload.py` will no longer be denied for non-allowlisted egress, although it still records network-related findings.
+
+## Relationship To The Sample Matrix
+
+The parent `examples/tool_safety_guard/samples/` directory remains the acceptance matrix with 12 focused cases. This quickstart is the narrative example: it shows how the scanner, filter, executor wrapper, report, and audit event fit together in a minimal application.

examples/tool_safety_guard/quickstart/agent/__init__.py

@@ -0,0 +1,10 @@
+# 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 guard quickstart agent package."""
+
+from .agent import run_quickstart
+
+__all__ = ["run_quickstart"]

examples/tool_safety_guard/quickstart/agent/agent.py

@@ -0,0 +1,145 @@
+# 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 guard quickstart application logic.
+
+This module is shaped like other quickstart ``agent/agent.py`` files, but it is
+model-free on purpose: the issue being demonstrated is pre-execution script
+safety for tools and code executors, so the sample can run without API keys.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Iterable
+from pathlib import Path
+from typing import Any
+
+from trpc_agent_sdk.code_executors import CodeBlock
+from trpc_agent_sdk.code_executors import CodeExecutionInput
+from trpc_agent_sdk.context import create_agent_context
+from trpc_agent_sdk.filter import FilterResult
+from trpc_agent_sdk.tools.safety import SafetyGuardedCodeExecutor
+from trpc_agent_sdk.tools.safety import ToolSafetyFilter
+from trpc_agent_sdk.tools.safety import ToolSafetyGuard
+from trpc_agent_sdk.tools.safety import ToolSafetyPolicy
+from trpc_agent_sdk.tools.safety import ToolSafetyScanRequest
+
+from .config import DEFAULT_CASES
+from .config import DEFAULT_OUT
+from .config import DEFAULT_POLICY
+from .config import QUICKSTART_DIR
+from .config import ScriptSafetyCase
+from .tools import DryRunScriptExecutor
+from .tools import dry_run_tool
+from .tools import read_script
+
+
+def _enum_value(value: Any) -> str:
+    return value.value if hasattr(value, "value") else str(value)
+
+
+def _write_json(path: Path, payload: Any) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+
+
+async def _run_filter_case(guard: ToolSafetyGuard, script: str, language: str) -> dict[str, Any]:
+    filter_ = ToolSafetyFilter(guard=guard)
+
+    async def handle() -> FilterResult:
+        return FilterResult(rsp=await dry_run_tool(script, language=language))
+
+    result = await filter_.run(
+        create_agent_context(),
+        {
+            "script": script,
+            "language": language,
+            "cwd": str(QUICKSTART_DIR),
+        },
+        handle,
+    )
+    return {
+        "continued": bool(result.is_continue),
+        "response": result.rsp,
+    }
+
+
+async def _run_executor_case(guard: ToolSafetyGuard, script: str, language: str) -> dict[str, Any]:
+    executor = SafetyGuardedCodeExecutor(
+        delegate=DryRunScriptExecutor(work_dir=str(QUICKSTART_DIR)),
+        guard=guard,
+    )
+    result = await executor.execute_code(
+        invocation_context=None,  # type: ignore[arg-type]
+        code_execution_input=CodeExecutionInput(code_blocks=[CodeBlock(language=language, code=script)]),
+    )
+    return {
+        "outcome": _enum_value(result.outcome),
+        "output": result.output,
+    }
+
+
+async def run_quickstart(
+    *,
+    policy_path: Path = DEFAULT_POLICY,
+    output_dir: Path = DEFAULT_OUT,
+    cases: Iterable[ScriptSafetyCase] = DEFAULT_CASES,
+) -> dict[str, Any]:
+    """Run the model-free tool safety quickstart and write a report."""
+
+    policy = ToolSafetyPolicy.load(policy_path)
+    audit_path = output_dir / "audit.jsonl"
+    report_path = output_dir / "quickstart_report.json"
+    guard = ToolSafetyGuard(policy=policy, audit_log_path=audit_path)
+    case_results = []
+
+    for case in cases:
+        script = read_script(case.script_path)
+        report = guard.scan(
+            ToolSafetyScanRequest(
+                script=script,
+                language=case.language,
+                cwd=str(QUICKSTART_DIR),
+                tool_metadata={
+                    "name": case.name,
+                    "script_path": str(case.script_path),
+                },
+            )
+        )
+        filter_result = await _run_filter_case(guard, script, case.language)
+        executor_result = await _run_executor_case(guard, script, case.language)
+        case_results.append(
+            {
+                "name": case.name,
+                "script": str(case.script_path.relative_to(QUICKSTART_DIR)),
+                "language": case.language,
+                "decision": _enum_value(report.decision),
+                "risk_level": _enum_value(report.risk_level),
+                "blocked": report.blocked,
+                "rule_ids": [finding.rule_id for finding in report.findings],
+                "summary": report.summary,
+                "filter": filter_result,
+                "code_executor": executor_result,
+            }
+        )
+
+    summary = {
+        "policy": {
+            "name": policy.name,
+            "version": policy.version,
+            "path": str(policy_path),
+        },
+        "case_count": len(case_results),
+        "decision_counts": {
+            decision: sum(1 for case in case_results if case["decision"] == decision)
+            for decision in sorted({case["decision"] for case in case_results})
+        },
+        "cases": case_results,
+        "report": str(report_path),
+        "audit_log": str(audit_path),
+    }
+    _write_json(report_path, summary)
+    return summary

examples/tool_safety_guard/quickstart/agent/config.py

@@ -0,0 +1,35 @@
+# 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.
+"""Configuration for the tool safety guard quickstart."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+
+
+QUICKSTART_DIR = Path(__file__).resolve().parents[1]
+DEFAULT_POLICY = QUICKSTART_DIR / "policy.yaml"
+DEFAULT_OUT = QUICKSTART_DIR / "out"
+SCRIPTS_DIR = QUICKSTART_DIR / "scripts"
+
+
+@dataclass(frozen=True)
+class ScriptSafetyCase:
+    """A script-like payload that the quickstart feeds into the guard."""
+
+    name: str
+    script_path: Path
+    language: str
+
+
+DEFAULT_CASES = (
+    ScriptSafetyCase("safe_report", SCRIPTS_DIR / "safe_report.py", "python"),
+    ScriptSafetyCase("external_upload", SCRIPTS_DIR / "external_upload.py", "python"),
+    ScriptSafetyCase("read_secret", SCRIPTS_DIR / "read_secret.py", "python"),
+    ScriptSafetyCase("review_subprocess", SCRIPTS_DIR / "review_subprocess.py", "python"),
+    ScriptSafetyCase("dangerous_cleanup", SCRIPTS_DIR / "dangerous_cleanup.sh", "bash"),
+)