diff --git a/.github/scripts/cloud_namespace.py b/.github/scripts/cloud_namespace.py
index c13d1e65e..456f649fc 100644
--- a/.github/scripts/cloud_namespace.py
+++ b/.github/scripts/cloud_namespace.py
@@ -51,8 +51,10 @@ async def wait_for_operation(
 
 async def create() -> None:
     client = await cloud_client()
-    namespace_name = "sdk-python-ci-{}-{}".format(
-        os.environ["GITHUB_RUN_ID"], os.environ["GITHUB_RUN_ATTEMPT"]
+    namespace_name = "sdk-python-ci-{}-{}{}".format(
+        os.environ["GITHUB_RUN_ID"],
+        os.environ["GITHUB_RUN_ATTEMPT"],
+        os.environ.get("TEMPORAL_CLOUD_NAMESPACE_SUFFIX", ""),
     )
     result = await client.cloud_service.create_namespace(
         CreateNamespaceRequest(
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index b1098dcdc..c632ab98b 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -254,8 +254,9 @@ jobs:
         env:
           TEMPORAL_CLIENT_CLOUD_API_KEY: ${{ secrets.TEMPORAL_CLIENT_CLOUD_API_KEY }}
           TEMPORAL_CLIENT_CLOUD_API_VERSION: v0.19.1
+          TEMPORAL_CLOUD_NAMESPACE_SUFFIX: -general
       - run: mkdir junit-xml
-      - run: poe test -s --workflow-environment envconfig --junit-xml=junit-xml/cloud.xml
+      - run: poe test -s --workflow-environment envconfig --ignore=tests/nexus --junit-xml=junit-xml/cloud.xml
         timeout-minutes: 15
         env:
           TEMPORAL_ADDRESS: ${{ steps.create-cloud-namespace.outputs.namespace }}.tmprl.cloud:7233
@@ -278,6 +279,80 @@ jobs:
           path: junit-xml
           retention-days: 14
 
+  # Nexus endpoint provisioning is slow on Cloud, so run these tests separately.
+  cloud-nexus-test:
+    if: ${{ github.event.pull_request.head.repo.full_name == '' || github.event.pull_request.head.repo.full_name == 'temporalio/sdk-python' }}
+    timeout-minutes: 60
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+        with:
+          submodules: recursive
+      - uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable
+      - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 # v5
+        with:
+          python-version: "3.14"
+      - uses: Swatinem/rust-cache@e18b497796c12c097a38f9edb9d0641fb99eee32 # v2
+        with:
+          workspaces: temporalio/bridge -> target
+          key: ${{ env.pythonLocation }}
+      - uses: arduino/setup-protoc@c65c819552d16ad3c9b72d9dfd5ba5237b9c906b # v3
+        with:
+          version: "23.x"
+          repo-token: ${{ secrets.GITHUB_TOKEN }}
+      - uses: astral-sh/setup-uv@cec208311dfd045dd5311c1add060b2062131d57 # v8
+      - run: uv tool install poethepoet
+      - run: uv sync --all-extras
+      - run: poe build-develop
+      - name: Generate Cloud test certificates
+        run: |
+          cert_dir="$RUNNER_TEMP/cloud-test-certs"
+          mkdir "$cert_dir"
+          openssl req -x509 -newkey rsa:2048 -nodes -days 1 \
+            -keyout "$cert_dir/ca.key" -out "$cert_dir/ca.pem" \
+            -subj '/CN=Temporal Python SDK Cloud CI CA'
+          openssl req -newkey rsa:2048 -nodes \
+            -keyout "$cert_dir/client.key" -out "$cert_dir/client.csr" \
+            -subj '/CN=Temporal Python SDK Cloud CI'
+          openssl x509 -req -days 1 -in "$cert_dir/client.csr" \
+            -CA "$cert_dir/ca.pem" -CAkey "$cert_dir/ca.key" -CAcreateserial \
+            -out "$cert_dir/client.pem" -extfile <(printf 'extendedKeyUsage=clientAuth')
+          {
+            echo "TEMPORAL_CLOUD_CLIENT_CA_PATH=$cert_dir/ca.pem"
+            echo "TEMPORAL_TLS_CLIENT_CERT_PATH=$cert_dir/client.pem"
+            echo "TEMPORAL_TLS_CLIENT_KEY_PATH=$cert_dir/client.key"
+          } >> "$GITHUB_ENV"
+      - name: Create Cloud namespace
+        id: create-cloud-namespace
+        run: uv run python .github/scripts/cloud_namespace.py create
+        env:
+          TEMPORAL_CLIENT_CLOUD_API_KEY: ${{ secrets.TEMPORAL_CLIENT_CLOUD_API_KEY }}
+          TEMPORAL_CLIENT_CLOUD_API_VERSION: v0.19.1
+          TEMPORAL_CLOUD_NAMESPACE_SUFFIX: -nexus
+      - run: mkdir junit-xml
+      - run: poe test -n 16 -s --workflow-environment envconfig tests/nexus --junit-xml=junit-xml/cloud-nexus.xml
+        timeout-minutes: 45
+        env:
+          TEMPORAL_ADDRESS: ${{ steps.create-cloud-namespace.outputs.namespace }}.tmprl.cloud:7233
+          TEMPORAL_NAMESPACE: ${{ steps.create-cloud-namespace.outputs.namespace }}
+          TEMPORAL_IS_CLOUD_TESTS: true
+          TEMPORAL_CLIENT_CLOUD_API_KEY: ${{ secrets.TEMPORAL_CLIENT_CLOUD_API_KEY }}
+          TEMPORAL_CLIENT_CLOUD_API_VERSION: v0.19.1
+          TEMPORAL_CLIENT_CLOUD_NAMESPACE: ${{ steps.create-cloud-namespace.outputs.namespace }}
+      - name: Delete Cloud namespace
+        if: ${{ always() && steps.create-cloud-namespace.outputs.namespace != '' }}
+        run: uv run python .github/scripts/cloud_namespace.py delete "${{ steps.create-cloud-namespace.outputs.namespace }}"
+        env:
+          TEMPORAL_CLIENT_CLOUD_API_KEY: ${{ secrets.TEMPORAL_CLIENT_CLOUD_API_KEY }}
+          TEMPORAL_CLIENT_CLOUD_API_VERSION: v0.19.1
+      - name: Upload junit-xml artifacts
+        uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4
+        if: always()
+        with:
+          name: junit-xml--${{github.run_id}}--${{github.run_attempt}}--cloud-nexus
+          path: junit-xml
+          retention-days: 14
+
   # Runs the sdk features repo tests with this repo's current SDK code
   features-tests:
     uses: temporalio/features/.github/workflows/python.yaml@main
diff --git a/tests/nexus/conftest.py b/tests/nexus/conftest.py
new file mode 100644
index 000000000..97b276a75
--- /dev/null
+++ b/tests/nexus/conftest.py
@@ -0,0 +1,193 @@
+import asyncio
+import os
+import time
+from collections.abc import AsyncGenerator
+from dataclasses import dataclass
+
+import pytest
+import pytest_asyncio
+
+from temporalio.api.cloud.cloudservice.v1 import (
+    CreateNexusEndpointRequest,
+    DeleteNexusEndpointRequest,
+    GetAsyncOperationRequest,
+    GetNamespaceRequest,
+    GetNexusEndpointRequest,
+)
+from temporalio.api.cloud.nexus.v1 import (
+    AllowedCloudNamespacePolicySpec,
+    Endpoint,
+    EndpointPolicySpec,
+    EndpointSpec,
+    EndpointTargetSpec,
+    WorkerTargetSpec,
+)
+from temporalio.api.cloud.operation.v1 import AsyncOperation
+from temporalio.api.cloud.resource.v1 import ResourceState
+from temporalio.api.operatorservice.v1 import (
+    GetNexusEndpointRequest as GetDataPlaneNexusEndpointRequest,
+)
+from temporalio.client import CloudOperationsClient
+from temporalio.service import RPCError, RPCStatusCode
+from temporalio.testing import WorkflowEnvironment
+
+
+@dataclass
+class _CloudNexusEndpointClient:
+    client: CloudOperationsClient
+    namespace_id: str
+
+    async def wait_for_operation(self, operation: AsyncOperation) -> None:
+        deadline = time.monotonic() + 10 * 60
+        while True:
+            operation = (
+                await self.client.cloud_service.get_async_operation(
+                    GetAsyncOperationRequest(async_operation_id=operation.id)
+                )
+            ).async_operation
+            if operation.state == AsyncOperation.STATE_FULFILLED:
+                return
+            if operation.state in {
+                AsyncOperation.STATE_FAILED,
+                AsyncOperation.STATE_CANCELLED,
+                AsyncOperation.STATE_REJECTED,
+            }:
+                raise RuntimeError(
+                    "Cloud operation "
+                    f"{operation.id} "
+                    f"{AsyncOperation.State.Name(operation.state).lower()}: "
+                    f"{operation.failure_reason}"
+                )
+            if time.monotonic() >= deadline:
+                raise TimeoutError(
+                    f"Timed out waiting for Cloud operation {operation.id}"
+                )
+            delay = max(
+                operation.check_duration.seconds
+                + operation.check_duration.nanos / 1_000_000_000,
+                1,
+            )
+            await asyncio.sleep(min(delay, deadline - time.monotonic()))
+
+    async def wait_for_endpoint(self, endpoint_id: str) -> Endpoint:
+        deadline = time.monotonic() + 10 * 60
+        while True:
+            endpoint = (
+                await self.client.cloud_service.get_nexus_endpoint(
+                    GetNexusEndpointRequest(endpoint_id=endpoint_id)
+                )
+            ).endpoint
+            if endpoint.state == ResourceState.RESOURCE_STATE_ACTIVE:
+                return endpoint
+            if endpoint.state in {
+                ResourceState.RESOURCE_STATE_ACTIVATION_FAILED,
+                ResourceState.RESOURCE_STATE_UPDATE_FAILED,
+                ResourceState.RESOURCE_STATE_DELETE_FAILED,
+                ResourceState.RESOURCE_STATE_SUSPENDED,
+                ResourceState.RESOURCE_STATE_EXPIRED,
+            }:
+                raise RuntimeError(
+                    "Cloud Nexus endpoint "
+                    f"{endpoint_id} "
+                    f"{ResourceState.Name(endpoint.state).lower()}"
+                )
+            if time.monotonic() >= deadline:
+                raise TimeoutError(
+                    f"Timed out waiting for Cloud Nexus endpoint {endpoint_id}"
+                )
+            await asyncio.sleep(1)
+
+    async def wait_for_data_plane_endpoint(
+        self, env: WorkflowEnvironment, endpoint_id: str
+    ) -> None:
+        deadline = time.monotonic() + 10 * 60
+        while True:
+            try:
+                await env.client.operator_service.get_nexus_endpoint(
+                    GetDataPlaneNexusEndpointRequest(id=endpoint_id)
+                )
+                return
+            except RPCError as err:
+                if err.status != RPCStatusCode.NOT_FOUND:
+                    raise
+            if time.monotonic() >= deadline:
+                raise TimeoutError(
+                    "Timed out waiting for Cloud Nexus endpoint "
+                    f"{endpoint_id} to reach the data plane"
+                )
+            await asyncio.sleep(1)
+
+
+@pytest_asyncio.fixture(scope="session")  # type: ignore[reportUntypedFunctionDecorator]
+async def cloud_nexus_endpoint_client() -> AsyncGenerator[
+    _CloudNexusEndpointClient | None, None
+]:
+    if "TEMPORAL_IS_CLOUD_TESTS" not in os.environ:
+        yield None
+        return
+
+    client = await CloudOperationsClient.connect(
+        api_key=os.environ["TEMPORAL_CLIENT_CLOUD_API_KEY"],
+        version=os.environ["TEMPORAL_CLIENT_CLOUD_API_VERSION"],
+    )
+    namespace = await client.cloud_service.get_namespace(
+        GetNamespaceRequest(namespace=os.environ["TEMPORAL_NAMESPACE"])
+    )
+    yield _CloudNexusEndpointClient(client, namespace.namespace.namespace)
+
+
+@pytest_asyncio.fixture(autouse=True)  # type: ignore[reportUntypedFunctionDecorator]
+async def cloud_nexus_endpoints(
+    cloud_nexus_endpoint_client: _CloudNexusEndpointClient | None,
+    env: WorkflowEnvironment,
+    monkeypatch: pytest.MonkeyPatch,
+) -> AsyncGenerator[None, None]:
+    if cloud_nexus_endpoint_client is None:
+        yield
+        return
+
+    endpoints: list[Endpoint] = []
+
+    async def create_nexus_endpoint(endpoint_name: str, task_queue: str) -> Endpoint:
+        response = await cloud_nexus_endpoint_client.client.cloud_service.create_nexus_endpoint(
+            CreateNexusEndpointRequest(
+                spec=EndpointSpec(
+                    name=endpoint_name,
+                    target_spec=EndpointTargetSpec(
+                        worker_target_spec=WorkerTargetSpec(
+                            namespace_id=cloud_nexus_endpoint_client.namespace_id,
+                            task_queue=task_queue,
+                        )
+                    ),
+                    policy_specs=[
+                        EndpointPolicySpec(
+                            allowed_cloud_namespace_policy_spec=AllowedCloudNamespacePolicySpec(
+                                namespace_id=cloud_nexus_endpoint_client.namespace_id
+                            )
+                        )
+                    ],
+                )
+            )
+        )
+        await cloud_nexus_endpoint_client.wait_for_operation(response.async_operation)
+        endpoint = await cloud_nexus_endpoint_client.wait_for_endpoint(
+            response.endpoint_id
+        )
+        await cloud_nexus_endpoint_client.wait_for_data_plane_endpoint(env, endpoint.id)
+        endpoints.append(endpoint)
+        return endpoint
+
+    monkeypatch.setattr(env, "create_nexus_endpoint", create_nexus_endpoint)
+    try:
+        yield
+    finally:
+        for endpoint in reversed(endpoints):
+            response = await cloud_nexus_endpoint_client.client.cloud_service.delete_nexus_endpoint(
+                DeleteNexusEndpointRequest(
+                    endpoint_id=endpoint.id,
+                    resource_version=endpoint.resource_version,
+                )
+            )
+            await cloud_nexus_endpoint_client.wait_for_operation(
+                response.async_operation
+            )
diff --git a/tests/nexus/test_dynamic_creation_of_user_handler_classes.py b/tests/nexus/test_dynamic_creation_of_user_handler_classes.py
index 214e02ab9..f7306a46b 100644
--- a/tests/nexus/test_dynamic_creation_of_user_handler_classes.py
+++ b/tests/nexus/test_dynamic_creation_of_user_handler_classes.py
@@ -10,10 +10,6 @@
 from temporalio.worker import Worker
 from tests.helpers.nexus import make_nexus_endpoint_name
 
-# Cloud CI's namespace credentials cannot manage Nexus endpoints.
-# See https://github.com/temporalio/sdk-python/issues/1704.
-pytestmark = pytest.mark.requires_local_server
-
 
 @workflow.defn
 class MyWorkflow:
diff --git a/tests/nexus/test_nexus_client_updates.py b/tests/nexus/test_nexus_client_updates.py
index 97dd251da..302ede54e 100644
--- a/tests/nexus/test_nexus_client_updates.py
+++ b/tests/nexus/test_nexus_client_updates.py
@@ -3,7 +3,6 @@
 import uuid
 
 import nexusrpc
-import pytest
 from nexusrpc.handler import StartOperationContext, service_handler, sync_operation
 
 import temporalio.nexus
@@ -12,10 +11,6 @@
 from temporalio.testing import WorkflowEnvironment
 from temporalio.worker import Worker
 
-# Cloud CI's namespace credentials cannot manage Nexus endpoints.
-# See https://github.com/temporalio/sdk-python/issues/1704.
-pytestmark = pytest.mark.requires_local_server
-
 
 @nexusrpc.service
 class ClientTestService:
diff --git a/tests/nexus/test_nexus_worker_shutdown.py b/tests/nexus/test_nexus_worker_shutdown.py
index 2a94027d5..bd9063237 100644
--- a/tests/nexus/test_nexus_worker_shutdown.py
+++ b/tests/nexus/test_nexus_worker_shutdown.py
@@ -23,10 +23,6 @@
     make_nexus_endpoint_name,
 )
 
-# Cloud CI's namespace credentials cannot manage Nexus endpoints.
-# See https://github.com/temporalio/sdk-python/issues/1704.
-pytestmark = pytest.mark.requires_local_server
-
 
 @nexusrpc.service
 class ShutdownTestService:
diff --git a/tests/nexus/test_signal_link_propagation_e2e.py b/tests/nexus/test_signal_link_propagation_e2e.py
index 9e51d4b93..e489ad8a7 100644
--- a/tests/nexus/test_signal_link_propagation_e2e.py
+++ b/tests/nexus/test_signal_link_propagation_e2e.py
@@ -56,10 +56,6 @@
     workflow_event_link_event_type,
 )
 
-# Cloud CI's namespace credentials cannot manage Nexus endpoints.
-# See https://github.com/temporalio/sdk-python/issues/1704.
-pytestmark = pytest.mark.requires_local_server
-
 EventType = temporalio.api.enums.v1.EventType
 
 
diff --git a/tests/nexus/test_standalone_operations.py b/tests/nexus/test_standalone_operations.py
index 26a8316b4..c71337b20 100644
--- a/tests/nexus/test_standalone_operations.py
+++ b/tests/nexus/test_standalone_operations.py
@@ -71,11 +71,6 @@
 # ---------------------------------------------------------------------------
 
 
-# Cloud CI's namespace credentials cannot manage Nexus endpoints.
-# See https://github.com/temporalio/sdk-python/issues/1704.
-pytestmark = pytest.mark.requires_local_server
-
-
 @dataclass
 class EchoInput:
     value: str
diff --git a/tests/nexus/test_temporal_extstore.py b/tests/nexus/test_temporal_extstore.py
index f23a69259..2afbac44c 100644
--- a/tests/nexus/test_temporal_extstore.py
+++ b/tests/nexus/test_temporal_extstore.py
@@ -41,10 +41,6 @@
 from tests.helpers.nexus import make_nexus_endpoint_name
 from tests.test_extstore import InMemoryTestDriver
 
-# Cloud CI's namespace credentials cannot manage Nexus endpoints.
-# See https://github.com/temporalio/sdk-python/issues/1704.
-pytestmark = pytest.mark.requires_local_server
-
 PAYLOAD_SIZE = 4096
 PAYLOAD_SIZE_THRESHOLD = 1024
 _STORE_FAILURE_MESSAGE = "external storage store failed"
diff --git a/tests/nexus/test_temporal_operation.py b/tests/nexus/test_temporal_operation.py
index 85948deb3..45a6fb1f0 100644
--- a/tests/nexus/test_temporal_operation.py
+++ b/tests/nexus/test_temporal_operation.py
@@ -40,10 +40,6 @@
     make_nexus_endpoint_name,
 )
 
-# Cloud CI's namespace credentials cannot manage Nexus endpoints.
-# See https://github.com/temporalio/sdk-python/issues/1704.
-pytestmark = pytest.mark.requires_local_server
-
 
 @dataclass
 class Input:
diff --git a/tests/nexus/test_use_existing_conflict_policy.py b/tests/nexus/test_use_existing_conflict_policy.py
index 8ffa9f8f8..e7cee9e1c 100644
--- a/tests/nexus/test_use_existing_conflict_policy.py
+++ b/tests/nexus/test_use_existing_conflict_policy.py
@@ -4,7 +4,6 @@
 import uuid
 from dataclasses import dataclass
 
-import pytest
 from nexusrpc.handler import service_handler
 
 from temporalio import nexus, workflow
@@ -14,10 +13,6 @@
 from temporalio.worker import Worker
 from tests.helpers.nexus import make_nexus_endpoint_name
 
-# Cloud CI's namespace credentials cannot manage Nexus endpoints.
-# See https://github.com/temporalio/sdk-python/issues/1704.
-pytestmark = pytest.mark.requires_local_server
-
 
 @dataclass
 class OpInput:
diff --git a/tests/nexus/test_workflow_caller.py b/tests/nexus/test_workflow_caller.py
index 0c47f17c1..328637525 100644
--- a/tests/nexus/test_workflow_caller.py
+++ b/tests/nexus/test_workflow_caller.py
@@ -90,11 +90,6 @@ class OpDefinitionType(IntEnum):
     LONGHAND = 1
 
 
-# Cloud CI's namespace credentials cannot manage Nexus endpoints.
-# See https://github.com/temporalio/sdk-python/issues/1704.
-pytestmark = pytest.mark.requires_local_server
-
-
 @dataclass
 class SyncResponse:
     op_definition_type: OpDefinitionType
diff --git a/tests/nexus/test_workflow_caller_cancellation_types.py b/tests/nexus/test_workflow_caller_cancellation_types.py
index eca269984..bf33983a5 100644
--- a/tests/nexus/test_workflow_caller_cancellation_types.py
+++ b/tests/nexus/test_workflow_caller_cancellation_types.py
@@ -25,10 +25,6 @@
 from tests.helpers import LogCapturer, assert_event_subsequence, assert_eventually
 from tests.helpers.nexus import make_nexus_endpoint_name
 
-# Cloud CI's namespace credentials cannot manage Nexus endpoints.
-# See https://github.com/temporalio/sdk-python/issues/1704.
-pytestmark = pytest.mark.requires_local_server
-
 
 @dataclass
 class TestContext:
diff --git a/tests/nexus/test_workflow_caller_cancellation_types_when_cancel_handler_fails.py b/tests/nexus/test_workflow_caller_cancellation_types_when_cancel_handler_fails.py
index 44d91a7d1..4cdeeeb15 100644
--- a/tests/nexus/test_workflow_caller_cancellation_types_when_cancel_handler_fails.py
+++ b/tests/nexus/test_workflow_caller_cancellation_types_when_cancel_handler_fails.py
@@ -30,10 +30,6 @@
     has_event,
 )
 
-# Cloud CI's namespace credentials cannot manage Nexus endpoints.
-# See https://github.com/temporalio/sdk-python/issues/1704.
-pytestmark = pytest.mark.requires_local_server
-
 
 @dataclass
 class TestContext:
diff --git a/tests/nexus/test_workflow_caller_error_chains.py b/tests/nexus/test_workflow_caller_error_chains.py
index 1012d8a94..9ff84f405 100644
--- a/tests/nexus/test_workflow_caller_error_chains.py
+++ b/tests/nexus/test_workflow_caller_error_chains.py
@@ -25,10 +25,6 @@
 from temporalio.worker import Worker
 from tests.helpers.nexus import make_nexus_endpoint_name
 
-# Cloud CI's namespace credentials cannot manage Nexus endpoints.
-# See https://github.com/temporalio/sdk-python/issues/1704.
-pytestmark = pytest.mark.requires_local_server
-
 
 @dataclass
 class ExpectedError:
diff --git a/tests/nexus/test_workflow_caller_errors.py b/tests/nexus/test_workflow_caller_errors.py
index 0f1b6a789..9246b9fd7 100644
--- a/tests/nexus/test_workflow_caller_errors.py
+++ b/tests/nexus/test_workflow_caller_errors.py
@@ -42,10 +42,6 @@
 from tests.helpers import LogCapturer, assert_eq_eventually
 from tests.helpers.nexus import make_nexus_endpoint_name
 
-# Cloud CI's namespace credentials cannot manage Nexus endpoints.
-# See https://github.com/temporalio/sdk-python/issues/1704.
-pytestmark = pytest.mark.requires_local_server
-
 operation_invocation_counts = Counter[str]()
 
 logger = getLogger(__name__)
diff --git a/tests/nexus/test_workflow_run_operation.py b/tests/nexus/test_workflow_run_operation.py
index 7135fde71..851f408ec 100644
--- a/tests/nexus/test_workflow_run_operation.py
+++ b/tests/nexus/test_workflow_run_operation.py
@@ -23,10 +23,6 @@
 from temporalio.worker import Worker
 from tests.helpers.nexus import make_nexus_endpoint_name
 
-# Cloud CI's namespace credentials cannot manage Nexus endpoints.
-# See https://github.com/temporalio/sdk-python/issues/1704.
-pytestmark = pytest.mark.requires_local_server
-
 
 @dataclass
 class Input: