langgraph_sdk

from langgraph_sdk import Auth

my_auth = Auth()

@my_auth.authenticate
async def authenticate(authorization: str) -> Auth.types.MinimalUserDict:
    user = await verify_token(authorization)  # Your token verification logic
    if not user:
        raise Auth.exceptions.HTTPException(
            status_code=401, detail="Unauthorized"
        )
    return {
        "identity": user["id"],
        "permissions": user.get("permissions", []),
    }

# Default deny: reject all requests that don't have a specific handler
@my_auth.on
async def deny_all(ctx: Auth.types.AuthContext, value: Any) -> False:
    return False

# Allow users to create threads with their own identity as owner
@my_auth.on.threads.create
async def allow_thread_create(
    ctx: Auth.types.AuthContext, value: Auth.types.on.threads.create.value
):
    metadata = value.setdefault("metadata", {})
    metadata["owner"] = ctx.user.identity

# Allow users to read and search their own threads
@my_auth.on.threads.read
async def allow_thread_read(
    ctx: Auth.types.AuthContext, value: Auth.types.on.threads.read.value
) -> Auth.types.FilterType:
    return {"owner": ctx.user.identity}

@my_auth.on.threads.search
async def allow_thread_search(
    ctx: Auth.types.AuthContext, value: Auth.types.on.threads.search.value
) -> Auth.types.FilterType:
    return {"owner": ctx.user.identity}

# Scope all store operations to the user's namespace
@my_auth.on.store
async def scope_store(ctx: Auth.types.AuthContext, value: Auth.types.on.store.value):
    namespace = tuple(value["namespace"]) if value.get("namespace") else ()
    if not namespace or namespace[0] != ctx.user.identity:
        namespace = (ctx.user.identity, *namespace)
    value["namespace"] = namespace

1. Authentication (your @auth.authenticate handler) is performed first on every request
2. For authorization, the most specific matching handler is called:
   • If a handler exists for the exact resource and action, it is used (e.g., @auth.on.threads.create)
   • Otherwise, if a handler exists for the resource with any action, it is used (e.g., @auth.on.threads)
   • Finally, if no specific handlers match, the global handler is used (e.g., @auth.on)
   • If no global handler is set, the request is accepted

This allows you to set default behavior with a global handler while overriding specific routes as needed.

JSON encryptors must not add or remove keys from the input dict. Only values may be transformed. This constraint is enforced at runtime by the server and exists because SQL JSONB merge operations (used for partial updates) work at the key level.

Correct (per-key encryption):

# Input:  {"secret": "value", "plain": "x"}
# Output: {"secret": "<encrypted>", "plain": "x"}  ✓ Keys preserved

Incorrect (key consolidation):

# Input:  {"secret": "value", "plain": "x"}
# Output: {"__encrypted__": "<blob>", "plain": "x"}  ✗ Key changed

If your encryptor needs to store auxiliary data (DEK, IV, etc.), embed it within the encrypted value itself, not as separate keys.

from langgraph_sdk import Encryption, EncryptionContext

my_encryption = Encryption()

SKIP_FIELDS = {"tenant_id", "owner", "thread_id", "assistant_id"}
ENCRYPTED_PREFIX = "encrypted:"

@my_encryption.encrypt.blob
async def encrypt_blob(ctx: EncryptionContext, blob: bytes) -> bytes:
    return your_encrypt_bytes(blob)

@my_encryption.decrypt.blob
async def decrypt_blob(ctx: EncryptionContext, blob: bytes) -> bytes:
    return your_decrypt_bytes(blob)

@my_encryption.encrypt.json
async def encrypt_json(ctx: EncryptionContext, data: dict) -> dict:
    result = {}
    for k, v in data.items():
        if k in SKIP_FIELDS or v is None:
            result[k] = v
        else:
            result[k] = ENCRYPTED_PREFIX + your_encrypt_string(v)
    return result

@my_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):
            result[k] = your_decrypt_string(v[len(ENCRYPTED_PREFIX):])
        else:
            result[k] = v
    return result

The ctx.model and ctx.field attributes tell you which model type and specific field is being encrypted, allowing different logic:

@my_encryption.encrypt.json
async def encrypt_json(ctx: EncryptionContext, data: dict) -> dict:
    if ctx.field == "metadata":
        # Metadata - standard encryption
        return encrypt_standard(data)
    elif ctx.field == "values":
        # Thread values - more sensitive, use stronger encryption
        return encrypt_sensitive(data)
    else:
        return encrypt_standard(data)

ENCRYPTED_PREFIX = "enc:"

@my_encryption.encrypt.json
async def encrypt_json(ctx: EncryptionContext, data: dict) -> dict:
    # Encrypt different keys depending on the field
    if ctx.field == "context":
        keys_to_encrypt = {"api_key", "secret_token"}
    else:
        keys_to_encrypt = {"email", "ssn"}
    return {
        k: ENCRYPTED_PREFIX + encrypt(v) if k in keys_to_encrypt else v
        for k, v in data.items()
    }

@my_encryption.decrypt.json
async def decrypt_json(ctx: EncryptionContext, data: dict) -> dict:
    # Decrypt ANY value with the prefix, regardless of model/field
    return {
        k: decrypt(v[len(ENCRYPTED_PREFIX):])
           if isinstance(v, str) and v.startswith(ENCRYPTED_PREFIX)
           else v
        for k, v in data.items()
    }