Repository: bentoml/BentoML
Pull request #3991 (2023-06-26 02:35:26)

feat: BentoCloudClient
## What does this PR address?
#### BentoCloudClient is implemented with the following functionalities:
```
client = bentoml.BentoCloudClient()

client.list_bentos()

client.list_models()

client.deployment.create()

client.deployment.get()

client.deployment.update()

client.deployment.terminate()

client.deployment.delete()

client.deployment.list()
```

#### CLI: 
`bentoml cloud login` is added, and `bentoml yatai login` is kept for backward compatibility. 

Deployment Subcommands Groups is implemented:
`bentoml deployment ...`:
```  Deployment Subcommands Groups
Options:
  -h, --help  Show this message and exit.

Commands:
  create     Create a deployment on BentoCloud.
  delete     Delete a deployment on BentoCloud.
  get        Get a deployment on BentoCloud.
  list       List existing deployments on BentoCloud.
  terminate  Terminate a deployment on BentoCloud.
  update     Update a deployment on BentoCloud.
```
<!--
Thanks for sending a pull request!

Congrats for making it this far! Here's a 🍱 for you. There are still a few steps ahead.

Please make sure to read the contribution guidelines, then fill out the blanks below before requesting a code review.

Name your Pull Request with one of the following prefixes, e.g. "feat: add support for PyTorch", to indicate the type of changes proposed. This is based on the [Conventional Commits specification](https://www.conventionalcommits.org/en/v1.0.0/#summary).
  - feat: (new feature for the user, not a new feature for build script)
  - fix: (bug fix for the user, not a fix to a build script)
  - docs: (changes to the documentation)
  - style: (formatting, missing semicolons, etc; no production code change)
  - refactor: (refactoring production code, eg. renaming a variable)
  - perf: (code changes that improve performance)
  - test: (adding missing tests, refactoring tests; no production code change)
  - chore: (updating grunt tasks etc; no production code change)
  - build: (changes that affect the build system or external dependencies)
  - ci: (changes to configuration files and scripts)
  - revert: (reverts a previous commit)

Describe your changes in detail. Attach screenshots here if appropriate.

Once you're done with this, someone from BentoML team or community member will help review your PR (see "Who can help review?" section for potential reviewers.). If no one has reviewed your PR after a week have passed, don't hesitate to post a new comment and ping @-the same person. Notifications sometimes get lost 🥲.
-->

<!-- Remove if not applicable -->

Fixes #(issue)

## Before submitting:

<!--- Go over all the following points, and put an `x` in all the boxes that apply. -->
<!--- If you're unsure about any of these, don't hesitate to ask. We're here to help! -->
<!--- If you plan to update documentation or tests in follow-up, please note -->

- [ ] Does the Pull Request follow [Conventional Commits specification](https://www.conventionalcommits.org/en/v1.0.0/#summary) naming? Here are [GitHub's
      guide](https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/proposing-changes-to-your-work-with-pull-requests/creating-a-pull-request) on how to create a pull request.
- [ ] Does the code follow BentoML's code style, `pre-commit run -a` script has passed ([instructions](https://github.com/bentoml/BentoML/blob/main/DEVELOPMENT.md#style-check-auto-formatting-type-checking))?
- [ ] Did you read through [contribution guidelines](https://github.com/bentoml/BentoML/blob/main/CONTRIBUTING.md#ways-to-contribute) and follow [development guidelines](https://github.com/bentoml/BentoML/blob/main/DEVELOPMENT.md#start-developing)?
- [ ] Did your changes require updates to the documentation? Have you updated
      those accordingly? Here are [documentation guidelines](https://github.com/bentoml/BentoML/tree/main/docs) and [tips on writting docs](https://github.com/bentoml/BentoML/tree/main/docs#writing-documentation).
- [ ] Did you write tests to cover your changes?


codecov[bot]: ## [Codecov](https://app.codecov.io/gh/bentoml/BentoML/pull/3991?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bentoml) Report
> Merging [#3991](https://app.codecov.io/gh/bentoml/BentoML/pull/3991?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bentoml) (ec151a7) into [main](https://app.codecov.io/gh/bentoml/BentoML/commit/df2b6032bf4cd5a7b3923231a921360054709858?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bentoml) (df2b603) will **not change** coverage.
> The diff coverage is `0.00%`.

[![Impacted file tree graph](https://app.codecov.io/gh/bentoml/BentoML/pull/3991/graphs/tree.svg?width=650&height=150&src=pr&token=GMzgXdpE5b&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bentoml)](https://app.codecov.io/gh/bentoml/BentoML/pull/3991?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bentoml)

```diff
@@           Coverage Diff           @@
##            main   #3991     +/-   ##
=======================================
  Coverage   0.00%   0.00%             
=======================================
  Files        157     166      +9     
  Lines      13126   15214   +2088     
=======================================
- Misses     13126   15214   +2088     
```

| [Impacted Files](https://app.codecov.io/gh/bentoml/BentoML/pull/3991?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bentoml) | Coverage Δ | |
|---|---|---|
| [src/bentoml/\_\_init\_\_.py](https://app.codecov.io/gh/bentoml/BentoML/pull/3991?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bentoml#diff-c3JjL2JlbnRvbWwvX19pbml0X18ucHk=) | `0.00% <0.00%> (ø)` | |
| [src/bentoml/\_internal/cloud/\_\_init\_\_.py](https://app.codecov.io/gh/bentoml/BentoML/pull/3991?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bentoml#diff-c3JjL2JlbnRvbWwvX2ludGVybmFsL2Nsb3VkL19faW5pdF9fLnB5) | `0.00% <0.00%> (ø)` | |
| [src/bentoml/\_internal/cloud/base.py](https://app.codecov.io/gh/bentoml/BentoML/pull/3991?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bentoml#diff-c3JjL2JlbnRvbWwvX2ludGVybmFsL2Nsb3VkL2Jhc2UucHk=) | `0.00% <0.00%> (ø)` | |
| [src/bentoml/\_internal/cloud/bentocloud.py](https://app.codecov.io/gh/bentoml/BentoML/pull/3991?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bentoml#diff-c3JjL2JlbnRvbWwvX2ludGVybmFsL2Nsb3VkL2JlbnRvY2xvdWQucHk=) | `0.00% <0.00%> (ø)` | |
| [src/bentoml/\_internal/cloud/client.py](https://app.codecov.io/gh/bentoml/BentoML/pull/3991?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bentoml#diff-c3JjL2JlbnRvbWwvX2ludGVybmFsL2Nsb3VkL2NsaWVudC5weQ==) | `0.00% <0.00%> (ø)` | |
| [src/bentoml/\_internal/cloud/config.py](https://app.codecov.io/gh/bentoml/BentoML/pull/3991?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bentoml#diff-c3JjL2JlbnRvbWwvX2ludGVybmFsL2Nsb3VkL2NvbmZpZy5weQ==) | `0.00% <0.00%> (ø)` | |
| [src/bentoml/\_internal/cloud/deployment.py](https://app.codecov.io/gh/bentoml/BentoML/pull/3991?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bentoml#diff-c3JjL2JlbnRvbWwvX2ludGVybmFsL2Nsb3VkL2RlcGxveW1lbnQucHk=) | `0.00% <0.00%> (ø)` | |
| [src/bentoml/\_internal/cloud/schemas.py](https://app.codecov.io/gh/bentoml/BentoML/pull/3991?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bentoml#diff-c3JjL2JlbnRvbWwvX2ludGVybmFsL2Nsb3VkL3NjaGVtYXMucHk=) | `0.00% <0.00%> (ø)` | |
| [src/bentoml/\_internal/cloud/yatai.py](https://app.codecov.io/gh/bentoml/BentoML/pull/3991?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bentoml#diff-c3JjL2JlbnRvbWwvX2ludGVybmFsL2Nsb3VkL3lhdGFpLnB5) | `0.00% <0.00%> (ø)` | |
| [src/bentoml/\_internal/configuration/containers.py](https://app.codecov.io/gh/bentoml/BentoML/pull/3991?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bentoml#diff-c3JjL2JlbnRvbWwvX2ludGVybmFsL2NvbmZpZ3VyYXRpb24vY29udGFpbmVycy5weQ==) | `0.00% <0.00%> (ø)` | |
| ... and [8 more](https://app.codecov.io/gh/bentoml/BentoML/pull/3991?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=bentoml) | |


Patch:
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 9ef49b7a..8e613da3 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 2d3d6287..1fdf353e 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 238f51aa..f625ba67 100644
--- a/src/bentoml/__init__.py
+++ b/src/bentoml/__init__.py
@@ -49,7 +49,7 @@ from ._internal.context import ServiceContext as Context
 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 @@ if TYPE_CHECKING:
     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 @@ else:
     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 @@ __all__ = [
     "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 00000000..dc1a77ad
--- /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 00000000..6c1c93d5
--- /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 a9c17142..fe8ab1fa 100644
--- a/src/bentoml/_internal/yatai_client/__init__.py
+++ b/src/bentoml/_internal/cloud/bentocloud.py
@@ -7,8 +7,6 @@ import tempfile
 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 @@ import requests
 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 @@ class YataiClient:
         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 @@ class YataiClient:
         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 @@ class YataiClient:
         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 @@ class YataiClient:
         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 @@ class YataiClient:
         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 @@ class YataiClient:
                 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 @@ class YataiClient:
                         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 1078696b..89d02670 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 UserSchema
 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 @@ class YataiRESTApiClient:
 
     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 @@ class YataiRESTApiClient:
         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 @@ class YataiRESTApiClient:
 
     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 @@ class YataiRESTApiClient:
         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 @@ class YataiRESTApiClient:
 
     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 @@ class YataiRESTApiClient:
         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 @@ class YataiRESTApiClient:
         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 00000000..e05d0822
--- /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 00000000..bf16f1d0
--- /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 00000000..1d1ea3d5
--- /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 00000000..477c0dee
--- /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 38b8aeaa..db322de4 100644
--- a/src/bentoml/_internal/configuration/containers.py
+++ b/src/bentoml/_internal/configuration/containers.py
@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import os
 import math
+from pathlib import Path
 import uuid
 import typing as t
 import logging
@@ -244,6 +245,11 @@ class _BentoMLContainerClass:
     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 @@ class _BentoMLContainerClass:
     @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 a8bef2b4..b1ace4b5 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 get_runnable(bento_model: bentoml.Model) -> t.Type[bentoml.Runnable]:
 
         @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 get_runnable(bento_model: bentoml.Model) -> t.Type[bentoml.Runnable]:
             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 16555764..5a338de6 100644
--- a/src/bentoml/_internal/service/service.py
+++ b/src/bentoml/_internal/service/service.py
@@ -181,7 +181,7 @@ class Service:
                 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 @@ class Service:
                         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 8ad5e4f7..a81ec41a 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 inject
 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 de7bc491..17890ba6 100644
--- a/src/bentoml/_internal/utils/pickle.py
+++ b/src/bentoml/_internal/utils/pickle.py
@@ -26,6 +26,7 @@ else:
 # 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 b2fac496..00000000
--- 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 5a7c236c..00000000
--- 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 84f0762f..2725b6fd 100644
--- a/src/bentoml/bentos.py
+++ b/src/bentoml/bentos.py
@@ -27,7 +27,7 @@ from ._internal.utils.analytics.usage_stats import _usage_event_debugging
 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 00000000..f8e90bbc
--- /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 4c02bb20..c06c251d 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 f5f7a008..5037c431 100644
--- a/src/bentoml/models.py
+++ b/src/bentoml/models.py
@@ -21,7 +21,7 @@ from ._internal.configuration.containers import BentoMLContainer
 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 64fdcf68..686e1ca8 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 add_bento_management_commands(cli: Group):
     )
     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 add_bento_management_commands(cli: Group):
         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 add_bento_management_commands(cli: Group):
         _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 69fd5019..0ae74c0c 100644
--- a/src/bentoml_cli/cli.py
+++ b/src/bentoml_cli/cli.py
@@ -7,10 +7,11 @@ from bentoml_cli.env import add_env_command
 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 create_bentoml_cli() -> click.Group:
 
     # 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 00000000..039d4006
--- /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 00000000..fb600b75
--- /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 85e5b962..be23d998 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 add_model_management_commands(cli: Group) -> None:
     )
     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 add_model_management_commands(cli: Group) -> None:
         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 ceeefb7c..a13098ae 100644
--- a/src/bentoml_cli/utils.py
+++ b/src/bentoml_cli/utils.py
@@ -16,6 +16,7 @@ from click.exceptions import UsageError
 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 @@ class BentoMLCommandGroup(click.Group):
 
         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 98b48735..00000000
--- 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 7c66b016..281710fe 100644
--- a/tests/unit/_internal/bento/test_service_serialization.py
+++ b/tests/unit/_internal/bento/test_service_serialization.py
@@ -2,6 +2,7 @@ from __future__ import annotations
 
 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 test_remote_bento_strategy_pull_yatai(build_bento):
         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 00000000..825bde7e
--- /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