저장 데이터 암호화 추가하기

Agent Server는 체크포인트 데이터와 메타데이터에 대한 저장 데이터 암호화를 지원해요. 단일 키를 사용한 기본 암호화 또는 고급 사용 사례를 위한 커스텀 암호화 중에서 선택할 수 있어요.

출처: 문서

본문

암호화 방법 선택

방법	암호화되는 것	사용 사례
기본 암호화	체크포인트 blob, 선택적으로 JSON 필드	단일 정적 키, 자동 AES 암호화, 선택적 필드 암호화
커스텀 암호화	체크포인트, 스레드, 런, 어시스턴트, 크론 및 스토어	테넌트별 키, KMS 통합

기본 암호화

단일 정적 키를 사용한 간단한 암호화를 위해 LANGGRAPH_AES_KEY 환경 변수를 설정하세요. LangGraph는 AES를 사용해 체크포인트 blob을 자동으로 암호화해요.

langgraph.json의 의존성에 pycryptodome을 추가하세요:

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

LANGGRAPH_AES_KEY 환경 변수를 16, 24 또는 32바이트 키로 설정하세요 (각각 AES-128, AES-192, AES-256용).

JSON 필드 암호화

특정 JSON 필드도 암호화하려면 LANGGRAPH_AES_JSON_KEYS를 암호화할 키의 쉼표로 구분된 목록으로 설정하세요:

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

이 키들은 스레드, 어시스턴트, 런, 크론 및 스토어 데이터에 나타나는 곳마다 암호화돼요.

암호화된 필드는 검색하거나 필터링할 수 없어요.

시스템 필드는 암호화할 수 없어요: langgraph_version, langgraph_api_version, langgraph_plan, langgraph_host, langgraph_api_url, langgraph_request_id, langgraph_auth_user_id 및 langgraph_auth_permissions.

커스텀 암호화

Agent Server 버전 0.6.22+와 Python SDK 버전 langgraph-sdk>=0.3.1이 필요해요.

Agent Server 버전 0.5.34–0.6.21에는 커스텀 암호화의 사전 릴리스 버전이 포함돼 있어요. 이 버전들로 암호화된 데이터는 0.6.22+로 업그레이드하면 손상돼요. 이 버전에서는 커스텀 암호화를 사용하지 마세요.

기본 암호화가 요구 사항을 충족하지 못하는 경우에만 커스텀 암호화를 사용하세요. 커스텀 암호화는 암호화 핸들러를 구현하고 유지 관리해야 하며 운영 복잡성을 추가해요. 선택적 선택 필드 암호화가 있는 단일 정적 키만 필요하다면 대신 기본 암호화를 사용하세요.

다음이 필요할 때 커스텀 암호화를 사용하세요:

테넌트별 키 격리 — 고객마다 다른 암호화 키
KMS 통합 — 키 관리, 교체, 감사 로깅을 위한 AWS KMS, Google Cloud KMS 또는 HashiCorp Vault

작동 방식

langgraph.json에서 암호화 모듈 경로 구성
blob 및 JSON 암호화를 위한 암호화 모듈 정의
X-Encryption-Context 헤더를 통해 암호화 컨텍스트 전달 (예: 테넌트 ID)
LangGraph가 데이터를 저장하기 전과 검색한 후에 핸들러를 호출

키 교체와 감사 로깅이 있는 프로덕션 배포는 AWS Encryption SDK를 사용한 봉투 암호화를 참고하세요.

구성

langgraph.json에 암호화 모듈을 추가하세요:

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

기본 암호화에서 마이그레이션 중이라면 LANGGRAPH_AES_KEY를 계속 구성해 두세요. 커스텀 암호화가 새 쓰기를 처리하는 동안 기존 AES 암호화 데이터는 읽을 수 있는 상태로 유지돼요.

암호화 모듈 정의

Blob 암호화 (체크포인트)

Blob 핸들러는 체크포인트 데이터, 즉 그래프 실행에서 직렬화된 상태를 암호화해요. 테넌트별 키를 사용한 간단한 예시입니다 (cryptography 라이브러리의 대칭 암호화 스킴인 Fernet 사용):

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)

ctx.metadata dict는 X-Encryption-Context 헤더에서 오며 암호화된 데이터와 함께 평문으로 저장되므로, 복호화 시 올바른 키가 사용돼요.

