Use declared types for transfer serialization

CHANGELOG.md

@@ -29,6 +29,10 @@ to include examples, links to docs, or any other relevant information.
 
 ### Fixed
 
+- Use declared argument and result types to select transfer type converters during
+  serialization, and skip transfer conversion when no hint is available. Preserve
+  existing data and payload converter method signatures.
+
 ### Security
 
 ## [1.33.0] - 2026-09-14

temporalio/client/_client.py

@@ -625,6 +625,11 @@ async def start_workflow(
         return await self._impl.start_workflow(
             StartWorkflowInput(
                 workflow=name,
+                arg_types=(
+                    temporalio.workflow._Definition.must_from_run_fn(workflow).arg_types
+                    if callable(workflow)
+                    else None
+                ),
                 args=temporalio.common._arg_or_args(arg, args),
                 id=id,
                 task_queue=task_queue,
@@ -1187,6 +1192,11 @@ async def _start_update_with_start(
         update_input = UpdateWithStartUpdateWorkflowInput(
             update_id=id,
             update=update_name,
+            arg_types=(
+                update._defn.arg_types
+                if isinstance(update, temporalio.workflow.UpdateMethodMultiParam)
+                else None
+            ),
             args=temporalio.common._arg_or_args(arg, args),
             headers={},
             ret_type=result_type or result_type_from_type_hint,
@@ -1521,6 +1531,13 @@ async def start_activity(
         return await self._impl.start_activity(
             StartActivityInput(
                 activity_type=name,
+                arg_types=(
+                    temporalio.activity._Definition.must_from_callable(
+                        activity
+                    ).arg_types
+                    if callable(activity)
+                    else None
+                ),
                 args=temporalio.common._arg_or_args(arg, args),
                 id=id,
                 task_queue=task_queue,

temporalio/client/_impl.py

@@ -284,7 +284,9 @@ async def _populate_start_workflow_execution_request(
         req.workflow_type.name = input.workflow
         req.task_queue.name = input.task_queue
         if input.args:
-            req.input.payloads.extend(await data_converter.encode(input.args))
+            req.input.payloads.extend(
+                await data_converter.encode_with_type_hints(input.args, input.arg_types)
+            )
         if input.execution_timeout is not None:
             req.workflow_execution_timeout.FromTimedelta(input.execution_timeout)
         if input.run_timeout is not None:
@@ -422,7 +424,7 @@ async def query_workflow(self, input: QueryWorkflowInput) -> Any:
         req.query.query_type = input.query
         if input.args:
             req.query.query_args.payloads.extend(
-                await data_converter.encode(input.args)
+                await data_converter.encode_with_type_hints(input.args, input.arg_types)
             )
         if input.headers is not None:  # type:ignore[reportUnnecessaryComparison]
             await self._apply_headers(input.headers, req.query.header.fields)
@@ -484,7 +486,9 @@ async def signal_workflow(self, input: SignalWorkflowInput) -> None:
             request_id=str(uuid.uuid4()),
         )
         if input.args:
-            req.input.payloads.extend(await data_converter.encode(input.args))
+            req.input.payloads.extend(
+                await data_converter.encode_with_type_hints(input.args, input.arg_types)
+            )
         if input.headers is not None:  # type:ignore[reportUnnecessaryComparison]
             await self._apply_headers(input.headers, req.header.fields)
         temporalio.nexus._operation_context._apply_nexus_context_to_signal_workflow_request(
@@ -626,7 +630,9 @@ async def _build_start_activity_execution_request(
 
         # Set input payloads
         if input.args:
-            req.input.payloads.extend(await data_converter.encode(input.args))
+            req.input.payloads.extend(
+                await data_converter.encode_with_type_hints(input.args, input.arg_types)
+            )
 
         # Set search attributes
         if input.search_attributes is not None:
@@ -939,7 +945,7 @@ async def _build_update_workflow_execution_request(
             )
         if input.args:
             req.request.input.args.payloads.extend(
-                await data_converter.encode(input.args)
+                await data_converter.encode_with_type_hints(input.args, input.arg_types)
             )
         if input.headers is not None:  # type:ignore[reportUnnecessaryComparison]
             await self._apply_headers(input.headers, req.request.input.header.fields)
@@ -1581,7 +1587,9 @@ async def start_nexus_operation(
             req.start_to_close_timeout.FromTimedelta(input.start_to_close_timeout)
 
         # Set input payload
-        encoded = await data_converter.encode([input.arg])
+        encoded = await data_converter.encode_with_type_hints(
+            [input.arg], [input.input_type]
+        )
         if encoded:
             req.input.CopyFrom(encoded[0])
 

temporalio/client/_interceptor.py

@@ -96,6 +96,7 @@ class StartWorkflowInput:
     request_eager_start: bool
     priority: temporalio.common.Priority
     versioning_override: temporalio.common.VersioningOverride | None = None
+    arg_types: list[type] | None = None
 
 
 @dataclass
@@ -170,6 +171,7 @@ class QueryWorkflowInput:
     ret_type: type | None
     rpc_metadata: Mapping[str, str | bytes]
     rpc_timeout: timedelta | None
+    arg_types: list[type] | None = None
 
 
 @dataclass
@@ -183,6 +185,7 @@ class SignalWorkflowInput:
     headers: Mapping[str, temporalio.api.common.v1.Payload]
     rpc_metadata: Mapping[str, str | bytes]
     rpc_timeout: timedelta | None
+    arg_types: list[type] | None = None
 
 
 @dataclass
@@ -221,6 +224,7 @@ class StartActivityInput:
     headers: Mapping[str, temporalio.api.common.v1.Payload]
     rpc_metadata: Mapping[str, str | bytes]
     rpc_timeout: timedelta | None
+    arg_types: list[type] | None = None
 
 
 @dataclass
@@ -342,6 +346,7 @@ class StartWorkflowUpdateInput:
     ret_type: type | None
     rpc_metadata: Mapping[str, str | bytes]
     rpc_timeout: timedelta | None
+    arg_types: list[type] | None = None
 
 
 @dataclass
@@ -354,6 +359,7 @@ class UpdateWithStartUpdateWorkflowInput:
     wait_for_stage: WorkflowUpdateStage
     headers: Mapping[str, temporalio.api.common.v1.Payload]
     ret_type: type | None
+    arg_types: list[type] | None = None
 
 
 @dataclass
@@ -386,6 +392,7 @@ class UpdateWithStartStartWorkflowInput:
     ret_type: type | None
     priority: temporalio.common.Priority
     versioning_override: temporalio.common.VersioningOverride | None = None
+    arg_types: list[type] | None = None
 
 
 @dataclass
@@ -605,6 +612,7 @@ class StartNexusOperationInput:
     headers: Mapping[str, str]
     rpc_metadata: Mapping[str, str | bytes]
     rpc_timeout: timedelta | None
+    input_type: type | None = None
 
 
 @dataclass

temporalio/client/_nexus.py

@@ -940,16 +940,16 @@ def __init__(
     def _resolve_operation(
         self,
         operation: nexusrpc.Operation[Any, Any] | str | Callable[..., Any],
-    ) -> tuple[str, type | None]:
-        """Resolve an operation to its name and output type."""
+    ) -> tuple[str, type | None, type | None]:
+        """Resolve an operation to its name, input type, and output type."""
         if isinstance(operation, str):
-            return operation, None
+            return operation, None, None
         elif isinstance(operation, nexusrpc.Operation):
-            return operation.name, operation.output_type
+            return operation.name, operation.input_type, operation.output_type
         elif callable(operation):
             _, op = temporalio.nexus._util.get_operation_factory(operation)
             if isinstance(op, nexusrpc.Operation):
-                return op.name, op.output_type
+                return op.name, op.input_type, op.output_type
             else:
                 raise ValueError(
                     f"Operation callable is not a Nexus operation: {operation}"
@@ -982,14 +982,15 @@ async def start_operation(
         .. warning::
            This API is experimental and unstable.
         """
-        op_name, output_type = self._resolve_operation(operation)
+        op_name, input_type, output_type = self._resolve_operation(operation)
         final_result_type: type | None = (
             result_type if isinstance(operation, str) else output_type
         )
 
         return await self._client._impl.start_nexus_operation(
             StartNexusOperationInput(
                 operation=op_name,
+                input_type=input_type,
                 arg=arg,
                 id=id,
                 endpoint=self._endpoint,

temporalio/client/_schedule.py

@@ -557,6 +557,7 @@ class ScheduleActionStartWorkflow(ScheduleAction):
     Headers may still be encoded by the payload codec if present.
     """
     _from_raw: bool = dataclasses.field(compare=False, init=False)
+    _arg_types: list[type] | None = dataclasses.field(compare=False, init=False)
 
     @staticmethod
     def _from_proto(  # pyright: ignore
@@ -682,6 +683,7 @@ def __init__(
         values.
         """
         super().__init__()
+        self._arg_types = None
         if raw_info:
             self._from_raw = True
             # Ignore other fields
@@ -753,6 +755,7 @@ def __init__(
                 defn = temporalio.workflow._Definition.must_from_run_fn(workflow)
                 if not defn.name:
                     raise ValueError("Cannot schedule dynamic workflow explicitly")
+                self._arg_types = defn.arg_types
                 workflow = defn.name
             elif not isinstance(workflow, str):
                 raise TypeError("Workflow must be a string or callable")  # type:ignore[reportUnreachable]
@@ -815,8 +818,15 @@ async def _to_proto(
                         payloads=[
                             a
                             if isinstance(a, temporalio.api.common.v1.Payload)
-                            else (await data_converter.encode([a]))[0]
-                            for a in self.args
+                            else (
+                                await data_converter.encode_with_type_hints(
+                                    [a],
+                                    [self._arg_types[index]]
+                                    if self._arg_types and index < len(self._arg_types)
+                                    else None,
+                                )
+                            )[0]
+                            for index, a in enumerate(self.args)
                         ]
                     )
                     if self.args

temporalio/client/_workflow.py

@@ -580,6 +580,7 @@ async def query(
         """
         query_name: str
         ret_type = result_type
+        arg_types: list[type] | None = None
         if callable(query):
             defn = temporalio.workflow._QueryDefinition.from_fn(query)
             if not defn:
@@ -592,6 +593,7 @@ async def query(
             # TODO(cretz): Check count/type of args at runtime?
             query_name = defn.name
             ret_type = defn.ret_type
+            arg_types = defn.arg_types
         else:
             query_name = str(query)
 
@@ -600,6 +602,7 @@ async def query(
                 id=self._id,
                 run_id=self._run_id,
                 query=query_name,
+                arg_types=arg_types,
                 args=temporalio.common._arg_or_args(arg, args),
                 reject_condition=reject_condition
                 or self._client._config["default_workflow_query_reject_condition"],
@@ -692,6 +695,12 @@ async def signal(
                 signal=temporalio.workflow._SignalDefinition.must_name_from_fn_or_str(
                     signal
                 ),
+                arg_types=(
+                    defn.arg_types
+                    if callable(signal)
+                    and (defn := temporalio.workflow._SignalDefinition.from_fn(signal))
+                    else None
+                ),
                 args=temporalio.common._arg_or_args(arg, args),
                 headers={},
                 rpc_metadata=rpc_metadata,
@@ -970,6 +979,11 @@ async def _start_update(
                 first_execution_run_id=self._first_execution_run_id,
                 update_id=id,
                 update=update_name,
+                arg_types=(
+                    update._defn.arg_types
+                    if isinstance(update, temporalio.workflow.UpdateMethodMultiParam)
+                    else None
+                ),
                 args=temporalio.common._arg_or_args(arg, args),
                 headers={},
                 ret_type=result_type or result_type_from_type_hint,
@@ -1201,6 +1215,11 @@ def __init__(
 
         self._start_workflow_input = UpdateWithStartStartWorkflowInput(
             workflow=name,
+            arg_types=(
+                temporalio.workflow._Definition.must_from_run_fn(workflow).arg_types
+                if callable(workflow)
+                else None
+            ),
             args=temporalio.common._arg_or_args(arg, args),
             id=id,
             task_queue=task_queue,

temporalio/converter/_data_converter.py

@@ -29,6 +29,7 @@
 from temporalio.converter._payload_converter import (
     PayloadConverter,
     _TemporalTransferTypePayloadConverter,
+    _with_serialization_type_hints,
 )
 from temporalio.converter._serialization_context import (
     SerializationContext,
@@ -98,6 +99,22 @@ def _new_payload_converter(self) -> PayloadConverter:
             self.payload_converter_class()
         )
 
+    async def encode_with_type_hints(
+        self,
+        values: Sequence[Any],
+        type_hints: Sequence[type | None] | None = None,
+    ) -> list[temporalio.api.common.v1.Payload]:
+        """Encode values using declared types for transfer converter selection.
+
+        Hints correspond to values by position. A missing or None hint disables
+        transfer conversion for that value; hints for omitted arguments are ignored.
+        Existing :py:meth:`encode` overrides are invoked unchanged; overrides
+        should forward the original sequence to preserve hints when delegating
+        to payload conversion.
+        """
+        with _with_serialization_type_hints(values, type_hints):
+            return await self.encode(values)
+
     async def encode(
         self, values: Sequence[Any]
     ) -> list[temporalio.api.common.v1.Payload]: