Add gRPC user agent interceptor

src/hiero_sdk_python/client/client.py

@@ -19,6 +19,7 @@
 from hiero_sdk_python.hbar import Hbar
 from hiero_sdk_python.logger.logger import Logger, LogLevel
 from hiero_sdk_python.transaction.transaction_id import TransactionId
+from hiero_sdk_python.user_agent_interceptor import _apply_user_agent_interceptor
 
 from .network import Network
 
@@ -163,6 +164,7 @@ 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 = _apply_user_agent_interceptor(self.mirror_channel)
         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

@@ -11,6 +11,7 @@
 from hiero_sdk_python.address_book.node_address import NodeAddress
 from hiero_sdk_python.channels import _Channel
 from hiero_sdk_python.managed_node_address import _ManagedNodeAddress
+from hiero_sdk_python.user_agent_interceptor import _apply_user_agent_interceptor
 
 
 # Timeout for fetching server certificates during TLS validation
@@ -147,6 +148,7 @@ def _get_channel(self):
         else:
             channel = grpc.insecure_channel(str(self._address))
 
+        channel = _apply_user_agent_interceptor(channel)
         self._channel = _Channel(channel)
 
         return self._channel

src/hiero_sdk_python/user_agent_interceptor.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+from collections import namedtuple
+from importlib import metadata as importlib_metadata
+
+import grpc
+
+
+_SDK_PACKAGE_NAME = "hiero-sdk-python"
+_USER_AGENT_HEADER = "x-user-agent"
+
+
+def _get_sdk_version() -> str:
+    """Return the installed SDK version, or dev for local checkouts."""
+    try:
+        return importlib_metadata.version(_SDK_PACKAGE_NAME)
+    except importlib_metadata.PackageNotFoundError:
+        return "dev"
+
+
+class _ClientCallDetails(
+    namedtuple(
+        "_ClientCallDetails",
+        ("method", "timeout", "metadata", "credentials", "wait_for_ready", "compression"),
+    ),
+    grpc.ClientCallDetails,
+):
+    """Concrete call details used to attach metadata in client interceptors."""
+
+
+class _UserAgentInterceptor(grpc.UnaryUnaryClientInterceptor, grpc.UnaryStreamClientInterceptor):
+    """gRPC client interceptor that identifies this SDK to Hiero nodes."""
+
+    def __init__(self) -> None:
+        self._user_agent = f"{_SDK_PACKAGE_NAME}/{_get_sdk_version()}"
+
+    def intercept_unary_unary(self, continuation, client_call_details, request):
+        return continuation(self._with_user_agent(client_call_details), request)
+
+    def intercept_unary_stream(self, continuation, client_call_details, request):
+        return continuation(self._with_user_agent(client_call_details), request)
+
+    def _with_user_agent(self, client_call_details):
+        metadata = list(client_call_details.metadata or ())
+        metadata.append((_USER_AGENT_HEADER, self._user_agent))
+
+        return _ClientCallDetails(
+            client_call_details.method,
+            client_call_details.timeout,
+            metadata,
+            client_call_details.credentials,
+            client_call_details.wait_for_ready,
+            client_call_details.compression,
+        )
+
+
+_USER_AGENT_INTERCEPTOR = _UserAgentInterceptor()
+
+
+def _apply_user_agent_interceptor(channel: grpc.Channel) -> grpc.Channel:
+    """Wrap a channel so every outgoing call includes the SDK user-agent header."""
+    return grpc.intercept_channel(channel, _USER_AGENT_INTERCEPTOR)

tests/unit/user_agent_interceptor_test.py

@@ -0,0 +1,126 @@
+from __future__ import annotations
+
+from concurrent import futures
+from importlib import metadata as importlib_metadata
+from unittest.mock import Mock, patch
+
+import grpc
+import pytest
+
+from hiero_sdk_python.user_agent_interceptor import (
+    _USER_AGENT_HEADER,
+    _apply_user_agent_interceptor,
+    _ClientCallDetails,
+    _get_sdk_version,
+    _UserAgentInterceptor,
+)
+
+
+pytestmark = pytest.mark.unit
+
+
+def _call_details(metadata=None):
+    return _ClientCallDetails(
+        method="/proto.Service/Method",
+        timeout=30,
+        metadata=metadata,
+        credentials=None,
+        wait_for_ready=None,
+        compression=None,
+    )
+
+
+def test_get_sdk_version_returns_installed_version():
+    assert _get_sdk_version()
+
+
+@patch("hiero_sdk_python.user_agent_interceptor.importlib_metadata.version")
+def test_get_sdk_version_falls_back_to_dev(mock_version):
+    mock_version.side_effect = importlib_metadata.PackageNotFoundError
+
+    assert _get_sdk_version() == "dev"
+
+
+@patch("hiero_sdk_python.user_agent_interceptor._get_sdk_version", return_value="1.2.3")
+def test_unary_unary_interceptor_adds_user_agent_header(mock_get_version):
+    interceptor = _UserAgentInterceptor()
+    continuation = Mock(return_value="response")
+
+    response = interceptor.intercept_unary_unary(continuation, _call_details(), request="request")
+
+    assert response == "response"
+    modified_call_details = continuation.call_args.args[0]
+    assert (_USER_AGENT_HEADER, "hiero-sdk-python/1.2.3") in modified_call_details.metadata
+
+
+@patch("hiero_sdk_python.user_agent_interceptor._get_sdk_version", return_value="1.2.3")
+def test_unary_stream_interceptor_adds_user_agent_header(mock_get_version):
+    interceptor = _UserAgentInterceptor()
+    continuation = Mock(return_value=iter(["response"]))
+
+    response = interceptor.intercept_unary_stream(continuation, _call_details(), request="request")
+
+    assert list(response) == ["response"]
+    modified_call_details = continuation.call_args.args[0]
+    assert (_USER_AGENT_HEADER, "hiero-sdk-python/1.2.3") in modified_call_details.metadata
+
+
+@patch("hiero_sdk_python.user_agent_interceptor._get_sdk_version", return_value="1.2.3")
+def test_interceptor_preserves_existing_metadata(mock_get_version):
+    interceptor = _UserAgentInterceptor()
+    continuation = Mock(return_value="response")
+    original_metadata = [("authorization", "token")]
+
+    interceptor.intercept_unary_unary(continuation, _call_details(original_metadata), request="request")
+
+    modified_metadata = continuation.call_args.args[0].metadata
+    assert modified_metadata == [
+        ("authorization", "token"),
+        (_USER_AGENT_HEADER, "hiero-sdk-python/1.2.3"),
+    ]
+
+
+@patch("grpc.intercept_channel")
+def test_apply_user_agent_interceptor_wraps_channel(mock_intercept_channel):
+    channel = Mock(spec=grpc.Channel)
+    intercepted_channel = Mock(spec=grpc.Channel)
+    mock_intercept_channel.return_value = intercepted_channel
+
+    assert _apply_user_agent_interceptor(channel) is intercepted_channel
+    mock_intercept_channel.assert_called_once()
+    assert mock_intercept_channel.call_args.args[0] is channel
+    assert isinstance(mock_intercept_channel.call_args.args[1], _UserAgentInterceptor)
+
+
+def test_interceptor_sends_valid_grpc_metadata():
+    received_metadata = []
+
+    def handle_request(request, context):
+        received_metadata.extend(context.invocation_metadata())
+        return b"response"
+
+    server = grpc.server(futures.ThreadPoolExecutor(max_workers=1))
+    method_handler = grpc.unary_unary_rpc_method_handler(
+        handle_request,
+        request_deserializer=lambda request: request,
+        response_serializer=lambda response: response,
+    )
+    server.add_generic_rpc_handlers((grpc.method_handlers_generic_handler("test.Service", {"Method": method_handler}),))
+    port = server.add_insecure_port("127.0.0.1:0")
+    server.start()
+
+    try:
+        channel = _apply_user_agent_interceptor(grpc.insecure_channel(f"127.0.0.1:{port}"))
+        method = channel.unary_unary(
+            "/test.Service/Method",
+            request_serializer=lambda request: request,
+            response_deserializer=lambda response: response,
+        )
+
+        assert method(b"request", timeout=5) == b"response"
+    finally:
+        server.stop(0)
+
+    user_agent_values = [value for key, value in received_metadata if key == _USER_AGENT_HEADER]
+    assert len(user_agent_values) == 1
+    assert user_agent_values[0].startswith("hiero-sdk-python/")