Python Contracts

Define contracts in Python using decorators when you need logic that YAML operators cannot express.

Define contracts in Python using decorators when you need logic that YAML operators cannot express. Python contracts enforce identically to YAML contracts -- the pipeline does not distinguish between them.

Verdict

Every Python contract returns a Verdict:

from edictum import Verdict

# Contract passed -- tool call proceeds
Verdict.pass_()

# Contract failed -- tool call denied with reason
Verdict.fail("Destructive command denied: rm -rf /")

# With metadata (appears in audit events)
Verdict.fail("Trade exceeds limit", trade_id="T-1234", amount=5000)

Method	Returns	Description
`Verdict.pass_()`	`Verdict(passed=True)`	Contract passed
`Verdict.fail(message, **metadata)`	`Verdict(passed=False, message=..., metadata=...)`	Contract failed. Message is truncated to 500 chars.

Messages are sent to the agent -- make them specific and instructive so the agent can self-correct. "Denied" is bad. "Refunds over $500 require supervisor role. Current role: analyst." is good.

Preconditions

Preconditions run before tool execution. If one fails, the tool never runs.

from edictum import Edictum, Verdict, precondition

@precondition(tool="bash")
def no_destructive_commands(envelope):
    cmd = envelope.args.get("command", "")
    destructive = ["rm", "rmdir", "shred", "mkfs"]
    first_token = cmd.split()[0] if cmd.split() else ""
    if first_token in destructive:
        return Verdict.fail(f"Destructive command denied: {cmd}")
    return Verdict.pass_()

guard = Edictum(contracts=[no_destructive_commands])

Decorator signature

@precondition(tool="tool_name", when=optional_filter)
def my_contract(envelope: ToolEnvelope) -> Verdict:
    ...

Parameter	Type	Description
`tool`	`str`	Tool name to match. Use `"*"` for all tools.
`when`	`Callable \| None`	Optional filter: `when(envelope) -> bool`. Contract only runs if `True`.

The ToolEnvelope

The envelope parameter is a frozen dataclass with everything about the tool call:

Field	Type	Description
`tool_name`	`str`	Name of the tool being called
`args`	`dict`	Deep-copied tool arguments (safe to inspect, cannot mutate original)
`principal`	`Principal \| None`	Identity context (user_id, role, org_id, ticket_ref, claims)
`environment`	`str`	Deployment environment ("production", "staging", etc.)
`side_effect`	`SideEffect`	`PURE`, `READ`, `WRITE`, or `IRREVERSIBLE`
`bash_command`	`str \| None`	Extracted command string (set when `tool_name == "Bash"`)
`file_path`	`str \| None`	Extracted file path (set for file tools)
`metadata`	`dict`	Arbitrary per-call metadata
`call_id`	`str`	Unique UUID for this call
`call_index`	`int`	Sequential position in the session
`timestamp`	`datetime`	UTC timestamp

Async preconditions

Preconditions can be async:

@precondition(tool="deploy_service")
async def check_change_freeze(envelope):
    is_frozen = await ops_api.is_change_freeze(envelope.environment)
    if is_frozen:
        return Verdict.fail("Change freeze active. No deploys until lifted.")
    return Verdict.pass_()

Conditional execution with `when`

Skip the contract entirely for certain calls:

@precondition(tool="*", when=lambda env: env.environment == "production")
def prod_only_check(envelope):
    if not envelope.principal or not envelope.principal.ticket_ref:
        return Verdict.fail("Production changes require a ticket reference.")
    return Verdict.pass_()

Postconditions

Postconditions run after tool execution. They inspect the tool's output.

from edictum import postcondition, Verdict
import re

@postcondition(tool="*")
def check_pii_in_output(envelope, tool_response):
    text = str(tool_response)
    if re.search(r'\b\d{3}-\d{2}-\d{4}\b', text):  # SSN pattern
        return Verdict.fail("PII detected in output: SSN pattern found.")
    return Verdict.pass_()

guard = Edictum(contracts=[check_pii_in_output])

Decorator signature

@postcondition(tool="tool_name", when=optional_filter)
def my_contract(envelope: ToolEnvelope, tool_response: Any) -> Verdict:
    ...

The callback receives two parameters: the envelope and the raw tool response.

For PURE/READ tools, a postcondition failure injects retry context. For WRITE/IRREVERSIBLE tools, failures produce warnings only -- the action already happened.

Session Contracts

Session contracts enforce cross-turn limits using persisted counters. They must be async because all Session methods are async.

from edictum import session_contract, Verdict

@session_contract
async def limit_operations(session):
    count = await session.execution_count()
    if count >= 50:
        return Verdict.fail("Session limit reached (50 tool calls). Summarize and stop.")
    return Verdict.pass_()

guard = Edictum(contracts=[limit_operations])

Session methods

Method	Returns	Description
`await session.execution_count()`	`int`	Total tool executions (successful calls only)
`await session.attempt_count()`	`int`	Total attempts including denied calls
`await session.tool_execution_count(tool)`	`int`	Executions for a specific tool
`await session.consecutive_failures()`	`int`	Resets on success, increments on failure
`session.session_id`	`str`	Session identifier (property, not async)

Per-tool limits

@session_contract
async def limit_deploys(session):
    deploys = await session.tool_execution_count("deploy_service")
    if deploys >= 3:
        return Verdict.fail("Maximum 3 deploys per session.")
    return Verdict.pass_()

Built-in: deny_sensitive_reads()

A ready-made precondition factory that blocks reads of common secret files:

from edictum import Edictum, deny_sensitive_reads

# Use defaults
guard = Edictum(contracts=[deny_sensitive_reads()])

