Execution security for Python AI agents — audit, policy enforcement, and DLP in one package. One decorator, zero config.
Works two ways:
@protect— add governance to any existing agent (LangChain, CrewAI, AutoGen, plain Python)Node9Agent— build a governed agent from scratch with tools, DLP, and audit built-in
pip install node9node9 automatically routes to the right backend:
| Environment | Routing |
|---|---|
NODE9_API_KEY set | → node9 SaaS (cloud / CI — no local daemon needed) |
| Local daemon running | → node9-proxy on localhost:7391 |
| Neither | → offline audit log at ~/.node9/audit.log (auto-approve, never blocks) |
No config required — it just works wherever your agent runs.
Drop @protect on any function your agent calls. node9 intercepts the call, logs it, and enforces policy before the function runs.
fromnode9importprotect, ActionDeniedException@protectdefwrite_file(path: str, content: str) ->None:
withopen(path, "w") asf:
f.write(content)
_ALLOWED_COMMANDS= {"pytest", "ruff", "mypy", "black"}
@protect("run_tests")defrun_tests(tool: str) ->str:
# Allowlist-based: only pre-approved CLI tools can be invoked.# Never pass raw LLM strings to subprocess — enumerate safe commands explicitly.iftoolnotin_ALLOWED_COMMANDS:
raiseValueError(f"Tool {tool!r} is not in the allowed list: {_ALLOWED_COMMANDS}")
importsubprocessreturnsubprocess.check_output([tool], text=True)
try:
write_file("/etc/hosts", "bad content")
exceptActionDeniedExceptionase:
print(f"Blocked: {e}")Works with async def out of the box.
fromnode9importconfigureconfigure(agent_name="my-langchain-agent", policy="audit")Or via environment variables:
NODE9_AGENT_NAME=my-langchain-agent
NODE9_AGENT_POLICY=audit| Policy | Behaviour |
|---|---|
audit | Log everything, auto-approve. Never blocks. Good for CI. |
require_approval | Block + notify human. Good for production actions. |
block_on_rules | Auto-block if rules match, audit otherwise. |
| (empty) | SaaS default behaviour. |
fromlangchain.toolsimportBaseToolfromnode9importprotectclassWriteFileTool(BaseTool):
name="write_file"description="Write content to a file."@protect("write_file")def_run(self, path: str, content: str) ->str:
withopen(path, "w") asf:
f.write(content)
returnf"Written to {path}"fromcrewai.toolsimporttoolfromnode9importprotect@tool("write_file")@protect("write_file")defwrite_file(path: str, content: str) ->str:
"""Write content to a file."""withopen(path, "w") asf:
f.write(content)
returnf"Written to {path}"See examples/ for full runnable examples including AutoGen and LangGraph.
Node9Agent is a governance base class — DLP, path safety, audit, and tool dispatch built-in. It does not include an LLM loop; that is your framework's responsibility. This keeps the SDK framework-agnostic with zero dependencies.
importanthropicfromnode9importNode9Agent, tool, internalclassCiAgent(Node9Agent):
agent_name="ci-code-review"policy="audit"_ALLOWED_SUITES= {"pytest", "pytest --tb=short", "ruff check ."}
@tool("run_tests")defrun_tests(self, suite: str) ->str:
"""Run an allowlisted test suite and return output."""importshlex, subprocessifsuitenotinself._ALLOWED_SUITES:
raiseValueError(f"Suite {suite!r} not in allowed list: {self._ALLOWED_SUITES}")
returnsubprocess.check_output(shlex.split(suite), text=True)
@tool("write_code")defwrite_code(self, filename: str, content: str) ->str:
"""Write content to a file in the workspace."""fromnode9importsafe_pathpath=safe_path(filename, workspace=self._workspace) # traversal-safewithopen(path, "w") asf:
f.write(content)
returnf"Written {filename}"@internaldef_git_push(self, branch: str) ->str:
"""Push to remote — infrastructure, not a governed action."""importsubprocesssubprocess.run(["git", "push", "origin", branch], check=True)
returnf"Pushed {branch}"agent=CiAgent(workspace="/path/to/repo")
client=anthropic.Anthropic()
# Get tool specs in the format your LLM expectstools=agent.build_tools_anthropic() # → input_schema format# tools = agent.build_tools_openai() # → {type: function, function: {...}}# tools = agent._build_tools() # → neutral (parameters key)# Your LLM loop — use whichever client you wantmessages= [{"role": "user", "content": "Fix the failing tests in this diff: ..."}]
whileTrue:
response=client.messages.create(model="claude-opus-4-6", tools=tools, messages=messages)
messages.append({"role": "assistant", "content": response.content})
ifresponse.stop_reason!="tool_use":
breakresults= []
forblockinresponse.content:
ifblock.type=="tool_use":
result=agent.dispatch(block.name, block.input) # DLP + audit happen hereresults.append({"type": "tool_result", "tool_use_id": block.id, "content": result})
messages.append({"role": "user", "content": results})See examples/ for complete runnable implementations per framework.
Every @tool-decorated method, before the function runs:
- DLP scan — blocks if
filenameorcontentcontains a secret or sensitive path - Path safety — rejects
../traversal attempts, raisesActionDeniedException - Audit / approval — calls
evaluate()which respects the agent'spolicy - Run ID — injects a UUID grouping all tool calls from one session in the dashboard
@internal is for git operations, workspace setup, and other infrastructure:
- Never calls
evaluate()— no SaaS call, no blocking - Logs locally only:
[node9 internal] _git_push(branch='main')
Node9Agent introspects @tool methods and builds tool specs automatically — parameter names, types from annotations, and descriptions from docstrings. No manual schema writing.
fromnode9importdlp_scan, safe_path# Scan content for secrets before writing to diskhit=dlp_scan("output.txt", content)
ifhit:
raiseValueError(f"Blocked: {hit}")
# Resolve a path safely within a workspace directorypath=safe_path("src/main.py", workspace="/tmp/repo")Patterns detected: AWS keys, GitHub tokens, Slack tokens, OpenAI keys, Stripe keys, PEM private keys, GCP service accounts, NPM auth tokens, Anthropic keys, and sensitive file paths (.ssh, .aws, .env, .kube, etc.).
ActionDeniedException has a negotiation property — feed it back to the LLM so it can try a different approach:
try:
agent.dispatch("delete_file", {"path": "/etc/hosts"})
exceptActionDeniedExceptionase:
# e.negotiation = "Action 'delete_file' was blocked by Node9: policy. Choose a different approach."response=llm.invoke(e.negotiation)| Variable | Default | Description |
|---|---|---|
NODE9_API_KEY | — | Routes to node9 SaaS. Required for cloud / CI. |
NODE9_AGENT_NAME | — | Agent identity — appears in audit logs and dashboard. |
NODE9_AGENT_POLICY | — | audit, require_approval, or block_on_rules. |
NODE9_DAEMON_PORT | 7391 | Local daemon port. |
NODE9_AUTO_START | — | Set to 1 to auto-launch the local daemon if not running. |
NODE9_SKIP | — | Set to 1 to bypass all checks. Never set in production — disables all governance. For unit tests only. If set, a warning is emitted at import time. |
After cloning, activate the git hooks (runs tests before every commit and push):
git config core.hooksPath .githooksRun tests manually:
python3 -m pytest tests/ -p no:anyio -qApache-2.0