# LangGraph Integration (Python)

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

Source: https://docs.atensecurity.com/docs/sdks/langgraph-integration

## 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

```bash
pip install atensec-thoth langgraph
```

`langgraph` is optional. If missing, Thoth raises:

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

## API

```python
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

```python
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

```python
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

```python
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):

```python
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):

```python
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",
)
```