Repository: PrefectHQ/prefect
Issue category: Feature Request

Start sending a User-Agent header on all API calls with the current `prefect` version
### First check

- [X] I added a descriptive title to this issue.
- [X] I used the GitHub search to find a similar request and didn't find it.
- [X] I searched the Prefect documentation for this feature.

### Prefect Version

2.x

### Describe the current behavior

Today we send a generic `User-Agent` header of `python-httpx/0.27.0`

### Describe the proposed behavior

We can send a `User-Agent` header capturing the `prefect` client version, like `prefect/2.16.1`.  This is a more accurate/precise approach and will give us more visibility into the distribution of client versions.

### Example Use

_No response_

### Additional context

_No response_


Functions to edit:
src/prefect/client/base.py:PrefectHttpxClient.__init__

Patch:
diff --git a/src/prefect/client/base.py b/src/prefect/client/base.py
index b3c883d8db17..27a75dc9608d 100644
--- a/src/prefect/client/base.py
+++ b/src/prefect/client/base.py
@@ -27,6 +27,8 @@
 from prefect._vendor.starlette import status
 from typing_extensions import Self
 
+import prefect
+from prefect.client import constants
 from prefect.client.schemas.objects import CsrfToken
 from prefect.exceptions import PrefectHTTPStatusError
 from prefect.logging import get_logger
@@ -199,6 +201,11 @@ def __init__(self, *args, enable_csrf_support: bool = False, **kwargs):
 
         super().__init__(*args, **kwargs)
 
+        user_agent = (
+            f"prefect/{prefect.__version__} (API {constants.SERVER_API_VERSION})"
+        )
+        self.headers["User-Agent"] = user_agent
+
     async def _send_with_retry(
         self,
         request: Request,

Test patch: diff --git a/tests/client/test_base_client.py b/tests/client/test_base_client.py
index 2cc1d9064a31..292d5130ed9f 100644
--- a/tests/client/test_base_client.py
+++ b/tests/client/test_base_client.py
@@ -1,6 +1,6 @@
 from contextlib import asynccontextmanager
 from datetime import datetime, timedelta, timezone
-from typing import AsyncGenerator, List, Tuple
+from typing import Any, AsyncGenerator, Dict, List, Tuple
 from unittest import mock
 
 import httpx
@@ -8,6 +8,9 @@
 from httpx import AsyncClient, Request, Response
 from prefect._vendor.starlette import status
 
+import prefect
+import prefect.client
+import prefect.client.constants
 from prefect.client.base import PrefectHttpxClient, PrefectResponse
 from prefect.client.schemas.objects import CsrfToken
 from prefect.exceptions import PrefectHTTPStatusError
@@ -507,7 +510,6 @@ async def test_prefect_httpx_client_returns_prefect_response(self, monkeypatch):
             )
         assert isinstance(response, PrefectResponse)
 
-    #
     async def test_prefect_httpx_client_raises_prefect_http_status_error(
         self, monkeypatch
     ):
@@ -534,10 +536,11 @@ async def test_prefect_httpx_client_raises_prefect_http_status_error(
 @asynccontextmanager
 async def mocked_client(
     responses: List[Response],
+    **client_kwargs: Dict[str, Any],
 ) -> AsyncGenerator[Tuple[PrefectHttpxClient, mock.AsyncMock], None]:
     with mock.patch("httpx.AsyncClient.send", autospec=True) as send:
         send.side_effect = responses
-        client = PrefectHttpxClient(enable_csrf_support=True)
+        client = PrefectHttpxClient(**client_kwargs)
         async with client:
             try:
                 yield client, send
@@ -545,9 +548,17 @@ async def mocked_client(
                 pass
 
 
+@asynccontextmanager
+async def mocked_csrf_client(
+    responses: List[Response],
+) -> AsyncGenerator[Tuple[PrefectHttpxClient, mock.AsyncMock], None]:
+    async with mocked_client(responses, enable_csrf_support=True) as (client, send):
+        yield client, send
+
+
 class TestCsrfSupport:
     async def test_no_csrf_headers_not_change_request(self):
-        async with mocked_client(responses=[RESPONSE_200]) as (client, send):
+        async with mocked_csrf_client(responses=[RESPONSE_200]) as (client, send):
             await client.get(url="fake.url/fake/route")
 
         request = send.call_args[0][1]
@@ -558,7 +569,7 @@ async def test_no_csrf_headers_not_change_request(self):
 
     @pytest.mark.parametrize("method", ["post", "put", "patch", "delete"])
     async def test_csrf_headers_on_change_request(self, method: str):
-        async with mocked_client(responses=[RESPONSE_CSRF, RESPONSE_200]) as (
+        async with mocked_csrf_client(responses=[RESPONSE_CSRF, RESPONSE_200]) as (
             client,
             send,
         ):
@@ -583,7 +594,7 @@ async def test_csrf_headers_on_change_request(self, method: str):
         assert request.headers["Prefect-Csrf-Client"] == str(client.csrf_client_id)
 
     async def test_refreshes_token_on_csrf_403(self):
-        async with mocked_client(
+        async with mocked_csrf_client(
             responses=[
                 RESPONSE_CSRF,
                 RESPONSE_INVALID_TOKEN,
@@ -629,7 +640,7 @@ async def test_refreshes_token_on_csrf_403(self):
         assert request.headers["Prefect-Csrf-Client"] == str(client.csrf_client_id)
 
     async def test_does_not_refresh_csrf_token_not_expired(self):
-        async with mocked_client(responses=[RESPONSE_200]) as (
+        async with mocked_csrf_client(responses=[RESPONSE_200]) as (
             client,
             send,
         ):
@@ -646,7 +657,7 @@ async def test_does_not_refresh_csrf_token_not_expired(self):
         assert request.headers["Prefect-Csrf-Client"] == str(client.csrf_client_id)
 
     async def test_does_refresh_csrf_token_when_expired(self):
-        async with mocked_client(responses=[RESPONSE_CSRF, RESPONSE_200]) as (
+        async with mocked_csrf_client(responses=[RESPONSE_CSRF, RESPONSE_200]) as (
             client,
             send,
         ):
@@ -672,7 +683,7 @@ async def test_does_refresh_csrf_token_when_expired(self):
         assert request.headers["Prefect-Csrf-Client"] == str(client.csrf_client_id)
 
     async def test_raises_exception_bad_csrf_token_response(self):
-        async with mocked_client(responses=[RESPONSE_400]) as (
+        async with mocked_csrf_client(responses=[RESPONSE_400]) as (
             client,
             _,
         ):
@@ -680,7 +691,7 @@ async def test_raises_exception_bad_csrf_token_response(self):
                 await client.post(url="fake.url/fake/route")
 
     async def test_disables_csrf_support_404_token_endpoint(self):
-        async with mocked_client(responses=[RESPONSE_404, RESPONSE_200]) as (
+        async with mocked_csrf_client(responses=[RESPONSE_404, RESPONSE_200]) as (
             client,
             send,
         ):
@@ -689,10 +700,39 @@ async def test_disables_csrf_support_404_token_endpoint(self):
             assert client.enable_csrf_support is False
 
     async def test_disables_csrf_support_422_csrf_disabled(self):
-        async with mocked_client(responses=[RESPONSE_CSRF_DISABLED, RESPONSE_200]) as (
+        async with mocked_csrf_client(
+            responses=[RESPONSE_CSRF_DISABLED, RESPONSE_200]
+        ) as (
             client,
             send,
         ):
             assert client.enable_csrf_support is True
             await client.post(url="fake.url/fake/route")
             assert client.enable_csrf_support is False
+
+
+class TestUserAgent:
+    @pytest.fixture
+    def prefect_version(self, monkeypatch: pytest.MonkeyPatch) -> str:
+        v = "42.43.44"
+        monkeypatch.setattr(prefect, "__version__", v)
+        return v
+
+    @pytest.fixture
+    def prefect_api_version(self, monkeypatch: pytest.MonkeyPatch) -> str:
+        v = "45.46.47"
+        monkeypatch.setattr(prefect.client.constants, "SERVER_API_VERSION", v)
+        return v
+
+    async def test_passes_informative_user_agent(
+        self,
+        prefect_version: str,
+        prefect_api_version: str,
+    ):
+        async with mocked_client(responses=[RESPONSE_200]) as (client, send):
+            await client.get(url="fake.url/fake/route")
+
+        request = send.call_args[0][1]
+        assert isinstance(request, httpx.Request)
+
+        assert request.headers["User-Agent"] == "prefect/42.43.44 (API 45.46.47)"