Repository: PrefectHQ/prefect
Pull request: Enable `EventsWorker` to emit events to Prefect servers
<!-- 
Thanks for opening a pull request to Prefect! We've got a few requests to help us review contributions:

- Make sure that your title neatly summarizes the proposed changes.
- Provide a short overview of the change and the value it adds.
- Share an example to help us understand the change in user experience.
- Confirm that you've done common tasks so we can give a timely review.
- Review our contribution guidelines: https://docs.prefect.io/latest/contributing/overview/

Happy engineering!
-->

<!-- Include an overview here -->
Adds a `PrefectEventsClient` that enables event emission to OSS Prefect servers. The existing `PrefectCloudEventsClient` inherits from this new client because emitting events to Prefect Cloud requires authentication.

Note that this PR requires a long-running Prefect server. Emitting events via the ephemeral API will be handled in a subsequent PR.

Closes https://github.com/PrefectHQ/prefect/issues/12631

### Example
<!-- 
Share an example of the change in action.

A code blurb is best. Changes to features should include an example that is executable by a new user.
If changing documentation, a link to a preview of the page is great.
 -->
1. Update the following settings in your Prefect profile:
```
PREFECT_API_SERVICES_EVENT_LOGGER_ENABLED='True'
PREFECT_API_URL='http://localhost:4200/api'
PREFECT_EXPERIMENTAL_EVENTS='True'
```
2. Start a Prefect server:
```
prefect server start
```
3. Run the following code:
```
>>> from prefect.events import emit_event
>>>
>>> def some_function(name: str="kiki") -> None:
...     print(f"hi {name}!")
...     emit_event(event=f"{name}.sent.event!", resource={"prefect.resource.id": f"coder{name}"})
...
>>> some_function()
hi kiki!
>>> some_function("alex")
hi alex!
```
4. You should see events in your Prefect server logs:
```
Event: 05b9977a 2024-04-09T21:46:59.839805+00:00  ( -0.07) [kiki.sent.event!]
coder.kiki
Event: 02c01cc1 2024-04-09T21:47:08.559323+00:00  ( -0.00) [alex.sent.event!]
coder.alex
```

### Checklist
<!-- These boxes may be checked after opening the pull request. -->

- [ ] This pull request references any related issue by including "closes `<link to issue>`"
	- If no issue exists and your change is not a small fix, please [create an issue](https://github.com/PrefectHQ/prefect/issues/new/choose) first.
- [x] This pull request includes tests or only affects documentation.
- [x] This pull request includes a label categorizing the change e.g. `maintenance`, `fix`, `feature`, `enhancement`, `docs`.
  <!-- If you do not have permission to add a label, a maintainer will add one for you -->

For documentation changes:

- [ ] This pull request includes redirect settings in `netlify.toml` for files that are removed or renamed.

For new functions or classes in the Python SDK:

- [ ] This pull request includes helpful docstrings.
- [ ] If a new Python file was added, this pull request contains a stub page in the Python SDK docs and an entry in `mkdocs.yml` navigation.

Problem statement:
Update OSS events client to write to OSS server
The `EventsWorker` only emits events when Prefect Cloud is configured as the backend. Now that we can send events to OSS Prefect servers, the `EventsWorker` needs to be updated to emit events to OSS Prefect servers too.

Hints from issue discussion:

Commit message: Rename function

Patch to review:
diff --git a/src/prefect/events/clients.py b/src/prefect/events/clients.py
index ce21dd9cc522..a0bdddefdb1e 100644
--- a/src/prefect/events/clients.py
+++ b/src/prefect/events/clients.py
@@ -118,8 +118,8 @@ def _get_api_url_and_key(
     return api_url, api_key
 
 
-class PrefectCloudEventsClient(EventsClient):
-    """A Prefect Events client that streams Events to a Prefect Cloud Workspace"""
+class PrefectEventsClient(EventsClient):
+    """A Prefect Events client that streams events to a Prefect server"""
 
     _websocket: Optional[WebSocketClientProtocol]
     _unconfirmed_events: List[Event]
@@ -127,20 +127,22 @@ class PrefectCloudEventsClient(EventsClient):
     def __init__(
         self,
         api_url: str = None,
-        api_key: str = None,
         reconnection_attempts: int = 10,
         checkpoint_every: int = 20,
     ):
         """
         Args:
-            api_url: The base URL for a Prefect Cloud workspace
-            api_key: The API of an actor with the manage_events scope
+            api_url: The base URL for a Prefect server
             reconnection_attempts: When the client is disconnected, how many times
                 the client should attempt to reconnect
             checkpoint_every: How often the client should sync with the server to
                 confirm receipt of all previously sent events
         """
-        api_url, api_key = _get_api_url_and_key(api_url, api_key)
+        api_url = api_url or PREFECT_API_URL.value()
+        if not api_url:
+            raise ValueError(
+                "api_url must be provided or set in the Prefect configuration"
+            )
 
         socket_url = (
             api_url.replace("https://", "wss://")
@@ -150,14 +152,13 @@ def __init__(
 
         self._connect = connect(
             socket_url + "/events/in",
-            extra_headers={"Authorization": f"bearer {api_key}"},
         )
         self._websocket = None
         self._reconnection_attempts = reconnection_attempts
         self._unconfirmed_events = []
         self._checkpoint_every = checkpoint_every
 
-    async def __aenter__(self) -> "PrefectCloudEventsClient":
+    async def __aenter__(self) -> "PrefectEventsClient":
         # Don't handle any errors in the initial connection, because these are most
         # likely a permission or configuration issue that should propagate
         await super().__aenter__()
@@ -238,6 +239,46 @@ async def _emit(self, event: Event) -> None:
                     await asyncio.sleep(1)
 
 
+class PrefectCloudEventsClient(PrefectEventsClient):
+    """A Prefect Events client that streams events to a Prefect Cloud Workspace"""
+
+    _websocket: Optional[WebSocketClientProtocol]
+    _unconfirmed_events: List[Event]
+
+    def __init__(
+        self,
+        api_url: str = None,
+        api_key: str = None,
+        reconnection_attempts: int = 10,
+        checkpoint_every: int = 20,
+    ):
+        """
+        Args:
+            api_url: The base URL for a Prefect Cloud workspace
+            api_key: The API of an actor with the manage_events scope
+            reconnection_attempts: When the client is disconnected, how many times
+                the client should attempt to reconnect
+            checkpoint_every: How often the client should sync with the server to
+                confirm receipt of all previously sent events
+        """
+        api_url, api_key = _get_api_url_and_key(api_url, api_key)
+
+        socket_url = (
+            api_url.replace("https://", "wss://")
+            .replace("http://", "ws://")
+            .rstrip("/")
+        )
+
+        self._connect = connect(
+            socket_url + "/events/in",
+            extra_headers={"Authorization": f"bearer {api_key}"},
+        )
+        self._websocket = None
+        self._reconnection_attempts = reconnection_attempts
+        self._unconfirmed_events = []
+        self._checkpoint_every = checkpoint_every
+
+
 SEEN_EVENTS_SIZE = 500_000
 SEEN_EVENTS_TTL = 120
 
diff --git a/src/prefect/events/utilities.py b/src/prefect/events/utilities.py
index 0eb3789f9291..b5afb6c2ff15 100644
--- a/src/prefect/events/utilities.py
+++ b/src/prefect/events/utilities.py
@@ -6,9 +6,13 @@
 
 from prefect._internal.schemas.fields import DateTimeTZ
 
-from .clients import AssertingEventsClient, PrefectCloudEventsClient
+from .clients import (
+    AssertingEventsClient,
+    PrefectCloudEventsClient,
+    PrefectEventsClient,
+)
 from .schemas.events import Event, RelatedResource
-from .worker import EventsWorker, emit_events_to_cloud
+from .worker import EventsWorker, should_emit_events
 
 TIGHT_TIMING = timedelta(minutes=5)
 
@@ -42,10 +46,14 @@ def emit_event(
         The event that was emitted if worker is using a client that emit
         events, otherwise None
     """
-    if not emit_events_to_cloud():
+    if not should_emit_events():
         return None
 
-    operational_clients = [AssertingEventsClient, PrefectCloudEventsClient]
+    operational_clients = [
+        AssertingEventsClient,
+        PrefectCloudEventsClient,
+        PrefectEventsClient,
+    ]
     worker_instance = EventsWorker.instance()
 
     if worker_instance.client_type not in operational_clients:
diff --git a/src/prefect/events/worker.py b/src/prefect/events/worker.py
index e7274667d3b7..a498560a0e92 100644
--- a/src/prefect/events/worker.py
+++ b/src/prefect/events/worker.py
@@ -5,17 +5,38 @@
 from typing_extensions import Self
 
 from prefect._internal.concurrency.services import QueueService
-from prefect.settings import PREFECT_API_KEY, PREFECT_API_URL, PREFECT_CLOUD_API_URL
+from prefect.settings import (
+    PREFECT_API_KEY,
+    PREFECT_API_URL,
+    PREFECT_CLOUD_API_URL,
+    PREFECT_EXPERIMENTAL_EVENTS,
+)
 from prefect.utilities.context import temporary_context
 
-from .clients import EventsClient, NullEventsClient, PrefectCloudEventsClient
+from .clients import (
+    EventsClient,
+    NullEventsClient,
+    PrefectCloudEventsClient,
+    PrefectEventsClient,
+)
 from .related import related_resources_from_run_context
 from .schemas.events import Event
 
 
+def should_emit_events() -> bool:
+    return emit_events_to_cloud() or should_emit_events_to_running_server()
+
+
 def emit_events_to_cloud() -> bool:
-    api = PREFECT_API_URL.value()
-    return api and api.startswith(PREFECT_CLOUD_API_URL.value())
+    api_url = PREFECT_API_URL.value()
+    return isinstance(api_url, str) and api_url.startswith(
+        PREFECT_CLOUD_API_URL.value()
+    )
+
+
+def should_emit_events_to_running_server() -> bool:
+    api_url = PREFECT_API_URL.value()
+    return isinstance(api_url, str) and PREFECT_EXPERIMENTAL_EVENTS
 
 
 class EventsWorker(QueueService[Event]):
@@ -62,6 +83,8 @@ def instance(
                     "api_url": PREFECT_API_URL.value(),
                     "api_key": PREFECT_API_KEY.value(),
                 }
+            elif should_emit_events_to_running_server():
+                client_type = PrefectEventsClient
 
             else:
                 client_type = NullEventsClient
diff --git a/src/prefect/testing/fixtures.py b/src/prefect/testing/fixtures.py
index 4b610e57c9ec..a5e9c737aa65 100644
--- a/src/prefect/testing/fixtures.py
+++ b/src/prefect/testing/fixtures.py
@@ -316,14 +316,19 @@ async def outgoing_events(socket: WebSocketServerProtocol):
 
 @pytest.fixture
 def events_api_url(events_server: WebSocketServer, unused_tcp_port: int) -> str:
+    return f"http://localhost:{unused_tcp_port}"
+
+
+@pytest.fixture
+def events_cloud_api_url(events_server: WebSocketServer, unused_tcp_port: int) -> str:
     return f"http://localhost:{unused_tcp_port}/accounts/A/workspaces/W"
 
 
 @pytest.fixture
-def mock_emit_events_to_cloud(monkeypatch) -> mock.Mock:
+def mock_should_emit_events(monkeypatch) -> mock.Mock:
     m = mock.Mock()
     m.return_value = True
-    monkeypatch.setattr("prefect.events.utilities.emit_events_to_cloud", m)
+    monkeypatch.setattr("prefect.events.utilities.should_emit_events", m)
     return m
 
 
diff --git a/tests/concurrency/test_concurrency_asyncio.py b/tests/concurrency/test_concurrency_asyncio.py
index 2f6f6abeb3f1..94d56d1082ec 100644
--- a/tests/concurrency/test_concurrency_asyncio.py
+++ b/tests/concurrency/test_concurrency_asyncio.py
@@ -94,7 +94,7 @@ async def test_concurrency_emits_events(
     concurrency_limit: ConcurrencyLimitV2,
     other_concurrency_limit: ConcurrencyLimitV2,
     asserting_events_worker: EventsWorker,
-    mock_emit_events_to_cloud,
+    mock_should_emit_events,
     reset_worker_events,
 ):
     executed = False
@@ -248,7 +248,7 @@ async def test_rate_limit_emits_events(
     concurrency_limit_with_decay: ConcurrencyLimitV2,
     other_concurrency_limit_with_decay: ConcurrencyLimitV2,
     asserting_events_worker: EventsWorker,
-    mock_emit_events_to_cloud,
+    mock_should_emit_events,
     reset_worker_events,
 ):
     async def resource_heavy():
diff --git a/tests/concurrency/test_concurrency_sync.py b/tests/concurrency/test_concurrency_sync.py
index 9816a383d925..781ab15a4a85 100644
--- a/tests/concurrency/test_concurrency_sync.py
+++ b/tests/concurrency/test_concurrency_sync.py
@@ -49,7 +49,7 @@ def test_concurrency_emits_events(
     concurrency_limit: ConcurrencyLimitV2,
     other_concurrency_limit: ConcurrencyLimitV2,
     asserting_events_worker: EventsWorker,
-    mock_emit_events_to_cloud,
+    mock_should_emit_events,
     reset_worker_events,
 ):
     def resource_heavy():
@@ -241,7 +241,7 @@ def test_rate_limit_emits_events(
     concurrency_limit_with_decay: ConcurrencyLimitV2,
     other_concurrency_limit_with_decay: ConcurrencyLimitV2,
     asserting_events_worker: EventsWorker,
-    mock_emit_events_to_cloud,
+    mock_should_emit_events,
     reset_worker_events,
 ):
     def resource_heavy():
diff --git a/tests/events/client/conftest.py b/tests/events/client/conftest.py
index d4db946a6fea..fd8e23298d84 100644
--- a/tests/events/client/conftest.py
+++ b/tests/events/client/conftest.py
@@ -10,8 +10,8 @@ def reset_asserting_events_client():
 
 
 @pytest.fixture(autouse=True)
-def mock_emit_events_to_cloud(mock_emit_events_to_cloud):
-    yield mock_emit_events_to_cloud
+def mock_should_emit_events(mock_should_emit_events):
+    yield mock_should_emit_events
 
 
 @pytest.fixture
diff --git a/tests/events/client/test_cloud_events_client.py b/tests/events/client/test_cloud_events_client.py
index 3dcef568c58e..562b7ab4546e 100644
--- a/tests/events/client/test_cloud_events_client.py
+++ b/tests/events/client/test_cloud_events_client.py
@@ -4,14 +4,25 @@
 from websockets.exceptions import ConnectionClosed
 
 from prefect.events import Event
-from prefect.events.clients import PrefectCloudEventsClient
+from prefect.events.clients import PrefectCloudEventsClient, PrefectEventsClient
 from prefect.testing.fixtures import Puppeteer, Recorder
 
 
-async def test_cloud_client_can_connect_and_emit(
+async def test_events_client_can_connect_and_emit(
     events_api_url: str, example_event_1: Event, recorder: Recorder
 ):
-    async with PrefectCloudEventsClient(events_api_url, "my-token") as client:
+    async with PrefectEventsClient(events_api_url) as client:
+        await client.emit(example_event_1)
+
+    assert recorder.connections == 1
+    assert recorder.path == "/events/in"
+    assert recorder.events == [example_event_1]
+
+
+async def test_cloud_client_can_connect_and_emit(
+    events_cloud_api_url: str, example_event_1: Event, recorder: Recorder
+):
+    async with PrefectCloudEventsClient(events_cloud_api_url, "my-token") as client:
         await client.emit(example_event_1)
 
     assert recorder.connections == 1
@@ -20,7 +31,7 @@ async def test_cloud_client_can_connect_and_emit(
 
 
 async def test_reconnects_and_resends_after_hard_disconnect(
-    events_api_url: str,
+    events_cloud_api_url: str,
     example_event_1: Event,
     example_event_2: Event,
     example_event_3: Event,
@@ -29,7 +40,7 @@ async def test_reconnects_and_resends_after_hard_disconnect(
     recorder: Recorder,
     puppeteer: Puppeteer,
 ):
-    client = PrefectCloudEventsClient(events_api_url, "my-token", checkpoint_every=1)
+    client = PrefectEventsClient(events_cloud_api_url, checkpoint_every=1)
     async with client:
         assert recorder.connections == 1
 
@@ -55,7 +66,7 @@ async def test_reconnects_and_resends_after_hard_disconnect(
 
 @pytest.mark.parametrize("attempts", [4, 1, 0])
 async def test_gives_up_after_a_certain_amount_of_tries(
-    events_api_url: str,
+    events_cloud_api_url: str,
     example_event_1: Event,
     example_event_2: Event,
     example_event_3: Event,
@@ -63,8 +74,10 @@ async def test_gives_up_after_a_certain_amount_of_tries(
     puppeteer: Puppeteer,
     attempts: int,
 ):
-    client = PrefectCloudEventsClient(
-        events_api_url, "my-token", checkpoint_every=1, reconnection_attempts=attempts
+    client = PrefectEventsClient(
+        events_cloud_api_url,
+        checkpoint_every=1,
+        reconnection_attempts=attempts,
     )
     async with client:
         assert recorder.connections == 1
@@ -86,7 +99,7 @@ async def test_gives_up_after_a_certain_amount_of_tries(
 
 
 async def test_giving_up_after_negative_one_tries_is_a_noop(
-    events_api_url: str,
+    events_cloud_api_url: str,
     example_event_1: Event,
     example_event_2: Event,
     example_event_3: Event,
@@ -95,8 +108,8 @@ async def test_giving_up_after_negative_one_tries_is_a_noop(
 ):
     """This is a nonsensical configuration, but covers a branch of client.emit where
     the primary reconnection loop does nothing (not even sending events)"""
-    client = PrefectCloudEventsClient(
-        events_api_url, "my-token", checkpoint_every=1, reconnection_attempts=-1
+    client = PrefectEventsClient(
+        events_cloud_api_url, checkpoint_every=1, reconnection_attempts=-1
     )
     async with client:
         assert recorder.connections == 1
@@ -114,15 +127,15 @@ async def test_giving_up_after_negative_one_tries_is_a_noop(
 
 
 async def test_handles_api_url_with_trailing_slash(
-    events_api_url: str, example_event_1: Event, recorder: Recorder
+    events_cloud_api_url: str, example_event_1: Event, recorder: Recorder
 ):
     # Regression test for https://github.com/PrefectHQ/prefect/issues/9662
     # where a configuration that has a trailing slash on the PREFECT_API_URL
     # would cause the client to fail to connect with a 403 as the path would
     # contain a double slash, ie: `wss:api.prefect.cloud/...//events/in`
 
-    events_api_url += "/"
-    async with PrefectCloudEventsClient(events_api_url, "my-token") as client:
+    events_cloud_api_url += "/"
+    async with PrefectEventsClient(events_cloud_api_url) as client:
         await client.emit(example_event_1)
 
     assert recorder.connections == 1
@@ -131,7 +144,7 @@ async def test_handles_api_url_with_trailing_slash(
 
 
 async def test_recovers_from_temporary_error_reconnecting(
-    events_api_url: str,
+    events_cloud_api_url: str,
     example_event_1: Event,
     example_event_2: Event,
     example_event_3: Event,
@@ -139,9 +152,9 @@ async def test_recovers_from_temporary_error_reconnecting(
     puppeteer: Puppeteer,
 ):
     """Regression test for an error where the client encountered assertion errors in
-    PrefectCloudEventsClient._emit because the websocket was None"""
-    client = PrefectCloudEventsClient(
-        events_api_url, "my-token", checkpoint_every=1, reconnection_attempts=3
+    PrefectEventsClient._emit because the websocket was None"""
+    client = PrefectEventsClient(
+        events_cloud_api_url, checkpoint_every=1, reconnection_attempts=3
     )
     async with client:
         assert recorder.connections == 1
@@ -179,7 +192,7 @@ async def test_recovers_from_temporary_error_reconnecting(
 
 
 async def test_recovers_from_long_lasting_error_reconnecting(
-    events_api_url: str,
+    events_cloud_api_url: str,
     example_event_1: Event,
     example_event_2: Event,
     example_event_3: Event,
@@ -187,9 +200,9 @@ async def test_recovers_from_long_lasting_error_reconnecting(
     puppeteer: Puppeteer,
 ):
     """Regression test for an error where the client encountered assertion errors in
-    PrefectCloudEventsClient._emit because the websocket was None"""
-    client = PrefectCloudEventsClient(
-        events_api_url, "my-token", checkpoint_every=1, reconnection_attempts=3
+    PrefectEventsClient._emit because the websocket was None"""
+    client = PrefectEventsClient(
+        events_cloud_api_url, checkpoint_every=1, reconnection_attempts=3
     )
     async with client:
         assert recorder.connections == 1
diff --git a/tests/events/client/test_cloud_events_subscriber.py b/tests/events/client/test_cloud_events_subscriber.py
index f6d658969d2f..ba5405666e13 100644
--- a/tests/events/client/test_cloud_events_subscriber.py
+++ b/tests/events/client/test_cloud_events_subscriber.py
@@ -9,14 +9,14 @@
 
 
 async def test_subscriber_can_connect_with_defaults(
-    events_api_url: str,
+    events_cloud_api_url: str,
     example_event_1: Event,
     example_event_2: Event,
     recorder: Recorder,
     puppeteer: Puppeteer,
 ):
     with temporary_settings(
-        updates={PREFECT_API_KEY: "my-token", PREFECT_API_URL: events_api_url}
+        updates={PREFECT_API_KEY: "my-token", PREFECT_API_URL: events_cloud_api_url}
     ):
         puppeteer.token = "my-token"
         puppeteer.outgoing_events = [example_event_1, example_event_2]
@@ -34,7 +34,7 @@ async def test_subscriber_can_connect_with_defaults(
 
 
 async def test_subscriber_complains_without_api_url_and_key(
-    events_api_url: str,
+    events_cloud_api_url: str,
     example_event_1: Event,
     example_event_2: Event,
     recorder: Recorder,
@@ -46,7 +46,7 @@ async def test_subscriber_complains_without_api_url_and_key(
 
 
 async def test_subscriber_can_connect_and_receive_one_event(
-    events_api_url: str,
+    events_cloud_api_url: str,
     example_event_1: Event,
     example_event_2: Event,
     recorder: Recorder,
@@ -58,7 +58,7 @@ async def test_subscriber_can_connect_and_receive_one_event(
     filter = EventFilter(event=EventNameFilter(name=["example.event"]))
 
     async with PrefectCloudEventSubscriber(
-        events_api_url,
+        events_cloud_api_url,
         "my-token",
         filter,
         reconnection_attempts=0,
@@ -74,7 +74,7 @@ async def test_subscriber_can_connect_and_receive_one_event(
 
 
 async def test_subscriber_specifying_negative_reconnects_gets_error(
-    events_api_url: str,
+    events_cloud_api_url: str,
     example_event_1: Event,
     example_event_2: Event,
     recorder: Recorder,
@@ -87,7 +87,7 @@ async def test_subscriber_specifying_negative_reconnects_gets_error(
 
     with pytest.raises(ValueError, match="non-negative"):
         PrefectCloudEventSubscriber(
-            events_api_url,
+            events_cloud_api_url,
             "my-token",
             filter,
             reconnection_attempts=-1,
@@ -97,7 +97,7 @@ async def test_subscriber_specifying_negative_reconnects_gets_error(
 
 
 async def test_subscriber_raises_on_invalid_auth_with_soft_denial(
-    events_api_url: str,
+    events_cloud_api_url: str,
     example_event_1: Event,
     example_event_2: Event,
     recorder: Recorder,
@@ -110,7 +110,7 @@ async def test_subscriber_raises_on_invalid_auth_with_soft_denial(
 
     with pytest.raises(Exception, match="Unable to authenticate"):
         subscriber = PrefectCloudEventSubscriber(
-            events_api_url,
+            events_cloud_api_url,
             "bogus",
             filter,
             reconnection_attempts=0,
@@ -124,7 +124,7 @@ async def test_subscriber_raises_on_invalid_auth_with_soft_denial(
 
 
 async def test_subscriber_raises_on_invalid_auth_with_hard_denial(
-    events_api_url: str,
+    events_cloud_api_url: str,
     example_event_1: Event,
     example_event_2: Event,
     recorder: Recorder,
@@ -138,7 +138,7 @@ async def test_subscriber_raises_on_invalid_auth_with_hard_denial(
 
     with pytest.raises(Exception, match="Unable to authenticate"):
         subscriber = PrefectCloudEventSubscriber(
-            events_api_url,
+            events_cloud_api_url,
             "bogus",
             filter,
             reconnection_attempts=0,
@@ -152,7 +152,7 @@ async def test_subscriber_raises_on_invalid_auth_with_hard_denial(
 
 
 async def test_subscriber_reconnects_on_hard_disconnects(
-    events_api_url: str,
+    events_cloud_api_url: str,
     example_event_1: Event,
     example_event_2: Event,
     recorder: Recorder,
@@ -165,7 +165,7 @@ async def test_subscriber_reconnects_on_hard_disconnects(
     filter = EventFilter(event=EventNameFilter(name=["example.event"]))
 
     async with PrefectCloudEventSubscriber(
-        events_api_url,
+        events_cloud_api_url,
         "my-token",
         filter,
         reconnection_attempts=2,
@@ -178,7 +178,7 @@ async def test_subscriber_reconnects_on_hard_disconnects(
 
 
 async def test_subscriber_gives_up_after_so_many_attempts(
-    events_api_url: str,
+    events_cloud_api_url: str,
     example_event_1: Event,
     example_event_2: Event,
     recorder: Recorder,
@@ -192,7 +192,7 @@ async def test_subscriber_gives_up_after_so_many_attempts(
 
     with pytest.raises(ConnectionClosedError):
         async with PrefectCloudEventSubscriber(
-            events_api_url,
+            events_cloud_api_url,
             "my-token",
             filter,
             reconnection_attempts=4,
@@ -205,7 +205,7 @@ async def test_subscriber_gives_up_after_so_many_attempts(
 
 
 async def test_subscriber_skips_duplicate_events(
-    events_api_url: str,
+    events_cloud_api_url: str,
     example_event_1: Event,
     example_event_2: Event,
     recorder: Recorder,
@@ -217,7 +217,7 @@ async def test_subscriber_skips_duplicate_events(
     filter = EventFilter(event=EventNameFilter(name=["example.event"]))
 
     async with PrefectCloudEventSubscriber(
-        events_api_url,
+        events_cloud_api_url,
         "my-token",
         filter,
     ) as subscriber:
diff --git a/tests/events/client/test_events_emit_event.py b/tests/events/client/test_events_emit_event.py
index fd8f436de914..8eea669a63d1 100644
--- a/tests/events/client/test_events_emit_event.py
+++ b/tests/events/client/test_events_emit_event.py
@@ -123,9 +123,9 @@ def test_noop_with_null_events_client():
         )
 
 
-def test_noop_with_non_cloud_client(mock_emit_events_to_cloud: mock.Mock):
+def test_noop_with_non_cloud_client(mock_should_emit_events: mock.Mock):
     with temporary_settings(updates={PREFECT_API_URL: "http://localhost:4242"}):
-        mock_emit_events_to_cloud.return_value = None
+        mock_should_emit_events.return_value = None
 
         assert (
             emit_event(
diff --git a/tests/events/client/test_events_worker.py b/tests/events/client/test_events_worker.py
index 3b6fb38aeda4..39b3c26db3d2 100644
--- a/tests/events/client/test_events_worker.py
+++ b/tests/events/client/test_events_worker.py
@@ -7,10 +7,12 @@
 from prefect.events.clients import (
     AssertingEventsClient,
     NullEventsClient,
+    PrefectEventsClient,
 )
 from prefect.events.worker import EventsWorker
 from prefect.settings import (
     PREFECT_API_URL,
+    PREFECT_EXPERIMENTAL_EVENTS,
     temporary_settings,
 )
 
@@ -44,6 +46,17 @@ def test_worker_instance_null_client_non_cloud_api_url():
         assert worker.client_type == NullEventsClient
 
 
+def test_worker_instance_null_client_non_cloud_api_url_events_enabled():
+    with temporary_settings(
+        updates={
+            PREFECT_API_URL: "http://localhost:8080/api",
+            PREFECT_EXPERIMENTAL_EVENTS: True,
+        }
+    ):
+        worker = EventsWorker.instance()
+        assert worker.client_type == PrefectEventsClient
+
+
 async def test_includes_related_resources_from_run_context(
     asserting_events_worker: EventsWorker, reset_worker_events, prefect_client
 ):


Code review comments:
diff_hunk: @@ -238,6 +239,46 @@ async def _emit(self, event: Event) -> None:
                     await asyncio.sleep(1)
 
 
+class PrefectCloudEventsClient(PrefectEventsClient):
+    """A Prefect Events client that streams events to a Prefect Cloud Workspace"""
+
+    _websocket: Optional[WebSocketClientProtocol]
+    _unconfirmed_events: List[Event]
path: src/prefect/events/clients.py
text: @user1:
Probably don't need these since they're in the base class

diff_hunk: @@ -150,14 +152,13 @@ def __init__(
 
         self._connect = connect(
             socket_url + "/events/in",
-            extra_headers={"Authorization": f"bearer {api_key}"},
         )
         self._websocket = None
         self._reconnection_attempts = reconnection_attempts
         self._unconfirmed_events = []
         self._checkpoint_every = checkpoint_every
 
-    async def __aenter__(self) -> "PrefectCloudEventsClient":
+    async def __aenter__(self) -> "PrefectEventsClient":
path: src/prefect/events/clients.py
text: @user1:
How about `Self` here:

```suggestion
    async def __aenter__(self) -> Self:
```