Metadata-Version: 2.4
Name: langgraph-plainid
Version: 1.0.0
Summary: LangGraph integration for PlainID authorization
Author: PlainID
License-Expression: MIT
Classifier: Development Status :: 5 - Production/Stable
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Security
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: <=3.12,>=3.10
Requires-Dist: core-plainid<2.0.0,>=1.0.0
Requires-Dist: langchain-plainid<2.0.0,>=1.0.0
Requires-Dist: langgraph<2.0.0,>=1.1.2
Description-Content-Type: text/markdown

# langgraph-plainid

[PlainID](https://www.plainid.com/) authorization integration for [LangGraph](https://langchain-ai.github.io/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](https://pypi.org/project/core-plainid/) and [langchain-plainid](https://pypi.org/project/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

```bash
pip install langgraph-plainid
```

`core-plainid` and `langchain-plainid` are installed automatically as dependencies.

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

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

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

Multiple identities (e.g. a User and an AI Agent) are supported for agentic scenarios through the `additional_identities` field in `RequestContext` — see the core-plainid README for details.

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

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

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

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

```python
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_id_type="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`:

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

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