moltrust-enforce

Reference client for the MolTrust runtime check POST /enforce/check (constraint_mode = "enforce").

Thin and explicit: no decorator, no framework hook, no hidden middleware. Two methods, and the operator sees what happens in both. Framework-agnostic — where the call sits in your own code is the operator's decision.

The SDK checks mandates. It does not issue them: creating and signing mandates is not part of it.

0.5.0 breaks every digest — for the second time, for a different reason. The digested core carries no free text any more: neither the verdict's reason nor the one on each predicate. Both stay in the response — the reason remains readable — only outside the value two implementations have to hit byte for byte (AAE -02 §2.5.2/§2.5.3). Affected are the core_digest of verdict, ratification and AER decision; action_binding, mandate_digest and transaction_digest stay. How to tell: enforce_version, ratify_version and aer_version now read "3.0". A record from 0.4.0 or earlier can no longer be recomputed with this package. Details in the CHANGELOG.

Installation

pip install -e sdk/python              # recompute and verify only
pip install -e "sdk/python[client]"    # plus the HTTP client for POST /enforce/check

The base carries jcs and cryptography — exactly what the recompute path needs. The HTTP client sits in the extra client from 0.4.0 on; a third party who only checks a verdict installs 5 packages instead of 12. [verify] is an empty extra and exists so that pip install "moltrust-enforce[verify]" runs and the answer sits in the package itself: the verifier needs nothing beyond the base.

Change against 0.3.0: there httpx came along unconditionally. Anyone who touches EnforceClient after the upgrade without the extra gets no bare ModuleNotFoundError, but:

EnforceClient needs the HTTP client, which is not installed. It moved into an extra
in 0.4.0: pip install 'moltrust-enforce[client]'. Recomputing and verifying work without it.

Releases are published to PyPI; the latest version there is the current one. Each publish is its own human-approved step. The current release digests a core without free text — see the note above and the CHANGELOG.

Pattern 1 — believe the server

The simple way. One call, one verdict.

from moltrust_enforce import EnforceClient

client = EnforceClient("https://api.moltrust.ch", api_key=API_KEY)

verdict = client.check(mandate, transaction)

if verdict.permitted:
    execute(transaction)
else:
    log.warning("blocked: %s (%s)", verdict.verdict, verdict.reason)

verdict.permitted is true for PERMIT only. PENDING is not permission, DENY even less so.

Pattern 2 — recompute it yourself

The actual point. The verdict hangs on mandate and transaction alone — no server state, no clock, no database. Whoever holds both inputs recomputes it locally and does not have to believe the server.

verdict = client.check(mandate, transaction)
result = client.verify(verdict, mandate, transaction)

if not result.ok:
    # The server said something other than the inputs give.
    alert("enforce server disagrees with local recompute", result.mismatches)
    return                       # do not execute

if verdict.permitted:
    execute(transaction)

verify() checks three things: whether the response carries itself (core_digest matches the core shipped with it), whether the local evaluation yields the same digest, and whether the server names the same verdict as the local evaluation. Every deviation lands in result.mismatches.

It also works without a server at all — the kernel is public:

from moltrust_enforce import enforce_check
local = enforce_check(mandate, transaction)

Fail-closed

A PERMIT arises exclusively from a 200 response that was read and says PERMIT. Everything else is DENY:

Situation	Result
Server unreachable, timeout, DNS failure	DENY, from_server=False
HTTP 4xx/5xx	DENY, from_server=False
Response is not JSON / has the wrong shape	DENY, from_server=False
verdict value unknown	DENY, from_server=False
No valid mandate in the request	DENY (the server answers 200 with a DENY record)

For anyone who prefers to handle the failure as an exception:

client = EnforceClient(..., on_transport_error="raise")   # raises EnforceTransportError

Both settings are fail-closed. A third, permitting behaviour does not exist — no switch that turns an unreachable check into permission.

PENDING

check() returns PENDING unchanged. The SDK does not resolve it, and permitted stays False — a PENDING action never passes through here quietly.

The optional hook reports, it does not decide: its return value is ignored, the verdict stays PENDING.

def queue_for_approval(verdict):
    approvals.put(verdict.core_digest)          # report

client = EnforceClient(..., on_pending=queue_for_approval)

verdict = client.check(mandate, transaction)
if verdict.pending:
    return                                       # the operator has to act

