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, install it from source until PyPI publication is verified:

The kill switch is the first thing to wire. 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://api.hummbl.io/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: