feat: support runtime exclusive gateway

agentkit/sdk/runtime/gateway.py

@@ -0,0 +1,90 @@
+# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from __future__ import annotations
+
+from typing import Any, Optional
+
+RUNTIME_GATEWAY_MODES = {"Shared", "Exclusive"}
+
+
+def normalize_runtime_gateway_mode(
+    value: Any,
+    label: str = "runtime.gateway_mode",
+) -> Optional[str]:
+    if value is None or value == "":
+        return None
+    if not isinstance(value, str):
+        raise ValueError(f"{label} must be Shared or Exclusive.")
+
+    normalized = value.strip().lower()
+    if normalized == "shared":
+        return "Shared"
+    if normalized == "exclusive":
+        return "Exclusive"
+    raise ValueError(f"{label} must be Shared or Exclusive.")
+
+
+def effective_runtime_gateway_mode(mode: Optional[str]) -> str:
+    return str(mode).strip() if mode and str(mode).strip() else "Shared"
+
+
+def validate_runtime_gateway_create_config(
+    *,
+    gateway_mode: Optional[str],
+    gateway_instance_id: Optional[str],
+    has_network_configuration: bool = False,
+    label: str = "runtime",
+) -> None:
+    mode = (
+        normalize_runtime_gateway_mode(gateway_mode, f"{label}.gateway_mode")
+        or "Shared"
+    )
+    gateway_instance_id = (gateway_instance_id or "").strip()
+
+    if mode == "Exclusive":
+        if not gateway_instance_id:
+            raise ValueError(
+                f"{label}.gateway_instance_id is required when gateway_mode is Exclusive."
+            )
+        if has_network_configuration:
+            raise ValueError(
+                f"{label}.network cannot be used when gateway_mode is Exclusive; "
+                "the Runtime follows the selected gateway instance network."
+            )
+        return
+
+    if gateway_instance_id:
+        raise ValueError(
+            f"{label}.gateway_instance_id is only valid when gateway_mode is Exclusive."
+        )
+
+
+def runtime_gateway_binding_changed(
+    *,
+    desired_mode: Optional[str],
+    desired_instance_id: Optional[str],
+    current_mode: Optional[str],
+    current_instance_id: Optional[str],
+) -> bool:
+    desired_mode = effective_runtime_gateway_mode(desired_mode)
+    current_mode = effective_runtime_gateway_mode(current_mode)
+
+    if desired_mode not in RUNTIME_GATEWAY_MODES or current_mode not in RUNTIME_GATEWAY_MODES:
+        return desired_mode != current_mode
+    if desired_mode != current_mode:
+        return True
+    if desired_mode != "Exclusive":
+        return False
+    return (desired_instance_id or "").strip() != (current_instance_id or "").strip()

agentkit/sdk/runtime/types.py

@@ -40,6 +40,11 @@ class AgentKitRuntimeVersionsForListRuntimeVersions(RuntimeTypeBaseModel):
     created_at: Optional[str] = Field(default=None, alias="CreatedAt")
     description: Optional[str] = Field(default=None, alias="Description")
     envs: Optional[list[EnvsForListRuntimeVersions]] = Field(default=None, alias="Envs")
+    gateway_instance_id: Optional[str] = Field(default=None, alias="GatewayInstanceId")
+    gateway_instance_name: Optional[str] = Field(
+        default=None, alias="GatewayInstanceName"
+    )
+    gateway_mode: Optional[str] = Field(default=None, alias="GatewayMode")
     memory_mb: Optional[int] = Field(default=None, alias="MemoryMb")
     role_name: Optional[str] = Field(default=None, alias="RoleName")
     runtime_id: Optional[str] = Field(default=None, alias="RuntimeId")
@@ -63,6 +68,11 @@ class AgentKitRuntimesForListRuntimes(RuntimeTypeBaseModel):
     )
     description: Optional[str] = Field(default=None, alias="Description")
     envs: Optional[list[EnvsForListRuntimes]] = Field(default=None, alias="Envs")
+    gateway_instance_id: Optional[str] = Field(default=None, alias="GatewayInstanceId")
+    gateway_instance_name: Optional[str] = Field(
+        default=None, alias="GatewayInstanceName"
+    )
+    gateway_mode: Optional[str] = Field(default=None, alias="GatewayMode")
     knowledge_id: Optional[str] = Field(default=None, alias="KnowledgeId")
     mcp_toolset_id: Optional[str] = Field(default=None, alias="MCPToolsetId")
     memory_id: Optional[str] = Field(default=None, alias="MemoryId")
@@ -326,6 +336,8 @@ class CreateRuntimeRequest(RuntimeTypeBaseModel):
     client_token: Optional[str] = Field(default=None, alias="ClientToken")
     cpu_milli: Optional[int] = Field(default=None, alias="CpuMilli")
     description: Optional[str] = Field(default=None, alias="Description")
+    gateway_instance_id: Optional[str] = Field(default=None, alias="GatewayInstanceId")
+    gateway_mode: Optional[str] = Field(default=None, alias="GatewayMode")
     knowledge_id: Optional[str] = Field(default=None, alias="KnowledgeId")
     mcp_toolset_id: Optional[str] = Field(default=None, alias="MCPToolsetId")
     max_concurrency: Optional[int] = Field(default=None, alias="MaxConcurrency")
@@ -388,6 +400,11 @@ class GetRuntimeResponse(RuntimeTypeBaseModel):
     description: Optional[str] = Field(default=None, alias="Description")
     envs: Optional[list[EnvsForGetRuntime]] = Field(default=None, alias="Envs")
     failed_log_file_url: Optional[str] = Field(default=None, alias="FailedLogFileUrl")
+    gateway_instance_id: Optional[str] = Field(default=None, alias="GatewayInstanceId")
+    gateway_instance_name: Optional[str] = Field(
+        default=None, alias="GatewayInstanceName"
+    )
+    gateway_mode: Optional[str] = Field(default=None, alias="GatewayMode")
     knowledge_id: Optional[str] = Field(default=None, alias="KnowledgeId")
     mcp_toolset_id: Optional[str] = Field(default=None, alias="MCPToolsetId")
     max_concurrency: Optional[int] = Field(default=None, alias="MaxConcurrency")
@@ -455,6 +472,11 @@ class GetRuntimeVersionResponse(RuntimeTypeBaseModel):
     description: Optional[str] = Field(default=None, alias="Description")
     endpoint: Optional[str] = Field(default=None, alias="Endpoint")
     envs: Optional[list[EnvsForGetRuntimeVersion]] = Field(default=None, alias="Envs")
+    gateway_instance_id: Optional[str] = Field(default=None, alias="GatewayInstanceId")
+    gateway_instance_name: Optional[str] = Field(
+        default=None, alias="GatewayInstanceName"
+    )
+    gateway_mode: Optional[str] = Field(default=None, alias="GatewayMode")
     max_concurrency: Optional[int] = Field(default=None, alias="MaxConcurrency")
     memory_mb: Optional[int] = Field(default=None, alias="MemoryMb")
     model_agent_name: Optional[str] = Field(default=None, alias="ModelAgentName")

agentkit/toolkit/cli/cli_config.py

@@ -192,6 +192,18 @@ def config_command(
         "--runtime_binding_mcp_toolset_id",
         help="Bind Runtime to an MCP Toolset ID (cloud/hybrid only)",
     ),
