diff --git a/temporalio/client.py b/temporalio/client.py
index 7d0ea9f6a..91cb89425 100644
--- a/temporalio/client.py
+++ b/temporalio/client.py
@@ -4336,12 +4336,20 @@ class SchedulePolicy:
     exhausted.
     """
 
+    keep_original_workflow_id: bool = False
+    """Whether a scheduled workflow keeps its original workflow ID.
+
+    If false, a timestamp may be appended to the scheduled workflow ID for
+    uniqueness.
+    """
+
     @staticmethod
     def _from_proto(pol: temporalio.api.schedule.v1.SchedulePolicies) -> SchedulePolicy:
         return SchedulePolicy(
             overlap=ScheduleOverlapPolicy(int(pol.overlap_policy)),
             catchup_window=pol.catchup_window.ToTimedelta(),
             pause_on_failure=pol.pause_on_failure,
+            keep_original_workflow_id=pol.keep_original_workflow_id,
         )
 
     def _to_proto(self) -> temporalio.api.schedule.v1.SchedulePolicies:
@@ -4353,6 +4361,7 @@ def _to_proto(self) -> temporalio.api.schedule.v1.SchedulePolicies:
             ),
             catchup_window=catchup_window,
             pause_on_failure=self.pause_on_failure,
+            keep_original_workflow_id=self.keep_original_workflow_id,
         )
 
 
diff --git a/tests/test_client.py b/tests/test_client.py
index 833c97fb0..eda7a9f8c 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -1022,6 +1022,88 @@ async def list_ids() -> list[str]:
     await assert_no_schedules(client)
 
 
+async def test_schedule_can_create_with_keep_original_workflow_id_policy(
+    client: Client, worker: ExternalWorker, env: WorkflowEnvironment
+):
+    if env.supports_time_skipping:
+        pytest.skip("Java test server doesn't support schedules")
+
+    schedule_id = f"can-create-schedule-with-keep-original-workflow-id-{uuid.uuid4()}"
+    handle = await client.create_schedule(
+        schedule_id,
+        Schedule(
+            spec=ScheduleSpec(
+                intervals=[ScheduleIntervalSpec(every=timedelta(hours=1))]
+            ),
+            action=ScheduleActionStartWorkflow(
+                "kitchen_sink",
+                KSWorkflowParams(
+                    actions=[KSAction(result=KSResultAction("some result"))]
+                ),
+                id=f"{schedule_id}-workflow",
+                task_queue=worker.task_queue,
+            ),
+            policy=SchedulePolicy(keep_original_workflow_id=True),
+        ),
+    )
+
+    try:
+        desc = await handle.describe()
+        assert desc.schedule.policy.keep_original_workflow_id
+        assert desc.raw_description.schedule.policies.keep_original_workflow_id
+    finally:
+        await handle.delete()
+
+
+async def test_schedule_can_update_keep_original_workflow_id_policy(
+    client: Client, worker: ExternalWorker, env: WorkflowEnvironment
+):
+    if env.supports_time_skipping:
+        pytest.skip("Java test server doesn't support schedules")
+
+    schedule_id = f"can-update-keep-original-workflow-id-{uuid.uuid4()}"
+    handle = await client.create_schedule(
+        schedule_id,
+        Schedule(
+            spec=ScheduleSpec(
+                intervals=[ScheduleIntervalSpec(every=timedelta(hours=5))]
+            ),
+            action=ScheduleActionStartWorkflow(
+                "kitchen_sink",
+                KSWorkflowParams(
+                    actions=[KSAction(result=KSResultAction("some result"))]
+                ),
+                id=f"{schedule_id}-workflow",
+                task_queue=worker.task_queue,
+            ),
+        ),
+    )
+
+    try:
+        desc = await handle.describe()
+        assert not desc.schedule.policy.keep_original_workflow_id
+
+        def update_enable(input: ScheduleUpdateInput) -> ScheduleUpdate:
+            input.description.schedule.policy.keep_original_workflow_id = True
+            return ScheduleUpdate(schedule=input.description.schedule)
+
+        await handle.update(update_enable)
+        desc = await handle.describe()
+        assert desc.schedule.policy.keep_original_workflow_id
+        assert desc.raw_description.schedule.policies.keep_original_workflow_id
+
+        def update_disable(input: ScheduleUpdateInput) -> ScheduleUpdate:
+            input.description.schedule.policy.keep_original_workflow_id = False
+            return ScheduleUpdate(schedule=input.description.schedule)
+
+        await handle.update(update_disable)
+        desc = await handle.describe()
+        assert not desc.schedule.policy.keep_original_workflow_id
+        assert not desc.raw_description.schedule.policies.keep_original_workflow_id
+    finally:
+        await handle.delete()
+
+
 async def test_schedule_calendar_spec_defaults(
     client: Client, worker: ExternalWorker, env: WorkflowEnvironment
 ):