Repository: bentoml/BentoML (language: python)

Issue:
feat: disable compression
Upload is tested and working, but image building isn't, pending money / default build issues.

/cc @ssheng @aarnphm 
fix: read uncompressed bentos
Allows for reading uncompressed tar files, now that we upload those.


Discussion / hints:

Fix patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 9ef49b7ab83..8e613da347a 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -1,23 +1,17 @@
 ci:
   autoupdate_schedule: monthly
-
 exclude: '.*\.(css|js|svg)$'
 repos:
   - repo: https://github.com/charliermarsh/ruff-pre-commit
-    rev: 'v0.0.263'
+    rev: 'v0.0.274'
     hooks:
       - id: ruff
         args: [--fix, --exit-non-zero-on-fix, --show-fixes]
-
   - repo: https://github.com/psf/black
     rev: 23.3.0
     hooks:
-      - id: black
-        additional_dependencies:  # black[jupyter] dependencies
-          - ipython>=7.8.0
-          - tokenize-rt>=3.2.0
+      - id: black-jupyter
         files: '/(src|tests|docs|examples|typings)/'
-
   - repo: https://github.com/pre-commit/pre-commit-hooks
     rev: v4.4.0
     hooks:
diff --git a/pyproject.toml b/pyproject.toml
index 2d3d62872d5..1fdf353e584 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -180,8 +180,6 @@ omit = [
     "src/bentoml/grpc/v1alpha1/*",
     "src/bentoml/grpc/v1/*",
     "src/bentoml/_internal/external_typing/*",
-    "src/bentoml/_internal/yatai_client/*",
-    "src/bentoml/_internal/yatai_rest_api_client/*",
 ]
 
 [tool.coverage.report]
@@ -197,8 +195,6 @@ omit = [
     "src/bentoml/grpc/v1alpha1/*",
     "src/bentoml/grpc/v1/*",
     "src/bentoml/_internal/external_typing/*",
-    "src/bentoml/_internal/yatai_client/*",
-    "src/bentoml/_internal/yatai_rest_api_client/*",
 ]
 exclude_lines = [
     "\\#\\s*pragma: no cover",
diff --git a/src/bentoml/__init__.py b/src/bentoml/__init__.py
index 238f51aa87c..f625ba67fff 100644
--- a/src/bentoml/__init__.py
+++ b/src/bentoml/__init__.py
@@ -49,7 +49,7 @@
 from ._internal.service import Service
 from ._internal.monitoring import monitor
 from ._internal.utils.http import Cookie
-from ._internal.yatai_client import YataiClient
+from ._internal.cloud import YataiClient
 from ._internal.service.loader import load
 
 # Framework specific modules, model management and IO APIs are lazily loaded upon import.
@@ -93,6 +93,7 @@
     from . import exceptions  # BentoML exceptions
     from . import server  # Server API
     from . import monitoring  # Monitoring API
+    from . import cloud  # Cloud API
 
     # isort: on
 else:
@@ -147,6 +148,7 @@
     server = _LazyLoader("bentoml.server", globals(), "bentoml.server")
     exceptions = _LazyLoader("bentoml.exceptions", globals(), "bentoml.exceptions")
     monitoring = _LazyLoader("bentoml.monitoring", globals(), "bentoml.monitoring")
+    cloud = _LazyLoader("bentoml.cloud", globals(), "bentoml.cloud")
 
     del _LazyLoader
 
@@ -212,6 +214,7 @@
     "xgboost",
     # integrations
     "ray",
+    "cloud",
     "triton",
     "monitor",
     "load_config",
diff --git a/src/bentoml/_internal/cloud/__init__.py b/src/bentoml/_internal/cloud/__init__.py
new file mode 100644
index 00000000000..dc1a77ad468
--- /dev/null
+++ b/src/bentoml/_internal/cloud/__init__.py
@@ -0,0 +1,4 @@
+from .base import CloudClient as CloudClient
+from .bentocloud import BentoCloudClient as BentoCloudClient
+from .yatai import YataiClient as YataiClient
+from .deployment import Resource as Resource
diff --git a/src/bentoml/_internal/cloud/base.py b/src/bentoml/_internal/cloud/base.py
new file mode 100644
index 00000000000..6c1c93d5d9b
--- /dev/null
+++ b/src/bentoml/_internal/cloud/base.py
@@ -0,0 +1,163 @@
+from __future__ import annotations
+
+import typing as t
+from abc import ABC
+from abc import abstractmethod
+from functools import wraps
+from contextlib import contextmanager
+
+from rich.panel import Panel
+from rich.console import Group
+from rich.progress import Progress
+from rich.progress import BarColumn
+from rich.progress import TextColumn
+from rich.progress import SpinnerColumn
+from rich.progress import DownloadColumn
+from rich.progress import TimeElapsedColumn
+from rich.progress import TimeRemainingColumn
+from rich.progress import TransferSpeedColumn
+
+from ..tag import Tag
+from ..bento import Bento
+from ..bento import BentoStore
+from ..models import Model
+from ..models import ModelStore
+
+FILE_CHUNK_SIZE = 100 * 1024 * 1024  # 100Mb
+
+
+class ObjectWrapper(object):
+    def __getattr__(self, name: str) -> t.Any:
+        return getattr(self._wrapped, name)
+
+    def __setattr__(self, name: str, value: t.Any) -> None:
+        return setattr(self._wrapped, name, value)
+
+    def wrapper_getattr(self, name: str):
+        """Actual `self.getattr` rather than self._wrapped.getattr"""
+        return getattr(self, name)
+
+    def wrapper_setattr(self, name: str, value: t.Any) -> None:
+        """Actual `self.setattr` rather than self._wrapped.setattr"""
+        return object.__setattr__(self, name, value)
+
+    def __init__(self, wrapped: t.Any):
+        """
+        Thin wrapper around a given object
+        """
+        self.wrapper_setattr("_wrapped", wrapped)
+
+
+class CallbackIOWrapper(ObjectWrapper):
+    def __init__(
+        self,
+        callback: t.Callable[[int], None],
+        stream: t.BinaryIO,
+        method: t.Literal["read", "write"] = "read",
+    ):
+        """
+        Wrap a given `file`-like object's `read()` or `write()` to report
+        lengths to the given `callback`
+        """
+        super().__init__(stream)
+        func = getattr(stream, method)
+        if method == "write":
+
+            @wraps(func)
+            def write(data: t.Union[bytes, bytearray], *args: t.Any, **kwargs: t.Any):
+                res = func(data, *args, **kwargs)
+                callback(len(data))
+                return res
+
+            self.wrapper_setattr("write", write)
+        elif method == "read":
+
+            @wraps(func)
+            def read(*args: t.Any, **kwargs: t.Any):
+                data = func(*args, **kwargs)
+                callback(len(data))
+                return data
+
+            self.wrapper_setattr("read", read)
+        else:
+            raise KeyError("Can only wrap read/write methods")
+
+
+class CloudClient(ABC):
+    log_progress = Progress(TextColumn("{task.description}"))
+
+    spinner_progress = Progress(
+        TextColumn("  "),
+        TimeElapsedColumn(),
+        TextColumn("[bold purple]{task.fields[action]}"),
+        SpinnerColumn("simpleDots"),
+    )
+
+    transmission_progress = Progress(
+        TextColumn("[bold blue]{task.description}", justify="right"),
+        BarColumn(bar_width=None),
+        "[progress.percentage]{task.percentage:>3.1f}%",
+        "•",
+        DownloadColumn(),
+        "•",
+        TransferSpeedColumn(),
+        "•",
+        TimeRemainingColumn(),
+    )
+
+    progress_group = Group(
+        Panel(Group(log_progress, spinner_progress)), transmission_progress
+    )
+
+    @contextmanager
+    def spin(self, *, text: str):
+        task_id = self.spinner_progress.add_task("", action=text)
+        try:
+            yield
+        finally:
+            self.spinner_progress.stop_task(task_id)
+            self.spinner_progress.update(task_id, visible=False)
+
+    @abstractmethod
+    def push_model(
+        self,
+        model: Model,
+        *,
+        force: bool = False,
+        threads: int = 10,
+        context: str | None = None,
+    ):
+        pass
+
+    @abstractmethod
+    def push_bento(
+        self,
+        bento: Bento,
+        *,
+        force: bool = False,
+        threads: int = 10,
+        context: str | None = None,
+    ):
+        pass
+
+    @abstractmethod
+    def pull_model(
+        self,
+        tag: str | Tag,
+        *,
+        force: bool = False,
+        context: str | None = None,
+        model_store: ModelStore,
+    ) -> Model:
+        pass
+
+    @abstractmethod
+    def pull_bento(
+        self,
+        tag: str | Tag,
+        *,
+        force: bool = False,
+        context: str | None = None,
+        bento_store: BentoStore,
+    ) -> Bento:
+        pass
diff --git a/src/bentoml/_internal/yatai_client/__init__.py b/src/bentoml/_internal/cloud/bentocloud.py
similarity index 88%
rename from src/bentoml/_internal/yatai_client/__init__.py
rename to src/bentoml/_internal/cloud/bentocloud.py
index a9c17142cf3..fe8ab1faed5 100644
--- a/src/bentoml/_internal/yatai_client/__init__.py
+++ b/src/bentoml/_internal/cloud/bentocloud.py
@@ -7,8 +7,6 @@
 import threading
 from pathlib import Path
 from tempfile import NamedTemporaryFile
-from functools import wraps
-from contextlib import contextmanager
 from concurrent.futures import ThreadPoolExecutor
 
 import fs
@@ -16,148 +14,54 @@
 from rich.live import Live
 from simple_di import inject
 from simple_di import Provide
-from rich.panel import Panel
-from rich.console import Group
-from rich.progress import Progress
-from rich.progress import BarColumn
-from rich.progress import TextColumn
-from rich.progress import SpinnerColumn
-from rich.progress import DownloadColumn
-from rich.progress import TimeElapsedColumn
-from rich.progress import TimeRemainingColumn
-from rich.progress import TransferSpeedColumn
 
 from ..tag import Tag
 from ..bento import Bento
 from ..bento import BentoStore
 from ..utils import calc_dir_size
+from .config import get_rest_api_client
 from ..models import Model
 from ..models import copy_model
 from ..models import ModelStore
+from .schemas import BentoApiSchema
+from .schemas import LabelItemSchema
+from .schemas import BentoRunnerSchema
+from .schemas import BentoUploadStatus
+from .schemas import CreateBentoSchema
+from .schemas import CreateModelSchema
+from .schemas import ModelUploadStatus
+from .schemas import UpdateBentoSchema
+from .schemas import CompletePartSchema
+from .schemas import BentoManifestSchema
+from .schemas import ModelManifestSchema
+from .schemas import TransmissionStrategy
+from .schemas import FinishUploadBentoSchema
+from .schemas import FinishUploadModelSchema
+from .schemas import BentoRunnerResourceSchema
+from .schemas import CreateBentoRepositorySchema
+from .schemas import CreateModelRepositorySchema
+from .schemas import CompleteMultipartUploadSchema
+from .schemas import PreSignMultipartUploadUrlSchema
+from .base import CloudClient
+from .base import FILE_CHUNK_SIZE
+from .base import CallbackIOWrapper
+from .deployment import Deployment
 from ...exceptions import NotFound
 from ...exceptions import BentoMLException
 from ..configuration.containers import BentoMLContainer
-from ..yatai_rest_api_client.config import get_yatai_rest_api_client
-from ..yatai_rest_api_client.schemas import BentoApiSchema
-from ..yatai_rest_api_client.schemas import LabelItemSchema
-from ..yatai_rest_api_client.schemas import BentoRunnerSchema
-from ..yatai_rest_api_client.schemas import BentoUploadStatus
-from ..yatai_rest_api_client.schemas import CreateBentoSchema
-from ..yatai_rest_api_client.schemas import CreateModelSchema
-from ..yatai_rest_api_client.schemas import ModelUploadStatus
-from ..yatai_rest_api_client.schemas import UpdateBentoSchema
-from ..yatai_rest_api_client.schemas import CompletePartSchema
-from ..yatai_rest_api_client.schemas import BentoManifestSchema
-from ..yatai_rest_api_client.schemas import ModelManifestSchema
-from ..yatai_rest_api_client.schemas import TransmissionStrategy
-from ..yatai_rest_api_client.schemas import FinishUploadBentoSchema
-from ..yatai_rest_api_client.schemas import FinishUploadModelSchema
-from ..yatai_rest_api_client.schemas import BentoRunnerResourceSchema
-from ..yatai_rest_api_client.schemas import CreateBentoRepositorySchema
-from ..yatai_rest_api_client.schemas import CreateModelRepositorySchema
-from ..yatai_rest_api_client.schemas import CompleteMultipartUploadSchema
-from ..yatai_rest_api_client.schemas import PreSignMultipartUploadUrlSchema
 
 if t.TYPE_CHECKING:
     from concurrent.futures import Future
 
     from rich.progress import TaskID
 
+    from .schemas import BentoWithRepositoryListSchema
+    from .schemas import ModelWithRepositoryListSchema
 
-FILE_CHUNK_SIZE = 100 * 1024 * 1024  # 100Mb
 
-
-class ObjectWrapper(object):
-    def __getattr__(self, name: str) -> t.Any:
-        return getattr(self._wrapped, name)
-
-    def __setattr__(self, name: str, value: t.Any) -> None:
-        return setattr(self._wrapped, name, value)
-
-    def wrapper_getattr(self, name: str):
-        """Actual `self.getattr` rather than self._wrapped.getattr"""
-        return getattr(self, name)
-
-    def wrapper_setattr(self, name: str, value: t.Any) -> None:
-        """Actual `self.setattr` rather than self._wrapped.setattr"""
-        return object.__setattr__(self, name, value)
-
-    def __init__(self, wrapped: t.Any):
-        """
-        Thin wrapper around a given object
-        """
-        self.wrapper_setattr("_wrapped", wrapped)
-
-
-class CallbackIOWrapper(ObjectWrapper):
-    def __init__(
-        self,
-        callback: t.Callable[[int], None],
-        stream: t.BinaryIO,
-        method: t.Literal["read", "write"] = "read",
-    ):
-        """
-        Wrap a given `file`-like object's `read()` or `write()` to report
-        lengths to the given `callback`
-        """
-        super().__init__(stream)
-        func = getattr(stream, method)
-        if method == "write":
-
-            @wraps(func)
-            def write(data: t.Union[bytes, bytearray], *args: t.Any, **kwargs: t.Any):
-                res = func(data, *args, **kwargs)
-                callback(len(data))
-                return res
-
-            self.wrapper_setattr("write", write)
-        elif method == "read":
-
-            @wraps(func)
-            def read(*args: t.Any, **kwargs: t.Any):
-                data = func(*args, **kwargs)
-                callback(len(data))
-                return data
-
-            self.wrapper_setattr("read", read)
-        else:
-            raise KeyError("Can only wrap read/write methods")
-
-
-class YataiClient:
-    log_progress = Progress(TextColumn("{task.description}"))
-
-    spinner_progress = Progress(
-        TextColumn("  "),
-        TimeElapsedColumn(),
-        TextColumn("[bold purple]{task.fields[action]}"),
-        SpinnerColumn("simpleDots"),
-    )
-
-    transmission_progress = Progress(
-        TextColumn("[bold blue]{task.description}", justify="right"),
-        BarColumn(bar_width=None),
-        "[progress.percentage]{task.percentage:>3.1f}%",
-        "•",
-        DownloadColumn(),
-        "•",
-        TransferSpeedColumn(),
-        "•",
-        TimeRemainingColumn(),
-    )
-
-    progress_group = Group(
-        Panel(Group(log_progress, spinner_progress)), transmission_progress
-    )
-
-    @contextmanager
-    def spin(self, *, text: str):
-        task_id = self.spinner_progress.add_task("", action=text)
-        try:
-            yield
-        finally:
-            self.spinner_progress.stop_task(task_id)
-            self.spinner_progress.update(task_id, visible=False)
+class BentoCloudClient(CloudClient):
+    def __init__(self):
+        self.deployment = Deployment()
 
     def push_bento(
         self,
@@ -184,7 +88,7 @@ def _do_push_bento(
         threads: int = 10,
         context: str | None = None,
     ):
-        yatai_rest_client = get_yatai_rest_api_client(context)
+        yatai_rest_client = get_rest_api_client(context)
         name = bento.tag.name
         version = bento.tag.version
         if version is None:
@@ -500,8 +404,8 @@ def pull_bento(
         tag: str | Tag,
         *,
         force: bool = False,
-        bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
         context: str | None = None,
+        bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     ) -> Bento:
         with Live(self.progress_group):
             download_task_id = self.transmission_progress.add_task(
@@ -541,7 +445,7 @@ def _do_pull_bento(
         if version is None:
             raise BentoMLException(f'Bento "{_tag}" version can not be None')
 
-        yatai_rest_client = get_yatai_rest_api_client(context)
+        yatai_rest_client = get_rest_api_client(context)
 
         with self.spin(text=f'Fetching bento "{_tag}"'):
             remote_bento = yatai_rest_client.get_bento(
@@ -680,7 +584,7 @@ def _do_push_model(
         threads: int = 10,
         context: str | None = None,
     ):
-        yatai_rest_client = get_yatai_rest_api_client(context)
+        yatai_rest_client = get_rest_api_client(context)
         name = model.tag.name
         version = model.tag.version
         if version is None:
@@ -943,8 +847,8 @@ def pull_model(
         tag: str | Tag,
         *,
         force: bool = False,
-        model_store: ModelStore = Provide[BentoMLContainer.model_store],
         context: str | None = None,
+        model_store: ModelStore = Provide[BentoMLContainer.model_store],
     ) -> Model:
         with Live(self.progress_group):
             download_task_id = self.transmission_progress.add_task(
@@ -979,7 +883,7 @@ def _do_pull_model(
                 model_store.delete(tag)
         except NotFound:
             pass
-        yatai_rest_client = get_yatai_rest_api_client(context)
+        yatai_rest_client = get_rest_api_client(context)
         _tag = Tag.from_taglike(tag)
         name = _tag.name
         version = _tag.version
@@ -1059,3 +963,27 @@ def _do_pull_model(
                         f'[bold green]Successfully pulled model "{_tag}"'
                     )
                     return model
+
+    def list_bentos(self, context: str | None = None) -> BentoWithRepositoryListSchema:
+        yatai_rest_client = get_rest_api_client(context)
+        res = yatai_rest_client.get_bentos_list()
+        if res is None:
+            raise BentoMLException("List bentos request failed")
+
+        res.items = [
+            bento
+            for bento in sorted(res.items, key=lambda x: x.created_at, reverse=True)
+        ]
+        return res
+
+    def list_models(self, context: str | None = None) -> ModelWithRepositoryListSchema:
+        yatai_rest_client = get_rest_api_client(context)
+        res = yatai_rest_client.get_models_list()
+        if res is None:
+            raise BentoMLException("List models request failed")
+
+        res.items = [
+            model
+            for model in sorted(res.items, key=lambda x: x.created_at, reverse=True)
+        ]
+        return res
diff --git a/src/bentoml/_internal/yatai_rest_api_client/yatai.py b/src/bentoml/_internal/cloud/client.py
similarity index 70%
rename from src/bentoml/_internal/yatai_rest_api_client/yatai.py
rename to src/bentoml/_internal/cloud/client.py
index 1078696b3a6..89d0267076d 100644
--- a/src/bentoml/_internal/yatai_rest_api_client/yatai.py
+++ b/src/bentoml/_internal/cloud/client.py
@@ -1,6 +1,7 @@
+from __future__ import annotations
+
 import io
 import logging
-from typing import Optional
 from urllib.parse import urljoin
 
 import requests
@@ -9,26 +10,34 @@
 from .schemas import BentoSchema
 from .schemas import ModelSchema
 from .schemas import schema_to_json
+from .schemas import DeploymentSchema
 from .schemas import schema_from_json
 from .schemas import CreateBentoSchema
 from .schemas import CreateModelSchema
 from .schemas import UpdateBentoSchema
 from .schemas import OrganizationSchema
+from .schemas import DeploymentListSchema
 from .schemas import BentoRepositorySchema
 from .schemas import ModelRepositorySchema
+from .schemas import CreateDeploymentSchema
+from .schemas import UpdateDeploymentSchema
 from .schemas import FinishUploadBentoSchema
 from .schemas import FinishUploadModelSchema
 from .schemas import CreateBentoRepositorySchema
 from .schemas import CreateModelRepositorySchema
+from .schemas import BentoWithRepositoryListSchema
 from .schemas import CompleteMultipartUploadSchema
+from .schemas import ModelWithRepositoryListSchema
 from .schemas import PreSignMultipartUploadUrlSchema
-from ...exceptions import YataiRESTApiClientError
+from .schemas import ClusterListSchema
+from .schemas import ClusterFullSchema
+from ...exceptions import CloudRESTApiClientError
 from ..configuration import BENTOML_VERSION
 
 logger = logging.getLogger(__name__)
 
 
-class YataiRESTApiClient:
+class RestApiClient:
     def __init__(self, endpoint: str, api_token: str) -> None:
         self.endpoint = endpoint
         self.session = requests.Session()
@@ -49,11 +58,11 @@ def _is_not_found(self, resp: requests.Response) -> bool:
 
     def _check_resp(self, resp: requests.Response) -> None:
         if resp.status_code != 200:
-            raise YataiRESTApiClientError(
+            raise CloudRESTApiClientError(
                 f"request failed with status code {resp.status_code}: {resp.text}"
             )
 
-    def get_current_user(self) -> Optional[UserSchema]:
+    def get_current_user(self) -> UserSchema | None:
         url = urljoin(self.endpoint, "/api/v1/auth/current")
         resp = self.session.get(url)
         if self._is_not_found(resp):
@@ -61,7 +70,7 @@ def get_current_user(self) -> Optional[UserSchema]:
         self._check_resp(resp)
         return schema_from_json(resp.text, UserSchema)
 
-    def get_current_organization(self) -> Optional[OrganizationSchema]:
+    def get_current_organization(self) -> OrganizationSchema | None:
         url = urljoin(self.endpoint, "/api/v1/current_org")
         resp = self.session.get(url)
         if self._is_not_found(resp):
@@ -71,7 +80,7 @@ def get_current_organization(self) -> Optional[OrganizationSchema]:
 
     def get_bento_repository(
         self, bento_repository_name: str
-    ) -> Optional[BentoRepositorySchema]:
+    ) -> BentoRepositorySchema | None:
         url = urljoin(
             self.endpoint, f"/api/v1/bento_repositories/{bento_repository_name}"
         )
@@ -89,9 +98,7 @@ def create_bento_repository(
         self._check_resp(resp)
         return schema_from_json(resp.text, BentoRepositorySchema)
 
-    def get_bento(
-        self, bento_repository_name: str, version: str
-    ) -> Optional[BentoSchema]:
+    def get_bento(self, bento_repository_name: str, version: str) -> BentoSchema | None:
         url = urljoin(
             self.endpoint,
             f"/api/v1/bento_repositories/{bento_repository_name}/bentos/{version}",
@@ -236,7 +243,7 @@ def download_bento(
 
     def get_model_repository(
         self, model_repository_name: str
-    ) -> Optional[ModelRepositorySchema]:
+    ) -> ModelRepositorySchema | None:
         url = urljoin(
             self.endpoint, f"/api/v1/model_repositories/{model_repository_name}"
         )
@@ -254,9 +261,7 @@ def create_model_repository(
         self._check_resp(resp)
         return schema_from_json(resp.text, ModelRepositorySchema)
 
-    def get_model(
-        self, model_repository_name: str, version: str
-    ) -> Optional[ModelSchema]:
+    def get_model(self, model_repository_name: str, version: str) -> ModelSchema | None:
         url = urljoin(
             self.endpoint,
             f"/api/v1/model_repositories/{model_repository_name}/models/{version}",
@@ -387,3 +392,121 @@ def download_model(
         resp = self.session.get(url, stream=True)
         self._check_resp(resp)
         return resp
+
+    def get_bento_repositories_list(
+        self, bento_repository_name: str
+    ) -> BentoWithRepositoryListSchema | None:
+        url = urljoin(self.endpoint, "/api/v1/bento_repositories")
+        resp = self.session.get(url)
+        if self._is_not_found(resp):
+            return None
+        self._check_resp(resp)
+        return schema_from_json(resp.text, BentoWithRepositoryListSchema)
+
+    def get_bentos_list(self) -> BentoWithRepositoryListSchema | None:
+        url = urljoin(self.endpoint, "/api/v1/bentos")
+        resp = self.session.get(url)
+        if self._is_not_found(resp):
+            return None
+        self._check_resp(resp)
+        return schema_from_json(resp.text, BentoWithRepositoryListSchema)
+
+    def get_models_list(self) -> ModelWithRepositoryListSchema | None:
+        url = urljoin(self.endpoint, "/api/v1/models")
+        resp = self.session.get(url)
+        if self._is_not_found(resp):
+            return None
+        self._check_resp(resp)
+        return schema_from_json(resp.text, ModelWithRepositoryListSchema)
+
+    def get_deployment_list(
+        self, cluster_name: str, **params: str | int | None
+    ) -> DeploymentListSchema | None:
+        url = urljoin(self.endpoint, f"/api/v1/clusters/{cluster_name}/deployments")
+        resp = self.session.get(url, params=params)
+        if self._is_not_found(resp):
+            return None
+        self._check_resp(resp)
+        return schema_from_json(resp.text, DeploymentListSchema)
+
+    def create_deployment(
+        self, cluster_name: str, create_schema: CreateDeploymentSchema
+    ) -> DeploymentSchema | None:
+        url = urljoin(self.endpoint, f"/api/v1/clusters/{cluster_name}/deployments")
+        resp = self.session.post(url, data=schema_to_json(create_schema))
+        self._check_resp(resp)
+        return schema_from_json(resp.text, DeploymentSchema)
+
+    def get_deployment(
+        self, cluster_name: str, kube_namespace: str, deployment_name: str
+    ) -> DeploymentSchema | None:
+        url = urljoin(
+            self.endpoint,
+            f"/api/v1/clusters/{cluster_name}/namespaces/{kube_namespace}/deployments/{deployment_name}",
+        )
+        resp = self.session.get(url)
+        if self._is_not_found(resp):
+            return None
+        self._check_resp(resp)
+        return schema_from_json(resp.text, DeploymentSchema)
+
+    def update_deployment(
+        self,
+        cluster_name: str,
+        kube_namespace: str,
+        deployment_name: str,
+        update_schema: UpdateDeploymentSchema,
+    ) -> DeploymentSchema | None:
+        url = urljoin(
+            self.endpoint,
+            f"/api/v1/clusters/{cluster_name}/namespaces/{kube_namespace}/deployments/{deployment_name}",
+        )
+        resp = self.session.patch(url, data=schema_to_json(update_schema))
+        if self._is_not_found(resp):
+            return None
+        self._check_resp(resp)
+        return schema_from_json(resp.text, DeploymentSchema)
+
+    def terminate_deployment(
+        self, cluster_name: str, kube_namespace: str, deployment_name: str
+    ) -> DeploymentSchema | None:
+        url = urljoin(
+            self.endpoint,
+            f"/api/v1/clusters/{cluster_name}/namespaces/{kube_namespace}/deployments/{deployment_name}/terminate",
+        )
+        resp = self.session.post(url)
+        if self._is_not_found(resp):
+            return None
+        self._check_resp(resp)
+        return schema_from_json(resp.text, DeploymentSchema)
+
+    def delete_deployment(
+        self, cluster_name: str, kube_namespace: str, deployment_name: str
+    ) -> DeploymentSchema | None:
+        url = urljoin(
+            self.endpoint,
+            f"/api/v1/clusters/{cluster_name}/namespaces/{kube_namespace}/deployments/{deployment_name}",
+        )
+        resp = self.session.delete(url)
+        if self._is_not_found(resp):
+            return None
+        self._check_resp(resp)
+        return schema_from_json(resp.text, DeploymentSchema)
+
+    def get_cluster_list(
+        self, params: dict[str, str | int] | None = None
+    ) -> ClusterListSchema | None:
+        url = urljoin(self.endpoint, "/api/v1/clusters")
+        resp = self.session.get(url, params=params)
+        if self._is_not_found(resp):
+            return None
+        self._check_resp(resp)
+        return schema_from_json(resp.text, ClusterListSchema)
+
+    def get_cluster(self, cluster_name: str) -> ClusterFullSchema | None:
+        url = urljoin(self.endpoint, f"/api/v1/clusters/{cluster_name}")
+        resp = self.session.get(url)
+        if self._is_not_found(resp):
+            return None
+        self._check_resp(resp)
+        return schema_from_json(resp.text, ClusterFullSchema)
diff --git a/src/bentoml/_internal/cloud/config.py b/src/bentoml/_internal/cloud/config.py
new file mode 100644
index 00000000000..e05d0822650
--- /dev/null
+++ b/src/bentoml/_internal/cloud/config.py
@@ -0,0 +1,129 @@
+from __future__ import annotations
+
+import typing as t
+import logging
+
+import attr
+import yaml
+from ..utils import bentoml_cattr
+from simple_di import inject
+from simple_di import Provide
+
+from .client import RestApiClient
+from ...exceptions import CloudRESTApiClientError
+from ..configuration.containers import BentoMLContainer
+
+logger = logging.getLogger(__name__)
+
+default_context_name = "default"
+
+if t.TYPE_CHECKING:
+    from pathlib import Path
+
+
+@attr.define
+class CloudClientContext:
+    name: str
+    endpoint: str
+    api_token: str
+    email: t.Optional[str] = attr.field(default=None)
+
+    def get_rest_api_client(self) -> RestApiClient:
+        return RestApiClient(self.endpoint, self.api_token)
+
+    def get_email(self) -> str:
+        if not self.email:
+            cli = self.get_rest_api_client()
+            user = cli.get_current_user()
+            if user is None:
+                raise CloudRESTApiClientError(
+                    "Unable to get current user from yatai server"
+                )
+            self.email = user.email
+            add_context(self, ignore_warning=True)
+        return self.email
+
+
+@attr.define
+class CloudClientConfig:
+    contexts: t.List[CloudClientContext] = attr.field(factory=list)
+    current_context_name: str = attr.field(default=default_context_name)
+
+    def get_context(self, context: t.Optional[str]) -> CloudClientContext:
+        for ctx in self.contexts:
+            if ctx.name == context:
+                return ctx
+        raise CloudRESTApiClientError(
+            f"Not found {context} cloud context, please login!"
+        )
+
+    def set_current_context(self, context: str | None) -> CloudClientContext:
+        """Set current context to a new context default."""
+        try:
+            new_context = self.get_context(context)
+        except CloudRESTApiClientError as err:
+            raise err from None
+        new_config = attr.evolve(self, current_context_name=new_context.name)
+        new_config.to_yaml_file()
+        return new_context
+
+    def get_current_context(self) -> CloudClientContext:
+        return self.get_context(self.current_context_name)
+
+    @inject
+    def to_yaml_file(
+        self, cloud_config: Path = Provide[BentoMLContainer.cloud_config]
+    ) -> None:
+        """Converts this config to yaml file at given path object.
+        Note that this method will not return the config object.
+        """
+        with cloud_config.open("w") as f:
+            yaml.dump(bentoml_cattr.unstructure(self), stream=f)
+
+    @classmethod
+    def default_config(cls) -> CloudClientConfig:
+        """Iinitialize a default config and write it to default yaml path."""
+        config = cls(contexts=[], current_context_name=default_context_name)
+        config.to_yaml_file()
+        return config
+
+    @classmethod
+    @inject
+    def get_config(
+        cls, cloud_config: Path = Provide[BentoMLContainer.cloud_config]
+    ) -> CloudClientConfig:
+        if not cloud_config.exists():
+            return cls.default_config()
+        with cloud_config.open("r") as stream:
+            try:
+                yaml_content = yaml.safe_load(stream)
+            except yaml.YAMLError as exc:
+                logger.error(exc)
+                raise
+
+            if not yaml_content:
+                return cls.default_config()
+            return bentoml_cattr.structure(yaml_content, cls)
+
+
+def add_context(context: CloudClientContext, *, ignore_warning: bool = False) -> None:
+    config = CloudClientConfig.get_config()
+    for idx, ctx in enumerate(config.contexts):
+        if ctx.name == context.name:
+            if not ignore_warning:
+                logger.warning("Overriding existing Yatai context config: %s", ctx.name)
+            config.contexts[idx] = context
+            break
+    else:
+        config.contexts.append(context)
+    config.to_yaml_file()
+
+
+def get_context(context: str | None) -> CloudClientContext:
+    return CloudClientConfig.get_config().get_context(context)
+
+
+def get_rest_api_client(context: str | None = None) -> RestApiClient:
+    cfg = CloudClientConfig.get_config()
+    ctx = cfg.get_context(context) if context else cfg.get_current_context()
+    return ctx.get_rest_api_client()
diff --git a/src/bentoml/_internal/cloud/deployment.py b/src/bentoml/_internal/cloud/deployment.py
new file mode 100644
index 00000000000..bf16f1d04b1
--- /dev/null
+++ b/src/bentoml/_internal/cloud/deployment.py
@@ -0,0 +1,555 @@
+from __future__ import annotations
+
+import json
+import typing as t
+import logging
+
+import attr
+
+from ..tag import Tag
+from ..utils import bentoml_cattr
+from ..utils import resolve_user_filepath
+from .config import get_rest_api_client
+from .schemas import DeploymentMode
+from .schemas import DeploymentSchema
+from .schemas import DeploymentListSchema
+from .schemas import DeploymentTargetType
+from .schemas import FullDeploymentSchema
+from .schemas import CreateDeploymentSchema
+from .schemas import DeploymentTargetConfig
+from .schemas import UpdateDeploymentSchema
+from .schemas import DeploymentTargetHPAConf
+from .schemas import DeploymentTargetCanaryRule
+from .schemas import DeploymentTargetRunnerConfig
+from deepmerge.merger import Merger
+from ..utils import first_not_none
+from ...exceptions import BentoMLException
+
+logger = logging.getLogger(__name__)
+
+config_merger = Merger(
+    # merge dicts
+    type_strategies=[(dict, "merge")],
+    # override all other types
+    fallback_strategies=["override"],
+    # override conflicting types
+    type_conflict_strategies=["override"],
+)
+
+
+@attr.define
+class Resource:
+    @classmethod
+    def for_hpa_conf(cls, **kwargs: t.Any) -> DeploymentTargetHPAConf:
+        return bentoml_cattr.structure(kwargs, DeploymentTargetHPAConf)
+
+    @classmethod
+    def for_runner(cls, **kwargs: t.Any) -> DeploymentTargetRunnerConfig:
+        exclusive_api_server_key = {
+            v for v in kwargs if v not in attr.fields_dict(DeploymentTargetRunnerConfig)
+        }
+        return bentoml_cattr.structure(
+            {k: v for k, v in kwargs.items() if k not in exclusive_api_server_key},
+            DeploymentTargetRunnerConfig,
+        )
+
+    @classmethod
+    def for_api_server(cls, **kwargs: t.Any) -> DeploymentTargetConfig:
+        return bentoml_cattr.structure(kwargs, DeploymentTargetConfig)
+
+
+class Deployment:
+    @classmethod
+    def _get_default_kube_namespace(
+        cls,
+        cluster_name: str,
+        context: str | None = None,
+    ) -> str:
+        cloud_rest_client = get_rest_api_client(context)
+        res = cloud_rest_client.get_cluster(cluster_name)
+        if not res:
+            raise BentoMLException("Cannot get default kube namespace")
+        return res.config.default_deployment_kube_namespace
+
+    @classmethod
+    def _get_default_cluster(cls, context: str | None = None) -> str:
+        cloud_rest_client = get_rest_api_client(context)
+        res = cloud_rest_client.get_cluster_list(params={"count": 1})
+        if not res:
+            raise BentoMLException("Failed to get list of clusters.")
+        if not res.items:
+            raise BentoMLException("Cannot get default clusters.")
+        return res.items[0].name
+
+    @classmethod
+    def _create_deployment(
+        cls,
+        create_deployment_schema: CreateDeploymentSchema,
+        context: str | None = None,
+        cluster_name: str | None = None,
+    ) -> DeploymentSchema:
+        cloud_rest_client = get_rest_api_client(context)
+        if cluster_name is None:
+            cluster_name = cls._get_default_cluster(context)
+        if create_deployment_schema.kube_namespace is None:
+            create_deployment_schema.kube_namespace = cls._get_default_kube_namespace(
+                cluster_name, context
+            )
+        for target in create_deployment_schema.targets:
+            if (
+                cloud_rest_client.get_bento(target.bento_repository, target.bento)
+                is None
+            ):
+                raise BentoMLException(
+                    f"Create deployment: {target.bento_repository}:{target.bento} does not exist"
+                )
+        if (
+            cloud_rest_client.get_deployment(
+                cluster_name,
+                create_deployment_schema.kube_namespace,
+                create_deployment_schema.name,
+            )
+            is not None
+        ):
+            raise BentoMLException("Create deployment: Deployment already exists")
+        res = cloud_rest_client.create_deployment(
+            cluster_name, create_deployment_schema
+        )
+        if res is None:
+            raise BentoMLException("Create deployment request failed")
+        logger.debug("Deployment Schema: %s", create_deployment_schema)
+        return res
+
+    @classmethod
+    def _update_deployment(
+        cls,
+        deployment_name: str,
+        update_deployment_schema: UpdateDeploymentSchema,
+        kube_namespace: str | None = None,
+        context: str | None = None,
+        cluster_name: str | None = None,
+    ) -> DeploymentSchema:
+        cloud_rest_client = get_rest_api_client(context)
+        if cluster_name is None:
+            cluster_name = cls._get_default_cluster(context)
+        if kube_namespace is None:
+            kube_namespace = cls._get_default_kube_namespace(cluster_name, context)
+        for target in update_deployment_schema.targets:
+            if (
+                cloud_rest_client.get_bento(target.bento_repository, target.bento)
+                is None
+            ):
+                raise BentoMLException(
+                    f"Update deployment: {target.bento_repository}:{target.bento} does not exist"
+                )
+            cloud_rest_client.get_deployment(
+                cluster_name,
+                kube_namespace,
+                deployment_name,
+            )
+
+        res = cloud_rest_client.update_deployment(
+            cluster_name, kube_namespace, deployment_name, update_deployment_schema
+        )
+        if res is None:
+            raise BentoMLException("Update deployment request failed")
+        logger.debug("%s is created.", deployment_name)
+        logger.debug("Deployment Schema: %s", update_deployment_schema)
+        return res
+
+    @classmethod
+    def update(
+        cls,
+        deployment_name: str,
+        bento: Tag | str | None = None,
+        description: str | None = None,
+        expose_endpoint: bool | None = None,
+        cluster_name: str | None = None,
+        kube_namespace: str | None = None,
+        resource_instance: str | None = None,
+        hpa_conf: DeploymentTargetHPAConf | None = None,
+        runners_config: dict[str, DeploymentTargetRunnerConfig] | None = None,
+        api_server_config: DeploymentTargetConfig | None = None,
+        mode: DeploymentMode | None = None,
+        type: DeploymentTargetType | None = None,
+        context: str | None = None,
+        labels: dict[str, str] | None = None,
+        canary_rules: t.List[DeploymentTargetCanaryRule] | None = None,
+    ) -> DeploymentSchema:
+        if mode is None:
+            mode = DeploymentMode.Function
+        if type is None:
+            type = DeploymentTargetType.STABLE
+
+        if cluster_name is None:
+            cluster_name = cls._get_default_cluster(context)
+        if kube_namespace is None:
+            kube_namespace = cls._get_default_kube_namespace(cluster_name, context)
+
+        base_schema = cls.get(deployment_name, context, cluster_name, kube_namespace)
+        # Deployment target always has length of 1
+        if base_schema.latest_revision is None:
+            raise BentoMLException(
+                f"Deployment {deployment_name} has no latest revision"
+            )
+        deployment_target = base_schema.latest_revision.targets[0]
+
+        if bento is None:
+            # NOTE: bento.repository.name is the bento.name, and bento.name is the bento.version
+            # from bentocloud to bentoml.Tag concept
+            bento = f"{deployment_target.bento.repository.name}:{deployment_target.bento.name}"
+        bento = Tag.from_taglike(bento)
+
+        updated_config = bentoml_cattr.unstructure(deployment_target.config)
+        if hpa_conf is not None:
+            hpa_conf_dct = bentoml_cattr.unstructure(hpa_conf)
+            if "hpa_conf" in updated_config:
+                if updated_config["hpa_conf"] is None:
+                    updated_config["hpa_conf"] = {}
+                config_merger.merge(updated_config["hpa_conf"], hpa_conf_dct)
+            if "runners" in updated_config and updated_config["runners"] is not None:
+                for _, runner in updated_config["runners"].items():
+                    if runner["hpa_conf"] is None:
+                        runner["hpa_conf"] = {}
+                    config_merger.merge(runner["hpa_conf"], hpa_conf_dct)
+        if resource_instance is not None:
+            updated_config["resource_instance"] = resource_instance
+            if updated_config.get("runners") is not None:
+                for runner in updated_config["runners"].values():
+                    runner["resource_instance"] = resource_instance
+        if expose_endpoint is not None:
+            updated_config["enable_ingress"] = expose_endpoint
+
+        if api_server_config is not None:
+            if runners_config is not None:
+                api_server_config.runners = runners_config
+            config_merger.merge(
+                updated_config, bentoml_cattr.unstructure(api_server_config)
+            )
+        elif runners_config is not None:
+            config_merger.merge(
+                updated_config,
+                {
+                    "runners": {
+                        k: bentoml_cattr.unstructure(v)
+                        for k, v in runners_config.items()
+                    }
+                },
+            )
+
+        dct_update: dict[str, t.Any] = {
+            "mode": first_not_none(mode, base_schema.mode),
+            "labels": first_not_none(
+                [{"key": key, "value": value} for key, value in labels.items()]
+                if labels
+                else None,
+                [bentoml_cattr.unstructure(i) for i in base_schema.labels],
+            ),
+            "description": description,
+        }
+        update_target = {
+            "type": first_not_none(type, deployment_target.type),
+            "bento": first_not_none(bento.version, deployment_target.bento.name),
+            "bento_repository": first_not_none(
+                bento.name, deployment_target.bento.repository.name
+            ),
+            "config": updated_config,
+        }
+
+        rules = first_not_none(
+            [bentoml_cattr.unstructure(i) for i in canary_rules]
+            if canary_rules
+            else None,
+            [bentoml_cattr.unstructure(i) for i in deployment_target.canary_rules]
+            if deployment_target.canary_rules
+            else None,
+        )
+        if rules:
+            update_target["canary_rules"] = rules
+
+        # update the target
+        dct_update["targets"] = [update_target]
+
+        return cls._update_deployment(
+            deployment_name=deployment_name,
+            update_deployment_schema=bentoml_cattr.structure(
+                dct_update, UpdateDeploymentSchema
+            ),
+            context=context,
+            cluster_name=cluster_name,
+            kube_namespace=kube_namespace,
+        )
+
+    @classmethod
+    def create(
+        cls,
+        deployment_name: str,
+        bento: Tag | str,
+        description: str | None = None,
+        expose_endpoint: bool | None = None,
+        cluster_name: str | None = None,
+        kube_namespace: str | None = None,
+        resource_instance: str | None = None,
+        hpa_conf: DeploymentTargetHPAConf | None = None,
+        runners_config: dict[str, DeploymentTargetRunnerConfig] | None = None,
+        api_server_config: DeploymentTargetConfig | None = None,
+        mode: DeploymentMode | None = None,
+        type: DeploymentTargetType | None = None,
+        context: str | None = None,
+        labels: dict[str, str] | None = None,
+        canary_rules: t.List[DeploymentTargetCanaryRule] | None = None,
+    ) -> DeploymentSchema:
+        if mode is None:
+            mode = DeploymentMode.Function
+        if type is None:
+            type = DeploymentTargetType.STABLE
+        bento_tag = Tag.from_taglike(bento)
+
+        dct: dict[str, t.Any] = {
+            "name": deployment_name,
+            "kube_namespace": kube_namespace,
+            "mode": mode,
+            "description": description,
+        }
+        if labels:
+            dct["labels"] = [
+                {"key": key, "value": value} for key, value in labels.items()
+            ]
+        if api_server_config is None:
+            _config: dict[str, t.Any] = {
+                "runners": {
+                    k: bentoml_cattr.unstructure(v) for k, v in runners_config.items()
+                }
+                if runners_config
+                else None,
+            }
+        else:
+            api_server_config.runners = runners_config
+            _config = bentoml_cattr.unstructure(api_server_config)
+
+        create_target: dict[str, t.Any] = {
+            "type": type,
+            "bento_repository": bento_tag.name,
+            "bento": bento_tag.version,
+            "config": _config,
+        }
+        if canary_rules:
+            create_target["canary_rules"] = [
+                bentoml_cattr.unstructure(i) for i in canary_rules
+            ]
+
+        # Only change the value by the top-level param if it is not provided already in api_server_config or runner_config
+        if hpa_conf:
+            hpa_conf_dct = bentoml_cattr.unstructure(hpa_conf)
+            _config_hpa_conf = _config.get("hpa_conf", None)
+            if _config_hpa_conf is None:
+                _config_hpa_conf = {}
+            for k, v in hpa_conf_dct.items():
+                if k not in _config_hpa_conf:
+                    _config_hpa_conf[k] = v
+                else:
+                    logger.warning(
+                        "Key %s is already set in API server config and will not be overwritten with hpa_conf.%s",
+                        k,
+                        k,
+                    )
+            _config["hpa_conf"] = _config_hpa_conf
+            if "runners" in _config and _config["runners"] is not None:
+                _runner_config = _config["runners"]
+                for runner in _runner_config.values():
+                    _runner_hpa_conf = runner.get("hpa_conf", None)
+                    if _runner_hpa_conf is None:
+                        _runner_hpa_conf = {}
+                    for k, v in hpa_conf_dct.items():
+                        if k not in _runner_hpa_conf:
+                            _runner_hpa_conf[k] = v
+                        else:
+                            logger.warning(
+                                "Key %s is already set in runner config and will not be overwritten with hpa_conf.%s",
+                                k,
+                                k,
+                            )
+                    runner["hpa_conf"] = _runner_hpa_conf
+
+        if resource_instance:
+            if "resource_instance" not in _config:
+                _config["resource_instance"] = resource_instance
+            if "runners" in _config and _config["runners"] is not None:
+                for runner in _config["runners"].values():
+                    if "resource_instance" not in runner:
+                        runner["resource_instance"] = resource_instance
+
+        if expose_endpoint is not None and _config.get("enable_ingress", None) is None:
+            _config["enable_ingress"] = expose_endpoint
+
+        # setup the create target itself
+        dct["targets"] = [create_target]
+
+        return cls._create_deployment(
+            context=context,
+            cluster_name=cluster_name,
+            create_deployment_schema=bentoml_cattr.structure(
+                dct, CreateDeploymentSchema
+            ),
+        )
+
+    @classmethod
+    def list(
+        cls,
+        context: str | None = None,
+        cluster_name: str | None = None,
+        query: str | None = None,
+        search: str | None = None,
+        count: int | None = None,
+        start: int | None = None,
+    ) -> DeploymentListSchema:
+        cloud_rest_client = get_rest_api_client(context)
+        if cluster_name is None:
+            cluster_name = cls._get_default_cluster(context)
+        if query or start or count or search:
+            params = {"start": start, "count": count, "search": search, "q": query}
+            res = cloud_rest_client.get_deployment_list(cluster_name, **params)
+            if res is None:
+                raise BentoMLException("List deployments request failed")
+            return res
+        else:
+            all_deployment = cloud_rest_client.get_deployment_list(cluster_name)
+            if all_deployment is None:
+                raise BentoMLException("List deployments request failed")
+            return all_deployment
+
+    @classmethod
+    def create_from_file(
+        cls,
+        path_or_stream: str | t.TextIO,
+        path_context: str | None = None,
+        context: str | None = None,
+    ) -> DeploymentSchema:
+        if isinstance(path_or_stream, str):
+            real_path = resolve_user_filepath(path_or_stream, path_context)
+            try:
+                with open(real_path, "r") as file:
+                    data = json.load(file)
+            except FileNotFoundError:
+                raise ValueError(f"File not found: {real_path}")
+            except json.JSONDecodeError as e:
+                raise ValueError(f"Error decoding JSON file: {real_path}\n{e}")
+            except Exception as e:
+                raise ValueError(
+                    f"An error occurred while reading the file: {real_path}\n{e}"
+                )
+        else:
+            # load the data from trusted stream
+            data = json.load(path_or_stream)
+        deployment_schema = bentoml_cattr.structure(data, FullDeploymentSchema)
+        return cls._create_deployment(
+            create_deployment_schema=deployment_schema,
+            context=context,
+            cluster_name=deployment_schema.cluster_name,
+        )
+
+    @classmethod
+    def update_from_file(
+        cls,
+        path_or_stream: str | t.TextIO,
+        path_context: str | None = None,
+        context: str | None = None,
+    ) -> DeploymentSchema:
+        if isinstance(path_or_stream, str):
+            real_path = resolve_user_filepath(path_or_stream, path_context)
+            try:
+                with open(real_path, "r") as file:
+                    data = json.load(file)
+            except FileNotFoundError:
+                raise ValueError(f"File not found: {real_path}")
+            except json.JSONDecodeError as e:
+                raise ValueError(f"Error decoding JSON file: {real_path}\n{e}")
+            except Exception as e:
+                raise ValueError(
+                    f"An error occurred while reading the file: {real_path}\n{e}"
+                )
+        else:
+            data = json.load(path_or_stream)
+        deployment_schema = bentoml_cattr.structure(data, FullDeploymentSchema)
+        return cls._update_deployment(
+            deployment_name=deployment_schema.name,
+            update_deployment_schema=deployment_schema,
+            context=context,
+            cluster_name=deployment_schema.cluster_name,
+            kube_namespace=deployment_schema.kube_namespace,
+        )
+
+    @classmethod
+    def get(
+        cls,
+        deployment_name: str,
+        context: str | None = None,
+        cluster_name: str | None = None,
+        kube_namespace: str | None = None,
+    ) -> DeploymentSchema:
+        cloud_rest_client = get_rest_api_client(context)
+        if cluster_name is None:
+            cluster_name = cls._get_default_cluster(context)
+        if kube_namespace is None:
+            kube_namespace = cls._get_default_kube_namespace(cluster_name, context)
+        res = cloud_rest_client.get_deployment(
+            cluster_name, kube_namespace, deployment_name
+        )
+        if res is None:
+            raise BentoMLException("Get deployment request failed")
+        return res
+
+    @classmethod
+    def delete(
+        cls,
+        deployment_name: str,
+        context: str | None = None,
+        cluster_name: str | None = None,
+        kube_namespace: str | None = None,
+    ) -> DeploymentSchema:
+        cloud_rest_client = get_rest_api_client(context)
+        if cluster_name is None:
+            cluster_name = cls._get_default_cluster(context)
+        if kube_namespace is None:
+            kube_namespace = cls._get_default_kube_namespace(cluster_name, context)
+        res = cloud_rest_client.get_deployment(
+            cluster_name,
+            kube_namespace,
+            deployment_name,
+        )
+        if res is None:
+            raise BentoMLException("Delete deployment: Deployment does not exist")
+
+        res = cloud_rest_client.delete_deployment(
+            cluster_name, kube_namespace, deployment_name
+        )
+        if res is None:
+            raise BentoMLException("Delete deployment request failed")
+        return res
+
+    @classmethod
+    def terminate(
+        cls,
+        deployment_name: str,
+        context: str | None = None,
+        cluster_name: str | None = None,
+        kube_namespace: str | None = None,
+    ) -> DeploymentSchema:
+        cloud_rest_client = get_rest_api_client(context)
+        if cluster_name is None:
+            cluster_name = cls._get_default_cluster(context)
+        if kube_namespace is None:
+            kube_namespace = cls._get_default_kube_namespace(cluster_name, context)
+        res = cloud_rest_client.get_deployment(
+            cluster_name,
+            kube_namespace,
+            deployment_name,
+        )
+        if res is None:
+            raise BentoMLException("Teminate deployment: Deployment does not exist")
+        res = cloud_rest_client.terminate_deployment(
+            cluster_name, kube_namespace, deployment_name
+        )
+        if res is None:
+            raise BentoMLException("Terminate deployment request failed")
+        return res
diff --git a/src/bentoml/_internal/cloud/schemas.py b/src/bentoml/_internal/cloud/schemas.py
new file mode 100644
index 00000000000..1d1ea3d5674
--- /dev/null
+++ b/src/bentoml/_internal/cloud/schemas.py
@@ -0,0 +1,703 @@
+from __future__ import annotations
+
+import json
+import typing as t
+from enum import Enum
+from typing import TYPE_CHECKING
+from datetime import datetime
+import attr
+import cattr
+from dateutil.parser import parse
+
+time_format = "%Y-%m-%d %H:%M:%S.%f"
+
+
+def datetime_encoder(time_obj: t.Optional[datetime]) -> t.Optional[str]:
+    if not time_obj:
+        return None
+    return time_obj.strftime(time_format)
+
+
+def datetime_decoder(datetime_str: t.Optional[str], _: t.Any) -> t.Optional[datetime]:
+    if not datetime_str:
+        return None
+    return parse(datetime_str)
+
+
+def dict_options_converter(
+    options_type: type[t.Any],
+) -> t.Callable[[type[t.Any] | dict[str, t.Any]], t.Any]:
+    def _converter(value: type[t.Any] | dict[str, t.Any] | None) -> options_type:
+        if value is None:
+            return options_type()
+        if isinstance(value, dict):
+            return options_type(**value)
+        return value
+
+    return _converter
+
+
+cloud_converter = cattr.Converter()
+
+cloud_converter.register_unstructure_hook(datetime, datetime_encoder)
+cloud_converter.register_structure_hook(datetime, datetime_decoder)
+
+
+T = t.TypeVar("T")
+
+
+def schema_from_json(json_content: str, cls: t.Type[T]) -> T:
+    dct = json.loads(json_content)
+    return cloud_converter.structure(dct, cls)
+
+
+def schema_to_json(obj: t.Any) -> str:
+    res = cloud_converter.unstructure(obj, obj.__class__)
+    return json.dumps(res)
+
+
+@attr.define
+class BaseSchema:
+    uid: str
+    created_at: datetime
+    updated_at: t.Optional[datetime]
+    deleted_at: t.Optional[datetime]
+
+
+@attr.define
+class BaseListSchema:
+    start: int
+    count: int
+    total: int
+
+
+class ResourceType(Enum):
+    USER = "user"
+    ORG = "organization"
+    CLUSTER = "cluster"
+    HostCluster = "host_cluster"
+    BENTO_REPOSITORY = "bento_repository"
+    BENTO = "bento"
+    MODEL_REPOSITORY = "model_repository"
+    MODEL = "model"
+    DEPLOYMENT = "deployment"
+    DEPLOYMENT_REVISION = "deployment_revision"
+    TERMINAL_RECORD = "terminal_record"
+    LABEL = "label"
+    API_TOKEN = "api_token"
+    YATAI_COMPONENT = "yatai_component"
+    LimitGroup = "limit_group"
+    ResourceInstance = "resource_instance"
+
+
+@attr.define
+class ResourceSchema(BaseSchema):
+    name: str
+    resource_type: ResourceType
+    labels: t.List[LabelItemSchema]
+
+
+@attr.define
+class UserSchema:
+    name: str
+    email: str
+    first_name: str
+    last_name: str
+
+    def get_name(self) -> str:
+        if not self.first_name and not self.last_name:
+            return self.name
+        return f"{self.first_name} {self.last_name}".strip()
+
+
+@attr.define
+class OrganizationSchema(ResourceSchema):
+    description: str
+
+
+@attr.define
+class OrganizationListSchema(BaseListSchema):
+    items: t.List[OrganizationSchema]
+
+
+@attr.define
+class ClusterSchema(ResourceSchema):
+    description: str
+    creator: UserSchema
+
+
+@attr.define
+class ClusterConfigSchema:
+    default_deployment_kube_namespace: str
+    ingress_ip: str
+    aws: str
+
+
+@attr.define
+class ClusterListSchema(BaseListSchema):
+    items: t.List[ClusterSchema]
+
+
+@attr.define
+class CreateBentoRepositorySchema:
+    name: str
+    description: str
+
+
+class BentoImageBuildStatus(Enum):
+    PENDING = "pending"
+    BUILDING = "building"
+    SUCCESS = "success"
+    FAILED = "failed"
+
+
+class BentoUploadStatus(Enum):
+    PENDING = "pending"
+    BUILDING = "uploading"
+    SUCCESS = "success"
+    FAILED = "failed"
+
+
+@attr.define
+class BentoApiSchema:
+    route: str
+    doc: str
+    input: str
+    output: str
+
+
+@attr.define
+class BentoRunnerResourceSchema:
+    cpu: t.Optional[t.Any]
+    nvidia_gpu: t.Optional[t.Any]
+    custom_resources: t.Optional[t.Any]
+
+
+@attr.define
+class BentoRunnerSchema:
+    name: str
+    runnable_type: t.Optional[str]
+    models: t.Optional[t.List[str]]
+    resource_config: t.Optional[BentoRunnerResourceSchema]
+
+
+@attr.define
+class BentoManifestSchema:
+    service: str
+    bentoml_version: str
+    size_bytes: int
+    apis: t.Dict[str, BentoApiSchema] = attr.field(factory=dict)
+    models: t.List[str] = attr.field(factory=list)
+    runners: t.Optional[t.List[BentoRunnerSchema]] = attr.field(factory=list)
+
+
+if TYPE_CHECKING:
+    TransmissionStrategy = t.Literal["presigned_url", "proxy"]
+else:
+    TransmissionStrategy = str
+
+
+@attr.define
+class BentoSchema(ResourceSchema):
+    description: str
+    version: str
+    image_build_status: BentoImageBuildStatus
+    upload_status: BentoUploadStatus
+    upload_finished_reason: str
+    presigned_upload_url: str
+    presigned_download_url: str
+    manifest: BentoManifestSchema
+    transmission_strategy: t.Optional[TransmissionStrategy] = attr.field(default=None)
+    upload_id: t.Optional[str] = attr.field(default=None)
+
+    upload_started_at: t.Optional[datetime] = attr.field(default=None)
+    upload_finished_at: t.Optional[datetime] = attr.field(default=None)
+    build_at: datetime = attr.field(factory=datetime.now)
+
+
+@attr.define
+class BentoRepositorySchema(ResourceSchema):
+    description: str
+    latest_bento: t.Optional[BentoSchema]
+
+
+@attr.define
+class BentoWithRepositorySchema(BentoSchema):
+    repository: BentoRepositorySchema = attr.field(default=None)
+
+
+@attr.define
+class BentoWithRepositoryListSchema(BaseListSchema):
+    items: t.List[BentoWithRepositorySchema] = attr.field(factory=list)
+
+
+@attr.define
+class CreateBentoSchema:
+    description: str
+    version: str
+    manifest: BentoManifestSchema
+    build_at: datetime = attr.field(factory=datetime.now)
+    labels: t.List[LabelItemSchema] = attr.field(factory=list)
+
+
+@attr.define
+class UpdateBentoSchema:
+    manifest: t.Optional[BentoManifestSchema] = attr.field(default=None)
+    labels: t.Optional[t.List[LabelItemSchema]] = attr.field(default=None)
+
+
+@attr.define
+class BentoFullSchema(BentoWithRepositorySchema):
+    models: t.List[ModelWithRepositorySchema] = attr.field(factory=list)
+
+
+@attr.define
+class PreSignMultipartUploadUrlSchema:
+    upload_id: str
+    part_number: int
+
+
+@attr.define
+class CompletePartSchema:
+    part_number: int
+    etag: str
+
+
+@attr.define
+class CompleteMultipartUploadSchema:
+    parts: t.List[CompletePartSchema]
+    upload_id: str
+
+
+@attr.define
+class FinishUploadBentoSchema:
+    status: t.Optional[BentoUploadStatus]
+    reason: t.Optional[str]
+
+
+@attr.define
+class CreateModelRepositorySchema:
+    name: str
+    description: str
+
+
+class ModelImageBuildStatus(Enum):
+    PENDING = "pending"
+    BUILDING = "building"
+    SUCCESS = "success"
+    FAILED = "failed"
+
+
+class ModelUploadStatus(Enum):
+    PENDING = "pending"
+    BUILDING = "uploading"
+    SUCCESS = "success"
+    FAILED = "failed"
+
+
+@attr.define
+class ModelManifestSchema:
+    module: str
+    api_version: str
+    bentoml_version: str
+    size_bytes: int
+    metadata: t.Dict[str, t.Any] = attr.field(factory=dict)
+    context: t.Dict[str, t.Any] = attr.field(factory=dict)
+    options: t.Dict[str, t.Any] = attr.field(factory=dict)
+
+
+@attr.define
+class ModelSchema(ResourceSchema):
+    description: str
+    version: str
+    image_build_status: ModelImageBuildStatus
+    upload_status: ModelUploadStatus
+    upload_finished_reason: str
+    presigned_upload_url: str
+    presigned_download_url: str
+    manifest: ModelManifestSchema
+
+    transmission_strategy: t.Optional[TransmissionStrategy] = attr.field(default=None)
+    upload_id: t.Optional[str] = attr.field(default=None)
+
+    upload_started_at: t.Optional[datetime] = attr.field(default=None)
+    upload_finished_at: t.Optional[datetime] = attr.field(default=None)
+    build_at: datetime = attr.field(factory=datetime.now)
+
+
+@attr.define
+class ModelRepositorySchema(ResourceSchema):
+    description: str
+    latest_model: t.Optional[ModelSchema]
+
+
+@attr.define
+class ModelWithRepositorySchema(ModelSchema):
+    repository: ModelRepositorySchema = attr.field(default=None)
+
+
+@attr.define
+class ModelWithRepositoryListSchema(BaseListSchema):
+    items: t.List[ModelWithRepositorySchema] = attr.field(factory=list)
+
+
+@attr.define
+class CreateModelSchema:
+    description: str
+    version: str
+    manifest: ModelManifestSchema
+    build_at: datetime = attr.field(factory=datetime.now)
+    labels: t.List[LabelItemSchema] = attr.field(factory=list)
+
+
+@attr.define
+class FinishUploadModelSchema:
+    status: t.Optional[ModelUploadStatus]
+    reason: t.Optional[str]
+
+
+@attr.define
+class BentoRepositoryListSchema(BaseListSchema):
+    items: t.List[BentoRepositorySchema]
+
+
+@attr.define
+class BentoListSchema(BaseListSchema):
+    items: t.List[BentoSchema]
+
+
+class DeploymentTargetCanaryRuleType(Enum):
+    WEIGHT = "weight"
+    HEADER = "header"
+    COOKIE = "cookie"
+
+
+@attr.define
+class DeploymentTargetCanaryRule:
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    type: DeploymentTargetCanaryRuleType
+    weight: int
+    header: str
+    cookie: str
+    header_value: str
+
+
+@attr.define
+class ApiServerBentoDeploymentOverrides:
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    monitorExporter: t.Optional[t.Dict[str, t.Any]] = attr.field(default=None)
+    extraPodMetadata: t.Optional[t.Dict[str, t.Any]] = attr.field(default=None)
+    extraPodSpec: t.Optional[t.Dict[str, t.Any]] = attr.field(default=None)
+
+
+@attr.define
+class RunnerBentoDeploymentOverrides:
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    extraPodMetadata: t.Optional[t.Dict[str, t.Any]] = attr.field(default=None)
+    extraPodSpec: t.Optional[t.Dict[str, t.Any]] = attr.field(default=None)
+
+
+@attr.define
+class BentoRequestOverrides:
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    imageBuildTimeout: int = attr.field(default=None)
+    imageBuilderExtraPodMetadata: t.Optional[t.Dict[str, t.Any]] = attr.field(
+        default=None
+    )
+    imageBuilderExtraPodSpec: t.Optional[t.Dict[str, t.Any]] = attr.field(default=None)
+    imageBuilderExtraContainerEnv: t.Optional[t.List[t.Dict[str, t.Any]]] = attr.field(
+        default=None
+    )
+    imageBuilderContainerResources: t.Optional[t.Dict[str, t.Any]] = attr.field(
+        default=None
+    )
+    dockerConfigJsonSecretName: t.Optional[str] = attr.field(default=None)
+    downloaderContainerEnvFrom: t.Optional[t.Dict[str, t.Any]] = attr.field(
+        default=None
+    )
+
+
+@attr.define
+class LabelItemSchema:
+    key: str
+    value: str
+
+
+class HPAMetricType(Enum):
+    MEMORY = "memory"
+    CPU = "cpu"
+    GPU = "gpu"
+    QPS = "qps"
+
+
+@attr.define
+class HPAMetric:
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    type: HPAMetricType  # enum
+    value: t.Any  # resource.Quantity
+
+
+class HPAScaleBehavior(Enum):
+    DISABLED = "disabled"
+    STABLE = "stable"
+    FAST = "fast"
+
+
+@attr.define
+class HPAPolicy:
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    metrics: t.Optional[t.List[HPAMetric]] = attr.field(default=None)
+    scale_down_behavior: t.Optional[HPAScaleBehavior] = attr.field(default=None)
+    scale_up_behavior: t.Optional[HPAScaleBehavior] = attr.field(default=None)
+
+
+@attr.define
+class DeploymentTargetHPAConf:
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    cpu: t.Optional[int] = attr.field(default=None)
+    gpu: t.Optional[int] = attr.field(default=None)
+    memory: t.Optional[str] = attr.field(default=None)
+    qps: t.Optional[int] = attr.field(default=None)
+    min_replicas: t.Optional[int] = attr.field(default=None)
+    max_replicas: t.Optional[int] = attr.field(default=None)
+    policy: t.Optional[HPAPolicy] = attr.field(default=None)
+
+
+@attr.define
+class DeploymentTargetResourceItem:
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    cpu: t.Optional[str] = attr.field(default=None)
+    memory: t.Optional[str] = attr.field(default=None)
+    gpu: t.Optional[str] = attr.field(default=None)
+    custom: t.Optional[t.Dict[str, str]] = attr.field(default=None)
+
+
+@attr.define
+class DeploymentTargetResources:
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    requests: t.Optional[DeploymentTargetResourceItem] = attr.field(default=None)
+    limits: t.Optional[DeploymentTargetResourceItem] = attr.field(default=None)
+
+
+@attr.define
+class RequestQueueConfig:
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    enabled: t.Optional[bool] = attr.field(default=None)
+    max_consume_concurrency: t.Optional[int] = attr.field(default=None)
+
+
+@attr.define
+class TrafficControlConfig:
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    timeout: t.Optional[str] = attr.field(default=None)
+    request_queue: t.Optional[RequestQueueConfig] = attr.field(default=None)
+
+
+class DeploymentStrategy(Enum):
+    RollingUpdate = "RollingUpdate"
+    Recreate = "Recreate"
+    RampedSlowRollout = "RampedSlowRollout"
+    BestEffortControlledRollout = "BestEffortControlledRollout"
+
+
+@attr.define
+class DeploymentTargetRunnerConfig:
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    resource_instance: t.Optional[str] = attr.field(default=None)
+    resources: t.Optional[DeploymentTargetResources] = attr.field(default=None)
+    hpa_conf: t.Optional[DeploymentTargetHPAConf] = attr.field(default=None)
+    envs: t.Optional[t.List[LabelItemSchema]] = attr.field(default=None)
+    enable_stealing_traffic_debug_mode: t.Optional[bool] = attr.field(default=None)
+    enable_debug_mode: t.Optional[bool] = attr.field(default=None)
+    enable_debug_pod_receive_production_traffic: t.Optional[bool] = attr.field(
+        default=None
+    )
+    deployment_strategy: t.Optional[DeploymentStrategy] = attr.field(default=None)
+    bento_deployment_overrides: t.Optional[RunnerBentoDeploymentOverrides] = attr.field(
+        default=None
+    )
+    traffic_control: t.Optional[TrafficControlConfig] = attr.field(default=None)
+    deployment_cold_start_wait_timeout: t.Optional[int] = attr.field(default=None)
+
+
+class DeploymentTargetType(Enum):
+    STABLE = "stable"
+    CANARY = "canary"
+
+
+@attr.define
+class DeploymentTargetConfig:
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    resources: DeploymentTargetResources = attr.field(
+        default=None, converter=dict_options_converter(DeploymentTargetResources)
+    )
+    kubeResourceUid: str = attr.field(default="")  # empty str
+    kubeResourceVersion: str = attr.field(default="")
+    resource_instance: t.Optional[str] = attr.field(default=None)
+    hpa_conf: t.Optional[DeploymentTargetHPAConf] = attr.field(default=None)
+    envs: t.Optional[t.List[LabelItemSchema]] = attr.field(default=None)
+    runners: t.Optional[t.Dict[str, DeploymentTargetRunnerConfig]] = attr.field(
+        default=None
+    )
+    enable_ingress: t.Optional[bool] = attr.field(default=None)  # false for enables
+    enable_stealing_traffic_debug_mode: t.Optional[bool] = attr.field(default=None)
+    enable_debug_mode: t.Optional[bool] = attr.field(default=None)
+    enable_debug_pod_receive_production_traffic: t.Optional[bool] = attr.field(
+        default=None
+    )
+    deployment_strategy: t.Optional[DeploymentStrategy] = attr.field(
+        default=None
+    )  # Specific
+    bento_deployment_overrides: t.Optional[
+        ApiServerBentoDeploymentOverrides
+    ] = attr.field(default=None)
+    bento_request_overrides: t.Optional[BentoRequestOverrides] = attr.field(
+        default=None
+    )  # Put into image builder
+    traffic_control: t.Optional[TrafficControlConfig] = attr.field(default=None)
+    deployment_cold_start_wait_timeout: t.Optional[int] = attr.field(default=None)
+
+
+@attr.define
+class CreateDeploymentTargetSchema:
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    type: DeploymentTargetType  # stable by default
+    bento_repository: str
+    bento: str
+    config: DeploymentTargetConfig
+    canary_rules: t.Optional[t.List[DeploymentTargetCanaryRule]] = attr.field(
+        default=None
+    )
+
+
+class DeploymentStatus(Enum):
+    Unknown = "unknown"
+    NonDeployed = "non-deployed"
+    Running = "running"
+    Unhealthy = "unhealthy"
+    Failed = "failed"
+    Deploying = "deploying"
+    Terminating = "terminating"
+    Terminated = "terminated"
+    ImageBuilding = "image-building"
+    ImageBuildFailed = "image-build-failed"
+    ImageBuildSucceeded = "image-build-succeeded"
+
+
+@attr.define
+class DeploymentSchema(ResourceSchema):
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    creator: UserSchema
+    cluster: ClusterSchema
+    status: DeploymentStatus
+    kube_namespace: str
+    latest_revision: t.Optional[DeploymentRevisionSchema] = attr.field(
+        default=None
+    )  # Delete returns no latest revision
+    mode: t.Optional[DeploymentMode] = attr.field(default=None)
+
+
+@attr.define
+class DeploymentTargetSchema(ResourceSchema):
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    creator: UserSchema
+    type: DeploymentTargetType
+    bento: BentoFullSchema
+    config: DeploymentTargetConfig
+    canary_rules: t.Optional[t.List[DeploymentTargetCanaryRule]] = attr.field(
+        default=None
+    )
+
+
+class DeploymentRevisionStatus(Enum):
+    ACTIVE = "active"
+    INACTIVE = "inactive"
+
+
+@attr.define
+class DeploymentRevisionSchema(ResourceSchema):
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    creator: UserSchema
+    status: DeploymentRevisionStatus
+    targets: t.List[DeploymentTargetSchema]
+
+
+@attr.define
+class ResourceInstanceConfigSchema:
+    group: str
+    resources: DeploymentTargetResources
+    price: str
+    node_selectors: t.Optional[t.Dict[str, str]] = attr.field(factory=dict)
+
+
+@attr.define
+class ResourceInstanceSchema(ResourceSchema):
+    display_name: str
+    description: str
+    config: ResourceInstanceConfigSchema
+
+
+@attr.define
+class ClusterFullSchema(ClusterSchema):
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    organization: OrganizationSchema
+    kube_config: str
+    config: ClusterConfigSchema
+    grafana_root_path: str
+    resource_instances: t.List[ResourceInstanceSchema]
+
+
+@attr.define
+class DeploymentListSchema(BaseListSchema):
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    items: t.List[DeploymentSchema]
+
+
+class DeploymentMode(Enum):
+    Deployment = "deployment"
+    Function = "function"
+
+
+@attr.define
+class UpdateDeploymentSchema:
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    targets: t.List[CreateDeploymentTargetSchema]
+    mode: t.Optional[DeploymentMode] = attr.field(default=None)
+    labels: t.Optional[t.List[LabelItemSchema]] = attr.field(default=None)
+    description: t.Optional[str] = attr.field(default=None)
+    do_not_deploy: t.Optional[bool] = attr.field(default=None)
+
+
+@attr.define
+class CreateDeploymentSchema(UpdateDeploymentSchema):
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    name: str = attr.field(default=None)
+    kube_namespace: t.Optional[str] = attr.field(default=None)
+
+
+@attr.define
+class FullDeploymentSchema(CreateDeploymentSchema):
+    __omit_if_default__ = True
+    __forbid_extra_keys__ = True
+    cluster_name: t.Optional[str] = attr.field(default=None)
diff --git a/src/bentoml/_internal/cloud/yatai.py b/src/bentoml/_internal/cloud/yatai.py
new file mode 100644
index 00000000000..477c0dee577
--- /dev/null
+++ b/src/bentoml/_internal/cloud/yatai.py
@@ -0,0 +1,958 @@
+from __future__ import annotations
+
+import io
+import typing as t
+import tarfile
+import tempfile
+import threading
+from pathlib import Path
+from tempfile import NamedTemporaryFile
+from concurrent.futures import ThreadPoolExecutor
+
+import fs
+import requests
+from rich.live import Live
+from simple_di import inject
+from simple_di import Provide
+
+from ..tag import Tag
+from ..bento import Bento
+from ..bento import BentoStore
+from ..utils import calc_dir_size
+from .config import get_rest_api_client
+from ..models import Model
+from ..models import copy_model
+from ..models import ModelStore
+from .schemas import BentoApiSchema
+from .schemas import LabelItemSchema
+from .schemas import BentoRunnerSchema
+from .schemas import BentoUploadStatus
+from .schemas import CreateBentoSchema
+from .schemas import CreateModelSchema
+from .schemas import ModelUploadStatus
+from .schemas import UpdateBentoSchema
+from .schemas import CompletePartSchema
+from .schemas import BentoManifestSchema
+from .schemas import ModelManifestSchema
+from .schemas import TransmissionStrategy
+from .schemas import FinishUploadBentoSchema
+from .schemas import FinishUploadModelSchema
+from .schemas import BentoRunnerResourceSchema
+from .schemas import CreateBentoRepositorySchema
+from .schemas import CreateModelRepositorySchema
+from .schemas import CompleteMultipartUploadSchema
+from .schemas import PreSignMultipartUploadUrlSchema
+from .base import CloudClient
+from .base import FILE_CHUNK_SIZE
+from .base import CallbackIOWrapper
+from ...exceptions import NotFound
+from ...exceptions import BentoMLException
+from ..configuration.containers import BentoMLContainer
+
+if t.TYPE_CHECKING:
+    from concurrent.futures import Future
+
+    from rich.progress import TaskID
+
+
+class YataiClient(CloudClient):
+    def push_bento(
+        self,
+        bento: Bento,
+        *,
+        force: bool = False,
+        threads: int = 10,
+        context: str | None = None,
+    ):
+        with Live(self.progress_group):
+            upload_task_id = self.transmission_progress.add_task(
+                f'Pushing Bento "{bento.tag}"', start=False, visible=False
+            )
+            self._do_push_bento(
+                bento, upload_task_id, force=force, threads=threads, context=context
+            )
+
+    def _do_push_bento(
+        self,
+        bento: Bento,
+        upload_task_id: TaskID,
+        *,
+        force: bool = False,
+        threads: int = 10,
+        context: str | None = None,
+    ):
+        yatai_rest_client = get_rest_api_client(context)
+        name = bento.tag.name
+        version = bento.tag.version
+        if version is None:
+            raise BentoMLException(f"Bento {bento.tag} version cannot be None")
+        info = bento.info
+        model_tags = [m.tag for m in info.models]
+        model_store = bento._model_store  # type: ignore
+        models = (model_store.get(name) for name in model_tags)
+        with ThreadPoolExecutor(max_workers=max(len(model_tags), 1)) as executor:
+
+            def push_model(model: Model) -> None:
+                model_upload_task_id = self.transmission_progress.add_task(
+                    f'Pushing model "{model.tag}"', start=False, visible=False
+                )
+                self._do_push_model(
+                    model,
+                    model_upload_task_id,
+                    force=force,
+                    threads=threads,
+                    context=context,
+                )
+
+            futures: t.Iterator[None] = executor.map(push_model, models)
+            list(futures)
+        with self.spin(text=f'Fetching Bento repository "{name}"'):
+            bento_repository = yatai_rest_client.get_bento_repository(
+                bento_repository_name=name
+            )
+        if not bento_repository:
+            with self.spin(text=f'Bento repository "{name}" not found, creating now..'):
+                bento_repository = yatai_rest_client.create_bento_repository(
+                    req=CreateBentoRepositorySchema(name=name, description="")
+                )
+        with self.spin(text=f'Try fetching Bento "{bento.tag}" from Yatai..'):
+            remote_bento = yatai_rest_client.get_bento(
+                bento_repository_name=name, version=version
+            )
+        if (
+            not force
+            and remote_bento
+            and remote_bento.upload_status == BentoUploadStatus.SUCCESS
+        ):
+            self.log_progress.add_task(
+                f'[bold blue]Push failed: Bento "{bento.tag}" already exists in Yatai'
+            )
+            return
+        labels: list[LabelItemSchema] = [
+            LabelItemSchema(key=key, value=value) for key, value in info.labels.items()
+        ]
+        apis: dict[str, BentoApiSchema] = {}
+        models = [str(m.tag) for m in info.models]
+        runners = [
+            BentoRunnerSchema(
+                name=r.name,
+                runnable_type=r.runnable_type,
+                models=r.models,
+                resource_config=BentoRunnerResourceSchema(
+                    cpu=r.resource_config.get("cpu"),
+                    nvidia_gpu=r.resource_config.get("nvidia.com/gpu"),
+                    custom_resources=r.resource_config.get("custom_resources"),
+                )
+                if r.resource_config
+                else None,
+            )
+            for r in info.runners
+        ]
+        manifest = BentoManifestSchema(
+            service=info.service,
+            bentoml_version=info.bentoml_version,
+            apis=apis,
+            models=models,
+            runners=runners,
+            size_bytes=calc_dir_size(bento.path),
+        )
+        if not remote_bento:
+            with self.spin(text=f'Registering Bento "{bento.tag}" with Yatai..'):
+                remote_bento = yatai_rest_client.create_bento(
+                    bento_repository_name=bento_repository.name,
+                    req=CreateBentoSchema(
+                        description="",
+                        version=version,
+                        build_at=info.creation_time,
+                        manifest=manifest,
+                        labels=labels,
+                    ),
+                )
+        else:
+            with self.spin(text=f'Updating Bento "{bento.tag}"..'):
+                remote_bento = yatai_rest_client.update_bento(
+                    bento_repository_name=bento_repository.name,
+                    version=version,
+                    req=UpdateBentoSchema(
+                        manifest=manifest,
+                        labels=labels,
+                    ),
+                )
+
+        transmission_strategy: TransmissionStrategy = "proxy"
+        presigned_upload_url: str | None = None
+
+        if remote_bento.transmission_strategy is not None:
+            transmission_strategy = remote_bento.transmission_strategy
+        else:
+            with self.spin(
+                text=f'Getting a presigned upload url for bento "{bento.tag}" ..'
+            ):
+                remote_bento = yatai_rest_client.presign_bento_upload_url(
+                    bento_repository_name=bento_repository.name, version=version
+                )
+                if remote_bento.presigned_upload_url:
+                    transmission_strategy = "presigned_url"
+                    presigned_upload_url = remote_bento.presigned_upload_url
+
+        with io.BytesIO() as tar_io:
+            with self.spin(text=f'Creating tar archive for bento "{bento.tag}"..'):
+                with tarfile.open(fileobj=tar_io, mode="w:") as tar:
+
+                    def filter_(
+                        tar_info: tarfile.TarInfo,
+                    ) -> t.Optional[tarfile.TarInfo]:
+                        if tar_info.path == "./models" or tar_info.path.startswith(
+                            "./models/"
+                        ):
+                            return None
+                        return tar_info
+
+                    tar.add(bento.path, arcname="./", filter=filter_)
+            tar_io.seek(0, 0)
+
+            with self.spin(text=f'Start uploading bento "{bento.tag}"..'):
+                yatai_rest_client.start_upload_bento(
+                    bento_repository_name=bento_repository.name, version=version
+                )
+
+            file_size = tar_io.getbuffer().nbytes
+
+            self.transmission_progress.update(
+                upload_task_id, completed=0, total=file_size, visible=True
+            )
+            self.transmission_progress.start_task(upload_task_id)
+
+            io_mutex = threading.Lock()
+
+            def io_cb(x: int):
+                with io_mutex:
+                    self.transmission_progress.update(upload_task_id, advance=x)
+
+            wrapped_file = CallbackIOWrapper(io_cb, tar_io, "read")
+
+            if transmission_strategy == "proxy":
+                try:
+                    yatai_rest_client.upload_bento(
+                        bento_repository_name=bento_repository.name,
+                        version=version,
+                        data=wrapped_file,
+                    )
+                except Exception as e:  # pylint: disable=broad-except
+                    self.log_progress.add_task(
+                        f'[bold red]Failed to upload bento "{bento.tag}"'
+                    )
+                    raise e
+                self.log_progress.add_task(
+                    f'[bold green]Successfully pushed bento "{bento.tag}"'
+                )
+                return
+            finish_req = FinishUploadBentoSchema(
+                status=BentoUploadStatus.SUCCESS,
+                reason="",
+            )
+            try:
+                if presigned_upload_url is not None:
+                    resp = requests.put(presigned_upload_url, data=wrapped_file)
+                    if resp.status_code != 200:
+                        finish_req = FinishUploadBentoSchema(
+                            status=BentoUploadStatus.FAILED,
+                            reason=resp.text,
+                        )
+                else:
+                    with self.spin(
+                        text=f'Start multipart uploading Bento "{bento.tag}"...'
+                    ):
+                        remote_bento = yatai_rest_client.start_bento_multipart_upload(
+                            bento_repository_name=bento_repository.name,
+                            version=version,
+                        )
+                        if not remote_bento.upload_id:
+                            raise BentoMLException(
+                                f'Failed to start multipart upload for Bento "{bento.tag}", upload_id is empty'
+                            )
+
+                        upload_id: str = remote_bento.upload_id
+
+                    chunks_count = file_size // FILE_CHUNK_SIZE + 1
+
+                    def chunk_upload(
+                        upload_id: str, chunk_number: int
+                    ) -> FinishUploadBentoSchema | tuple[str, int]:
+                        with self.spin(
+                            text=f'({chunk_number}/{chunks_count}) Presign multipart upload url of Bento "{bento.tag}"...'
+                        ):
+                            remote_bento = (
+                                yatai_rest_client.presign_bento_multipart_upload_url(
+                                    bento_repository_name=bento_repository.name,
+                                    version=version,
+                                    req=PreSignMultipartUploadUrlSchema(
+                                        upload_id=upload_id,
+                                        part_number=chunk_number,
+                                    ),
+                                )
+                            )
+                        with self.spin(
+                            text=f'({chunk_number}/{chunks_count}) Uploading chunk of Bento "{bento.tag}"...'
+                        ):
+                            chunk = (
+                                tar_io.getbuffer()[
+                                    (chunk_number - 1)
+                                    * FILE_CHUNK_SIZE : chunk_number
+                                    * FILE_CHUNK_SIZE
+                                ]
+                                if chunk_number < chunks_count
+                                else tar_io.getbuffer()[
+                                    (chunk_number - 1) * FILE_CHUNK_SIZE :
+                                ]
+                            )
+
+                            with io.BytesIO(chunk) as chunk_io:
+                                wrapped_file = CallbackIOWrapper(
+                                    io_cb, chunk_io, "read"
+                                )
+
+                                resp = requests.put(
+                                    remote_bento.presigned_upload_url, data=wrapped_file
+                                )
+                                if resp.status_code != 200:
+                                    return FinishUploadBentoSchema(
+                                        status=BentoUploadStatus.FAILED,
+                                        reason=resp.text,
+                                    )
+                                return resp.headers["ETag"], chunk_number
+
+                    futures_: list[
+                        Future[FinishUploadBentoSchema | tuple[str, int]]
+                    ] = []
+
+                    with ThreadPoolExecutor(
+                        max_workers=min(max(chunks_count, 1), threads)
+                    ) as executor:
+                        for i in range(1, chunks_count + 1):
+                            future = executor.submit(
+                                chunk_upload,
+                                upload_id,
+                                i,
+                            )
+                            futures_.append(future)
+
+                    parts: list[CompletePartSchema] = []
+
+                    for future in futures_:
+                        result = future.result()
+                        if isinstance(result, FinishUploadBentoSchema):
+                            finish_req = result
+                            break
+                        else:
+                            etag, chunk_number = result
+                            parts.append(
+                                CompletePartSchema(
+                                    part_number=chunk_number,
+                                    etag=etag,
+                                )
+                            )
+
+                    with self.spin(
+                        text=f'Completing multipart upload of Bento "{bento.tag}"...'
+                    ):
+                        remote_bento = (
+                            yatai_rest_client.complete_bento_multipart_upload(
+                                bento_repository_name=bento_repository.name,
+                                version=version,
+                                req=CompleteMultipartUploadSchema(
+                                    upload_id=upload_id,
+                                    parts=parts,
+                                ),
+                            )
+                        )
+
+            except Exception as e:  # pylint: disable=broad-except
+                finish_req = FinishUploadBentoSchema(
+                    status=BentoUploadStatus.FAILED,
+                    reason=str(e),
+                )
+            if finish_req.status is BentoUploadStatus.FAILED:
+                self.log_progress.add_task(
+                    f'[bold red]Failed to upload Bento "{bento.tag}"'
+                )
+            with self.spin(text="Submitting upload status to Yatai"):
+                yatai_rest_client.finish_upload_bento(
+                    bento_repository_name=bento_repository.name,
+                    version=version,
+                    req=finish_req,
+                )
+            if finish_req.status != BentoUploadStatus.SUCCESS:
+                self.log_progress.add_task(
+                    f'[bold red]Failed pushing Bento "{bento.tag}": {finish_req.reason}'
+                )
+            else:
+                self.log_progress.add_task(
+                    f'[bold green]Successfully pushed Bento "{bento.tag}"'
+                )
+
+    @inject
+    def pull_bento(
+        self,
+        tag: str | Tag,
+        *,
+        force: bool = False,
+        bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
+        context: str | None = None,
+    ) -> Bento:
+        with Live(self.progress_group):
+            download_task_id = self.transmission_progress.add_task(
+                f'Pulling bento "{tag}"', start=False, visible=False
+            )
+            return self._do_pull_bento(
+                tag,
+                download_task_id,
+                force=force,
+                bento_store=bento_store,
+                context=context,
+            )
+
+    @inject
+    def _do_pull_bento(
+        self,
+        tag: str | Tag,
+        download_task_id: TaskID,
+        *,
+        force: bool = False,
+        bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
+        context: str | None = None,
+    ) -> Bento:
+        try:
+            bento = bento_store.get(tag)
+            if not force:
+                self.log_progress.add_task(
+                    f'[bold blue]Bento "{tag}" exists in local model store'
+                )
+                return bento
+            bento_store.delete(tag)
+        except NotFound:
+            pass
+        _tag = Tag.from_taglike(tag)
+        name = _tag.name
+        version = _tag.version
+        if version is None:
+            raise BentoMLException(f'Bento "{_tag}" version can not be None')
+
+        yatai_rest_client = get_rest_api_client(context)
+
+        with self.spin(text=f'Fetching bento "{_tag}"'):
+            remote_bento = yatai_rest_client.get_bento(
+                bento_repository_name=name, version=version
+            )
+        if not remote_bento:
+            raise BentoMLException(f'Bento "{_tag}" not found on Yatai')
+
+        with tempfile.TemporaryDirectory() as temp_dir:
+            # Download models to a temporary directory
+            model_store = ModelStore(temp_dir)
+            with ThreadPoolExecutor(
+                max_workers=max(len(remote_bento.manifest.models), 1)
+            ) as executor:
+
+                def pull_model(model_tag: Tag):
+                    model_download_task_id = self.transmission_progress.add_task(
+                        f'Pulling model "{model_tag}"', start=False, visible=False
+                    )
+                    self._do_pull_model(
+                        model_tag,
+                        model_download_task_id,
+                        force=force,
+                        model_store=model_store,
+                        context=context,
+                    )
+
+                futures = executor.map(pull_model, remote_bento.manifest.models)
+                list(futures)
+
+            # Download bento files from yatai
+            transmission_strategy: TransmissionStrategy = "proxy"
+            presigned_download_url: str | None = None
+
+            if remote_bento.transmission_strategy is not None:
+                transmission_strategy = remote_bento.transmission_strategy
+            else:
+                with self.spin(
+                    text=f'Getting a presigned download url for bento "{_tag}"'
+                ):
+                    remote_bento = yatai_rest_client.presign_bento_download_url(
+                        name, version
+                    )
+                    if remote_bento.presigned_download_url:
+                        presigned_download_url = remote_bento.presigned_download_url
+                        transmission_strategy = "presigned_url"
+
+            if transmission_strategy == "proxy":
+                response = yatai_rest_client.download_bento(
+                    bento_repository_name=name,
+                    version=version,
+                )
+            else:
+                if presigned_download_url is None:
+                    with self.spin(
+                        text=f'Getting a presigned download url for bento "{_tag}"'
+                    ):
+                        remote_bento = yatai_rest_client.presign_bento_download_url(
+                            name, version
+                        )
+                        presigned_download_url = remote_bento.presigned_download_url
+                response = requests.get(presigned_download_url, stream=True)
+
+            if response.status_code != 200:
+                raise BentoMLException(
+                    f'Failed to download bento "{_tag}": {response.text}'
+                )
+            total_size_in_bytes = int(response.headers.get("content-length", 0))
+            block_size = 1024  # 1 Kibibyte
+            with NamedTemporaryFile() as tar_file:
+                self.transmission_progress.update(
+                    download_task_id,
+                    completed=0,
+                    total=total_size_in_bytes,
+                    visible=True,
+                )
+                self.transmission_progress.start_task(download_task_id)
+                for data in response.iter_content(block_size):
+                    self.transmission_progress.update(
+                        download_task_id, advance=len(data)
+                    )
+                    tar_file.write(data)
+                self.log_progress.add_task(
+                    f'[bold green]Finished downloading all bento "{_tag}" files'
+                )
+                tar_file.seek(0, 0)
+                tar = tarfile.open(fileobj=tar_file, mode="r")
+                with self.spin(text=f'Extracting bento "{_tag}" tar file'):
+                    with fs.open_fs("temp://") as temp_fs:
+                        for member in tar.getmembers():
+                            f = tar.extractfile(member)
+                            if f is None:
+                                continue
+                            p = Path(member.name)
+                            if p.parent != Path("."):
+                                temp_fs.makedirs(str(p.parent), recreate=True)
+                            temp_fs.writebytes(member.name, f.read())
+                        bento = Bento.from_fs(temp_fs)
+                        for model_tag in remote_bento.manifest.models:
+                            with self.spin(
+                                text=f'Copying model "{model_tag}" to bento'
+                            ):
+                                copy_model(
+                                    model_tag,
+                                    src_model_store=model_store,
+                                    target_model_store=bento._model_store,  # type: ignore
+                                )
+                        bento = bento.save(bento_store)
+                        self.log_progress.add_task(
+                            f'[bold green]Successfully pulled bento "{_tag}"'
+                        )
+                        return bento
+
+    def push_model(
+        self,
+        model: Model,
+        *,
+        force: bool = False,
+        threads: int = 10,
+        context: str | None = None,
+    ):
+        with Live(self.progress_group):
+            upload_task_id = self.transmission_progress.add_task(
+                f'Pushing model "{model.tag}"', start=False, visible=False
+            )
+            self._do_push_model(
+                model, upload_task_id, force=force, threads=threads, context=context
+            )
+
+    def _do_push_model(
+        self,
+        model: Model,
+        upload_task_id: TaskID,
+        *,
+        force: bool = False,
+        threads: int = 10,
+        context: str | None = None,
+    ):
+        yatai_rest_client = get_rest_api_client(context)
+        name = model.tag.name
+        version = model.tag.version
+        if version is None:
+            raise BentoMLException(f'Model "{model.tag}" version cannot be None')
+        info = model.info
+        with self.spin(text=f'Fetching model repository "{name}"'):
+            model_repository = yatai_rest_client.get_model_repository(
+                model_repository_name=name
+            )
+        if not model_repository:
+            with self.spin(text=f'Model repository "{name}" not found, creating now..'):
+                model_repository = yatai_rest_client.create_model_repository(
+                    req=CreateModelRepositorySchema(name=name, description="")
+                )
+        with self.spin(text=f'Try fetching model "{model.tag}" from Yatai..'):
+            remote_model = yatai_rest_client.get_model(
+                model_repository_name=name, version=version
+            )
+        if (
+            not force
+            and remote_model
+            and remote_model.upload_status == ModelUploadStatus.SUCCESS
+        ):
+            self.log_progress.add_task(
+                f'[bold blue]Model "{model.tag}" already exists in Yatai, skipping'
+            )
+            return
+        if not remote_model:
+            labels: list[LabelItemSchema] = [
+                LabelItemSchema(key=key, value=value)
+                for key, value in info.labels.items()
+            ]
+            with self.spin(text=f'Registering model "{model.tag}" with Yatai..'):
+                remote_model = yatai_rest_client.create_model(
+                    model_repository_name=model_repository.name,
+                    req=CreateModelSchema(
+                        description="",
+                        version=version,
+                        build_at=info.creation_time,
+                        manifest=ModelManifestSchema(
+                            module=info.module,
+                            metadata=info.metadata,
+                            context=info.context.to_dict(),
+                            options=info.options.to_dict(),
+                            api_version=info.api_version,
+                            bentoml_version=info.context.bentoml_version,
+                            size_bytes=calc_dir_size(model.path),
+                        ),
+                        labels=labels,
+                    ),
+                )
+
+        transmission_strategy: TransmissionStrategy = "proxy"
+        presigned_upload_url: str | None = None
+
+        if remote_model.transmission_strategy is not None:
+            transmission_strategy = remote_model.transmission_strategy
+        else:
+            with self.spin(
+                text=f'Getting a presigned upload url for Model "{model.tag}" ..'
+            ):
+                remote_model = yatai_rest_client.presign_model_upload_url(
+                    model_repository_name=model_repository.name, version=version
+                )
+                if remote_model.presigned_upload_url:
+                    transmission_strategy = "presigned_url"
+                    presigned_upload_url = remote_model.presigned_upload_url
+
+        with io.BytesIO() as tar_io:
+            with self.spin(text=f'Creating tar archive for model "{model.tag}"..'):
+                with tarfile.open(fileobj=tar_io, mode="w:") as tar:
+                    tar.add(model.path, arcname="./")
+            tar_io.seek(0, 0)
+            with self.spin(text=f'Start uploading model "{model.tag}"..'):
+                yatai_rest_client.start_upload_model(
+                    model_repository_name=model_repository.name, version=version
+                )
+            file_size = tar_io.getbuffer().nbytes
+            self.transmission_progress.update(
+                upload_task_id,
+                description=f'Uploading model "{model.tag}"',
+                total=file_size,
+                visible=True,
+            )
+            self.transmission_progress.start_task(upload_task_id)
+
+            io_mutex = threading.Lock()
+
+            def io_cb(x: int):
+                with io_mutex:
+                    self.transmission_progress.update(upload_task_id, advance=x)
+
+            wrapped_file = CallbackIOWrapper(io_cb, tar_io, "read")
+            if transmission_strategy == "proxy":
+                try:
+                    yatai_rest_client.upload_model(
+                        model_repository_name=model_repository.name,
+                        version=version,
+                        data=wrapped_file,
+                    )
+                except Exception as e:  # pylint: disable=broad-except
+                    self.log_progress.add_task(
+                        f'[bold red]Failed to upload model "{model.tag}"'
+                    )
+                    raise e
+                self.log_progress.add_task(
+                    f'[bold green]Successfully pushed model "{model.tag}"'
+                )
+                return
+            finish_req = FinishUploadModelSchema(
+                status=ModelUploadStatus.SUCCESS,
+                reason="",
+            )
+            try:
+                if presigned_upload_url is not None:
+                    resp = requests.put(presigned_upload_url, data=wrapped_file)
+                    if resp.status_code != 200:
+                        finish_req = FinishUploadModelSchema(
+                            status=ModelUploadStatus.FAILED,
+                            reason=resp.text,
+                        )
+                else:
+                    with self.spin(
+                        text=f'Start multipart uploading Model "{model.tag}"...'
+                    ):
+                        remote_model = yatai_rest_client.start_model_multipart_upload(
+                            model_repository_name=model_repository.name,
+                            version=version,
+                        )
+                        if not remote_model.upload_id:
+                            raise BentoMLException(
+                                f'Failed to start multipart upload for model "{model.tag}", upload_id is empty'
+                            )
+
+                        upload_id: str = remote_model.upload_id
+
+                    chunks_count = file_size // FILE_CHUNK_SIZE + 1
+
+                    def chunk_upload(
+                        upload_id: str, chunk_number: int
+                    ) -> FinishUploadModelSchema | tuple[str, int]:
+                        with self.spin(
+                            text=f'({chunk_number}/{chunks_count}) Presign multipart upload url of model "{model.tag}"...'
+                        ):
+                            remote_model = (
+                                yatai_rest_client.presign_model_multipart_upload_url(
+                                    model_repository_name=model_repository.name,
+                                    version=version,
+                                    req=PreSignMultipartUploadUrlSchema(
+                                        upload_id=upload_id,
+                                        part_number=chunk_number,
+                                    ),
+                                )
+                            )
+
+                        with self.spin(
+                            text=f'({chunk_number}/{chunks_count}) Uploading chunk of model "{model.tag}"...'
+                        ):
+                            chunk = (
+                                tar_io.getbuffer()[
+                                    (chunk_number - 1)
+                                    * FILE_CHUNK_SIZE : chunk_number
+                                    * FILE_CHUNK_SIZE
+                                ]
+                                if chunk_number < chunks_count
+                                else tar_io.getbuffer()[
+                                    (chunk_number - 1) * FILE_CHUNK_SIZE :
+                                ]
+                            )
+
+                            with io.BytesIO(chunk) as chunk_io:
+                                wrapped_file = CallbackIOWrapper(
+                                    io_cb,
+                                    chunk_io,
+                                    "read",
+                                )
+
+                                resp = requests.put(
+                                    remote_model.presigned_upload_url, data=wrapped_file
+                                )
+                                if resp.status_code != 200:
+                                    return FinishUploadModelSchema(
+                                        status=ModelUploadStatus.FAILED,
+                                        reason=resp.text,
+                                    )
+                                return resp.headers["ETag"], chunk_number
+
+                    futures_: list[
+                        Future[FinishUploadModelSchema | tuple[str, int]]
+                    ] = []
+
+                    with ThreadPoolExecutor(
+                        max_workers=min(max(chunks_count, 1), threads)
+                    ) as executor:
+                        for i in range(1, chunks_count + 1):
+                            future = executor.submit(
+                                chunk_upload,
+                                upload_id,
+                                i,
+                            )
+                            futures_.append(future)
+
+                    parts: list[CompletePartSchema] = []
+
+                    for future in futures_:
+                        result = future.result()
+                        if isinstance(result, FinishUploadModelSchema):
+                            finish_req = result
+                            break
+                        else:
+                            etag, chunk_number = result
+                            parts.append(
+                                CompletePartSchema(
+                                    part_number=chunk_number,
+                                    etag=etag,
+                                )
+                            )
+
+                    with self.spin(
+                        text=f'Completing multipart upload of model "{model.tag}"...'
+                    ):
+                        remote_model = (
+                            yatai_rest_client.complete_model_multipart_upload(
+                                model_repository_name=model_repository.name,
+                                version=version,
+                                req=CompleteMultipartUploadSchema(
+                                    upload_id=upload_id,
+                                    parts=parts,
+                                ),
+                            )
+                        )
+
+            except Exception as e:  # pylint: disable=broad-except
+                finish_req = FinishUploadModelSchema(
+                    status=ModelUploadStatus.FAILED,
+                    reason=str(e),
+                )
+            if finish_req.status is ModelUploadStatus.FAILED:
+                self.log_progress.add_task(
+                    f'[bold red]Failed to upload model "{model.tag}"'
+                )
+            with self.spin(text="Submitting upload status to Yatai"):
+                yatai_rest_client.finish_upload_model(
+                    model_repository_name=model_repository.name,
+                    version=version,
+                    req=finish_req,
+                )
+            if finish_req.status != ModelUploadStatus.SUCCESS:
+                self.log_progress.add_task(
+                    f'[bold red]Failed pushing model "{model.tag}" : {finish_req.reason}'
+                )
+            else:
+                self.log_progress.add_task(
+                    f'[bold green]Successfully pushed model "{model.tag}"'
+                )
+
+    @inject
+    def pull_model(
+        self,
+        tag: str | Tag,
+        *,
+        force: bool = False,
+        context: str | None = None,
+        model_store: ModelStore = Provide[BentoMLContainer.model_store],
+    ) -> Model:
+        with Live(self.progress_group):
+            download_task_id = self.transmission_progress.add_task(
+                f'Pulling model "{tag}"', start=False, visible=False
+            )
+            return self._do_pull_model(
+                tag,
+                download_task_id,
+                force=force,
+                model_store=model_store,
+                context=context,
+            )
+
+    @inject
+    def _do_pull_model(
+        self,
+        tag: str | Tag,
+        download_task_id: TaskID,
+        *,
+        force: bool = False,
+        model_store: ModelStore = Provide[BentoMLContainer.model_store],
+        context: str | None = None,
+    ) -> Model:
+        try:
+            model = model_store.get(tag)
+            if not force:
+                self.log_progress.add_task(
+                    f'[bold blue]Model "{tag}" already exists locally, skipping'
+                )
+                return model
+            else:
+                model_store.delete(tag)
+        except NotFound:
+            pass
+        yatai_rest_client = get_rest_api_client(context)
+        _tag = Tag.from_taglike(tag)
+        name = _tag.name
+        version = _tag.version
+        if version is None:
+            raise BentoMLException(f'Model "{_tag}" version cannot be None')
+        with self.spin(text=f'Getting a presigned download url for model "{_tag}"..'):
+            remote_model = yatai_rest_client.presign_model_download_url(name, version)
+
+        if not remote_model:
+            raise BentoMLException(f'Model "{_tag}" not found on Yatai')
+
+        # Download model files from yatai
+        transmission_strategy: TransmissionStrategy = "proxy"
+        presigned_download_url: str | None = None
+
+        if remote_model.transmission_strategy is not None:
+            transmission_strategy = remote_model.transmission_strategy
+        else:
+            with self.spin(text=f'Getting a presigned download url for model "{_tag}"'):
+                remote_model = yatai_rest_client.presign_model_download_url(
+                    name, version
+                )
+                if remote_model.presigned_download_url:
+                    presigned_download_url = remote_model.presigned_download_url
+                    transmission_strategy = "presigned_url"
+
+        if transmission_strategy == "proxy":
+            response = yatai_rest_client.download_model(
+                model_repository_name=name, version=version
+            )
+        else:
+            if presigned_download_url is None:
+                with self.spin(
+                    text=f'Getting a presigned download url for model "{_tag}"'
+                ):
+                    remote_model = yatai_rest_client.presign_model_download_url(
+                        name, version
+                    )
+                    presigned_download_url = remote_model.presigned_download_url
+
+            response = requests.get(presigned_download_url, stream=True)
+            if response.status_code != 200:
+                raise BentoMLException(
+                    f'Failed to download model "{_tag}": {response.text}'
+                )
+
+        total_size_in_bytes = int(response.headers.get("content-length", 0))
+        block_size = 1024  # 1 Kibibyte
+        with NamedTemporaryFile() as tar_file:
+            self.transmission_progress.update(
+                download_task_id,
+                description=f'Downloading model "{_tag}"',
+                total=total_size_in_bytes,
+                visible=True,
+            )
+            self.transmission_progress.start_task(download_task_id)
+            for data in response.iter_content(block_size):
+                self.transmission_progress.update(download_task_id, advance=len(data))
+                tar_file.write(data)
+            self.log_progress.add_task(
+                f'[bold green]Finished downloading model "{_tag}" files'
+            )
+            tar_file.seek(0, 0)
+            tar = tarfile.open(fileobj=tar_file, mode="r")
+            with self.spin(text=f'Extracting model "{_tag}" tar file'):
+                with fs.open_fs("temp://") as temp_fs:
+                    for member in tar.getmembers():
+                        f = tar.extractfile(member)
+                        if f is None:
+                            continue
+                        p = Path(member.name)
+                        if p.parent != Path("."):
+                            temp_fs.makedirs(str(p.parent), recreate=True)
+                        temp_fs.writebytes(member.name, f.read())
+                    model = Model.from_fs(temp_fs).save(model_store)
+                    self.log_progress.add_task(
+                        f'[bold green]Successfully pulled model "{_tag}"'
+                    )
+                    return model
diff --git a/src/bentoml/_internal/configuration/containers.py b/src/bentoml/_internal/configuration/containers.py
index 38b8aeaa1e2..db322de4c18 100644
--- a/src/bentoml/_internal/configuration/containers.py
+++ b/src/bentoml/_internal/configuration/containers.py
@@ -2,6 +2,7 @@
 
 import os
 import math
+from pathlib import Path
 import uuid
 import typing as t
 import logging
@@ -244,6 +245,11 @@ def model_store(base_dir: str = Provide[model_store_dir]) -> "ModelStore":
     def session_id() -> str:
         return uuid.uuid1().hex
 
+    @providers.SingletonFactory
+    @staticmethod
+    def cloud_config(bentoml_home: str = Provide[bentoml_home]) -> Path:
+        return Path(bentoml_home) / ".yatai.yaml"
+
     api_server_config = config.api_server
     runners_config = config.runners
 
@@ -257,10 +263,17 @@ def session_id() -> str:
     @providers.SingletonFactory
     @staticmethod
     def yatai_client():
-        from ..yatai_client import YataiClient
+        from ..cloud.yatai import YataiClient
 
         return YataiClient()
 
+    @providers.SingletonFactory
+    @staticmethod
+    def bentocloud_client():
+        from ..cloud.bentocloud import BentoCloudClient
+
+        return BentoCloudClient()
+
     @providers.SingletonFactory
     @staticmethod
     def serve_info() -> ServeInfo:
diff --git a/src/bentoml/_internal/frameworks/diffusers.py b/src/bentoml/_internal/frameworks/diffusers.py
index a8bef2b4d9e..b1ace4b5588 100644
--- a/src/bentoml/_internal/frameworks/diffusers.py
+++ b/src/bentoml/_internal/frameworks/diffusers.py
@@ -76,7 +76,9 @@ def _str2cls(
     return cls
 
 
-def _extract_commit_hash(resolved_dir: str, regex_commit_hash: t.Pattern[str]) -> str | None:
+def _extract_commit_hash(
+    resolved_dir: str, regex_commit_hash: t.Pattern[str]
+) -> str | None:
     """
     Extracts the commit hash from a resolved filename toward a cache file.
     modified from https://github.com/huggingface/transformers/blob/0b7b4429c78de68acaf72224eb6dae43616d820c/src/transformers/utils/hub.py#L219
@@ -94,7 +96,7 @@ def _extract_commit_hash(resolved_dir: str, regex_commit_hash: t.Pattern[str]) -
 
 def _try_import_huggingface_hub():
     try:
-        import huggingface_hub # noqa: F401
+        import huggingface_hub  # noqa: F401
     except ImportError:  # pragma: no cover
         raise MissingDependencyException(
             "'huggingface_hub' is required in order to download pretrained diffusion models, install with 'pip install huggingface-hub'. For more information, refer to https://huggingface.co/docs/huggingface_hub/quick-start",
@@ -340,7 +342,9 @@ def import_model(
 
     if sync_with_hub_version:
         if tag.version is not None:
-            logger.warn(f"sync_with_hub_version is True, user provided version {tag.version} may be overridden by huggingface hub's commit hash")
+            logger.warn(
+                f"sync_with_hub_version is True, user provided version {tag.version} may be overridden by huggingface hub's commit hash"
+            )
 
     context = ModelContext(
         framework_name="diffusers",
@@ -369,7 +373,9 @@ def import_model(
     if os.path.isdir(model_name_or_path):
         src_dir = model_name_or_path
         if sync_with_hub_version:
-            raise BentoMLException("Cannot sync version with huggingface hub when importing a local model")
+            raise BentoMLException(
+                "Cannot sync version with huggingface hub when importing a local model"
+            )
 
     elif pipeline_class:
         _try_import_huggingface_hub()
@@ -380,6 +386,7 @@ def import_model(
 
         if sync_with_hub_version:
             from huggingface_hub.file_download import REGEX_COMMIT_HASH
+
             version = _extract_commit_hash(src_dir, REGEX_COMMIT_HASH)
             if version is not None:
                 if variant is not None:
@@ -398,11 +405,11 @@ def import_model(
 
         if sync_with_hub_version:
             from huggingface_hub.file_download import REGEX_COMMIT_HASH
+
             version = _extract_commit_hash(src_dir, REGEX_COMMIT_HASH)
             if version is not None:
                 tag.version = version
 
-
     with bentoml.models.create(
         tag,
         module=MODULE_NAME,
@@ -580,7 +587,6 @@ def __init__(self):
 
         @bentoml.Runnable.method(batchable=False)
         def _replace_scheduler(self, scheduler_txt: str):
-
             try:
                 scheduler_cls = _str2cls(scheduler_txt)
                 if isinstance(self.pipeline.scheduler, scheduler_cls):
@@ -599,7 +605,8 @@ def _replace_scheduler(self, scheduler_txt: str):
             except (ModuleNotFoundError, ValueError, AttributeError):
                 logger.info(f"Cannot import {scheduler_txt}")
                 return dict(
-                    success=False, error_message="cannot import scheduler class",
+                    success=False,
+                    error_message="cannot import scheduler class",
                 )
 
     def make_run_method(
diff --git a/src/bentoml/_internal/service/service.py b/src/bentoml/_internal/service/service.py
index 16555764a3b..5a338de6578 100644
--- a/src/bentoml/_internal/service/service.py
+++ b/src/bentoml/_internal/service/service.py
@@ -181,7 +181,7 @@ def load_exported_bento(bento_tag: Tag, content: bytes):
                 return (load, (self.bento.tag,))
             else:
                 # serialization_strategy == REMOTE_BENTO
-                def get_or_pull(bento_tag):
+                def get_or_pull(bento_tag: Tag) -> Service:
                     try:
                         return load(bento_tag)
                     except NotFound:
@@ -191,8 +191,8 @@ def get_or_pull(bento_tag):
                         bento = tmp_bento_store.get(bento_tag)
                         return load_bento_dir(bento.path)
                     except NotFound:
-                        yatai_client = BentoMLContainer.yatai_client.get()
-                        yatai_client.pull_bento(bento_tag, bento_store=tmp_bento_store)
+                        cloud_client = BentoMLContainer.bentocloud_client.get()
+                        cloud_client.pull_bento(bento_tag, bento_store=tmp_bento_store)
                         return get_or_pull(bento_tag)
 
                 return (get_or_pull, (self.bento.tag,))
diff --git a/src/bentoml/_internal/utils/analytics/schemas.py b/src/bentoml/_internal/utils/analytics/schemas.py
index 8ad5e4f7262..a81ec41acd6 100644
--- a/src/bentoml/_internal/utils/analytics/schemas.py
+++ b/src/bentoml/_internal/utils/analytics/schemas.py
@@ -1,9 +1,9 @@
+from __future__ import annotations
 import os
 import re
 import uuid
 import typing as t
 from abc import ABC
-from typing import TYPE_CHECKING
 from datetime import datetime
 from datetime import timezone
 from platform import platform
@@ -18,12 +18,13 @@
 from simple_di import Provide
 
 from ...utils import bentoml_cattr
+from ...cloud.config import CloudClientConfig
 from ...configuration import BENTOML_VERSION
 from ...configuration.containers import BentoMLContainer
-from ...yatai_rest_api_client.config import get_config_path
-from ...yatai_rest_api_client.config import get_current_context
 
-if TYPE_CHECKING:
+if t.TYPE_CHECKING:
+    from pathlib import Path
+
     P = t.ParamSpec("P")
     GenericFunction = t.Callable[P, t.Any]
 
@@ -74,10 +75,13 @@ def get_client_info(
         return new_client_info
 
 
+@inject
 @lru_cache(maxsize=1)
-def get_yatai_user_email() -> t.Optional[str]:
-    if os.path.exists(get_config_path()):
-        return get_current_context().email
+def get_yatai_user_email(
+    cloud_config: Path = Provide[BentoMLContainer.cloud_config],
+) -> str | None:
+    if cloud_config.exists():
+        return CloudClientConfig.get_config(cloud_config).get_current_context().email
 
 
 @lru_cache(maxsize=1)
diff --git a/src/bentoml/_internal/utils/pickle.py b/src/bentoml/_internal/utils/pickle.py
index de7bc4913e8..17890ba668e 100644
--- a/src/bentoml/_internal/utils/pickle.py
+++ b/src/bentoml/_internal/utils/pickle.py
@@ -26,6 +26,7 @@
 # be huge, hence we barely save any time while having a large header
 # (that may cause errors).
 
+
 def pep574_dumps(obj: t.Any) -> tuple[bytes, bytes, list[int]]:
     buffers: list[pickle.PickleBuffer] = []
     main_bytes: bytes = pickle.dumps(obj, protocol=5, buffer_callback=buffers.append)
diff --git a/src/bentoml/_internal/yatai_rest_api_client/config.py b/src/bentoml/_internal/yatai_rest_api_client/config.py
deleted file mode 100644
index b2fac496041..00000000000
--- a/src/bentoml/_internal/yatai_rest_api_client/config.py
+++ /dev/null
@@ -1,122 +0,0 @@
-import os
-import typing as t
-import logging
-from typing import List
-from typing import Optional
-from pathlib import Path
-
-import attr
-import yaml
-import cattr
-from simple_di import inject
-from simple_di import Provide
-
-from bentoml.exceptions import YataiRESTApiClientError
-
-from .yatai import YataiRESTApiClient
-from ..configuration.containers import BentoMLContainer
-
-logger = logging.getLogger(__name__)
-
-default_context_name = "default"
-
-
-@inject
-def get_config_path(bentoml_home: str = Provide[BentoMLContainer.bentoml_home]) -> Path:
-    return Path(bentoml_home) / ".yatai.yaml"
-
-
-@attr.define
-class YataiClientContext:
-    name: str
-    endpoint: str
-    api_token: str
-    email: Optional[str] = attr.field(default=None)
-
-    def get_yatai_rest_api_client(self) -> YataiRESTApiClient:
-        return YataiRESTApiClient(self.endpoint, self.api_token)
-
-    def get_email(self) -> str:
-        if not self.email:
-            cli = self.get_yatai_rest_api_client()
-            user = cli.get_current_user()
-            if user is None:
-                raise YataiRESTApiClientError(
-                    "Unable to get current user from yatai server"
-                )
-            self.email = user.email
-            add_context(self, ignore_warning=True)
-        return self.email
-
-
-@attr.define
-class YataiClientConfig:
-    contexts: List[YataiClientContext] = attr.field(factory=list)
-    current_context_name: str = attr.field(default=default_context_name)
-
-    def get_context(self, context: t.Optional[str]) -> YataiClientContext:
-        for ctx in self.contexts:
-            if ctx.name == context:
-                return ctx
-        raise YataiRESTApiClientError(
-            f"Not found {context} yatai context, please login!"
-        )
-
-    def get_current_context(self) -> YataiClientContext:
-        return self.get_context(self.current_context_name)
-
-
-_config: YataiClientConfig = YataiClientConfig()
-
-
-def store_config(config: YataiClientConfig) -> None:
-    with open(get_config_path(), "w") as f:
-        dct = cattr.unstructure(config)
-        yaml.dump(dct, stream=f)
-
-
-def init_config() -> YataiClientConfig:
-    config = YataiClientConfig(contexts=[], current_context_name=default_context_name)
-    store_config(config)
-    return config
-
-
-def get_config() -> YataiClientConfig:
-    if not os.path.exists(get_config_path()):
-        return init_config()
-    with open(get_config_path(), "r") as f:
-        dct = yaml.safe_load(f)
-        if not dct:
-            return init_config()
-        return cattr.structure(dct, YataiClientConfig)
-
-
-def add_context(context: YataiClientContext, *, ignore_warning: bool = False) -> None:
-    config = get_config()
-    for idx, ctx in enumerate(config.contexts):
-        if ctx.name == context.name:
-            if not ignore_warning:
-                logger.warning("Overriding existing Yatai context config: %s", ctx.name)
-            config.contexts[idx] = context
-            break
-    else:
-        config.contexts.append(context)
-    store_config(config)
-
-
-def get_current_context() -> YataiClientContext:
-    config = get_config()
-    return config.get_current_context()
-
-
-def get_context(context: t.Optional[str]) -> YataiClientContext:
-    config = get_config()
-    return config.get_context(context)
-
-
-def get_yatai_rest_api_client(context: t.Optional[str]) -> YataiRESTApiClient:
-    if context:
-        ctx = get_context(context)
-    else:
-        ctx = get_current_context()
-    return ctx.get_yatai_rest_api_client()
diff --git a/src/bentoml/_internal/yatai_rest_api_client/schemas.py b/src/bentoml/_internal/yatai_rest_api_client/schemas.py
deleted file mode 100644
index 5a7c236c544..00000000000
--- a/src/bentoml/_internal/yatai_rest_api_client/schemas.py
+++ /dev/null
@@ -1,301 +0,0 @@
-import json
-import typing as t
-from enum import Enum
-from typing import TYPE_CHECKING
-from datetime import datetime
-
-import attr
-import cattr
-from dateutil.parser import parse
-
-time_format = "%Y-%m-%d %H:%M:%S.%f"
-
-
-def datetime_encoder(time_obj: t.Optional[datetime]) -> t.Optional[str]:
-    if not time_obj:
-        return None
-    return time_obj.strftime(time_format)
-
-
-def datetime_decoder(datetime_str: t.Optional[str], _: t.Any) -> t.Optional[datetime]:
-    if not datetime_str:
-        return None
-    return parse(datetime_str)
-
-
-converter = cattr.Converter()
-
-converter.register_unstructure_hook(datetime, datetime_encoder)
-converter.register_structure_hook(datetime, datetime_decoder)
-
-
-T = t.TypeVar("T")
-
-
-def schema_from_json(json_content: str, cls: t.Type[T]) -> T:
-    dct = json.loads(json_content)
-    return converter.structure(dct, cls)
-
-
-def schema_to_json(obj: t.Any) -> str:
-    res = converter.unstructure(obj, obj.__class__)
-    return json.dumps(res)
-
-
-@attr.define
-class BaseSchema:
-    uid: str
-    created_at: datetime
-    updated_at: t.Optional[datetime]
-    deleted_at: t.Optional[datetime]
-
-
-@attr.define
-class BaseListSchema:
-    start: int
-    count: int
-    total: int
-
-
-class ResourceType(Enum):
-    USER = "user"
-    ORG = "organization"
-    CLUSTER = "cluster"
-    BENTO_REPOSITORY = "bento_repository"
-    BENTO = "bento"
-    MODEL_REPOSITORY = "model_repository"
-    MODEL = "model"
-
-
-@attr.define
-class ResourceSchema(BaseSchema):
-    name: str
-    resource_type: ResourceType
-
-
-@attr.define
-class LabelItemSchema:
-    key: str
-    value: str
-
-
-@attr.define
-class UserSchema:
-    name: str
-    email: str
-    first_name: str
-    last_name: str
-
-    def get_name(self) -> str:
-        if not self.first_name and not self.last_name:
-            return self.name
-        return f"{self.first_name} {self.last_name}".strip()
-
-
-@attr.define
-class OrganizationSchema(ResourceSchema):
-    description: str
-
-
-@attr.define
-class OrganizationListSchema(BaseListSchema):
-    items: t.List[OrganizationSchema]
-
-
-@attr.define
-class ClusterSchema(ResourceSchema):
-    description: str
-
-
-@attr.define
-class CreateBentoRepositorySchema:
-    name: str
-    description: str
-
-
-class BentoImageBuildStatus(Enum):
-    PENDING = "pending"
-    BUILDING = "building"
-    SUCCESS = "success"
-    FAILED = "failed"
-
-
-class BentoUploadStatus(Enum):
-    PENDING = "pending"
-    BUILDING = "uploading"
-    SUCCESS = "success"
-    FAILED = "failed"
-
-
-@attr.define
-class BentoApiSchema:
-    route: str
-    doc: str
-    input: str
-    output: str
-
-
-@attr.define
-class BentoRunnerResourceSchema:
-    cpu: t.Optional[t.Any]
-    nvidia_gpu: t.Optional[t.Any]
-    custom_resources: t.Optional[t.Any]
-
-
-@attr.define
-class BentoRunnerSchema:
-    name: str
-    runnable_type: t.Optional[str]
-    models: t.Optional[t.List[str]]
-    resource_config: t.Optional[BentoRunnerResourceSchema]
-
-
-@attr.define
-class BentoManifestSchema:
-    service: str
-    bentoml_version: str
-    size_bytes: int
-    apis: t.Dict[str, BentoApiSchema] = attr.field(factory=dict)
-    models: t.List[str] = attr.field(factory=list)
-    runners: t.Optional[t.List[BentoRunnerSchema]] = attr.field(factory=list)
-
-
-if TYPE_CHECKING:
-    TransmissionStrategy = t.Literal["presigned_url", "proxy"]
-else:
-    TransmissionStrategy = str
-
-
-@attr.define
-class BentoSchema(ResourceSchema):
-    description: str
-    version: str
-    image_build_status: BentoImageBuildStatus
-    upload_status: BentoUploadStatus
-    upload_finished_reason: str
-    presigned_upload_url: str
-    presigned_download_url: str
-    manifest: BentoManifestSchema
-
-    transmission_strategy: t.Optional[TransmissionStrategy] = attr.field(default=None)
-    upload_id: t.Optional[str] = attr.field(default=None)
-
-    upload_started_at: t.Optional[datetime] = attr.field(default=None)
-    upload_finished_at: t.Optional[datetime] = attr.field(default=None)
-    build_at: datetime = attr.field(factory=datetime.now)
-
-
-@attr.define
-class BentoRepositorySchema(ResourceSchema):
-    description: str
-    latest_bento: t.Optional[BentoSchema]
-
-
-@attr.define
-class CreateBentoSchema:
-    description: str
-    version: str
-    manifest: BentoManifestSchema
-    build_at: datetime = attr.field(factory=datetime.now)
-    labels: t.List[LabelItemSchema] = attr.field(factory=list)
-
-
-@attr.define
-class UpdateBentoSchema:
-    manifest: t.Optional[BentoManifestSchema] = attr.field(default=None)
-    labels: t.Optional[t.List[LabelItemSchema]] = attr.field(default=None)
-
-
-@attr.define
-class PreSignMultipartUploadUrlSchema:
-    upload_id: str
-    part_number: int
-
-
-@attr.define
-class CompletePartSchema:
-    part_number: int
-    etag: str
-
-
-@attr.define
-class CompleteMultipartUploadSchema:
-    parts: t.List[CompletePartSchema]
-    upload_id: str
-
-
-@attr.define
-class FinishUploadBentoSchema:
-    status: t.Optional[BentoUploadStatus]
-    reason: t.Optional[str]
-
-
-@attr.define
-class CreateModelRepositorySchema:
-    name: str
-    description: str
-
-
-class ModelImageBuildStatus(Enum):
-    PENDING = "pending"
-    BUILDING = "building"
-    SUCCESS = "success"
-    FAILED = "failed"
-
-
-class ModelUploadStatus(Enum):
-    PENDING = "pending"
-    BUILDING = "uploading"
-    SUCCESS = "success"
-    FAILED = "failed"
-
-
-@attr.define
-class ModelManifestSchema:
-    module: str
-    api_version: str
-    bentoml_version: str
-    size_bytes: int
-    metadata: t.Dict[str, t.Any] = attr.field(factory=dict)
-    context: t.Dict[str, t.Any] = attr.field(factory=dict)
-    options: t.Dict[str, t.Any] = attr.field(factory=dict)
-
-
-@attr.define
-class ModelSchema(ResourceSchema):
-    description: str
-    version: str
-    image_build_status: ModelImageBuildStatus
-    upload_status: ModelUploadStatus
-    upload_finished_reason: str
-    presigned_upload_url: str
-    presigned_download_url: str
-    manifest: ModelManifestSchema
-
-    transmission_strategy: t.Optional[TransmissionStrategy] = attr.field(default=None)
-    upload_id: t.Optional[str] = attr.field(default=None)
-
-    upload_started_at: t.Optional[datetime] = attr.field(default=None)
-    upload_finished_at: t.Optional[datetime] = attr.field(default=None)
-    build_at: datetime = attr.field(factory=datetime.now)
-
-
-@attr.define
-class ModelRepositorySchema(ResourceSchema):
-    description: str
-    latest_model: t.Optional[ModelSchema]
-
-
-@attr.define
-class CreateModelSchema:
-    description: str
-    version: str
-    manifest: ModelManifestSchema
-    build_at: datetime = attr.field(factory=datetime.now)
-    labels: t.List[LabelItemSchema] = attr.field(factory=list)
-
-
-@attr.define
-class FinishUploadModelSchema:
-    status: t.Optional[ModelUploadStatus]
-    reason: t.Optional[str]
diff --git a/src/bentoml/bentos.py b/src/bentoml/bentos.py
index 84f0762f677..2725b6fd66f 100644
--- a/src/bentoml/bentos.py
+++ b/src/bentoml/bentos.py
@@ -27,7 +27,7 @@
 if t.TYPE_CHECKING:
     from .server import Server
     from ._internal.bento import BentoStore
-    from ._internal.yatai_client import YataiClient
+    from ._internal.cloud import BentoCloudClient
     from ._internal.bento.build_config import CondaOptions
     from ._internal.bento.build_config import DockerOptions
     from ._internal.bento.build_config import PythonOptions
@@ -246,13 +246,13 @@ def push(
     *,
     force: bool = False,
     _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
-    _yatai_client: YataiClient = Provide[BentoMLContainer.yatai_client],
+    _cloud_client: BentoCloudClient = Provide[BentoMLContainer.bentocloud_client],
 ):
     """Push Bento to a yatai server."""
     bento = _bento_store.get(tag)
     if not bento:
         raise BentoMLException(f"Bento {tag} not found in local store")
-    _yatai_client.push_bento(bento, force=force)
+    _cloud_client.push_bento(bento, force=force)
 
 
 @inject
@@ -261,9 +261,9 @@ def pull(
     *,
     force: bool = False,
     _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
-    _yatai_client: YataiClient = Provide[BentoMLContainer.yatai_client],
+    _cloud_client: BentoCloudClient = Provide[BentoMLContainer.bentocloud_client],
 ):
-    _yatai_client.pull_bento(tag, force=force, bento_store=_bento_store)
+    _cloud_client.pull_bento(tag, force=force, bento_store=_bento_store)
 
 
 @inject
diff --git a/src/bentoml/cloud.py b/src/bentoml/cloud.py
new file mode 100644
index 00000000000..f8e90bbc60e
--- /dev/null
+++ b/src/bentoml/cloud.py
@@ -0,0 +1,3 @@
+from ._internal.cloud import YataiClient as YataiClient
+from ._internal.cloud import BentoCloudClient as BentoCloudClient
+from ._internal.cloud import Resource as Resource
diff --git a/src/bentoml/exceptions.py b/src/bentoml/exceptions.py
index 4c02bb20a28..c06c251d489 100644
--- a/src/bentoml/exceptions.py
+++ b/src/bentoml/exceptions.py
@@ -109,8 +109,8 @@ class CLIException(BentoMLException):
     """Raised when CLI encounters an issue"""
 
 
-class YataiRESTApiClientError(BentoMLException):
-    """Raised when communicating with BentoCloud server."""
+class CloudRESTApiClientError(BentoMLException):
+    """Raised when communicating with Yatai or BentoCloud server."""
 
     pass
 
diff --git a/src/bentoml/models.py b/src/bentoml/models.py
index f5f7a008716..5037c4312a5 100644
--- a/src/bentoml/models.py
+++ b/src/bentoml/models.py
@@ -21,7 +21,7 @@
 if TYPE_CHECKING:
     from ._internal.models import ModelStore
     from ._internal.models.model import ModelSignaturesType
-    from ._internal.yatai_client import YataiClient
+    from ._internal.cloud import BentoCloudClient
 
 
 @inject
@@ -206,12 +206,12 @@ def push(
     *,
     force: bool = False,
     _model_store: "ModelStore" = Provide[BentoMLContainer.model_store],
-    _yatai_client: YataiClient = Provide[BentoMLContainer.yatai_client],
+    _cloud_client: BentoCloudClient = Provide[BentoMLContainer.bentocloud_client],
 ):
     model_obj = _model_store.get(tag)
     if not model_obj:
         raise BentoMLException(f"Model {tag} not found in local store")
-    _yatai_client.push_model(model_obj, force=force)
+    _cloud_client.push_model(model_obj, force=force)
 
 
 @inject
@@ -219,9 +219,9 @@ def pull(
     tag: t.Union[Tag, str],
     *,
     force: bool = False,
-    _yatai_client: YataiClient = Provide[BentoMLContainer.yatai_client],
+    _cloud_client: BentoCloudClient = Provide[BentoMLContainer.bentocloud_client],
 ) -> Model:
-    return _yatai_client.pull_model(tag, force=force)
+    return _cloud_client.pull_model(tag, force=force)
 
 
 @inject
diff --git a/src/bentoml_cli/bentos.py b/src/bentoml_cli/bentos.py
index 64fdcf6886e..686e1ca8da0 100644
--- a/src/bentoml_cli/bentos.py
+++ b/src/bentoml_cli/bentos.py
@@ -67,7 +67,7 @@ def add_bento_management_commands(cli: Group):
     from bentoml._internal.utils.analytics.usage_stats import _usage_event_debugging
 
     bento_store = BentoMLContainer.bento_store.get()
-    yatai_client = BentoMLContainer.yatai_client.get()
+    cloud_client = BentoMLContainer.bentocloud_client.get()
 
     @cli.command()
     @click.argument("bento_tag", type=click.STRING)
@@ -255,7 +255,7 @@ def import_bento_(bento_path: str) -> None:  # type: ignore (not accessed)
     )
     def pull(bento_tag: str, force: bool, context: str) -> None:  # type: ignore (not accessed)
         """Pull Bento from a yatai server."""
-        yatai_client.pull_bento(bento_tag, force=force, context=context)
+        cloud_client.pull_bento(bento_tag, force=force, context=context)
 
     @cli.command()
     @click.argument("bento_tag", type=click.STRING)
@@ -280,7 +280,7 @@ def push(bento_tag: str, force: bool, threads: int, context: str) -> None:  # ty
         bento_obj = bento_store.get(bento_tag)
         if not bento_obj:
             raise click.ClickException(f"Bento {bento_tag} not found in local store")
-        yatai_client.push_bento(
+        cloud_client.push_bento(
             bento_obj, force=force, threads=threads, context=context
         )
 
@@ -316,7 +316,7 @@ def build(  # type: ignore (not accessed)
         _bento_store: BentoStore = Provide[BentoMLContainer.bento_store],
     ):
         """Build a new Bento from current directory."""
-        if output == 'tag':
+        if output == "tag":
             from bentoml._internal.configuration import set_quiet_mode
 
             from bentoml._internal.log import configure_logging
diff --git a/src/bentoml_cli/cli.py b/src/bentoml_cli/cli.py
index 69fd50196a4..0ae74c0cbf9 100644
--- a/src/bentoml_cli/cli.py
+++ b/src/bentoml_cli/cli.py
@@ -7,10 +7,11 @@
 from bentoml_cli.serve import add_serve_command
 from bentoml_cli.start import add_start_command
 from bentoml_cli.utils import BentoMLCommandGroup
-from bentoml_cli.yatai import add_login_command
+from bentoml_cli.cloud import add_cloud_command
 from bentoml_cli.bentos import add_bento_management_commands
 from bentoml_cli.models import add_model_management_commands
 from bentoml_cli.containerize import add_containerize_command
+from bentoml_cli.deployment import add_deployment_command
 
 
 def create_bentoml_cli() -> click.Group:
@@ -36,12 +37,13 @@ def bentoml_cli():
 
     # Add top-level CLI commands
     add_env_command(bentoml_cli)
-    add_login_command(bentoml_cli)
+    add_cloud_command(bentoml_cli)
     add_bento_management_commands(bentoml_cli)
     add_model_management_commands(bentoml_cli)
     add_start_command(bentoml_cli)
     add_serve_command(bentoml_cli)
     add_containerize_command(bentoml_cli)
+    add_deployment_command(bentoml_cli)
 
     if psutil.WINDOWS:
         import sys
diff --git a/src/bentoml_cli/cloud.py b/src/bentoml_cli/cloud.py
new file mode 100644
index 00000000000..039d4006ba8
--- /dev/null
+++ b/src/bentoml_cli/cloud.py
@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+import click
+import json
+import click_option_group as cog
+
+
+def add_cloud_command(cli: click.Group) -> click.Group:
+    from bentoml_cli.utils import BentoMLCommandGroup
+    from bentoml.exceptions import CLIException
+    from bentoml._internal.configuration import get_quiet_mode
+    from bentoml._internal.utils import bentoml_cattr
+    from bentoml._internal.cloud.client import RestApiClient
+    from bentoml._internal.cloud.config import add_context
+    from bentoml._internal.cloud.config import CloudClientContext
+    from bentoml._internal.cloud.config import default_context_name
+    from bentoml._internal.cloud.config import CloudClientConfig
+
+    @cli.group(name="cloud", aliases=["yatai"], cls=BentoMLCommandGroup)
+    def cloud():
+        """BentoCloud Subcommands Groups
+
+        \b
+        Note that 'bentoml yatai' is mainly for backward compatible reason. It is recommended
+        to use 'bentoml cloud'.
+        """
+
+    @cloud.command()
+    @cog.optgroup.group(
+        "Login", help="Required login options", cls=cog.RequiredAllOptionGroup
+    )
+    @cog.optgroup.option(
+        "--endpoint",
+        type=click.STRING,
+        help="BentoCloud or Yatai endpoint, i.e: https://cloud.bentoml.com",
+    )
+    @cog.optgroup.option(
+        "--api-token",
+        type=click.STRING,
+        help="BentoCloud or Yatai user API token",
+    )
+    @click.option(
+        "--context",
+        type=click.STRING,
+        help="BentoCloud or Yatai context name for the endpoint and API token",
+        default=default_context_name,
+    )
+    def login(endpoint: str, api_token: str, context: str) -> None:  # type: ignore (not accessed)
+        """Login to BentoCloud or Yatai server."""
+        cloud_rest_client = RestApiClient(endpoint, api_token)
+        user = cloud_rest_client.get_current_user()
+
+        if user is None:
+            raise CLIException("current user is not found")
+
+        org = cloud_rest_client.get_current_organization()
+
+        if org is None:
+            raise CLIException("current organization is not found")
+
+        ctx = CloudClientContext(
+            name=context,
+            endpoint=endpoint,
+            api_token=api_token,
+            email=user.email,
+        )
+
+        add_context(ctx)
+        click.echo(
+            f"Successfully logged in to BentoCloud for {user.name} in {org.name}"
+        )
+
+    @cloud.command(aliases=["current-context"])
+    def get_current_context() -> None:  # type: ignore (not accessed)
+        """Get current cloud context."""
+        cur = CloudClientConfig.get_config().get_current_context()
+        if not get_quiet_mode():
+            click.echo(json.dumps(bentoml_cattr.unstructure(cur), indent=2))
+
+    @cloud.command()
+    @click.argument("context", type=click.STRING)
+    def update_current_context(context: str) -> None:  # type: ignore (not accessed)
+        """Update current context"""
+        ctx = CloudClientConfig.get_config().set_current_context(context)
+        click.echo(f"Successfully switched to context: {ctx.name}")
+
+    return cli
diff --git a/src/bentoml_cli/deployment.py b/src/bentoml_cli/deployment.py
new file mode 100644
index 00000000000..fb600b754e4
--- /dev/null
+++ b/src/bentoml_cli/deployment.py
@@ -0,0 +1,257 @@
+from __future__ import annotations
+
+import click
+import typing as t
+
+if t.TYPE_CHECKING:
+    TupleStrAny = tuple[str, ...]
+    from bentoml._internal.cloud.schemas import DeploymentSchema
+    from bentoml._internal.cloud.schemas import DeploymentListSchema
+else:
+    TupleStrAny = tuple
+
+
+def add_deployment_command(cli: click.Group) -> None:
+    import json
+    from bentoml_cli.utils import BentoMLCommandGroup
+    from bentoml._internal.configuration.containers import BentoMLContainer
+    from bentoml._internal.utils import rich_console as console, bentoml_cattr
+    from rich.table import Table
+
+    client = BentoMLContainer.bentocloud_client.get()
+    output_option = click.option(
+        "-o",
+        "--output",
+        type=click.Choice(["json", "default"]),
+        default="default",
+        help="Display the output of this command.",
+    )
+
+    def shared_decorator(f: t.Callable[..., t.Any]):
+        options = [
+            click.argument(
+                "deployment-name",
+                type=click.STRING,
+                required=True,
+            ),
+            click.option(
+                "--context", type=click.STRING, default=None, help="Yatai context name."
+            ),
+            click.option(
+                "--cluster-name",
+                type=click.STRING,
+                default=None,
+                help="Name of the cluster.",
+            ),
+            click.option(
+                "--kube-namespace",
+                type=click.STRING,
+                default=None,
+                help="Kubernetes namespace.",
+            ),
+            output_option,
+        ]
+        for opt in reversed(options):
+            f = opt(f)
+        return f
+
+    @cli.group(name="deployment", cls=BentoMLCommandGroup)
+    def deployment_cli():
+        """Deployment Subcommands Groups"""
+
+    @deployment_cli.command()
+    @click.option(
+        "-f",
+        "--file",
+        type=click.File(),
+        help="JSON file path for the deployment configuration",
+    )
+    @click.option(
+        "--context", type=click.STRING, default=None, help="Yatai context name."
+    )
+    @output_option
+    def create(  # type: ignore
+        file: str,
+        context: str,
+        output: t.Literal["json", "default"],
+    ) -> DeploymentSchema:
+        """Create a deployment on BentoCloud.
+
+        \b
+        A deployment can be created using a json file with configurations.
+        The json file has the exact format as the one on BentoCloud Deployment UI.
+        """
+        res = client.deployment.create_from_file(path_or_stream=file, context=context)
+        if output == "default":
+            console.print(res)
+        elif output == "json":
+            click.echo(json.dumps(bentoml_cattr.unstructure(res), indent=2))
+        return res
+
+    @deployment_cli.command()
+    @click.option(
+        "-f",
+        "--file",
+        type=click.File(),
+        help="JSON file path for the deployment configuration",
+    )
+    @click.option(
+        "--context", type=click.STRING, default=None, help="Yatai context name."
+    )
+    @output_option
+    def update(  # type: ignore
+        file: str,
+        context: str,
+        output: t.Literal["json", "default"],
+    ) -> DeploymentSchema:
+        """Update a deployment on BentoCloud.
+
+        \b
+        A deployment can be updated using a json file with needed configurations.
+        The json file has the exact format as the one on BentoCloud Deployment UI.
+        """
+        res = client.deployment.update_from_file(
+            path_or_stream=file,
+            context=context,
+        )
+        if output == "default":
+            console.print(res)
+        elif output == "json":
+            unstructured = bentoml_cattr.unstructure(res)
+            click.echo(json.dumps(unstructured, indent=2))
+        return res
+
+    @deployment_cli.command()
+    @shared_decorator
+    def get(  # type: ignore
+        deployment_name: str,
+        context: str,
+        cluster_name: str,
+        kube_namespace: str,
+        output: t.Literal["json", "default"],
+    ) -> DeploymentSchema:
+        """Get a deployment on BentoCloud."""
+        res = client.deployment.get(
+            deployment_name=deployment_name,
+            context=context,
+            cluster_name=cluster_name,
+            kube_namespace=kube_namespace,
+        )
+        if output == "default":
+            console.print(res)
+        elif output == "json":
+            unstructured = bentoml_cattr.unstructure(res)
+            click.echo(json.dumps(unstructured, indent=2))
+        return res
+
+    @deployment_cli.command()
+    @shared_decorator
+    def terminate(  # type: ignore
+        deployment_name: str,
+        context: str,
+        cluster_name: str,
+        kube_namespace: str,
+        output: t.Literal["json", "default"],
+    ) -> DeploymentSchema:
+        """Terminate a deployment on BentoCloud."""
+        res = client.deployment.terminate(
+            deployment_name=deployment_name,
+            context=context,
+            cluster_name=cluster_name,
+            kube_namespace=kube_namespace,
+        )
+        if output == "default":
+            console.print(res)
+        elif output == "json":
+            unstructured = bentoml_cattr.unstructure(res)
+            click.echo(json.dumps(unstructured, indent=2))
+        return res
+
+    @deployment_cli.command()
+    @shared_decorator
+    def delete(  # type: ignore
+        deployment_name: str,
+        context: str,
+        cluster_name: str,
+        kube_namespace: str,
+        output: t.Literal["json", "default"],
+    ) -> DeploymentSchema:
+        """Delete a deployment on BentoCloud."""
+        res = client.deployment.delete(
+            deployment_name=deployment_name,
+            context=context,
+            cluster_name=cluster_name,
+            kube_namespace=kube_namespace,
+        )
+        if output == "default":
+            console.print(res)
+        elif output == "json":
+            unstructured = bentoml_cattr.unstructure(res)
+            click.echo(json.dumps(unstructured, indent=2))
+        return res
+
+    @deployment_cli.command()
+    @click.option(
+        "--context", type=click.STRING, default=None, help="Yatai context name."
+    )
+    @click.option(
+        "--cluster-name", type=click.STRING, default=None, help="Name of the cluster."
+    )
+    @click.option(
+        "--query", type=click.STRING, default=None, help="Query for list request."
+    )
+    @click.option(
+        "--search", type=click.STRING, default=None, help="Search for list request."
+    )
+    @click.option(
+        "--start", type=click.STRING, default=None, help="Start for list request."
+    )
+    @click.option(
+        "--count", type=click.STRING, default=None, help="Count for list request."
+    )
+    @click.option(
+        "-o",
+        "--output",
+        help="Display the output of this command.",
+        type=click.Choice(["json", "default", "table"]),
+        default="table",
+    )
+    def list(  # type: ignore
+        context: str,
+        cluster_name: str,
+        query: str,
+        search: str,
+        count: int,
+        start: int,
+        output: t.Literal["json", "default", "table"],
+    ) -> DeploymentListSchema:
+        """List existing deployments on BentoCloud."""
+        res = client.deployment.list(
+            context=context,
+            cluster_name=cluster_name,
+            query=query,
+            search=search,
+            count=count,
+            start=start,
+        )
+        if output == "table":
+            table = Table(box=None)
+            table.add_column("Deployment")
+            table.add_column("Bento")
+            table.add_column("Status")
+            table.add_column("Created At")
+            for deployment in res.items:
+                target = deployment.latest_revision.targets[0]
+                table.add_row(
+                    deployment.name,
+                    f"{target.bento.repository.name}:{target.bento.name}",
+                    deployment.status.value,
+                    deployment.created_at.astimezone().strftime("%Y-%m-%d %H:%M:%S"),
+                )
+            console.print(table)
+        elif output == "default":
+            console.print(res)
+        elif output == "json":
+            unstructured = bentoml_cattr.unstructure(res)
+            click.echo(json.dumps(unstructured, indent=2))
+        return res
diff --git a/src/bentoml_cli/models.py b/src/bentoml_cli/models.py
index 85e5b962230..be23d998334 100644
--- a/src/bentoml_cli/models.py
+++ b/src/bentoml_cli/models.py
@@ -49,7 +49,7 @@ def add_model_management_commands(cli: Group) -> None:
     from bentoml._internal.configuration.containers import BentoMLContainer
 
     model_store = BentoMLContainer.model_store.get()
-    yatai_client = BentoMLContainer.yatai_client.get()
+    cloud_client = BentoMLContainer.bentocloud_client.get()
 
     @cli.group(name="models", cls=BentoMLCommandGroup)
     def model_cli():
@@ -235,7 +235,7 @@ def import_from(model_path: str) -> None:  # type: ignore (not accessed)
     )
     def pull(model_tag: str, force: bool, context: str):  # type: ignore (not accessed)
         """Pull Model from a yatai server."""
-        yatai_client.pull_model(model_tag, force=force, context=context)
+        cloud_client.pull_model(model_tag, force=force, context=context)
 
     @model_cli.command()
     @click.argument("model_tag", type=click.STRING)
@@ -260,6 +260,6 @@ def push(model_tag: str, force: bool, threads: int, context: str):  # type: igno
         model_obj = model_store.get(model_tag)
         if not model_obj:
             raise click.ClickException(f"Model {model_tag} not found in local store")
-        yatai_client.push_model(
+        cloud_client.push_model(
             model_obj, force=force, threads=threads, context=context
         )
diff --git a/src/bentoml_cli/utils.py b/src/bentoml_cli/utils.py
index ceeefb7c39c..a13098ae044 100644
--- a/src/bentoml_cli/utils.py
+++ b/src/bentoml_cli/utils.py
@@ -16,6 +16,7 @@
 if TYPE_CHECKING:
     from click import Option
     from click import Command
+    from click import Group
     from click import Context
     from click import Parameter
     from click import HelpFormatter
@@ -195,7 +196,7 @@ class BentoMLCommandGroup(click.Group):
     Click command class customized for BentoML CLI, allow specifying a default
     command for each group defined.
 
-    This command groups will also introduce support for aliases for commands.
+    This command groups will also introduce support for aliases for commands and groups.
 
     Example:
 
@@ -204,6 +205,9 @@ class BentoMLCommandGroup(click.Group):
         @click.group(cls=BentoMLCommandGroup)
         def cli(): ...
 
+        @click.group(name="cloud", aliases=["cloud"], cls=BentoMLCommandGroup)
+        def cli(): ...
+
         @cli.command(aliases=["serve-http"])
         def serve(): ...
     """
@@ -377,6 +381,22 @@ def wrapper(func: F[P]) -> Command:
 
         return wrapper
 
+    def group(self, *args: t.Any, **kwargs: t.Any) -> t.Callable[[F[P]], Group]:
+        aliases = kwargs.pop("aliases", None)
+
+        def decorator(f: F[P]):
+            # create the main group
+            grp = super(BentoMLCommandGroup, self).group(*args, **kwargs)(f)
+
+            if aliases is not None:
+                assert grp.name
+                self._commands[grp.name] = aliases
+                self._aliases.update({k: grp.name for k in aliases})
+
+            return grp
+
+        return decorator
+
     def resolve_alias(self, cmd_name: str):
         return self._aliases[cmd_name] if cmd_name in self._aliases else cmd_name
 
diff --git a/src/bentoml_cli/yatai.py b/src/bentoml_cli/yatai.py
deleted file mode 100644
index 98b48735a9e..00000000000
--- a/src/bentoml_cli/yatai.py
+++ /dev/null
@@ -1,59 +0,0 @@
-from __future__ import annotations
-
-import click
-
-
-def add_login_command(cli: click.Group) -> None:
-    from bentoml_cli.utils import BentoMLCommandGroup
-    from bentoml.exceptions import CLIException
-    from bentoml._internal.yatai_rest_api_client.yatai import YataiRESTApiClient
-    from bentoml._internal.yatai_rest_api_client.config import add_context
-    from bentoml._internal.yatai_rest_api_client.config import YataiClientContext
-    from bentoml._internal.yatai_rest_api_client.config import default_context_name
-
-    @cli.group(name="yatai", cls=BentoMLCommandGroup)
-    def yatai_cli():
-        """Yatai Subcommands Groups"""
-
-    @yatai_cli.command()
-    @click.option(
-        "--endpoint", type=click.STRING, help="Yatai endpoint, i.e: https://yatai.com"
-    )
-    @click.option("--api-token", type=click.STRING, help="Yatai user API token")
-    @click.option(
-        "--context",
-        type=click.STRING,
-        help="Yatai context name for the endpoint and API token",
-        default=default_context_name,
-    )
-    def login(endpoint: str, api_token: str, context: str) -> None:  # type: ignore (not accessed)
-        """Login to Yatai server."""
-        if not endpoint:
-            raise CLIException("need --endpoint")
-
-        if not api_token:
-            raise CLIException("need --api-token")
-
-        yatai_rest_client = YataiRESTApiClient(endpoint, api_token)
-        user = yatai_rest_client.get_current_user()
-
-        if user is None:
-            raise CLIException("current user is not found")
-
-        org = yatai_rest_client.get_current_organization()
-
-        if org is None:
-            raise CLIException("current organization is not found")
-
-        ctx = YataiClientContext(
-            name=context,
-            endpoint=endpoint,
-            api_token=api_token,
-            email=user.email,
-        )
-
-        add_context(ctx)
-
-        click.echo(
-            f'Successfully logged in as user "{user.name}" in organization "{org.name}".'
-        )


Test patch:
diff --git a/tests/unit/_internal/bento/test_service_serialization.py b/tests/unit/_internal/bento/test_service_serialization.py
index 7c66b01637c..281710fede8 100644
--- a/tests/unit/_internal/bento/test_service_serialization.py
+++ b/tests/unit/_internal/bento/test_service_serialization.py
@@ -2,6 +2,7 @@
 
 import os
 import sys
+import typing as t
 from unittest.mock import ANY
 from unittest.mock import MagicMock
 
@@ -51,7 +52,7 @@ def reload_service_instance():
 @pytest.mark.usefixtures(
     "change_test_dir", "reset_serialization_strategy", "reload_service_instance"
 )
-def test_export_bento_strategy(build_bento):
+def test_export_bento_strategy(build_bento: bentoml.Bento):
     bentoml.set_serialization_strategy("EXPORT_BENTO")
     svc = bentoml.load(build_bento.tag)
     del sys.modules["simplebento"]
@@ -62,7 +63,7 @@ def test_export_bento_strategy(build_bento):
 @pytest.mark.usefixtures(
     "change_test_dir", "reset_serialization_strategy", "reload_service_instance"
 )
-def test_local_bento_strategy(build_bento):
+def test_local_bento_strategy(build_bento: bentoml.Bento):
     bentoml.set_serialization_strategy("LOCAL_BENTO")
     svc = bentoml.load(build_bento.tag)
     del sys.modules["simplebento"]
@@ -73,7 +74,7 @@ def test_local_bento_strategy(build_bento):
 @pytest.mark.usefixtures(
     "change_test_dir", "reset_serialization_strategy", "reload_service_instance"
 )
-def test_remote_bento_strategy_with_local_store_hit(build_bento):
+def test_remote_bento_strategy_with_local_store_hit(build_bento:bentoml.Bento):
     bentoml.set_serialization_strategy("REMOTE_BENTO")
     svc = bentoml.load(build_bento.tag)
     del sys.modules["simplebento"]
@@ -84,7 +85,7 @@ def test_remote_bento_strategy_with_local_store_hit(build_bento):
 @pytest.mark.usefixtures(
     "change_test_dir", "reset_serialization_strategy", "reload_service_instance"
 )
-def test_remote_bento_strategy_with_tmp_store_hit(build_bento):
+def test_remote_bento_strategy_with_tmp_store_hit(build_bento: bentoml.Bento):
     bentoml.set_serialization_strategy("REMOTE_BENTO")
     svc = bentoml.load(build_bento.tag)
     serialized_svc = cloudpickle.dumps(svc)
@@ -101,13 +102,13 @@ def test_remote_bento_strategy_with_tmp_store_hit(build_bento):
 @pytest.mark.usefixtures(
     "change_test_dir", "reset_serialization_strategy", "reload_service_instance"
 )
-def test_remote_bento_strategy_pull_yatai(build_bento):
+def test_remote_bento_strategy_pull_yatai(build_bento: bentoml.Bento):
     bentoml.set_serialization_strategy("REMOTE_BENTO")
     svc = bentoml.load(build_bento.tag)
 
-    mock_yatai_client = MagicMock()
+    mock_bentocloud_client = MagicMock()
 
-    def pull_bento(*args, **kwargs):
+    def pull_bento(*args: t.Any, **kwargs: t.Any):
         working_dir = os.getcwd()
         bento = build_test_bento()
         os.chdir(working_dir)
@@ -115,19 +116,19 @@ def pull_bento(*args, **kwargs):
         if "simplebento" in sys.modules:
             del sys.modules["simplebento"]
 
-    mock_yatai_client.pull_bento.side_effect = pull_bento
+    mock_bentocloud_client.pull_bento.side_effect = pull_bento
 
-    _yatai_client = BentoMLContainer.yatai_client.get()
-    BentoMLContainer.yatai_client.set(mock_yatai_client)
+    _bentocloud_client = BentoMLContainer.bentocloud_client.get()
+    BentoMLContainer.bentocloud_client.set(mock_bentocloud_client)
 
     serialized_svc = cloudpickle.dumps(svc)
     bentoml.bentos.delete(build_bento.tag)
 
     del sys.modules["simplebento"]
     assert svc == cloudpickle.loads(serialized_svc)
-    mock_yatai_client.pull_bento.assert_called_once_with(
+    mock_bentocloud_client.pull_bento.assert_called_once_with(
         Tag("test.simplebento", "1.0"), bento_store=ANY
     )
 
-    # Reset mock yatai_client
-    BentoMLContainer.yatai_client.set(_yatai_client)
+    # Reset mock bentocloud_client
+    BentoMLContainer.bentocloud_client.set(_bentocloud_client)
diff --git a/src/bentoml/_internal/yatai_rest_api_client/__init__.py b/tests/unit/_internal/cloud/__init__.py
similarity index 100%
rename from src/bentoml/_internal/yatai_rest_api_client/__init__.py
rename to tests/unit/_internal/cloud/__init__.py
diff --git a/tests/unit/_internal/cloud/test_deployment.py b/tests/unit/_internal/cloud/test_deployment.py
new file mode 100644
index 00000000000..825bde7e68b
--- /dev/null
+++ b/tests/unit/_internal/cloud/test_deployment.py
@@ -0,0 +1,830 @@
+from __future__ import annotations
+from datetime import datetime
+
+import typing as t
+from unittest.mock import patch
+import pytest
+from bentoml._internal.cloud.schemas import (
+    BentoFullSchema,
+    BentoImageBuildStatus,
+    BentoManifestSchema,
+    BentoRepositorySchema,
+    BentoUploadStatus,
+    ClusterSchema,
+    DeploymentMode,
+    DeploymentRevisionSchema,
+    DeploymentRevisionStatus,
+    DeploymentSchema,
+    DeploymentStatus,
+    DeploymentTargetSchema,
+    ResourceType,
+)
+from bentoml._internal.cloud.schemas import DeploymentTargetType
+from bentoml._internal.cloud.schemas import UpdateDeploymentSchema
+from bentoml.cloud import BentoCloudClient
+from bentoml.cloud import Resource
+from bentoml._internal.cloud.schemas import CreateDeploymentSchema
+from bentoml._internal.cloud.schemas import CreateDeploymentTargetSchema
+from bentoml._internal.cloud.schemas import DeploymentTargetConfig
+from bentoml._internal.cloud.schemas import DeploymentTargetRunnerConfig
+from bentoml._internal.cloud.schemas import LabelItemSchema
+from bentoml._internal.cloud.schemas import DeploymentTargetCanaryRule
+from bentoml._internal.cloud.schemas import DeploymentTargetCanaryRuleType
+from bentoml._internal.cloud.schemas import DeploymentTargetHPAConf
+from bentoml._internal.cloud.schemas import UserSchema
+import attr
+
+if t.TYPE_CHECKING:
+    from unittest.mock import MagicMock
+
+
+def f_create(
+    create_deployment_schema: CreateDeploymentSchema,
+    context: str | None = None,
+    cluster_name: str | None = None,
+):
+    return create_deployment_schema
+
+
+def f_update(
+    deployment_name: str,
+    update_deployment_schema: UpdateDeploymentSchema,
+    kube_namespace: str | None = None,
+    context: str | None = None,
+    cluster_name: str | None = None,
+):
+    return update_deployment_schema
+
+
+@pytest.fixture(name="get_schema", scope="function")
+def fixture_get_schema() -> DeploymentSchema:
+    user = UserSchema(name="", email="", first_name="", last_name="")
+    return DeploymentSchema(
+        latest_revision=DeploymentRevisionSchema(
+            targets=[
+                DeploymentTargetSchema(
+                    type=DeploymentTargetType.STABLE,
+                    bento=BentoFullSchema(
+                        uid="",
+                        created_at=datetime(2023, 5, 25),
+                        updated_at=None,
+                        deleted_at=None,
+                        name="12345",
+                        resource_type=ResourceType.BENTO,
+                        labels=[],
+                        description="",
+                        version="",
+                        image_build_status=BentoImageBuildStatus.PENDING,
+                        upload_status=BentoUploadStatus.SUCCESS,
+                        upload_finished_reason="",
+                        presigned_upload_url="",
+                        presigned_download_url="",
+                        manifest=BentoManifestSchema(
+                            service="",
+                            bentoml_version="",
+                            size_bytes=0,
+                            apis={},
+                            models=["iris_clf:ddaex6h2vw6kwcvj"],
+                        ),
+                        build_at=datetime(2023, 5, 25),
+                        repository=BentoRepositorySchema(
+                            uid="",
+                            created_at="",
+                            updated_at=None,
+                            deleted_at=None,
+                            name="iris_classifier",
+                            resource_type=ResourceType.BENTO_REPOSITORY,
+                            labels=[],
+                            description="",
+                            latest_bento="",
+                        ),
+                    ),
+                    config=DeploymentTargetConfig(
+                        resource_instance="t3-micro",
+                        enable_ingress=True,
+                        hpa_conf=DeploymentTargetHPAConf(
+                            min_replicas=2, max_replicas=10
+                        ),
+                        runners={
+                            "runner1": DeploymentTargetRunnerConfig(
+                                resource_instance="t3-small",
+                                hpa_conf=DeploymentTargetHPAConf(
+                                    min_replicas=3, max_replicas=10
+                                ),
+                            ),
+                            "runner2": DeploymentTargetRunnerConfig(
+                                resource_instance="t3-medium",
+                                hpa_conf=DeploymentTargetHPAConf(
+                                    min_replicas=5, max_replicas=10
+                                ),
+                            ),
+                        },
+                    ),
+                    canary_rules=[],
+                    uid="",
+                    created_at=datetime(2023, 5, 1),
+                    updated_at=None,
+                    deleted_at=None,
+                    name="",
+                    resource_type=ResourceType.DEPLOYMENT_REVISION,
+                    labels=[],
+                    creator=user,
+                )
+            ],
+            uid="",
+            created_at=datetime(2023, 5, 1),
+            updated_at=None,
+            deleted_at=None,
+            name="test=xxx",
+            resource_type=ResourceType.DEPLOYMENT_REVISION,
+            labels=[],
+            creator=user,
+            status=DeploymentRevisionStatus.ACTIVE,
+        ),
+        uid="",
+        created_at=datetime(2023, 5, 1),
+        updated_at=None,
+        deleted_at=None,
+        name="test=xxx",
+        resource_type=ResourceType.DEPLOYMENT_REVISION,
+        labels=[],
+        creator=user,
+        status=DeploymentStatus.Running,
+        cluster=ClusterSchema(
+            uid="",
+            name="default",
+            resource_type=ResourceType.CLUSTER,
+            labels=[],
+            description="",
+            creator=user,
+            created_at=datetime(2023, 5, 1),
+            updated_at=None,
+            deleted_at=None,
+        ),
+        kube_namespace="",
+    )
+
+
+@pytest.fixture(scope="function", name="cloudclient")
+def fixture_cloudclient() -> BentoCloudClient:
+    return BentoCloudClient()
+
+
+@patch("bentoml._internal.cloud.deployment.Deployment._create_deployment")
+def test_create_deployment(
+    mock_create_deployment: MagicMock, cloudclient: BentoCloudClient
+):
+    mock_create_deployment.side_effect = f_create
+
+    res = cloudclient.deployment.create(
+        deployment_name="test-xxx", bento="iris_classifier:dqjxjyx2vweogcvj"
+    )
+    assert res == CreateDeploymentSchema(
+        targets=[
+            CreateDeploymentTargetSchema(
+                type=DeploymentTargetType.STABLE,
+                bento_repository="iris_classifier",
+                bento="dqjxjyx2vweogcvj",
+                config=DeploymentTargetConfig(),
+            )
+        ],
+        mode=DeploymentMode.Function,
+        name="test-xxx",
+    )
+
+
+@patch("bentoml._internal.cloud.deployment.Deployment._create_deployment")
+def test_create_deployment_canary_rules(
+    mock_create_deployment: MagicMock, cloudclient: BentoCloudClient
+):
+    mock_create_deployment.side_effect = f_create
+    rules = [
+        DeploymentTargetCanaryRule(DeploymentTargetCanaryRuleType.WEIGHT, 3, "", "", "")
+    ]
+    res = cloudclient.deployment.create(
+        deployment_name="test-xxx",
+        bento="iris_classifier:dqjxjyx2vweogcvj",
+        canary_rules=rules,
+    )
+    assert res == CreateDeploymentSchema(
+        targets=[
+            CreateDeploymentTargetSchema(
+                type=DeploymentTargetType.STABLE,
+                bento_repository="iris_classifier",
+                bento="dqjxjyx2vweogcvj",
+                config=DeploymentTargetConfig(),
+                canary_rules=rules,
+            )
+        ],
+        mode=DeploymentMode.Function,
+        name="test-xxx",
+    )
+
+
+@patch("bentoml._internal.cloud.deployment.Deployment._create_deployment")
+def test_create_deployment_labels(
+    mock_create_deployment: MagicMock, cloudclient: BentoCloudClient
+):
+    mock_create_deployment.side_effect = f_create
+
+    res = cloudclient.deployment.create(
+        deployment_name="test-xxx",
+        bento="iris_classifier:dqjxjyx2vweogcvj",
+        labels={"user": "steve"},
+    )
+    assert res == CreateDeploymentSchema(
+        targets=[
+            CreateDeploymentTargetSchema(
+                type=DeploymentTargetType.STABLE,
+                bento_repository="iris_classifier",
+                bento="dqjxjyx2vweogcvj",
+                config=DeploymentTargetConfig(),
+            )
+        ],
+        mode=DeploymentMode.Function,
+        name="test-xxx",
+        labels=[LabelItemSchema("user", "steve")],
+    )
+
+
+@patch("bentoml._internal.cloud.deployment.Deployment._create_deployment")
+def test_create_deployment_resource_instance(
+    mock_create_deployment: MagicMock, cloudclient: BentoCloudClient
+):
+    mock_create_deployment.side_effect = f_create
+
+    res = cloudclient.deployment.create(
+        deployment_name="test-xxx",
+        bento="iris_classifier:dqjxjyx2vweogcvj",
+        resource_instance="test-instance",
+    )
+    assert res == CreateDeploymentSchema(
+        targets=[
+            CreateDeploymentTargetSchema(
+                type=DeploymentTargetType.STABLE,
+                bento_repository="iris_classifier",
+                bento="dqjxjyx2vweogcvj",
+                config=DeploymentTargetConfig(resource_instance="test-instance"),
+            )
+        ],
+        mode=DeploymentMode.Function,
+        name="test-xxx",
+    )
+
+
+@patch("bentoml._internal.cloud.deployment.Deployment._create_deployment")
+def test_create_deployment_resource_instance_runner(
+    mock_create_deployment: MagicMock, cloudclient: BentoCloudClient
+):
+    mock_create_deployment.side_effect = f_create
+    runner = Resource.for_runner(enable_debug_mode=True)
+
+    res = cloudclient.deployment.create(
+        deployment_name="test-xxx",
+        bento="iris_classifier:dqjxjyx2vweogcvj",
+        resource_instance="test-instance",
+        runners_config={"runner": runner},
+    )
+    assert res == CreateDeploymentSchema(
+        targets=[
+            CreateDeploymentTargetSchema(
+                type=DeploymentTargetType.STABLE,
+                bento_repository="iris_classifier",
+                bento="dqjxjyx2vweogcvj",
+                config=DeploymentTargetConfig(
+                    resource_instance="test-instance",
+                    runners={
+                        "runner": DeploymentTargetRunnerConfig(
+                            resource_instance="test-instance", enable_debug_mode=True
+                        )
+                    },
+                ),
+            )
+        ],
+        mode=DeploymentMode.Function,
+        name="test-xxx",
+    )
+
+
+@patch("bentoml._internal.cloud.deployment.Deployment._create_deployment")
+def test_create_deployment_resource_instance_api_server(
+    mock_create_deployment: MagicMock, cloudclient: BentoCloudClient
+):
+    mock_create_deployment.side_effect = f_create
+    api_server = Resource.for_api_server(enable_ingress=True)
+
+    res = cloudclient.deployment.create(
+        deployment_name="test-xxx",
+        bento="iris_classifier:dqjxjyx2vweogcvj",
+        resource_instance="test-resource",
+        api_server_config=api_server,
+    )
+    assert res == CreateDeploymentSchema(
+        targets=[
+            CreateDeploymentTargetSchema(
+                type=DeploymentTargetType.STABLE,
+                bento_repository="iris_classifier",
+                bento="dqjxjyx2vweogcvj",
+                config=DeploymentTargetConfig(
+                    resource_instance="test-resource", enable_ingress=True
+                ),
+            )
+        ],
+        mode=DeploymentMode.Function,
+        name="test-xxx",
+    )
+
+
+@patch("bentoml._internal.cloud.deployment.Deployment._create_deployment")
+def test_create_deployment_api_server(
+    mock_create_deployment: MagicMock, cloudclient: BentoCloudClient
+):
+    mock_create_deployment.side_effect = f_create
+    api_server_conf = Resource.for_api_server(resource_instance="t3-micro")
+    res = cloudclient.deployment.create(
+        deployment_name="test-xxx",
+        bento="iris_classifier:dqjxjyx2vweogcvj",
+        api_server_config=api_server_conf,
+    )
+
+    assert res == CreateDeploymentSchema(
+        targets=[
+            CreateDeploymentTargetSchema(
+                type=DeploymentTargetType.STABLE,
+                bento_repository="iris_classifier",
+                bento="dqjxjyx2vweogcvj",
+                config=DeploymentTargetConfig(resource_instance="t3-micro"),
+            )
+        ],
+        mode=DeploymentMode.Function,
+        name="test-xxx",
+    )
+
+
+@patch("bentoml._internal.cloud.deployment.Deployment._create_deployment")
+def test_create_deployment_hpa_conf(
+    mock_create_deployment: MagicMock, cloudclient: BentoCloudClient
+):
+    mock_create_deployment.side_effect = f_create
+    hpa_conf = Resource.for_hpa_conf(min_replicas=2, max_replicas=10)
+    res = cloudclient.deployment.create(
+        deployment_name="test-xxx",
+        bento="iris_classifier:dqjxjyx2vweogcvj",
+        hpa_conf=hpa_conf,
+    )
+    assert res == CreateDeploymentSchema(
+        targets=[
+            CreateDeploymentTargetSchema(
+                type=DeploymentTargetType.STABLE,
+                bento_repository="iris_classifier",
+                bento="dqjxjyx2vweogcvj",
+                config=DeploymentTargetConfig(hpa_conf=hpa_conf),
+            )
+        ],
+        mode=DeploymentMode.Function,
+        name="test-xxx",
+    )
+
+
+@patch("bentoml._internal.cloud.deployment.Deployment._create_deployment")
+def test_create_deployment_runner(
+    mock_create_deployment: MagicMock, cloudclient: BentoCloudClient
+):
+    mock_create_deployment.side_effect = f_create
+    runner = Resource.for_runner(resource_instance="t3-micro", enable_debug_mode=True)
+    res = cloudclient.deployment.create(
+        deployment_name="test-xxx",
+        bento="iris_classifier:dqjxjyx2vweogcvj",
+        runners_config={"runner1": runner},
+    )
+    assert res == CreateDeploymentSchema(
+        targets=[
+            CreateDeploymentTargetSchema(
+                type=DeploymentTargetType.STABLE,
+                bento_repository="iris_classifier",
+                bento="dqjxjyx2vweogcvj",
+                config=DeploymentTargetConfig(runners={"runner1": runner}),
+            )
+        ],
+        mode=DeploymentMode.Function,
+        name="test-xxx",
+    )
+
+
+@patch("bentoml._internal.cloud.deployment.Deployment._create_deployment")
+def test_create_deployment_runner_hpa_conf(
+    mock_create_deployment: MagicMock, cloudclient: BentoCloudClient
+):
+    mock_create_deployment.side_effect = f_create
+    hpa_conf = Resource.for_hpa_conf(min_replicas=2, max_replicas=10)
+    runner = Resource.for_runner(resource_instance="t3-micro", enable_debug_mode=True)
+    res = cloudclient.deployment.create(
+        deployment_name="test-xxx",
+        bento="iris_classifier:dqjxjyx2vweogcvj",
+        runners_config={"runner1": runner},
+        hpa_conf=hpa_conf,
+    )
+    assert res == CreateDeploymentSchema(
+        targets=[
+            CreateDeploymentTargetSchema(
+                type=DeploymentTargetType.STABLE,
+                bento_repository="iris_classifier",
+                bento="dqjxjyx2vweogcvj",
+                config=DeploymentTargetConfig(
+                    hpa_conf=hpa_conf,
+                    runners={
+                        "runner1": DeploymentTargetRunnerConfig(
+                            resource_instance="t3-micro",
+                            hpa_conf=hpa_conf,
+                            enable_debug_mode=True,
+                        )
+                    },
+                ),
+            )
+        ],
+        mode=DeploymentMode.Function,
+        name="test-xxx",
+    )
+
+
+@patch("bentoml._internal.cloud.deployment.Deployment._create_deployment")
+def test_create_deployment_api_server_runner(
+    mock_create_deployment: MagicMock, cloudclient: BentoCloudClient
+):
+    mock_create_deployment.side_effect = f_create
+    api_server = Resource.for_api_server(
+        resource_instance="t3-micro", enable_stealing_traffic_debug_mode=True
+    )
+    runner = Resource.for_runner(resource_instance="t3-micro", enable_debug_mode=True)
+    res = cloudclient.deployment.create(
+        deployment_name="test-xxx",
+        bento="iris_classifier:dqjxjyx2vweogcvj",
+        runners_config={"runner1": runner},
+        api_server_config=api_server,
+    )
+    assert res == CreateDeploymentSchema(
+        targets=[
+            CreateDeploymentTargetSchema(
+                type=DeploymentTargetType.STABLE,
+                bento_repository="iris_classifier",
+                bento="dqjxjyx2vweogcvj",
+                config=DeploymentTargetConfig(
+                    resource_instance="t3-micro",
+                    enable_stealing_traffic_debug_mode=True,
+                    runners={
+                        "runner1": DeploymentTargetRunnerConfig(
+                            resource_instance="t3-micro", enable_debug_mode=True
+                        )
+                    },
+                ),
+            )
+        ],
+        mode=DeploymentMode.Function,
+        name="test-xxx",
+    )
+
+
+@patch("bentoml._internal.cloud.deployment.Deployment._create_deployment")
+def test_create_deployment_api_server_hpa_conf(
+    mock_create_deployment: MagicMock, cloudclient: BentoCloudClient
+):
+    mock_create_deployment.side_effect = f_create
+    api_server = Resource.for_api_server(resource_instance="t3-micro")
+    hpa_conf = Resource.for_hpa_conf(min_replicas=2, max_replicas=10)
+    res = cloudclient.deployment.create(
+        deployment_name="test-xxx",
+        bento="iris_classifier:dqjxjyx2vweogcvj",
+        api_server_config=api_server,
+        hpa_conf=hpa_conf,
+    )
+    assert res == CreateDeploymentSchema(
+        targets=[
+            CreateDeploymentTargetSchema(
+                type=DeploymentTargetType.STABLE,
+                bento_repository="iris_classifier",
+                bento="dqjxjyx2vweogcvj",
+                config=DeploymentTargetConfig(
+                    resource_instance="t3-micro", hpa_conf=hpa_conf
+                ),
+            )
+        ],
+        mode=DeploymentMode.Function,
+        name="test-xxx",
+    )
+
+
+@patch("bentoml._internal.cloud.deployment.Deployment._create_deployment")
+def test_create_deployment_api_server_runner_hpa_conf(
+    mock_create_deployment: MagicMock, cloudclient: BentoCloudClient
+):
+    mock_create_deployment.side_effect = f_create
+    api_server = Resource.for_api_server(resource_instance="t3-micro")
+    runner = Resource.for_runner(
+        resource_instance="t3-small", hpa_conf={"min_replicas": 3}
+    )
+    runner2 = Resource.for_runner(
+        resource_instance="t3-medium", hpa_conf={"min_replicas": 5}
+    )
+    hpa_conf = Resource.for_hpa_conf(min_replicas=2, max_replicas=10)
+    res = cloudclient.deployment.create(
+        deployment_name="test-xxx",
+        bento="iris_classifier:dqjxjyx2vweogcvj",
+        api_server_config=api_server,
+        hpa_conf=hpa_conf,
+        runners_config={"runner1": runner, "runner2": runner2},
+        expose_endpoint=True,
+        labels={"user": "steve"},
+    )
+    assert res == CreateDeploymentSchema(
+        labels=[LabelItemSchema("user", "steve")],
+        targets=[
+            CreateDeploymentTargetSchema(
+                type=DeploymentTargetType.STABLE,
+                bento_repository="iris_classifier",
+                bento="dqjxjyx2vweogcvj",
+                config=DeploymentTargetConfig(
+                    resource_instance="t3-micro",
+                    enable_ingress=True,
+                    hpa_conf=hpa_conf,
+                    runners={
+                        "runner1": DeploymentTargetRunnerConfig(
+                            resource_instance="t3-small",
+                            hpa_conf=DeploymentTargetHPAConf(
+                                min_replicas=3, max_replicas=10
+                            ),
+                        ),
+                        "runner2": DeploymentTargetRunnerConfig(
+                            resource_instance="t3-medium",
+                            hpa_conf=DeploymentTargetHPAConf(
+                                min_replicas=5, max_replicas=10
+                            ),
+                        ),
+                    },
+                ),
+            )
+        ],
+        mode=DeploymentMode.Function,
+        name="test-xxx",
+    )
+
+
+@pytest.fixture(name="update_schema", scope="function")
+def fixture_update_schema() -> UpdateDeploymentSchema:
+    return UpdateDeploymentSchema(
+        targets=[
+            CreateDeploymentTargetSchema(
+                type=DeploymentTargetType.STABLE,
+                bento_repository="iris_classifier",
+                bento="12345",
+                config=DeploymentTargetConfig(
+                    resource_instance="t3-micro",
+                    enable_ingress=True,
+                    hpa_conf=DeploymentTargetHPAConf(min_replicas=2, max_replicas=10),
+                    runners={
+                        "runner1": DeploymentTargetRunnerConfig(
+                            resource_instance="t3-small",
+                            hpa_conf=DeploymentTargetHPAConf(
+                                min_replicas=3, max_replicas=10
+                            ),
+                        ),
+                        "runner2": DeploymentTargetRunnerConfig(
+                            resource_instance="t3-medium",
+                            hpa_conf=DeploymentTargetHPAConf(
+                                min_replicas=5, max_replicas=10
+                            ),
+                        ),
+                    },
+                ),
+            )
+        ],
+        mode=DeploymentMode.Function,
+        labels=[],
+    )
+
+
+@patch("bentoml._internal.cloud.deployment.Deployment.get")
+@patch("bentoml._internal.cloud.deployment.Deployment._update_deployment")
+def test_update_deployment_bento(
+    mock_update_deployment: MagicMock,
+    mock_get: MagicMock,
+    update_schema: UpdateDeploymentSchema,
+    get_schema: DeploymentSchema,
+    cloudclient: BentoCloudClient,
+):
+    mock_update_deployment.side_effect = f_update
+    mock_get.return_value = get_schema
+    res = cloudclient.deployment.update(
+        deployment_name="test-xxx",
+        bento="iris_classifier:dqjxjyx2vweogcvj",
+        cluster_name="",
+        kube_namespace="",
+    )
+    update_schema.targets[0].bento = "dqjxjyx2vweogcvj"
+    assert res == update_schema
+
+
+@patch("bentoml._internal.cloud.deployment.Deployment.get")
+@patch("bentoml._internal.cloud.deployment.Deployment._update_deployment")
+def test_update_deployment_runner(
+    mock_update_deployment: MagicMock,
+    mock_get: MagicMock,
+    update_schema: UpdateDeploymentSchema,
+    get_schema: DeploymentSchema,
+    cloudclient: BentoCloudClient,
+):
+    mock_update_deployment.side_effect = f_update
+    mock_get.return_value = get_schema
+    new_runnner = Resource.for_runner(
+        resource_instance="new-resource", hpa_conf={"min_replicas": 6}
+    )
+    res = cloudclient.deployment.update(
+        deployment_name="test-xxx",
+        cluster_name="",
+        kube_namespace="",
+        runners_config={"runner1": new_runnner},
+    )
+    update_schema.targets[0].config.runners["runner1"].hpa_conf.min_replicas = 6
+    update_schema.targets[0].config.runners[
+        "runner1"
+    ].resource_instance = "new-resource"
+    assert res == update_schema
+
+
+@patch("bentoml._internal.cloud.deployment.Deployment.get")
+@patch("bentoml._internal.cloud.deployment.Deployment._update_deployment")
+def test_update_deployment_runner_hpa_conf(
+    mock_update_deployment: MagicMock,
+    mock_get: MagicMock,
+    update_schema: UpdateDeploymentSchema,
+    get_schema: DeploymentSchema,
+    cloudclient: BentoCloudClient,
+):
+    mock_update_deployment.side_effect = f_update
+    mock_get.return_value = get_schema
+    hpa_conf = Resource.for_hpa_conf(min_replicas=5)
+    new_runnner = Resource.for_runner(
+        resource_instance="new-resource", hpa_conf={"min_replicas": 7}
+    )
+    res = cloudclient.deployment.update(
+        deployment_name="test-xxx",
+        cluster_name="",
+        kube_namespace="",
+        runners_config={"runner1": new_runnner},
+        hpa_conf=hpa_conf,
+    )
+    update_schema.targets[0].config.hpa_conf.min_replicas = 5
+    for k, v in update_schema.targets[0].config.runners.items():
+        if k == "runner1":
+            v.hpa_conf.min_replicas = 7
+            v.resource_instance = "new-resource"
+        else:
+            v.hpa_conf.min_replicas = 5
+    assert res == update_schema
+
+
+@patch("bentoml._internal.cloud.deployment.Deployment.get")
+@patch("bentoml._internal.cloud.deployment.Deployment._update_deployment")
+def test_update_deployment_api_server(
+    mock_update_deployment: MagicMock,
+    mock_get: MagicMock,
+    update_schema: UpdateDeploymentSchema,
+    get_schema: DeploymentSchema,
+    cloudclient: BentoCloudClient,
+):
+    mock_update_deployment.side_effect = f_update
+    mock_get.return_value = get_schema
+    api_server = Resource.for_api_server(
+        enable_ingress=False, hpa_conf={"min_replicas": 5}
+    )
+    res = cloudclient.deployment.update(
+        deployment_name="test-xxx",
+        cluster_name="",
+        kube_namespace="",
+        api_server_config=api_server,
+    )
+    update_schema.targets[0].config.hpa_conf.min_replicas = 5
+    update_schema.targets[0].config.enable_ingress = False
+    assert res == update_schema
+
+
+@patch("bentoml._internal.cloud.deployment.Deployment.get")
+@patch("bentoml._internal.cloud.deployment.Deployment._update_deployment")
+def test_update_deployment_api_server_hpa_conf(
+    mock_update_deployment: MagicMock,
+    mock_get: MagicMock,
+    update_schema: UpdateDeploymentSchema,
+    get_schema: DeploymentSchema,
+    cloudclient: BentoCloudClient,
+):
+    mock_update_deployment.side_effect = f_update
+    mock_get.return_value = get_schema
+    api_server = Resource.for_api_server(hpa_conf={"min_replicas": 9})
+    hpa_conf = Resource.for_hpa_conf(min_replicas=8)
+    res = cloudclient.deployment.update(
+        deployment_name="test-xxx",
+        cluster_name="",
+        kube_namespace="",
+        api_server_config=api_server,
+        hpa_conf=hpa_conf,
+    )
+    UpdateDeploymentSchema(
+        targets=[
+            CreateDeploymentTargetSchema(
+                type=DeploymentTargetType.STABLE,
+                bento_repository="iris_classifier",
+                bento="12345",
+                config=DeploymentTargetConfig(
+                    resource_instance="t3-micro",
+                    enable_ingress=True,
+                    hpa_conf=DeploymentTargetHPAConf(min_replicas=2, max_replicas=10),
+                    runners={
+                        "runner1": DeploymentTargetRunnerConfig(
+                            resource_instance="t3-small",
+                            hpa_conf=DeploymentTargetHPAConf(
+                                min_replicas=3, max_replicas=10
+                            ),
+                        ),
+                        "runner2": DeploymentTargetRunnerConfig(
+                            resource_instance="t3-medium",
+                            hpa_conf=DeploymentTargetHPAConf(
+                                min_replicas=5, max_replicas=10
+                            ),
+                        ),
+                    },
+                ),
+            )
+        ],
+        mode=DeploymentMode.Function,
+        labels=[],
+    )
+    update_schema.targets[0].config.hpa_conf.min_replicas = 9
+    for _, v in update_schema.targets[0].config.runners.items():
+        v.hpa_conf.min_replicas = 8
+    assert res == update_schema
+
+
+@patch("bentoml._internal.cloud.deployment.Deployment.get")
+@patch("bentoml._internal.cloud.deployment.Deployment._update_deployment")
+def test_update_deployment_resource_instance(
+    mock_update_deployment: MagicMock,
+    mock_get: MagicMock,
+    update_schema: UpdateDeploymentSchema,
+    get_schema: DeploymentSchema,
+    cloudclient: BentoCloudClient,
+):
+    mock_update_deployment.side_effect = f_update
+    mock_get.return_value = get_schema
+    res = cloudclient.deployment.update(
+        deployment_name="test-xxx",
+        cluster_name="",
+        kube_namespace="",
+        resource_instance="test-resource",
+    )
+    update_schema.targets[0].config.resource_instance = "test-resource"
+    for _, v in update_schema.targets[0].config.runners.items():
+        v.resource_instance = "test-resource"
+    assert res == update_schema
+
+
+@patch("bentoml._internal.cloud.deployment.Deployment.get")
+@patch("bentoml._internal.cloud.deployment.Deployment._update_deployment")
+def test_update_deployment_labels(
+    mock_update_deployment: MagicMock,
+    mock_get: MagicMock,
+    update_schema: UpdateDeploymentSchema,
+    get_schema: DeploymentSchema,
+    cloudclient: BentoCloudClient,
+):
+    mock_update_deployment.side_effect = f_update
+    mock_get.return_value = get_schema
+    res = cloudclient.deployment.update(
+        deployment_name="test-xxx",
+        cluster_name="",
+        kube_namespace="",
+        labels={"user": "steve"},
+    )
+    assert res == attr.evolve(update_schema, labels=[LabelItemSchema("user", "steve")])
+
+
+@patch("bentoml._internal.cloud.deployment.Deployment.get")
+@patch("bentoml._internal.cloud.deployment.Deployment._update_deployment")
+def test_update_deployment_canary_rules(
+    mock_update_deployment: MagicMock,
+    mock_get: MagicMock,
+    update_schema: UpdateDeploymentSchema,
+    get_schema: DeploymentSchema,
+    cloudclient: BentoCloudClient,
+):
+    mock_update_deployment.side_effect = f_update
+    mock_get.return_value = get_schema
+    rules = [
+        DeploymentTargetCanaryRule(DeploymentTargetCanaryRuleType.WEIGHT, 3, "", "", "")
+    ]
+    res = cloudclient.deployment.update(
+        deployment_name="test-xxx",
+        cluster_name="",
+        kube_namespace="",
+        canary_rules=rules,
+    )
+    update_schema.targets[0].canary_rules = rules
+    assert res == update_schema