+    runtime_gateway_mode: Optional[str] = typer.Option(
+        None,
+        "--runtime_gateway_mode",
+        "--runtime-gateway-mode",
+        help="Runtime gateway mode (cloud/hybrid, CreateRuntime only): Shared|Exclusive",
+    ),
+    runtime_gateway_instance_id: Optional[str] = typer.Option(
+        None,
+        "--runtime_gateway_instance_id",
+        "--runtime-gateway-instance-id",
+        help="Exclusive Runtime gateway instance ID (cloud/hybrid, CreateRuntime only)",
+    ),
     # Runtime network configuration (advanced, CreateRuntime only)
     runtime_network_mode: Optional[str] = typer.Option(
         None,
@@ -344,6 +356,8 @@ def config_command(
             runtime_vpc_id=runtime_vpc_id,
             runtime_subnet_ids=runtime_subnet_ids,
             runtime_enable_shared_internet_access=runtime_enable_shared_internet_access,
+            runtime_gateway_mode=runtime_gateway_mode,
+            runtime_gateway_instance_id=runtime_gateway_instance_id,
         )
 
         has_cli_params = ConfigParamHandler.has_cli_params(cli_params)

agentkit/toolkit/cli/cli_runtime.py

@@ -25,6 +25,10 @@
 from rich.panel import Panel
 
 from agentkit.sdk.runtime.client import AgentkitRuntimeClient
+from agentkit.sdk.runtime.gateway import (
+    normalize_runtime_gateway_mode,
+    validate_runtime_gateway_create_config,
+)
 from agentkit.sdk.runtime import types as rt
 
 console = Console()
@@ -92,6 +96,60 @@ def _build_network_for_create_runtime(
     )
 
 
+def _has_network_configuration_for_create_runtime(
+    vpc_id: Optional[str],
+    subnet_ids: Optional[str],
+    enable_private_network: bool,
+    enable_public_network: bool,
+    enable_shared_internet_access: bool,
+) -> bool:
+    return (
+        enable_private_network
+        or bool((vpc_id or "").strip())
+        or bool((subnet_ids or "").strip())
+        or enable_shared_internet_access
+        or enable_public_network is False
+    )
+
+
+def _payload_value(payload: object, *keys: str) -> object:
+    if not isinstance(payload, dict):
+        return None
+    for key in keys:
+        if key in payload:
+            return payload[key]
+    return None
+
+
+def _payload_has_non_null_key(payload: object, *keys: str) -> bool:
+    if not isinstance(payload, dict):
+        return False
+    return any(key in payload and payload[key] is not None for key in keys)
+
+
+def _unsupported_runtime_update_fields_message(payload: object) -> Optional[str]:
+    if not isinstance(payload, dict):
+        return None
+    unsupported = [
+        "GatewayMode",
+        "GatewayInstanceId",
+        "NetworkConfiguration",
+        "gateway_mode",
+        "gateway_instance_id",
+        "network_configuration",
+        "gatewayMode",
+        "gatewayInstanceId",
+        "networkConfiguration",
+    ]
+    found = [key for key in unsupported if key in payload]
+    if not found:
+        return None
+    return (
+        f"{', '.join(found)} can only be set when creating a Runtime; "
+        "gateway mode, gateway instance, and network configuration cannot be updated."
+    )
+
+
 def _validate_runtime_create_authorizer_options(
     api_key_name: Optional[str],
     jwt_discovery_url: Optional[str],
@@ -132,6 +190,16 @@ def create_runtime_command(
     model_agent_name: Optional[str] = typer.Option(
         None, "--model-agent-name", help="Model agent name"
     ),
+    gateway_mode: Optional[str] = typer.Option(
+        None,
+        "--gateway-mode",
+        help="Runtime gateway mode: Shared | Exclusive",
+    ),
+    gateway_instance_id: Optional[str] = typer.Option(
+        None,
+        "--gateway-instance-id",
+        help="Exclusive Runtime gateway instance ID",
+    ),
     vpc_id: Optional[str] = typer.Option(None, "--vpc-id", help="VPC ID"),
     subnet_ids: Optional[str] = typer.Option(
         None, "--subnet-ids", help="Subnet IDs (comma-separated)"
@@ -187,12 +255,49 @@ def create_runtime_command(
     try:
         if json_body:
             payload = json.loads(json_body)
+            validate_runtime_gateway_create_config(
+                gateway_mode=_payload_value(
+                    payload,
+                    "GatewayMode",
+                    "gateway_mode",
+                    "gatewayMode",
+                ),
+                gateway_instance_id=_payload_value(
+                    payload,
+                    "GatewayInstanceId",
+                    "gateway_instance_id",
+                    "gatewayInstanceId",
+                ),
+                has_network_configuration=_payload_has_non_null_key(
+                    payload,
+                    "NetworkConfiguration",
+                    "network_configuration",
+                    "networkConfiguration",
+                ),
+                label="runtime",
+            )
             req = rt.CreateRuntimeRequest(**payload)
         else:
             _validate_runtime_create_authorizer_options(
                 api_key_name=api_key_name,
                 jwt_discovery_url=jwt_discovery_url,
             )
+            normalized_gateway_mode = normalize_runtime_gateway_mode(
+                gateway_mode,
+                "--gateway-mode",
+            )
+            validate_runtime_gateway_create_config(
+                gateway_mode=normalized_gateway_mode,
+                gateway_instance_id=gateway_instance_id,
+                has_network_configuration=_has_network_configuration_for_create_runtime(
+                    vpc_id=vpc_id,
+                    subnet_ids=subnet_ids,
+                    enable_private_network=enable_private_network,
+                    enable_public_network=enable_public_network,
+                    enable_shared_internet_access=enable_shared_internet_access,
+                ),
+                label="runtime",
+            )
 
             authorizer = None
             if any([api_key_name, api_key_location, jwt_discovery_url]):
@@ -253,6 +358,10 @@ def create_runtime_command(
                 tool_id=tool_id,
                 mcp_toolset_id=mcp_toolset_id,
                 model_agent_name=model_agent_name,
+                gateway_mode=normalized_gateway_mode,
+                gateway_instance_id=(
+                    gateway_instance_id.strip() if gateway_instance_id else None
+                ),
                 authorizer_configuration=authorizer,
                 network_configuration=network,
                 envs=envs,
@@ -343,6 +452,9 @@ def update_runtime_command(
         client = AgentkitRuntimeClient(region=(region or "").strip())
         if json_body:
             payload = json.loads(json_body)
+            unsupported = _unsupported_runtime_update_fields_message(payload)
+            if unsupported:
+                raise ValueError(unsupported)
             req = rt.UpdateRuntimeRequest(**payload)
         else:
             envs = None
@@ -467,6 +579,16 @@ def list_runtimes_command(
     project_name: Optional[str] = typer.Option(
         None, "--project-name", help="Filter by project name"
     ),
+    gateway_mode: Optional[str] = typer.Option(
+        None,
+        "--gateway-mode",
+        help="Filter by runtime gateway mode: Shared | Exclusive",
+    ),
+    gateway_instance_id: Optional[str] = typer.Option(
+        None,
+        "--gateway-instance-id",
+        help="Filter by runtime gateway instance ID",
+    ),
     # Sorting options
     sort_by: Optional[str] = typer.Option(
         None, "--sort-by", help="Sort by field: Name|Status|CreatedAt|UpdatedAt"
@@ -618,6 +740,26 @@ def list_runtimes_command(
                     rt.FiltersItemForListRuntimes(name="Status", values=statuses)
                 )
 
+        if gateway_mode:
+            filters.append(
+                rt.FiltersItemForListRuntimes(
+                    name="GatewayMode",
+                    values=[
+                        normalize_runtime_gateway_mode(
+                            gateway_mode,
+                            "--gateway-mode",
+                        )
+                    ],
+                )
+            )
+        if gateway_instance_id:
+            filters.append(
+                rt.FiltersItemForListRuntimes(
+                    name="GatewayInstanceId",
+                    values=[gateway_instance_id],
+                )
+            )
+
         # Setup console with color control
         local_console = console if not no_color else Console(no_color=True)
 
@@ -696,6 +838,8 @@ def build_request(next_token_val):
             ("Name", "Name", "white"),
             ("Status", "Status", "green"),
             ("ProjectName", "ProjectName", "yellow"),
+            ("GatewayMode", "GatewayMode", "cyan"),
+            ("GatewayInstanceName", "GatewayInstanceName", "blue"),
             ("UpdatedAt", "UpdatedAt", "magenta"),
         ]