tenet-client 0.2.0

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=...,
    )
    ...

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.

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.