Enterprise LangChain deployments in 2026 need audit logs and policy enforcement on every tool call. This tutorial wraps the Scavio LangChain tool with a governance layer: per-agent rate limits, query allowlists, and structured audit logs feeding a SIEM.
Prerequisites
- Python 3.10+
- LangChain 0.2+
- A Scavio API key
- A log destination (Loki, Datadog, CloudWatch)
Walkthrough
Step 1: Wrap the Scavio tool
Subclass the LangChain tool with policy hooks.
Python
from langchain.tools import Tool
import os, requests, json, time
def scavio_raw(query):
r = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': os.environ['SCAVIO_API_KEY']},
json={'query': query})
return r.json()Step 2: Add audit logging
Every call logs agent_id, query, result count.
Python
def audit(agent_id, query, result):
print(json.dumps({'ts': time.time(), 'agent': agent_id, 'query': query, 'results': len(result.get('organic_results', []))}))Step 3: Enforce allowlist
Block queries that contain blocked patterns.
Python
BLOCKED = ['ssn', 'credit card']
def allowed(query):
return not any(b in query.lower() for b in BLOCKED)Step 4: Apply rate limits
Per-agent token bucket.
Python
from collections import defaultdict
from time import time
BUCKET = defaultdict(list)
def under_limit(agent_id, limit=100, window=3600):
now = time()
BUCKET[agent_id] = [t for t in BUCKET[agent_id] if t > now - window]
if len(BUCKET[agent_id]) >= limit: return False
BUCKET[agent_id].append(now); return TrueStep 5: Register the governed tool
Agents use the governed wrapper, not the raw one.
Python
def governed(agent_id, query):
if not allowed(query): return {'error': 'policy_block'}
if not under_limit(agent_id): return {'error': 'rate_limit'}
result = scavio_raw(query)
audit(agent_id, query, result)
return result
tool = Tool.from_function(lambda q: governed('agent-1', q), name='search', description='Governed web search')Python Example
Python
import os, requests, json, time
API_KEY = os.environ['SCAVIO_API_KEY']
def governed_search(agent_id, query):
if 'ssn' in query.lower(): return {'error': 'policy'}
r = requests.post('https://api.scavio.dev/api/v1/search',
headers={'x-api-key': API_KEY},
json={'query': query})
print(json.dumps({'ts': time.time(), 'agent': agent_id, 'query': query}))
return r.json()
print(governed_search('agent-42', 'best serp api'))JavaScript Example
JavaScript
const API_KEY = process.env.SCAVIO_API_KEY;
const BLOCKED = ['ssn', 'credit card'];
export async function governedSearch(agentId, query) {
if (BLOCKED.some(b => query.toLowerCase().includes(b))) return { error: 'policy' };
const r = await fetch('https://api.scavio.dev/api/v1/search', {
method: 'POST',
headers: { 'x-api-key': API_KEY, 'Content-Type': 'application/json' },
body: JSON.stringify({ query })
});
const d = await r.json();
console.log(JSON.stringify({ ts: Date.now(), agent: agentId, query }));
return d;
}Expected Output
JSON
Structured audit log per call, policy blocks for sensitive queries, per-agent rate limit enforcement. Ready for SOC 2 evidence collection.