diff --git a/api/controllers/console/wraps.py b/api/controllers/console/wraps.py
index 4b5fb7ca5b..ef2931ce9b 100644
--- a/api/controllers/console/wraps.py
+++ b/api/controllers/console/wraps.py
@@ -20,7 +20,7 @@ from models.account import AccountStatus
 from models.dataset import RateLimitLog
 from models.model import DifySetup
 from services.feature_service import FeatureService, LicenseStatus
-from services.operation_service import OperationService
+from services.operation_service import OperationService, UtmInfo
 
 from .error import NotInitValidateError, NotSetupError, UnauthorizedAndForceLogout
 
@@ -205,7 +205,7 @@ def cloud_utm_record[**P, R](view: Callable[P, R]) -> Callable[P, R]:
                 utm_info = request.cookies.get("utm_info")
 
                 if utm_info:
-                    utm_info_dict: dict = json.loads(utm_info)
+                    utm_info_dict: UtmInfo = json.loads(utm_info)
                     OperationService.record_utm(current_tenant_id, utm_info_dict)
 
         return view(*args, **kwargs)
diff --git a/api/services/operation_service.py b/api/services/operation_service.py
index c05e9d555c..903efd26ae 100644
--- a/api/services/operation_service.py
+++ b/api/services/operation_service.py
@@ -1,8 +1,22 @@
 import os
+from typing import TypedDict
 
 import httpx
 
 
+class UtmInfo(TypedDict, total=False):
+    """Expected shape of the utm_info dict passed to record_utm.
+
+    All fields are optional; missing keys default to an empty string.
+    """
+
+    utm_source: str
+    utm_medium: str
+    utm_campaign: str
+    utm_content: str
+    utm_term: str
+
+
 class OperationService:
     base_url = os.environ.get("BILLING_API_URL", "BILLING_API_URL")
     secret_key = os.environ.get("BILLING_API_SECRET_KEY", "BILLING_API_SECRET_KEY")
@@ -17,7 +31,7 @@ class OperationService:
         return response.json()
 
     @classmethod
-    def record_utm(cls, tenant_id: str, utm_info: dict):
+    def record_utm(cls, tenant_id: str, utm_info: UtmInfo):
         params = {
             "tenant_id": tenant_id,
             "utm_source": utm_info.get("utm_source", ""),