Add encryption at rest

Agent Server supports encryption at rest for checkpoint data and metadata. You can choose between basic encryption with a single key or custom encryption for advanced use cases.

Choosing an encryption method

Method	What’s encrypted	Use case
Basic encryption	Checkpoint blobs, optionally JSON fields	Single static key, automatic AES encryption, selective field encryption
Custom encryption	Checkpoints, threads, runs, assistants, crons and stores	Per-tenant keys, KMS integration

Basic encryption

For simple encryption with a single static key, set the LANGGRAPH_AES_KEY environment variable. LangGraph will automatically encrypt checkpoint blobs using AES.

1. Add pycryptodome to your dependencies in langgraph.json:

   {
     "dependencies": [".", "pycryptodome"],
     "graphs": {
       "agent": "./agent.py:graph"
     }
   }
   

2. Set the LANGGRAPH_AES_KEY environment variable to a 16, 24, or 32-byte key (for AES-128, AES-192, or AES-256 respectively).

Encrypting JSON fields

To also encrypt specific JSON fields, set LANGGRAPH_AES_JSON_KEYS to a comma-separated list of keys to encrypt:

export LANGGRAPH_AES_KEY="your-16-24-or-32-byte-key"
export LANGGRAPH_AES_JSON_KEYS="api_key,secret_token,user_credentials"

These keys are encrypted wherever they appear in thread, assistant, run, cron, and store data.

Encrypted fields cannot be searched or filtered.

System fields cannot be encrypted: langgraph_version, langgraph_api_version, langgraph_plan, langgraph_host, langgraph_api_url, langgraph_request_id, langgraph_auth_user_id, and langgraph_auth_permissions.

Custom encryption

Requires Agent Server version 0.6.22+ and Python SDK version langgraph-sdk>=0.3.1.

Agent Server versions 0.5.34–0.6.21 included a pre-release version of custom encryption. Data encrypted with these versions will be corrupted when upgrading to 0.6.22+. Do not use custom encryption on these versions.

Only use custom encryption if basic encryption doesn’t meet your needs. Custom encryption requires you to implement and maintain encryption handlers, and adds operational complexity. If you only need a single static key with optional selective field encryption, use basic encryption instead.

Use custom encryption when you need:

• Per-tenant key isolation — different encryption keys for different customers
• KMS integration — AWS KMS, Google Cloud KMS, or HashiCorp Vault for key management, rotation, and audit logging

How it works

1. Configure the encryption module path in langgraph.json
2. Define your encryption module with handlers for blob and JSON encryption
3. Pass encryption context (like tenant ID) via the X-Encryption-Context header
4. LangGraph calls your handlers before storing and after retrieving data

For production deployments with key rotation and audit logging, see Envelope encryption with AWS Encryption SDK.

Configuration

Add your encryption module to langgraph.json:

{
  "dependencies": ["."],
  "graphs": {
    "agent": "./agent.py:graph"
  },
  "encryption": {
    "path": "./encryption.py:encryption"
  }
}

If you’re migrating from basic encryption, keep LANGGRAPH_AES_KEY configured. Custom encryption handles new writes while existing AES-encrypted data remains readable.

Defining your encryption module

Blob encryption (checkpoints)

Blob handlers encrypt checkpoint data—the serialized state from graph execution. Here’s a simplified example using per-tenant keys with Fernet (a symmetric encryption scheme from the cryptography library):

import os
from cryptography.fernet import Fernet
from langgraph_sdk import Encryption, EncryptionContext

encryption = Encryption()

# In production, fetch from a secrets manager
TENANT_KEYS = {
    "tenant-a": Fernet(os.environ["TENANT_A_KEY"]),
    "tenant-b": Fernet(os.environ["TENANT_B_KEY"]),
}


def _get_fernet(ctx: EncryptionContext) -> Fernet:
    tenant_id = ctx.metadata.get("tenant_id")
    if not tenant_id or tenant_id not in TENANT_KEYS:
        raise ValueError(f"Unknown tenant: {tenant_id}")
    return TENANT_KEYS[tenant_id]


@encryption.encrypt.blob
async def encrypt_blob(ctx: EncryptionContext, data: bytes) -> bytes:
    return _get_fernet(ctx).encrypt(data)


@encryption.decrypt.blob
async def decrypt_blob(ctx: EncryptionContext, data: bytes) -> bytes:
    return _get_fernet(ctx).decrypt(data)

The ctx.metadata dict comes from the X-Encryption-Context header and is stored in plaintext alongside encrypted data, so the correct key is used on decryption.

JSON encryption (metadata)

JSON handlers encrypt structured data like thread metadata, assistant context, and run kwargs. Unlike blob encryption, you choose which fields to encrypt—keeping some unencrypted for search and filtering.

import json
import os
from cryptography.fernet import Fernet
from langgraph_sdk import Encryption, EncryptionContext

