Implementation:Langchain ai Langgraph Encryption Types

Attribute	Value
Source	`libs/sdk-py/langgraph_sdk/encryption/types.py` (147 lines)
Domain	Encryption, Types
Principle	Data_Encryption
Library	sdk-py
Import	`from langgraph_sdk.encryption.types import EncryptionContext, BlobEncryptor`

Overview

The `types.py` module in the `langgraph_sdk.encryption` package defines the core types for LangGraph's custom at-rest encryption system. It provides the `EncryptionContext` class, typed callback aliases for blob and JSON encryption/decryption handlers, and a `ContextHandler` type for deriving encryption context from authenticated user information.

Description

This module establishes the type contracts that encryption providers must implement to integrate with LangGraph's data-at-rest encryption. All encryption and decryption handlers are async callables, reflecting the expectation that they will interact with external key management services (KMS).

EncryptionContext

`EncryptionContext` -- A slot-based class that provides context to encryption/decryption handlers. It contains non-secret key-value data intended to be sent to an external service managing the actual encryption.

Attributes:

• `model: str | None` -- The model type being encrypted (e.g., `"assistant"`, `"thread"`, `"run"`, `"checkpoint"`).
• `field: str | None` -- The specific field being encrypted (e.g., `"metadata"`, `"context"`, `"kwargs"`, `"values"`).
• `metadata: dict[str, Any]` -- Additional context metadata for encryption decisions. Defaults to `{}`.

Callback Type Aliases

`BlobEncryptor` -- `Callable[[EncryptionContext, bytes], Awaitable[bytes]]`. Encrypts opaque blob data such as checkpoints.

`BlobDecryptor` -- `Callable[[EncryptionContext, bytes], Awaitable[bytes]]`. Decrypts opaque blob data.

`JsonEncryptor` -- `Callable[[EncryptionContext, Json], Awaitable[Json]]`. Encrypts structured JSON data. Designed to support selective field-level encryption where some fields (e.g., `"owner"`) remain in plaintext for search/filtering while others (with a specific prefix like `"my.customer.org/"`) are encrypted.

`JsonDecryptor` -- `Callable[[EncryptionContext, Json], Awaitable[Json]]`. Decrypts structured JSON data. Must be the inverse of the corresponding `JsonEncryptor`.

`ContextHandler` -- `Callable[[BaseUser, EncryptionContext], Awaitable[dict[str, Any]]]`. Derives encryption context metadata from the authenticated user. Called once per request in middleware (after auth), allowing encryption context to be derived from JWT claims or user properties. The return value becomes `ctx.metadata` for subsequent operations.

`Json` -- Type alias for `dict[str, Any]`, representing JSON-serializable dictionary data.

Important Design Notes

• All handlers must be async functions, as encryption typically involves I/O operations with external KMS services.
• Encrypted field values cannot be reliably searched, since most real-world encryption implementations use nonces (non-deterministic encryption). Only unencrypted fields can be used in search queries.
• `ContextHandler` runs when `ctx.model` and `ctx.field` are `None`, since it executes once per request before specific model/field operations.

Usage

from langgraph_sdk.encryption.types import (
    EncryptionContext,
    BlobEncryptor,
    BlobDecryptor,
    JsonEncryptor,
    JsonDecryptor,
)

async def my_blob_encryptor(ctx: EncryptionContext, data: bytes) -> bytes:
    # Call external KMS to encrypt
    return await kms_encrypt(data, context=ctx.metadata)

async def my_blob_decryptor(ctx: EncryptionContext, data: bytes) -> bytes:
    # Call external KMS to decrypt
    return await kms_decrypt(data, context=ctx.metadata)

Code Reference

EncryptionContext Class

Attribute	Type	Default	Description
`model`	None`	`None`	The model type being encrypted (e.g., `"assistant"`, `"thread"`).
`field`	None`	`None`	The specific field being encrypted (e.g., `"metadata"`, `"values"`).
`metadata`	`dict[str, Any]`	`{}`	Additional context metadata for encryption decisions.

Method	Signature	Description
`__init__`	`(model=None, metadata=None, field=None)`	Initialize the encryption context.
`__repr__`	`() -> str`	Returns `"EncryptionContext(model=..., field=..., metadata=...)"`.

Type Aliases

Name	Type Signature	Description
`Json`	`dict[str, Any]`	JSON-serializable dictionary type.
`BlobEncryptor`	`Callable[[EncryptionContext, bytes], Awaitable[bytes]]`	Encrypts raw byte data.
`BlobDecryptor`	`Callable[[EncryptionContext, bytes], Awaitable[bytes]]`	Decrypts raw byte data.
`JsonEncryptor`	`Callable[[EncryptionContext, Json], Awaitable[Json]]`	Encrypts structured JSON dictionaries.
`JsonDecryptor`	`Callable[[EncryptionContext, Json], Awaitable[Json]]`	Decrypts structured JSON dictionaries.
`ContextHandler`	`Callable[[BaseUser, EncryptionContext], Awaitable[dict[str, Any]]]`	Derives encryption metadata from authenticated user.

I/O Contract

Aspect	Detail
Input	`EncryptionContext` provides model, field, and metadata to handlers. `BlobEncryptor`/`BlobDecryptor` receive raw bytes. `JsonEncryptor`/`JsonDecryptor` receive JSON dicts. `ContextHandler` receives `BaseUser` and `EncryptionContext`.
Output	Blob handlers return encrypted/decrypted bytes. JSON handlers return encrypted/decrypted dicts. `ContextHandler` returns a metadata dict.
Side Effects	Handlers typically perform network I/O to external KMS services.
Async Requirement	All handlers must be async functions (return `Awaitable`).

Usage Examples

Selective JSON Field Encryption

from langgraph_sdk.encryption.types import EncryptionContext, Json

PREFIX = "my.customer.org/"

async def selective_json_encryptor(ctx: EncryptionContext, data: Json) -> Json:
    result = {}
    for key, value in data.items():
        if key.startswith(PREFIX):
            result[key] = await kms_encrypt_value(value, ctx.metadata)
        else:
            result[key] = value
    return result

# Input:  {"owner": "user123", "my.customer.org/email": "john@example.com"}
# Output: {"owner": "user123", "my.customer.org/email": "ENCRYPTED_VALUE"}

Context Handler from JWT Claims

from langgraph_sdk.encryption.types import EncryptionContext, ContextHandler

async def derive_context(user, ctx: EncryptionContext) -> dict:
    return {
        "tenant_id": user.tenant_id,
        "key_version": "v2",
        **ctx.metadata,
    }

Blob Encryption for Checkpoints

from langgraph_sdk.encryption.types import EncryptionContext

async def encrypt_checkpoint(ctx: EncryptionContext, data: bytes) -> bytes:
    key = await get_encryption_key(ctx.metadata.get("tenant_id"))
    return aes_encrypt(data, key)

async def decrypt_checkpoint(ctx: EncryptionContext, data: bytes) -> bytes:
    key = await get_encryption_key(ctx.metadata.get("tenant_id"))
    return aes_decrypt(data, key)

Related Pages

• Langchain_ai_Langgraph_Auth_HTTPException -- Auth exceptions used in the same authentication pipeline.
• Langchain_ai_Langgraph_ServerRuntime -- Server runtime providing user context for encryption handlers.
• Langchain_ai_Langgraph_SDK_Errors -- SDK error hierarchy for communication failures.