diff --git a/sdks/python/apache_beam/io/watch.py b/sdks/python/apache_beam/io/watch.py
index 2fcee7a8080f..b0ff2bd6a075 100644
--- a/sdks/python/apache_beam/io/watch.py
+++ b/sdks/python/apache_beam/io/watch.py
@@ -59,12 +59,11 @@ def poll(prefix) -> PollResult[str]:
 import inspect
 import time
 import typing
+from collections.abc import Iterable
 from typing import Any
 from typing import Callable
 from typing import Generic
-from typing import Iterable
 from typing import Optional
-from typing import Tuple
 from typing import TypeVar
 
 from apache_beam import coders
@@ -111,7 +110,7 @@ class PollResult(Generic[OutputT]):
   The ``OutputT`` type parameter can annotate a poll function's return type,
   as in ``-> PollResult[str]``; the transform infers the output coder from it.
   """
-  outputs: Tuple[TimestampedValue, ...]
+  outputs: tuple[TimestampedValue, ...]
   watermark: Optional[Timestamp] = None
 
   @property
@@ -119,7 +118,7 @@ def is_complete(self) -> bool:
     return self.watermark == MAX_TIMESTAMP
 
   @staticmethod
-  def _normalize(outputs, timestamp) -> Tuple[TimestampedValue, ...]:
+  def _normalize(outputs, timestamp) -> tuple[TimestampedValue, ...]:
     if timestamp is None:
       default_ts = Timestamp.now()
     else:
@@ -428,7 +427,7 @@ def _hash(self, value: Any) -> bytes:
   def current_restriction(self) -> _GrowthState:
     return self._restriction
 
-  def try_claim(self, position: Tuple[PollResult, Any]) -> bool:
+  def try_claim(self, position: tuple[PollResult, Any]) -> bool:
     """Claims one poll round; at most one claim succeeds per ``process()``.
 
     The claim is rejected after a checkpoint already stopped this invocation,
@@ -723,7 +722,7 @@ def expand(self, pcoll):
             output_coder,
             key_fn,
             key_coder,
-            self._now)).with_output_types(Tuple[input_type, value_type])
+            self._now)).with_output_types(tuple[input_type, value_type])
 
 
 def _as_duration(value) -> Duration: