---
title: "LangGraph"
slug: "langgraph"
updated: 2026-07-19T13:02:35Z
published: 2026-07-19T13:02:35Z
canonical: "docs.plainid.io/langgraph"
---

> ## Documentation Index
> Fetch the complete documentation index at: https://docs.plainid.io/llms.txt
> Use this file to discover all available pages before exploring further.

# LangGraph

PlainID integrates with LangGraph to enforce authorization across agent workflows, including prompt categorization, data anonymization, and Policy-based retrieval.

This integration enables you to apply centralized authorization decisions at each step of a stateful agent graph, ensuring that execution paths, data access, and outputs are governed by Policy.

The integration is built on top of:

- `core-plainid` for authorization and Policy evaluation.
- `langchain-plainid` for retrieval and filtering.

See the following articles for more information:

> [Integration Core](/v1/docs/integration-core) [LangChain](/v1/docs/langchain-agentic-ai)

## Install the SDK

Install the SDK to add the required PlainID and LangGraph integration packages.

**To install the SDK:**

1. Run the installation command.

```
pip install langgraph-plainid
```

This installs:

- `core-plainid`
- `langchain-plainid`
- `langgraph-plainid`

## Define Agent State

All nodes operate on a shared `AgentState` `TypedDict` that flows through the graph. The state contains the `request_context` for identity information and optional sub-states for each node type. Alternatively, `request_context` can be provided at construction time to the underlying components, for example, `PlainIDPermissionsProvider` and `FilterDirectiveProvider`. See `core-plainid`.

**To define Agent State:**

1. Import the required class.

```
from langgraph_plainid.models.state.agent_state import AgentState
```

1. Review the `AgentState` fields.

| Field | Type | Description |
| --- | --- | --- |
| request_context | RequestContext | Identity context (entity ID, type, additional identities) |
| categorization | CategorizationState | Sub-state with query and optional error_details |
| anonymization | AnonymizationState | Sub-state with query, optional output_text, and error_details |
| retrieval | RetrievalState | Sub-state with query, resource_types, optional retrieved_documents, and error_details |

The state typically includes:

- `request_context` for identity information.
- `categorization` for prompt classification input.
- `anonymization` for text processing input and output.
- `retrieval` for the query and retrieved documents.

## Initialize Permissions Provider

All PlainID nodes rely on the same permissions provider.

**To initialize the permissions provider:**

1. Import the required class.

```
from core_plainid.utils.plainid_permissions_provider import PlainIDPermissionsProvider
```

1. Configure the `PlainIDPermissionsProvider`.

```
permissions_provider = PlainIDPermissionsProvider(
    base_url="https://platform-product.us1.plainid.io",
    client_id="your_client_id",
    client_secret="your_client_secret",
)
```

## Add Categorization Node

The categorization node validates whether a prompt is allowed based on PlainID Policies.

**To add a Categorization Node:**

1. Import the required classes.

```
from core_plainid.categorization.categorizer import Categorizer
from langgraph_plainid.nodes.categorization_node import CategorizationNode
```

1. Create the `Categorizer`.

```
categorizer = Categorizer(
    classifier_provider=classifier,
    permissions_provider=permissions_provider,
    all_categories=["contract", "HR", "finance"],
)
```

1. Create the `CategorizationNode`.

```
categorization_node = CategorizationNode(
    categorizer=categorizer,
    next_node="anonymizer",
)
```

Note that `next_node` is optional.

## Add an Anonymization Node

The anonymization node applies Policy-driven masking or encryption to sensitive data.

**To add an Anonymization Node:**

1. Import the required classes.

```
from core_plainid.anonymization.presidio_anonymizer import PresidioAnonymizer
from langgraph_plainid.nodes.anonymizer_node import AnonymizerNode
```

1. Create the anonymizer.

```
anonymizer = PresidioAnonymizer(
    permissions_provider=permissions_provider,
    encrypt_key="your_16_char_key!",
)
```

1. Create the `AnonymizerNode`.

```
anonymizer_node = AnonymizerNode(
    anonymizer=anonymizer,
    next_node="retrieval",
)
```

This node:

- Reads from `state["anonymization"]["query"]`.
- Writes to `state["anonymization"]["output_text"]`.

## Add Retrieval Node

The retrieval node enforces PlainID filtering when querying vector stores.

**To add a Retrieval Node:**

1. Import the required classes.

