tenet-langchain 0.4.0

Tenet PII governance integration for LangChain agents

tenet-langchain

LangChain integration for the Tenet compliance SDK. Built on top of the framework-agnostic tenet-client package — install this one if you're building a LangChain agent; install tenet-client directly if you're not.

pip install tenet-langchain

You own the agent lifecycle

Tenet evaluates and returns a verdict. Your application policy decides what to do with it — allow, review, block, override the call, fail-open, hand off to a human reviewer. Modeled loosely on Galileo Protect's stage/action split: the centralized judge produces the verdict, your local policy maps it to a runtime action.

That means you keep calling langchain.agents.create_agent yourself. Tenet ships as a middleware you drop into the middleware=[...] list, not a façade that owns agent creation.

Three control levels

Level	Surface	When to use
1. Raw client	TenetCloudJudgeClient + evaluate_phase	Custom agents (no LangChain), FastAPI middleware, queue workers, anywhere you want full control.
2. LangChain middleware	TenetJudgeMiddleware + JudgePolicy	The primary path. Drop into a create_agent(...) middleware list; policy callbacks decide every outcome.
3. Convenience helper	wrap() (deprecated)	Preserved for alpha integrations. Owns the agent lifecycle for you, which removes control over middleware order and policy dispatch. New code should use level 2.

The local-server path (TenetMiddleware) remains unchanged for self-hosted Tenet desktop / on-prem deployments.

Primary path — create_agent + TenetJudgeMiddleware + JudgePolicy

from langchain.agents import create_agent
from tenet_client import (
    JudgeAction,
    JudgePolicy,
    TenetCloudJudgeClient,
)
from tenet_langchain import TenetJudgeMiddleware

judge = TenetCloudJudgeClient.from_env()

policy = JudgePolicy(
    on_block=lambda ctx, verdict: JudgeAction.tool_error(
        verdict.safe_message or "Blocked by Tenet."
    ),
    on_review=lambda ctx, verdict: JudgeAction.allow(metadata={"tenet_review": True}),
    on_unavailable=lambda ctx, error: JudgeAction.raise_error(error),
    on_timeout=lambda ctx, error: JudgeAction.raise_error(error),
)

agent = create_agent(
    model="claude-sonnet-4-5-20250514",
    tools=[],
    middleware=[
        TenetJudgeMiddleware(judge, policy=policy),
    ],
)

JudgePolicy() with no callbacks gives you the safe defaults — block → tool_error, review → soft-allow with metadata, unavailable / timeout → re-raise (fail-closed). Override only what you want to change.

from_env() environment variables

TenetCloudJudgeClient.from_env() needs two variables — the credentials Tenet provisions for your account:

• TENET_CLIENT_ID
• TENET_CLIENT_SECRET

Both are required; from_env() raises ValueError naming any that are missing. Everything else (cloud URL, Auth0 audience, Auth0 issuer) points at Tenet-operated infrastructure and uses sensible production defaults — you should not need to set them.

Optional tuning knobs, if you want them: TENET_JUDGE_ID (defaults to hr-1; set healthcare-1 for the PHI / clinical-safety judge) and TENET_JUDGE_TIMEOUT_SECONDS (defaults to 3.0).

If you'd rather pass credentials explicitly, call TenetCloudJudgeClient(...) directly with the same keyword arguments.

Per-tool and per-phase policies

Stricter rules for high-risk tools, softer rules for low-risk ones:

from tenet_client import JudgeAction, JudgePolicy
from tenet_langchain import TenetJudgeMiddleware

policy = JudgePolicy(
    per_tool={
        "send_email": JudgePolicy(
            on_block=lambda ctx, v: JudgeAction.tool_error("Cannot send PHI."),
        ),
        "search": JudgePolicy(
            on_block=lambda ctx, v: JudgeAction.allow(),
        ),
    },
    per_phase={
        "tool_post": JudgePolicy(
            on_block=lambda ctx, v: JudgeAction.override_output(
                "Sorry, I can't share that."
            ),
        ),
    },
)

