Encryption

Payload codecs for encrypting data in Temporal's event history.

EncryptionCodec

EncryptionCodec

EncryptionCodec(
    key: bytes | None = None, *, key_id: str = "default"
)

Bases: PayloadCodec

Temporal PayloadCodec that encrypts/decrypts payloads using AES-GCM.

All data flowing through Temporal's event history (Activity inputs/outputs, Workflow state, Signal payloads) will be encrypted with the provided key.

This is critical for workloads containing PII, as Temporal's event history is queryable and exportable. Without encryption, all channel state (including messages lists containing user data) is stored in plaintext.

Parameters:

Name	Type	Description	Default
key	bytes | None	32-byte encryption key for AES-256-GCM. If not provided, falls back to the LANGGRAPH_TEMPORAL_ENCRYPTION_KEY environment variable (hex-encoded).	None

Raises:

Type	Description
ValueError	If no key is provided and the environment variable is not set.
ValueError	If the key is not exactly 32 bytes.

Source code in langgraph/temporal/encryption.py

def __init__(
    self,
    key: bytes | None = None,
    *,
    key_id: str = "default",
) -> None:
    env_key = os.environ.get("LANGGRAPH_TEMPORAL_ENCRYPTION_KEY")
    if key is not None:
        self._key = key
    elif env_key is not None:
        self._key = bytes.fromhex(env_key)
    else:
        raise ValueError(
            "Encryption key required. Provide `key` parameter or set "
            "LANGGRAPH_TEMPORAL_ENCRYPTION_KEY environment variable "
            "(hex-encoded 32-byte key)."
        )

    if len(self._key) != 32:
        raise ValueError(
            f"Encryption key must be exactly 32 bytes (got {len(self._key)}). "
            "Use `os.urandom(32)` to generate a suitable key."
        )

    self._key_id = key_id

encode async

encode(payloads: Sequence[Payload]) -> list[Payload]

Encrypt payloads before they enter Temporal's event history.

Source code in langgraph/temporal/encryption.py

async def encode(self, payloads: Sequence[Payload]) -> list[Payload]:
    """Encrypt payloads before they enter Temporal's event history."""
    return [self._encrypt_payload(p) for p in payloads]

decode async

decode(payloads: Sequence[Payload]) -> list[Payload]

Decrypt payloads retrieved from Temporal's event history.

Source code in langgraph/temporal/encryption.py

async def decode(self, payloads: Sequence[Payload]) -> list[Payload]:
    """Decrypt payloads retrieved from Temporal's event history."""
    return [self._decrypt_payload(p) for p in payloads]

FernetEncryptionCodec

FernetEncryptionCodec

FernetEncryptionCodec(key: str | bytes)

Bases: PayloadCodec

Temporal PayloadCodec using Fernet symmetric encryption.

Fernet is simpler to use than AES-GCM (key is a URL-safe base64 string) and provides authenticated encryption. Use this for simpler deployments where AES-GCM key management is not needed.

Parameters:

Name	Type	Description	Default
key	str | bytes	Fernet key (URL-safe base64-encoded 32-byte key). Generate with Fernet.generate_key().	required

Source code in langgraph/temporal/encryption.py

def __init__(self, key: str | bytes) -> None:
    self._key = key if isinstance(key, bytes) else key.encode()

encode async

encode(payloads: Sequence[Payload]) -> list[Payload]

Encrypt payloads using Fernet.

Source code in langgraph/temporal/encryption.py

async def encode(self, payloads: Sequence[Payload]) -> list[Payload]:
    """Encrypt payloads using Fernet."""
    return [self._encrypt_payload(p) for p in payloads]

decode async

decode(payloads: Sequence[Payload]) -> list[Payload]

Decrypt payloads using Fernet.

Source code in langgraph/temporal/encryption.py

async def decode(self, payloads: Sequence[Payload]) -> list[Payload]:
    """Decrypt payloads using Fernet."""
    return [self._decrypt_payload(p) for p in payloads]

generate_encryption_key

generate_encryption_key

generate_encryption_key() -> bytes

Generate a random 32-byte encryption key for AES-256-GCM.

Returns:

Type	Description
bytes	A 32-byte random key suitable for use with EncryptionCodec.

Source code in langgraph/temporal/encryption.py

def generate_encryption_key() -> bytes:
    """Generate a random 32-byte encryption key for AES-256-GCM.

    Returns:
        A 32-byte random key suitable for use with `EncryptionCodec`.
    """
    return os.urandom(32)