Workflows: Add payload compression encoding (#526)

Add zstd compression support for workflow payload encoding.

- add payload compression config and compressed encoding option
- wrap compressed payloads with self-describing msgpack metadata
- decode compressed payloads independently from local encoder config
- run compression before offloading and after partial encryption
- keep compression dependencies optional under workflow_payload_compression
- add tests for compression, encryption/offloading ordering, and invalid payloads

Author: lucasmrdt

docs/models/encodedpayloadoptions.md

@@ -17,3 +17,4 @@ This is an open enum. Unrecognized values will not fail type checks.
 - `"offloaded"`
 - `"encrypted"`
 - `"encrypted-partial"`
+- `"compressed"`

pyproject.toml

@@ -49,6 +49,10 @@ workflow_payload_offloading = [
 workflow_payload_encryption = [
     "cryptography>=41.0.0,<47.0.0",
 ]
+workflow_payload_compression = [
+    "msgpack>=1.1.0,<2.0.0",
+    "zstandard>=0.25.0,<0.26",
+]
 
 
 [project.urls]
@@ -59,6 +63,7 @@ dev = [
     "invoke>=2.2.0,<3",
     "pyyaml>=6.0.2,<7",
     "mypy==1.15.0",
+    "msgpack>=1.1.0,<2.0.0",
     "opentelemetry-sdk (>=1.33.1,<2.0.0)",
     "pylint==3.2.3",
     "pytest>=8.2.2,<9",
@@ -70,6 +75,7 @@ dev = [
     "griffe>=1.7.3,<2",
     "authlib>=1.5.2,<2",
     "websockets >=13.0",
+    "zstandard>=0.25.0,<0.26",
 ]
 lint = [
     "ruff>=0.11.10,<0.12",
@@ -137,6 +143,7 @@ ignore_missing_imports = true
 [[tool.mypy.overrides]]
 module = [
     "jsonpath.*",
+    "msgpack.*",
     "typing_inspect.*",
     "authlib.*",
     "websockets.*",

src/mistralai/client/models/encodedpayloadoptions.py

@@ -11,6 +11,7 @@
         "offloaded",
         "encrypted",
         "encrypted-partial",
+        "compressed",
     ],
     UnrecognizedStr,
 ]

src/mistralai/extra/exceptions.py

@@ -28,6 +28,10 @@ class WorkflowPayloadEncryptionException(MistralClientException):
     """Workflow payload encryption exception"""
 
 
+class WorkflowPayloadCompressionException(MistralClientException):
+    """Workflow payload compression exception"""
+
+
 class RunException(MistralClientException):
     """Conversation run errors."""
 

src/mistralai/extra/tests/fixtures/__init__.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+class InMemoryBlobStorage:
+    def __init__(self) -> None:
+        self.blobs: dict[str, bytes] = {}
+
+    async def __aenter__(self) -> "InMemoryBlobStorage":
+        return self
+
+    async def __aexit__(self, *_args: Any) -> None:
+        pass
+
+    async def upload_blob(self, key: str, content: bytes) -> str:
+        self.blobs[key] = content
+        return key
+
+    async def get_blob(self, key: str) -> bytes:
+        return self.blobs[key]
+
+    async def get_blob_properties(self, key: str) -> dict[str, Any] | None:
+        if key not in self.blobs:
+            return None
+        return {"size": len(self.blobs[key]), "last_modified": "test"}