Quickstart

One package, no third-party dependencies.

Requires Python 3.11+. Works in any environment including Docker without internet access and air-gapped deployments. If you want the mental models layer too:

The kill switch is the first thing to wire. It's the one control that cannot be overridden by any agent. Every agent action should check it before executing.

from pathlib import Path
from hummbl_governance import KillSwitch, KillSwitchMode

# Create once; state is file-system-persisted
ks = KillSwitch(state_dir=Path("_state"))

# The check: every agent loop starts here
if ks.engaged():
    print("halted:", ks.get_status())
    raise SystemExit(0)

# Your agent logic goes here
print("system is running normally")

Delegation tokens give each agent a signed scope. Every action is traceable to an authorized delegation. Enable the IDP feature flag first.

import os
from hummbl_governance import DelegationTokenManager

os.environ["ENABLE_IDP"] = "true"

mgr = DelegationTokenManager()

# Issue a token for this agent session
token = mgr.create_token(
    agent_id="research-agent-1",
    scope=["read:docs", "write:drafts"],
    ttl=3600,      # 1 hour
    chain_depth=1, # can sub-delegate once
)

# Verify scope before each action
if mgr.check_least_privilege(token, op="write:drafts"):
    write_draft(content)
else:
    raise PermissionError("agent not authorized for write:drafts")

The governance bus is the audit trail. Append-only JSONL. Every agent action gets a timestamped, identity-tagged entry. This is what an assessor reads.

from pathlib import Path
from hummbl_governance import BusWriter

bus = BusWriter(
    log_path=Path("_state/governance_bus.jsonl"),
    agent_id="research-agent-1",
)

# Log before acting
bus.write(
    msg_type="ACTION",
    message="writing draft: product-spec-v1.md",
    metadata={"token": token[:16] + "...", "scope": "write:drafts"},
)

# The action
write_draft(content, path="product-spec-v1.md")

# Log the result
bus.write(
    msg_type="RECEIPT",
    message="draft complete",
    metadata={"lines": 142, "path": "product-spec-v1.md"},
)

All four primitives wired together. Copy-paste this into a single file, run it, and you have a governed agent loop.

"""
governed_agent.py — minimal governed AI agent
Requires: pip install hummbl-governance
Python 3.11+, stdlib-only
"""
import os
from pathlib import Path
from hummbl_governance import (
    KillSwitch, KillSwitchMode,
    DelegationTokenManager,
    BusWriter,
    CircuitBreaker,
)

os.environ["ENABLE_IDP"] = "true"

# ── 1. State directory ────────────────────────────────────────────
state = Path("_state")
state.mkdir(exist_ok=True)

# ── 2. Kill switch ────────────────────────────────────────────────
ks = KillSwitch(state_dir=state)
if ks.engaged():
    raise SystemExit(f"kill switch active: {ks.get_status()}")

# ── 3. Delegation token ───────────────────────────────────────────
mgr = DelegationTokenManager()
token = mgr.create_token(
    agent_id="research-agent-1",
    scope=["read:web", "write:drafts"],
    ttl=3600,
    chain_depth=1,
)

# ── 4. Governance bus ─────────────────────────────────────────────
bus = BusWriter(
    log_path=state / "governance_bus.jsonl",
    agent_id="research-agent-1",
)

# ── 5. Circuit breaker (wrap any external call) ───────────────────
cb = CircuitBreaker(failure_threshold=3, recovery_timeout=60.0)

# ── 6. Governed agent loop ────────────────────────────────────────
def fetch_page(url: str) -> str:
    import urllib.request
    with urllib.request.urlopen(url, timeout=10) as r:
        return r.read().decode()

tasks = [
    "https://hummbl-api.hummbl.workers.dev/health",
]

for url in tasks:
    # Check kill switch before each task
    if ks.engaged():
        bus.write("HALT", f"kill switch engaged mid-loop: {url}")
        break

    # Check authorization
    if not mgr.check_least_privilege(token, op="read:web"):
        bus.write("DENIED", f"no scope for read:web — skipping {url}")
        continue

    bus.write("ACTION", f"fetching {url}", metadata={"token_prefix": token[:12]})

    try:
        result = cb.call(fetch_page, url)
        bus.write("RECEIPT", f"fetched {url}", metadata={"bytes": len(result)})
        print(f"[OK] {url}: {len(result)} bytes")
    except Exception as e:
        bus.write("ERROR", f"failed {url}: {e}")
        print(f"[ERR] {url}: {e}")

print(f"\nAudit log: {state / 'governance_bus.jsonl'}")

Run it:

Then read the audit log: