feat: add user agent header using gRPC interceptor

src/hiero_sdk_python/channels.py

@@ -1,5 +1,10 @@
 from __future__ import annotations
 
+from collections import namedtuple
+from importlib.metadata import PackageNotFoundError, version
+
+import grpc
+
 from hiero_sdk_python.hapi.services import (
     address_book_service_pb2_grpc,
     consensus_service_pb2_grpc,
@@ -14,6 +19,84 @@
 )
 
 
+class _UserAgentInterceptor(grpc.UnaryUnaryClientInterceptor, grpc.UnaryStreamClientInterceptor):
+    """
+    gRPC interceptor that appends an x-user-agent header to all outgoing requests.
+    """
+
+    _HEADER_KEY = "x-user-agent"
+    _SDK_NAME = "hiero-sdk-python"
+    _CallDetails = namedtuple(
+        "_CallDetails",
+        ("method", "timeout", "metadata", "credentials", "wait_for_ready", "compression"),
+    )
+
+    def __init__(self) -> None:
+        """
+        Initialize the interceptor and compute the user agent value.
+        The user agent is computed once during initialization to avoid repeated package metadata lookups on every request.
+        """
+
+        try:
+            sdk_version = version(self._SDK_NAME)
+        except PackageNotFoundError:
+            sdk_version = "dev"
+        self._user_agent = f"{self._SDK_NAME}/{sdk_version}"
+
+    def _with_user_agent(self, details: grpc.ClientCallDetails) -> grpc.ClientCallDetails:
+        """
+        Append the user agent header to the call details.
+
+        Args:
+            details: The original gRPC call details.
+
+        Returns:
+            A new ClientCallDetails object with the x-user-agent header included in the metadata.
+        """
+        metadata = [] if details.metadata is None else list(details.metadata)
+        metadata = [entry for entry in metadata if entry[0] != self._HEADER_KEY]
+        metadata.append((self._HEADER_KEY, self._user_agent))
+
+        return self._CallDetails(
+            details.method,
+            details.timeout,
+            metadata,
+            getattr(details, "credentials", None),
+            getattr(details, "wait_for_ready", None),
+            getattr(details, "compression", None),
+        )
+
+    def intercept_unary_unary(self, continuation, client_call_details, request):
+        """
+        Intercept unary-unary calls and append the user agent header.
+
+        Args:
+            continuation: The gRPC continuation function to call the next interceptor or actual RPC.
+            client_call_details: The details of the gRPC call, including method, timeout, metadata, etc.
+            request: The request object being sent.
+
+        Returns:
+            The result of the gRPC call after appending the user agent header.
+        """
+
+        return continuation(self._with_user_agent(client_call_details), request)
+
+    def intercept_unary_stream(self, continuation, client_call_details, request):
+        """
+        Intercept unary-stream calls and append the user agent header.
+
+        Args:
+            continuation: The gRPC continuation function to call the next interceptor or actual RPC.
+            client_call_details: The details of the gRPC call, including method, timeout, metadata, etc.
+            request: The request object being sent.
+
+        Returns:
+            The result of the gRPC call after appending the user agent header.
+        """
+
+        return continuation(self._with_user_agent(client_call_details), request)
+
+
 class _Channel:
     """
     The _Channel class is a wrapper around gRPC channels that provides access to various

src/hiero_sdk_python/client/client.py

@@ -12,6 +12,7 @@
 from dotenv import load_dotenv
 
 from hiero_sdk_python.account.account_id import AccountId
+from hiero_sdk_python.channels import _UserAgentInterceptor
 from hiero_sdk_python.crypto.private_key import PrivateKey
 from hiero_sdk_python.hapi.mirror import (
     consensus_service_pb2_grpc as mirror_consensus_grpc,
@@ -163,6 +164,8 @@ def _init_mirror_stub(self) -> None:
             self.mirror_channel = grpc.secure_channel(mirror_address, grpc.ssl_channel_credentials())
         else:
             self.mirror_channel = grpc.insecure_channel(mirror_address)
+
+        self.mirror_channel = grpc.intercept_channel(self.mirror_channel, _UserAgentInterceptor())
         self.mirror_stub = mirror_consensus_grpc.ConsensusServiceStub(self.mirror_channel)
 
     def set_operator(self, account_id: AccountId, private_key: PrivateKey) -> None:

src/hiero_sdk_python/node.py

@@ -9,7 +9,7 @@
 
 from hiero_sdk_python.account.account_id import AccountId
 from hiero_sdk_python.address_book.node_address import NodeAddress
-from hiero_sdk_python.channels import _Channel
+from hiero_sdk_python.channels import _Channel, _UserAgentInterceptor
 from hiero_sdk_python.managed_node_address import _ManagedNodeAddress
 
 
@@ -147,6 +147,8 @@ def _get_channel(self):
         else:
             channel = grpc.insecure_channel(str(self._address))
 
+        channel = grpc.intercept_channel(channel, _UserAgentInterceptor())
+
         self._channel = _Channel(channel)
 
         return self._channel

tests/unit/channel_interceptor_test.py

@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+import grpc
+import pytest
+
+from hiero_sdk_python import channels
+from hiero_sdk_python.channels import _UserAgentInterceptor
+
+
+pytestmark = pytest.mark.unit
+
+
+class _DummyCallDetails:
+    def __init__(
+        self,
+        metadata=None,
+        method: str = "/proto.Service/Method",
+        timeout: float | None = None,
+        credentials=None,
+        wait_for_ready: bool | None = None,
+        compression=None,
+    ):
+        self.method = method
+        self.timeout = timeout
+        self.metadata = metadata
+        self.credentials = credentials
+        self.wait_for_ready = wait_for_ready
+        self.compression = compression
+
+
+def test_user_agent_value_from_installed_version(monkeypatch):
+    """Interceptor should build x-user-agent from package version when available."""
+
+    monkeypatch.setattr(channels, "version", lambda _name: "0.0.0")
+
+    interceptor = _UserAgentInterceptor()
+    assert interceptor._user_agent == "hiero-sdk-python/0.0.0"
+
+
+def test_user_agent_value_falls_back_to_dev(monkeypatch):
+    """Interceptor should fall back to dev when package metadata is unavailable."""
+
+    def _raise_not_found(_name):
+        raise channels.PackageNotFoundError
+
+    monkeypatch.setattr(channels, "version", _raise_not_found)
+
+    interceptor = _UserAgentInterceptor()
+    assert interceptor._user_agent == "hiero-sdk-python/dev"
+
+
+def test_intercept_unary_unary_appends_user_agent_metadata(monkeypatch):
+    """Unary calls should forward metadata including x-user-agent."""
+
+    monkeypatch.setattr(channels, "version", lambda _name: "0.0.0")
+
+    interceptor = _UserAgentInterceptor()
+    request = object()
+    captured = {}
+
+    def continuation(client_call_details, req):
+        captured["details"] = client_call_details
+        captured["request"] = req
+        return "ok"
+
+    details = _DummyCallDetails(metadata=[("existing", "value")])
+    result = interceptor.intercept_unary_unary(continuation, details, request)
+
+    assert result == "ok"
+    assert captured["request"] is request
+    assert ("existing", "value") in captured["details"].metadata
+    expected_header = (interceptor._HEADER_KEY, f"{interceptor._SDK_NAME}/0.0.0")
+    assert expected_header in captured["details"].metadata
+
+
+def test_intercept_unary_stream_adds_metadata_when_none(monkeypatch):
+    """Unary-stream calls should work even when original metadata is None."""
+
+    monkeypatch.setattr(channels, "version", lambda _name: "0.0.0")
+
+    interceptor = _UserAgentInterceptor()
+    request = object()
+    captured = {}
+
+    def continuation(client_call_details, req):
+        captured["details"] = client_call_details
+        captured["request"] = req
+        return "stream"
+
+    details = _DummyCallDetails(metadata=None)
+    result = interceptor.intercept_unary_stream(continuation, details, request)
+
+    assert result == "stream"
+    assert captured["request"] is request
+    expected_header = (interceptor._HEADER_KEY, f"{interceptor._SDK_NAME}/0.0.0")
+    assert captured["details"].metadata == [expected_header]
+
+
+def test_interceptor_implements_required_grpc_interfaces():
+    """Guard against accidental loss of gRPC interceptor interface inheritance."""
+    interceptor = _UserAgentInterceptor()
+    assert isinstance(interceptor, grpc.UnaryUnaryClientInterceptor)
+    assert isinstance(interceptor, grpc.UnaryStreamClientInterceptor)