Per-tool wins over per-phase wins over the top-level callbacks.

Read-intent — read_tools / intent_resolver (0.4.0)

The engine reads a top-level intent field on the /decide body as its highest-precedence intent signal. On a benign first-party read (e.g. "what have we said to this candidate") a missing intent resolves to unknown, which grades the call against the strict action ceiling and needlessly escalates. Declaring which tools are reads lets the middleware assert intent="read" so those calls fast-pass.

Declare the read taxonomy once at construction. Either enumerate the read tools, or supply a resolver (for a computed taxonomy):

from tenet_langchain import TenetJudgeMiddleware  # or CloudJudgeMiddleware

mw = TenetJudgeMiddleware(
    judge,
    policy=policy,
    judge_tools_pre={"schedule_interview", "send_phone_screen"},  # mutating → pre
    judge_tools_post={"top_candidates", "read_messages"},         # reads → post
    read_tools={"top_candidates", "read_messages"},               # ← intent="read"
    # or, for a computed taxonomy (wins over read_tools):
    # intent_resolver=lambda tool_name: "read" if tool_name.startswith("read_") else None,
)

The intent wire field takes read | action | unknown (default unset == unknown, omitted from the body). read / action are emitted; unknown / None are omitted so the envelope is byte-identical to not sending it.

Safety contract — tool_post-only, enforced in code

A caller-asserted intent="read" is stamped only at phase == "tool_post" for a tool in read_tools (or one the resolver classifies read). It is never emitted at agent_input or tool_pre — even if the tool name matches read_tools.

This is not a convention; it is enforced by resolve_read_intent (tenet_langchain._intent), which applies the phase gate before consulting read_tools / intent_resolver. The reason: a read is authoritative only where the judged tool's read-only nature is verifiable (at tool_post, backed by the engine's operator read-tool allowlist + the tool_input protected-class ratchet). At agent_input a caller read is a guardrail bypass — the engine proved that forcing intent="read" on the agent_input traps leaked 8/10 (probe 709cd6a) — so the engine now treats an agent_input caller intent as advisory, re-derives read-eligibility from the prompt itself, and the SDK never sends it there. tool_pre (actuating, pre-execution) is likewise never a read.

intent_resolver is consulted only at tool_post; it must return read, action, unknown, or None. Never blanket intent="read" onto an actuating tool — resolve it from the single audited taxonomy above. See the wire contract for the field semantics and the X-Tenet-Api-Version header.

Recipes

Block hard. Default. Returns a ToolMessage(status="error") to the agent.

from tenet_client import JudgeAction, JudgePolicy

policy = JudgePolicy(
    on_block=lambda ctx, v: JudgeAction.tool_error(v.safe_message or "blocked"),
)

Review only — never block, just tag for audit.

from tenet_client import JudgeAction, JudgePolicy

policy = JudgePolicy(
    on_block=lambda ctx, v: JudgeAction.allow(metadata={"flagged": True}),
    on_review=lambda ctx, v: JudgeAction.allow(metadata={"flagged": True}),
)

Fail-open dev mode. When the cloud is flaky in dev.

from tenet_client import JudgeAction, JudgePolicy

dev_policy = JudgePolicy(
    on_unavailable=lambda ctx, err: JudgeAction.allow(),
    on_timeout=lambda ctx, err: JudgeAction.allow(),
)

Human review handoff.

from tenet_client import JudgeAction, JudgePolicy

def queue_for_review(ctx, verdict):
    queue.push({
        "session_id": ctx.session_id,
        "tool_name": ctx.tool_name,
        "verdict": verdict.model_dump(),
    })
    return JudgeAction.tool_error("Waiting on human review.")

policy = JudgePolicy(on_block=queue_for_review)

Human-in-the-loop review prompt.

Use this when a review verdict should pause the LangChain agent and ask the end user whether to proceed. This uses LangGraph interrupts under the hood, so the agent needs a checkpointer and each run needs a stable thread_id. You do not need to build a custom LangGraph graph; LangChain agents created with create_agent(...) are already graph-backed.

