diff --git a/dapr/ext/workflow/dapr_workflow_context.py b/dapr/ext/workflow/dapr_workflow_context.py
index 7ceee0cbf..3cde70d55 100644
--- a/dapr/ext/workflow/dapr_workflow_context.py
+++ b/dapr/ext/workflow/dapr_workflow_context.py
@@ -15,6 +15,7 @@
 
 from datetime import datetime, timedelta
 from typing import Any, Callable, List, Optional, TypeVar, Union
+from uuid import UUID
 
 from dapr.ext.workflow._durabletask import task
 from dapr.ext.workflow.logger import Logger, LoggerOptions
@@ -57,6 +58,9 @@ def set_custom_status(self, custom_status: str) -> None:
         self._logger.debug(f'{self.instance_id}: Setting custom status to {custom_status}')
         self.__obj.set_custom_status(custom_status)
 
+    def new_guid(self) -> UUID:
+        return self.__obj.new_guid()
+
     def create_timer(self, fire_at: Union[datetime, timedelta]) -> task.Task:
         self._logger.debug(f'{self.instance_id}: Creating timer to fire at {fire_at} time')
         return self.__obj.create_timer(fire_at)
diff --git a/dapr/ext/workflow/workflow_context.py b/dapr/ext/workflow/workflow_context.py
index e3e98fe9e..956bac326 100644
--- a/dapr/ext/workflow/workflow_context.py
+++ b/dapr/ext/workflow/workflow_context.py
@@ -18,6 +18,7 @@
 from abc import ABC, abstractmethod
 from datetime import datetime, timedelta
 from typing import Any, Callable, Generator, Optional, TypeVar, Union
+from uuid import UUID
 
 from dapr.ext.workflow._durabletask import task
 from dapr.ext.workflow.propagation import PropagatedHistory, PropagationScope
@@ -90,6 +91,22 @@ def set_custom_status(self, custom_status: str) -> None:
         """Set the custom status."""
         pass
 
+    @abstractmethod
+    def new_guid(self) -> UUID:
+        """Create a new GUID that is safe for replay within a workflow.
+
+        The GUID is deterministically derived from the workflow instance ID,
+        the current replay-safe time, and a counter that increments on each
+        call, so calling this repeatedly returns different values within a
+        single execution while remaining stable across replays.
+
+        Returns
+        -------
+        uuid.UUID
+            A replay-safe, deterministically generated GUID.
+        """
+        pass
+
     @abstractmethod
     def create_timer(self, fire_at: Union[datetime, timedelta]) -> task.Task:
         """Create a Timer Task to fire after at the specified deadline.
diff --git a/tests/ext/workflow/test_dapr_workflow_context.py b/tests/ext/workflow/test_dapr_workflow_context.py
index 22350dda8..7236a4588 100644
--- a/tests/ext/workflow/test_dapr_workflow_context.py
+++ b/tests/ext/workflow/test_dapr_workflow_context.py
@@ -16,6 +16,7 @@
 import unittest
 from datetime import datetime
 from unittest import mock
+from uuid import UUID
 
 import dapr.ext.workflow._durabletask.internal.protos as pb
 from dapr.ext.workflow import PropagatedHistory
@@ -54,6 +55,9 @@ def set_custom_status(self, custom_status):
     def get_propagated_history(self):
         return self._propagated_history
 
+    def new_guid(self):
+        return UUID('12345678-1234-5678-1234-567812345678')
+
 
 class DaprWorkflowContextTest(unittest.TestCase):
     def mock_client_activity(ctx: WorkflowActivityContext, input):
@@ -83,6 +87,9 @@ def test_workflow_context_functions(self):
             dapr_wf_ctx.set_custom_status(mock_custom_status)
             assert fakeContext.custom_status == mock_custom_status
 
+            new_guid_result = dapr_wf_ctx.new_guid()
+            assert new_guid_result == UUID('12345678-1234-5678-1234-567812345678')
+
     def test_get_propagated_history_proxies_inner_context(self):
         with mock.patch(
             'dapr.ext.workflow._durabletask.worker._RuntimeOrchestrationContext',