diff --git a/Test/DurableTask.Core.Tests/ScheduleTaskOptionsTests.cs b/Test/DurableTask.Core.Tests/ScheduleTaskOptionsTests.cs
new file mode 100644
index 000000000..c61b99033
--- /dev/null
+++ b/Test/DurableTask.Core.Tests/ScheduleTaskOptionsTests.cs
@@ -0,0 +1,201 @@
+// ---------------------------------------------------------------
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// ---------------------------------------------------------------
+
+namespace DurableTask.Core.Tests
+{
+    using Microsoft.VisualStudio.TestTools.UnitTesting;
+    using System;
+    using System.Collections.Generic;
+
+    [TestClass]
+    public class ScheduleTaskOptionsTests
+    {
+        [TestMethod]
+        public void CreateBuilder_ShouldReturnBuilderInstance()
+        {
+            // Act
+            ScheduleTaskOptions.Builder builder = ScheduleTaskOptions.CreateBuilder();
+
+            // Assert
+            Assert.IsNotNull(builder);
+            Assert.IsInstanceOfType(builder, typeof(ScheduleTaskOptions.Builder));
+        }
+
+        [TestMethod]
+        public void Build_ShouldCreateInstanceWithNullProperties()
+        {
+            // Act
+            ScheduleTaskOptions options = ScheduleTaskOptions.CreateBuilder().Build();
+
+            // Assert
+            Assert.IsNotNull(options);
+            Assert.IsNull(options.Tags);
+            Assert.IsNull(options.RetryOptions);
+        }
+
+        [TestMethod]
+        public void WithTags_ShouldSetTagsProperty()
+        {
+            // Arrange
+            Dictionary<string, string> tags = new Dictionary<string, string>
+            {
+                { "key1", "value1" },
+                { "key2", "value2" }
+            };
+
+            // Act
+            ScheduleTaskOptions options = ScheduleTaskOptions.CreateBuilder()
+                .WithTags(tags)
+                .Build();
+
+            // Assert
+            Assert.IsNotNull(options.Tags);
+            Assert.AreEqual(2, options.Tags.Count);
+            Assert.AreEqual("value1", options.Tags["key1"]);
+            Assert.AreEqual("value2", options.Tags["key2"]);
+        }
+
+        [TestMethod]
+        public void AddTag_WithNullTags_ShouldInitializeTagsCollection()
+        {
+            // Act
+            ScheduleTaskOptions options = ScheduleTaskOptions.CreateBuilder()
+                .AddTag("key1", "value1")
+                .Build();
+
+            // Assert
+            Assert.IsNotNull(options.Tags);
+            Assert.AreEqual(1, options.Tags.Count);
+            Assert.AreEqual("value1", options.Tags["key1"]);
+        }
+
+        [TestMethod]
+        public void AddTag_WithExistingTags_ShouldAddToCollection()
+        {
+            // Arrange
+            Dictionary<string, string> tags = new Dictionary<string, string>
+            {
+                { "key1", "value1" }
+            };
+
+            // Act
+            ScheduleTaskOptions options = ScheduleTaskOptions.CreateBuilder()
+                .WithTags(tags)
+                .AddTag("key2", "value2")
+                .Build();
+
+            // Assert
+            Assert.IsNotNull(options.Tags);
+            Assert.AreEqual(2, options.Tags.Count);
+            Assert.AreEqual("value1", options.Tags["key1"]);
+            Assert.AreEqual("value2", options.Tags["key2"]);
+        }
+
+        [TestMethod]
+        public void AddTag_OverwriteExistingKey_ShouldUpdateValue()
+        {
+            // Act
+            ScheduleTaskOptions options = ScheduleTaskOptions.CreateBuilder()
+                .AddTag("key1", "originalValue")
+                .AddTag("key1", "newValue")
+                .Build();
+
+            // Assert
+            Assert.IsNotNull(options.Tags);
+            Assert.AreEqual(1, options.Tags.Count);
+            Assert.AreEqual("newValue", options.Tags["key1"]);
+        }
+
+        [TestMethod]
+        public void WithRetryOptions_Instance_ShouldSetRetryOptionsProperty()
+        {
+            // Arrange
+            RetryOptions retryOptions = new RetryOptions(TimeSpan.FromSeconds(5), 3);
+
+            // Act
+            ScheduleTaskOptions options = ScheduleTaskOptions.CreateBuilder()
+                .WithRetryOptions(retryOptions)
+                .Build();
+
+            // Assert
+            Assert.IsNotNull(options.RetryOptions);
+            Assert.AreEqual(TimeSpan.FromSeconds(5), options.RetryOptions.FirstRetryInterval);
+            Assert.AreEqual(3, options.RetryOptions.MaxNumberOfAttempts);
+        }
+
+        [TestMethod]
+        public void WithRetryOptions_Parameters_ShouldCreateAndSetRetryOptions()
+        {
+            // Act
+            ScheduleTaskOptions options = ScheduleTaskOptions.CreateBuilder()
+                .WithRetryOptions(TimeSpan.FromSeconds(5), 3)
+                .Build();
+
+            // Assert
+            Assert.IsNotNull(options.RetryOptions);
+            Assert.AreEqual(TimeSpan.FromSeconds(5), options.RetryOptions.FirstRetryInterval);
+            Assert.AreEqual(3, options.RetryOptions.MaxNumberOfAttempts);
+            Assert.AreEqual(1, options.RetryOptions.BackoffCoefficient);
+            Assert.AreEqual(TimeSpan.MaxValue, options.RetryOptions.MaxRetryInterval);
+        }
+
+        [TestMethod]
+        public void WithRetryOptions_WithConfigureAction_ShouldConfigureRetryOptions()
+        {
+            // Act
+            ScheduleTaskOptions options = ScheduleTaskOptions.CreateBuilder()
+                .WithRetryOptions(TimeSpan.FromSeconds(5), 3, retryOptions =>
+                {
+                    retryOptions.BackoffCoefficient = 2.0;
+                    retryOptions.MaxRetryInterval = TimeSpan.FromMinutes(1);
+                })
+                .Build();
+
+            // Assert
+            Assert.IsNotNull(options.RetryOptions);
+            Assert.AreEqual(TimeSpan.FromSeconds(5), options.RetryOptions.FirstRetryInterval);
+            Assert.AreEqual(3, options.RetryOptions.MaxNumberOfAttempts);
+            Assert.AreEqual(2.0, options.RetryOptions.BackoffCoefficient);
+            Assert.AreEqual(TimeSpan.FromMinutes(1), options.RetryOptions.MaxRetryInterval);
+        }
+
+        [TestMethod]
+        public void WithRetryOptions_PassNullConfigureAction_ShouldStillCreateRetryOptions()
+        {
+            // Act
+            ScheduleTaskOptions options = ScheduleTaskOptions.CreateBuilder()
+                .WithRetryOptions(TimeSpan.FromSeconds(5), 3, null)
+                .Build();
+
+            // Assert
+            Assert.IsNotNull(options.RetryOptions);
+            Assert.AreEqual(TimeSpan.FromSeconds(5), options.RetryOptions.FirstRetryInterval);
+            Assert.AreEqual(3, options.RetryOptions.MaxNumberOfAttempts);
+        }
+
+        [TestMethod]
+        public void FluentInterface_CombiningAllMethods_ShouldBuildCorrectInstance()
+        {
+            // Arrange
+            RetryOptions retryOptions = new RetryOptions(TimeSpan.FromSeconds(1), 2);
+
+            // Act
+            ScheduleTaskOptions options = ScheduleTaskOptions.CreateBuilder()
+                .AddTag("env", "test")
+                .WithRetryOptions(retryOptions)
+                .AddTag("priority", "high")
+                .Build();
+
+            // Assert
+            Assert.IsNotNull(options);
+            Assert.IsNotNull(options.Tags);
+            Assert.AreEqual(2, options.Tags.Count);
+            Assert.AreEqual("test", options.Tags["env"]);
+            Assert.AreEqual("high", options.Tags["priority"]);
+            Assert.IsNotNull(options.RetryOptions);
+            Assert.AreEqual(retryOptions.FirstRetryInterval, options.RetryOptions.FirstRetryInterval);
+            Assert.AreEqual(retryOptions.MaxNumberOfAttempts, options.RetryOptions.MaxNumberOfAttempts);
+        }
+    }
+}
\ No newline at end of file
diff --git a/Test/DurableTask.Core.Tests/TaskOrchestrationContextTests.cs b/Test/DurableTask.Core.Tests/TaskOrchestrationContextTests.cs
new file mode 100644
index 000000000..a19d76a20
--- /dev/null
+++ b/Test/DurableTask.Core.Tests/TaskOrchestrationContextTests.cs
@@ -0,0 +1,228 @@
+// ---------------------------------------------------------------
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// ---------------------------------------------------------------
+
+namespace DurableTask.Core.Tests
+{
+    using DurableTask.Core.Exceptions;
+    using DurableTask.Core.History;
+    using Microsoft.VisualStudio.TestTools.UnitTesting;
+    using System;
+    using System.Collections.Generic;
+    using System.Threading;
+    using System.Threading.Tasks;
+
+    [TestClass]
+    public class TaskOrchestrationContextTests
+    {
+        private MockTaskOrchestrationContext context;
+        private OrchestrationInstance instance;
+
+        [TestInitialize]
+        public void Initialize()
+        {
+            instance = new OrchestrationInstance { InstanceId = "TestInstance", ExecutionId = Guid.NewGuid().ToString() };
+            context = new MockTaskOrchestrationContext(instance, TaskScheduler.Default);
+        }
+
+        [TestMethod]
+        public async Task ScheduleTask_Basic_ShouldScheduleTask()
+        {
+            // Act
+            Task<int> resultTask = context.ScheduleTask<int>("TestActivity", "1.0", 10, 20);
+            Assert.IsFalse(resultTask.IsCompleted);
+
+            // Verify task was scheduled
+            Assert.AreEqual(1, context.ScheduledTasks.Count);
+            ScheduledTaskInfo scheduledTask = context.ScheduledTasks[0];
+            Assert.AreEqual("TestActivity", scheduledTask.Name);
+            Assert.AreEqual("1.0", scheduledTask.Version);
+            CollectionAssert.AreEqual(new object[] { 10, 20 }, scheduledTask.Parameters);
+
+            // Complete the task and verify result
+            context.CompleteTask<int>(0, 30);
+            int result = await resultTask;
+            Assert.AreEqual(30, result);
+        }
+
+        [TestMethod]
+        public async Task ScheduleTask_WithNullOptions_ShouldScheduleTask()
+        {
+            // Act
+            ScheduleTaskOptions options = ScheduleTaskOptions.CreateBuilder().Build();
+            Task<int> resultTask = context.ScheduleTask<int>("TestActivity", "1.0", options, 10, 20);
+            Assert.IsFalse(resultTask.IsCompleted);
+
+            // Verify task was scheduled
+            Assert.AreEqual(1, context.ScheduledTasks.Count);
+            ScheduledTaskInfo scheduledTask = context.ScheduledTasks[0];
+            Assert.AreEqual("TestActivity", scheduledTask.Name);
+            Assert.AreEqual("1.0", scheduledTask.Version);
+            CollectionAssert.AreEqual(new object[] { 10, 20 }, scheduledTask.Parameters);
+            Assert.IsNull(scheduledTask.Options.Tags);
+            Assert.IsNull(scheduledTask.Options.RetryOptions);
+
+            // Complete the task and verify result
+            context.CompleteTask<int>(0, 30);
+            int result = await resultTask;
+            Assert.AreEqual(30, result);
+        }
+
+        [TestMethod]
+        public async Task ScheduleTask_WithTags_ShouldPassTags()
+        {
+            // Arrange
+            ScheduleTaskOptions options = ScheduleTaskOptions.CreateBuilder()
+                .AddTag("key1", "value1")
+                .AddTag("key2", "value2")
+                .Build();
+
+            // Act
+            Task<int> resultTask = context.ScheduleTask<int>("TestActivity", "1.0", options, 10, 20);
+
+            // Verify task was scheduled with tags
+            Assert.AreEqual(1, context.ScheduledTasks.Count);
+            ScheduledTaskInfo scheduledTask = context.ScheduledTasks[0];
+
+            Assert.IsNotNull(scheduledTask.Options.Tags);
+            Assert.AreEqual(2, scheduledTask.Options.Tags.Count);
+            Assert.AreEqual("value1", scheduledTask.Options.Tags["key1"]);
+            Assert.AreEqual("value2", scheduledTask.Options.Tags["key2"]);
+
+            // Complete the task
+            context.CompleteTask<int>(0, 30);
+            await resultTask;
+        }
+
+        [TestMethod]
+        public async Task ScheduleTask_WithDefaultReturnValue_ShouldHandleNullResult()
+        {
+            // Act
+            Task<int> resultTask = context.ScheduleTask<int>("TestActivity", "1.0", 10, 20);
+
+            // Complete the task with null result
+            context.CompleteTaskWithNullResult(0);
+
+            // Verify default value is returned
+            int result = await resultTask;
+            Assert.AreEqual(0, result); // default for int is 0
+        }
+
+        [TestMethod]
+        public async Task ScheduleTask_WithReferenceType_ShouldHandleNullResult()
+        {
+            // Act
+            Task<string> resultTask = context.ScheduleTask<string>("TestActivity", "1.0", 10, 20);
+
+            // Complete the task with null result
+            context.CompleteTaskWithNullResult(0);
+
+            // Verify null is returned
+            string result = await resultTask;
+            Assert.IsNull(result); // default for reference type is null
+        }
+
+        [TestMethod]
+        public async Task ScheduleTask_WithFailure_ShouldPropagateException()
+        {
+            // Act
+            Task<int> resultTask = context.ScheduleTask<int>("TestActivity", "1.0", 10, 20);
+
+            // Fail the task
+            InvalidOperationException expectedException = new InvalidOperationException("Expected failure");
+            context.FailTask(0, expectedException);
+
+            // Verify exception is propagated
+            try
+            {
+                await resultTask;
+                Assert.Fail("Task should have failed");
+            }
+            catch (TaskFailedException ex)
+            {
+                Assert.AreEqual("TestActivity", ex.Name);
+                Assert.AreEqual("1.0", ex.Version);
+                Assert.IsInstanceOfType(ex.InnerException, typeof(InvalidOperationException));
+                Assert.AreEqual("Expected failure", ex.InnerException.Message);
+            }
+        }
+
+        private class MockTaskOrchestrationContext : TaskOrchestrationContext
+        {
+            public List<ScheduledTaskInfo> ScheduledTasks { get; } = new List<ScheduledTaskInfo>();
+            public List<TimeSpan> Delays { get; } = new List<TimeSpan>();
+
+            public MockTaskOrchestrationContext(OrchestrationInstance orchestrationInstance, TaskScheduler taskScheduler)
+                : base(orchestrationInstance, taskScheduler)
+            {
+                CurrentUtcDateTime = DateTime.UtcNow;
+            }
+
+            public void CompleteTask<T>(int taskIndex, T result)
+            {
+                string serializedResult = MessageDataConverter.SerializeInternal(result);
+                TaskCompletedEvent taskCompletedEvent = new TaskCompletedEvent(0, taskIndex, serializedResult);
+                HandleTaskCompletedEvent(taskCompletedEvent);
+            }
+
+            public void CompleteTaskWithNullResult(int taskIndex)
+            {
+                TaskCompletedEvent taskCompletedEvent = new TaskCompletedEvent(0, taskIndex, null);
+                HandleTaskCompletedEvent(taskCompletedEvent);
+            }
+
+            public void FailTask(int taskIndex, Exception exception)
+            {
+                string details = ErrorDataConverter.SerializeInternal(exception);
+                TaskFailedEvent taskFailedEvent = new TaskFailedEvent(0, taskIndex, exception.Message, details);
+                HandleTaskFailedEvent(taskFailedEvent);
+            }
+
+            public override async Task<T> CreateTimer<T>(DateTime fireAt, T state, CancellationToken cancelToken)
+            {
+                TimeSpan delay = fireAt - CurrentUtcDateTime;
+                Delays.Add(delay);
+
+                CurrentUtcDateTime = fireAt; // Advance the time
+
+                return await Task.FromResult(state);
+            }
+
+            public override async Task<TResult> ScheduleTask<TResult>(string name, string version, params object[] parameters)
+            {
+                ScheduledTasks.Add(new ScheduledTaskInfo
+                {
+                    Name = name,
+                    Version = version,
+                    Parameters = parameters,
+                    Options = null
+                });
+
+                // This just sets up the infrastructure needed for completing the task
+                return await base.ScheduleTask<TResult>(name, version, parameters);
+            }
+
+            public override async Task<TResult> ScheduleTask<TResult>(string name, string version, ScheduleTaskOptions options, params object[] parameters)
+            {
+                ScheduledTasks.Add(new ScheduledTaskInfo
+                {
+                    Name = name,
+                    Version = version,
+                    Parameters = parameters,
+                    Options = options
+                });
+
+                // This will go through TaskOrchestrationContext's implementation, which handles retries
+                return await base.ScheduleTask<TResult>(name, version, options, parameters);
+            }
+        }
+
+        private class ScheduledTaskInfo
+        {
+            public string Name { get; set; }
+            public string Version { get; set; }
+            public object[] Parameters { get; set; }
+            public ScheduleTaskOptions Options { get; set; }
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/DurableTask.Core/Command/ScheduleTaskOrchestratorAction.cs b/src/DurableTask.Core/Command/ScheduleTaskOrchestratorAction.cs
index f4c65b84c..4e3c0bf85 100644
--- a/src/DurableTask.Core/Command/ScheduleTaskOrchestratorAction.cs
+++ b/src/DurableTask.Core/Command/ScheduleTaskOrchestratorAction.cs
@@ -11,6 +11,8 @@
 //  limitations under the License.
 //  ----------------------------------------------------------------------------------
 #nullable enable
+using System.Collections.Generic;
+
 namespace DurableTask.Core.Command
 {
     /// <summary>
@@ -41,5 +43,10 @@ public class ScheduleTaskOrchestratorAction : OrchestratorAction
 
         // TODO: This property is not used and should be removed or made obsolete
         internal string? Tasklist { get; set; }
+
+        /// <summary>
+        /// Gets or sets a dictionary of tags of string, string
+        /// </summary>
+        public IDictionary<string, string>? Tags { get; set; }
     }
 }
\ No newline at end of file
diff --git a/src/DurableTask.Core/History/TaskScheduledEvent.cs b/src/DurableTask.Core/History/TaskScheduledEvent.cs
index c0f9108d3..73f84ce15 100644
--- a/src/DurableTask.Core/History/TaskScheduledEvent.cs
+++ b/src/DurableTask.Core/History/TaskScheduledEvent.cs
@@ -13,6 +13,7 @@
 #nullable enable
 namespace DurableTask.Core.History
 {
+    using System.Collections.Generic;
     using System.Runtime.Serialization;
     using DurableTask.Core.Tracing;
 
@@ -84,5 +85,11 @@ public TaskScheduledEvent(int eventId)
         /// </summary>
         [DataMember]
         public DistributedTraceContext? ParentTraceContext { get; set; }
+        
+        /// <summary>
+        /// Gets or sets a dictionary of tags of string, string
+        /// </summary>
+        [DataMember]
+        public IDictionary<string, string>? Tags { get; set; }
     }
 }
\ No newline at end of file
diff --git a/src/DurableTask.Core/OrchestrationContext.cs b/src/DurableTask.Core/OrchestrationContext.cs
index 4b2d8c7fe..97c6f6f6d 100644
--- a/src/DurableTask.Core/OrchestrationContext.cs
+++ b/src/DurableTask.Core/OrchestrationContext.cs
@@ -286,6 +286,19 @@ public virtual Task<TResult> ScheduleTask<TResult>(Type activityType, params obj
                 NameVersionHelper.GetDefaultVersion(activityType), parameters);
         }
 
