Stdlib-only safety patterns for agentic AI systems. Zero third-party runtime dependencies.
Built from production experience coordinating 5+ AI agents (Claude, Codex, Gemini) in a governed execution environment with 6,000+ tests.
| Pattern | Module | What it does |
|---|---|---|
| Circuit Breaker | circuit_breaker |
Automatic failure detection and recovery. CLOSED/OPEN/HALF_OPEN state machine. |
| Kill Switch | kill_switch |
Emergency halt with 4 modes (DISENGAGED/HALT_NONCRITICAL/HALT_ALL/EMERGENCY). |
| Coordination Bus | coordination_bus |
Append-only TSV message bus with flock locking and HMAC signing. |
| Schema Validator | schema_validator |
JSON Schema validation (Draft 2020-12 subset) without jsonschema dependency. |
| Health Probe | health_probe |
Composable health check interface with latency tracking. |
git clone https://github.com/hummbl-dev/agentic-patterns.git
cd agentic-patterns
pip install -e ".[test]"
python -m pytest tests/ -vTry the examples:
python examples/failure_injection.py # Circuit breaker lifecycle demo
python examples/agent_runner.py # Full agent with guardrailsfrom agentic_patterns.circuit_breaker import CircuitBreaker, CircuitBreakerOpen
breaker = CircuitBreaker(failure_threshold=3, recovery_timeout=30.0)
try:
result = breaker.call(external_api, request)
except CircuitBreakerOpen:
result = cached_fallbackfrom agentic_patterns.kill_switch import KillSwitch, KillSwitchMode
ks = KillSwitch(critical_tasks=frozenset({"health_check", "alerting"}))
ks.engage(KillSwitchMode.HALT_NONCRITICAL, reason="Error rate spike")
if ks.is_task_allowed("data_export"):
run_export() # Blocked during HALT_NONCRITICALfrom agentic_patterns.coordination_bus import BusWriter, sign_message
bus = BusWriter("coordination.tsv")
bus.post("agent-1", "all", "STATUS", "Task complete")
# Optional: sign messages for tamper detection
secret = "shared-secret"
sig = sign_message("payload", secret)from agentic_patterns.schema_validator import validate, ValidationError
schema = {
"type": "object",
"required": ["name", "status"],
"properties": {
"name": {"type": "string", "minLength": 1},
"status": {"type": "string", "enum": ["active", "inactive"]},
},
}
validate({"name": "agent-1", "status": "active"}, schema) # OKfrom agentic_patterns.health_probe import HealthCollector, HealthProbe, ProbeResult
class APIProbe(HealthProbe):
@property
def name(self): return "external-api"
def check(self):
# Your health check logic
return ProbeResult(name=self.name, healthy=True, message="200 OK")
collector = HealthCollector([APIProbe()])
report = collector.check_all()
print(report.overall_healthy) # TrueEvery pattern in this library uses only Python's standard library. No requests, no pydantic, no jsonschema.
This constraint is intentional:
- Zero supply chain risk for safety-critical components
- No version conflicts with the host application
- Deployable anywhere Python 3.11+ runs
- Auditable -- the entire dependency tree is Python itself
Agent Task
|
v
[Kill Switch] -- Is this task allowed?
|
v
[Circuit Breaker] -- Is the downstream service healthy?
|
v
[External Call] -- Execute with failure tracking
|
v
[Bus Writer] -- Log the event (append-only, signed)
|
v
[Health Probe] -- Report status to collectors
- Python 3.11+
- Unix-like OS for coordination bus (uses
fcntl.flock) - No runtime dependencies
This repo is part of the HUMMBL cognitive AI architecture. Related repos:
| Repo | Purpose |
|---|---|
| hummbl-governance | Production governance runtime built on these patterns |
| base120 | Deterministic cognitive framework -- 120 mental models across 6 transformations |
| mcp-server | Model Context Protocol server for Base120 integration |
| arbiter | Agent-aware code quality scoring and attribution |
| governed-iac-reference | Reference architecture for governed infrastructure-as-code |
Learn more at hummbl.io.
MIT -- see LICENSE.