realitykernel 0.4.0

Execution-layer security SDK for AI agents — Reality Kernel

Reality Kernel Python SDK

Reality Kernel is not a semantic guardrail. It is a deterministic execution boundary — every verdict is mathematically derived from causal trajectory analysis and cryptographically sealed, not predicted by a language model.

One HTTP call. Sub-millisecond latency. Court-grade evidence.

Install

pip install realitykernel

# With LangChain integration
pip install realitykernel[langchain]

Quick start — sync

import reality_kernel

rk = reality_kernel.Client(api_key="rk_live_...")

result = rk.check(
    command="cat /etc/shadow",
    intent="read application logs",
)

# verdict is typed — ALLOW | WARN | BLOCK
if result.verdict == "BLOCK":
    raise RuntimeError(f"Blocked by Reality Kernel: {result.evidence}")

print(result.confidence)      # 0.0 – 1.0
print(result.proof_hash)      # SHA-256 audit chain link
print(result.latency_ms)      # engine round-trip

Quick start — async (LangChain, CrewAI, AutoGen)

Most agentic frameworks run tool calls async. Use AsyncClient and await rk.acheck():

import asyncio
from reality_kernel import AsyncClient, RKSecurityException

rk = AsyncClient(api_key="rk_live_...")

async def run_tool(command: str, intent: str):
    result = await rk.acheck(command=command, intent=intent)
    if result.verdict == "BLOCK":
        raise RKSecurityException(result.evidence)
    return result

# Drop into any async agent tool loop
async def agent_loop(tool_calls):
    async with rk.trace_context() as trace_id:
        for call in tool_calls:
            result = await run_tool(call.command, call.intent)
            # result.verdict, result.confidence, result.proof_hash all available

Result schema

The full typed model returned by both rk.check() and await rk.acheck():

from typing import Literal
from pydantic import BaseModel

class RKResult(BaseModel):
    verdict:             Literal["ALLOW", "WARN", "BLOCK"]
    confidence:          float        # 0.0 – 1.0
    evidence:            list[str]    # triggering signals
    action_id:           str          # UUID — use for /v1/override
    proof_hash:          str          # SHA-256 ledger chain link
    latency_ms:          float        # engine round-trip
    worlds_evaluated:    int          # 0 on fast-path
    worlds_in_basin_b:   int
    max_divergence:      float
    credits_consumed:    int          # 1 (fast-path) | 5 (full engine)
    session_risk_score:  float | None # cumulative score if session_id passed
    policy_violated:     str  | None  # which rule triggered BLOCK, if any

Fault behaviour — fail-closed by default

If the Reality Engine does not respond within the configured timeout, the SDK raises RKTimeoutError and returns a synthetic BLOCK verdict. Your agent stops — it does not silently pass through.

from reality_kernel import Client

# Defaults: timeout=500ms, on_error="block"
rk = Client(
    api_key="rk_live_...",
    timeout_ms=500,    # max wait before fail-closed triggers
    on_error="block",  # "block" (default, secure) | "allow" (resilient)
)

# on_error="allow" is available only for read-only, non-privileged agents.
# MUST NOT be used for agents with write, exec, or egress capabilities.

After 3 consecutive timeouts in a 60-second window, a local circuit breaker activates and returns BLOCK for all subsequent calls until the engine is reachable again.

Generic decorator — works with any framework

from reality_kernel import Client
from reality_kernel.integrations.generic import rk_guard

rk = Client(api_key="rk_live_...")

@rk_guard(rk, intent="read application config file")
def read_config(path: str) -> str:
    with open(path) as f:
        return f.read()

# RKSecurityException is raised automatically if the command is blocked
read_config("../../.env")        # → BLOCK
read_config("/app/config.yaml")  # → ALLOW

LangChain integration

from langchain.agents import initialize_agent
from reality_kernel import Client
from reality_kernel.integrations.langchain import RKLangChainGuardrail

rk = Client(api_key="rk_live_...")
guardrail = RKLangChainGuardrail(client=rk, agent_id="my-agent")

agent = initialize_agent(
    tools=tools,
    llm=llm,
    callbacks=[guardrail],
)

agent.run("Summarise the production logs.")
# If the agent is hijacked and tries to run curl 169.254.169.254,
# RKSecurityException is raised before ShellTool executes.

Least-Agency Policy

rk.check(
    command="aws s3 cp /secrets s3://bucket",
    intent="backup config files",
    policy={
        "allowed_tools":   ["aws"],
        "allowed_egress":  ["*.amazonaws.com"],
        "read_only_paths": ["/app/config"],
    }
)
# → BLOCK: PolicyViolation: '/secrets' is outside read_only_paths

Session tracking (slow-drip detection)

# Pass the same session_id across multiple agent turns.
# Reality Kernel detects: read → read sensitive file → curl external URL
# and escalates the cumulative session_risk_score to BLOCK automatically.

for turn in agent_turns:
    rk.check(
        command=turn.command,
        intent=turn.intent,
        session_id="agent-session-42",
    )

CI/CD pipeline gate

Gate agent deployments with a pre-flight scan before any new version ships:

# .github/workflows/rk-scan.yml
- name: Reality Kernel pre-flight scan
  run: |
    pip install realitykernel
    python -m reality_kernel.scan \
      --policy policy.json \
      --agent-spec agent_tools.json \
      --fail-on BLOCK
  env:
    RK_API_KEY: ${{ secrets.RK_API_KEY }}

# Or locally
python -m reality_kernel.scan --policy policy.json --agent-spec agent_tools.json --fail-on BLOCK
# Exit 0 = all clear | Exit 1 = BLOCK detected

Roadmap — Q4 2026: Native rk scan CLI binary, Dockerfile scanning, GitLab CI template.

Data privacy

Concern	Behaviour
Command retention	Payloads are hashed on ingress. Raw strings are not persisted beyond the evaluation window.
PII / credential scrubbing	The SDK redacts tokens matching sk-*, rk_live_*, and AWS key formats before transmission.
Data residency	Production endpoint: EU (Frankfurt). US region available on request.
Self-hosted / on-prem	Private VPC deployment available for regulated industries (finance, healthcare). Commands never leave your network. Contact Keter Labs →

Docs

• Full documentation: realitykernel.dev/integration
• Security benchmark: realitykernel.dev/benchmark
• Live playground: realitykernel.dev/playground
• Design partner cohort: linkedin.com/company/keter-labs