DEV Community

Mavericksantander
Mavericksantander

Posted on

Add Agent Safety to Any LangChain Tool in Two Lines

#ai #agents #llm #python

You have a LangChain agent with tool access. It can run shell commands, call APIs, modify files. It works great in development.

Then you give it production credentials and it does something you didn't expect.

The fix is two lines.

The Problem With Tool Access Today

When you define a LangChain tool, there's nothing between the model's decision and the execution:

from langchain.tools import tool

@tool
def run_bash(command: str) -> str:
    """Execute a bash command and return the output."""
    import subprocess
    return subprocess.check_output(command, shell=True).decode()
Enter fullscreen mode Exit fullscreen mode

The model decides to call run_bash. It runs. No questions asked.

If the model decides rm -rf /tmp/important_data is the right move, that's what happens. No log. No gate. No way to know it happened until something is broken.

The Fix: @safe_tool

Canopy 0.2.1 ships a decorator that wraps any function with a policy check before execution:

from langchain.tools import tool
from canopy import safe_tool

@tool
@safe_tool(
    action_type="execute_shell",
    agent_ctx={"env": "production", "role": "deploy_bot"}
)
def run_bash(command: str) -> str:
    """Execute a bash command and return the output."""
    import subprocess
    return subprocess.check_output(command, shell=True).decode()
Enter fullscreen mode Exit fullscreen mode

That's it. Two lines added, zero changes to the agent.

LangChain sees a normal tool. Canopy intercepts every call before execution and makes a policy decision: ALLOW, DENY, or REQUIRE_APPROVAL.

What Happens on Each Decision

ALLOW — function executes normally, decision written to audit log.

DENY — raises PermissionError with the reason and an avid (audit ID). LangChain catches it as a tool error and the agent can decide what to do next.

REQUIRE_APPROVAL — by default also raises PermissionError. You can override this with a callback.

def notify_slack_and_wait(canopy_result, func, *args, **kwargs):
    # Send a Slack message, wait for approval, then execute
    approved = send_slack_approval_request(
        reason=canopy_result["reason"],
        avid=canopy_result["avid"],
        command=kwargs.get("command"),
    )
    if approved:
        return func(*args, **kwargs)
    return "Action was not approved by operator."

@tool
@safe_tool(
    action_type="execute_shell",
    agent_ctx={"env": "production"},
    on_require_approval="callback",
    approval_callback=notify_slack_and_wait,
)
def run_bash(command: str) -> str:
    """Execute a bash command."""
    import subprocess
    return subprocess.check_output(command, shell=True).decode()
Enter fullscreen mode Exit fullscreen mode

Now your agent pauses on dangerous commands and waits for a human. Not because the model decided to — because the runtime enforces it regardless of what the model decided.

Dynamic Context at Runtime

In real multi-agent systems, the context changes. Different agents have different roles. The same tool gets called by a deploy bot in one flow and a research agent in another.

@safe_tool accepts agent_ctx as a callable:

from contextvars import ContextVar
from canopy import safe_tool

current_agent_ctx: ContextVar[dict] = ContextVar("agent_ctx", default={})

@tool
@safe_tool(
    action_type="execute_shell",
    agent_ctx=lambda: current_agent_ctx.get()
)
def run_bash(command: str) -> str:
    """Execute a bash command."""
    import subprocess
    return subprocess.check_output(command, shell=True).decode()
Enter fullscreen mode Exit fullscreen mode

Before running each agent, set the context:

token = current_agent_ctx.set({
    "env": "production",
    "role": "research_agent"
})
# run agent...
current_agent_ctx.reset(token)
Enter fullscreen mode Exit fullscreen mode

The decorator evaluates agent_ctx at call time, not at definition time. Same tool, different policy behavior depending on which agent is calling it.

The Policy Behind It

@safe_tool uses the same YAML policy engine as authorize_action. The default policy for execute_shell already covers the most dangerous patterns out of the box:

execute_shell:
  rules:
    - decision: DENY
      when_any:
        - 'payload contains "rm -rf"'
        - 'payload contains "mkfs"'
        - 'payload contains "dd if="'
        - 'payload contains "> /dev/sd"'
      reason: "Destructive shell pattern blocked"

    - decision: REQUIRE_APPROVAL
      when_any:
        - 'payload contains "pip install"'
        - 'payload contains "curl"'
        - 'payload contains "wget"'
        - 'payload contains "npm install"'
      reason: "Network/install command requires approval"
Enter fullscreen mode Exit fullscreen mode

You can override with your own policy file:

CANOPY_POLICY_FILE=/path/to/my-policy.yaml python my_agent.py
Enter fullscreen mode Exit fullscreen mode

The Audit Log

Every @safe_tool call writes to the hash-chain audit log, including the function name, serialized arguments, decision, reason, and avid. The chain is tamper-evident — each entry is cryptographically linked to the previous one.

After your agent runs:

canopy-verify audit.log
# exit 0: chain valid
# exit 1: tampered or broken
Enter fullscreen mode Exit fullscreen mode

You get a complete, verifiable record of every action your agent attempted and what Canopy decided.

Install 

pip install --upgrade canopy-runtime
Enter fullscreen mode Exit fullscreen mode 
from canopy import safe_tool, authorize_action
Enter fullscreen mode Exit fullscreen mode

Full changelog: CHANGELOG.md

GitHub: https://github.com/Mavericksantander/Canopy

If you're stacking @safe_tool with CrewAI or AutoGen instead of LangChain, the pattern is the same — wrap the function before it gets registered as a tool. Drop your setup in the comments if you run into anything unexpected.

Top comments (0)