LangChain's tool abstraction makes it easy to give an AI agent access to external capabilities — web search, database queries, trade execution, file operations. That same abstraction is also the place where scope violations happen: an agent that should only read data ends up writing it, or a tool scoped to test data runs against production.
This tutorial shows how to add cryptographic scope enforcement to LangChain tool calls using the Kakunin Python SDK, so that every tool execution is gated on a verified X.509 certificate.
What You Will Build
By the end of this tutorial, your LangChain agent will:
- Verify its X.509 certificate is active before executing any tool
- Enforce that the agent holds the required scope for each specific tool
- Raise a
ScopeViolationError— not silently fail — if the check does not pass - Log every tool execution as a behavioural event for compliance audit
Prerequisites
- Python 3.11+
- A Kakunin account with an API key (
kak_live_...) - A registered and certified agent (see the quickstart)
- LangChain installed:
pip install langchain langchain-openai
Step 1: Install the Kakunin SDK
pip install kakuninSet your environment variables:
export KKN_API_KEY=kak_live_...
export KKN_AGENT_ID=agt_... # from your Kakunin dashboard
export OPENAI_API_KEY=sk-...Step 2: Understand the Scope Model
Every Kakunin-certified agent carries its permitted scopes in its metadata. When you register an agent, you define what it is authorised to do:
import asyncio
import kakunin
kkn = kakunin.Kakunin(api_key="kak_live_...")
async def register():
agent = await kkn.agents.create(
name="Trading Research Agent",
model="gpt-4o",
version="v1.0.0",
model_hash=kakunin.Kakunin.compute_model_hash("gpt-4o"),
metadata={
"scopes": ["market_data:read", "research:write", "portfolio:read"]
# No "trades:execute" — this agent is read-only
}
)
cert = await kkn.agents.certify(agent.id)
print(f"Agent {agent.id} certified: {cert.serial_number}")
asyncio.run(register())The scopes list in metadata defines what the agent is authorised to do. verify_agent_scope checks this list before every tool execution.
Step 3: Wrap a LangChain Tool with KakuninToolGuard
KakuninToolGuard wraps any LangChain BaseTool and adds a pre-execution scope check. If the agent does not hold the required scope, execution is blocked before the tool's _run method is ever called.
import os
from langchain.tools import BaseTool
from kakunin.integrations.langchain import KakuninToolGuard
import kakunin
kkn = kakunin.Kakunin(api_key=os.environ["KKN_API_KEY"])
AGENT_ID = os.environ["KKN_AGENT_ID"]
# Define your underlying tool
class MarketDataTool(BaseTool):
name: str = "get_market_data"
description: str = "Fetch current price and volume data for a given ticker symbol."
def _run(self, ticker: str) -> str:
# Your actual market data integration here
return f"AAPL: $189.42, Vol: 54.2M"
async def _arun(self, ticker: str) -> str:
return self._run(ticker)
# Wrap it — requires "market_data:read" scope
market_data_tool = KakuninToolGuard(
tool=MarketDataTool(),
kakunin=kkn,
agent_id=AGENT_ID,
required_scopes=["market_data:read"],
)Now market_data_tool behaves like a standard LangChain tool — you pass it to your agent exactly the same way — but every invocation first verifies the agent's certificate.
Step 4: Define a Restricted Tool
Some tools should only be available to agents with elevated scope. Here is a trade execution tool that requires trades:execute:
class TradeExecutionTool(BaseTool):
name: str = "execute_trade"
description: str = "Execute a market order for the given ticker and quantity."
def _run(self, ticker: str, quantity: int) -> str:
# Real execution logic here
return f"Order submitted: BUY {quantity} {ticker}"
async def _arun(self, ticker: str, quantity: int) -> str:
return self._run(ticker, quantity)
# Wrap with scope enforcement — requires "trades:execute"
trade_tool = KakuninToolGuard(
tool=TradeExecutionTool(),
kakunin=kkn,
agent_id=AGENT_ID,
required_scopes=["trades:execute"],
)When the research agent (which has ["market_data:read", "research:write", "portfolio:read"]) tries to call execute_trade, it will receive:
ScopeViolationError: Agent 'agt_...' missing required scopes: ['trades:execute']The trade is never attempted.
Step 5: Build the Agent
Wire both tools into a standard LangChain agent:
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from kakunin.exceptions import ScopeViolationError
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system", "You are a trading research assistant. Gather market data and write research notes."),
MessagesPlaceholder("chat_history", optional=True),
("human", "{input}"),
MessagesPlaceholder("agent_scratchpad"),
])
# Only provide tools the agent is scoped for
# Including trade_tool here would be caught at runtime by KakuninToolGuard
tools = [market_data_tool]
agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)Step 6: Handle ScopeViolationError
ScopeViolationError carries structured context so you can handle it precisely:
from kakunin.exceptions import ScopeViolationError
try:
result = executor.invoke({"input": "What is the current price of AAPL?"})
print(result["output"])
except ScopeViolationError as e:
print(f"Scope violation: {e}")
print(f" Agent ID: {e.agent_id}")
print(f" Agent status: {e.agent_status}")
print(f" Missing scopes: {e.missing_scopes}")
# Log to your incident system, alert on-call, etc.The error also fires if the agent's certificate has been revoked — e.agent_status will be "revoked" rather than "active". This means a compromised or non-compliant agent is blocked from all tool execution automatically, with no code change required.
Step 7: Add the Scope Callback for Full Chain Coverage
KakuninToolGuard protects individual tools. If you want a single check that gates the entire agent invocation before any tool is called, use langchain_scope_callback:
from kakunin.integrations.langchain import langchain_scope_callback
# Create a callback that checks scope before every LLM call
scope_cb = langchain_scope_callback(
kakunin=kkn,
agent_id=AGENT_ID,
required_scopes=["market_data:read"], # minimum scope for this chain
)
result = executor.invoke(
{"input": "Research AAPL and write a one-paragraph summary."},
config={"callbacks": [scope_cb]},
)The callback raises ScopeViolationError on the first LLM call if the certificate is invalid — before any tokens are generated or any tools are invoked.
Step 8: Emit Behavioural Events for Compliance
Scope enforcement satisfies the EU AI Act Article 15 cybersecurity requirement. The Article 12 audit logging requirement needs a separate instrumentation step — emit a behavioural event for each tool execution:
import asyncio
from langchain.callbacks.base import BaseCallbackHandler
class KakuninAuditCallback(BaseCallbackHandler):
def __init__(self, kkn, agent_id):
self.kkn = kkn
self.agent_id = agent_id
def on_tool_end(self, output, **kwargs):
# Fire-and-forget — do not block the agent
asyncio.ensure_future(
self.kkn.events.ingest({
"agent_id": self.agent_id,
"action_type": "tool_call",
"metadata": {"output_length": len(str(output))},
})
)
audit_cb = KakuninAuditCallback(kkn, AGENT_ID)
result = executor.invoke(
{"input": "What is the current price of AAPL?"},
config={"callbacks": [scope_cb, audit_cb]},
)Each event creates an immutable audit record stored in WORM-backed storage, satisfying the five-year retention requirement under EU AI Act Article 12.
Complete Working Example
import os
import asyncio
from langchain.tools import BaseTool
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from kakunin.integrations.langchain import KakuninToolGuard, langchain_scope_callback
from kakunin.exceptions import ScopeViolationError
import kakunin
kkn = kakunin.Kakunin(api_key=os.environ["KKN_API_KEY"])
AGENT_ID = os.environ["KKN_AGENT_ID"]
class MarketDataTool(BaseTool):
name: str = "get_market_data"
description: str = "Get current price and volume for a ticker."
def _run(self, ticker: str) -> str:
return f"{ticker}: $189.42, Vol: 54.2M"
async def _arun(self, ticker: str) -> str:
return self._run(ticker)
tools = [
KakuninToolGuard(
tool=MarketDataTool(),
kakunin=kkn,
agent_id=AGENT_ID,
required_scopes=["market_data:read"],
)
]
llm = ChatOpenAI(model="gpt-4o", temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system", "You are a trading research assistant."),
("human", "{input}"),
MessagesPlaceholder("agent_scratchpad"),
])
agent = create_openai_tools_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
scope_cb = langchain_scope_callback(kkn, AGENT_ID, required_scopes=["market_data:read"])
try:
result = executor.invoke(
{"input": "What is the current price of AAPL?"},
config={"callbacks": [scope_cb]},
)
print(result["output"])
except ScopeViolationError as e:
print(f"Blocked: {e.missing_scopes}")What Happens When a Certificate Is Revoked
When Kakunin auto-revokes a certificate (risk score ≥ 0.85), the revocation propagates to the CDN within 60 seconds. On the next tool invocation, KakuninToolGuard calls kkn.agents.get(agent_id) and receives status: "revoked". The ScopeViolationError is raised with agent_status="revoked", and no tool is executed.
Your error handler can use this to trigger an incident alert, pause the agent's job queue, and notify a human operator — satisfying the EU AI Act Article 14 human oversight requirement.
Next Steps
- Add the
verify_agent_scopedecorator to standalone async functions: see the Python SDK docs - Extend to LlamaIndex, CrewAI, or AutoGen: all four framework integrations are in
kakunin.integrations - Set up the Supabase RLS integration to enforce database-level scope at query time: see the Supabase integration guide
- Review the full EU AI Act compliance roadmap for AI agent operators: Navigating EU AI Act Compliance