tenet-langchain 0.1.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.

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.

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.