Enforce runtime policies on LangChain tools by wrapping each tool in a policy layer that checks rate limits, content filters, budget caps, and access permissions before execution. LangChain agents make autonomous tool decisions, which means unconstrained tools can exhaust API budgets, leak sensitive queries, or call tools inappropriately. A policy wrapper intercepts every tool call, applies rules, logs violations, and blocks unauthorized actions without modifying the underlying tool code.
Prerequisites
- Python 3.8+ installed
- langchain library installed
- requests library installed
- A Scavio API key from scavio.dev
Walkthrough
Step 1: Define policy rules
Create a policy configuration that specifies rate limits, blocked queries, budget caps, and required permissions for each tool.
Python
import os, time, json, requests
from collections import defaultdict
from functools import wraps
API_KEY = os.environ['SCAVIO_API_KEY']
POLICIES = {
'web_search': {
'rate_limit': 30, # calls per minute
'budget_limit': 100, # max calls per session
'blocked_patterns': ['password', 'ssn', 'credit card'],
'required_role': 'user',
},
}
usage = defaultdict(lambda: {'count': 0, 'window_start': time.time(), 'total': 0})
violation_log = []Step 2: Create the policy wrapper
Build a decorator that intercepts tool calls and applies all policy checks before allowing execution.
Python
def policy_check(tool_name: str, query: str, user_role: str = 'user') -> dict:
policy = POLICIES.get(tool_name, {})
now = time.time()
u = usage[tool_name]
# Rate limit check
if now - u['window_start'] > 60:
u['count'] = 0
u['window_start'] = now
if u['count'] >= policy.get('rate_limit', 999):
return {'allowed': False, 'reason': 'rate_limit_exceeded'}
# Budget check
if u['total'] >= policy.get('budget_limit', 9999):
return {'allowed': False, 'reason': 'budget_exceeded'}
# Content filter
query_lower = query.lower()
for pattern in policy.get('blocked_patterns', []):
if pattern in query_lower:
return {'allowed': False, 'reason': f'blocked_pattern: {pattern}'}
# Role check
required_role = policy.get('required_role', 'user')
roles_hierarchy = {'admin': 3, 'user': 2, 'guest': 1}
if roles_hierarchy.get(user_role, 0) < roles_hierarchy.get(required_role, 0):
return {'allowed': False, 'reason': 'insufficient_permissions'}
u['count'] += 1
u['total'] += 1
return {'allowed': True}
print(policy_check('web_search', 'best crm 2026'))
print(policy_check('web_search', 'find password reset link'))Step 3: Apply policies to LangChain tools
Wrap LangChain tool functions with the policy layer so every call is checked before execution.
Python
def enforced_tool(tool_name: str, func):
@wraps(func)
def wrapper(query: str, **kwargs) -> str:
check = policy_check(tool_name, query)
if not check['allowed']:
violation = {
'tool': tool_name, 'query': query[:100],
'reason': check['reason'], 'time': time.time(),
}
violation_log.append(violation)
return f'[BLOCKED] {check["reason"]}'
return func(query, **kwargs)
return wrapper
def web_search(query: str) -> str:
resp = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY},
json={'platform': 'google', 'query': query}, timeout=10)
results = resp.json().get('organic_results', [])[:3]
return '\n'.join(f"{r['title']}: {r.get('snippet', '')}" for r in results)
safe_search = enforced_tool('web_search', web_search)
print(safe_search('best crm 2026'))
print(safe_search('find password reset'))Step 4: Log violations for review
Record all policy violations with context for security review and policy tuning.
Python
def log_violation(tool: str, query: str, reason: str):
entry = {
'tool': tool,
'query': query[:200],
'reason': reason,
'timestamp': time.time(),
}
violation_log.append(entry)
print(f'VIOLATION: {tool} - {reason} - query: {query[:50]}')
def get_violation_report() -> dict:
report = {
'total_violations': len(violation_log),
'by_reason': defaultdict(int),
'by_tool': defaultdict(int),
}
for v in violation_log:
report['by_reason'][v['reason']] += 1
report['by_tool'][v['tool']] += 1
report['by_reason'] = dict(report['by_reason'])
report['by_tool'] = dict(report['by_tool'])
return report
print(json.dumps(get_violation_report(), indent=2))Step 5: Test the enforcement
Run test scenarios to verify rate limits, content filters, and budget caps all work correctly.
Python
def test_policies():
# Test content filter
result = safe_search('what is my ssn number')
assert 'BLOCKED' in result, 'Content filter should block SSN query'
print('Content filter: PASS')
# Test normal query
result = safe_search('python tutorial 2026')
assert 'BLOCKED' not in result, 'Normal query should succeed'
print('Normal query: PASS')
# Test rate limiting (exhaust limit)
original_limit = POLICIES['web_search']['rate_limit']
POLICIES['web_search']['rate_limit'] = 2
usage['web_search']['count'] = 0
usage['web_search']['window_start'] = time.time()
safe_search('test query 1')
safe_search('test query 2')
result = safe_search('test query 3')
assert 'BLOCKED' in result, 'Rate limit should block third query'
print('Rate limiting: PASS')
POLICIES['web_search']['rate_limit'] = original_limit
report = get_violation_report()
print(f'Total violations logged: {report["total_violations"]}')
print('All policy tests passed')
test_policies()Python Example
Python
import requests, os, time
from collections import defaultdict
H = {'x-api-key': os.environ['SCAVIO_API_KEY']}
usage = defaultdict(int)
def enforced_search(query, limit=30):
if usage['search'] >= limit:
return 'Rate limit exceeded'
blocked = ['password', 'ssn', 'credit card']
if any(b in query.lower() for b in blocked):
return 'Blocked by content policy'
usage['search'] += 1
data = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
json={'platform': 'google', 'query': query}).json()
return data.get('organic_results', [])[:3]
print(enforced_search('best crm 2026'))JavaScript Example
JavaScript
const H = {'x-api-key': process.env.SCAVIO_API_KEY, 'Content-Type': 'application/json'};
let usage = 0;
const BLOCKED = ['password', 'ssn', 'credit card'];
async function enforcedSearch(query, limit = 30) {
if (usage >= limit) return 'Rate limit exceeded';
if (BLOCKED.some(b => query.toLowerCase().includes(b))) return 'Blocked';
usage++;
const r = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST', headers: H, body: JSON.stringify({platform: 'google', query})
});
return ((await r.json()).organic_results || []).slice(0, 3);
}
enforcedSearch('best crm 2026').then(console.log);Expected Output
JSON
A policy enforcement layer for LangChain tools that blocks unauthorized queries, enforces rate and budget limits, and logs all violations for review.