Without the hook the same happens, only without the report: PENDING comes back, permitted is False, nothing is executed.

Mandate and transaction

A mandate carries grants. A grant binds to an action (action_binding), declares its fields (type_fields), holds constraints and a disposition:

from moltrust_enforce import action_digest

action = {"verb": "transfer", "asset": "USDC", "chain": "base"}

mandate = {
    "grants": [{
        "action_binding": action_digest(action),
        "type_fields": ["verb", "asset", "chain"],    # what the action consists of
        "disposition": "allow",                       # allow | hold | forbid
        "constraints": [
            {"type": "exact", "field": "to", "value": "0xABC…"},
            {"type": "enum",  "field": "region", "values": ["CH", "DE"]},
            {"type": "range", "field": "amount", "lo": 0, "hi": 1000},
        ],
    }],
}

transaction = {"action": action, "to": "0xABC…", "region": "CH", "amount": 500}

type_fields separates the action from its arguments. The action must be an object and carry exactly these keys — none missing, none extra. verb is mandatory. Recipient and amount stay siblings of the action and are checked through constraints; if the amount moves into the action, it becomes part of the digest and every payment would be a different action. Without type_fields the grant is invalid, and a mandate consisting of it alone carries nothing.

exact compares exactly — no prefix, no case folding, no normalization; a vanity address with the same beginning fails. enum compares every element exactly. range is a closed integer interval lo ≤ arg ≤ hi; floating-point numbers are rejected because they break recomputability.

PERMIT exists only when a grant matches via action_binding, all of its constraints hold and the disposition is allow. An action that no grant addresses is DENY and never PENDING. forbid takes precedence over a permitting grant.

AER — verdicts over live preconditions

From 0.4.0 on the SDK also evaluates constraints whose answer does not sit in the transaction: whether an authorization has been revoked, whether a recipient is on a sanctions list, which exchange rate applied to a fiat limit. Such facts live outside the decision and change; so that a third party can still recompute the verdict, every fact value is carried along as a signed statement with a validity window — an evidence item, packaged as a DSSE envelope (Dead Simple Signing Envelope, the signature format from the supply-chain world). All items of a decision sit in a bundle, the bundle has a hash bundle_commit, and that sits in the verdict record.

from moltrust_enforce import build_bundle, f_ext, verify_record

# Four constraint types point at the bundle instead of at the transaction.
mandate = {"grants": [{
    "action_binding": action_digest(action),
    "type_fields": ["verb", "asset", "chain"],
    "disposition": "allow",
    "constraints": [
        {"type": "exact", "field": "to", "value": "0xABC…"},
        {"type": "evidence_bool", "query": {"kind": "revocation", "subject": "aae:0f3a"},
         "expect": False},
        {"type": "evidence_enum", "query": {"kind": "jurisdiction", "subject": "0xABC…"},
         "values": ["CH", "DE"]},
        {"type": "evidence_scaled_range", "field": "amount", "rate_scale": 6,
         "query": {"kind": "fx", "pair": "USDC/EUR"}, "lo": 0, "hi": 500},
    ],
}]}

bundle = build_bundle(items, mandate, transaction, "2026-08-31T12:00:00Z")
record = f_ext(mandate, transaction, bundle)     # pure, without network and without clock

evidence_bool, evidence_enum and evidence_range compare the value from the bundle. evidence_scaled_range converts an amount from the transaction with a rate from the bundle and checks it against a limit: the rate sits as an integer in units of 10**rate_scale, and amount * rate is compared against limit * 10**rate_scale. 500 USDC minor units at a rate of 0.92 give 460 EUR minor units and hold under a limit of 500. The arithmetic runs without division and without rounding, because a floating-point intermediate step can give a different verdict depending on the platform.

Every evidence constraint additionally checks the window of its item against the decision_timestamp of the bundle. If an item for a question is missing, if the value has the wrong type, or if the timestamp lies outside the window, the result is DENY — the same fail-closed rule as in the static case. A mandate without evidence constraints gets the same verdict from f_ext as from enforce_check; tests/test_aer_ext_core.py holds that over a case corpus.

Recomputing without a server: moltrust-verify

The verifier receives record, bundle, mandate, transaction and a trust list — a file that says which sources the checking party believes. It opens no connection and reads no clock:

