Add STS session token support for temporary Volcengine credentials

agentkit/auth/_openapi.py

@@ -93,7 +93,7 @@ def __init__(
             harden_default_ssl_context()
         self.ak = access_key or os.getenv("VOLCENGINE_ACCESS_KEY") or os.getenv("VOLC_ACCESSKEY")
         self.sk = secret_key or os.getenv("VOLCENGINE_SECRET_KEY") or os.getenv("VOLC_SECRETKEY")
-        self.token = session_token or os.getenv("VOLCENGINE_SESSION_TOKEN")
+        self.token = session_token or os.getenv("VOLCENGINE_SESSION_TOKEN") or os.getenv("VOLC_SESSIONTOKEN")
         if not (self.ak and self.sk):
             raise AuthError(
                 "admin provisioning needs Volcengine credentials.",

agentkit/auth/admin.py

@@ -381,7 +381,7 @@ def publish_discovery(
         ) from exc
     ak = access_key or _os.getenv("VOLCENGINE_ACCESS_KEY")
     sk = secret_key or _os.getenv("VOLCENGINE_SECRET_KEY")
-    token = session_token or _os.getenv("VOLCENGINE_SESSION_TOKEN")
+    token = session_token or _os.getenv("VOLCENGINE_SESSION_TOKEN") or _os.getenv("VOLC_SESSIONTOKEN")
     endpoint = f"tos-{coords.region}.volces.com"
     client = tos.TosClientV2(ak, sk, endpoint, coords.region, security_token=token)
     try:

agentkit/client/base_service_client.py

@@ -123,6 +123,7 @@ def __init__(
             service=service,
             explicit_access_key=access_key or None,
             explicit_secret_key=secret_key or None,
+            explicit_session_token=session_token or None,
             platform_config=platform_config,
         )
         self._credential_source = getattr(creds, "source", None)

agentkit/platform/__init__.py

@@ -14,6 +14,7 @@
 
 from __future__ import annotations
 
+import os
 from typing import Optional
 
 from agentkit.platform.configuration import VolcConfiguration, Endpoint, Credentials
@@ -80,16 +81,23 @@ def resolve_credentials(
     *,
     explicit_access_key: Optional[str] = None,
     explicit_secret_key: Optional[str] = None,
+    explicit_session_token: Optional[str] = None,
     platform_config: Optional[VolcConfiguration] = None,
 ) -> Credentials:
     """
     Resolves credentials for a service.
     """
     # 1. Explicit args take absolute precedence
     if explicit_access_key and explicit_secret_key:
+        token = (
+            explicit_session_token
+            or os.getenv("VOLCENGINE_SESSION_TOKEN")
+            or os.getenv("VOLC_SESSIONTOKEN")
+        )
         return Credentials(
             access_key=explicit_access_key,
             secret_key=explicit_secret_key,
+            session_token=token or None,
             source="explicit_args",
         )
 

agentkit/platform/configuration.py

@@ -424,12 +424,17 @@ def _get_config_file_credentials(self) -> Optional[Credentials]:
         if self._provider == CloudProvider.BYTEPLUS:
             gc_ak = get_global_config_str("byteplus", "access_key")
             gc_sk = get_global_config_str("byteplus", "secret_key")
+            gc_token = get_global_config_str("byteplus", "session_token")
         else:
             gc_ak = get_global_config_str("volcengine", "access_key")
             gc_sk = get_global_config_str("volcengine", "secret_key")
+            gc_token = get_global_config_str("volcengine", "session_token")
         if gc_ak and gc_sk:
             return Credentials(
-                access_key=gc_ak, secret_key=gc_sk, source="global_config"
+                access_key=gc_ak,
+                secret_key=gc_sk,
+                session_token=gc_token or None,
+                source="global_config",
             )
         return None
 

agentkit/toolkit/cli/cli_deploy.py

@@ -53,6 +53,11 @@ def deploy_command(
     volcengine_secret_key: Optional[str] = typer.Option(
         None, "--volcengine-secret-key", help="Volcengine secret key (harness deploy)."
     ),
+    volcengine_session_token: Optional[str] = typer.Option(
+        None,
+        "--volcengine-session-token",
+        help="Volcengine session token (STS, for temporary credentials).",
+    ),
     discovery_url: Optional[str] = typer.Option(
         None,
         "--discovery-url",
@@ -81,6 +86,7 @@ def deploy_command(
             region=region,
             access_key=volcengine_access_key,
             secret_key=volcengine_secret_key,
+            session_token=volcengine_session_token,
             discovery_url=discovery_url,
             allowed_id=allowed_id,
             assume_yes=yes,
@@ -123,6 +129,7 @@ def _deploy_harness(
     region,
     access_key,
     secret_key,
+    session_token,
     discovery_url,
     allowed_id,
     assume_yes: bool = False,
@@ -157,6 +164,7 @@ def _deploy_harness(
             region=region,
             access_key=access_key,
             secret_key=secret_key,
+            session_token=session_token,
             discovery_url=discovery_url,
             allowed_id=allowed_id,
             reporter=reporter,

agentkit/toolkit/config/global_config.py

@@ -46,18 +46,21 @@ class VolcengineCredentials:
 
     access_key: str = ""
     secret_key: str = ""
+    session_token: str = ""
 
     def to_dict(self):
         return {
             "access_key": self.access_key,
             "secret_key": self.secret_key,
+            "session_token": self.session_token,
         }
 
     @classmethod
     def from_dict(cls, data: dict):
         return cls(
             access_key=data.get("access_key", ""),
             secret_key=data.get("secret_key", ""),
+            session_token=data.get("session_token", ""),
         )
 
 

agentkit/toolkit/harness/deploy.py

@@ -196,6 +196,7 @@ def deploy_harness(
     region: Optional[str] = None,
     access_key: Optional[str] = None,
     secret_key: Optional[str] = None,
+    session_token: Optional[str] = None,
     discovery_url: Optional[str] = None,
     allowed_id: Optional[str] = None,
     reporter: Optional[Reporter] = None,
@@ -226,7 +227,8 @@ def deploy_harness(
         name: Harness name; locates ``<name>.harness.json`` and names the runtime.
         path: Directory containing the spec and Dockerfile (default: cwd).
         region: AgentKit region (default ``cn-beijing`` or ``VOLCENGINE_REGION``).
-        access_key / secret_key: Volcengine credentials (default: ``VOLCENGINE_*`` env).
+        access_key / secret_key / session_token: Volcengine credentials
+            (default: ``VOLCENGINE_*`` env; session_token is for STS temporary creds).
         discovery_url / allowed_id: OAuth2/JWT overrides for the spec ``auth`` block.
         reporter: Progress reporter forwarded to the launch (default: silent).
         on_conflict: Callback consulted when a single same-name harness exists;
@@ -264,13 +266,20 @@ def deploy_harness(
     auth = _resolve_auth(spec.get("auth"), discovery_url, allowed_id)
 
     # AgentKit authenticates via the Volcengine SDK, which reads VOLC_ACCESSKEY /
-    # VOLC_SECRETKEY from the environment. Mirror whatever AK/SK was passed (or
-    # already set as VOLCENGINE_*) into those names.
+    # VOLC_SECRETKEY / VOLC_SESSIONTOKEN from the environment. Mirror whatever
+    # AK/SK/token was passed (or already set as VOLCENGINE_*) into those names.
     ak = access_key or os.getenv("VOLCENGINE_ACCESS_KEY", "")
     sk = secret_key or os.getenv("VOLCENGINE_SECRET_KEY", "")
+    token = (
+        session_token
+        or os.getenv("VOLCENGINE_SESSION_TOKEN", "")
+        or os.getenv("VOLC_SESSIONTOKEN", "")
+    )
     if ak and sk:
         os.environ["VOLC_ACCESSKEY"] = ak
         os.environ["VOLC_SECRETKEY"] = sk
+        if token:
+            os.environ["VOLC_SESSIONTOKEN"] = token
     if not os.getenv("VOLC_ACCESSKEY") or not os.getenv("VOLC_SECRETKEY"):
         raise ValueError(
             "Volcengine credentials are required. Pass access_key / secret_key, "
@@ -282,7 +291,7 @@ def deploy_harness(
     # Resolve a name collision into a deploy mode. The harness config defaults to
     # `runtime_id: Auto` (create new); an existing same-name harness can instead
     # be updated in place (new version) after confirmation.
-    client = AgentkitRuntimeClient(region=resolved_region)
+    client = AgentkitRuntimeClient(region=resolved_region, session_token=token or "")
     matches = _find_runtimes_by_name(client, runtime_name)
     update_runtime_id = None
     if len(matches) > 1: