LangGraph Integration (Python)

First-class Thoth governance for LangGraph StateGraph, CompiledStateGraph, and ToolNode patterns.

Why LangGraph Needs Special Handling

LangGraph executes tools through ToolNode and tool objects (.invoke() / .ainvoke()), not only plain Python callables. Compared to LangChain AgentExecutor, production graphs add:

Stateful multi-step execution (StateGraph / compiled graphs)
Sync + async tool invocation paths
Parallel branches that require session-safe call history

Thoth's instrument_langgraph() instruments the graph or tools in one call and preserves one session across the full workflow.

Installation

pip install atensec-thoth langgraph

langgraph is optional. If missing, Thoth raises:

ImportError: langgraph is required for this integration. Install it with: pip install langgraph

API

def instrument_langgraph(
    graph_or_tools,
    *,
    agent_id: str,
    approved_scope: list[str],
    tenant_id: str,
    user_id: str = "system",
    enforcement: str = "block",
    api_key: str | None = None,
    api_url: str | None = None,
    session_id: str | None = None,
    session_intent: str | None = None,
    environment: str = "prod",
    enforcement_trace_id: str | None = None,
    purpose: str | None = None,
    data_classification: str | None = None,
    task_context: str | dict[str, Any] | None = None,
    fail_open: bool = False,
)

Accepted inputs:

StateGraph (pre-compile)
CompiledStateGraph
list of @tool tools

Return type matches input type.

Pattern A: Instrument a Tool List

from langchain_core.tools import tool
from thoth import instrument_langgraph
 
@tool
def retrieve_patient_record(patient_id: str) -> dict:
    """Retrieve patient profile."""
    return {"patient_id": patient_id}
 
@tool
def check_formulary(medication: str, patient_id: str) -> dict:
    """Check medication coverage."""
    return {"covered": True}
 
governed_tools = instrument_langgraph(
    [retrieve_patient_record, check_formulary],
    agent_id="healthcare-workflow",
    approved_scope=["retrieve_patient_record", "check_formulary"],
    tenant_id="abridge",
    session_intent="clinical-workflow-automation",
    purpose="clinical-documentation",
    data_classification="PHI",
    enforcement="step_up",
    api_url="https://enforce.example",
)

Pattern B: Instrument `ToolNode` Directly

from langchain_core.tools import tool
from thoth.integrations.langgraph import instrument_tool_node
 
@tool
def search_docs(query: str) -> str:
    """Search docs."""
    return f"result:{query}"
 
tool_node = instrument_tool_node(
    [search_docs],
    agent_id="docs-agent",
    approved_scope=["search_docs"],
    tenant_id="acme",
    enforcement="block",
    api_url="https://enforce.example",
)

Pattern C: `@thoth_graph` Decorator

from langchain_core.tools import tool
from langgraph.graph import START, StateGraph
from langgraph.prebuilt import ToolNode
from thoth.integrations.langgraph import thoth_graph
 
@tool
def retrieve_patient(patient_id: str) -> dict:
    """Retrieve patient."""
    return {"patient_id": patient_id}
 
@thoth_graph(
    agent_id="healthcare-workflow",
    approved_scope=["retrieve_patient"],
    tenant_id="abridge",
    session_intent="clinical-workflow-automation",
    enforcement="step_up",
    api_url="https://enforce.example",
)
def build_graph():
    g = StateGraph(dict)
    g.add_node("tools", ToolNode([retrieve_patient]))
    g.add_edge(START, "tools")
    return g.compile()
 
app = build_graph()

Session Continuity

A single instrument_langgraph() call creates one SessionContext. All tool calls in that graph execution share:

one session_id
one accumulating session_tool_calls history

This enables accumulation-style policy logic (for example MOSES risk progression) across multi-step flows.

Sync + Async Handling

Thoth wraps both:

tool.invoke(...)
tool.ainvoke(...)

Async calls use async enforcement (acheck) and async step-up polling.

STEP_UP in Graph Execution

When the enforcer returns STEP_UP with a hold_token, Thoth polls /v1/enforce/hold/{hold_token} until resolution.

Approved: tool executes and graph continues
Denied/timeout: ThothPolicyViolation is raised and graph execution stops unless your graph/app handles it

MODIFY Behavior

For MODIFY decisions, Thoth rewrites tool arguments from decision.modified_tool_args, executes with modified args, and emits telemetry with:

original_tool_args
modified_tool_args
modification_reason

DEFER Behavior

For DEFER, Thoth does not execute the tool. It raises ThothDeferredError and includes defer metadata (defer_reason, defer_timeout_seconds).

Regulatory Context (HIPAA / Clinical)

Set regulatory context fields directly at instrumentation time:

session_intent
purpose
data_classification
task_context

These are propagated to enforcement payloads and tool telemetry.

Fail-Open vs Fail-Closed

fail_open=False (default): unreachable enforcer returns fail-closed behavior in enforcing modes.
fail_open=True: retryable transport/status failures can allow execution.
observe mode always executes tools and logs warnings when enforcer is unavailable.

Mock Mode

Set THOTH_MOCK_MODE=true to simulate decisions locally without a live enforcer.

Mock behavior:

tool in scope -> ALLOW
out-of-scope + observe -> ALLOW
out-of-scope + step_up/progressive -> STEP_UP then approval
out-of-scope + block -> BLOCK
tool name contains write|delete|modify|export with data_classification="PHI" -> STEP_UP

Compatibility Matrix

Validated in this SDK against:

langgraph prebuilt ToolNode
StateGraph and CompiledStateGraph
sync and async tool invocation paths

Migration: `Tracer.wrap_tool()` -> `instrument_langgraph()`

Before (manual wiring):

config = ThothConfig(...)
session = SessionContext(config)
emitter = HttpEmitter(...)
enforcer = EnforcerClient(config)
step_up = StepUpClient(config)
tracer = Tracer(config, session, emitter, enforcer, step_up)
search = tracer.wrap_tool("search", search)
retrieve = tracer.wrap_tool("retrieve", retrieve)

After (one call):

from thoth import instrument_langgraph
 
governed_tools = instrument_langgraph(
    [search, retrieve],
    agent_id="agent",
    approved_scope=["search", "retrieve"],
    tenant_id="acme",
    api_url="https://enforce.example",
)

LangGraph Integration (Python)

On this page