from typing import Any

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command, interrupt

from tenet_client import (
    JudgeAction,
    JudgeContext,
    JudgePolicy,
    JudgeVerdict,
    TenetCloudJudgeClient,
)
from tenet_langchain import TenetJudgeMiddleware


def ask_user_on_review(ctx: JudgeContext, verdict: JudgeVerdict) -> JudgeAction:
    answer: Any = interrupt(
        {
            "kind": "tenet_review",
            "question": "Tenet recommends human review. Proceed with this tool call?",
            "tool_name": ctx.tool_name,
            "tool_input": ctx.tool_input,
            "reason": verdict.reason,
            "choices": [
                {"id": "proceed", "label": "Yes, proceed"},
                {"id": "reject", "label": "No, choose another action"},
            ],
        }
    )

    if isinstance(answer, dict) and answer.get("proceed") is True:
        return JudgeAction.allow(metadata={"tenet_review_approved": True})

    if isinstance(answer, dict):
        message = answer.get("message")
    else:
        message = None

    return JudgeAction.tool_error(
        message or "The user rejected this tool call after Tenet review."
    )


judge = TenetCloudJudgeClient.from_env()
policy = JudgePolicy(on_review=ask_user_on_review)

agent = create_agent(
    model="claude-sonnet-4-5-20250514",
    tools=[send_claim_to_payer],
    middleware=[TenetJudgeMiddleware(judge, policy=policy)],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "case-123"}}

result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "Submit this prior-auth packet to the payer.",
            }
        ]
    },
    config=config,
    version="v2",
)

# If Tenet returns review, expose result.interrupts[0].value to your UI.
# The user chooses yes/no, then you resume with the same thread_id.
if result.interrupts:
    user_selected_yes = True

    if user_selected_yes:
        resume_payload = {"proceed": True}
    else:
        resume_payload = {
            "proceed": False,
            "message": "Do not submit yet. Ask me for the missing CPT code first.",
        }

    agent.invoke(
        Command(resume=resume_payload),
        config=config,
        version="v2",
    )

Keep the on_review callback idempotent. LangGraph restarts the node that called interrupt(...) when the run resumes, so any side effects before the interrupt can happen more than once.

Safe override response — replace the tool output instead of erroring.

from tenet_client import JudgeAction, JudgePolicy

policy = JudgePolicy(
    on_block=lambda ctx, v: JudgeAction.override_output(
        "I can't share that information. Please contact support."
    ),
)

Retry nudge — let the agent self-correct.

A retry_nudge action returns a ToolMessage(status="error") carrying corrective guidance to the agent loop. The agent reads the guidance on its next iteration and re-issues the tool call, typically with different arguments. Distinct from tool_error (terminal denial) and from override_output (replacement output). Forge research found this mechanic produced large completion-accuracy lifts on identical model weights, so it ships as a first-class JudgeAction kind.

from tenet_client import JudgeAction, JudgePolicy

policy = JudgePolicy(
    on_block=lambda ctx, v: JudgeAction.retry_nudge_from_verdict(v),
)

The canonical constructor assembles the guidance from verdict.reason and verdict.safe_message via a documented two-line template:

Cannot proceed as requested: {reason}
Suggested correction: {safe_message_or_reason}

Pass template=... to retry_nudge_from_verdict to override.

Severity-routed recipe.

For partner-configurable severity → action mapping, use the severity_routed_policy recipe. Default mapping:

Severity	Action
low	retry_nudge_from_verdict
medium	retry_nudge_from_verdict
high	tool_error
critical	tool_error
None	tool_error (preserves v1 default)

Each bucket is keyword-configurable. human_review_severities is empty by default and, when populated, takes precedence over tool_error:

from tenet_client.recipes import severity_routed_policy

policy = severity_routed_policy(
    retry_nudge_severities=("low", "medium"),
    tool_error_severities=("high", "critical"),
    human_review_severities=(),
)

Pass through any other JudgePolicy kwargs (on_unresolved, on_unavailable, per_tool, ...) via the same call.

