tenet-client 0.3.2

Tenet SDK — cloud judge client and local server client. Framework-agnostic.

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

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 and TENET_JUDGE_TIMEOUT_SECONDS. (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].
• Additional domains (disclosure, prompt injection, ...) 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.

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.