feat: add request source header X-Custom-Request-Context (#32)

Author: YumengBao

agentkit/client/base_service_client.py

@@ -26,7 +26,10 @@
 from volcengine.Credentials import Credentials
 from volcengine.ServiceInfo import ServiceInfo
 
-from agentkit.utils.ve_sign import get_volc_ak_sk_region
+from agentkit.utils.ve_sign import (
+    get_volc_ak_sk_region,
+    ensure_x_custom_source_header,
+)
 
 T = TypeVar("T")
 
@@ -73,7 +76,7 @@ def __init__(
         session_token: str = "",
         service_name: str = "",
         credential_env_prefix: str = "",
-        header: Optional[Dict[str, Any]] = {"Accept": "application/json"},
+        header: Optional[Dict[str, Any]] = None,
     ) -> None:
         """
         Initialize the service client.
@@ -105,6 +108,15 @@ def __init__(
         self.session_token = session_token
         self.service_name = service_name
 
+        if header is None:
+            effective_header: Dict[str, Any] = {"Accept": "application/json"}
+        else:
+            effective_header = header.copy()
+            if "Accept" not in effective_header:
+                effective_header = {"Accept": "application/json", **effective_header}
+
+        effective_header = ensure_x_custom_source_header(effective_header)
+
         # Get service-specific configuration from subclass
         config = self._get_service_config()
         self.host = config["host"]
@@ -115,7 +127,7 @@ def __init__(
         # Create ServiceInfo
         self.service_info = ServiceInfo(
             host=self.host,
-            header=header,
+            header=effective_header,
             credentials=Credentials(
                 ak=self.access_key,
                 sk=self.secret_key,

agentkit/toolkit/cli/cli.py

@@ -14,6 +14,8 @@
 
 """AgentKit CLI - Main entry point for AgentKit Starter Toolkit."""
 
+import os
+
 import typer
 from rich.panel import Panel
 from rich.console import Console
@@ -77,6 +79,7 @@ def main(
     ),
 ):
     """AgentKit CLI - Deploy AI agents with ease."""
+    os.environ.setdefault("AGENTKIT_CLIENT_TYPE", "cli")
     # If no subcommand is provided, show logo
     if ctx.invoked_subcommand is None:
         show_logo()

agentkit/utils/ve_sign.py

@@ -16,6 +16,7 @@
 import hashlib
 import hmac
 import os
+import platform
 import requests
 from urllib.parse import quote
 from agentkit.utils.global_config_io import get_global_config_str
@@ -29,6 +30,63 @@
 Scheme = "https"
 
 
+MAX_X_CUSTOM_SOURCE_LENGTH = 256
+
+
+def _get_os_tag() -> str:
+    system = platform.system().lower()
+    if "linux" in system:
+        return "linux"
+    if "windows" in system:
+        return "windows"
+    if "darwin" in system or "mac" in system:
+        return "macos"
+    return "unknown"
+
+
+def _get_entry() -> str:
+    value = os.getenv("AGENTKIT_CLIENT_TYPE", "").strip().lower()
+    if value in ("cli", "sdk"):
+        return value
+    return "sdk"
+
+
+def _get_sdk_version() -> str:
+    try:
+        from agentkit.version import VERSION
+
+        if VERSION:
+            return VERSION
+    except Exception:
+        pass
+    return "unknown"
+
+
+def build_x_custom_source_header() -> str:
+    sdk_name = "agentkit-sdk-python"
+    version = _get_sdk_version()
+    entry = _get_entry()
+    os_tag = _get_os_tag()
+    product = f"{sdk_name}/{version}"
+    parts = ["schema=v1", f"entry={entry}", f"os={os_tag}"]
+    inner = "; ".join(parts)
+    value = product
+    if inner:
+        value = f"{product} ({inner})"
+    if len(value) <= MAX_X_CUSTOM_SOURCE_LENGTH:
+        return value
+    if len(product) <= MAX_X_CUSTOM_SOURCE_LENGTH:
+        return product
+    return value[:MAX_X_CUSTOM_SOURCE_LENGTH]
+
+
+def ensure_x_custom_source_header(header: dict | None) -> dict:
+    base = header.copy() if header is not None else {}
+    if "X-Custom-Request-Context" not in base:
+        base["X-Custom-Request-Context"] = build_x_custom_source_header()
+    return base
+
+
 def norm_query(params):
     query = ""
     for key in sorted(params.keys()):
@@ -146,6 +204,7 @@ def request(method, date, query, header, ak, sk, action, body):
             signature,
         )
     )
+    header = ensure_x_custom_source_header(header)
     header = {**header, **sign_result}
     # header = {**header, **{"X-Security-Token": SessionToken}}
     # 第六步：将 Signature 签名写入 HTTP Header 中，并发送 HTTP 请求。
@@ -168,7 +227,7 @@ def ve_request(
     version: str,
     region: str,
     host: str,
-    header: dict = {},
+    header: dict | None = None,
     content_type: str = "application/json",
     scheme: str = "https",
 ):
@@ -199,7 +258,7 @@ def ve_request(
 
     try:
         response_body = request(
-            "POST", now, {}, header, AK, SK, action, json.dumps(request_body)
+            "POST", now, {}, header or {}, AK, SK, action, json.dumps(request_body)
         )
         check_error(response_body)
         return response_body