ACEL — Agent Contract Enforcement Layer (core)

Runtime verification for AI agent tool calls. Declare temporal ordering contracts and Hoare-style pre/postconditions in plain Python, and have them enforced live against the stream of tool calls an agent makes — halting the agent the moment a rule is broken.

acel-core ships the transport-independent monitor (Phase 1) plus a live MCP proxy (Phase 2): the same contracts enforced against a real MCP server, via the official MCP Python SDK's request-middleware pipeline.

This is runtime verification — checking each concrete execution against a specification as it happens. It does not prove the agent correct in general; it guarantees that this run did not violate the rules you declared.

Why

Statistical agent-eval tools answer "how often does this agent behave well, on average?" ACEL answers the production question they can't: "did this execution just violate a rule we cannot allow to be violated?" — before the bad tool call lands.

Install

pip install -e .

Quickstart

from acel import Session, must_precede, at_most_n_times

session = Session(state={"authenticated": False})

# Temporal ordering rules (no logic syntax required):
session.add_contract(must_precede("validate_record", "delete_record"))
session.add_contract(at_most_n_times("send_payment", n=1))

# State-based gate: reads only allowed once authenticated.
session.register_tool(
    "read_user_data",
    precondition=lambda s: s.get("authenticated") is True,
)
# Authentication commits trusted info into session state.
session.register_tool(
    "authenticate",
    commit=lambda s, args, result: s.set("authenticated", result["ok"]),
)

session.call("authenticate", {"user": "p"}, result={"ok": True})
session.call("read_user_data", {"query": "SELECT ..."}, result={"rows": []})

# This halts: delete before validate.
session.call("delete_record", {"id": "r_42"})   # raises ContractViolation

On violation, a ContractViolation is raised carrying a Violation record:

from acel import ContractViolation

try:
    session.call("delete_record", {"id": "r_42"})
except ContractViolation as exc:
    v = exc.violation
    print(v.kind)            # "temporal"
    print(v.spec)            # "must_precede(validate_record, delete_record)"
    print(v.step)            # index of the offending call
    print(v.trace)           # every call up to the violation
    print(v.state_snapshot)  # symbolic state at the moment it broke

The six temporal templates

Each template is a deterministic automaton advanced in O(1) per tool call, with a three-valued verdict (SATISFIED / VIOLATED / UNKNOWN) over the finite trace.

Pre/postconditions with decorators

from acel import Session, precondition, postcondition

@precondition(lambda s: s.get("authenticated") is True)
@postcondition(lambda s, r: r["tenant_id"] == s.get("current_tenant"))
def search_database(query): ...

session = Session(state={"authenticated": True, "current_tenant": "t_9"})
session.register(search_database)

Offline analysis / CI mode

Session.replay runs a recorded trace and returns every violation without executing anything — the basis for the coming acel replay trace.json CLI and for testing contracts against known-bad traces.

violations = session.replay([
    {"tool": "delete_record", "args": {"id": "1"}},
])

Live MCP proxy (Phase 2)

ACEL can gate a real MCP server's tool calls, live, via the official MCP Python SDK's ServerMiddleware hook. Every tools/call request passes through ACEL's gate before the real tool handler runs — a blocked call has zero side effects.

pip install "acel-core[mcp]"

from mcp.server.mcpserver import MCPServer
from acel import Session, must_precede
from acel.mcp_middleware import ACELMiddleware

session = Session()
session.add_contract(must_precede("validate_record", "delete_record"))

server = MCPServer("my-server", middleware=[ACELMiddleware(session)])

@server.tool()
def delete_record(record_id: str) -> dict: ...

See examples/toy_server.py for a complete toy server (5 tools, 3 contracts) and tests/test_mcp_proxy.py for an end-to-end demo: a real ClientSession talking to this server, with ACEL catching an ordering violation, a cardinality violation, and a state-precondition violation — each one halted before the tool it would have run.

Shadow mode

The recommended way to roll out a new set of contracts: shadow mode detects and records every violation exactly as enforce mode does — same evidence log, same hash chain — but never blocks a call. Run it against real traffic first, see what it would have caught, then switch to enforce once you trust the rules.

session = Session(mode="shadow")  # default is "enforce"

acel serve examples/toy_server.py --shadow

Session.call(), .precheck()/.postcheck() (the MCP proxy path), and the CLI all respect mode. Session.replay() does not — it's a retrospective CI-gate tool ("would this recorded trace have been blocked"), not a live session, so it always reports every violation regardless of mode.

Config-driven contracts (no code required)

Temporal contracts can be declared in a plain JSON or YAML file instead of Python — useful for trying ACEL against your own tools without writing any code, or for keeping the rule set separate from your server implementation:

acel init-config rules.yaml     # writes a starter file
acel validate rules.yaml        # parses it, prints the contracts it declares

state:
  authenticated: false

contracts:
  - template: must_precede
    args: [validate_record, delete_record]
  - template: at_most_n_times
    args: [send_payment]
    kwargs: {n: 1}

Layer a rules file on top of a live server (--contracts adds to whatever build_server() already sets up, and merges the state block in):

acel serve examples/toy_server.py --contracts rules.yaml

Or check a recorded trace against a rules file directly (the same format acel replay has always used, now also parseable as YAML):

acel replay trace.json --rules rules.yaml

Why preconditions/postconditions aren't in the config file: they evaluate real logic over state (lambda s: s.get("authenticated") is True), and there's no safe way to deserialize arbitrary logic from a data file without either an eval-style security hole or a bespoke expression language. Temporal contracts have no such problem — every template is fully described by tool names and simple parameters, so building one from a config file is just constructing an object from validated data, no code execution involved. Pre/postconditions stay in Python, wired directly to your tools — install YAML support with pip install "acel-core[config]".

Correctness

python benchmarks/correctness.py

A labeled dataset of 51 synthetic tool-call traces spanning all 6 temporal templates (valid sequences, violating sequences, and edge cases like empty traces and multiple simultaneous contracts) — measured at 100% precision and 100% recall. Since the monitor is deterministic automaton checking, not statistical detection, that's the expected result; the suite exists to prove it and to catch any future regression (it's also wired into pytest as tests/test_correctness_suite.py, so a miss fails CI directly).

Performance

python benchmarks/latency.py

Measured on the reference dev machine, 20,000 iterations, discarding a 1,000-call warmup: added p95 latency per tool call is ~0.005ms at 1 active contract and ~0.04ms at 50 concurrently active contracts — well under the <5ms target. Each temporal contract is a deterministic automaton advanced in O(1) per event, so overhead scales linearly with the number of active contracts, not with session length.

Tests

pip install pytest
pytest                    # core monitor + evidence (no extra deps)
pip install "acel-core[mcp]"
pytest tests/test_mcp_proxy.py tests/test_cli_serve.py   # live MCP proxy + CLI

License

MIT