encryption = Encryption()

TENANT_KEYS = {
    "tenant-a": Fernet(os.environ["TENANT_A_KEY"]),
    "tenant-b": Fernet(os.environ["TENANT_B_KEY"]),
}

SKIP_FIELDS = {
    "tenant_id", "owner",
    "run_id", "thread_id", "graph_id", "assistant_id", "user_id", "checkpoint_id",
    "source", "step", "parents", "run_attempt",
    "langgraph_version", "langgraph_api_version", "langgraph_plan", "langgraph_host",
    "langgraph_api_url", "langgraph_request_id", "langgraph_auth_user",
    "langgraph_auth_user_id", "langgraph_auth_permissions",
}
ENCRYPTED_PREFIX = "encrypted:"


def _get_fernet(ctx: EncryptionContext) -> Fernet:
    tenant_id = ctx.metadata.get("tenant_id")
    if not tenant_id or tenant_id not in TENANT_KEYS:
        raise ValueError(f"Unknown tenant: {tenant_id}")
    return TENANT_KEYS[tenant_id]


@encryption.encrypt.json
async def encrypt_json(ctx: EncryptionContext, data: dict) -> dict:
    fernet = _get_fernet(ctx)
    result = {}
    for k, v in data.items():
        if k in SKIP_FIELDS or v is None:
            result[k] = v
        else:
            value_json = json.dumps(v)
            encrypted = fernet.encrypt(value_json.encode()).decode()
            result[k] = ENCRYPTED_PREFIX + encrypted
    return result


@encryption.decrypt.json
async def decrypt_json(ctx: EncryptionContext, data: dict) -> dict:
    fernet = _get_fernet(ctx)
    result = {}
    for k, v in data.items():
        if isinstance(v, str) and v.startswith(ENCRYPTED_PREFIX):
            encrypted_value = v[len(ENCRYPTED_PREFIX):]
            decrypted = fernet.decrypt(encrypted_value.encode()).decode()
            result[k] = json.loads(decrypted)
        else:
            result[k] = v
    return result

JSON encryption considerations

Encrypted fields cannot be searched or filtered. Design your metadata schema so that fields you need to query remain unencrypted.

JSON encryptors must preserve key structure. SQL JSONB merge operations work at the key level. Encryptors that change keys—whether by consolidating fields (e.g., moving sensitive data into __encrypted__) or by encrypting key names themselves—cause data loss during merges. Use per-key encryption: transform values in-place while preserving keys.

Migration consideration: Use a recognizable prefix or format in encrypted values so your decryptor can detect and skip unencrypted data. This allows you to encrypt additional fields in the future without re-encrypting existing records. The example above uses this pattern.

Performance consideration: Per-key encryption means one encryption call per field. If your encryption involves round-trips to an external service (e.g., KMS), this can significantly impact latency. Consider caching data keys locally or using envelope encryption where you encrypt a local data key with KMS and use it for multiple fields.

User-defined fields for authorization (e.g., tenant_id, owner) should generally be left unencrypted, as should fields used for search and filtering. Additionally, some system-managed fields will never be encrypted:

• Resource identifiers (thread_id, run_id, assistant_id, graph_id, checkpoint_id, task_id)
• Most fields beginning with langgraph_ (except for langgraph_auth_user)
• Required checkpoint metadata (source, step, parents, run_attempt)
• Internal fields used for scheduling and orchestration (__after_seconds__, __request_start_time_ms__, most fields beginning with __pregel)
• Run-level execution limits (max_concurrency, recursion_limit) specified in a run’s config
• Thread TTL updates (ttl) specified in a run’s config.configurable

What gets encrypted

JSON handlers (@encryption.encrypt.json / @encryption.decrypt.json) are applied recursively to the following fields:

• thread.metadata, thread.values
• assistant.metadata, assistant.context
• run.metadata, run.kwargs
• cron.metadata, cron.payload
• store.value

Some fields are excluded from encryption. Unless otherwise noted, these exclusions apply at every level of a nested JSON object, not just the root level.

Blob handlers (@encryption.encrypt.blob / @encryption.decrypt.blob) are applied to checkpoint blobs (graph execution state).

Deriving context from authentication

Instead of passing X-Encryption-Context explicitly, derive encryption context from the authenticated user:

from langgraph_sdk import Encryption, EncryptionContext
from starlette.authentication import BaseUser

encryption = Encryption()

@encryption.context
async def get_encryption_context(user: BaseUser, ctx: EncryptionContext) -> dict:
    return {
        **ctx.metadata,
        "tenant_id": user["tenant_id"],
    }

This handler runs once per request after authentication. The returned dict becomes ctx.metadata for all encryption operations in that request.

Passing encryption context

Pass encryption context via the X-Encryption-Context header. The context is arbitrary data that you define—you control the schema and can include any fields your encryption logic needs (e.g., tenant_id, key_version). The context is available in your handlers as ctx.metadata and is stored in plaintext for use during decryption.

import base64
import json
from langgraph_sdk import get_client

encryption_context = base64.b64encode(
    json.dumps({"tenant_id": "tenant-a"}).encode()
).decode()

client = get_client(url="http://localhost:2024")

result = await client.runs.wait(
    thread_id=None,
    assistant_id="agent",
    input={"messages": [{"role": "user", "content": "Hello"}]},
    headers={"X-Encryption-Context": encryption_context},
)

The encryption context is stored in plaintext. On decryption, it’s automatically restored—callers don’t need to pass the header when reading.

Envelope encryption with AWS Encryption SDK

For production deployments on AWS, use the AWS Encryption SDK with AWS KMS, or an equivalent within your cloud provider. This approach:

• Handles envelope encryption automatically (no manual key packing)
• Provides key rotation and audit logging
• Binds ciphertext to encryption context (tenant isolation)
• Caches data keys locally to avoid repeated KMS calls, latency and rate limits

Complete example

import base64
import json
import os

import aws_encryption_sdk
from aws_encryption_sdk import (
    CachingCryptoMaterialsManager,
    CommitmentPolicy,
    LocalCryptoMaterialsCache,
    StrictAwsKmsMasterKeyProvider,
)
from langgraph_sdk import Encryption, EncryptionContext

encryption = Encryption()

# The SDK uses envelope encryption: one KMS API call generates a data key,
# then encrypts/decrypts locally. The cache reuses data keys across operations.
client = aws_encryption_sdk.EncryptionSDKClient(
    commitment_policy=CommitmentPolicy.REQUIRE_ENCRYPT_REQUIRE_DECRYPT
)
key_provider = StrictAwsKmsMasterKeyProvider(key_ids=[os.environ["KMS_KEY_ARN"]])
cache = LocalCryptoMaterialsCache(capacity=100)
cmm = CachingCryptoMaterialsManager(
    master_key_provider=key_provider,
    cache=cache,
    max_age=300.0,
    max_messages_encrypted=100,
)

SKIP_FIELDS = {
    "tenant_id", "owner",
    "run_id", "thread_id", "graph_id", "assistant_id", "user_id", "checkpoint_id",
    "source", "step", "parents", "run_attempt",
    "langgraph_version", "langgraph_api_version", "langgraph_plan", "langgraph_host",
    "langgraph_api_url", "langgraph_request_id", "langgraph_auth_user",
    "langgraph_auth_user_id", "langgraph_auth_permissions",
}
ENCRYPTED_PREFIX = "encrypted:"


@encryption.encrypt.blob
async def encrypt_blob(ctx: EncryptionContext, data: bytes) -> bytes:
    ciphertext, _ = client.encrypt(
        source=data,
        materials_manager=cmm,
        encryption_context={"tenant_id": ctx.metadata["tenant_id"]},
    )
    return ciphertext


@encryption.decrypt.blob
async def decrypt_blob(ctx: EncryptionContext, data: bytes) -> bytes:
    plaintext, _ = client.decrypt(source=data, key_provider=key_provider)
    return plaintext


@encryption.encrypt.json
async def encrypt_json(ctx: EncryptionContext, data: dict) -> dict:
    tenant_id = ctx.metadata["tenant_id"]
    result = {}
    for k, v in data.items():
        if k in SKIP_FIELDS or v is None:
            result[k] = v
        else:
            ciphertext, _ = client.encrypt(
                source=json.dumps(v).encode(),
                materials_manager=cmm,
                encryption_context={"tenant_id": tenant_id},
            )
            result[k] = ENCRYPTED_PREFIX + base64.b64encode(ciphertext).decode()
    return result


@encryption.decrypt.json
async def decrypt_json(ctx: EncryptionContext, data: dict) -> dict:
    result = {}
    for k, v in data.items():
        if isinstance(v, str) and v.startswith(ENCRYPTED_PREFIX):
            ciphertext = base64.b64decode(v[len(ENCRYPTED_PREFIX):])
            plaintext, _ = client.decrypt(source=ciphertext, key_provider=key_provider)
            result[k] = json.loads(plaintext.decode())
        else:
            result[k] = v
    return result

The encryption_context is cryptographically bound to the ciphertext via KMS—decryption fails if the context doesn’t match. The context is embedded in the ciphertext, so decrypt handlers don’t need to reference ctx.metadata.

Key rotation

KMS handles master key rotation automatically. When you enable automatic rotation on your KMS key, old encrypted data keys can still be decrypted while new operations use the rotated key material. No re-encryption of existing data is required.

Related

• Custom authentication

---

Connect these docs to Claude, VSCode, and more via MCP for real-time answers.

Edit this page on GitHub or file an issue.