Implement gRPC `X-User-Agent` Header for SDK Identification

[rwalworth]

🧑‍🎓 Beginner Issue

Welcome! This is a Beginner Issue designed to help you learn the codebase.

🏁 When this issue is complete, you will have:

✅ Researched gRPC interceptors and metadata in Python
✅ Followed SDK patterns to implement a solution
✅ Researched and identified appropriate programming constructs to solve the problem
✅ Created basic tests
✅ Delivered a clean, review-ready pull request

---

🐞 Problem Description

Every gRPC call the Python SDK makes to a Hiero consensus node or mirror node is currently sent without any identifying information about the SDK. The consensus node has a built-in usage tracker (GrpcUsageTracker) that reads a X-User-Agent gRPC metadata header from incoming requests and uses it to log SDK type and version metrics. Without this header, the Python SDK is logged as UNKNOWN in the consensus node's telemetry.

The consensus node reads the header using this exact key:

Key.of("X-User-Agent", Metadata.ASCII_STRING_MARSHALLER)  // GrpcUsageTracker.java

And parses values in the format {sdk-name}/{version}, for example:

hiero-sdk-python/2.1.5

The supported SDK name for Python is hiero-sdk-python (see UserAgentType.java).

Current behavior: gRPC calls are made with no X-User-Agent header, so all Python SDK traffic appears as UNKNOWN in consensus node metrics.

Desired behavior: Every outgoing gRPC call from the Python SDK includes a X-User-Agent: hiero-sdk-python/{version} metadata header, enabling the consensus node to correctly attribute traffic.

Files to look at:

• src/hiero_sdk_python/node.py — where consensus node gRPC channels are created (grpc.secure_channel, grpc.insecure_channel)
• src/hiero_sdk_python/client/client.py — where mirror node gRPC channels are created
• src/hiero_sdk_python/channels.py — the _Channel wrapper around gRPC channels
• src/hiero_sdk_python/executable.py — where stub methods are actually called (_execute_method)

---

💡 Expected Solution

Implement a gRPC client interceptor that injects the X-User-Agent: hiero-sdk-python/{version} metadata header on every outgoing gRPC call. Apply this interceptor to all channels created in the SDK (both consensus node and mirror node channels).

The header value must exactly match the format expected by the consensus node:

hiero-sdk-python/2.1.5

where the version comes from the installed package version at runtime.

---

🔍 Research Pointers

gRPC Python interceptors (start here):

• gRPC Python Interceptors documentation — specifically UnaryUnaryClientInterceptor and UnaryStreamClientInterceptor
• grpc.intercept_channel() — wraps an existing channel with one or more interceptors (this is the cleanest approach)
• grpc-interceptor PyPI package — not used in this repo, but a useful reference for understanding interceptor patterns

How other Hiero SDKs implement this:

• hiero-sdk-js — uses gRPC metadata injected per-call
• hiero-sdk-go — uses a UnaryClientInterceptor that appends user-agent metadata; look for the interceptor implementation

How the consensus node processes this header:

• GrpcUsageTracker.java — the server interceptor that reads the header
• UserAgent.java — how the header value is parsed (format: sdk-name/semver, e.g. hiero-sdk-python/2.1.5; "dev" is also accepted as a version)
• UserAgentType.java — confirms hiero-sdk-python is a recognised SDK name

Getting the package version at runtime in Python:

• importlib.metadata — standard library module; use importlib.metadata.version("hiero-sdk-python") to get the installed version at runtime

---

🛠️ Implementation Notes

1. Research how Python gRPC interceptors work before writing any code — specifically how grpc.intercept_channel() wraps a channel and how ClientCallDetails is used to pass metadata. Look at the Go SDK interceptor as an analogous example.

2. Create an interceptor class that implements the required interceptor interfaces and injects the X-User-Agent header into the gRPC call metadata. Consider which call types this SDK uses (unary? server-streaming?) and implement the appropriate interfaces for each.

3. Look up the package version at runtime using importlib.metadata. Handle the case where the version is unavailable gracefully (e.g., in a development checkout where the package is not installed — fall back to "dev").

4. Apply the interceptor when channels are created. Find where grpc.secure_channel() and grpc.insecure_channel() are called in node.py and client/client.py, and wrap the resulting channel with grpc.intercept_channel().

5. Keep it minimal — the interceptor's only job is to append this one header. It should not modify anything else about the call.

---

🧠 Beginner Contributors — Prerequisites & Expectations

Caution

Beginner issues are low-risk but we expect a working solution.

Important

We recommend completing at least 3 good first issues before attempting a beginner issue.

Difficulty:

• Requires building knowledge of specific files in the gRPC channel setup
• Beginner to intermediate coding skills; familiarity with Python class inheritance

Tip

You should be comfortable with:

• Forking, branching, committing (with DCO + GPG signing), and opening a pull request without a tutorial
• Reading beginner to intermediate code you did not write and following its patterns
• Handling simple merge conflicts
• Keeping your fork up to date with main by rebasing
• Looking things up on your own before asking for help

If any of that feels unfamiliar, good first issues might still be the most rewarding path for you right now.

⚠️ AI Usage Policy

• AI can help you understand the codebase and the problem
• Using AI to generate code for beginner issues is strictly discouraged
• Using AI as the main source of research is strictly discouraged — refer to language and library documentation, protobuf definitions, and other SDKs

⏱️ Timeline & Workflow

• Typical time: ~1 week / ~8 hours
  🧩 Difficulty: Manageable but requires investigation and testing
  🎓 Best for: Beginner contributors

• 🔴 Completing this issue in 1–3 hours is a red flag

Tip

Suggested: share your proposed implementation approach as a comment before writing code to get early feedback and avoid wasted effort.

---

🧪 Testing Requirements

Important

At the beginner level, testing will be necessary but a small component.
The methods you implement should be verified (happy path at least).

Source code changes (in src/):

• Write unit tests that verify the interceptor injects the correct X-User-Agent header value
• Test that the version string is included correctly and falls back gracefully when unavailable
• Run tests locally: uv run pytest tests/unit/ -v
• Integration tests will run automatically when you push

---

🛡️ Quality & Review Standards

Beginner PRs must be working and follow best practices.

1. Working: The implementation must inject the correct header on all outgoing gRPC calls.
2. SDK Best Practices: The interceptor should follow the patterns used in node.py and channels.py — do not introduce a new style or abstraction.
3. Testing: Unit tests that verify the header is injected with the correct key and value format.

⚠️ Breaking changes

• This change should be transparent to SDK users — it adds metadata to gRPC calls but does not change any public API.

---

✅ PR Quality Checklist

Before opening your PR, confirm:

• I have spent the majority of my time researching the problem and building my understanding of the codebase and methods.
• My implementation works and is tested.
• Every line of code is personally understood and explainable.
• My changes address what the issue asked for — nothing more, nothing less.
• I have applied appropriate linting, code quality, and formatting tools used in this repo.

Before requesting a review, confirm:

• I did not modify files unrelated to this issue
• Clean git history — no rebase artifacts, merge commits, or unrelated files
• My commits are signed: git commit -S -s -m "chore: description" — Signing guide
• I added a CHANGELOG.md entry under [Unreleased] — Changelog guide
• I have included appropriate tests and all CI checks pass.

---

📋 Workflow Quick Reference

Step	Guide
Claim this issue	Comment /assign below
Sync with main	Rebasing guide
Open a PR and link this issue	PR guide
Resolve merge conflicts	Merge conflicts guide

---

📚 Resources & Support

🆘 Stuck?

Tip

Comment on this issue and describe what you have tried. A maintainer will respond.

Python SDK References:

• Windows Setup Guide
• Pylance guide
• Running examples
• Browse closed beginner PRs

References:

• gRPC Python Interceptors
• importlib.metadata — Python docs
• Community Calls
• Discord

[github-actions]

@coderabbitai plan

[coderabbitai]

Coding Plan

Summary

• Create a new user_agent_interceptor.py module containing a version utility and a dual-interface gRPC interceptor class
• The interceptor appends X-User-Agent: hiero-sdk-python/{version} metadata to all outgoing calls, supporting both unary and server-streaming RPCs
• Apply the interceptor via grpc.intercept_channel() at both channel creation points: _Node._get_channel() for consensus nodes and Client._init_mirror_stub() for mirror nodes
• Version is resolved at runtime via importlib.metadata with a "dev" fallback for development environments

Design Choices

Design Choice 1: Which gRPC interceptor interfaces need to be implemented?

Options Considered:

1. Implement only UnaryUnaryClientInterceptor (covers transactions/queries but not topic subscriptions)
2. Implement both UnaryUnaryClientInterceptor and UnaryStreamClientInterceptor (covers all SDK call types)

Chosen Option: 2

Rationale: Both interfaces must be implemented because the mirror node uses server-streaming for subscribeTopic, and all gRPC calls should include the user-agent header.

💡 User Tips

Regenerate the plan with different choices with @coderabbitai <feedback>.

Implementation Steps

Phase 1: Create User-Agent Interceptor Module

This phase creates the core interceptor functionality in a new module, including version detection and the gRPC interceptor class that injects the X-User-Agent header.

Task 1: Create Interceptor Module with Version Utility

Create a new module to house the user-agent interceptor and a helper function for runtime version detection.

• Create new file src/hiero_sdk_python/user_agent_interceptor.py following the flat module pattern used by channels.py and node.py
• Implement a _get_sdk_version() function that uses importlib.metadata.version("hiero-sdk-python") to retrieve the installed package version
• Handle PackageNotFoundError gracefully by returning "dev" as the fallback version (for development checkouts where the package isn't installed)
• Define the header key constant X-User-Agent and format the header value as hiero-sdk-python/{version}

Task 2: Implement gRPC Client Interceptor Class

Implement the interceptor class that injects the user-agent metadata on every outgoing call.

• Create a class (e.g., _UserAgentInterceptor) that implements both grpc.UnaryUnaryClientInterceptor and grpc.UnaryStreamClientInterceptor interfaces
• Implement intercept_unary_unary() to handle standard transaction and query calls used by executable.py
• Implement intercept_unary_stream() to handle server-streaming calls used by topic_message_query.py for topic subscriptions
• In both methods, create new ClientCallDetails with the X-User-Agent metadata appended to any existing metadata, then delegate to the continuation callable
• The interceptor should be stateless — the version string can be computed once at module load or on first use

🤖 Prompt for AI agents

Create a new file `src/hiero_sdk_python/user_agent_interceptor.py` following the
flat module pattern used by `channels.py` and `node.py`. This phase implements
the version detection utility and the dual-interface gRPC interceptor class.

- Implement a `_get_sdk_version()` function:
- Use `importlib.metadata.version("hiero-sdk-python")` to retrieve the installed
package version at runtime
- Catch `PackageNotFoundError` and return `"dev"` as the fallback version for
development checkouts
- Define a header key constant for `X-User-Agent` and format the header value as
`hiero-sdk-python/{version}`
- Implement a `_UserAgentInterceptor` class that inherits from both
`grpc.UnaryUnaryClientInterceptor` and `grpc.UnaryStreamClientInterceptor`:
- Implement `intercept_unary_unary()` to handle standard transaction and query
calls (used by `executable.py`)
- Implement `intercept_unary_stream()` to handle server-streaming calls (used by
`topic_message_query.py` for topic subscriptions)
- In both methods, construct new `ClientCallDetails` with the `X-User-Agent`
metadata appended to any existing metadata, then delegate to the `continuation`
callable
- The interceptor should be stateless; the version string may be computed once
at module load or on first use

Phase 2: Apply Interceptor to Channels and Add Tests

This phase integrates the interceptor into the existing channel creation points and adds unit tests to verify correct behavior.

Task 1: Apply Interceptor to Consensus Node Channels

Modify consensus node channel creation to wrap channels with the user-agent interceptor.

• Update src/hiero_sdk_python/node.py in the _Node._get_channel() method
• After creating the channel via grpc.secure_channel() or grpc.insecure_channel(), wrap it with grpc.intercept_channel(channel, interceptor) before passing it to _Channel()
• Import the interceptor from the new module; consider creating a single shared interceptor instance to avoid repeated instantiation

Task 2: Apply Interceptor to Mirror Node Channels

Modify mirror node channel creation to wrap channels with the same interceptor.

• Update src/hiero_sdk_python/client/client.py in the _init_mirror_stub() method
• After creating the channel via grpc.secure_channel() or grpc.insecure_channel(), wrap it with grpc.intercept_channel(channel, interceptor) before creating the stub
• Use the same interceptor class/instance as the consensus node channels for consistency

Task 3: Create Unit Tests for the Interceptor

Add unit tests to verify the interceptor correctly injects the header and handles version fallback.

• Create new test file tests/unit/user_agent_interceptor_test.py following existing test patterns (pytest.mark.unit, from __future__ import annotations)
• Test that _get_sdk_version() returns a version string (either real or "dev" fallback)
• Test that _get_sdk_version() returns "dev" when importlib.metadata.version() raises PackageNotFoundError (use @patch to simulate)
• Test that the interceptor's intercept_unary_unary() method adds the X-User-Agent header with the correct format to ClientCallDetails.metadata
• Test that the interceptor's intercept_unary_stream() method similarly adds the header
• Test that existing metadata (if any) is preserved when the header is appended
• Use mock continuations to verify the modified ClientCallDetails is passed through correctly

🤖 Prompt for AI agents

Integrate the `_UserAgentInterceptor` into both channel creation points in the
existing codebase, then create a unit test file to verify behavior. This phase
ensures the interceptor is applied consistently across all outgoing gRPC calls.

**`src/hiero_sdk_python/node.py`** — update `_Node._get_channel()`:
- Import the interceptor from `user_agent_interceptor.py`; consider creating a
single shared interceptor instance at module level to avoid repeated
instantiation
- After creating the channel via `grpc.secure_channel()` or
`grpc.insecure_channel()`, wrap it using `grpc.intercept_channel(channel,
interceptor)` before passing it to `_Channel()`

**`src/hiero_sdk_python/client/client.py`** — update `_init_mirror_stub()`:
- Import and use the same interceptor class/instance as the consensus node
channels for consistency
- After creating the channel via `grpc.secure_channel()` or
`grpc.insecure_channel()`, wrap it using `grpc.intercept_channel(channel,
interceptor)` before creating the stub

**`tests/unit/user_agent_interceptor_test.py`** — create new test file:
- Follow existing test patterns: `from __future__ import annotations`,
`pytest.mark.unit` markers, and `@patch` decorators for mocking
- Test that `_get_sdk_version()` returns a version string (real or `"dev"`
fallback)
- Test that `_get_sdk_version()` returns `"dev"` when
`importlib.metadata.version()` raises `PackageNotFoundError` (simulated via
`@patch`)
- Test that `intercept_unary_unary()` adds the `X-User-Agent` header with the
correct format to `ClientCallDetails.metadata`
- Test that `intercept_unary_stream()` similarly adds the header
- Test that any pre-existing metadata is preserved when the header is appended
- Use mock continuations to verify the modified `ClientCallDetails` is passed
through correctly

Research

The Python SDK is organized under src/hiero_sdk_python/ with gRPC channel management split across several files: node.py handles consensus node channel creation (with TLS support via _Node._get_channel()), client/client.py manages mirror node channels in _init_mirror_stub(), and channels.py provides a lazy stub wrapper (_Channel) around raw gRPC channels. The SDK uses unary gRPC calls for all transactions/queries (via executable.py) and server-streaming for topic subscriptions (via topic_message_query.py). No interceptors currently exist in the codebase. Tests follow pytest conventions with *_test.py naming, pytest.mark.unit markers, and @patch decorators for mocking gRPC functions.

---

🚀 Next Steps

🤖 All AI agent prompts combined

Task: 1

Create a new file `src/hiero_sdk_python/user_agent_interceptor.py` following the
flat module pattern used by `channels.py` and `node.py`. This phase implements
the version detection utility and the dual-interface gRPC interceptor class.

- Implement a `_get_sdk_version()` function:
- Use `importlib.metadata.version("hiero-sdk-python")` to retrieve the installed
package version at runtime
- Catch `PackageNotFoundError` and return `"dev"` as the fallback version for
development checkouts
- Define a header key constant for `X-User-Agent` and format the header value as
`hiero-sdk-python/{version}`
- Implement a `_UserAgentInterceptor` class that inherits from both
`grpc.UnaryUnaryClientInterceptor` and `grpc.UnaryStreamClientInterceptor`:
- Implement `intercept_unary_unary()` to handle standard transaction and query
calls (used by `executable.py`)
- Implement `intercept_unary_stream()` to handle server-streaming calls (used by
`topic_message_query.py` for topic subscriptions)
- In both methods, construct new `ClientCallDetails` with the `X-User-Agent`
metadata appended to any existing metadata, then delegate to the `continuation`
callable
- The interceptor should be stateless; the version string may be computed once
at module load or on first use
===============================================================================

Task: 2

Integrate the `_UserAgentInterceptor` into both channel creation points in the
existing codebase, then create a unit test file to verify behavior. This phase
ensures the interceptor is applied consistently across all outgoing gRPC calls.

**`src/hiero_sdk_python/node.py`** — update `_Node._get_channel()`:
- Import the interceptor from `user_agent_interceptor.py`; consider creating a
single shared interceptor instance at module level to avoid repeated
instantiation
- After creating the channel via `grpc.secure_channel()` or
`grpc.insecure_channel()`, wrap it using `grpc.intercept_channel(channel,
interceptor)` before passing it to `_Channel()`

**`src/hiero_sdk_python/client/client.py`** — update `_init_mirror_stub()`:
- Import and use the same interceptor class/instance as the consensus node
channels for consistency
- After creating the channel via `grpc.secure_channel()` or
`grpc.insecure_channel()`, wrap it using `grpc.intercept_channel(channel,
interceptor)` before creating the stub

**`tests/unit/user_agent_interceptor_test.py`** — create new test file:
- Follow existing test patterns: `from __future__ import annotations`,
`pytest.mark.unit` markers, and `@patch` decorators for mocking
- Test that `_get_sdk_version()` returns a version string (real or `"dev"`
fallback)
- Test that `_get_sdk_version()` returns `"dev"` when
`importlib.metadata.version()` raises `PackageNotFoundError` (simulated via
`@patch`)
- Test that `intercept_unary_unary()` adds the `X-User-Agent` header with the
correct format to `ClientCallDetails.metadata`
- Test that `intercept_unary_stream()` similarly adds the header
- Test that any pre-existing metadata is preserved when the header is appended
- Use mock continuations to verify the modified `ClientCallDetails` is passed
through correctly

💡 Iterate on the plan with: @coderabbitai <feedback>

Example Feedback
- `@coderabbitai` You can skip phase 3. Add a simple unit test case for phase 2.
- `@coderabbitai` For design choice 1 go ahead with option 3 and replan.

---

💬 Have feedback or questions? Drop into our discord!

[github-actions]

👋 Hi @shady-2004! Thanks for your interest in contributing 💡

Before taking on a beginner issue, we ask contributors to complete at least one Good First Issue to get familiar with the workflow.

👉 Find a Good First Issue here

Please try a GFI first, then come back — we’ll be happy to assign this! 😊

[github-actions]

👋 Hi @shady-2004! If you'd like to work on this issue, please comment /assign to get assigned.

[oGranny]

/assign

[github-actions]

Hi @oGranny 👋

This issue has been assigned but no pull request has been created yet.
Are you still planning on working on it?
If you are, please create a draft PR linked to this issue or comment /working to let us know.
If you’re no longer able to work on this issue, you can comment /unassign to release it.

From the Python SDK Team

[github-actions]

Hello, this is CommunityCallBot.

This is a reminder that the Hiero Python SDK Community Call will begin in approximately 4 hours (14:00 UTC).

The call is an open forum where contributors and users can discuss topics, raise issues, and influence the direction of the Python SDK.

Details:

• Time: 14:00 UTC
• Join Link: Zoom Meeting

Disclaimer: This is an automated reminder. Please verify the schedule here for any changes.