diff --git a/dapr/ext/workflow/_durabletask/task.py b/dapr/ext/workflow/_durabletask/task.py
index e3a190c41..6a3dc7451 100644
--- a/dapr/ext/workflow/_durabletask/task.py
+++ b/dapr/ext/workflow/_durabletask/task.py
@@ -13,6 +13,7 @@
 from __future__ import annotations
 
 import math
+import uuid
 from abc import ABC, abstractmethod
 from datetime import datetime, timedelta
 from typing import Any, Callable, Generator, Generic, Optional, TypeVar, Union
@@ -89,6 +90,17 @@ def set_custom_status(self, custom_status: str) -> None:
         """
         pass
 
+    @abstractmethod
+    def new_guid(self) -> uuid.UUID:
+        """Create a new GUID that is safe for replay within an orchestration.
+
+        Returns
+        -------
+        uuid.UUID
+            A new, deterministically generated GUID.
+        """
+        pass
+
     @abstractmethod
     def create_timer(self, fire_at: Union[datetime, timedelta]) -> Task:
         """Create a Timer Task to fire after at the specified deadline.
diff --git a/dapr/ext/workflow/_durabletask/worker.py b/dapr/ext/workflow/_durabletask/worker.py
index 090eb747a..e0c0946e1 100644
--- a/dapr/ext/workflow/_durabletask/worker.py
+++ b/dapr/ext/workflow/_durabletask/worker.py
@@ -17,6 +17,7 @@
 import random
 import threading
 import time
+import uuid
 import warnings
 from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timedelta, timezone
@@ -1609,6 +1610,9 @@ def current_utc_datetime(self, value: datetime):
     def is_replaying(self) -> bool:
         return self._is_replaying
 
+    def new_guid(self) -> uuid.UUID:
+        return self.uuid4()
+
     def set_custom_status(self, custom_status: str) -> None:
         if custom_status is not None and not isinstance(custom_status, str):
             warnings.warn(
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',