Implementation:Langchain ai Langgraph EncryptedSerializer

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

Overview

A decorator-pattern serializer that wraps any `SerializerProtocol` with AES encryption via a `CipherProtocol`, providing transparent encryption and decryption of checkpoint data.

Description

The EncryptedSerializer class implements `SerializerProtocol` by composing an inner serializer with a cipher. During `dumps_typed`, the object is first serialized by the inner serializer (defaulting to `JsonPlusSerializer`) to obtain a `(type, bytes)` pair, then the bytes are encrypted via the cipher's `encrypt` method. The cipher name is appended to the type string with a `+` separator, producing a composite type like `"json+aes"`. This allows the `loads_typed` method to detect whether data is encrypted by checking for the `+` delimiter, enabling transparent reading of both encrypted and unencrypted data.

The class provides a convenient `from_pycryptodome_aes` class method that creates an `EncryptedSerializer` pre-configured with AES encryption using the PyCryptodome library. The AES key can be supplied directly via the `key` keyword argument or read from the `LANGGRAPH_AES_KEY` environment variable. The default mode is AES-EAX, which provides both confidentiality and authenticity. The encrypted output format concatenates the nonce (16 bytes), authentication tag (16 bytes), and ciphertext, ensuring all components needed for decryption are stored together.

This design supports transparent key rotation and algorithm migration: data encrypted with different ciphers can coexist in the same checkpoint store, and the cipher name in the type string ensures the correct decryption path is always used. Unencrypted legacy data is handled gracefully since the absence of a `+` in the type string triggers a direct pass-through to the inner serializer.

Usage

Use `EncryptedSerializer` when checkpoint data must be encrypted at rest, such as when storing sensitive conversation data, PII, or API keys in checkpoint backends. It is typically passed as the `serde` parameter to a checkpoint saver. The `from_pycryptodome_aes` factory is the recommended way to set up AES encryption with minimal configuration.

Code Reference

Source Location

• Repository: Langchain_ai_Langgraph
• File: libs/checkpoint/langgraph/checkpoint/serde/encrypted.py

Signature

class EncryptedSerializer(SerializerProtocol):
    def __init__(
        self,
        cipher: CipherProtocol,
        serde: SerializerProtocol = JsonPlusSerializer(),
    ) -> None: ...

    def dumps_typed(self, obj: Any) -> tuple[str, bytes]: ...
    def loads_typed(self, data: tuple[str, bytes]) -> Any: ...

    @classmethod
    def from_pycryptodome_aes(
        cls,
        serde: SerializerProtocol = JsonPlusSerializer(),
        **kwargs: Any,
    ) -> "EncryptedSerializer": ...

Import

from langgraph.checkpoint.serde.encrypted import EncryptedSerializer

I/O Contract

Constructor Parameters

Parameter	Type	Required	Description
cipher	`CipherProtocol`	Yes	Encryption/decryption implementation
serde	`SerializerProtocol`	No	Inner serializer (default: `JsonPlusSerializer()`)

dumps_typed

Input	Output	Description
`obj: Any`	`tuple[str, bytes]`	Returns `("{type}+{cipher_name}", encrypted_bytes)`

loads_typed

Input	Output	Description
`data: tuple[str, bytes]`	`Any`	If type contains `+`, decrypts then deserializes; otherwise passes through to inner serde

from_pycryptodome_aes Parameters

Parameter	Type	Required	Description
serde	`SerializerProtocol`	No	Inner serializer (default: `JsonPlusSerializer()`)
key	`bytes`	No	AES key (16, 24, or 32 bytes); falls back to `LANGGRAPH_AES_KEY` env var
mode	AES mode constant	No	PyCryptodome AES mode (default: `AES.MODE_EAX`)

AES-EAX Wire Format

Offset	Size	Content
0	16 bytes	Nonce
16	16 bytes	Authentication tag
32	variable	Ciphertext

Usage Examples

from langgraph.checkpoint.serde.encrypted import EncryptedSerializer

# Using the convenience factory with environment variable
# Set LANGGRAPH_AES_KEY to a 16/24/32-byte string
import os
os.environ["LANGGRAPH_AES_KEY"] = "my-secret-key-16"

serde = EncryptedSerializer.from_pycryptodome_aes()

# Serialize and encrypt
type_str, encrypted = serde.dumps_typed({"user": "Alice", "score": 42})
print(type_str)  # "json+aes"

# Decrypt and deserialize
original = serde.loads_typed((type_str, encrypted))
print(original)  # {"user": "Alice", "score": 42}

# Using with an explicit key
serde_custom = EncryptedSerializer.from_pycryptodome_aes(
    key=b"0123456789abcdef"  # 16-byte AES key
)

# Pass to a checkpoint saver
# from langgraph.checkpoint.sqlite import SqliteSaver
# saver = SqliteSaver.from_conn_string(":memory:", serde=serde)

Related Pages

• Principle:Langchain_ai_Langgraph_Checkpoint_Encryption