diff --git a/sdk/core/azure-core/HISTORY.md b/sdk/core/azure-core/HISTORY.md
index c46f17aa4b37..25a794018eb4 100644
--- a/sdk/core/azure-core/HISTORY.md
+++ b/sdk/core/azure-core/HISTORY.md
@@ -5,6 +5,7 @@
 
 ### Features
 
+- Add user_agent & sdk_moniker kwargs in UserAgentPolicy init   #9355
 - Support OPTIONS HTTP verb     #9322
 - Add tracing_attributes to tracing decorator   #9297
 - Support auto_request_id in RequestIdPolicy   #9163
diff --git a/sdk/core/azure-core/azure/core/pipeline/policies/_universal.py b/sdk/core/azure-core/azure/core/pipeline/policies/_universal.py
index 6d656c636ffa..cc5b3274eada 100644
--- a/sdk/core/azure-core/azure/core/pipeline/policies/_universal.py
+++ b/sdk/core/azure-core/azure/core/pipeline/policies/_universal.py
@@ -164,6 +164,9 @@ class UserAgentPolicy(SansIOHTTPPolicy):
 
     :keyword bool user_agent_overwrite: Overwrites User-Agent when True. Defaults to False.
     :keyword bool user_agent_use_env: Gets user-agent from environment. Defaults to True.
+    :keyword str user_agent: If specified, this will be added in front of the user agent string.
+    :keyword str sdk_moniker: If specified, the user agent string will be
+        azsdk-python-[sdk_moniker] Python/[python_version] ([platform_version])
 
     .. admonition:: Example:
 
@@ -178,18 +181,23 @@ class UserAgentPolicy(SansIOHTTPPolicy):
     _ENV_ADDITIONAL_USER_AGENT = 'AZURE_HTTP_USER_AGENT'
 
     def __init__(self, base_user_agent=None, **kwargs):  # pylint: disable=super-init-not-called
-        # type: (Optional[str], bool) -> None
+        # type: (Optional[str], **Any) -> None
         self.overwrite = kwargs.pop('user_agent_overwrite', False)
         self.use_env = kwargs.pop('user_agent_use_env', True)
+        application_id = kwargs.pop('user_agent', None)
+        sdk_moniker = kwargs.pop('sdk_moniker', 'core/{}'.format(azcore_version))
 
-        if base_user_agent is None:
-            self._user_agent = "azsdk-python-core/{} Python/{} ({})".format(
-                azcore_version,
+        if base_user_agent:
+            self._user_agent = base_user_agent
+        else:
+            self._user_agent = "azsdk-python-{} Python/{} ({})".format(
+                sdk_moniker,
                 platform.python_version(),
                 platform.platform()
             )
-        else:
-            self._user_agent = base_user_agent
+
+        if application_id:
+            self._user_agent = "{} {}".format(application_id, self._user_agent)
 
     @property
     def user_agent(self):
@@ -222,7 +230,7 @@ def on_request(self, request):
             if options_dict.pop('user_agent_overwrite', self.overwrite):
                 http_request.headers[self._USERAGENT] = user_agent
             else:
-                user_agent = "{} {}".format(self.user_agent, user_agent)
+                user_agent = "{} {}".format(user_agent, self.user_agent)
                 http_request.headers[self._USERAGENT] = user_agent
 
         elif self.overwrite or self._USERAGENT not in http_request.headers:
diff --git a/sdk/core/azure-core/tests/test_universal_pipeline.py b/sdk/core/azure-core/tests/test_universal_pipeline.py
index fe2db8c515a2..0093bef899b1 100644
--- a/sdk/core/azure-core/tests/test_universal_pipeline.py
+++ b/sdk/core/azure-core/tests/test_universal_pipeline.py
@@ -56,16 +56,6 @@
     HTTPPolicy,
 )
 
-def test_user_agent():
-
-    with mock.patch.dict('os.environ', {'AZURE_HTTP_USER_AGENT': "mytools"}):
-        policy = UserAgentPolicy(None)
-        assert policy.user_agent.endswith("mytools")
-
-        request = HttpRequest('GET', 'http://127.0.0.1/')
-        policy.on_request(PipelineRequest(request, PipelineContext(None)))
-        assert request.headers["user-agent"].endswith("mytools")
-
 def test_request_history():
     class Non_deep_copiable(object):
         def __deepcopy__(self, memodict={}):
diff --git a/sdk/core/azure-core/tests/test_user_agent_policy.py b/sdk/core/azure-core/tests/test_user_agent_policy.py
new file mode 100644
index 000000000000..239daffbe63d
--- /dev/null
+++ b/sdk/core/azure-core/tests/test_user_agent_policy.py
@@ -0,0 +1,40 @@
+# ------------------------------------
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+# ------------------------------------
+"""Tests for the user agent policy."""
+from azure.core.pipeline.policies import UserAgentPolicy
+from azure.core.pipeline.transport import HttpRequest
+from azure.core.pipeline import PipelineRequest, PipelineContext
+try:
+    from unittest import mock
+except ImportError:
+    import mock
+
+def test_user_agent_policy():
+    user_agent = UserAgentPolicy(base_user_agent='foo')
+    assert user_agent._user_agent == 'foo'
+
+    user_agent = UserAgentPolicy(sdk_moniker='foosdk/1.0.0')
+    assert user_agent._user_agent.startswith('azsdk-python-foosdk/1.0.0 Python')
+
+    user_agent = UserAgentPolicy(base_user_agent='foo', user_agent='bar', user_agent_use_env=False)
+    assert user_agent._user_agent == 'bar foo'
+
+    request = HttpRequest('GET', 'http://127.0.0.1/')
+    pipeline_request = PipelineRequest(request, PipelineContext(None))
+
+    pipeline_request.context.options['user_agent'] = 'xyz'
+    user_agent.on_request(pipeline_request)
+    assert request.headers['User-Agent'] == 'xyz bar foo'
+
+
+def test_user_agent_environ():
+
+    with mock.patch.dict('os.environ', {'AZURE_HTTP_USER_AGENT': "mytools"}):
+        policy = UserAgentPolicy(None)
+        assert policy.user_agent.endswith("mytools")
+
+        request = HttpRequest('GET', 'http://127.0.0.1/')
+        policy.on_request(PipelineRequest(request, PipelineContext(None)))
+        assert request.headers["user-agent"].endswith("mytools")