Code examples
Real, working snippets you can copy into a new project. Each example shows a structured event, the Rego rule that decides it, and a one-line note on when you'd reach for it.
Install and minimal adapter
These snippets are concrete instances of the patterns described in governance for AI agents: deterministic policy at the action layer, with an audit trail per decision. Pick the adapter for your framework, point it at a directory of Rego policies, and every governed event flows through them automatically.
pip install kitelogikfrom openai import OpenAI
from kitelogik.adapters.openai import govern
client = govern(
OpenAI(),
policies="./policies", # directory of .rego files
audit_log="./audit.jsonl", # immutable per-event log
)
# From here on, every tool call this client makes is
# evaluated against your Rego policies before execution.
response = client.responses.create(
model="gpt-5",
input="Refund order #4821",
tools=[refund_tool, lookup_order_tool],
)Adapters ship for OpenAI, OpenAI Agents SDK, LangChain, LangGraph, CrewAI, Pydantic AI, LlamaIndex, Semantic Kernel, Haystack, Google ADK, and Dify. All adapters share the same policy engine and audit log.
Block writes outside an allowed directory
Classic sandboxing: the agent can write files, but only inside /tmp. Anything else is denied with a structured reason the agent can react to.
package kitelogik.tool_call
# Default: allow tool calls. Deny rules below override.
default decision := {"allow": true}
deny[reason] {
input.tool == "fs.write"
not startswith(input.args.path, "/tmp/")
reason := sprintf("write outside /tmp denied: %v", [input.args.path])
}
decision := {"allow": false, "deny": deny[_]} { deny[_] }Cap delegation depth
Stop runaway agent trees. A child agent can spawn its own children, but only up to two levels deep — anything beyond that needs human approval.
package kitelogik.agent_spawn
default decision := {"allow": true}
decision := {"allow": false, "reason": reason} {
input.parent.depth >= 2
reason := sprintf("delegation depth %v exceeds limit (2)", [input.parent.depth])
}Require approval for finance writes above $1000
Human-in-the-loop on sensitive operations. The call pauses until a reviewer approves or rejects asynchronously — small transactions go through unattended.
package kitelogik.tool_call
default decision := {"allow": true}
# Always require approval for tools tagged finance.write above the threshold.
decision := {"escalate": true, "reviewers": ["finance-oncall"]} {
"finance.write" in input.tool_tags
input.args.amount > 1000
}Cap a session's token budget
Stop runaway loops by enforcing a hard ceiling on per-session token spend. Once the budget is exhausted, the next governed event is denied.
package kitelogik.agent_resource
default decision := {"allow": true}
decision := {"allow": false, "reason": "session token budget exhausted"} {
input.session.tokens_spent > 100000
}MCP per-tool allowlist
Connecting an MCP server expands your agent's blast radius. This policy narrows that surface to a specific allowlist per agent role.
package kitelogik.tool_call
# Default deny for MCP calls — every allowed tool must be explicit.
default decision := {"allow": false, "reason": "MCP tool not in allowlist"}
allowed_for_role := {
"billing": {"stripe.refund", "stripe.lookup_charge"},
"support": {"zendesk.search", "zendesk.read_ticket"},
"analyst": {"warehouse.read_query"},
}
decision := {"allow": true} {
input.transport == "mcp"
input.tool in allowed_for_role[input.session.agent_role]
}Read more about the architecture in MCP governance with OPA/Rego.
Plan-level constraints
Evaluate a proposed action sequence before any step runs. Useful for agent planners that emit a tool plan and ask permission to execute it.
package kitelogik.agent_plan
default decision := {"allow": true}
write_tool_steps := count([s | s := input.plan.steps[_]; s.type == "write"])
decision := {"allow": false, "reason": reason} {
count(input.plan.steps) > 20
reason := sprintf("plan has %v steps; limit is 20", [count(input.plan.steps)])
}
decision := {"escalate": true, "reason": reason} {
write_tool_steps > 3
reason := sprintf("plan contains %v write steps; require approval", [write_tool_steps])
}Querying the audit log
Every governed decision is recorded as an immutable JSONL event with the policy version that decided it. Replay or query it like any structured log.
import json
from collections import Counter
# Load the audit log emitted by the adapter.
events = [json.loads(line) for line in open("audit.jsonl")]
# How many times did each tool fire — and how many were denied?
denied_by_tool = Counter(
e["tool"] for e in events
if e["event"] == "tool_call" and e["decision"]["allow"] is False
)
print(denied_by_tool.most_common(10))
# Audit answer: who denied this and which policy version?
for e in events:
if e.get("decision", {}).get("allow") is False:
print(e["timestamp"], e["policy_version"], e["decision"]["reason"])Frequently asked questions
Where do I put these Rego policies?
Anywhere your OPA evaluation pipeline can load them — typically a `policies/` directory in your repo. Kite Logik can load policies from a local directory, an OCI bundle, or a remote OPA server.
Do I need an OPA server running?
No. Kite Logik can evaluate Rego in-process using the embedded engine, or push evaluation to a remote OPA server / OPAL distribution if you already operate one. The same policies work in either mode.
How do I test a policy?
OPA ships first-class unit testing for Rego. Write `_test.rego` files alongside your policies and run `opa test policies/`. Kite Logik's policies use the same toolchain — your CI runs the tests like any other code.
Can policies depend on session context?
Yes. Every governed event includes the session metadata (agent identity, user, tenant, scope, prior tool calls in this session). Your Rego rules can branch on any of it.