TL;DR: The framework is 20% of the work. This post shows how to wrap
LangGraph or CrewAI with out-of-process governance that passes
enterprise security audits. Working code included.
I've spent 28 years building production AI systems. The pattern I'm watching repeat in 2026 is familiar: we get excited about capability, underinvest in control, and pay for it later.
This time the stakes are higher. 88% of enterprises running AI agents have already had a security incident. When the filters fail, there's no human reviewing the output—just an autonomous agent executing commands at machine speed.
Here's how to build agents that actually work in production.
The Problem with "Secure" Prompts
Most tutorials show you something like this:
system_prompt = """
You are a helpful assistant.
IMPORTANT: Never delete files or access sensitive data.
Always ask for confirmation before taking destructive actions.
"""
agent = create_agent(llm, tools, system_prompt)
This feels secure. It is not.
Why this fails:
In-process prompts are advisory. The model can be manipulated to ignore them.
Invisible to audit. No external system knows what rules were "set."
Overridable by context. Tool outputs can inject instructions that override the system prompt.
No enforcement mechanism. Nothing actually prevents the delete_file tool from executing.
In February 2026, researchers found 7 CVEs in OpenClaw. The root cause was architectural: no certified security baseline, no mandatory audit trail, no access control enforcement by default.
The Governance-First Architecture
[INSERT IMAGE: devto_architecture.png]
Here's the architecture that passes enterprise security audits:
Layer 1: Agent Runtime (Your Framework)
Use whatever framework fits your use case:
from langgraph.graph import StateGraph
from crewai import Agent, Crew
# Your orchestration logic lives here
# This is the 20% of the work
Layer 2: Policy Engine (Out-of-Process)
Every tool call goes through an external policy engine BEFORE execution:
# policy_engine.py
from typing import Callable
import json
class PolicyEngine:
def __init__(self, policy_file: str):
with open(policy_file) as f:
self.policies = json.load(f)
def evaluate(self, agent_id: str, tool_name: str, args: dict) -> bool:
"""
Returns True if action is allowed, False otherwise.
This runs OUT OF PROCESS from the agent.
"""
# Check agent permissions
agent_policy = self.policies.get(agent_id, {})
allowed_tools = agent_policy.get("allowed_tools", [])
if tool_name not in allowed_tools:
self.log_denied(agent_id, tool_name, args)
return False
# Check data classification
if self._contains_pii(args):
if not agent_policy.get("pii_access", False):
self.log_denied(agent_id, tool_name, args, "PII access denied")
return False
self.log_allowed(agent_id, tool_name, args)
return True
Policy file (policies.json):
{
"agent_researcher": {
"allowed_tools": ["web_search", "read_file"],
"pii_access": false,
"max_cost_per_call": 0.10
},
"agent_writer": {
"allowed_tools": ["read_file", "write_file"],
"pii_access": false,
"write_paths": ["/tmp/drafts/*"]
},
"agent_admin": {
"allowed_tools": ["*"],
"pii_access": true,
"requires_human_approval": ["delete_*", "send_email"]
}
}
Layer 3: Tool Wrapper with Policy Enforcement
Wrap every tool to enforce policy before execution:
# secure_tools.py
from functools import wraps
from policy_engine import PolicyEngine
from identity import get_current_agent_id
from audit import AuditLogger
policy = PolicyEngine("policies.json")
audit = AuditLogger()
def secure_tool(func: Callable) -> Callable:
"""
Decorator that enforces policy before tool execution.
"""
@wraps(func)
def wrapper(*args, **kwargs):
agent_id = get_current_agent_id()
tool_name = func.__name__
# Log the attempt
trace_id = audit.start_trace(agent_id, tool_name, kwargs)
# Evaluate policy BEFORE execution
if not policy.evaluate(agent_id, tool_name, kwargs):
audit.log_denied(trace_id)
raise PolicyViolationError(
f"Agent {agent_id} not authorized for {tool_name}"
)
try:
result = func(*args, **kwargs)
audit.log_success(trace_id, result)
return result
except Exception as e:
audit.log_error(trace_id, e)
raise
return wrapper
# Apply to all tools
@secure_tool
def read_file(path: str) -> str:
with open(path) as f:
return f.read()
@secure_tool
def write_file(path: str, content: str) -> None:
with open(path, 'w') as f:
f.write(content)
@secure_tool
def delete_file(path: str) -> None:
os.remove(path)
Layer 4: Per-Agent Identity
No shared API keys. Every agent gets a cryptographic identity:
# identity.py
import jwt
from datetime import datetime, timedelta
from contextvars import ContextVar
_current_agent: ContextVar[str] = ContextVar('current_agent')
def create_agent_token(agent_id: str, permissions: list) -> str:
"""
Create a signed JWT for an agent instance.
"""
payload = {
"agent_id": agent_id,
"permissions": permissions,
"iat": datetime.utcnow(),
"exp": datetime.utcnow() + timedelta(hours=1)
}
return jwt.encode(payload, SECRET_KEY, algorithm="HS256")
def authenticate_agent(token: str) -> str:
"""
Verify agent token and set context.
"""
payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
agent_id = payload["agent_id"]
_current_agent.set(agent_id)
return agent_id
def get_current_agent_id() -> str:
return _current_agent.get()
Layer 5: OpenTelemetry Observability
Full tracing for every agent action:
# observability.py
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
# Initialize tracer
provider = TracerProvider()
exporter = OTLPSpanExporter(endpoint="http://otel-collector:4317")
provider.add_span_processor(BatchSpanProcessor(exporter))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("agentmesh")
def trace_agent_action(func):
@wraps(func)
def wrapper(*args, **kwargs):
with tracer.start_as_current_span(func.__name__) as span:
span.set_attribute("agent.id", get_current_agent_id())
span.set_attribute("tool.name", func.__name__)
span.set_attribute("tool.args", str(kwargs))
try:
result = func(*args, **kwargs)
span.set_attribute("tool.success", True)
return result
except Exception as e:
span.set_attribute("tool.success", False)
span.set_attribute("tool.error", str(e))
raise
return wrapper
Putting It Together
Here's the full integration with LangGraph:
# main.py
from langgraph.graph import StateGraph, END
from secure_tools import read_file, write_file, secure_tool
from identity import create_agent_token, authenticate_agent
from observability import trace_agent_action
# Create agent with identity
researcher_token = create_agent_token(
"agent_researcher",
permissions=["web_search", "read_file"]
)
# Authenticate before running
authenticate_agent(researcher_token)
# Define your graph with secure tools
def research_node(state):
# All tool calls go through policy engine
content = read_file(state["input_path"]) # Policy checked here
return {"research": content}
def write_node(state):
# This would FAIL for researcher agent
# Policy engine blocks write_file for this agent
write_file(state["output_path"], state["research"])
return state
# Build graph
graph = StateGraph()
graph.add_node("research", research_node)
graph.add_node("write", write_node)
graph.add_edge("research", "write")
graph.add_edge("write", END)
# Run with full governance
app = graph.compile()
result = app.invoke({"input_path": "/data/source.txt"})
The Kill Switch
For production, you need the ability to stop agents immediately:
# kill_switch.py
import redis
from functools import wraps
r = redis.Redis(host='localhost', port=6379, db=0)
def check_kill_switch(func):
@wraps(func)
def wrapper(*args, **kwargs):
agent_id = get_current_agent_id()
# Check if agent is killed
if r.get(f"kill:{agent_id}"):
raise AgentKilledError(f"Agent {agent_id} has been terminated")
# Check if all agents are paused
if r.get("kill:all"):
raise AgentKilledError("All agents paused by administrator")
return func(*args, **kwargs)
return wrapper
def kill_agent(agent_id: str):
"""Emergency stop for a specific agent."""
r.set(f"kill:{agent_id}", "1")
# Also revoke all active sessions
revoke_agent_tokens(agent_id)
def kill_all_agents():
"""Emergency stop for all agents."""
r.set("kill:all", "1")
What You Get
With this architecture:
✅ Every action is traceable to a specific agent with a specific identity
✅ Policy is enforced before execution, not advised in prompts
✅ Audit trails are immutable and compatible with compliance requirements
✅ Agents can be killed individually or globally in milliseconds
✅ Tool access is scoped per agent, not per prompt
✅ You pass security audits because governance is out-of-process
The Open Source Implementation
I've packaged this architecture as AgentMesh:
pip install agentmesh
# Or clone the repo
git clone https://github.com/anilatambharii/agentmesh
AgentMesh wraps any orchestration framework with the governance layer. You keep your existing LangGraph/CrewAI code and add production safety.
Basic usage:
from agentmesh import SecureAgent, PolicyEngine
# Load policy
policy = PolicyEngine.from_file("policies.yaml")
# Wrap your existing agent
agent = SecureAgent(
base_agent=your_langgraph_agent,
policy=policy,
identity="agent_researcher",
audit_backend="otlp://localhost:4317"
)
# Run with full governance
result = agent.invoke(task)
Key Takeaways
The framework is 20% of the work. LangGraph, CrewAI, and others are great for orchestration. They are not governance infrastructure.
In-process prompts are advisory. Out-of-process policy is enforceable. This is the architectural principle that matters.
Identity is non-negotiable. No shared API keys. Every agent gets cryptographic credentials.
Audit everything. If you can't trace what happened, you can't pass compliance and you can't do forensics.
Build kill switches from day one. You will need them at 3 AM.
The code is open source: github.com/anilatambharii/agentmesh
If you're building AI agents for production, I'd love to hear what patterns you're using. Drop a comment or reach out on Twitter @anilsprasad.
Anil Prasad has spent 28 years building production AI infrastructure. He is the creator of AgentMesh and founder of Ambharii Labs.
Top comments (0)