diff --git a/examples/workflow/monitor.py b/examples/workflow/monitor.py
index d18ad929..150deae8 100644
--- a/examples/workflow/monitor.py
+++ b/examples/workflow/monitor.py
@@ -37,7 +37,7 @@ def status_monitor_workflow(ctx: wf.DaprWorkflowContext, job: JobStatus):
             ctx.call_activity(send_alert, input=f"Job '{job.job_id}' is unhealthy!")
         next_sleep_interval = 5  # check more frequently when unhealthy
 
-    yield ctx.create_timer(fire_at=ctx.current_utc_datetime + timedelta(seconds=next_sleep_interval))
+    yield ctx.create_timer(fire_at=timedelta(seconds=next_sleep_interval))
 
     # restart from the beginning with a new JobStatus input
     ctx.continue_as_new(job)
diff --git a/ext/dapr-ext-workflow/dapr/ext/workflow/dapr_workflow_context.py b/ext/dapr-ext-workflow/dapr/ext/workflow/dapr_workflow_context.py
index 96cab225..c65a4818 100644
--- a/ext/dapr-ext-workflow/dapr/ext/workflow/dapr_workflow_context.py
+++ b/ext/dapr-ext-workflow/dapr/ext/workflow/dapr_workflow_context.py
@@ -13,8 +13,8 @@
 limitations under the License.
 """
 
-from typing import Any, Callable, List, Optional, TypeVar
-from datetime import datetime
+from typing import Any, Callable, List, Optional, TypeVar, Union
+from datetime import datetime, timedelta
 
 from durabletask import task
 
@@ -48,7 +48,7 @@ def current_utc_datetime(self) -> datetime:
     def is_replaying(self) -> bool:
         return self.__obj.is_replaying
 
-    def create_timer(self, fire_at: datetime) -> task.Task:
+    def create_timer(self, fire_at: Union[datetime, timedelta]) -> task.Task:
         return self.__obj.create_timer(fire_at)
 
     def call_activity(self, activity: Callable[[WorkflowActivityContext, TInput], TOutput], *,
diff --git a/ext/dapr-ext-workflow/dapr/ext/workflow/workflow_context.py b/ext/dapr-ext-workflow/dapr/ext/workflow/workflow_context.py
index 59dfbd11..66ee03a5 100644
--- a/ext/dapr-ext-workflow/dapr/ext/workflow/workflow_context.py
+++ b/ext/dapr-ext-workflow/dapr/ext/workflow/workflow_context.py
@@ -15,7 +15,7 @@
 
 from __future__ import annotations
 from abc import ABC, abstractmethod
-from datetime import datetime
+from datetime import datetime, timedelta
 from typing import Any, Callable, Generator, Optional, TypeVar, Union
 
 from durabletask import task
@@ -85,13 +85,13 @@ def is_replaying(self) -> bool:
         pass
 
     @abstractmethod
-    def create_timer(self, fire_at: datetime) -> task.Task:
+    def create_timer(self, fire_at: Union[datetime, timedelta]) -> task.Task:
         """Create a Timer Task to fire after at the specified deadline.
 
         Parameters
         ----------
-        fire_at: datetime.datetime
-            The time for the timer to trigger
+        fire_at: datetime.datetime | datetime.timedelta
+            The time for the timer to trigger. Can be specified as a `datetime` or a `timedelta`.
 
         Returns
         -------