tenet-client

Pure-Python SDK for the Tenet cloud judge service and local Tenet server. No framework dependencies — works inside LangChain, LlamaIndex, FastAPI, plain scripts, anywhere.

For LangChain-specific middleware/callback adapters, install tenet-langchain instead — it depends on this package and ships the LangChain integration classes.

Install

pip install tenet-client

Optional: local PII pre-filtering

Local PII pre-filtering (audit-only) is provided by the standalone tenet-pii-engine package, which lives in a private repository. It is not published to PyPI and is therefore not exposed as a tenet-client install extra (PyPI forbids the direct-VCS references such an extra would require). Install it separately from the private source alongside tenet-client:

# From the private tenet-pii-engine source (regex-only, recommended for SDKs):
pip install "tenet-pii-engine[regex]"
# ...or opt into the heavier ONNX PII model:
pip install "tenet-pii-engine[onnx]"

tenet-pii-engine requires Python 3.12+ (the engine's floor). The SDK core still supports 3.11; LocalPIIMiddleware is a graceful no-op when the engine isn't installed.

Local PII pre-filtering (optional)

LocalPIIMiddleware runs the tenet-pii-engine detector on payload text before the client dispatches a /redact call — an audit-only pre-filter (RFC #429): it never blocks, raises, or mutates the dispatched payload. It defaults to the regex-only backend (no model download).

from tenet_client import TenetClient, LocalPIIMiddleware

def audit(text, findings):
    for f in findings:
        print(f.label, f.span, f.source)

client = TenetClient(pii_middleware=LocalPIIMiddleware(on_findings=audit))
client.redact("Email jane@acme.com")   # findings emitted to `audit` first

When tenet-pii-engine is absent, LocalPIIMiddleware().available is False and scan() returns [], so the SDK works unchanged.

Quickstart — cloud judge

The cloud judge runs at https://api.tenetlabs.com/v1/judge/evaluate. Your customer credentials (M2M client_id / client_secret) are issued by Tenet at provisioning time.

from tenet_client import TenetCloudJudgeClient, JudgeUnavailableError

client = TenetCloudJudgeClient(
    client_id="<your-m2m-client-id>",
    client_secret="<your-m2m-client-secret>",
)

# Recommended: warm the client at process start so the first user-facing
# call doesn't pay the Auth0 token-mint cold path.
client.warmup()

try:
    verdict = client.evaluate(
        phase="tool_pre",
        tool_name="search_resumes",
        tool_input={"query": "5+ years Python experience"},
    )
    if verdict.decision == "block":
        # Integrator decides what to do — return an error to the agent,
        # surface a safe message to the user, log + halt, etc.
        ...
except JudgeUnavailableError as e:
    # Cloud judge unreachable. The SDK does NOT have a fail_open flag —
    # it's a policy decision. Catch and either retry, halt, or proceed
    # depending on your environment.
    ...

Or read credentials from env. from_env() requires TENET_CLIENT_ID and TENET_CLIENT_SECRET; optional tuning vars are TENET_JUDGE_ID, TENET_JUDGE_TIMEOUT_SECONDS, and TENET_WORKFLOW (see Workflow label). (TENET_CLOUD_URL, TENET_AUTH0_AUDIENCE, and TENET_AUTH0_ISSUER_URL exist as escape hatches for dev / staging but should not be set in production.)

client = TenetCloudJudgeClient.from_env()

Async

Every method has an _async counterpart:

from contextlib import asynccontextmanager
from fastapi import FastAPI
from tenet_client import TenetCloudJudgeClient


@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.judge = TenetCloudJudgeClient.from_env()
    await app.state.judge.warmup_async()
    yield
    await app.state.judge.aclose()


app = FastAPI(lifespan=lifespan)


@app.post("/run-tool")
async def run_tool(req: ...):
    verdict = await app.state.judge.evaluate_async(
        phase="tool_pre", tool_name=..., tool_input=...,
    )
    ...

Reading multi-verdict responses

The judge returns a per-domain findings list alongside the top-level verdict. Simple callers can keep reading just the top level; callers who need to dispatch on a specific domain (EEO, disclosure, prompt injection, ...) read findings.

verdict = client.evaluate(phase="tool_pre", tool_input={"q": "..."})

# 1) Top-level still works. Backwards-compat is a guarantee: existing
#    code reading `verdict.verdict` / `verdict.decision` keeps working.
if verdict.verdict == "route_to_human":
    ...

# 2) Per-domain detail — the EEO finding is always present (the engine
#    synthesizes one from the top-level on legacy responses).
eeo = verdict.eeo_finding
print(eeo.verdict, eeo.severity, eeo.rule_ids, eeo.evidence_span)

# Index by name when you care about a specific domain.
disclosure = verdict.findings_by_domain.get("disclosure")
if disclosure is not None:
    ...

# Convenience helpers read well in agent code.
if verdict.blocked_by_domain("eeo"):
    return safe_message
span = verdict.evidence_for("eeo")  # str | None

Audit-only vs enforce

Every finding carries an enforcement field:

• enforce — this finding's verdict participated in the merged top-level verdict (most-restrictive-wins across all enforcing findings).
• audit_only — wire-visible for observability but never gates the top-level verdict. A new domain ships in audit-only until its promotion gate; an audit-only block does not block the caller.

verdict.enforcing_findings and verdict.audit_only_findings partition the list. blocked_by_domain(name) returns True only for an enforcing blocking finding — audit-only blockers return False on purpose; that's the semantic difference.

Which domains exist today

• EEO — live, enforce. Always findings[0]. Genuine EEO signals stay on domain="eeo".

• safety — live. Deterministic safety guards (prompt injection + exfil detection) are attributed to domain="safety". This finding carries a nested sub_findings list breaking the block down per detector — each SubFinding names its sub_domain ("injection" / "exfil"), the rule_ids that fired (e.g. ["SAFETY-EXFIL-01"]), its verdict, and enforcement:

  safety = verdict.findings_by_domain.get("safety")
  if safety is not None:
      for sf in safety.sub_findings:
          print(sf.sub_domain, sf.rule_ids, sf.verdict, sf.enforcement)
  

• Additional domains (disclosure, ...) append to findings as they ship, each starting in audit_only and graduating to enforce per its promotion gate.

Out-of-vocab verdicts normalize to route_to_human for the merged top-level value; the per-finding verdict is preserved verbatim so an audit can reconstruct what each detector actually returned.

Streaming (verdict-first)

The judge exposes an SSE surface that emits the verdict ~4× sooner than the full response. Two opt-in paths:

judge = TenetCloudJudgeClient.from_env()  # or use_stream=True / TENET_USE_STREAM=1


async def decide(prompt: dict):
    # 1. Drop-in: route through the stream, identical return type.
    verdict = await judge.evaluate_stream_to_response(
        phase="tool_pre", tool_input=prompt
    )

    # 2. Verdict-first: act on the verdict event before the body streams in.
    async for event in judge.evaluate_stream(phase="tool_pre", tool_input=prompt):
        if event.event_type == "verdict":
            ...  # dispatch your routing decision at ~TTV

    return verdict

Streaming is async-only; the sync evaluate() stays on the non-streaming route. The aggregated response additionally carries canonical_id, client_action, safe_message, explanation, and rewrite.

When a per-domain finding tightens the merged verdict after the initial verdict event, the server emits verdict_revised carrying the same shape plus a domain field identifying which finding drove the revision. The aggregator takes the latest revision as terminal; verdict- first consumers can either ignore verdict_revised (act at the first verdict) or re-dispatch on it. The domain key may be absent on older deployments — tolerate it.

Multi-turn history

Pass prior recruiter turns so the stateless judge sees the conversation context (oldest first); they become extra {"role": "user"} messages ahead of the latest turn. Never send assistant turns.

async def decide_with_history(judge):
    return await judge.evaluate_async(
        phase="agent_input",
        tool_input="now only the recent grads",
        prior_user_turns=["find backend engineers", "ones who can start now"],
    )

Session grouping

Pass a session_id — a conversation id — so the server groups every judge call in one conversation into a single Langfuse session. Use a stable id per conversation (e.g. your LangGraph thread_id):

async def decide_in_conversation(judge, conversation_id: str):
    return await judge.evaluate_async(
        phase="agent_input",
        tool_input="now only the recent grads",
        session_id=conversation_id,
    )

The server tenant-namespaces and length-caps the id, so send it raw; omit it (don't send empty) for an ungrouped, standalone trace. Works on the sync, async, and streaming paths.

Workflow label

If several of your integrations call the judge (e.g. a batch apply-summary job and a recruiter-agent chat agent), name each one with workflow so the Tenet traces page can filter per workflow instead of mixing them together. Set it once on the client (or via TENET_WORKFLOW) and override per call if one client serves several workflows:

judge = TenetCloudJudgeClient.from_env()  # TENET_WORKFLOW=apply-summary
judge.evaluate(phase="agent_input", tool_input="...", workflow="recruiter-agent")

It is a trace label only: it never affects the verdict and is unrelated to surface. Use a lowercase slug (letters, digits, ., _, -; at most 64 chars). Anything else raises ValueError at construction or call time, before any request is sent. Omit it and the call is recorded as unlabeled.

Support correlation — canonical_id

Each streamed decision carries a 32-hex canonical_id (also on the X-Tenet-Canonical-Id response header) — the same key the server emits to Langfuse, Metronome, and OTel. Log it on every call so support can grep one key across all three; never show it to the end user.

@judged — gate any function

from tenet_client import TenetCloudJudgeClient, JudgeBlocked, judged

judge = TenetCloudJudgeClient.from_env()
judge.warmup()

@judged(judge, fail_open=False)
def search_resumes(query: str) -> list[dict]:
    return _real_search(query)

try:
    hits = search_resumes("5+ years Python")
except JudgeBlocked as e:
    # e.reason / e.phase / e.tool_name / e.judge_id available
    ...

The decorator gates the function at tool_pre (before the body runs) and tool_post (after it returns). Works on sync and async functions — the wrapper auto-detects via inspect.iscoroutinefunction.

Integration recipes

For LangChain agents, install tenet-langchain — it ships CloudJudgeMiddleware, the wrap() one-call helper, and re-exports @judged. For other frameworks (LlamaIndex, FastAPI, raw loops), see the cloud judge recipes for copy-pasteable wiring patterns.

License

Apache-2.0.