diff --git a/firebase_admin/_utils.py b/firebase_admin/_utils.py
index 0277b9e5..1c7c3337 100644
--- a/firebase_admin/_utils.py
+++ b/firebase_admin/_utils.py
@@ -15,6 +15,8 @@
 """Internal utilities common to all modules."""
 
 import json
+import os
+import re
 from platform import python_version
 from typing import Callable, Optional
 
@@ -345,3 +347,16 @@ def __init__(self):
 
     def refresh(self, request):
         pass
+
+
+def get_emulator_host(env_var_name: str) -> Optional[str]:
+    """Retrieves and validates the host from the specified emulator environment variable."""
+    emulator_host = os.environ.get(env_var_name)
+    if emulator_host:
+        if not re.match(r'^(?:\[[a-fA-F0-9:]+\]|[a-zA-Z0-9._-]+):[0-9]+$', emulator_host):
+            raise ValueError(
+                f'Invalid {env_var_name}: "{emulator_host}". It must follow format '
+                '"host:port".'
+            )
+        return emulator_host
+    return None
diff --git a/firebase_admin/dataconnect.py b/firebase_admin/dataconnect.py
index 201e3b12..a3ee51a5 100644
--- a/firebase_admin/dataconnect.py
+++ b/firebase_admin/dataconnect.py
@@ -18,14 +18,39 @@
 Firebase apps.
 """
 
-from dataclasses import dataclass
-from typing import Dict, Optional
+from collections.abc import Mapping
+from dataclasses import dataclass, asdict, is_dataclass
+import typing
+from typing import Any, Dict, Generic, Optional, Type, TypeVar, Union
+import firebase_admin
+from firebase_admin import _utils, _http_client, App
 
-from firebase_admin import _utils, App
-
-__all__ = ['ConnectorConfig', 'DataConnect', 'client']
+__all__ = [
+    'ConnectorConfig',
+    'DataConnect',
+    'client',
+    'GraphqlOptions',
+    'Impersonation',
+    'ExecuteGraphqlResponse',
+]
 
 _DATA_CONNECT_ATTRIBUTE = '_data_connect'
+_DATA_CONNECT_PROD_URL = 'https://firebasedataconnect.googleapis.com'
+_API_VERSION = 'v1'
+
+_SERVICES_URL_FORMAT = (
+    '{host}/{version}/projects/{project_id}/locations/{location_id}'
+    '/services/{service_id}:{endpoint_id}'
+)
+
+_EMULATOR_SERVICES_URL_FORMAT = (
+    'http://{host}/{version}/projects/{project_id}/locations/{location_id}'
+    '/services/{service_id}:{endpoint_id}'
+)
+
+# Generic Type Parameters
+_Data = TypeVar("_Data")
+_Variables = TypeVar("_Variables")
 
 @dataclass(frozen=True)
 class ConnectorConfig:
@@ -122,3 +147,190 @@ def client(config: ConnectorConfig, app: Optional[App] = None) -> DataConnect:
     dc_service = _utils.get_app_service(app, _DATA_CONNECT_ATTRIBUTE, _DataConnectService)
 
     return dc_service.get_client(config)
+
+
+
+class Impersonation(dict):
+    """Represents impersonation configuration for DataConnect requests."""
+
+    @staticmethod
+    def unauthenticated() -> 'Impersonation':
+        """Returns impersonation configuration for unauthenticated requests."""
+        return Impersonation(unauthenticated=True)
+
+    @staticmethod
+    def authenticated(auth_claims: Dict[str, Any]) -> 'Impersonation':
+        """Returns impersonation configuration for authenticated requests.
+
+        # TODO: More strongly type auth_claims later.
+        """
+        return Impersonation(authClaims=auth_claims)
+
+
+@dataclass
+class GraphqlOptions(Generic[_Variables]):
+    variables: Optional[_Variables] = None
+    operation_name: Optional[str] = None
+    impersonate: Optional[Union[Impersonation, Dict[str, Any]]] = None
+
+
+@dataclass
+class ExecuteGraphqlResponse(Generic[_Data]):
+    data: _Data
+
+
+def _get_emulator_host() -> Optional[str]:
+    return _utils.get_emulator_host("DATA_CONNECT_EMULATOR_HOST")
+
+
+class _DataConnectApiClient:
+    """Internal client for sending requests to the Firebase Data Connect backend.
+
+    Attributes:
+        connector_config: The connector configuration specifying the service,
+            location, and connector name.
+        app: The Firebase App instance associated with this client.
+    """
+
+    def __init__(self, connector_config: ConnectorConfig, app: App) -> None:
+        if not isinstance(app, App):
+            raise ValueError(
+                'Second argument passed to DataConnectApiClient must be a valid '
+                'Firebase app instance.'
+            )
+        self._connector_config = connector_config
+        self._app = app
+
+        self._project_id = app.project_id
+        if not self._project_id:
+            raise ValueError(
+                'Failed to determine project ID. Initialize the SDK with service '
+                'account credentials or set project ID as an app option. Alternatively, set the '
+                'GOOGLE_CLOUD_PROJECT environment variable.')
+
+        self._emulator_host = _get_emulator_host()
+        if self._emulator_host:
+            self._credential = _utils.EmulatorAdminCredentials()
+        else:
+            self._credential = app.credential.get_credential()
+
+        self._http_client = _http_client.JsonHttpClient(credential=self._credential)
+
+    def _validate_variables_type(
+        self,
+        variables: Any,
+        variable_type: Optional[Type[Any]] = None
+    ) -> None:
+        """Validates variables against expected type."""
+        if variables is not None:
+            if not (isinstance(variables, Mapping) or is_dataclass(variables)):
+                raise ValueError("variables must be a collections.abc.Mapping or a dataclass")
+            if variable_type is not None:
+                expected_type = typing.get_origin(variable_type) or variable_type
+                if not isinstance(variables, expected_type):
+                    type_name = getattr(expected_type, '__name__', str(expected_type))
+                    raise ValueError(f"variables must be of type {type_name}")
+
+    def _validate_impersonation_options(self, impersonate: Any) -> None:
+        """Validates impersonation dictionary options."""
+        if impersonate is not None:
+            if not isinstance(impersonate, dict):
+                raise ValueError('impersonate option must be a dictionary')
+            if 'unauthenticated' not in impersonate and 'authClaims' not in impersonate:
+                raise ValueError(
+                    "impersonate option must contain either "
+                    "'unauthenticated' or 'authClaims'"
+                )
+            if 'unauthenticated' in impersonate and 'authClaims' in impersonate:
+                raise ValueError(
+                    "impersonate option cannot contain both "
+                    "'unauthenticated' and 'authClaims'"
+                )
+            if 'unauthenticated' in impersonate:
+                if not isinstance(impersonate['unauthenticated'], bool):
+                    raise ValueError("'unauthenticated' claim must be a boolean")
+            if 'authClaims' in impersonate:
+                if not isinstance(impersonate['authClaims'], dict):
+                    raise ValueError("'authClaims' claim must be a dictionary")
+
+    def _validate_graphql_options(
+        self,
+        graphql_options: Optional[GraphqlOptions[Any]],
+        variable_type: Optional[Type[Any]] = None
+    ) -> None:
+        """Validates GraphqlOptions inputs at runtime."""
+        if graphql_options is not None:
+            if not isinstance(graphql_options, GraphqlOptions):
+                raise ValueError('options must be a GraphqlOptions instance')
+
+            # Validate Variables against expected variable_type
+            self._validate_variables_type(graphql_options.variables, variable_type)
+
+            # Validate Operation Name (if it exists)
+            operation_name = graphql_options.operation_name
+            if operation_name is not None:
+                if not isinstance(operation_name, str):
+                    raise ValueError('operation_name must be a string')
+                if not operation_name.strip():
+                    raise ValueError('operation_name must be a non-empty string')
+
+            # Validate Impersonation (if it exists)
+            self._validate_impersonation_options(graphql_options.impersonate)
+
+    def _prepare_graphql_payload(
+        self,
+        graphql_query: str,
+        graphql_options: Optional[GraphqlOptions[_Variables]]
+    ) -> Dict[str, Any]:
+        """Serializes input query and options to JSON-compatible dictionary."""
+        payload = {
+            "query": graphql_query
+        }
+
+        if graphql_options is not None:
+            if graphql_options.variables is not None:
+                if is_dataclass(graphql_options.variables):
+                    payload["variables"] = asdict(graphql_options.variables)
+                else:
+                    payload["variables"] = graphql_options.variables
+
+            if graphql_options.operation_name is not None:
+                payload["operationName"] = graphql_options.operation_name.strip()
+
+            if graphql_options.impersonate is not None:
+                payload["extensions"] = {
+                    "impersonate": graphql_options.impersonate
+                }
+
+        return payload
+
+    def _get_firebase_dataconnect_service_url(self, method_name: str) -> str:
+        """Build and return the URL for a Firebase Data Connect API method."""
+        project_id = self._project_id
+        location = self._connector_config.location
+        service_id = self._connector_config.service_id
+
+        if self._emulator_host:
+            return _EMULATOR_SERVICES_URL_FORMAT.format(
+                host=self._emulator_host,
+                version=_API_VERSION,
+                project_id=project_id,
+                location_id=location,
+                service_id=service_id,
+                endpoint_id=method_name
+            )
+        return _SERVICES_URL_FORMAT.format(
+            host=_DATA_CONNECT_PROD_URL,
+            version=_API_VERSION,
+            project_id=project_id,
+            location_id=location,
+            service_id=service_id,
+            endpoint_id=method_name
+        )
+
+    def _get_headers(self) -> Dict[str, str]:
+        """Build and return the headers for a Firebase Data Connect API call."""
+        return {
+            "X-Firebase-Client": f"fire-admin-python/{firebase_admin.__version__}",
+            "x-goog-api-client": _utils.get_metrics_header(),
+        }
diff --git a/firebase_admin/functions.py b/firebase_admin/functions.py
index 66ba700b..b111d88f 100644
--- a/firebase_admin/functions.py
+++ b/firebase_admin/functions.py
@@ -18,7 +18,6 @@
 from datetime import datetime, timedelta, timezone
 from urllib import parse
 import re
-import os
 import json
 from base64 import b64encode
 from typing import Any, Optional, Dict
@@ -62,14 +61,7 @@
 _DEFAULT_LOCATION = 'us-central1'
 
 def _get_emulator_host() -> Optional[str]:
-    emulator_host = os.environ.get(_EMULATOR_HOST_ENV_VAR)
-    if emulator_host:
-        if '//' in emulator_host:
-            raise ValueError(
-                f'Invalid {_EMULATOR_HOST_ENV_VAR}: "{emulator_host}". It must follow format '
-                '"host:port".')
-        return emulator_host
-    return None
+    return _utils.get_emulator_host(_EMULATOR_HOST_ENV_VAR)
 
 
 def _get_functions_service(app) -> _FunctionsService:
diff --git a/tests/test_data_connect.py b/tests/test_data_connect.py
index c226b12b..9e1faf04 100644
--- a/tests/test_data_connect.py
+++ b/tests/test_data_connect.py
@@ -14,14 +14,20 @@
 
 """Test cases for the firebase_admin.dataconnect module."""
 
+from dataclasses import dataclass
+from typing import Any, Dict, Mapping
 from unittest import mock
+
+from google.auth import credentials as google_auth_credentials
 import pytest
 
+
 import firebase_admin
 from firebase_admin import _utils
 from firebase_admin import dataconnect
 from tests import testutils
 
+
 BASE_CONFIG = dataconnect.ConnectorConfig(
     service_id="starterproject",
     location="us-east4",
@@ -331,3 +337,390 @@ def test_overall_client_retrieval_and_caching(self):
         assert client1_app2.app is self.app2
         assert client1_app2.config is self.config1
         assert client1_app2 is not client1a
+
+
+class TestDataConnectApiClientConstructor:
+
+    def setup_method(self):
+        self.cred = testutils.MockCredential()
+
+    def teardown_method(self, method):
+        del method
+        testutils.cleanup_apps()
+
+    def test_constructor_invalid_app(self):
+        msg = (
+            "Second argument passed to DataConnectApiClient must be a valid "
+            "Firebase app instance."
+        )
+        with pytest.raises(ValueError, match=msg):
+            dataconnect._DataConnectApiClient(BASE_CONFIG, None)
+
+    def test_constructor_missing_project_id(self):
+        class CredentialWithoutProjectId(firebase_admin.credentials.Base):
+            def get_credential(self):
+                class DummyGoogleCred(google_auth_credentials.Credentials):
+                    def refresh(self, request):
+                        pass
+                return DummyGoogleCred()
+
+        app_no_project_id = firebase_admin.initialize_app(
+            CredentialWithoutProjectId(),
+            name="no-project-id-app"
+        )
+        try:
+            with pytest.raises(ValueError, match="Failed to determine project ID"):
+                dataconnect._DataConnectApiClient(BASE_CONFIG, app_no_project_id)
+        finally:
+            firebase_admin.delete_app(app_no_project_id)
+
+    def test_constructor_connector_config(self):
+        app = firebase_admin.initialize_app(self.cred, options={'projectId': 'test-project'})
+        api_client = dataconnect._DataConnectApiClient(BASE_CONFIG, app)
+        assert api_client._connector_config is BASE_CONFIG
+
+    def test_constructor_emulator_host_invalid(self, monkeypatch):
+        monkeypatch.setenv("DATA_CONNECT_EMULATOR_HOST", "http://localhost:9399")
+        app = firebase_admin.initialize_app(self.cred, options={'projectId': 'test-project'})
+        with pytest.raises(ValueError, match="Invalid DATA_CONNECT_EMULATOR_HOST"):
+            dataconnect._DataConnectApiClient(BASE_CONFIG, app)
+
+
+class TestDataConnectApiClientValidateGraphqlOptions:
+
+    def setup_method(self):
+        self.cred = testutils.MockCredential()
+        self.app = firebase_admin.initialize_app(
+            self.cred, options={'projectId': 'test-project'}
+        )
+        self.api_client = dataconnect._DataConnectApiClient(BASE_CONFIG, self.app)
+
+    def teardown_method(self, method):
+        del method
+        testutils.cleanup_apps()
+
+    def test_validate_graphql_options_valid(self):
+        # Valid with no options
+        self.api_client._validate_graphql_options(None)
+
+        # Valid with default options (no arguments)
+        options = dataconnect.GraphqlOptions()
+        self.api_client._validate_graphql_options(options)
+
+    def test_validate_graphql_options_valid_impersonate(self):
+        # Valid unauthenticated impersonation
+        imp_unauth = dataconnect.Impersonation.unauthenticated()
+        options = dataconnect.GraphqlOptions(impersonate=imp_unauth)
+        self.api_client._validate_graphql_options(options)
+
+        # Valid authenticated impersonation
+        imp_auth = dataconnect.Impersonation.authenticated(
+            {"sub": "authenticated-UUID"}
+        )
+        options = dataconnect.GraphqlOptions(impersonate=imp_auth)
+        self.api_client._validate_graphql_options(options)
+
+    def test_validate_graphql_options_valid_dataclass_variables(self):
+        @dataclass
+        class UserProfile:
+            address: str
+            phone: str
+
+        @dataclass
+        class CreateUserVariables:
+            user_id: str
+            name: str
+            profile: UserProfile
+
+        profile_val = UserProfile(address="123 Road", phone="332-3233-0199")
+        valid_variables = CreateUserVariables(
+            user_id="1", name="Fred", profile=profile_val
+        )
+        options = dataconnect.GraphqlOptions(variables=valid_variables)
+        self.api_client._validate_graphql_options(options, CreateUserVariables)
+
+    def test_validate_graphql_options_valid_mapping_variables(self):
+        options = dataconnect.GraphqlOptions(variables={"user_id": "1", "name": "Fred"})
+        self.api_client._validate_graphql_options(options)
+
+    def test_validate_graphql_options_valid_generic_variables(self):
+        options = dataconnect.GraphqlOptions(variables={"user_id": "1", "name": "Fred"})
+        self.api_client._validate_graphql_options(options, Dict[str, Any])
+        self.api_client._validate_graphql_options(options, Mapping[str, Any])
+
+    def test_validate_graphql_options_invalid_options(self):
+        with pytest.raises(ValueError, match="options must be a GraphqlOptions instance"):
+            self.api_client._validate_graphql_options("invalid-options")
+
+    def test_validate_graphql_options_invalid_impersonate(self):
+        # impersonate must be dict
+        options = dataconnect.GraphqlOptions(impersonate="invalid")
+        with pytest.raises(ValueError, match="impersonate option must be a dictionary"):
+            self.api_client._validate_graphql_options(options)
+
+        # impersonate must have either unauthenticated or authClaims
+        options = dataconnect.GraphqlOptions(impersonate={"invalid_key": True})
+        msg = (
+            "impersonate option must contain either "
+            "'unauthenticated' or 'authClaims'"
+        )
+        with pytest.raises(ValueError, match=msg):
+            self.api_client._validate_graphql_options(options)
+
+        # unauthenticated must be boolean
+        options = dataconnect.GraphqlOptions(impersonate={"unauthenticated": "not-bool"})
+        with pytest.raises(ValueError, match="'unauthenticated' claim must be a boolean"):
+            self.api_client._validate_graphql_options(options)
+
+        # authClaims must be a dict
+        options = dataconnect.GraphqlOptions(impersonate={"authClaims": "not-dict"})
+        with pytest.raises(ValueError, match="'authClaims' claim must be a dictionary"):
+            self.api_client._validate_graphql_options(options)
+
+        # impersonate cannot contain both unauthenticated and authClaims
+        options = dataconnect.GraphqlOptions(
+            impersonate={"unauthenticated": True, "authClaims": {"uid": "123"}}
+        )
+        msg = (
+            "impersonate option cannot contain both "
+            "'unauthenticated' and 'authClaims'"
+        )
+        with pytest.raises(ValueError, match=msg):
+            self.api_client._validate_graphql_options(options)
+
+    def test_validate_graphql_options_invalid_operation_name(self):
+        # Test type validation
+        options = dataconnect.GraphqlOptions(operation_name=123)
+        with pytest.raises(ValueError, match="operation_name must be a string"):
+            self.api_client._validate_graphql_options(options)
+
+        # Test empty string validation
+        options = dataconnect.GraphqlOptions(operation_name="")
+        with pytest.raises(ValueError, match="operation_name must be a non-empty string"):
+            self.api_client._validate_graphql_options(options)
+
+        # Test stripped whitespace validation
+        options = dataconnect.GraphqlOptions(operation_name="   ")
+        with pytest.raises(ValueError, match="operation_name must be a non-empty string"):
+            self.api_client._validate_graphql_options(options)
+
+    def test_validate_graphql_options_invalid_variables(self):
+        @dataclass
+        class UserProfile:
+            address: str
+            phone: str
+
+        @dataclass
+        class CreateUserVariables:
+            user_id: str
+            name: str
+            profile: UserProfile
+
+        # Test invalid variable format (not Mapping or dataclass)
+        options = dataconnect.GraphqlOptions(variables="invalid-string-format")
+        msg = "variables must be a collections.abc.Mapping or a dataclass"
+        with pytest.raises(ValueError, match=msg):
+            self.api_client._validate_graphql_options(options)
+
+        # Test valid Mapping format but type mismatch against expected dataclass type
+        options = dataconnect.GraphqlOptions(variables={"foo": "bar"})
+        with pytest.raises(ValueError, match="variables must be of type CreateUserVariables"):
+            self.api_client._validate_graphql_options(options, CreateUserVariables)
+
+        # Test type mismatch when variable_type is a tuple of classes (no __name__ attribute)
+        options = dataconnect.GraphqlOptions(variables={"foo": "bar"})
+        msg = r"variables must be of type \(\<class 'list'\>, \<class 'tuple'\>\)"
+        with pytest.raises(ValueError, match=msg):
+            self.api_client._validate_graphql_options(options, (list, tuple))
+
+        # Test type mismatch when a dataclass is passed but a Dict is expected
+        profile_val = UserProfile(address="123 Road", phone="332-3233-0199")
+        valid_variables = CreateUserVariables(
+            user_id="1", name="Fred", profile=profile_val
+        )
+        options = dataconnect.GraphqlOptions(variables=valid_variables)
+        with pytest.raises(ValueError, match="variables must be of type dict"):
+            self.api_client._validate_graphql_options(options, Dict[str, Any])
+
+
+class TestDataConnectApiClientPrepareGraphqlPayload:
+
+    def setup_method(self):
+        self.cred = testutils.MockCredential()
+        self.app = firebase_admin.initialize_app(self.cred, options={'projectId': 'test-project'})
+        self.api_client = dataconnect._DataConnectApiClient(BASE_CONFIG, self.app)
+
+    def teardown_method(self, method):
+        del method
+        testutils.cleanup_apps()
+
+    def test_prepare_graphql_payload_only_query(self):
+        payload = self.api_client._prepare_graphql_payload("query { hello }", None)
+        assert payload == {"query": "query { hello }"}
+
+    def test_prepare_graphql_payload_with_variables(self):
+        options = dataconnect.GraphqlOptions(variables={"foo": "bar"})
+        payload = self.api_client._prepare_graphql_payload("query { hello }", options)
+        assert payload == {
+            "query": "query { hello }",
+            "variables": {"foo": "bar"}
+        }
+
+    def test_prepare_graphql_payload_with_dataclass_variables(self):
+        @dataclass
+        class UserProfile:
+            address: str
+            phone: str
+
+        @dataclass
+        class CreateUserVariables:
+            user_id: str
+            name: str
+            profile: UserProfile
+
+        profile_val = UserProfile(address="123 Road", phone="332-3233-0199")
+        valid_variables = CreateUserVariables(
+            user_id="1", name="Fred", profile=profile_val
+        )
+        options = dataconnect.GraphqlOptions(variables=valid_variables)
+        payload = self.api_client._prepare_graphql_payload("query { hello }", options)
+        assert payload == {
+            "query": "query { hello }",
+            "variables": {
+                "user_id": "1",
+                "name": "Fred",
+                "profile": {
+                    "address": "123 Road",
+                    "phone": "332-3233-0199"
+                }
+            }
+        }
+
+    def test_prepare_graphql_payload_with_operation_name(self):
+        options = dataconnect.GraphqlOptions(operation_name="  myOp  ")
+        self.api_client._validate_graphql_options(options)
+        assert options.operation_name == "  myOp  "
+
+        payload = self.api_client._prepare_graphql_payload("query { hello }", options)
+        assert payload == {
+            "query": "query { hello }",
+            "operationName": "myOp"
+        }
+        assert options.operation_name == "  myOp  "
+
+    def test_prepare_graphql_payload_with_impersonate_unauthenticated(self):
+        imp_unauth = dataconnect.Impersonation.unauthenticated()
+        options = dataconnect.GraphqlOptions(impersonate=imp_unauth)
+        payload = self.api_client._prepare_graphql_payload("query { hello }", options)
+        assert payload == {
+            "query": "query { hello }",
+            "extensions": {
+                "impersonate": {"unauthenticated": True}
+            }
+        }
+
+    def test_prepare_graphql_payload_with_impersonate_authenticated(self):
+        imp_auth = dataconnect.Impersonation.authenticated(
+            {"sub": "authenticated-UUID"}
+        )
+        options = dataconnect.GraphqlOptions(impersonate=imp_auth)
+        payload = self.api_client._prepare_graphql_payload("query { hello }", options)
+        assert payload == {
+            "query": "query { hello }",
+            "extensions": {
+                "impersonate": {"authClaims": {"sub": "authenticated-UUID"}}
+            }
+        }
+
+    def test_prepare_graphql_payload_with_all_fields(self):
+        @dataclass
+        class UserProfile:
+            address: str
+            phone: str
+
+        @dataclass
+        class CreateUserVariables:
+            user_id: str
+            name: str
+            profile: UserProfile
+
+        profile_val = UserProfile(address="123 Road", phone="332-3233-0199")
+        valid_variables = CreateUserVariables(
+            user_id="1", name="Fred", profile=profile_val
+        )
+        imp_auth = dataconnect.Impersonation.authenticated(
+            {"sub": "authenticated-UUID"}
+        )
+        options = dataconnect.GraphqlOptions(
+            variables=valid_variables,
+            operation_name="getUsers",
+            impersonate=imp_auth
+        )
+        payload = self.api_client._prepare_graphql_payload("query { hello }", options)
+        assert payload == {
+            "query": "query { hello }",
+            "operationName": "getUsers",
+            "variables": {
+                "user_id": "1",
+                "name": "Fred",
+                "profile": {
+                    "address": "123 Road",
+                    "phone": "332-3233-0199"
+                }
+            },
+            "extensions": {
+                "impersonate": {"authClaims": {"sub": "authenticated-UUID"}}
+            }
+        }
+
+
+class TestDataConnectApiClientServiceUrl:
+
+    def setup_method(self):
+        self.cred = testutils.MockCredential()
+        self.app = firebase_admin.initialize_app(self.cred, options={'projectId': 'test-project'})
+        self.api_client = dataconnect._DataConnectApiClient(BASE_CONFIG, self.app)
+
+    def teardown_method(self, method):
+        del method
+        testutils.cleanup_apps()
+
+    def test_get_firebase_dataconnect_service_url_production(self):
+        url = self.api_client._get_firebase_dataconnect_service_url("executeGraphql")
+        expected = (
+            "https://firebasedataconnect.googleapis.com/v1"
+            "/projects/test-project/locations/us-east4"
+            "/services/starterproject:executeGraphql"
+        )
+        assert url == expected
+
+    def test_get_firebase_dataconnect_service_url_emulator(self, monkeypatch):
+        monkeypatch.setenv("DATA_CONNECT_EMULATOR_HOST", "localhost:9399")
+        api_client = dataconnect._DataConnectApiClient(BASE_CONFIG, self.app)
+        url = api_client._get_firebase_dataconnect_service_url("executeGraphql")
+        expected = (
+            "http://localhost:9399/v1"
+            "/projects/test-project/locations/us-east4"
+            "/services/starterproject:executeGraphql"
+        )
+        assert url == expected
+
+
+class TestDataConnectApiClientGetHeaders:
+
+    def setup_method(self):
+        self.cred = testutils.MockCredential()
+        self.app = firebase_admin.initialize_app(
+            self.cred, options={'projectId': 'test-project'}
+        )
+        self.api_client = dataconnect._DataConnectApiClient(BASE_CONFIG, self.app)
+
+    def teardown_method(self, method):
+        del method
+        testutils.cleanup_apps()
+
+    def test_get_headers(self):
+        headers = self.api_client._get_headers()
+        assert isinstance(headers, dict)
+        assert headers.get("X-Firebase-Client") == f"fire-admin-python/{firebase_admin.__version__}"
+        assert headers.get("x-goog-api-client") == _utils.get_metrics_header()
diff --git a/tests/test_utils.py b/tests/test_utils.py
new file mode 100644
index 00000000..4d57e44a
--- /dev/null
+++ b/tests/test_utils.py
@@ -0,0 +1,55 @@
+# Copyright 2026 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Test cases for the firebase_admin._utils module."""
+
+import pytest
+from firebase_admin import _utils
+
+class TestGetEmulatorHost:
+
+    @pytest.mark.parametrize('host', [
+        'localhost:8080',
+        '127.0.0.1:8080',
+        '[::1]:8080',
+        '[2001:db8::1]:8080',
+        'my-host:9000',
+        'my_host:9000',
+        'my.host.name:12345',
+        'host_with_underscores:8080',
+    ])
+    def test_get_emulator_host_valid(self, monkeypatch, host):
+        monkeypatch.setenv('TEST_EMULATOR_HOST', host)
+        assert _utils.get_emulator_host('TEST_EMULATOR_HOST') == host
+
+    @pytest.mark.parametrize('host', [
+        'http://localhost:8080',
+        'localhost',
+        '127.0.0.1',
+        '[::1]',
+        'my_host',
+        'localhost:abc',
+        'localhost:',
+        ':8080',
+        'invalid_host_name_with_chars$:8080',
+        'host@name:8080',
+    ])
+    def test_get_emulator_host_invalid(self, monkeypatch, host):
+        monkeypatch.setenv('TEST_EMULATOR_HOST', host)
+        with pytest.raises(ValueError, match='Invalid TEST_EMULATOR_HOST'):
+            _utils.get_emulator_host('TEST_EMULATOR_HOST')
+
+    def test_get_emulator_host_not_set(self, monkeypatch):
+        monkeypatch.delenv('TEST_EMULATOR_HOST', raising=False)
+        assert _utils.get_emulator_host('TEST_EMULATOR_HOST') is None