+        /// <summary>
+        ///     Schedule a TaskActivity by type, version, and tags.
+        /// </summary>
+        /// <typeparam name="TResult">Return Type of the TaskActivity.Execute method</typeparam>
+        /// <param name="name">Name of the orchestration as specified by the ObjectCreator</param>
+        /// <param name="version">Name of the orchestration as specified by the ObjectCreator</param>
+        /// <param name="parameters">Parameters for the TaskActivity.Execute method</param>
+        /// <param name="options">Options for scheduling a task</param>
+        public virtual Task<TResult> ScheduleTask<TResult>(string name, string version, ScheduleTaskOptions options, params object[] parameters)
+        {
+            throw new NotImplementedException();
+        }
+
         /// <summary>
         ///     Schedule a TaskActivity by name and version.
         /// </summary>
diff --git a/src/DurableTask.Core/OrchestrationRuntimeState.cs b/src/DurableTask.Core/OrchestrationRuntimeState.cs
index 494071dbd..3c10fe693 100644
--- a/src/DurableTask.Core/OrchestrationRuntimeState.cs
+++ b/src/DurableTask.Core/OrchestrationRuntimeState.cs
@@ -368,6 +368,7 @@ HistoryEvent GenerateAbridgedEvent(HistoryEvent evt)
                     Name = taskScheduledEvent.Name,
                     Version = taskScheduledEvent.Version,
                     Input = "[..snipped..]",
+                    Tags = taskScheduledEvent.Tags,
                 };
             }
             else if (evt is TaskCompletedEvent taskCompletedEvent)
diff --git a/src/DurableTask.Core/RetryOptions.cs b/src/DurableTask.Core/RetryOptions.cs
index 9f8ee2eac..6aa71fa0a 100644
--- a/src/DurableTask.Core/RetryOptions.cs
+++ b/src/DurableTask.Core/RetryOptions.cs
@@ -43,6 +43,26 @@ public RetryOptions(TimeSpan firstRetryInterval, int maxNumberOfAttempts)
             Handle = e => true;
         }
 
+        /// <summary>
+        /// Creates a new instance of RetryOptions by copying values from an existing instance
+        /// </summary>
+        /// <param name="source">The RetryOptions instance to copy from</param>
+        /// <exception cref="ArgumentNullException"></exception>
+        public RetryOptions(RetryOptions source)
+        {
+            if (source == null)
+            {
+                throw new ArgumentNullException(nameof(source));
+            }
+
+            FirstRetryInterval = source.FirstRetryInterval;
+            MaxNumberOfAttempts = source.MaxNumberOfAttempts;
+            MaxRetryInterval = source.MaxRetryInterval;
+            BackoffCoefficient = source.BackoffCoefficient;
+            RetryTimeout = source.RetryTimeout;
+            Handle = source.Handle;
+        }
+
         /// <summary>
         /// Gets or sets the first retry interval
         /// </summary>