on_unresolved — workflow correction, not a compliance outcome.

A tool can return a valid but empty result — "no records found", empty list, null. The middleware detects these heuristically (see below) and routes the case to JudgePolicy.on_unresolved independently of the verdict's allow|review|block. The default callback returns retry_nudge_from_verdict(v) so the agent gets a workflow correction prompt; override to taste:

from tenet_client import JudgeAction, JudgePolicy

policy = JudgePolicy(
    on_unresolved=lambda ctx, v: JudgeAction.retry_nudge(
        "No matches. Broaden the query or try a synonym."
    ),
)

Resolution-status detection

After every tool call, TenetJudgeMiddleware classifies the return as resolved / unresolved / (error — reserved for a future wave) and threads the value to the judge so server-side rules can branch on it. The default heuristics (case-insensitive substring match on the stringified output):

• empty list [], empty dict {}, empty string, or literal "None" → unresolved
• "not found", "no results", "no match", "0 results" → unresolved
• anything else → resolved

Opt out per tool or globally on the middleware constructor:

mw = TenetJudgeMiddleware(
    judge,
    policy=policy,
    resolution_detection=True,                              # default
    resolution_detection_excluded_tools={"send_email"},     # skip these tools
)

When opted out, the middleware threads resolution_status=None, which preserves the pre-Phase-4.5 wire envelope and dispatch byte-for-byte.

Retry-nudge budget + telemetry

To prevent runaway loops where the agent keeps re-issuing the same broken call, the middleware tracks consecutive retry-nudges per tool and escalates to tool_error once the budget is hit:

mw = TenetJudgeMiddleware(
    judge,
    policy=policy,
    retry_nudge_max_consecutive=3,   # default; escalate on the 4th
)

Three counters live on mw.telemetry, each keyed by (surface, tool):

• mw.telemetry.emitted — every retry-nudge synthesized
• mw.telemetry.succeeded — every chain that self-corrected (a non-nudge tool call followed a nudge on the same tool)
• mw.telemetry.exhausted — every chain that hit the budget and was escalated to tool_error

Read these directly to wire into whatever observability stack you have; the middleware ships no external dependency.

Verdict-first streaming

The judge exposes an SSE surface (POST /eeo/decide/stream) that emits the verdict ~4× sooner than the full response. Opt in with stream=True and run the agent with an async invocation (ainvoke / astream) — the sync hooks have no streaming variant and ignore the flag.

mw = TenetJudgeMiddleware(judge, policy=policy, stream=True)

The middleware branches on the stream's route event:

• act_at_verdict (route_to_human): the policy is dispatched the instant the verdict event lands — before the safe-message / explanation / rewrite body streams in — so a routing decision is available at ~TTV. The body is still drained and folded onto the verdict context for the HITL payload.
• wait_for_done (include / exclude / block): the rewrite or explanation is the actionable payload, so the stream drains fully and the policy is dispatched once on the complete verdict.

Behaviour and return types are otherwise identical to the non-streaming path — it is a drop-in, A/B-able latency optimization. (For the raw client, the equivalent toggles are use_stream=True, plus evaluate_stream(...) for a verdict-first event iterator.)

Multi-turn conversation history

The judge is stateless per call. On multi-turn BFOQ flows, sending only the latest turn can read as discriminatory out of context. Set send_history=True to send the prior recruiter turns (every prior HumanMessage, oldest first) as additional {"role": "user"} messages ahead of the latest turn:

mw = TenetJudgeMiddleware(judge, policy=policy, send_history=True)

Assistant turns are never sent — they carry no recruiter intent. Off by default, in which case the wire envelope is byte-for-byte the single-turn shape.

Langfuse session grouping

Every judge call carries a conversation session_id so the server groups a conversation's traces into one Langfuse session. The middleware uses your LangGraph thread_id when a checkpointer is configured (so each conversation groups on its own), falling back to the middleware's session_id constructor arg otherwise — the same conversation identity the verdict cache and turn-approval features scope on. No configuration required; to pin a fixed id for an out-of-graph deployment, pass TenetJudgeMiddleware(judge, session_id="…").

