diff --git a/CHANGES.md b/CHANGES.md
index f98790bc2e19..14c7e676f180 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -84,6 +84,7 @@
   `jakarta.jms.*`. The module now requires Java 17 at runtime ([#38729](https://github.com/apache/beam/issues/38729)).
 * (Python) `Regex.find_all` with `group=Regex.ALL` now returns every group, as `(match, group1, group2, ...)`, instead of only group 1, and with `outputEmpty=False` it drops a match only when the whole match is empty, matching the `group=0` behavior ([#40399](https://github.com/apache/beam/pull/40399)).
 * (Python) `Regex.all_matches` now returns every group of the regex, instead of failing with a `TypeError` when the regex has no groups and dropping nested or unmatched trailing groups ([#40562](https://github.com/apache/beam/issues/40562)).
+* (Python) `--dataflow_endpoint` now defaults to `''` instead of `https://dataflow.googleapis.com`. When unset, the Google Cloud client library resolves the Dataflow endpoint and honors `GOOGLE_CLOUD_UNIVERSE_DOMAIN`, so Dataflow jobs can be submitted in non-default universes. Code that reads `GoogleCloudOptions.dataflow_endpoint` should handle an empty value ([#40569](https://github.com/apache/beam/issues/40569)).
 
 ## Deprecations
 
diff --git a/sdks/python/apache_beam/options/pipeline_options.py b/sdks/python/apache_beam/options/pipeline_options.py
index 84c0743ceaaa..08ba34dabee4 100644
--- a/sdks/python/apache_beam/options/pipeline_options.py
+++ b/sdks/python/apache_beam/options/pipeline_options.py
@@ -1038,10 +1038,12 @@ class GoogleCloudOptions(PipelineOptions):
   def _add_argparse_args(cls, parser):
     parser.add_argument(
         '--dataflow_endpoint',
-        default=cls.DATAFLOW_ENDPOINT,
+        default='',
         help=(
-            'The URL for the Dataflow API. If not set, the default public URL '
-            'will be used.'))
+            'The URL for the Dataflow API. If not set, the endpoint is '
+            'resolved by the Google Cloud client library, which honors '
+            'GOOGLE_CLOUD_UNIVERSE_DOMAIN and otherwise uses the default '
+            'public URL (%s).' % cls.DATAFLOW_ENDPOINT))
     # Remote execution must check that this option is not None.
     parser.add_argument(
         '--project',
diff --git a/sdks/python/apache_beam/options/pipeline_options_validator.py b/sdks/python/apache_beam/options/pipeline_options_validator.py
index 9b0d2e6109d6..545051546413 100644
--- a/sdks/python/apache_beam/options/pipeline_options_validator.py
+++ b/sdks/python/apache_beam/options/pipeline_options_validator.py
@@ -168,9 +168,9 @@ def is_service_runner(self):
 
     dataflow_endpoint = (
         self.options.view_as(GoogleCloudOptions).dataflow_endpoint)
-    if dataflow_endpoint is None:
-      return False
-    else:
+    # An unset endpoint means the default Dataflow service endpoint,
+    # so only an explicit localhost endpoint opts out.
+    if dataflow_endpoint:
       endpoint_parts = urlparse(dataflow_endpoint, allow_fragments=False)
       if endpoint_parts.netloc.startswith("localhost"):
         return False
@@ -274,10 +274,9 @@ def validate_cloud_options(self, view):
       else:
         view.region = default_region
     dataflow_endpoint = view.dataflow_endpoint
-    if dataflow_endpoint is None:
-      errors.extend(
-          self._validate_error(self.ERR_MISSING_OPTION, dataflow_endpoint))
-    else:
+    # When dataflow_endpoint is unset, the client library resolves
+    # the endpoint (honoring GOOGLE_CLOUD_UNIVERSE_DOMAIN).
+    if dataflow_endpoint:
       valid_endpoint = self.validate_endpoint_url(dataflow_endpoint)
       if valid_endpoint is False:
         errors.extend(
diff --git a/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py b/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py
index 654808373f24..8a478f3fb633 100644
--- a/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py
+++ b/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py
@@ -180,12 +180,7 @@ def __init__(
               name=package))
 
     pool = dataflow.WorkerPool(
-        kind='local' if self.local else 'harness',
-        packages=package_descriptors,
-        taskrunner_settings=dataflow.TaskRunnerSettings(
-            parallel_worker_settings=dataflow.WorkerSettings(
-                base_url=GoogleCloudOptions.DATAFLOW_ENDPOINT,
-                service_path=self.google_cloud_options.dataflow_endpoint)))
+        kind='local' if self.local else 'harness', packages=package_descriptors)
 
     pool.autoscaling_settings = dataflow.AutoscalingSettings()
     # Set worker pool options received through command line.
@@ -902,7 +897,7 @@ def submit_job_description(self, job: Job) -> dataflow.Job:
           'HTTP status %d trying to create job'
           ' at dataflow service endpoint %s',
           e.code,
-          self.google_cloud_options.dataflow_endpoint)
+          self._jobs_client.api_endpoint)
       _LOGGER.fatal('details of server error: %s', e)
       raise
 
diff --git a/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py b/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py
index fdce35b4b5ef..b82c7f550673 100644
--- a/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py
+++ b/sdks/python/apache_beam/runners/dataflow/internal/apiclient_test.py
@@ -1320,12 +1320,19 @@ def test_template_file_generation_with_upload_graph(self):
           self.assertFalse(template_obj.get('steps'))
           self.assertTrue(template_obj['stepsLocation'])
 
+  @mock.patch.dict(
+      os.environ, {'GOOGLE_CLOUD_UNIVERSE_DOMAIN': 'custom-universe.com'})
   def test_dataflow_endpoint_clean(self):
+    # An explicit endpoint is cleaned and passed as an api_endpoint override to
+    # every Dataflow client, regardless of GOOGLE_CLOUD_UNIVERSE_DOMAIN.
     endpoints_and_expectations = [
         # (input_endpoint, expected_endpoint, expected_transport)
         ('https://dataflow.googleapis.com/', 'dataflow.googleapis.com', None),
         ('https://dataflow.googleapis.com   ', 'dataflow.googleapis.com', None),
         ('dataflow.googleapis.com/', 'dataflow.googleapis.com', None),
+        ('https://dataflow.example.com/', 'dataflow.example.com', None),
+        ('https://dataflow.example.com   ', 'dataflow.example.com', None),
+        ('dataflow.example.com/', 'dataflow.example.com', None),
         ('http://localhost:8080/', 'http://localhost:8080', 'rest'),
         ('localhost:8080/', 'localhost:8080', 'rest'),
     ]
@@ -1341,20 +1348,55 @@ def test_dataflow_endpoint_clean(self):
           '--no_auth',
       ])
       with mock.patch('apache_beam.runners.dataflow.internal.apiclient.dataflow'
-                      '.JobsV1Beta3Client') as mock_jobs:
-        with mock.patch(
-            'apache_beam.runners.dataflow.internal.apiclient.dataflow'
-            '.MessagesV1Beta3Client'):
-          with mock.patch(
-              'apache_beam.runners.dataflow.internal.apiclient.dataflow'
-              '.MetricsV1Beta3Client'):
-            apiclient.DataflowApplicationClient(pipeline_options)
-            mock_jobs.assert_called_once()
-            called_kwargs = mock_jobs.call_args.kwargs
-            client_opts = called_kwargs.get('client_options')
-            self.assertIsNotNone(client_opts)
-            self.assertEqual(client_opts.api_endpoint, expected_ep)
-            self.assertEqual(called_kwargs.get('transport'), expected_transport)
+                      '.JobsV1Beta3Client') as mock_jobs, \
+          mock.patch('apache_beam.runners.dataflow.internal.apiclient.dataflow'
+                     '.MessagesV1Beta3Client') as mock_messages, \
+          mock.patch('apache_beam.runners.dataflow.internal.apiclient.dataflow'
+                     '.MetricsV1Beta3Client') as mock_metrics:
+        apiclient.DataflowApplicationClient(pipeline_options)
+        for mock_client in (mock_jobs, mock_messages, mock_metrics):
+          mock_client.assert_called_once()
+          called_kwargs = mock_client.call_args.kwargs
+          client_opts = called_kwargs.get('client_options')
+          self.assertIsNotNone(client_opts)
+          self.assertEqual(client_opts.api_endpoint, expected_ep)
+          self.assertEqual(called_kwargs.get('transport'), expected_transport)
+
+  @mock.patch.dict(
+      os.environ,
+      {
+          # Pin mTLS off so endpoint resolution doesn't depend on the host.
+          'GOOGLE_API_USE_CLIENT_CERTIFICATE': 'false',
+          'GOOGLE_API_USE_MTLS_ENDPOINT': 'never',
+      })
+  def test_dataflow_endpoint_unset_uses_client_library_default(self):
+    # Without an explicit endpoint, no api_endpoint override is passed, so the
+    # client library resolves the endpoint and honors
+    # GOOGLE_CLOUD_UNIVERSE_DOMAIN.
+    pipeline_options = PipelineOptions([
+        '--project',
+        'test-project',
+        '--temp_location',
+        'gs://test-location/temp',
+        '--no_auth',
+    ])
+    self.assertEqual(
+        pipeline_options.view_as(GoogleCloudOptions).dataflow_endpoint, '')
+
+    def resolved_endpoints():
+      client = apiclient.DataflowApplicationClient(pipeline_options)
+      return {
+          c.api_endpoint
+          for c in (
+              client._jobs_client, client._messages_client,
+              client._metrics_client)
+      }
+
+    os.environ.pop('GOOGLE_CLOUD_UNIVERSE_DOMAIN', None)
+    self.assertEqual(resolved_endpoints(), {'dataflow.googleapis.com'})
+
+    os.environ['GOOGLE_CLOUD_UNIVERSE_DOMAIN'] = 'custom-universe.com'
+    self.assertEqual(resolved_endpoints(), {'dataflow.custom-universe.com'})
 
   def test_stage_resources(self):
     pipeline_options = PipelineOptions([