langgraph-plainid

PlainID authorization integration for LangGraph. Provides LangGraph nodes for prompt categorization, text anonymization, and policy-based document retrieval that can be composed into stateful agent graphs.

This library depends on core-plainid and langchain-plainid for the underlying authorization components. Please refer to the core-plainid README for setting up permissions, categorizers, anonymizers, classifiers, and PlainID rulesets, and the langchain-plainid README for retrieval and vector store configuration.

All nodes fully support both synchronous and asynchronous execution. The examples below use the async API; replace ainvoke with invoke for synchronous usage.

Installation

pip install langgraph-plainid

core-plainid and langchain-plainid are installed automatically as dependencies. Optional extras from core-plainid can be installed directly through langgraph-plainid:

pip install langgraph-plainid[categorization-llm]        # LLM-based categorization via LiteLLM
pip install langgraph-plainid[categorization-zeroshot]   # Zero-shot classification via Hugging Face
pip install langgraph-plainid[anonymization]             # Presidio-based PII anonymization
pip install langgraph-plainid[anonymization-ahds]        # Anonymization + Azure Health De-identification
pip install langgraph-plainid[all]                       # Everything

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 (e.g. PlainIDPermissionsProvider, FilterDirectiveProvider) — see the core-plainid README for details.

from langgraph_plainid.models.state.agent_state import AgentState

Multiple identities (e.g. a User and an AI Agent) are supported for agentic scenarios through the additional_identities field in RequestContext. Identity can also be resolved via HTTP headers that are matched against configured values in PlainID — see the Identity Context section in the core-plainid README for details.

All core-plainid components constructed in this library (e.g. PlainIDPermissionsProvider, FilterDirectiveProvider) support three authentication modes: client credentials, per-request JWT token, and automatic IDP token management via IdpAuthProvider — see the Authentication section in the core-plainid README.

Base Node

All PlainID nodes extend BaseNode, which provides common behavior:

• next_node — optional name of the next node to route to via LangGraph Command. If not set, the node returns the updated state directly and graph edges determine the flow.
• next_node_on_error — optional name of a node to route to when an error occurs. If set, errors are caught and routed as ErrorDetails in the sub-state. If not set, exceptions propagate normally.

Categorization Node

The CategorizationNode classifies the input prompt against PlainID policies. If the categories are not allowed, a PlainIDCategorizerException is raised (or routed to the error handler node if configured).

For setting up the categorizer, classifier providers, and the PlainID Prompt_Control ruleset, see the Category Filtering section in the core-plainid README.

from core_plainid.categorization.categorizer import Categorizer
from core_plainid.utils.plainid_permissions_provider import PlainIDPermissionsProvider
from langgraph_plainid.nodes.categorization_node import CategorizationNode

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

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

categorization_node = CategorizationNode(
    categorizer=categorizer,
    next_node="anonymizer",
    next_node_on_error="error_handler",
)

The node reads its input from state["categorization"]["query"].

Anonymization Node

The AnonymizerNode detects and anonymizes PII in the input text based on PlainID policies. The anonymized text is written to state["anonymization"]["output_text"].

For setting up the anonymizer, encryption key, AHDS, and the PlainID Output_Control ruleset, see the Anonymization section in the core-plainid README.

from core_plainid.anonymization.presidio_anonymizer import PresidioAnonymizer
from core_plainid.utils.plainid_permissions_provider import PlainIDPermissionsProvider
from langgraph_plainid.nodes.anonymizer_node import AnonymizerNode

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

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

anonymizer_node = AnonymizerNode(
    anonymizer=anonymizer,
    next_node="retrieval",
    next_node_on_error="error_handler",
)

The node reads its input from state["anonymization"]["query"] and writes the result to state["anonymization"]["output_text"].

Retrieval Node

The RetrievalNode retrieves documents from vector stores with PlainID-enforced filters. The retrieved documents are written to state["retrieval"]["retrieved_documents"].

For setting up the retriever, filter provider, and vector store configuration, see the Retrieval section in the langchain-plainid README.

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

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

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

retrieval_node = RetrievalNode(
    retriever=retriever,
    next_node_on_error="error_handler",
)

The node reads its input from state["retrieval"]["query"] and writes the result to state["retrieval"]["retrieved_documents"].

Building a Graph

Nodes are composed into a LangGraph StateGraph to define the agent's execution flow. Here is an example graph that categorizes a prompt, anonymizes it, and then retrieves documents:

from langgraph.graph import END, START, StateGraph
from core_plainid.models.context.request_context import RequestContext
from langgraph_plainid.models.state.agent_state import AgentState

graph = StateGraph(AgentState)

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

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

app = graph.compile()

request_context = RequestContext(
    entity_id="your_entity_id",
    entity_type_id="your_entity_type",
)

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?"},
})

print(result["anonymization"]["output_text"])       # anonymized text
print(result["retrieval"]["retrieved_documents"])    # retrieved documents

Using Command-Based Routing

Instead of defining edges between all nodes, you can use the next_node parameter to let nodes route to the next step via LangGraph Command:

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

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

retrieval_node = RetrievalNode(retriever=retriever)

graph = StateGraph(AgentState)

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

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

app = graph.compile()

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:

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 {}

graph.add_node("error_handler", error_handler)

If next_node_on_error is not set, exceptions propagate normally and can be caught in the calling context of invoke / ainvoke.

Exceptions

All exceptions are defined in the core-plainid library. See the Exceptions section in the core-plainid README for the full list.