moltrust-verify --input decision.json --trust-list sources.json

V1 PASS bundle commit and input binding hold
V2 PASS every item carries a signature from a trusted source
V3 PASS every item window covers the decision timestamp
V4 PASS recomputed PERMIT from the same inputs

PASS — recomputed verdict PERMIT

A finished decision to try out sits in examples/aer/ — decision.json, trust.json and the script that produces both.

V1 checks that the commit matches the bundle content and that bundle and record mean the same mandate and the same transaction. V2 checks per item an Ed25519 signature over the DSSE PAE against a key from the trust list. V3 checks per item the validity window against the decision timestamp. V4 recomputes f_ext and compares core_digest and verdict. Exit code 0 means all four hold; 1 means at least one fails; 2 means the input was not readable in the first place. The same checks as a library: verify_record(record, bundle, mandate, transaction, trust_list).

What this establishes: the operator used exactly this evidence, it was valid at the decision timestamp, and exactly this verdict follows from it. What stays open: whether a named source told the truth. Whoever holds the key of a listed source can sign a wrong value inside the window, and a fact can change within a valid window — against that a short window helps, or a fresh fetch immediately before execution. Trust is thereby moved onto named, auditable sources and not removed.

The SDK checks evidence and does not issue any. Signing source adapters do not belong in the package; the trust list is brought by the checking party, because a verifier that first had to resolve keys online would be no offline verifier.

The check path also loads no HTTP stack: EnforceClient arrives on access (PEP 562), so import moltrust_enforce.cli pulls neither httpx nor socket or ssl into the process. Loaded are jcs and cryptography. tests/test_aer_verify.py measures that on the process and not on the source text. Without the extra client, httpx is not installed in the first place.

Coupling to the server signature

The SDK is coupled to the signature from PR #306:

• Request {"mandate": …, "transaction": …, "prev_core_digest": "sha256:<64 hex>"|null} plus, for a delegated mandate, an optional "ancestors": [<parent mandate>, …] (root first). Each hop is checked for grant attenuation and recorded as a grant_attenuation predicate; without ancestors the core is unchanged.
• Response {"verdict", "reason", "grant_index", "trace", "record": {"core", "core_digest"}}

src/moltrust_enforce/_core.py is an unchanged copy of app/enforcement/enforce_check.py. Exactly one line differs: the import of the JCS canonicalization points directly at jcs here instead of at app.signature, because the SDK has to work without the server package. tests/test_core_parity.py checks both — that no second line differs, and that both versions deliver the same digests over a case corpus.

If the signature changes, the SDK has to follow.

The gate — require_moltrust

A different job from the rest of this package. The mandate path asks may this action proceed; the gate asks who is calling, and what does MolTrust say about them — at the door of your own paid endpoint, with no call to us in the request path.

from moltrust_enforce.gate import require_moltrust, load_jwks

JWKS = load_jwks("/etc/moltrust/jwks.json")
gate = require_moltrust(min_score=60, jwks=JWKS)

decision = gate(request.method, request.url.path, request.headers)
if not decision.allowed:
    return JSONResponse({"error": decision.reason, "detail": decision.detail},
                        status_code=403)

The caller sends three headers: X-MolTrust-Attestation (the gate_attestation field from GET /skill/trust-score/<did>), X-MolTrust-Timestamp, and X-MolTrust-Proof — an Ed25519 signature over method, path, DID and timestamp, made with the DID's own key.

The attestation carries the DID's public key, which is what makes the offline check possible at all: did:moltrust:<hex> is a random identifier, not a hash of a key, so without a signed statement binding the two you could verify that a score exists for some DID and never that the caller is that DID.

A discount for verified agents

Two prices on one route. Nothing is refused — the gate sets the price.

verified = require_moltrust(min_score=60, jwks=JWKS)

@app.get("/guard/api/agent/score/{address}")
async def agent_score(address: str, request: Request):
    d = verified("GET", request.url.path, request.headers)
    price = "0.05" if d.allowed else "0.20"
    return await charge_and_answer(address, price, caller=d.did)

And the other shape, where the credential is the requirement and the score is not consulted at all:

auditors = require_moltrust(credential_type="SkillAuditCredential", jwks=JWKS)

Deny by default

Every path that is not an explicit allow returns a denial, including a malformed header and an unknown key id. Each denial names a reason (attestation_missing, attestation_invalid, proof_invalid, score_withheld, score_below_minimum, track_record_invalid, credential_missing, …) so a caller can fix it without asking you what happened.

A withheld score is a denial. A score we have not computed is not a low score and it is not a pass. allow_withheld=True lets those agents through, which is a reasonable choice for a discount tier and a bad one for a spend authorisation. It does not bypass min_score: a withheld score is None, so a numeric threshold still denies.

How an agent qualifies

A score stays withheld until three endorsers exist, and an agent that registered this morning has none. allow_track_record=True is the way past that, and it is off by default. An agent gets there by registering, binding a Base wallet (GET /identity/nonce, then POST /identity/bind), and issuing one credential over it with POST /credentials/track-record — one call. Anchoring runs in a batch every two hours, and the anchor shows up in GET /skill/trust-score/<did>; that is what to poll. Calling the issuing endpoint again inside a week with nothing changed returns the same credential and answers reused: true.

The wallet clears two published thresholds: at least one transaction it sent itself, and at least seven days of age. They are a cost rather than a quality bar — a wallet with its own history cannot be produced on demand. The first one bites: agent marketplaces relay gaslessly, so a busy worker wallet can still sit at nonce 0.

With the option on, a well-formed track_record in the attestation stands in for the score, and min_score is not consulted for that caller. decision.via says which requirement carried the allow. The substitute covers a score nobody computed, never one that came out low, and it appears only after the anchoring batch has run.

Replay

The proof is fresh within max_age_seconds (default 300). Inside that window the same proof can be presented twice unless you pass seen, a callable that records a proof and returns False if it has seen it before. Without it the gate is replay-resistant, not replay-proof.

Keeping the key set current

load_jwks reads a file or a dict and never fetches. Refreshing is an operational step on your schedule:

curl -fsS https://api.moltrust.ch/.well-known/jwks.json > /etc/moltrust/jwks.json.new \
  && mv /etc/moltrust/jwks.json.new /etc/moltrust/jwks.json

A rotated key surfaces as attestation_invalid with no key for kid …, which is the one denial that means refresh the file rather than blame the caller.

The Node port is @moltrust/x402; both verify the same attestation and the same binding string, and their test suites assert the same denials.

Amazon Ads — moltrust_enforce.adapters.amazon_ads

A gate in front of an agent's own Amazon Ads MCP calls. Your code keeps the Amazon credentials and makes the call; MolTrust only answers allow, hold or forbid. Nothing in this package talks to Amazon.

from moltrust_enforce import EnforceClient
from moltrust_enforce.adapters.amazon_ads import (
    DOCUMENTED_TOOLS, AdsProfile, AmazonAdsGuard, Forbidden, HoldRequired)

profile = AdsProfile(
    mandate_id="agency-acme-2026q4",
    principal={"did": AGENCY_DID, "public_key": AGENCY_PUBKEY_HEX},
    tools=dict(DOCUMENTED_TOOLS),          # or a narrower pick
    account_ids=("ENTITY1ABC",), brands=("Acme Outdoor",),
    daily_budget_cap=50_000, hold_threshold=20_000,     # cents: 500.00 / 200.00
    total_daily_budget_cap=200_000, period_budget_cap=1_500_000,
    max_bid=500, max_bid_change_bp=2_500,               # 5.00, +/- 25 %
    valid_from="2026-10-01T00:00:00Z", valid_until="2026-12-31T23:59:59Z")

guard = AmazonAdsGuard(EnforceClient("https://api.moltrust.ch", api_key=KEY), profile)

try:
    guard.call(tool, args, my_amazon_call, context=my_account_state)
except HoldRequired as e:   # above the hold threshold: a person approves via /enforce/ratify
    queue_for_approval(e.outcome)
except Forbidden as e:      # outside the mandate; the reason names the limit
    log(e.outcome.reason)

context is your own view of the account (current budgets, current bids, brand per campaign); values the adapter cannot derive count as unknown and end in forbid. One call that changes several campaigns is checked per campaign, and the strictest verdict decides. Encoding and the Amazon facts behind it: docs/spec-fakten/aae-ads-profile.md in the server repository. Eight scenarios with verifiable AER records: python examples/amazon_ads/demo.py --out demo-output.

Tests

cd sdk/python && pip install -e ".[test]" && pytest

License

MIT