Merge branch 'main' into gzip-compression

Author: Sushisource

temporalio/client/_client.py

@@ -1196,8 +1196,6 @@ async def _start_update_with_start(
             args=temporalio.common._arg_or_args(arg, args),
             headers={},
             ret_type=result_type or result_type_from_type_hint,
-            rpc_metadata=rpc_metadata,
-            rpc_timeout=rpc_timeout,
             wait_for_stage=wait_for_stage,
         )
 
@@ -1222,6 +1220,8 @@ def on_start_error(
         input = StartWorkflowUpdateWithStartInput(
             start_workflow_input=start_workflow_operation._start_workflow_input,
             update_workflow_input=update_input,
+            rpc_metadata=rpc_metadata,
+            rpc_timeout=rpc_timeout,
             _on_start=on_start,
             _on_start_error=on_start_error,
         )

temporalio/client/_impl.py

@@ -852,7 +852,11 @@ def on_start(
 
         try:
             return await self._start_workflow_update_with_start(
-                input.start_workflow_input, input.update_workflow_input, on_start
+                input.start_workflow_input,
+                input.update_workflow_input,
+                input.rpc_metadata,
+                input.rpc_timeout,
+                on_start,
             )
         except asyncio.CancelledError as _err:
             err = _err
@@ -914,6 +918,8 @@ async def _start_workflow_update_with_start(
         self,
         start_input: UpdateWithStartStartWorkflowInput,
         update_input: UpdateWithStartUpdateWorkflowInput,
+        rpc_metadata: Mapping[str, str | bytes],
+        rpc_timeout: timedelta | None,
         on_start: Callable[
             [temporalio.api.workflowservice.v1.StartWorkflowExecutionResponse], None
         ],
@@ -941,7 +947,12 @@ async def _start_workflow_update_with_start(
         # Repeatedly try to invoke ExecuteMultiOperation until the update is durable
         while True:
             multiop_response = (
-                await self._client.workflow_service.execute_multi_operation(multiop_req)
+                await self._client.workflow_service.execute_multi_operation(
+                    multiop_req,
+                    retry=True,
+                    metadata=rpc_metadata,
+                    timeout=rpc_timeout,
+                )
             )
             start_response = multiop_response.responses[0].start_workflow
             update_response = multiop_response.responses[1].update_workflow

temporalio/client/_interceptor.py

@@ -334,8 +334,6 @@ class UpdateWithStartUpdateWorkflowInput:
     wait_for_stage: WorkflowUpdateStage
     headers: Mapping[str, temporalio.api.common.v1.Payload]
     ret_type: type | None
-    rpc_metadata: Mapping[str, str | bytes]
-    rpc_timeout: timedelta | None
 
 
 @dataclass
@@ -366,18 +364,23 @@ class UpdateWithStartStartWorkflowInput:
     static_details: str | None
     # Type may be absent
     ret_type: type | None
-    rpc_metadata: Mapping[str, str | bytes]
-    rpc_timeout: timedelta | None
     priority: temporalio.common.Priority
     versioning_override: temporalio.common.VersioningOverride | None = None
 
 
 @dataclass
 class StartWorkflowUpdateWithStartInput:
-    """Input for :py:meth:`OutboundInterceptor.start_update_with_start_workflow`."""
+    """Input for :py:meth:`OutboundInterceptor.start_update_with_start_workflow`.
+
+    The ``rpc_metadata`` and ``rpc_timeout`` fields are authoritative for the
+    ``execute_multi_operation`` gRPC call.  Interceptors that wish to set RPC
+    metadata should modify :py:attr:`rpc_metadata` on this object.
+    """
 
     start_workflow_input: UpdateWithStartStartWorkflowInput
     update_workflow_input: UpdateWithStartUpdateWorkflowInput
+    rpc_metadata: Mapping[str, str | bytes]
+    rpc_timeout: timedelta | None
     _on_start: Callable[
         [temporalio.api.workflowservice.v1.StartWorkflowExecutionResponse], None
     ]

temporalio/client/_workflow.py

@@ -1065,8 +1065,6 @@ def __init__(
         static_summary: str | None = None,
         static_details: str | None = None,
         start_delay: timedelta | None = None,
-        rpc_metadata: Mapping[str, str | bytes] = {},
-        rpc_timeout: timedelta | None = None,
         priority: temporalio.common.Priority = temporalio.common.Priority.default,
         versioning_override: temporalio.common.VersioningOverride | None = None,
     ) -> None: ...
@@ -1095,8 +1093,6 @@ def __init__(
         static_summary: str | None = None,
         static_details: str | None = None,
         start_delay: timedelta | None = None,
-        rpc_metadata: Mapping[str, str | bytes] = {},
-        rpc_timeout: timedelta | None = None,
         priority: temporalio.common.Priority = temporalio.common.Priority.default,
         versioning_override: temporalio.common.VersioningOverride | None = None,
     ) -> None: ...
@@ -1127,8 +1123,6 @@ def __init__(
         static_summary: str | None = None,
         static_details: str | None = None,
         start_delay: timedelta | None = None,
-        rpc_metadata: Mapping[str, str | bytes] = {},
-        rpc_timeout: timedelta | None = None,
         priority: temporalio.common.Priority = temporalio.common.Priority.default,
         versioning_override: temporalio.common.VersioningOverride | None = None,
     ) -> None: ...
@@ -1159,8 +1153,6 @@ def __init__(
         static_summary: str | None = None,
         static_details: str | None = None,
         start_delay: timedelta | None = None,
-        rpc_metadata: Mapping[str, str | bytes] = {},
-        rpc_timeout: timedelta | None = None,
         priority: temporalio.common.Priority = temporalio.common.Priority.default,
         versioning_override: temporalio.common.VersioningOverride | None = None,
     ) -> None: ...
@@ -1189,8 +1181,6 @@ def __init__(
         static_summary: str | None = None,
         static_details: str | None = None,
         start_delay: timedelta | None = None,
-        rpc_metadata: Mapping[str, str | bytes] = {},
-        rpc_timeout: timedelta | None = None,
         priority: temporalio.common.Priority = temporalio.common.Priority.default,
         versioning_override: temporalio.common.VersioningOverride | None = None,
         stack_level: int = 2,
@@ -1228,8 +1218,6 @@ def __init__(
             start_delay=start_delay,
             headers={},
             ret_type=result_type or result_type_from_run_fn,
-            rpc_metadata=rpc_metadata,
-            rpc_timeout=rpc_timeout,
             priority=priority,
             versioning_override=versioning_override,
         )

temporalio/contrib/openai_agents/_invoke_model_activity.py

@@ -30,6 +30,7 @@
 from agents.items import TResponseStreamEvent
 from agents.tool import (
     ApplyPatchTool,
+    CustomTool,
     LocalShellTool,
     ShellTool,
     ShellToolEnvironment,
@@ -39,6 +40,7 @@
     APIStatusError,
     AsyncOpenAI,
 )
+from openai.types.responses import CustomToolParam
 from openai.types.responses.tool_param import Mcp
 from typing_extensions import Required, TypedDict
 
@@ -112,6 +114,15 @@ class ApplyPatchToolInput:
     name: str = "apply_patch"
 
 
+@dataclass
+class CustomToolInput:
+    """Data conversion friendly representation of a CustomTool. Contains only the fields which are needed by the model
+    execution to determine what tool to call, not the actual tool invocation, which remains in the workflow context.
+    """
+
+    tool_config: CustomToolParam
+
+
 ToolInput = (
     FunctionToolInput
     | FileSearchTool
@@ -122,6 +133,7 @@ class ApplyPatchToolInput:
     | ShellToolInput
     | LocalShellTool
     | ApplyPatchToolInput
+    | CustomToolInput
     | ToolSearchTool
 )
 
@@ -235,6 +247,14 @@ def _build_tool(tool: ToolInput) -> Tool:
         return ApplyPatchTool(name=tool.name, editor=_NoopApplyPatchEditor())
     elif isinstance(tool, HostedMCPToolInput):
         return HostedMCPTool(tool_config=tool.tool_config)
+    elif isinstance(tool, CustomToolInput):
+        return CustomTool(
+            name=tool.tool_config["name"],
+            description=tool.tool_config.get("description", ""),
+            on_invoke_tool=_empty_on_invoke_tool,
+            format=tool.tool_config.get("format"),
+            defer_loading=tool.tool_config.get("defer_loading", False),
+        )
     elif isinstance(tool, FunctionToolInput):
         return FunctionTool(
             name=tool.name,

temporalio/contrib/openai_agents/_temporal_model_stub.py

@@ -22,14 +22,21 @@
     WebSearchTool,
 )
 from agents.items import TResponseStreamEvent
-from agents.tool import ApplyPatchTool, LocalShellTool, ShellTool, ToolSearchTool
+from agents.tool import (
+    ApplyPatchTool,
+    CustomTool,
+    LocalShellTool,
+    ShellTool,
+    ToolSearchTool,
+)
 from openai.types.responses.response_prompt_param import ResponsePromptParam
 
 from temporalio import workflow
 from temporalio.contrib.openai_agents._invoke_model_activity import (
     ActivityModelInput,
     AgentOutputSchemaInput,
     ApplyPatchToolInput,
+    CustomToolInput,
     FunctionToolInput,
     HandoffInput,
     HostedMCPToolInput,
@@ -92,6 +99,8 @@ def make_tool_info(tool: Tool) -> ToolInput:
                 return ApplyPatchToolInput(name=tool.name)
             elif isinstance(tool, HostedMCPTool):
                 return HostedMCPToolInput(tool_config=tool.tool_config)
+            elif isinstance(tool, CustomTool):
+                return CustomToolInput(tool_config=tool.tool_config)
             elif isinstance(tool, FunctionTool):
                 return FunctionToolInput(
                     name=tool.name,

temporalio/contrib/opentelemetry/_otel_interceptor.py

@@ -200,7 +200,7 @@ def workflow_interceptor_class(
         provider = get_tracer_provider()
         if not isinstance(provider, ReplaySafeTracerProvider):
             raise ValueError(
-                "When using OpenTelemetryPlugin, the global trace provider must be a ReplaySafeTracerProvider. Use init_tracer_provider to create one."
+                "When using OpenTelemetryPlugin, the global trace provider must be a ReplaySafeTracerProvider. Use create_tracer_provider to create one."
             )
 
         class InterceptorWithState(_TracingWorkflowInboundInterceptor):