JSON 암호화 (메타데이터)

JSON 핸들러는 스레드 메타데이터, 어시스턴트 컨텍스트, 런 kwargs 같은 구조화된 데이터를 암호화해요. blob 암호화와 달리 어떤 필드를 암호화할지 선택하고, 일부는 검색과 필터링을 위해 암호화하지 않고 남겨둘 수 있어요.

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 암호화 고려 사항

암호화된 필드는 검색하거나 필터링할 수 없어요. 쿼리해야 하는 필드가 암호화되지 않은 상태로 남도록 메타데이터 스키마를 설계하세요.

JSON 암호화기는 키 구조를 보존해야 해요. SQL JSONB 병합 연산은 키 수준에서 작동해요. 필드를 통합(예: 민감한 데이터를 __encrypted__로 이동)하거나 키 이름 자체를 암호화하는 등 키를 변경하는 암호화기는 병합 중 데이터 손실을 일으켜요. 키별 암호화를 사용하세요: 키를 보존하면서 값을 제자리에서 변환하세요.

마이그레이션 고려 사항: 복호화기가 암호화되지 않은 데이터를 감지하고 건너뛸 수 있도록 암호화된 값에 인식 가능한 접두사 또는 형식을 사용하세요. 이렇게 하면 기존 레코드를 다시 암호화하지 않고도 향후 추가 필드를 암호화할 수 있어요. 위 예시는 이 패턴을 사용해요.

성능 고려 사항: 키별 암호화는 필드당 암호화 호출 한 번을 의미해요. 암호화에 외부 서비스(예: KMS)로의 왕복이 포함되면 대기 시간에 큰 영향을 줄 수 있어요. 데이터 키를 로컬에 캐시하거나 봉투 암호화(로컬 데이터 키를 KMS로 암호화하고 여러 필드에 사용)를 고려하세요.

권한 부여를 위한 사용자 정의 필드(예: 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로 시작하는 대부분의 필드)
런의 config에 지정된 런 수준 실행 제한(max_concurrency, recursion_limit)
런의 config.configurable에 지정된 스레드 TTL 업데이트(ttl)

무엇이 암호화되는가

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

이 핸들러는 인증 후 요청당 한 번 실행돼요. 반환된 dict는 해당 요청의 모든 암호화 연산에 대해 ctx.metadata가 돼요.

암호화 컨텍스트 전달

X-Encryption-Context 헤더를 통해 암호화 컨텍스트를 전달하세요. 컨텍스트는 여러분이 정의하는 임의의 데이터로, 스키마를 제어하고 암호화 로직에 필요한 모든 필드(예: tenant_id, key_version)를 포함할 수 있어요. 컨텍스트는 핸들러에서 ctx.metadata로 사용할 수 있고 복호화 중 사용을 위해 평문으로 저장돼요.

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

암호화 컨텍스트는 평문으로 저장돼요. 복호화 시 자동으로 복원되므로 호출자는 읽을 때 헤더를 전달할 필요가 없어요.

AWS Encryption SDK를 사용한 봉투 암호화

AWS의 프로덕션 배포에서는 AWS KMS와 함께 AWS Encryption SDK를 사용하거나 클라우드 프로바이더 내에서 이에 상응하는 것을 사용하세요. 이 접근 방식은:

봉투 암호화를 자동으로 처리 (수동 키 패킹 불필요)
키 교체 및 감사 로깅 제공
암호문을 암호화 컨텍스트에 바인딩 (테넌트 격리)
반복 KMS 호출, 대기 시간 및 속도 제한을 피하기 위해 데이터 키를 로컬에 캐시

완전한 예시

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

encryption_context는 KMS를 통해 암호문에 암호화적으로 바인딩되므로, 컨텍스트가 일치하지 않으면 복호화가 실패해요. 컨텍스트는 암호문에 내장되므로 복호화 핸들러는 ctx.metadata를 참조할 필요가 없어요.

키 교체

KMS는 마스터 키 교체를 자동으로 처리해요. KMS 키에서 자동 교체를 활성화하면, 새 작업이 교체된 키 자료를 사용하는 동안 이전에 암호화된 데이터 키는 여전히 복호화할 수 있어요. 기존 데이터를 다시 암호화할 필요는 없어요.

더 알아보기 (Learn more)

커스텀 인증

저장 데이터 암호화 추가하기