Python Expect DSL
Use expect() to describe what an agent should do, then evaluate the chain to check what actually happened. Start with deterministic checks of outputs, tool calls and effects; add an embedding or judge only when the answer needs a semantic check.
from attest import TraceBuilder, expect
def test_greeting(attest): trace = TraceBuilder("greeter").set_output(message="Hello").build() attest.evaluate(expect(trace).output_contains("hello"))Creating and Evaluating a Chain
Section titled “Creating and Evaluating a Chain”expect(result: AgentResult | Trace) -> ExpectChain accepts an agent result or a raw trace. A raw trace is wrapped in an AgentResult. Assertion methods collect checks; they do not evaluate them.
| Fixture method | Behavior |
|---|---|
attest.evaluate(chain, *, budget=None) | Evaluate and enforce the outcome; returns an AgentResult if the test continues |
attest.inspect(chain, *, budget=None) | Evaluate and return an AgentResult without enforcing its assertion outcome |
attest.evaluate_async(chain, *, budget=None) | Async evaluate and enforce |
attest.inspect_async(chain, *, budget=None) | Async inspect, safe within an existing event loop |
attest.enforce(verdict) | Enforce an inspected result or an aggregate trial verdict |
budget is an optional cumulative session-cost limit in USD; exceeding it can still fail an inspect() call. This is separate from the engine’s model-call budgets in Configuration.
Any hard_fail fails the test. Otherwise any not_evaluated skips it (fails with --attest-strict); otherwise it passes, including soft failures. inspect() is useful for collecting trials before enforcing their aggregate result. During a --attest-trials trial, enforcement is deferred to the aggregate verdict.
In 1.1.0, leaving a chain unevaluated fails an otherwise passing test. This includes chains created in fixtures and assertions appended after evaluation. Evaluate the final chain, or use inspect() and explicitly check/enforce the result. chain.evaluated is true only when its current assertion list has been submitted. chain.assertions returns a copy of the list; chain.trace returns the underlying trace. Evaluation assigns stable assertion IDs (scheme 2) from the test scope, type, specification and occurrence, not the position among unrelated checks.
Layer 1: Schema
Section titled “Layer 1: Schema”Schema methods return the chain and do not accept soft.
| Method | Target |
|---|---|
output_matches_schema(schema: dict) | output.structured, not the whole output object |
output_field_matches_schema(field: str, schema: dict) | output.<field>; supports a path such as structured.items |
tool_args_match_schema(tool_name: str, schema: dict) | Named steps’ arguments |
tool_result_matches_schema(tool_name: str, schema: dict) | Named steps’ results |
def test_structured_output(attest): trace = TraceBuilder().set_output(structured={"answer": 42}).build() attest.evaluate(expect(trace).output_matches_schema({ "type": "object", "required": ["answer"], "properties": {"answer": {"type": "integer"}}, }))Layer 2: Constraints
Section titled “Layer 2: Constraints”These methods return the chain. Each accepts keyword-only soft: bool = False in addition to the parameters shown.
| Method | Parameters | Check |
|---|---|---|
cost_under | max_cost: float | metadata.cost_usd <= max_cost |
latency_under | max_ms: int | metadata.latency_ms <= max_ms |
tokens_under | max_tokens: int | metadata.total_tokens <= max_tokens |
tokens_between | min_tokens: int, max_tokens: int | Inclusive token range |
step_count | operator: str, value: int | Number of steps |
tool_call_count | operator: str, value: int | Number of tool_call steps |
constraint | field: str, operator: str, value: float | None = None, *, min=None, max=None | Generic numeric constraint |
Use eq, lt, lte, gt, or gte with a value; use between with constraint(..., min=..., max=...).
chain = (expect(trace) .cost_under(0.10) .latency_under(5000) .constraint("metadata.total_tokens", "between", min=100, max=500))# Inside a test: attest.evaluate(chain)Missing metadata is not a measured zero: populate the metrics your constraints need.
Layer 3: Trace Checks
Section titled “Layer 3: Trace Checks”The original trace checks match all step names, including LLM and retrieval steps, not just tool calls. For contracts that must count only tool calls, use the behavior methods below. Each method accepts keyword-only soft=False.
| Method | Parameters | Check |
|---|---|---|
tools_called_in_order | tools: list[str] | Names appear as an ordered subsequence |
tools_called_exactly | tools: list[str] | Names appear as one contiguous subsequence; other steps may occur before or after |
required_tools | tools: list[str] | All names appear |
forbidden_tools | tools: list[str] | No listed name appears |
no_tool_loops | tool: str, max_repetitions: int = 1 | Bounds the total occurrences of a name, not just consecutive repetitions |
no_duplicate_tools | No positional arguments | No step name repeats |
Behavior Contracts
Section titled “Behavior Contracts”Behavior contracts examine only tool_call steps and count retries as calls. Each returns the chain and accepts keyword-only soft=False.
| Method | Parameters | Check |
|---|---|---|
tool_cardinality | tool: str, *, per: str | None = None, at_most: int | None = None, at_least: int | None = None | Bound calls overall or per argument value; at least one bound required |
trajectory | tools: list[str], *, mode="strict" | strict: same sequence; unordered: same calls in any order; subset: no calls outside the list; superset: every listed call made |
trajectory_partial_order | order: list[tuple[str, str]], *, tools: list[str] | None = None | Require the named tools and each before/after pair |
tool_called_before | before: str, after: str | Every after call has an earlier before call; passes if after is absent |
no_tool_cycles | period: int, *, min_repeats: int = 2 | No consecutive repeated block of 1–4 calls |
max_tool_failure_streak | max_streak: int | Bound consecutive errors; needs trace-v2 error evidence |
tool_arg_satisfies | tool: str, arg: str, op: str, value: Any | Every matching call satisfies a dot-path argument predicate; passes if the tool is absent |
Argument operators are eq, ne, lt, lte, gt, gte, in, and not_in.
def test_refund_calls(attest): trace = (TraceBuilder("refund") .add_tool_call("check_eligibility", args={"order_id": "O-1042"}) .add_tool_call("issue_refund", args={"order_id": "O-1042", "amount_cents": 500}) .set_output(message="Refund issued").build()) attest.evaluate(expect(trace) .tool_cardinality("issue_refund", per="order_id", at_most=1) .tool_called_before("check_eligibility", "issue_refund") .tool_arg_satisfies("issue_refund", "amount_cents", "lte", 500))Pair a vacuously passing predicate with tool_cardinality(..., at_least=1) when a call must occur. Missing required argument/error evidence produces not_evaluated: missing_evidence, not proof that the contract passed.
Layer 3: World and Effect Contracts
Section titled “Layer 3: World and Effect Contracts”Tool calls show attempts; a world snapshot shows the state the agent left behind. chain.world(snapshot) returns WorldChecks; each check below adds an effect assertion and returns the parent chain. snapshot is world.state() or the result of world.stop() from the world helper. Passing None yields not_evaluated: missing_evidence.
| Method | Signature after world(snapshot) |
|---|---|
count | (resource: str, /, *, equals=None, at_most=None, at_least=None, where=None, **filters) |
unique | (resource: str, /, *, by: str | list[str], where=None, **filters) |
idempotent | (resource: str, /, *, key="idempotency_key", where=None, **filters) |
sum | (resource: str, field: str, /, *, at_most: int | float | Related, where=None, **filters) |
matches | (resource: str, field: str, target: Related, /, *, where=None, **filters) |
count needs equals or at least one of at_most / at_least; equals cannot be combined with either. unique checks distinct values of one or several fields; idempotent checks distinct idempotency-key values. Keyword filters and where are exact matches; duplicate filter names are an error. These effect methods do not accept soft.
related(resource: str, field: str, *, on: str) names the corresponding record in another resource, joined by on. With a numeric bound, sum checks the filtered total. With related(...), it groups by the join key and checks each group’s total against the related record’s field. matches compares every selected record’s field to the corresponding record’s field.
from attest import related
def assert_refund_effects(attest, trace, snapshot): # Pass the snapshot returned by your world's state() or stop(). chain = expect(trace) chain.world(snapshot).count("refunds", equals=1, order_id="O-1042") chain.world(snapshot).unique("refunds", by="idempotency_key") chain.world(snapshot).idempotent("refunds") chain.world(snapshot).sum("refunds", "amount_cents", at_most=related("orders", "amount_cents", on="order_id")) chain.world(snapshot).matches("notifications", "customer_id", related("orders", "customer_id", on="order_id")) attest.evaluate(chain)A None snapshot is not evaluated. Invalid snapshots, absent resources, non-numeric sum evidence and ambiguous related keys fail rather than pass; a missing corresponding record is a relational contract violation. Empty selected records can satisfy a bound without proving a record exists; add a count check when existence matters. Each check embeds the snapshot; the SDK rejects chains whose repeated snapshot payloads exceed 9 MiB.
Layer 4: Content
Section titled “Layer 4: Content”Output helpers target output.message. Case-insensitive matching is the default. All return the chain.
| Method | Signature |
|---|---|
output_contains | (value: str, *, case_sensitive=False, soft=False) |
output_not_contains | (value: str, *, case_sensitive=False, soft=False) |
output_matches_regex | (pattern: str, *, soft=False) |
output_has_all_keywords | (keywords: list[str], *, case_sensitive=False, soft=False) |
output_has_any_keyword | (keywords: list[str], *, case_sensitive=False, soft=False) |
output_forbids | (terms: list[str]); no soft option |
content_contains | (target: str, value: str, *, case_sensitive=False, soft=False) |
chain = (expect(trace) .output_contains("refund") .output_forbids(["password", "ssn"]) .content_contains("output.message", "issued"))Layers 5–6: Semantic Checks
Section titled “Layers 5–6: Semantic Checks”| Method | Signature |
|---|---|
output_similar_to | (reference: str, *, threshold=0.8, model=None, soft=False) |
passes_judge | (criteria: str, *, rubric="default", threshold=0.8, model=None, soft=False) |
Both target output.message. threshold accepts a numeric score or "dynamic"; model overrides the provider model. Layer 5 needs an embedding provider; Layer 6 needs a judge. ONNX embeddings are local; hosted calls can cost money. Configure providers and request/USD limits in Configuration.
chain = (expect(trace) .output_similar_to("The refund was issued", threshold=0.85) .passes_judge("The answer clearly explains the refund", rubric="default"))An abstaining judge, malformed reply, rejected credential or provider failure is not_evaluated, never an invented passing score. Only transient failures are retried. ATTEST_SIMULATION=1 still evaluates deterministic checks but marks Layers 5–6 not_evaluated: simulation. Tests skip unevaluated assertions unless strict; the default report policy only warns for not_evaluated, so set not_evaluated: block in policy if CI must reject it.
With threshold="dynamic", the engine uses the newest 50 passing or soft-failing scores for the stable assertion ID. At least 10 are required; the bar is max(0.8, mean − 2·max(σ, 0.02)), otherwise 0.8. After upgrading to 1.1.0, older ordinal-ID baselines do not match the new IDs and must rebuild. See Drift Detection.
Trace Tree Checks (Layer 3)
Section titled “Trace Tree Checks (Layer 3)”These checks use nested agent_call steps. All accept keyword-only soft=False.
| Method | Parameters |
|---|---|
agent_called | agent_id: str |
delegation_depth | max_depth: int (root depth is 0) |
agent_output_contains | agent_id: str, value: str, *, case_sensitive=False |
cross_agent_data_flow | from_agent: str, to_agent: str, field: str |
follows_transitions | transitions: list[tuple[str, str]] |
aggregate_cost_under | max_cost: float |
aggregate_tokens_under | max_tokens: int |
agent_ordered_before | agent_a: str, agent_b: str |
agents_overlap | agent_a: str, agent_b: str |
agent_wall_time_under | agent_id: str, max_ms: int |
ordered_agents | groups: list[list[str]] |
Ordering and overlap need recorded step timestamps. See TraceBuilder and TraceTree for construction and aggregation.
Plugin Assertions
Section titled “Plugin Assertions”plugin(plugin_id: str, config: dict | None = None, *, soft=False) invokes a registered assertion plugin. The built-in plugin helpers aggregate_latency_under(max_ms: int, *, soft=False) and all_tools_called(tool_names: list[str], *, soft=False) check summed trace-tree latency and tools across the whole tree.
chain = (expect(trace) .aggregate_latency_under(5000) .all_tools_called(["search", "summarize"]) .plugin("my_plugin", {"threshold": 0.9}))See Writing a Plugin for the plugin contract.
Soft Failures and Diagnostics
Section titled “Soft Failures and Diagnostics”Only methods that expose soft accept it. Schema methods, effect methods and output_forbids do not. A soft_fail is recorded but does not itself fail the test; it does not turn missing evidence or an unavailable provider into a pass. The session soft-failure limit can still fail a batch.
Use the returned AgentResult’s outcome(), hard_failures, soft_failures, not_evaluated and render_diagnostics() for custom reporting. passed is stricter than the test verdict: it is true only when every assertion passed.