Implementation:Langchain ai Langgraph Encryption Module

Knowledge Sources	Langchain_ai_Langgraph
Domains	Encryption, Security
Last Updated	2026-02-11 16:00 GMT

Overview

The `Encryption` class provides a decorator-based framework for implementing custom at-rest encryption and decryption of checkpoint blobs and structured JSON data in LangGraph Server applications.

Description

The Encryption module enables developers to add custom at-rest encryption to LangGraph deployments. Similar in structure to the Auth module, it uses a decorator pattern: developers create an `Encryption` instance, register handler functions via `@encryption.encrypt.blob`, `@encryption.decrypt.blob`, `@encryption.encrypt.json`, and `@encryption.decrypt.json` decorators, and reference the file in `langgraph.json` under the `"encryption"` key.

The module distinguishes between two data categories. Blob encryption handles opaque binary data such as serialized checkpoint state. The `encrypt.blob` and `decrypt.blob` decorators register async handlers with the signature `(EncryptionContext, bytes) -> bytes`. JSON encryption handles structured dictionaries such as thread metadata, assistant context, run kwargs, and store values. The `encrypt.json` and `decrypt.json` decorators register async handlers with the signature `(EncryptionContext, dict) -> dict`. A critical constraint for JSON encryptors is that they must preserve dictionary keys -- only values may be transformed -- because the server uses SQL JSONB merge operations that work at the key level.

Each handler receives an `EncryptionContext` object containing `model` (the entity type being encrypted, e.g., `"thread"`, `"checkpoint"`), `field` (the specific field, e.g., `"metadata"`, `"values"`), and `metadata` (a dict of non-secret context values persisted alongside encrypted data). The optional `@encryption.context` decorator registers a handler that derives encryption context from the authenticated user, allowing tenant-specific encryption keys to be resolved from JWT claims without requiring a separate header.

Internally, the `_EncryptDecorators` and `_DecryptDecorators` helper classes manage handler registration and enforce the one-handler-per-slot constraint via `DuplicateHandlerError`. All handlers are validated at registration time to ensure they are async functions with exactly two positional parameters. The module emits a `LangGraphBetaWarning` on first instantiation since the API is in beta.

Usage

Use this module when deploying LangGraph Server with requirements for at-rest encryption of checkpoint data, thread state, or metadata. Create an `encryption.py` file, instantiate `Encryption()`, register your encrypt/decrypt handlers, and add `"encryption": {"path": "./encryption.py:my_encryption"}` to `langgraph.json`.

Code Reference

Source Location

• Repository: Langchain_ai_Langgraph
• File: libs/sdk-py/langgraph_sdk/encryption/__init__.py

Signature

class Encryption:
    types = types  # Reference to encryption type definitions

    def __init__(self) -> None: ...

    # Decorator namespaces
    encrypt: _EncryptDecorators   # .blob(fn) and .json(fn)
    decrypt: _DecryptDecorators   # .blob(fn) and .json(fn)

    def context(self, fn: types.ContextHandler) -> types.ContextHandler:
        """Register a context handler to derive encryption context from auth."""
        ...

    def get_json_encryptor(self, _model: str | None = None) -> types.JsonEncryptor | None: ...
    def get_json_decryptor(self, _model: str | None = None) -> types.JsonDecryptor | None: ...

class _EncryptDecorators:
    def blob(self, fn: types.BlobEncryptor) -> types.BlobEncryptor: ...
    def json(self, fn: types.JsonEncryptor) -> types.JsonEncryptor: ...

class _DecryptDecorators:
    def blob(self, fn: types.BlobDecryptor) -> types.BlobDecryptor: ...
    def json(self, fn: types.JsonDecryptor) -> types.JsonDecryptor: ...

Import

from langgraph_sdk import Encryption
from langgraph_sdk.encryption.types import EncryptionContext

I/O Contract

Blob Encryption/Decryption Handlers

Direction	Name	Type	Description
Input	ctx	`EncryptionContext`	Context with `model`, `field`, and `metadata`
Input	blob	`bytes`	Raw bytes to encrypt or encrypted bytes to decrypt
Output	(return)	`bytes`	Encrypted or decrypted bytes

JSON Encryption/Decryption Handlers

Direction	Name	Type	Description
Input	ctx	`EncryptionContext`	Context with `model`, `field`, and `metadata`
Input	data	`dict[str, Any]`	Plaintext or encrypted JSON dictionary
Output	(return)	`dict[str, Any]`	Encrypted or decrypted JSON dictionary (must preserve keys)

Context Handler

Direction	Name	Type	Description
Input	user	`BaseUser`	The authenticated user from Starlette's AuthenticationMiddleware
Input	ctx	`EncryptionContext`	Current context (with `metadata` from `X-Encryption-Context` header)
Output	(return)	`dict[str, Any]`	Dict that becomes `ctx.metadata` for all subsequent encrypt/decrypt calls

EncryptionContext Attributes

Attribute	Type	Description
model	None`	The model type being encrypted (e.g., `"assistant"`, `"thread"`, `"checkpoint"`)
field	None`	The specific field being encrypted (e.g., `"metadata"`, `"context"`, `"values"`)
metadata	`dict[str, Any]`	Non-secret key-value pairs persisted with the encrypted data

Usage Examples

from langgraph_sdk import Encryption
from langgraph_sdk.encryption.types import EncryptionContext

encryption = Encryption()

ENCRYPTED_PREFIX = "enc:"

@encryption.encrypt.blob
async def encrypt_blob(ctx: EncryptionContext, blob: bytes) -> bytes:
    """Encrypt checkpoint blobs using an external KMS."""
    return await kms_client.encrypt(
        blob, context={"tenant": ctx.metadata.get("tenant_id")}
    )

@encryption.decrypt.blob
async def decrypt_blob(ctx: EncryptionContext, blob: bytes) -> bytes:
    """Decrypt checkpoint blobs."""
    return await kms_client.decrypt(
        blob, context={"tenant": ctx.metadata.get("tenant_id")}
    )

SKIP_FIELDS = {"tenant_id", "owner", "thread_id", "assistant_id"}

@encryption.encrypt.json
async def encrypt_json(ctx: EncryptionContext, data: dict) -> dict:
    """Encrypt JSON field values, preserving keys and skip fields."""
    result = {}
    for k, v in data.items():
        if k in SKIP_FIELDS or v is None:
            result[k] = v
        else:
            result[k] = ENCRYPTED_PREFIX + await kms_client.encrypt_string(str(v))
    return result

@encryption.decrypt.json
async def decrypt_json(ctx: EncryptionContext, data: dict) -> dict:
    """Decrypt JSON field values by detecting the encrypted prefix."""
    result = {}
    for k, v in data.items():
        if isinstance(v, str) and v.startswith(ENCRYPTED_PREFIX):
            result[k] = await kms_client.decrypt_string(v[len(ENCRYPTED_PREFIX):])
        else:
            result[k] = v
    return result

@encryption.context
async def derive_context(user, ctx: EncryptionContext) -> dict:
    """Derive encryption context from JWT claims."""
    return {**ctx.metadata, "tenant_id": getattr(user, "tenant_id", "default")}

Related Pages

• Principle:Langchain_ai_Langgraph_Data_Encryption