EngineManager

EngineManager starts and stops the engine subprocess that evaluates your checks. Pytest normally manages it for you; use it directly when building a custom integration. Use AttestClient for concurrent requests rather than adding another reader to the engine’s stdout.

import asyncio
from attest.engine_manager import EngineManager


async def main():
    async with EngineManager() as engine:
        response = await engine.send_request("evaluate_batch", {
            "trace": {"trace_id": "greeting", "output": {"message": "Hello"}, "steps": []},
            "assertions": [{"assertion_id": "greeting-check", "type": "content",
                "spec": {"target": "output.message", "check": "contains", "value": "hello"}}],
        })
        print(response["results"][0]["status"])


asyncio.run(main())

Constructor

EngineManager(engine_path: str | None = None, log_level: str = "warn")

engine_path selects an explicit binary; otherwise discovery runs. log_level is passed to the engine (debug, info, warn, or error).

Binary Discovery

resolve_engine(explicit_path: str | None = None, *, allow_download: bool = True) -> str returns the resolved path. The first binary found must report exactly ENGINE_VERSION, 1.1.0 for this SDK. A mismatch raises EngineVersionMismatchError; discovery does not skip it.

Priority	Source
1	Explicit path; a missing file raises FileNotFoundError
2	ATTEST_ENGINE_PATH; a missing file raises FileNotFoundError
3	Executable attest-engine on PATH (attest-engine.exe on Windows); empty PATH entries skipped
4	Version-specific ~/.attest/bin/<ENGINE_VERSION>/ cache
5	Automatic release download, unless disabled

ATTEST_ENGINE_NO_DOWNLOAD=1 (also true or yes) disables download. Neither a monorepo/local bin/ directory nor the legacy unversioned cache is searched.

Download Verification

Published SDK packages carry engine-checksums.json. Downloads must match that package-shipped SHA256; the release’s downloaded checksum file is not trusted. The binary is atomically installed into the version-specific cache. Downloads require HTTPS (including redirects), allow at most five redirects, and have a 10-second connect / 120-second total deadline. Source/editable installs without the checksum table cannot auto-download: build the engine and set ATTEST_ENGINE_PATH.

Supported release assets:

OS	Architecture	Asset
macOS	ARM64	attest-engine-darwin-arm64
macOS	AMD64	attest-engine-darwin-amd64
Linux	ARM64	attest-engine-linux-arm64
Linux	AMD64	attest-engine-linux-amd64
Windows	ARM64	attest-engine-windows-arm64.exe
Windows	AMD64	attest-engine-windows-amd64.exe

Lifecycle and Requests

All methods below are async unless described as a property or synchronous method.

API	Behavior
start(env_overrides: Mapping[str, str] | None = None) -> InitializeResult	Spawn, send initialize, check version/capabilities; overrides affect the child environment only. A failed initialization stops the process before propagating the error.
restart(env_overrides: Mapping[str, str] | None = None) -> InitializeResult	Kill the old process and start a fresh engine; used for timeout recovery
stop() -> None	Release an attached client’s stdout reader, request shutdown, and stop the process; kill if termination exceeds five seconds
send_request(method: str, params: dict[str, Any]) -> Any	Send a request and return the result with the matching request ID
is_running: bool	Property: subprocess is alive
spawn_env: Mapping[str, str]	Property: environment used for the current engine; do not log it because it may contain credentials
attach_client(client: AttestClient) -> None	Synchronous: associate the single client that owns stdout

An async context manager calls start() on entry and stop() on exit. Calling a non-initialize request before startup raises RuntimeError. Once a client is attached, send_request() delegates to that client’s correlated request loop; attaching a second different client raises RuntimeError.

Without an attached client, the manager uses the startup/shutdown handshake reader, which drops responses whose IDs do not match. Do not issue concurrent direct handshake requests. AttestClient supplies request correlation, timeout-triggered restart, and conservative model-budget accounting for normal operation.

InitializeResult

Field	Type	Meaning in 1.1.0
engine_version	str	"1.1.0"
protocol_version	int	1
capabilities	list[str]	Accepted features and configured-provider markers
missing	list[str]	Requested but unavailable capabilities
compatible	bool	Required capabilities are present
encoding	str	"json"
max_concurrent_requests	int	Advertised limit 1
max_trace_size_bytes	int	10485760 (10 MiB)
max_steps_per_trace	int	10000

embedding / llm_judge capabilities mean the assertion types are accepted, not that providers are configured; inspect embedding_configured / judge_configured. Incompatible required capabilities raise RuntimeError. See JSON-RPC Protocol for wire details.

Pytest Integration

The session-scoped attest_engine fixture handles engine lifecycle; the attest fixture provides evaluate, inspect, async variants, and enforce.

Option	Default	Purpose
--attest-engine PATH	Discovery	Select binary
--attest-log-level LEVEL	warn	Engine logging
--attest-tier N	All	Select tests tagged at or below the tier
--attest-budget USD	Unlimited	Cumulative session-cost limit
--attest-cost-report	Off	Session cost summary
--attest-strict	Off	Fail rather than skip not-evaluated tests (also ATTEST_STRICT)
--attest-report PATH	Off	Report-v2 JSON, including every test outcome
--attest-diagnostics	Off	Expanded assertion diagnostics
--attest-trials N	1	Fixed-N trials for synchronous tests using attest
--attest-min-pass-rate RATE	1.0	Minimum evaluated-trial pass rate
--attest-egress block	ATTEST_EGRESS, else allow	Best-effort guard against provider connections from test code

Environment and Versions

ATTEST_ENGINE_TIMEOUT is the SDK response timeout in seconds (default 30). On an ordinary client timeout the engine restarts and pending requests fail. ATTEST_SIMULATION=1 still starts the engine and evaluates deterministic assertions, including trace trees and effects; Layers 5–6 return not_evaluated: simulation. It does not prevent the agent itself from calling providers.

from attest import __version__, ENGINE_VERSION

assert __version__ == ENGINE_VERSION == "1.1.0"

See Configuration for all provider, budget, cache and egress settings.