diff --git a/src/DurableTask.Core/ScheduleTaskOptions.cs b/src/DurableTask.Core/ScheduleTaskOptions.cs
new file mode 100644
index 000000000..5261ec80a
--- /dev/null
+++ b/src/DurableTask.Core/ScheduleTaskOptions.cs
@@ -0,0 +1,143 @@
+//  ----------------------------------------------------------------------------------
+//  Copyright Microsoft Corporation
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//  http://www.apache.org/licenses/LICENSE-2.0
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+//  ----------------------------------------------------------------------------------
+
+#nullable enable
+
+namespace DurableTask.Core
+{
+    using System;
+    using System.Collections.Generic;
+
+    /// <summary>
+    /// Options for scheduling a task.
+    /// </summary>
+    public class ScheduleTaskOptions
+    {
+        /// <summary>
+        /// Initializes a new instance of the <see cref="ScheduleTaskOptions"/> class.
+        /// </summary>
+        protected ScheduleTaskOptions()
+        {
+        }
+
+        /// <summary>
+        /// Dictionary of key/value tags associated with this instance.
+        /// </summary>
+        public IDictionary<string, string>? Tags { get; internal set; }
+
+        /// <summary>
+        /// Gets or sets the retry options for the scheduled task.
+        /// </summary>
+        public RetryOptions? RetryOptions { get; internal set; }
+
+        /// <summary>
+        /// Creates a new builder for constructing a <see cref="ScheduleTaskOptions"/> instance.
+        /// </summary>
+        /// <returns>A new builder for creating schedule task options.</returns>
+        public static Builder CreateBuilder()
+        {
+            return new Builder();
+        }
+
+        /// <summary>
+        /// Builder class for creating instances of <see cref="ScheduleTaskOptions"/>.
+        /// </summary>
+        public class Builder
+        {
+            private readonly ScheduleTaskOptions options;
+
+            /// <summary>
+            /// Initializes a new instance of the <see cref="Builder"/> class.
+            /// </summary>
+            internal Builder()
+            {
+                this.options = new ScheduleTaskOptions();
+            }
+
+            /// <summary>
+            /// Sets the tags for the schedule task options.
+            /// </summary>
+            /// <param name="tags">The dictionary of key/value tags.</param>
+            /// <returns>The builder instance.</returns>
+            public Builder WithTags(IDictionary<string, string> tags)
+            {
+                this.options.Tags = new Dictionary<string, string>(tags);
+                return this;
+            }
+
+            /// <summary>
+            /// Adds a tag to the schedule task options.
+            /// </summary>
+            /// <param name="key">The tag key.</param>
+            /// <param name="value">The tag value.</param>
+            /// <returns>The builder instance.</returns>
+            public Builder AddTag(string key, string value)
+            {
+                if (this.options.Tags == null)
+                {
+                    this.options.Tags = new Dictionary<string, string>();
+                }
+
+                this.options.Tags[key] = value;
+                return this;
+            }
+
+            /// <summary>
+            /// Sets the retry options for the scheduled task.
+            /// </summary>
+            /// <param name="retryOptions">The retry options to use.</param>
+            /// <returns>The builder instance.</returns>
+            public Builder WithRetryOptions(RetryOptions retryOptions)
+            {
+                this.options.RetryOptions = retryOptions == null ? null : new RetryOptions(retryOptions);
+                return this;
+            }
+
+            /// <summary>
+            /// Sets the retry options for the scheduled task with the specified parameters.
+            /// </summary>
+            /// <param name="firstRetryInterval">Timespan to wait for the first retry.</param>
+            /// <param name="maxNumberOfAttempts">Max number of attempts to retry.</param>
+            /// <returns>The builder instance.</returns>
+            public Builder WithRetryOptions(TimeSpan firstRetryInterval, int maxNumberOfAttempts)
+            {
+                this.options.RetryOptions = new RetryOptions(firstRetryInterval, maxNumberOfAttempts);
+                return this;
+            }
+
+            /// <summary>
+            /// Sets the retry options for the scheduled task with the specified parameters and configures additional properties.
+            /// </summary>
+            /// <param name="firstRetryInterval">Timespan to wait for the first retry.</param>
+            /// <param name="maxNumberOfAttempts">Max number of attempts to retry.</param>
+            /// <param name="configureRetryOptions">Action to configure additional retry option properties.</param>
+            /// <returns>The builder instance.</returns>
+            public Builder WithRetryOptions(TimeSpan firstRetryInterval, int maxNumberOfAttempts, Action<RetryOptions> configureRetryOptions)
+            {
+                var retryOptions = new RetryOptions(firstRetryInterval, maxNumberOfAttempts);
+                configureRetryOptions?.Invoke(retryOptions);
+                this.options.RetryOptions = retryOptions;
+                return this;
+            }
+
+            /// <summary>
+            /// Builds the <see cref="ScheduleTaskOptions"/> instance.
+            /// </summary>
+            /// <returns>The built schedule task options.</returns>
+            public ScheduleTaskOptions Build()
+            {
+                return this.options;
+            }
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/DurableTask.Core/TaskOrchestrationContext.cs b/src/DurableTask.Core/TaskOrchestrationContext.cs
index 7908eeb07..b12cb1b08 100644
--- a/src/DurableTask.Core/TaskOrchestrationContext.cs
+++ b/src/DurableTask.Core/TaskOrchestrationContext.cs
@@ -86,10 +86,25 @@ public override async Task<TResult> ScheduleTask<TResult>(string name, string ve
             return result;
         }
 
+        public override async Task<TResult> ScheduleTask<TResult>(string name, string version,
+            ScheduleTaskOptions options, params object[] parameters)
+        {
+            if (options.RetryOptions != null)
+            {
+                Task<TResult> RetryCall() => ScheduleTask<TResult>(name, version, ScheduleTaskOptions.CreateBuilder().WithTags(options.Tags).Build(), parameters);
+                var retryInterceptor = new RetryInterceptor<TResult>(this, options.RetryOptions, RetryCall);
+                return await retryInterceptor.Invoke();
+            }
+ 
+            TResult result = await ScheduleTaskToWorker<TResult>(name, version, null, options, parameters);
+ 
+            return result;
+        }
+
         public async Task<TResult> ScheduleTaskToWorker<TResult>(string name, string version, string taskList,
-            params object[] parameters)
+            ScheduleTaskOptions options, params object[] parameters)
         {
-            object result = await ScheduleTaskInternal(name, version, taskList, typeof(TResult), parameters);
+            object result = await ScheduleTaskInternal(name, version, taskList, typeof(TResult), options, parameters);
 
             if (result == null)
             {
@@ -99,8 +114,14 @@ public async Task<TResult> ScheduleTaskToWorker<TResult>(string name, string ver
             return (TResult)result;
         }
 
-        public async Task<object> ScheduleTaskInternal(string name, string version, string taskList, Type resultType,
+        public async Task<TResult> ScheduleTaskToWorker<TResult>(string name, string version, string taskList,
             params object[] parameters)
+        {
+            return await ScheduleTaskToWorker<TResult>(name, version, taskList, ScheduleTaskOptions.CreateBuilder().Build(), parameters);
+        }
+
+        public async Task<object> ScheduleTaskInternal(string name, string version, string taskList, Type resultType,
+            ScheduleTaskOptions options, params object[] parameters)
         {
             int id = this.idCounter++;
             string serializedInput = this.MessageDataConverter.SerializeInternal(parameters);
@@ -111,6 +132,7 @@ public async Task<object> ScheduleTaskInternal(string name, string version, stri
                 Version = version,
                 Tasklist = taskList,
                 Input = serializedInput,
+                Tags = options.Tags,
             };
 
             this.orchestratorActionsMap.Add(id, scheduleTaskTaskAction);
@@ -123,6 +145,13 @@ public async Task<object> ScheduleTaskInternal(string name, string version, stri
             return this.MessageDataConverter.Deserialize(serializedResult, resultType);
         }
 
+
+        public async Task<object> ScheduleTaskInternal(string name, string version, string taskList, Type resultType,
+            params object[] parameters)
+        {
+            return await ScheduleTaskInternal(name, version, taskList, resultType, ScheduleTaskOptions.CreateBuilder().Build(), parameters);
+        }
+
         public override Task<T> CreateSubOrchestrationInstance<T>(
             string name,
             string version,
diff --git a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs
index d1cb03f87..64f926f7a 100644
--- a/src/DurableTask.Core/TaskOrchestrationDispatcher.cs
+++ b/src/DurableTask.Core/TaskOrchestrationDispatcher.cs
@@ -1050,12 +1050,13 @@ TaskMessage ProcessScheduleTaskDecision(
             }
 
             var taskMessage = new TaskMessage();
-
             var scheduledEvent = new TaskScheduledEvent(
                 eventId: scheduleTaskOrchestratorAction.Id,
                 name: scheduleTaskOrchestratorAction.Name,
                 version: scheduleTaskOrchestratorAction.Version,
-                input: scheduleTaskOrchestratorAction.Input);
+                input: scheduleTaskOrchestratorAction.Input) {
+                    Tags = scheduleTaskOrchestratorAction.Tags
+                };
 
             ActivitySpanId clientSpanId = ActivitySpanId.CreateRandom();
 
@@ -1074,7 +1075,9 @@ TaskMessage ProcessScheduleTaskDecision(
                 scheduledEvent = new TaskScheduledEvent(
                     eventId: scheduleTaskOrchestratorAction.Id,
                     name: scheduleTaskOrchestratorAction.Name,
-                    version: scheduleTaskOrchestratorAction.Version);
+                    version: scheduleTaskOrchestratorAction.Version) {
+                        Tags = scheduleTaskOrchestratorAction.Tags
+                    };
 
                 if (parentTraceActivity != null)
                 {
diff --git a/test/DurableTask.AzureStorage.Tests/ScheduleTaskTests.cs b/test/DurableTask.AzureStorage.Tests/ScheduleTaskTests.cs
index a607933f4..c7e01b971 100644
--- a/test/DurableTask.AzureStorage.Tests/ScheduleTaskTests.cs
+++ b/test/DurableTask.AzureStorage.Tests/ScheduleTaskTests.cs
@@ -13,9 +13,10 @@
 
 namespace DurableTask.AzureStorage.Tests
 {
-    using System.Threading.Tasks;
     using DurableTask.Core;
     using Microsoft.VisualStudio.TestTools.UnitTesting;
+    using System.Collections.Generic;
+    using System.Threading.Tasks;
 
     [TestClass]
     public class ScheduleTaskTests