Support correlation — X-Tenet-Canonical-Id

Every judge call is tagged with a 32-hex canonical_id — the same key the server emits to Langfuse (trace id), Metronome (transaction id), and OTel (tenet.canonical_id). It is returned on the X-Tenet-Canonical-Id response header and the request_accepted / done stream events, surfaced on CloudJudgeResponse.canonical_id, and folded onto the verdict context as ctx.metadata["tenet_canonical_id"].

Runbook: log the canonical_id on every judge call so support can grep the same key across Langfuse and Metronome when triaging a decision. The middleware emits it at DEBUG (logging.getLogger("tenet_langchain.tenet_judge_middleware")). Never show it to the end user — it is an internal correlation key, not a user-facing reference.

Level 1 — raw client, no LangChain

For custom agents, FastAPI middleware, queue workers — anywhere you don't want LangChain in the call path:

from tenet_client import (
    JudgeAction,
    JudgePolicy,
    TenetCloudJudgeClient,
)

judge = TenetCloudJudgeClient.from_env()
policy = JudgePolicy()

verdict, action = judge.evaluate_phase(
    policy=policy,
    phase="tool_pre",
    tool_name="search",
    tool_input={"q": "patient lookup"},
)

if action.kind == "allow":
    result = run_search()
elif action.kind == "tool_error":
    result = action.message
elif action.kind == "raise_error":
    raise action.error

evaluate_phase_async is the async sibling.

Level 1 — @judged for standalone tools

from tenet_client import (
    JudgeAction,
    JudgePolicy,
    JudgeBlocked,
    TenetCloudJudgeClient,
    judged,
)

judge = TenetCloudJudgeClient.from_env()
policy = JudgePolicy(
    on_block=lambda ctx, v: JudgeAction.override_output(
        "I cannot share that data."
    ),
)

@judged(judge, policy=policy)
def get_patient_chart(patient_id: str) -> str:
    return _fetch_chart(patient_id)

# When the judge blocks, the override_output replaces the return value.
chart = get_patient_chart("MRN12345")

Cold-start avoidance

The first evaluate() call from a fresh TenetCloudJudgeClient pays a TLS handshake to Auth0, the Auth0 token mint, and a TLS handshake to the cloud judge — typically 600–900 ms vs. ~50–150 ms once warm. Call warmup() (or warmup_async()) at process start:

from tenet_client import TenetCloudJudgeClient

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

warmup() is idempotent and doubles as a credential smoke test.

Local server — TenetMiddleware

If you're running the Tenet desktop app or an on-prem server (default http://127.0.0.1:19990) instead of the cloud judge:

from langchain.agents import create_agent
from tenet_langchain import TenetMiddleware

agent = create_agent(
    model="claude-sonnet-4-5-20250514",
    tools=[],
    middleware=[TenetMiddleware(tenet_ids=["hipaa-safe-harbor"])],
)

Right surface for self-hosted deployments where data must not leave your network.

Migration from wrap()

wrap() still works but emits a DeprecationWarning. The translation is mechanical:

# Before
from tenet_langchain import wrap

agent = wrap(
    model=model,
    tools=tools,
    judge=judge,
    fail_open=False,
)

# After
from langchain.agents import create_agent
from tenet_client import JudgePolicy
from tenet_langchain import TenetJudgeMiddleware

agent = create_agent(
    model=model,
    tools=tools,
    middleware=[TenetJudgeMiddleware(judge, policy=JudgePolicy())],
)

The default JudgePolicy() is fail-closed — same as the old fail_open=False. To restore the old fail_open=True behavior, set on_unavailable=lambda ctx, err: JudgeAction.allow().

Examples

• examples/healthcare_agent/ — end-to-end LangChain healthcare agent gated by the healthcare-1 cloud judge.
• examples/human_review_interrupt.py — policy-level review verdict handling with a LangGraph interrupt and yes/no resume payload.

License

Apache-2.0.