hummbl-governance

Install

Zero third-party runtime dependencies. Works on air-gapped deployments, Docker without internet access, and regulated environments that prohibit external packages.

Test dependencies (pytest, coverage) are optional: pip install hummbl-governance[test].

Module index

Six primitives in one package. Import what you need:

DelegationTokenManager

HMAC-SHA256 signed capability tokens for agent authority. Every token carries scope, chain depth limit, and expiry. Feature-flagged via ENABLE_IDP=true.

Primitive page: Delegation Tokens →

Import

from hummbl_governance import DelegationTokenManager
from hummbl_governance.delegation import TokenBinding

Constructor

DelegationTokenManager(
    secret: bytes | None = None,
    # If None, generates a random secret. Pass bytes for reproducible tokens.
)

Key methods

Example

import os
from hummbl_governance import DelegationTokenManager

os.environ["ENABLE_IDP"] = "true"
mgr = DelegationTokenManager()

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

# Validate before executing
binding = mgr.validate_token(token)
if binding and mgr.check_least_privilege(token, op="write:drafts"):
    write_draft(content)

BusWriter

Append-only JSONL audit log. Every write is a line. Lines are never deleted or modified. The log is the proof of governance.

Primitive page: Governance Bus →

Import

from hummbl_governance import BusWriter

Constructor

BusWriter(
    bus_path: str | Path,                    # Path to the JSONL audit file
    *,
    policy: PolicyLevel = PolicyLevel.PERMISSIVE,
)

Key methods

Example

from pathlib import Path
from hummbl_governance import BusWriter

bus = BusWriter(
    bus_path=Path("_state/governance_bus.jsonl"),
)

# Log an action
bus.post(
    msg_type="ACTION",
    message="fetched arxiv paper 2401.00001",
    metadata={"source": "arxiv", "doi": "2401.00001"},
)

# Read recent activity
for entry in bus.read_since("2026-04-14T00:00:00Z"):
    print(entry.from_id, entry.msg_type, entry.message)

KillSwitch

Four-mode emergency halt. File-system-persisted state survives process restart. Non-delegable: no agent can override the kill switch.

Primitive page: Kill Switch →

Import

from hummbl_governance import KillSwitch, KillSwitchMode

Modes

Constructor

KillSwitch(
    state_dir: Path | None = None,       # Default: ./_state/
    require_hmac: bool = True,
    signing_secret: bytes | None = None,
    critical_tasks: frozenset[str] | None = None,
)

Example

from pathlib import Path
from hummbl_governance import KillSwitch, KillSwitchMode

ks = KillSwitch(state_dir=Path("_state"))

# Check before every agent action
if ks.check_task_allowed(task.id):
    agent.execute(task)
else:
    logger.info("halted: %s", ks.get_status())

# Engage from anywhere in your system
ks.engage(
    mode=KillSwitchMode.HALT_ALL,
    reason="anomaly detected in prod",
    triggered_by="ops-agent-1",
)

# Release after incident resolution
ks.disengage(reason="incident resolved", triggered_by="ops-agent-1")

CircuitBreaker

Wraps any callable — LLM API, vector store, webhook — and isolates failures. Prevents cascading degradation and runaway cost from retrying failing paid APIs.

Primitive page: Circuit Breaker →

Import

from hummbl_governance import CircuitBreaker, CircuitBreakerState

Constructor

CircuitBreaker(
    failure_threshold: int = 5,          # failures before OPEN
    recovery_timeout: float = 30.0,      # seconds before HALF_OPEN probe
    on_state_change: Callable | None = None,
)

Key methods

Example

from hummbl_governance import CircuitBreaker, CircuitBreakerState

cb = CircuitBreaker(failure_threshold=5, recovery_timeout=30.0)

try:
    result = cb.call(anthropic_client.messages.create, **params)
except Exception:
    result = fallback_handler()

if cb.state == CircuitBreakerState.OPEN:
    logger.warning("anthropic-api breaker is open")

CapabilityFence

Scope-narrowing context manager for multi-agent delegation chains. Prevents privilege escalation. Can be created standalone or bound to a delegation token.

Primitive page: Delegation Context →

Import

from hummbl_governance import CapabilityFence, CapabilityDenied

Constructor

CapabilityFence(
    allowed: list[str] | None = None,
    denied: list[str] | None = None,
    audit_log: list[CapabilityAuditEntry] | None = None,
)

Key methods

Example

from hummbl_governance import CapabilityFence, CapabilityDenied

fence = CapabilityFence(
    allowed=["read:docs", "write:drafts"],
    denied=["deploy:prod", "delete:*"],
)

# Explicit check
if fence.check("write:drafts"):
    write_draft(content)

# Context manager — raises on violation
with fence.guard("read:docs"):
    read_docs()

# Bind to a delegation token
fence_from_token = CapabilityFence.from_delegation_token(
    token,
    denied=["deploy:prod"],
)

MCP Attestation

Local-first governance for MCP-native agents. Server allowlists, capability tokens per server, tool-poisoning defense, and audit receipts for every tool call.

Primitive page: MCP Attestation →

Import

from hummbl_governance.mcp import Attest, Policy

Example

from hummbl_governance.mcp import Attest, Policy

policy = Policy.from_file("~/.hummbl/mcp-allowlist.yaml")
attest = Attest(policy=policy, bus_path="_state/mcp_bus.jsonl")

# Verify before connecting to a server
if attest.verify_server("postmark-mcp", fingerprint=FINGERPRINT):
    client.connect("postmark-mcp")
else:
    log.error("server not in allowlist")

# Generate an audit receipt for every tool call
receipt = attest.record(
    server="filesystem-mcp",
    tool="write_file",
    args={"path": "/tmp/x.md"},
    capability_token=current_token,
)

Feature flags

Some primitives are feature-flagged to allow gradual adoption in existing codebases:

Set in environment or .env. Feature flags are checked at runtime, not import time.

Contracts

Wire formats for BusWriter and DelegationTokenManager are frozen under fm-contracts-v0.1. Breaking changes require a SemVer major bump and a new baseline tag.

The contracts directory (contracts/ in the source repo) is the source of truth. If your code checks for specific JSONL fields, pin to a released version and test against the schema.

Source: github.com/hummbl-dev/hummbl-governance →