feat(sdk-py): add decrypt replacement result (#8598)

## Summary

- add `DecryptResult` with plaintext and optional replacement ciphertext
- keep raw decrypt return values backward compatible
- export the result from `langgraph_sdk`

## Merge order

Independent of langgraph-api#3884. Merge and release this SDK change
before LSD-1489 consumes the result in langgraph-api.

## Test plan

- `make format`
- `make lint`
- `make test`

.gitignore

@@ -77,6 +77,10 @@ __pypackages__/
 # Environments
 .env
 .envrc
+*.crt
+*.key
+*.pem
+credentials.json
 .venv
 .venvs
 env/
@@ -98,6 +102,7 @@ dmypy.json
 
 .vercel
 .turbo
+node_modules/
 .editorconfig
 .scratch
 .worktrees/

libs/sdk-py/langgraph_sdk/__init__.py

@@ -1,8 +1,15 @@
 from langgraph_sdk.auth import Auth
 from langgraph_sdk.client import get_client, get_sync_client
 from langgraph_sdk.encryption import Encryption
-from langgraph_sdk.encryption.types import EncryptionContext
+from langgraph_sdk.encryption.types import DecryptResult, EncryptionContext
 
 __version__ = "0.4.2"
 
-__all__ = ["Auth", "Encryption", "EncryptionContext", "get_client", "get_sync_client"]
+__all__ = [
+    "Auth",
+    "DecryptResult",
+    "Encryption",
+    "EncryptionContext",
+    "get_client",
+    "get_sync_client",
+]

libs/sdk-py/langgraph_sdk/encryption/__init__.py

@@ -18,6 +18,9 @@ import warnings
 
 from langgraph_sdk.encryption import types
 
+_BlobDecryptorT = typing.TypeVar("_BlobDecryptorT", bound=types.BlobDecryptor)
+_JsonDecryptorT = typing.TypeVar("_JsonDecryptorT", bound=types.JsonDecryptor)
+
 
 class LangGraphBetaWarning(UserWarning):
     """Warning for beta features in LangGraph SDK."""
@@ -141,7 +144,7 @@ class _DecryptDecorators:
     def __init__(self, parent: Encryption):
         self._parent = parent
 
-    def blob(self, fn: types.BlobDecryptor) -> types.BlobDecryptor:
+    def blob(self, fn: _BlobDecryptorT) -> _BlobDecryptorT:
         """Register a blob decryption handler.
 
         The handler will be called to decrypt opaque data like checkpoint blobs.
@@ -149,7 +152,9 @@ class _DecryptDecorators:
         Example:
             ```python
             @encryption.decrypt.blob
-            async def decrypt_blob(ctx: EncryptionContext, blob: bytes) -> bytes:
+            async def decrypt_blob(
+                ctx: EncryptionContext, blob: bytes
+            ) -> bytes | DecryptResult[bytes]:
                 # Decrypt the blob using your encryption service
                 return decrypted_blob
             ```
@@ -170,13 +175,15 @@ class _DecryptDecorators:
         self._parent._blob_decryptor = fn
         return fn
 
-    def json(self, fn: types.JsonDecryptor) -> types.JsonDecryptor:
+    def json(self, fn: _JsonDecryptorT) -> _JsonDecryptorT:
         """Register the JSON decryption handler.
 
         Example:
             ```python
             @encryption.decrypt.json
-            async def decrypt_json(ctx: EncryptionContext, data: dict) -> dict:
+            async def decrypt_json(
+                ctx: EncryptionContext, data: dict
+            ) -> dict | DecryptResult[dict]:
                 # Decrypt the data
                 return decrypt_data(data)
             ```
@@ -369,7 +376,7 @@ class Encryption:
     """Reference to encryption type definitions.
 
     Provides access to all type definitions used in the encryption system,
-    including EncryptionContext, BlobEncryptor, BlobDecryptor,
+    including EncryptionContext, DecryptResult, BlobEncryptor, BlobDecryptor,
     JsonEncryptor, and JsonDecryptor.
     """
 

libs/sdk-py/langgraph_sdk/encryption/types.py

@@ -9,10 +9,30 @@ from __future__ import annotations
 
 import typing
 from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
 
 Json = dict[str, typing.Any]
 """JSON-serializable dictionary type for structured data encryption."""
 
+T = typing.TypeVar("T")
+
+
+@dataclass(frozen=True, slots=True)
+class DecryptResult(typing.Generic[T]):
+    """Decrypted data and optional replacement ciphertext.
+
+    Return this from a decrypt handler when encrypted data should be replaced,
+    such as after rotating its encryption key. Returning plaintext directly
+    remains supported when no replacement is needed.
+
+    Attributes:
+        plaintext: Decrypted data returned to the caller
+        replacement: New encrypted data to persist in place of the input
+    """
+
+    plaintext: T
+    replacement: T | None = None
+
 
 class EncryptionContext:
     """Context passed to encryption/decryption handlers.
@@ -57,7 +77,9 @@ Returns:
     Awaitable that resolves to encrypted bytes
 """
 
-BlobDecryptor = Callable[[EncryptionContext, bytes], Awaitable[bytes]]
+BlobDecryptor = Callable[
+    [EncryptionContext, bytes], Awaitable[bytes | DecryptResult[bytes]]
+]
 """Handler for decrypting opaque blob data like checkpoints.
 
 Note: Must be an async function. Decryption typically involves I/O operations
@@ -68,7 +90,8 @@ Args:
     blob: The encrypted bytes to decrypt
 
 Returns:
-    Awaitable that resolves to decrypted bytes
+    Awaitable that resolves to decrypted bytes, or a DecryptResult containing
+    decrypted bytes and replacement ciphertext
 """
 
 JsonEncryptor = Callable[[EncryptionContext, Json], Awaitable[Json]]
@@ -101,7 +124,9 @@ Returns:
     Awaitable that resolves to encrypted JSON dictionary
 """
 
-JsonDecryptor = Callable[[EncryptionContext, Json], Awaitable[Json]]
+JsonDecryptor = Callable[
+    [EncryptionContext, Json], Awaitable[Json | DecryptResult[Json]]
+]
 """Handler for decrypting structured JSON data.
 
 Note: Must be an async function. Decryption typically involves I/O operations
@@ -115,7 +140,8 @@ Args:
     data: The encrypted JSON dictionary
 
 Returns:
-    Awaitable that resolves to decrypted JSON dictionary
+    Awaitable that resolves to a decrypted JSON dictionary, or a DecryptResult
+    containing decrypted JSON and replacement ciphertext
 """
 
 if typing.TYPE_CHECKING:

libs/sdk-py/tests/test_encryption.py

@@ -1,8 +1,40 @@
+from collections.abc import Awaitable, Callable
+
 import pytest
 
+from langgraph_sdk import DecryptResult, EncryptionContext
 from langgraph_sdk.encryption import DuplicateHandlerError, Encryption
 
 
+def test_decrypt_result():
+    result = DecryptResult(plaintext=b"plain", replacement=b"rotated")
+
+    assert result.plaintext == b"plain"
+    assert result.replacement == b"rotated"
+    assert DecryptResult(plaintext={"plain": True}).replacement is None
+
+
+def test_decrypt_decorators_preserve_return_types():
+    encryption = Encryption()
+
+    @encryption.decrypt.blob
+    async def blob_dec(_ctx: EncryptionContext, data: bytes) -> bytes:
+        return data
+
+    @encryption.decrypt.json
+    async def json_dec(
+        _ctx: EncryptionContext, data: dict[str, object]
+    ) -> dict[str, object]:
+        return data
+
+    blob_handler: Callable[[EncryptionContext, bytes], Awaitable[bytes]] = blob_dec
+    json_handler: Callable[
+        [EncryptionContext, dict[str, object]], Awaitable[dict[str, object]]
+    ] = json_dec
+    assert blob_handler is blob_dec
+    assert json_handler is json_dec
+
+
 class TestHandlerValidation:
     """Test duplicate handler and signature validation."""