feat!: standardize remote execution on JSON

README.md

@@ -84,6 +84,24 @@ make setup
 make test
 ```
 
+## Remote values
+
+Queue and task calls use JSON for arguments, return values, and streamed
+chunks. Values may contain `None`, booleans, strings, integers, finite
+floats, lists, and dictionaries with string keys. Convert tuples to lists,
+dates to strings, and model objects to JSON-compatible dictionaries before
+calling `.remote()` or `.spawn()`. Store large binary data on a volume or
+in object storage and pass its path or URL.
+
+The same validation applies in live development, deployed workers, and
+same-worker remote calls. Unsupported values, circular references, and
+non-finite numbers produce explicit errors. `.local()` is a direct Python
+call and does not apply the remote serialization contract. Custom ASGI
+applications retain their own HTTP response types.
+
+The SDK and `runpod-sdk-runtime` share the JSON protocol implementation;
+use compatible releases of both packages when building worker images.
+
 ## ⚡ | Serverless Worker (SDK)
 
 This python package can also be used to create a serverless worker that can be deployed to Runpod as a custom endpoint API.

requirements.txt

@@ -5,7 +5,6 @@ backoff >= 2.2.1
 boto3 >= 1.43.118
 certifi >= 2026.10.19
 click >= 8.5.0
-cloudpickle >= 3.1.2
 colorama >= 0.4.6, < 0.4.7
 cryptography >= 50.0.2
 fastapi[all] >= 0.162.0

runpod/_serialization.py

@@ -0,0 +1,39 @@
+"""strict json values shared by remote callers and workers."""
+
+import math
+from typing import Any
+
+
+def validate_json(value: Any, label: str = "value") -> None:
+    """require finite json primitives, lists, and string-keyed dictionaries."""
+    active = set()
+
+    def visit(item: Any, path: str) -> None:
+        if item is None or isinstance(item, (str, bool, int)):
+            return
+        if isinstance(item, float):
+            if not math.isfinite(item):
+                raise TypeError(f"{path} must be a finite number")
+            return
+        if not isinstance(item, (list, dict)):
+            raise TypeError(f"{path} has unsupported type {type(item).__name__}")
+        identity = id(item)
+        if identity in active:
+            raise TypeError(f"{path} contains a circular reference")
+        active.add(identity)
+        try:
+            if isinstance(item, list):
+                for index, child in enumerate(item):
+                    visit(child, f"{path}[{index}]")
+            else:
+                for key, child in item.items():
+                    if not isinstance(key, str):
+                        raise TypeError(f"{path} must have string keys")
+                    visit(child, f"{path}[{key!r}]")
+        finally:
+            active.remove(identity)
+
+    try:
+        visit(value, label)
+    except (TypeError, RecursionError) as exc:
+        raise TypeError(f"{label} must be json-serializable: {exc}") from exc

runpod/apps/build.py

@@ -50,7 +50,6 @@
     "aiohttp[speedups]",
     "aiohttp-retry",
     "backoff",
-    "cloudpickle",
     "py-cpuinfo",
     "requests",
     "tomli",

runpod/apps/handles.py

@@ -31,13 +31,15 @@ def load_model(): ...                    # worker startup hook
     Type,
 )
 
+from .._serialization import validate_json
 from .context import Context, block, current_context
 from .discovery_state import DiscoveryInvocationError, in_discovery
 from .errors import InvalidResourceError
 from .invoker import Invoker, StreamInvoker
 from .job import Job
 from .markers import INIT_ATTR, is_init, route_of
 from .schedule import SCHEDULE_ATTR
+from .serialization import require_json_args
 from .spec import ResourceKind, ResourceSpec, RouteSpec
 
 if TYPE_CHECKING:
@@ -120,15 +122,34 @@ def local(self, *args: Any, **kwargs: Any) -> Any:
 
     def _call_local_sync(self, *args: Any, **kwargs: Any) -> Any:
         result = self._fn(*args, **kwargs)
-        return list(result) if inspect.isgenerator(result) else result
+        if inspect.isgenerator(result):
+            chunks = []
+            for chunk in result:
+                validate_json(chunk, "return value")
+                chunks.append(chunk)
+            result = chunks
+        if not inspect.isawaitable(result) and not inspect.isasyncgen(result):
+            validate_json(result, "return value")
+        return result
 
     @staticmethod
     async def _await_local_result(result: Any) -> Any:
         if inspect.isawaitable(result):
             result = await result
         if inspect.isasyncgen(result):
-            return [chunk async for chunk in result]
-        return list(result) if inspect.isgenerator(result) else result
+            chunks = []
+            async for chunk in result:
+                validate_json(chunk, "return value")
+                chunks.append(chunk)
+            result = chunks
+        elif inspect.isgenerator(result):
+            chunks = []
+            for chunk in result:
+                validate_json(chunk, "return value")
+                chunks.append(chunk)
+            result = chunks
+        validate_json(result, "return value")
+        return result
 
     def _remote_sync(self, *args: Any, **kwargs: Any) -> Any:
         if (
@@ -139,6 +160,7 @@ def _remote_sync(self, *args: Any, **kwargs: Any) -> Any:
         ):
             self._guard_discovery()
             self.spec.validate()
+            require_json_args(self._fn, args, kwargs)
             result = self._call_local_sync(*args, **kwargs)
             if inspect.isawaitable(result) or inspect.isasyncgen(result):
                 return block(self._await_local_result(result))
@@ -148,6 +170,7 @@ def _remote_sync(self, *args: Any, **kwargs: Any) -> Any:
     async def _remote_async(self, *args: Any, **kwargs: Any) -> Any:
         self._guard_discovery()
         self.spec.validate()
+        require_json_args(self._fn, args, kwargs)
         ctx = current_context()
 
         if ctx is Context.WORKER and self._is_current_worker():
@@ -169,15 +192,18 @@ async def _stream_async(self, *args: Any, **kwargs: Any) -> AsyncIterator[Any]:
         self._guard_discovery()
         self.spec.validate()
         self._guard_generator()
+        require_json_args(self._fn, args, kwargs)
         ctx = current_context()
 
         if ctx is Context.WORKER and self._is_current_worker():
             gen = self._fn(*args, **kwargs)
             if inspect.isasyncgen(gen):
                 async for chunk in gen:
+                    validate_json(chunk, "stream chunk")
                     yield chunk
             else:
                 for chunk in gen:
+                    validate_json(chunk, "stream chunk")
                     yield chunk
             return
 
@@ -208,6 +234,7 @@ def _apply_options(self, payload: Dict[str, Any]) -> Dict[str, Any]:
 
     async def _spawn_async(self, *args: Any, **kwargs: Any) -> Any:
         self._guard_discovery()
+        require_json_args(self._fn, args, kwargs)
         target = await self._app._resolve(self.spec)
         payload = target.build_payload(self._fn, self.spec, args, kwargs)
         data = await target.submit(self._apply_options(payload))

runpod/apps/images.py

@@ -38,11 +38,9 @@ def runtime_tag() -> str:
 def local_python_version() -> str:
     """the local interpreter's version, which the worker must match.
 
-    dev sessions and tasks ship cloudpickle payloads produced on the
-    client interpreter; those are not reliably portable across python
-    versions, so a silent fallback would surface as cryptic
-    deserialization failures on the worker. mismatches fail here,
-    loudly, instead.
+    live execution ships python source from the client. matching the
+    interpreter keeps syntax and language behavior consistent between
+    local development and the worker.
     """
     version = f"{sys.version_info.major}.{sys.version_info.minor}"
     if version not in SUPPORTED_PYTHON_VERSIONS:

runpod/apps/protocol.py

@@ -1,21 +1,20 @@
 """wire protocol for remote function execution.
 
 this is the contract between the sdk and the worker runtime images.
-the worker repo imports these shapes from the runpod package so client
-and worker can never drift.
+the worker runtime and client share these shapes and validation rules.
 
 live mode ships function source with every request; deployed mode omits
 the code (the worker unpacked the build and resolves the function by
-name). args/kwargs are base64 cloudpickle unless serialization_format
-is "json".
+name). arguments, results, and stream chunks are json values.
 """
 
-from dataclasses import asdict, dataclass, field
+from dataclasses import asdict, dataclass, field, fields
 from typing import Any, Dict, List, Optional
 
+from .._serialization import validate_json
+
 EXECUTION_FUNCTION = "function"
 
-FORMAT_CLOUDPICKLE = "cloudpickle"
 FORMAT_JSON = "json"
 
 
@@ -31,10 +30,28 @@ class FunctionRequest:
     system_dependencies: Optional[List[str]] = None
     execution_type: str = EXECUTION_FUNCTION
     accelerate_downloads: bool = True
-    serialization_format: str = FORMAT_CLOUDPICKLE
+    serialization_format: str = FORMAT_JSON
+
+    def validate(self) -> None:
+        if self.serialization_format != FORMAT_JSON:
+            raise ValueError("serialization_format must be 'json'")
+        if not isinstance(self.args, list) or not isinstance(self.kwargs, dict):
+            raise TypeError("args must be a list and kwargs must be a dictionary")
+        validate_json(
+            {f.name: getattr(self, f.name) for f in fields(self)}, "function request"
+        )
+
+    @classmethod
+    def from_input(cls, data: Dict[str, Any]) -> "FunctionRequest":
+        if not isinstance(data, dict):
+            raise TypeError("function request must be a dictionary")
+        request = cls(**{f.name: data[f.name] for f in fields(cls) if f.name in data})
+        request.validate()
+        return request
 
     def to_input(self) -> Dict[str, Any]:
         """serialize as the job input dict."""
+        self.validate()
         data = asdict(self)
         return {k: v for k, v in data.items() if v is not None}
 
@@ -44,16 +61,23 @@ class FunctionResponse:
     """the worker's result for one function execution."""
 
     success: bool
-    result: Optional[str] = None
     json_result: Any = None
     error: Optional[str] = None
     stdout: Optional[str] = None
 
     @classmethod
     def from_output(cls, output: Dict[str, Any]) -> "FunctionResponse":
+        if not isinstance(output, dict) or not isinstance(output.get("success"), bool):
+            raise TypeError("function response must contain a boolean success field")
+        if output.get("serialization_format", FORMAT_JSON) != FORMAT_JSON:
+            raise ValueError("serialization_format must be 'json'")
+        if output.get("result") is not None:
+            raise ValueError("function responses must use json_result")
+        if output["success"] and "json_result" not in output:
+            raise ValueError("successful function response is missing json_result")
+        validate_json(output, "function response")
         return cls(
-            success=output.get("success", False),
-            result=output.get("result"),
+            success=output["success"],
             json_result=output.get("json_result"),
             error=output.get("error"),
             stdout=output.get("stdout"),

runpod/apps/serialization.py

@@ -1,63 +1,32 @@
-"""argument and result serialization for remote execution.
-
-args/kwargs cross the wire as base64-encoded cloudpickle strings, the
-format the worker runtime images expect. function source is extracted
-with decorators stripped so the worker can exec it standalone.
-"""
+"""json argument validation, response decoding, and function source shipping."""
 
 import ast
-import base64
 import inspect
 import os
 import textwrap
-from typing import Any, Callable, Dict, List
-
-import cloudpickle
-
-
-def serialize_arg(arg: Any) -> str:
-    return base64.b64encode(cloudpickle.dumps(arg)).decode("utf-8")
-
-
-def serialize_args(args: tuple) -> List[str]:
-    return [serialize_arg(a) for a in args]
-
+from typing import Any, Callable
 
-def serialize_kwargs(kwargs: dict) -> Dict[str, str]:
-    return {k: serialize_arg(v) for k, v in kwargs.items()}
+from .._serialization import validate_json
+from .errors import RemoteExecutionError
+from .protocol import FunctionResponse
 
 
-def deserialize_result(result_b64: str) -> Any:
+def unwrap_function_response(output: dict) -> Any:
     try:
-        return cloudpickle.loads(base64.b64decode(result_b64))
-    except ModuleNotFoundError as exc:
-        from .errors import RemoteExecutionError
-
+        response = FunctionResponse.from_output(output)
+    except (TypeError, ValueError) as exc:
+        raise RemoteExecutionError(f"invalid remote response: {exc}") from exc
+    if not response.success:
         raise RemoteExecutionError(
-            f"the remote result contains an object from the "
-            f"'{exc.name}' package, which is not installed locally. "
-            f"return plain python types (e.g. str(...)) or install "
-            f"'{exc.name}' on this machine."
-        ) from exc
+            f"remote execution failed: {response.error or 'unknown'}"
+        )
+    return response.json_result
 
 
 def require_json_args(fn: Callable, args: tuple, kwargs: dict) -> None:
-    """reject arguments that cannot cross the queue data plane.
-
-    queue jobs are json on the wire in every mode, so non-json
-    arguments fail at call time with a pointed error instead of a
-    worker-side decode failure.
-    """
-    import json
-
-    try:
-        json.dumps(list(args))
-        json.dumps(kwargs)
-    except (TypeError, ValueError) as exc:
-        raise TypeError(
-            f"{fn.__name__}() arguments must be json-serializable "
-            f"(queue jobs are plain json): {exc}"
-        ) from exc
+    """validate queue and task arguments before remote execution."""
+    validate_json(list(args), f"{fn.__name__}() arguments")
+    validate_json(kwargs, f"{fn.__name__}() keyword arguments")
 
 
 def get_function_source(fn: Callable) -> str: