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-coreships 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.
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.
pip install -e .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 ContractViolationOn 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| Template | Meaning |
|---|---|
must_precede(a, b) |
every b must be preceded by some a |
at_most_n_times(a, n) |
a occurs at most n times per session |
at_most_total(a, field, limit) |
the sum of args[field] across all calls to a must not exceed limit |
never_after(a, b) |
a must never occur after b |
required_before_session_end(a) |
a must occur at least once before the session ends |
cannot_follow_without(a, b) |
a may not occur unless b occurred earlier |
mutually_exclusive(a, b) |
a and b must not both occur in one session |
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. at_most_total is the odd one out — it's the only template that reads
call arguments rather than just the tool name, since it has to sum a
numeric field (e.g. a payment amount) across calls. It fails closed: a call
missing the field, or with a non-numeric value there, is treated as a
violation rather than silently let through — for a contract whose whole
purpose is capping spend, silently ignoring an unreadable amount would be the
actually dangerous failure mode.
from acel import Session, at_most_total
session = Session()
session.add_contract(at_most_total("send_payment", "amount", limit=500))
session.call("send_payment", {"amount": 300}, result={"sent": True})
session.call("send_payment", {"amount": 250}, result={"sent": True}) # raises: 550 > 500from 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)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"}},
])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.
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 --shadowSession.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.
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 declaresstate:
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.yamlOr 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.yamlWhy 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]".
Every violation is recorded as a tamper-evident, hash-chained bundle. Save
one to disk and check it later — from a completely fresh process, with no
in-memory state — with acel verify:
acel replay trace.json --rules rules.json --save-evidence evidence.json
acel verify evidence.jsonOK — 3 bundle(s) verified. Hash chain is intact, no tampering detected.
If any field in any bundle was altered after the fact, acel verify fails
and reports the exact bundle index where the chain first breaks — everything
from that point onward is untrustworthy, but pinpointing where it broke is
what actually helps you investigate:
FAIL — tampering detected. Bundle 1 (of 5) is the first to break the chain...
- Evidence bundles embed full call arguments, results, and state snapshots. That's what makes them useful evidence, but it also means anything sensitive passed as a tool argument (a password, a raw token, a secret) ends up persisted verbatim if you save an evidence log to disk or share it. ACEL doesn't redact argument values — it has no way to know which fields are sensitive without you telling it. Keep secrets out of tool arguments entirely; pass a reference/ID instead and resolve the real secret inside your own tool implementation.
ed25519_signer()generates a fresh, unpersisted key every call. Only the public key comes back — the private key never leaves memory and isn't saved anywhere. That's fine for signing within one process's lifetime, but restart the process (or call it again) and old signatures are no longer verifiable against the new public key. If you need signatures that stay verifiable across restarts, generate and store your own long-lived Ed25519 keypair rather than relying on this convenience function.- Config files (
--rules,--contracts) are parsed withyaml.safe_loadandjson.loadsonly — neveryaml.loadoreval. There is no code execution path from a rules file; that's exactly why pre/postconditions can't be declared there (see above) — only tool names, counts, and plain values are ever deserialized.
python benchmarks/correctness.pyA labeled dataset of 59 synthetic tool-call traces spanning all 7 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).
python benchmarks/latency.pyMeasured 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.
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 + CLIEverything above proves ACEL works against scripted tool calls. For the
stronger version — a real LLM in Claude Desktop or Claude Code actually
driving the tool calls, and ACEL blocking a mistake the model made itself —
see docs/TESTING_WITH_REAL_AGENTS.md.
It walks through wiring up examples/support_agent_server.py (a realistic
customer-support/refund scenario) and gives adversarial prompts designed to
actually trigger each contract.
MIT