diff --git a/dbt/adapters/snowflake/connections.py b/dbt/adapters/snowflake/connections.py
index fcd16aafc..97cfc1b55 100644
--- a/dbt/adapters/snowflake/connections.py
+++ b/dbt/adapters/snowflake/connections.py
@@ -94,6 +94,7 @@ class SnowflakeCredentials(Credentials):
     retry_all: bool = False
     insecure_mode: Optional[bool] = False
     reuse_connections: Optional[bool] = None
+    application: Optional[str] = None
 
     def __post_init__(self):
         if self.authenticator != "oauth" and (
@@ -352,7 +353,7 @@ def connect():
                 role=creds.role,
                 autocommit=True,
                 client_session_keep_alive=creds.client_session_keep_alive,
-                application="dbt",
+                application=creds.application if creds.application else "dbt",
                 insecure_mode=creds.insecure_mode,
                 session_parameters=session_parameters,
                 **creds.auth_args(),
diff --git a/tests/unit/test_snowflake_adapter.py b/tests/unit/test_snowflake_adapter.py
index 2666e7557..efd04a314 100644
--- a/tests/unit/test_snowflake_adapter.py
+++ b/tests/unit/test_snowflake_adapter.py
@@ -580,6 +580,36 @@ def test_query_tag(self):
             ]
         )
 
+    def test_application(self):
+        self.config.credentials = self.config.credentials.replace(
+            password="test_password", query_tag="test_application"
+        )
+        self.adapter = SnowflakeAdapter(self.config, get_context("spawn"))
+        conn = self.adapter.connections.set_connection_name(name="new_connection_with_new_config")
+
+        self.snowflake.assert_not_called()
+        conn.handle
+        self.snowflake.assert_has_calls(
+            [
+                mock.call(
+                    account="test_account",
+                    autocommit=True,
+                    client_session_keep_alive=False,
+                    database="test_database",
+                    password="test_password",
+                    role=None,
+                    schema="public",
+                    user="test_user",
+                    warehouse="test_warehouse",
+                    private_key=None,
+                    application="test_application",
+                    insecure_mode=False,
+                    session_parameters={},
+                    reuse_connections=None,
+                )
+            ]
+        )
+
     def test_reuse_connections_with_keep_alive(self):
         self.config.credentials = self.config.credentials.replace(
             reuse_connections=True, client_session_keep_alive=True