# Custom paths and commands
guard = Edictum(contracts=[
    deny_sensitive_reads(
        sensitive_paths=["/.private/", "/secrets/", "/.env"],
        sensitive_commands=["printenv", "env", "set"],
    )
])

Default denied paths (substring match)

/.ssh/, /var/run/secrets/, /.env, /.aws/credentials, /.git-credentials, /id_rsa, /id_ed25519

Default denied commands (exact match or with args)

printenv, env

The function checks both envelope.file_path and envelope.bash_command, so it catches cat ~/.ssh/id_rsa as a bash command even though it's not a file read tool.

Factory Pattern

Create parameterized contracts for reuse:

from edictum import precondition, Verdict

def make_require_target_dir(allowed_base: str):
    """Factory: deny mv/cp commands targeting paths outside allowed_base."""

    @precondition(tool="bash")
    def require_target_dir(envelope):
        cmd = envelope.args.get("command", "")
        tokens = cmd.split()
        if tokens and tokens[0] in ("mv", "cp") and len(tokens) >= 3:
            target = tokens[-1]
            if not target.startswith(allowed_base):
                return Verdict.fail(
                    f"Target '{target}' is outside {allowed_base}. "
                    f"Move files to {allowed_base} instead."
                )
        return Verdict.pass_()

    return require_target_dir

# Different base paths for different contexts
guard = Edictum(contracts=[
    make_require_target_dir("/tmp/organized/"),
])

Mixing Python and YAML

Load YAML contracts and add Python contracts to the same guard:

from edictum import Edictum, deny_sensitive_reads
from edictum.yaml_engine.loader import load_bundle
from edictum.yaml_engine.compiler import compile_contracts

# Load and compile YAML contracts
bundle_data, bundle_hash = load_bundle("contracts.yaml")
compiled = compile_contracts(bundle_data)

# Combine YAML + Python contracts in the constructor
guard = Edictum(
    contracts=compiled.preconditions
    + compiled.postconditions
    + compiled.session_contracts
    + [
        deny_sensitive_reads(),
        no_destructive_commands,
        limit_operations,
    ],
    limits=compiled.limits,
)

Python contracts are evaluated alongside YAML contracts in the pipeline. Preconditions run in registration order; the first denial wins.

Low-Level API

BashClassifier

Classifies bash commands by side-effect level using an allowlist:

from edictum import BashClassifier, SideEffect

BashClassifier.classify("ls -la")              # SideEffect.READ
BashClassifier.classify("git status")          # SideEffect.READ
BashClassifier.classify("rm -rf /tmp")         # SideEffect.IRREVERSIBLE
BashClassifier.classify("cat foo | grep x")    # SideEffect.IRREVERSIBLE (pipe operator)
BashClassifier.classify("echo $AWS_SECRET_KEY") # SideEffect.IRREVERSIBLE (bare $VAR)

READ allowlist: ls, cat, head, tail, wc, find, grep, rg, git status/log/diff/show/branch/remote/tag, echo, pwd, whoami, date, which, file, stat, du, df, tree, less, more

Any command containing shell operators (|, ;, &&, ||, >, >>, `, $(, $, etc.) is classified as IRREVERSIBLE regardless of the command itself. Bare $VAR expansions are treated identically to ${VAR} and $(cmd) — a command like echo $AWS_SECRET_KEY is IRREVERSIBLE even though echo is allowlisted.

This catches a specific exfiltration pattern: an allowlisted command passing a secret through its argument. This is a defense-in-depth measure for variable expansion — any $ token in the command triggers IRREVERSIBLE classification. Use sandbox contracts for broader security-critical path enforcement.

create_envelope()

Preferred factory for creating ToolEnvelope instances. Adapters call this internally -- you rarely need it directly:

from edictum import create_envelope, Principal

envelope = create_envelope(
    tool_name="read_file",
    tool_input={"path": "/etc/passwd"},
    run_id="session-123",
    call_index=0,
    principal=Principal(user_id="alice", role="analyst"),
    environment="production",
)

The factory deep-copies all arguments and metadata, validates the tool name (no null bytes, control characters, backslashes, or slashes), and classifies the tool's side effect via the registry or BashClassifier.

Direct construction (ToolEnvelope(tool_name=..., args=...)) is also permitted -- it validates tool_name with the same rules -- but does not deep-copy args. Prefer create_envelope() when deep-copy guarantees matter.

Testing Python Contracts

Use guard.evaluate() for synchronous dry-run testing without executing tools:

import pytest
from edictum import Edictum, Principal

def test_destructive_denied():
    guard = Edictum(contracts=[no_destructive_commands])
    result = guard.evaluate("bash", {"command": "rm -rf /"})
    assert result.verdict == "deny"
    assert "Destructive command denied" in result.deny_reasons[0]

def test_safe_command_allowed():
    guard = Edictum(contracts=[no_destructive_commands])
    result = guard.evaluate("bash", {"command": "ls -la"})
    assert result.verdict == "allow"

def test_role_gated():
    guard = Edictum(contracts=[prod_only_check])
    result = guard.evaluate(
        "deploy_service",
        {"env": "production"},
        principal=Principal(role="analyst"),  # no ticket_ref
        environment="production",
    )
    assert result.verdict == "deny"

For end-to-end tests with actual tool execution, use guard.run() with pytest.raises(EdictumDenied). See Adversarial Testing for red-team patterns.

Reference Implementation

The edictum-demo DevOps scenario contains a complete working example with Python contracts, including deny_sensitive_reads(), custom preconditions, factory patterns, and a governed vs unguarded comparison. See Industry Scenarios for details.

On this page