zentest-sdk

The Python SDK for Zentest, Nxzen's platform for testing AI apps (chatbots, RAG, voice agents). Run your app's test suite in your own process, or let the platform call its HTTP endpoint, and gate the release on a judged verdict with a confidence interval. The behaviour is fixed by the SDK contract v1 (SDK-01 to SDK-20). A TypeScript SDK for Node 18+ implements the same contract in sdk-ts/ (contract §2b); offline runs and the CLI are Python only. You need a Zentest workspace and an API token (console: Settings → API tokens).

Links: contract · changelog · CI gate guide · span format · examples

Install

Python 3.11 or later.

pip install zentest-sdk                 # core: httpx only
pip install "zentest-sdk[otel]"         # + OpenTelemetry exporter (zentest.otel)
pip install "zentest-sdk[yaml]"         # + YAML suite files for the CLI
pip install "zentest-sdk[schema]"       # + offline runs of a format metric with a JSON Schema
pip install "zentest-sdk[judge]"        # + the local Claude judge (open-core mode, no platform)

Install it into a virtual environment, then check it: zentest --version prints the installed version. Add only the extras you need.

Working on the SDK itself? Install it from a checkout of the repository instead: pip install -e sdk (extras as pip install -e "sdk[otel,yaml]", and sdk[dev] for the tests).

Dependencies (SDK-20). The client is installed into your application's environment, so the floor is thin: httpx and nothing else. Evaluation libraries stay on the platform; the deterministic checks used by offline runs are standard library only. Extras: otel (opentelemetry-sdk), yaml (pyyaml), schema (jsonschema). The package ships type hints (py.typed).

Evaluate in your own tests (0.4.0)

from zentest import LLMTestCase, assert_test
from zentest.metrics import PIILeak, Faithfulness, AnswerRelevancy, GEval

def test_burst_pipe():
    case = LLMTestCase(input="Is a burst pipe covered?", actual_output=app(q), retrieval_context=chunks)
    assert_test(case, [PIILeak(), Faithfulness(threshold=0.8), AnswerRelevancy(),
                       GEval("Plain English", "Short answer a customer understands")])

• pytest prints a Zentest summary table; --zentest-sync --zentest-product P --zentest-version V records the session as a platform run. Deterministic metrics always run before judge metrics.
• Judges: ZENTEST_JUDGE=platform (the platform's calibrated judge, needs ZENTEST_API_KEY), local (Claude with ANTHROPIC_API_KEY, or OpenRouter with OPENROUTER_API_KEY) or heuristic (no model call). A local judge is not calibrated: gate releases on the platform.
• zentest init then zentest test zentest.yaml: the same checks as a YAML file in Git (--baseline, --junit, --sync).
• zentest redteam --product P --version V --agent app:answer: the product's red-team suite against your agent.
• zentest.guard.Guard(...).protect(answer): block prompt injection and personal data in production.
• zentest.simulate(answer, persona=..., goal=...): a seeded multi-turn conversation, scored per turn.

A worked example on a real app lives in the repository under sdk/examples (it runs with no network in mock mode).

Quick start (release gate)

from zentest import Zentest, AgentInput, AgentResponse

def agent(x: AgentInput) -> AgentResponse:        # or (user_message: str) -> str, or (messages: list[dict]) -> str
    reply, chunks = my_app.answer(x.messages[-1]["content"])
    return AgentResponse(output=reply, retrieval_context=chunks,
                         model="claude-sonnet-4.5", prompt_version="r12", config_version="v6")

with Zentest(api_key="zt_live_…", base_url="http://localhost:8620", region="uk-south") as zt:
    v = zt.gates.run("Claims Copilot", "v6", "Claims FAQ Golden Set", agent, repeats=3, deterministic=True)
print(v.passed, v.reason, v.interval)

Arguments

• repeats: required, with no default. A single run gives no interval.
• deterministic=True: you are stating that the app runs with temperature 0 and a fixed seed. Without it the platform refuses (PrerequisiteError).
• agent=None: the platform calls the product's HTTP connection instead of your code (the endpoint runner, below).
• batch_size=50, flush_every_s=None: results go to the platform every batch_size trials, or sooner once flush_every_s seconds pass without a submission, so a slow app's run shows progress and time left in the console as it goes. None takes the client's (Zentest(flush_every_s=...), else ZENTEST_FLUSH_EVERY_S, else 30 s); 0 sends full batches only.
• packs=[...]: compliance pack ids from the workspace's catalogue (eu-ai-act, owasp-llm, …; any other id is a 422), recorded on the run and returned as v.packs.
• connection_id="…" (with agent=None): the HTTP connection the platform calls when the environment has several; without it, the environment's default connection, else its only one.
• corpus=[...], corpus_data_source="synthetic" | "masked_production" | "production": your app's retrieval corpus (chunks as strings or {"text", "id"}), scanned when the run starts. A test case found in it is skipped and reported on v.contamination; too many and the run is refused (UK-31).
• osa_scope={...}: the three Online Safety Act scope conditions for this release; a change comes back as zt.gates.last_run.scope_alert (UK-10). trigger="ci" marks a run started from a pipeline.
• timeout=600: seconds to wait for the verdict before ZentestTimeout.

Version attribution (contract §2a): return AgentResponse(output=…, model=…, prompt_version=…, config_version=…) from a shape C agent so every trial says what produced it. The values must equal what is registered on the version under test. Attribution is required by default: on a product created from P2a on, a trial that reports nothing does not count, and a run where nothing is reported does not pass (the verdict's reason says what to report). Products that predate it keep counting such trials, as declared, until the owner turns attribution required on; v.attribution["coverage"] shows the share the app reported. usage and metadata on the response are stored with each trial too (credential-named keys such as api_key, accessToken or clientSecret as [REDACTED], personal data redacted like any output). They are optional and never fail a trial: nested usage (OpenAI's prompt_tokens_details) is sent flat as prompt_tokens_details.cached_tokens, at most 32 values, and metadata over 16 KB as JSON is trimmed key by key; each cut is a warning on zt.gates.last_run.warnings.

Tracing (SDK-15): @zt.traced / with zt.span(name, kind=...) record steps (tool calls, retrievals, LLM calls, planning) as zentest.span.v1 spans. The SDK runner uploads one trace per trial and links it to the trial's result (zt.results(run) gives trace_id and session_id; a trial whose spans the platform did not store has no trace_id, and the run handle says which). Without spans the run still works, but the SDK warns when the run starts that trajectory and tool-choice metrics are unavailable, and zt.gates.last_run.traced is False.

@zt.traced(kind="tool", name="lookup_claim")
def lookup(claim_id: str) -> dict: ...

with zt.span("answer", kind="llm", region="uk-south") as s:
    s.set_usage(input_tokens=120, output_tokens=30, provider="anthropic")
    s.set_output(text)

Already on OpenTelemetry (SDK-16): keep your instrumentation (GenAI gen_ai.* or OpenInference) and add zentest.otel.install(zt, product="Claims Copilot", version="v6"). Spans started inside a trial are stamped with the trial's ids. A collector can instead export to POST /v1/traces with the API token and X-Zentest-Product (or the resource attribute zentest.product), as OTLP/JSON only: set OTEL_EXPORTER_OTLP_TRACES_PROTOCOL=http/json (protobuf, the exporters' default, gets 415). /v1/traces is outside /api, so a proxy or ingress needs a route for it; see zentest-span-v1 §4a.

Endpoint runner (no code on your side): the platform calls the product's HTTP connection (Connections in the console, per environment). For the mock app:

Manual loop: run = zt.start(...), then zt.submit(run, case, output) for each case in zt.cases(run), then zt.finish(run). zt.cancel(run) stops a run and returns what completed (trials_received, kept).

Catalogue and authoring: zt.products(), zt.versions(product), zt.suites(product), zt.suite_cases(suite_id), zt.add_cases(suite_id, [...]), zt.update_case(case_id, input=…), zt.results(run), zt.span_schema(), zt.upload_spans(product, spans, otlp=…), zt.instrumentation(product), zt.export_suite(product, version, suite), zt.submit_corpus(run, chunks, data_source=…) (UK-31 corpus scan before gating) and zt.submit_session(product, session_id, input, transcript=…) (REQ-066 production sessions, a sessions:write token).

Errors: PrerequisiteError, CalibrationError, ResidencyError, IdempotencyConflictError, MissingInputsError (.missing lists each gap), VersionError, RunFailedError, RunCancelledError, AuthError, NotFoundError, ZentestTimeout; all are ZentestError.

Minimum version (SDK-19): the client reads GET /api/version once before its first call and raises VersionError with the upgrade command when it is older than the platform's min_sdk_version.

CLI (zentest, CI gate guide): zentest run gate.zt (a platform gate in CI), zentest cancel RUN_ID, and offline (SDK-08):

zentest export --product "Claims Copilot" --version v6 --suite "Claims FAQ Golden Set" -o claims.ztx
zentest run --offline claims.ztx --agent my_app.gate:agent --repeats 3      # exit 0 passed, 1 not passed, 2 error
                                                                            #   or incomplete

The export holds cases, metric configs and rubrics, never a connection or secret. It does hold every case input, and once written it is outside the platform's retention and erasure: keep it where the suite's data may live, delete it after the job, and export again after an erasure. Offline, deterministic metrics are scored with the platform's own check library and the platform's decision rule. Offline mode cannot score judge metrics (they need the platform's judge gateway: calibration, family guard, residency), apply the rubric guards, span-level checks, the power check or the override chain, or produce an evidence pack; nothing is uploaded.

Incomplete runs exit 2 by default. When judge metrics were skipped the result is status: "incomplete" (exit 2): a deterministic-only pass is not a release verdict. A definite not-pass still exits 1 (blocked by a gating deterministic metric, or not passed with every trial scored). --allow-incomplete accepts the deterministic-only result (exit 0/1). A trial with no metric scorable offline is not passed, and a suite with no deterministic metric has no offline verdict (passed: null, exit 2 even with --allow-incomplete). --strict is the default and still accepted.

Examples

• examples/mock_app.py: a local stand-in app that reports attribution (meta).
• examples/quickstart.py: runs it through both runners, with tracing.
• examples/http_app_gate.py: wraps any login-protected chat API as an SDK agent from environment variables (logs in, caches and refreshes the JWT, threads conversation ids, reads answer and context with JSONPaths, reports attribution). python sdk/examples/http_app_gate.py --smoke "question" checks the configuration with one call.

Tests (from a checkout of the repository): cd sdk && python3 -m pytest -q tests (no network: httpx.MockTransport; the span and OTel tests also run the platform's own validator from backend/ when it is importable). pip install -e "sdk[dev]" adds build and twine, which the packaging tests use to build and check the release artifacts.

Versions and licence

zentest.__version__ is the installed version; CHANGELOG.md lists what each release changed. Releases are additive within contract v1, and the platform tells an SDK that is too old to upgrade (VersionError, above). Copyright Nxzen; proprietary licence (see the package metadata).