以编程方式使用文档

Agent Server 支持对检查点数据和元数据进行静态加密。您可以在使用单一密钥的基础加密和使用自定义加密的高级用例之间进行选择。

选择加密方法

方法加密内容使用场景
**基础加密**检查点 blob，可选的 JSON 字段单一静态密钥、自动 AES 加密、选择性字段加密
**自定义加密**检查点、线程、运行、助手、计划任务和存储每租户密钥、KMS 集成

基础加密

对于使用单一静态密钥的简单加密，请设置 LANGGRAPH_AES_KEY 环境变量。LangGraph 将自动使用 AES 加密检查点 blob。

1. 添加 pycryptodome 到您的依赖项中，位于 langgraph.json: 

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

  1. 设置 LANGGRAPH_AES_KEY 环境变量为 16、24 或 32 字节密钥（分别对应 AES-128、AES-192 或 AES-256）。

加密 JSON 字段

要同时加密特定的 JSON 字段，请设置 LANGGRAPH_AES_JSON_KEYS 为要加密的密钥的逗号分隔列表：

这些密钥在线程、助手、运行、计划任务和存储数据中出现时都会被加密。

系统字段无法加密： langgraph_version, langgraph_api_version, langgraph_plan, langgraph_host, langgraph_api_url, langgraph_request_id, langgraph_auth_user_id和 langgraph_auth_permissions.

自定义加密

在需要以下情况时使用自定义加密：

  • 每租户密钥隔离 — 为不同客户提供不同的加密密钥
  • KMS 集成 — 使用 AWS KMS、Google Cloud KMS 或 HashiCorp Vault 进行密钥管理、密钥轮换和审计日志记录

工作原理

  1. 在 中配置加密模块路径 langgraph.json
  2. 定义您的加密模块 以及用于 blob 和 JSON 加密的处理程序
  3. 通过 标头传递加密上下文（如租户 ID） X-Encryption-Context header
  4. LangGraph 在存储数据前和检索数据后调用您的处理程序

有关密钥轮换和审计日志记录的生产部署，请参阅 使用 AWS Encryption SDK 的信封加密.

配置

将您的加密模块添加到 langgraph.json:

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

定义您的加密模块

Blob 加密（检查点）

Blob 处理器加密检查点数据——即图形执行中的序列化状态。以下是使用 Fernet （来自 cryptography 库的对称加密方案）的简化示例，使用每个租户的密钥：

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)

该 ctx.metadata 字典来自 X-Encryption-Context 标头，并以明文形式存储在加密数据旁边，以便在解密时使用正确的密钥。

JSON 加密（元数据）

JSON 处理器加密结构化数据，如线程元数据、助手上下文和运行参数。与 blob 加密不同，您可以选择要加密的字段——保留部分未加密以便于搜索和过滤。

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 加密注意事项

用于授权的用户定义字段（例如 tenant_id, owner）通常应保持 **未加密**，用于搜索和过滤的字段也应如此。此外， **某些系统管理的字段永远不会加密**:

  • - 资源标识符（thread_id, run_id, assistant_id, graph_id, checkpoint_id, task_id)
  • - 大多数以 langgraph_ 开头的字段（ langgraph_auth_user)
  • - 必需的检查点元数据（source, step, parents, run_attempt)
  • - 用于调度和编排的内部字段（__after_seconds__, __request_start_time_ms__、大多数以 __pregel)
  • - 运行级别执行限制（max_concurrency, recursion_limit）在运行的 config
  • - 线程 TTL 更新（ttl）在运行的 config.configurable

哪些内容会被加密

JSON 处理器 (@encryption.encrypt.json / @encryption.decrypt.json）被递归应用于以下字段： - thread.metadata, thread.values - assistant.metadata, assistant.context - run.metadata, run.kwargs - cron.metadata, cron.payload - store.value

某些字段被排除在加密之外。 除非另有说明，这些排除适用于嵌套 JSON 对象的每个级别，而不仅仅是根级别。

Blob 处理器 (@encryption.encrypt.blob / @encryption.decrypt.blob) 应用于检查点 blob（图形执行状态）。

从身份验证中派生上下文

不要传递 X-Encryption-Context 显式传递，而是从经过身份验证的用户派生加密上下文：

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"],
    }

此处理器在每次请求的身份验证后运行一次。返回的字典成为 ctx.metadata 用于该请求中的所有加密操作。

传递加密上下文

通过以下方式传递加密上下文 X-Encryption-Context 标头。上下文是你定义的任意数据——你控制着模式，可以包含加密逻辑所需的任何字段（例如， tenant_id, key_version）。上下文在你的处理器中可作为 ctx.metadata 获取，并在解密期间以明文形式存储。

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},
)

使用 AWS Encryption SDK 进行信封加密

对于 AWS 上的生产部署，请使用 AWS Encryption SDK 配合 AWS KMS，或使用你的云提供商中的等效服务。此方法：

  • - 自动处理信封加密（无需手动打包密钥）
  • - 提供密钥轮换和审计日志记录
  • - 将密文绑定到加密上下文（租户隔离）
  • - 在本地缓存数据密钥以避免重复调用 KMS、减少延迟并避免速率限制

完整示例

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

该 encryption_context 通过 KMS 在密码学上绑定到密文——如果上下文不匹配则解密失败。上下文嵌入在密文中，因此解密处理器无需引用 ctx.metadata.

密钥轮换

KMS 自动处理主密钥轮换。当你在 KMS 密钥上启用自动轮换时，旧的加密数据密钥仍可被解密，而新操作使用轮换后的密钥材料。无需重新加密现有数据。

相关内容