safe-exec

Safe tool execution for LLM agents.
Prevents catastrophic mistakes by detecting when a session is "tired" and automatically gating dangerous tools.

pip install safe-exec

Zero dependencies. Python 3.10+. Works with any LLM framework.

---

The problem

You gave your LLM agent access to run_bash. It works great — until session 3, when context is bloated, the model has already made two errors, and it decides to rm -rf the wrong directory. You had no warning. There was no checkpoint.

safe-exec puts a gate between the LLM and every tool it can call. The gate uses three observable signals — context size, session duration, and recent error rate — to compute a fatigue score. When fatigue is high, dangerous tools require confirmation. When fatigue is critical, they're denied outright.

Every decision is written to an append-only witness.log.

---

Quick start

fromsafe_execimportSafeExecutorfrompathlibimportPathexecutor=SafeExecutor(workspace=Path("/tmp/my_workspace"))
@executor.register("read_file", risk="low")defread_file(path: str) ->str:
returnopen(path).read()
@executor.register("write_file", risk="medium")defwrite_file(path: str, content: str) ->str:
open(path, "w").write(content)
returnf"Written {len(content)} chars"@executor.register("run_bash", risk="high")defrun_bash(cmd: str) ->str:
importsubprocessreturnsubprocess.check_output(cmd, shell=True, text=True)
# Execute — the gate decides automaticallyresult=executor.execute("run_bash", cmd="ls -la")
ifresult.ok:
print(result.output)
elifresult.skipped:
print(f"Skipped: {result.decision.value}") # ask or deny

---

Built-in tools

A standard set of file and shell tools, all workspace-sandboxed:

fromsafe_execimportSafeExecutorfromsafe_exec.built_in_toolsimportregister_defaultsfrompathlibimportPathexecutor=SafeExecutor(workspace=Path("/tmp/workspace"))
register_defaults(executor)
# Now you have: read_file, write_file, append_file, list_dir,# delete_file, copy_file, run_bash, run_python, delete_dir

All file operations are restricted to the workspace. Path traversal attempts raise ValueError.

---

Gate policy

The default policy table:

Risk	Fatigue < 0.35	0.35 ≤ Fatigue < 0.70	Fatigue ≥ 0.70
low	ALLOW	ALLOW	ASK
medium	ALLOW	ASK	DENY
high	ASK	ASK	DENY

High-risk tools always require confirmation by default (always_ask_high=True).

Fatigue score

Three signals, fully tuneable:

fromsafe_execimportSystemMetricsmetrics=SystemMetrics(
max_context=4000, # tokens → score 1.0max_duration=1800.0, # seconds → score 1.0max_errors=5, # error count → score 1.0weight_context=0.40,
weight_duration=0.30,
weight_errors=0.30,
)

Update metrics after each LLM turn:

executor.update_metrics(context_tokens=1800) # after each responseexecutor.update_metrics(context_tokens=2400, error_occurred=True) # on errorexecutor.reset_metrics() # after context flush

---

Custom gate policy

fromsafe_execimportSafeExecutor, GateDecision, RiskLevel, SystemMetricsfromsafe_exec.gateimportGatePolicyclassStrictPolicy(GatePolicy):
defdecide(self, risk: RiskLevel, metrics: SystemMetrics) ->GateDecision:
# High risk always denied, everything else askifrisk==RiskLevel.HIGH:
returnGateDecision.DENYifmetrics.fatigue_score() >0.2:
returnGateDecision.ASKreturnGateDecision.ALLOWexecutor=SafeExecutor(policy=StrictPolicy())

---

Custom confirmation handler

# Non-interactive: always deny ASK (useful in CI)executor=SafeExecutor(confirm_fn=lambdatool, rationale: False)
# Slack webhook, Telegram bot, etc.defnotify_and_wait(tool_name: str, rationale: str) ->bool:
send_slack_message(f"Agent wants to run `{tool_name}`: {rationale}")
returnwait_for_approval(tool_name)
executor=SafeExecutor(confirm_fn=notify_and_wait)

---

Witness log

Every gate decision is recorded:

{"ts":"2026-04-21T10:23:01Z","event":"gate","tool":"run_bash","decision":"ask","details":{"rationale":"risk=high fatigue=0.41 ..."}}
{"ts":"2026-04-21T10:23:04Z","event":"execute","tool":"run_bash","decision":"allow","details":{...}}
{"ts":"2026-04-21T10:31:12Z","event":"gate","tool":"delete_dir","decision":"deny","details":{"rationale":"risk=high fatigue=0.78 ..."}}

Query it with standard tools:

# All denials
grep '"decision":"deny"' witness.log | jq .# Stats
python3 -c "from safe_exec import SafeExecutor; e=SafeExecutor(); print(e.log_stats())"

---

Integration examples

OpenAI tool calling

importopenai, jsonfromsafe_execimportSafeExecutorfromsafe_exec.built_in_toolsimportregister_defaultsexecutor=SafeExecutor()
register_defaults(executor)
defhandle_tool_call(tool_name: str, args: dict):
result=executor.execute(tool_name, **args)
ifresult.ok:
returnstr(result.output)
returnf"Tool {result.decision.value}: {result.erroror'gated'}"# Use handle_tool_call as your tool_call dispatcher in the response loop.

Ollama (local LLMs)

fromsafe_execimportSafeExecutorfromsafe_exec.built_in_toolsimportregister_defaultsimporturllib.request, jsonexecutor=SafeExecutor()
register_defaults(executor)
# Update fatigue after each Ollama responsedefollama_turn(prompt: str, context_tokens: int) ->str:
executor.update_metrics(context_tokens=context_tokens)
# ... call Ollama, parse tool calls, dispatch through executor ...

---

Status

Feature	Status
Gate policy (ALLOW/ASK/DENY)	✅
Fatigue scoring	✅
Witness log	✅
Built-in tools (file + shell)	✅
Path traversal protection	✅
Custom policy	✅
Custom confirm handler	✅
Async support	🔧 planned
Token counter integration	🔧 planned

---

License

MIT — use it, modify it, ship it.