```
from langchain_plainid.retrieval.filter_directive_provider import FilterDirectiveProvider
from langchain_plainid.retrieval.multi_store_retriever import MultiStoreRetriever
from langgraph_plainid.nodes.retrieval_node import RetrievalNode
```

1. Configure the `FilterDirectiveProvider`.

```
filter_provider = FilterDirectiveProvider(
    base_url="https://platform-product.us1.plainid.io",
    client_id="your_client_id",
    client_secret="your_client_secret",
)
```

1. Create the `MultiStoreRetriever`.

```
retriever = MultiStoreRetriever(
    filter_provider=filter_provider,
    resource_types=["customer"],
    vector_stores=[customer_store],
    k=4,
)
```

1. Create the `RetrievalNode`.

```
retrieval_node = RetrievalNode(retriever=retriever)
```

This node:

- Reads from `state["retrieval"]["query"]`.
- Writes to `state["retrieval"]["retrieved_documents"]`.

## Build Graph

Compose the nodes into a `StateGraph`.

**To build the graph:**

1. Import the required classes.

```
from langgraph.graph import START, END, StateGraph
from core_plainid.models.context.request_context import RequestContext
```

1. Create the graph.

```
graph = StateGraph(AgentState)
```

1. Add the nodes.

```
graph.add_node("categorization", categorization_node)
graph.add_node("anonymizer", anonymizer_node)
graph.add_node("retrieval", retrieval_node)
```

1. Add the edges.

```
graph.add_edge(START, "categorization")
graph.add_edge("categorization", "anonymizer")
graph.add_edge("anonymizer", "retrieval")
graph.add_edge("retrieval", END)
```

1. Compile the graph.

```
app = graph.compile()
```

## Execute Workflow

Invoke the graph with a Request Context and input queries.

**To execute the workflow:**

1. Create the `RequestContext`.

```
request_context = RequestContext(
    entity_id="your_entity_id",
    entity_type_id="your_entity_type",
    additional_identities=[
        AdditionalIdentity(
            entity_id="your_additional_entity_id",
            entity_type_id="your_additional_entity_type",
        ),
    ],
)
```

1. Invoke the graph.

```
result = await app.ainvoke({
    "request_context": request_context,
    "categorization": {"query": "What is John Smith's contract status?"},
    "anonymization": {"query": "What is John Smith's contract status?"},
    "retrieval": {"query": "What is John Smith's contract status?"},
})
```

1. Access the results.

```
print(result["anonymization"]["output_text"])
print(result["retrieval"]["retrieved_documents"])
```

## Routing Between Nodes

Nodes can define the next execution step using `next_node`.

**To route between nodes:**

1. Configure the `CategorizationNode` with `next_node`.

```
categorization_node = CategorizationNode(
    categorizer=categorizer,
    next_node="anonymizer",
)
```

1. Configure the `AnonymizerNode` with `next_node`.

```
anonymizer_node = AnonymizerNode(
    anonymizer=anonymizer,
    next_node="retrieval",
)
```

This allows the graph to dynamically control execution flow without explicitly defining all edges.

## Architecture Summary

The PlainID and LangGraph integration introduces a **Policy enforcement layer embedded into the agent graph execution model**. It includes a:

- **State Layer (`AgentState`)** A shared, structured state object carries identity, inputs, and outputs across all nodes.
- **Policy Decision Layer (PlainID)** Centralized authorization engine evaluates:
  - Prompt intent (categorization).
  - Data sensitivity (anonymization).
  - Data access (retrieval filters).
- **Execution Layer (LangGraph Nodes)** Each node enforces a specific control point:
  - **Categorization Node** → governs *what can be asked*.
  - **Anonymization Node** → governs *what data is exposed*.
  - **Retrieval Node** → governs *what data can be accessed*.
- **Flow Control Layer (Graph Orchestration)** LangGraph manages execution paths, while PlainID Policies influence:
  - Whether execution continues.
  - What transformations occur.
  - What data is returned.

## Error Handling

When `next_node_on_error` is set, errors are caught and the graph routes to the specified error handler node. The error details are available in the sub-state.

**To handle errors:**

1. Define the error handler.

```
def error_handler(state: AgentState) -> dict:
    for key in ["categorization", "anonymization", "retrieval"]:
        sub_state = state.get(key)
        
        if sub_state and sub_state.get("error_details"):
            error_details = sub_state["error_details"]
            
            print(f"Error in {key}: {error_details['error_message']}")
            print(f"Exception: {error_details['error']}")

    return {}
```

1. Add the error handler node to the graph.

```
graph.add_node("error_handler", error_handler)
```