feat: add update deployment to dynamo deploy API and CLI (#1048)

.devcontainer/post-create.sh

@@ -62,6 +62,8 @@ ln -sf $HOME/dynamo/.build/target/debug/llmctl $HOME/dynamo/deploy/sdk/src/dynam
 cd $HOME/dynamo/lib/bindings/python && retry uv pip install -e .
 cd $HOME/dynamo && retry env DYNAMO_BIN_PATH=$HOME/dynamo/.build/target/debug uv pip install -e .
 
+export PYTHONPATH=/home/ubuntu/dynamo/components/planner/src:$PYTHONPATH
+
 # source the venv and set the VLLM_KV_CAPI_PATH in bashrc
 echo "source /opt/dynamo/venv/bin/activate" >> ~/.bashrc
 echo "export VLLM_KV_CAPI_PATH=$HOME/dynamo/.build/target/debug/libdynamo_llm_capi.so" >> ~/.bashrc

deploy/cloud/api-store/ai_dynamo_store/api/deployments.py

@@ -14,7 +14,7 @@
 #  limitations under the License.
 
 from datetime import datetime
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, Optional
 
 from fastapi import APIRouter, HTTPException, Query
 
@@ -23,6 +23,7 @@ from ..models.schemas import (
     DeploymentFullSchema,
     DeploymentListResponse,
     ResourceSchema,
+    UpdateDeploymentSchema,
     create_default_cluster,
     create_default_user,
 )
@@ -32,7 +33,9 @@ from .k8s import (
     get_dynamo_deployment,
     get_namespace,
     list_dynamo_deployments,
+    update_dynamo_deployment,
 )
+from .utils import build_latest_revision_from_cr, get_deployment_status, get_urls
 
 router = APIRouter(prefix="/api/v2/deployments", tags=["deployments"])
 
@@ -116,7 +119,7 @@ async def create_deployment(deployment: CreateDeploymentSchema):
             kube_namespace=kube_namespace,
             creator=creator,
             cluster=cluster,
-            latest_revision=None,
+            latest_revision=build_latest_revision_from_cr(created_crd),
             manifest=None,
         )
 
@@ -130,6 +133,15 @@ async def create_deployment(deployment: CreateDeploymentSchema):
 
 @router.get("/{name}", response_model=DeploymentFullSchema)
 def get_deployment(name: str) -> DeploymentFullSchema:
+    """
+    Retrieve a deployment by name.
+
+    Args:
+        name: The name of the deployment to retrieve
+
+    Returns:
+        DeploymentFullSchema: The deployment details
+    """
     try:
         kube_namespace = get_namespace()
         cr = get_dynamo_deployment(
@@ -147,7 +159,7 @@ def get_deployment(name: str) -> DeploymentFullSchema:
             urls=get_urls(cr),
             creator=create_default_user(),
             cluster=create_default_cluster(create_default_user()),
-            latest_revision=None,
+            latest_revision=build_latest_revision_from_cr(cr),
             manifest=None,
         )
         return deployment_schema
@@ -159,58 +171,17 @@ def get_deployment(name: str) -> DeploymentFullSchema:
         raise HTTPException(status_code=500, detail=str(e))
 
 
-def get_deployment_status(resource: Dict[str, Any]) -> str:
-    """
-    Get the current status of a deployment.
-    Maps operator status to BentoML status values.
-    Returns lowercase status values matching BentoML's DeploymentStatus enum.
-    """
-    status = resource.get("status", {})
-    conditions = status.get("conditions", [])
-    state = status.get("state", "")
-
-    # First check Ready condition
-    for condition in conditions:
-        if condition.get("type") == "Ready":
-            if condition.get("status") == "True":
-                # If state is "successful", map to "running"
-                if state == "successful":
-                    return "running"
-                return condition.get("message", "running").lower()
-            elif condition.get("message"):
-                return condition.get("message").lower()
-
-    # If no Ready condition or not True, check state
-    if state == "failed":
-        return "failed"
-    elif state == "pending":
-        return "deploying"  # map pending to deploying to match BentoML states
-
-    # Default fallback
-    return "unknown"
-
-
-def get_urls(resource: Dict[str, Any]) -> List[str]:
-    """
-    Get the URLs for a deployment.
-    Returns URLs as soon as they are available from EndpointExposed condition.
+@router.delete("/{name}", response_model=DeploymentFullSchema)
+def delete_deployment(name: str) -> DeploymentFullSchema:
     """
-    urls = []
-    conditions = resource.get("status", {}).get("conditions", [])
-
-    # Check for EndpointExposed condition
-    for condition in conditions:
-        if (
-            condition.get("type") == "EndpointExposed"
-            and condition.get("status") == "True"
-        ):
-            if message := condition.get("message"):
-                urls.append(message)
-    return urls
+    Delete a deployment by name.
 
+    Args:
+        name: The name of the deployment to delete
 
-@router.delete("/{name}", response_model=DeploymentFullSchema)
-def delete_deployment(name: str) -> DeploymentFullSchema:
+    Returns:
+        DeploymentFullSchema: The deleted deployment details
+    """
     try:
         kube_namespace = get_namespace()
         # Get deployment details before deletion
@@ -226,7 +197,7 @@ def delete_deployment(name: str) -> DeploymentFullSchema:
             urls=get_urls(cr),
             creator=create_default_user(),
             cluster=create_default_cluster(create_default_user()),
-            latest_revision=None,
+            latest_revision=build_latest_revision_from_cr(cr),
             manifest=None,
         )
         # Delete the deployment
@@ -295,7 +266,7 @@ def list_deployments(
                 urls=get_urls(cr),
                 creator=create_default_user(),
                 cluster=create_default_cluster(create_default_user()),
-                latest_revision=None,
+                latest_revision=build_latest_revision_from_cr(cr),
                 manifest=None,
             )
 
@@ -334,3 +305,65 @@ def list_deployments(
         print("Error listing deployments:")
         print(e)
         raise HTTPException(status_code=500, detail=str(e))
+
+
+@router.put("/{name}", response_model=DeploymentFullSchema)
+def update_deployment(name: str, deployment: UpdateDeploymentSchema):
+    """
+    Update an existing deployment.
+
+    Args:
+        name: The name of the deployment to update (path param)
+        deployment: The new deployment configuration (body)
+
+    Returns:
+        updated deployment details
+    """
+    try:
+        ownership = {"organization_id": "default-org", "user_id": "default-user"}
+        kube_namespace = get_namespace()
+        existing_deployment = get_deployment(name)
+        if existing_deployment.bento != deployment.bento:
+            raise HTTPException(
+                status_code=422,
+                detail="Cannot update the Dynamo components of a deployment.",
+            )
+
+        deployment_name = sanitize_deployment_name(name, deployment.bento)
+        updated_crd = update_dynamo_deployment(
+            name=deployment_name,
+            namespace=kube_namespace,
+            dynamo_nim=deployment.bento,
+            labels={
+                "ngc-organization": ownership["organization_id"],
+                "ngc-user": ownership["user_id"],
+            },
+            envs=deployment.envs,
+        )
+        resource = ResourceSchema(
+            uid=updated_crd["metadata"]["uid"],
+            name=updated_crd["metadata"]["name"],
+            created_at=updated_crd["metadata"].get(
+                "creationTimestamp", datetime.utcnow()
+            ),
+            updated_at=datetime.utcnow(),
+            resource_type="deployment",
+            labels=[],
+        )
+        creator = create_default_user()
+        cluster = create_default_cluster(creator)
+        deployment_schema = DeploymentFullSchema(
+            **resource.dict(),
+            status=get_deployment_status(updated_crd),
+            kube_namespace=kube_namespace,
+            creator=creator,
+            cluster=cluster,
+            latest_revision=build_latest_revision_from_cr(updated_crd),
+            manifest=None,
+            urls=get_urls(updated_crd),
+        )
+        return deployment_schema
+    except Exception as e:
+        print("Error updating deployment:")
+        print(e)
+        raise HTTPException(status_code=500, detail=str(e))

deploy/cloud/api-store/ai_dynamo_store/api/k8s.py

@@ -14,6 +14,7 @@
 #  limitations under the License.
 
 import os
+from functools import wraps
 from typing import Any, Dict, List, Optional
 
 from fastapi import HTTPException
@@ -34,6 +35,19 @@ DynamoGraphDeployment = K8sResource(
 )
 
 
+def ensure_kube_config(func):
+    @wraps(func)
+    def wrapper(*args, **kwargs):
+        try:
+            config.load_incluster_config()
+        except config.config_exception.ConfigException:
+            config.load_kube_config()
+        return func(*args, **kwargs)
+
+    return wrapper
+
+
+@ensure_kube_config
 def create_custom_resource(
     group: str, version: str, namespace: str, plural: str, body: Dict[str, Any]
 ) -> Dict[str, Any]:
@@ -50,11 +64,6 @@ def create_custom_resource(
     Returns:
         Created resource
     """
-    try:
-        config.load_incluster_config()
-    except config.config_exception.ConfigException:
-        config.load_kube_config()
-
     api = client.CustomObjectsApi()
     return api.create_namespaced_custom_object(
         group=group, version=version, namespace=namespace, plural=plural, body=body
@@ -101,6 +110,7 @@ def create_dynamo_deployment(
     )
 
 
+@ensure_kube_config
 def get_dynamo_deployment(name: str, namespace: str) -> Dict[str, Any]:
     """
     Get a DynamoGraphDeployment custom resource.
@@ -115,11 +125,6 @@ def get_dynamo_deployment(name: str, namespace: str) -> Dict[str, Any]:
     Raises:
         HTTPException: If the deployment is not found or an error occurs
     """
-    try:
-        config.load_incluster_config()
-    except config.config_exception.ConfigException:
-        config.load_kube_config()
-
     api = client.CustomObjectsApi()
     try:
         return api.get_namespaced_custom_object(
@@ -143,15 +148,11 @@ def get_namespace() -> str:
     return os.getenv("DEFAULT_KUBE_NAMESPACE", "dynamo")
 
 
+@ensure_kube_config
 def delete_dynamo_deployment(name: str, namespace: str) -> Dict[str, Any]:
     """
     Delete a DynamoGraphDeployment custom resource.
     """
-    try:
-        config.load_incluster_config()
-    except config.config_exception.ConfigException:
-        config.load_kube_config()
-
     api = client.CustomObjectsApi()
     try:
         return api.delete_namespaced_custom_object(
@@ -168,6 +169,7 @@ def delete_dynamo_deployment(name: str, namespace: str) -> Dict[str, Any]:
             raise HTTPException(status_code=500, detail=str(e))
 
 
+@ensure_kube_config
 def list_dynamo_deployments(
     namespace: str,
     label_selector: Optional[str] = None,
@@ -185,11 +187,6 @@ def list_dynamo_deployments(
     Raises:
         HTTPException: If an error occurs during listing
     """
-    try:
-        config.load_incluster_config()
-    except config.config_exception.ConfigException:
-        config.load_kube_config()
-
     api = client.CustomObjectsApi()
     try:
         response = api.list_namespaced_custom_object(
@@ -202,3 +199,62 @@ def list_dynamo_deployments(
         return response["items"]
     except client.rest.ApiException as e:
         raise HTTPException(status_code=500, detail=str(e))
+
+
+@ensure_kube_config
+def update_dynamo_deployment(
+    name: str,
+    namespace: str,
+    dynamo_nim: str,
+    labels: Dict[str, str],
+    envs: Optional[List[Dict[str, str]]] = None,
+) -> Dict[str, Any]:
+    """
+    Update a DynamoGraphDeployment custom resource.
+
+    Args:
+        name: Deployment name
+        namespace: Target namespace
+        dynamo_nim: Bento name and version (format: name:version)
+        labels: Resource labels
+        envs: Optional list of environment variables
+
+    Returns:
+        Updated deployment
+    """
+    # Fetch the current resource to get resourceVersion
+    current = get_dynamo_deployment(name, namespace)
+    resource_version = current["metadata"].get("resourceVersion")
+    if not resource_version:
+        raise RuntimeError("resourceVersion not found in current resource")
+
+    body = {
+        "apiVersion": "nvidia.com/v1alpha1",
+        "kind": "DynamoGraphDeployment",
+        "metadata": {
+            "name": name,
+            "namespace": namespace,
+            "labels": labels,
+            "resourceVersion": resource_version,  # Required for update
+        },
+        "spec": {
+            "dynamoGraph": dynamo_nim,
+            "services": {},
+            "envs": envs if envs else [],
+        },
+    }
+    api = client.CustomObjectsApi()
+    try:
+        return api.replace_namespaced_custom_object(
+            group=DynamoGraphDeployment.group,
+            version=DynamoGraphDeployment.version,
+            namespace=namespace,
+            plural=DynamoGraphDeployment.plural,
+            name=name,
+            body=body,
+        )
+    except client.rest.ApiException as e:
+        if e.status == 404:
+            raise HTTPException(status_code=404, detail="Deployment not found")
+        else:
+            raise HTTPException(status_code=500, detail=str(e))

deploy/cloud/api-store/ai_dynamo_store/api/utils.py

+#  SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+#  SPDX-License-Identifier: Apache-2.0
+#  #
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  #
+#  http://www.apache.org/licenses/LICENSE-2.0
+#  #
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+
+from datetime import datetime
+from typing import Any, Dict, List
+
+
+def get_deployment_status(resource: Dict[str, Any]) -> str:
+    """
+    Get the current status of a deployment.
+    Maps operator status to BentoML status values.
+    Returns lowercase status values matching BentoML's DeploymentStatus enum.
+    """
+    status = resource.get("status", {})
+    conditions = status.get("conditions", [])
+    state = status.get("state", "")
+
+    # First check Ready condition
+    for condition in conditions:
+        if condition.get("type") == "Ready":
+            if condition.get("status") == "True":
+                # If state is "successful", map to "running"
+                if state == "successful":
+                    return "running"
+                return condition.get("message", "running").lower()
+            elif condition.get("message"):
+                return condition.get("message").lower()
+
+    # If no Ready condition or not True, check state
+    if state == "failed":
+        return "failed"
+    elif state == "pending":
+        return "deploying"  # map pending to deploying to match BentoML states
+
+    # Default fallback
+    return "unknown"
+
+
+def get_urls(resource: Dict[str, Any]) -> List[str]:
+    """
+    Get the URLs for a deployment.
+    Returns URLs as soon as they are available from EndpointExposed condition.
+    """
+    urls = []
+    conditions = resource.get("status", {}).get("conditions", [])
+
+    # Check for EndpointExposed condition
+    for condition in conditions:
+        if (
+            condition.get("type") == "EndpointExposed"
+            and condition.get("status") == "True"
+        ):
+            if message := condition.get("message"):
+                urls.append(message)
+    return urls
+
+
+def build_latest_revision_from_cr(cr: dict) -> dict:
+    spec = cr.get("spec", {})
+    meta = cr.get("metadata", {})
+    now = datetime.utcnow().isoformat() + "Z"
+    bento_str = spec.get("dynamoGraph", "unknown:unknown")
+    if ":" in bento_str:
+        bento_name, bento_version = bento_str.split(":", 1)
+    else:
+        bento_name, bento_version = "unknown", "unknown"
+    # Dummy creator
+    creator = {"name": "system", "email": "", "first_name": "", "last_name": ""}
+    # Dummy repository
+    repository = {
+        "uid": "dummy-repo-uid",
+        "created_at": now,
+        "updated_at": now,
+        "deleted_at": None,
+        "name": bento_name,
+        "resource_type": "bento_repository",
+        "labels": [],
+        "description": "",
+        "latest_bento": None,
+    }
+    # Dummy bento
+    bento = {
+        "uid": "dummy-bento-uid",
+        "created_at": now,
+        "updated_at": now,
+        "deleted_at": None,
+        "name": bento_version,
+        "resource_type": "bento",
+        "labels": [],
+        "description": "",
+        "repository": repository,
+        "version": bento_version,
+        "image_build_status": "",
+        "upload_status": "",
+        "upload_finished_reason": "",
+        "presigned_upload_url": "",
+        "presigned_download_url": "",
+    }
+    # Target
+    target = {
+        "uid": "dummy-target-uid",
+        "created_at": now,
+        "updated_at": now,
+        "deleted_at": None,
+        "name": "default-target",
+        "resource_type": "deployment_target",
+        "labels": [],
+        "creator": creator,
+        "status": "running",
+        "config": {
+            "services": spec.get("services", {}),
+            "access_authorization": True,
+            "envs": spec.get("envs", []),
+        },
+        "bento": bento,
+    }
+    # Revision
+    return {
+        "uid": meta.get("uid", "dummy-uid"),
+        "created_at": meta.get("creationTimestamp", now),
+        "updated_at": meta.get("creationTimestamp", now),
+        "deleted_at": None,
+        "name": meta.get("name", "dummy-revision"),
+        "resource_type": "deployment_revision",
+        "labels": [],
+        "creator": creator,
+        "status": "running",
+        "targets": [target],
+    }

deploy/cloud/api-store/ai_dynamo_store/models/__init__.py

+#  SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+#  SPDX-License-Identifier: Apache-2.0
+#  #
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  #
+#  http://www.apache.org/licenses/LICENSE-2.0
+#  #
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.

deploy/cloud/api-store/ai_dynamo_store/tests/__init__.py

+#  SPDX-FileCopyrightText: Copyright (c) 2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+#  SPDX-License-Identifier: Apache-2.0
+#  #
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  #
+#  http://www.apache.org/licenses/LICENSE-2.0
+#  #
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.

deploy/cloud/api-store/ai_dynamo_store/api/test_deployments.py → deploy/cloud/api-store/ai_dynamo_store/tests/test_utils.py

@@ -13,7 +13,8 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-from .deployments import get_deployment_status, get_urls
+
+from ..api.utils import build_latest_revision_from_cr, get_deployment_status, get_urls
 
 
 def test_get_deployment_status():
@@ -57,3 +58,43 @@ def test_get_urls():
         }
     }
     assert get_urls(resource) == ["https://example.com"]
+
+
+def test_build_latest_revision_from_cr_minimal():
+    cr = {
+        "metadata": {
+            "uid": "u1",
+            "name": "n1",
+            "creationTimestamp": "2024-01-01T00:00:00Z",
+        },
+        "spec": {
+            "dynamoGraph": "repo:ver",
+            "services": {"svc": {}},
+            "envs": [{"name": "A", "value": "B"}],
+        },
+    }
+    rev = build_latest_revision_from_cr(cr)
+    assert rev["uid"] == "u1"
+    assert rev["name"] == "n1"
+    assert rev["targets"][0]["bento"]["repository"]["name"] == "repo"
+    assert rev["targets"][0]["bento"]["name"] == "ver"
+    assert rev["targets"][0]["config"]["services"] == {"svc": {}}
+    assert rev["targets"][0]["config"]["envs"] == [{"name": "A", "value": "B"}]
+
+
+def test_build_latest_revision_from_cr_missing_fields():
+    cr = {"spec": {}}
+    rev = build_latest_revision_from_cr(cr)
+    assert rev["uid"] == "dummy-uid"
+    assert rev["name"] == "dummy-revision"
+    assert rev["targets"][0]["bento"]["repository"]["name"] == "unknown"
+    assert rev["targets"][0]["bento"]["name"] == "unknown"
+    assert rev["targets"][0]["config"]["services"] == {}
+    assert rev["targets"][0]["config"]["envs"] == []
+
+
+def test_build_latest_revision_from_cr_bento_colonless():
+    cr = {"spec": {"dynamoGraph": "justrepo"}}
+    rev = build_latest_revision_from_cr(cr)
+    assert rev["targets"][0]["bento"]["repository"]["name"] == "unknown"
+    assert rev["targets"][0]["bento"]["name"] == "unknown"

deploy/sdk/src/dynamo/sdk/cli/deployment.py

@@ -68,33 +68,85 @@ def raise_deployment_config_error(err: BentoMLException, action: str) -> t.NoRet
     ) from None
 
 
-@inject
-def create_deployment(
-    pipeline: Optional[str] = None,
-    name: Optional[str] = None,
+def _get_urls(deployment: Deployment) -> List[str]:
+    """Get URLs from deployment."""
+    latest = deployment._client.v2.get_deployment(deployment.name, deployment.cluster)
+    urls = latest.urls if hasattr(latest, "urls") else None
+    return urls if urls is not None else []
+
+
+def _display_deployment_info(spinner: Spinner, deployment: Deployment) -> None:
+    """Helper function to display deployment status and URLs consistently."""
+    # Get status directly from schema and escape any Rich markup
+    status = deployment._schema.status if deployment._schema.status else "unknown"
+    # Escape any characters that are interpreted as markup
+    reformatted_status = status.replace("[", "\\[")
+    spinner.log(f"[bold]Status:[/] {reformatted_status}")
+
+    # Get URLs directly from schema
+    spinner.log("[bold]Ingress URLs:[/]")
+    try:
+        # Get latest deployment info for URLs
+        urls = _get_urls(deployment)
+        if urls:
+            for url in urls:
+                spinner.log(f"  - {url}")
+        else:
+            spinner.log("    No URLs available")
+    except Exception:
+        # If refresh fails, fall back to existing URLs
+        if deployment._urls:
+            for url in deployment._urls:
+                spinner.log(f"  - {url}")
+        else:
+            spinner.log("    No URLs available")
+
+
+def _build_env_dicts(
     config_file: Optional[TextIO] = None,
-    wait: bool = True,
-    timeout: int = 3600,
-    dev: bool = False,
-    args: Optional[List[str]] = None,
-    envs: Optional[List[str]] = None,
-    _cloud_client: BentoCloudClient = Provide[BentoMLContainer.bentocloud_client],
-) -> Deployment:
-    # Load config from file and serialize to env
+    args: Optional[list[str]] = None,
+    envs: Optional[list[str]] = None,
+) -> list[dict]:
+    """
+    Build a list of environment variable dicts from config file, args, and env strings.
+
+    Args:
+        config_file: Optional configuration file
+        args: Optional list of extra arguments
+        envs: Optional list of environment variable strings (KEY=VALUE)
+
+    Returns:
+        List of dicts suitable for use as envs
+    """
     service_configs = resolve_service_config(config_file=config_file, args=args)
     env_dicts = []
     if service_configs:
         config_json = json.dumps(service_configs)
         logger.info(f"Deployment service configuration: {config_json}")
         env_dicts.append({"name": "DYN_DEPLOYMENT_CONFIG", "value": config_json})
-
-    # Add user-supplied envs
     if envs:
         for env in envs:
             if "=" not in env:
                 raise CLIException(f"Invalid env format: {env}. Use KEY=VALUE.")
             key, value = env.split("=", 1)
             env_dicts.append({"name": key, "value": value})
+    return env_dicts
+
+
+@inject
+def create_deployment(
+    pipeline: Optional[str] = None,
+    name: Optional[str] = None,
+    config_file: Optional[TextIO] = None,
+    wait: bool = True,
+    timeout: int = 3600,
+    dev: bool = False,
+    args: Optional[List[str]] = None,
+    envs: Optional[List[str]] = None,
+    _cloud_client: BentoCloudClient = Provide[BentoMLContainer.bentocloud_client],
+) -> Deployment:
+    # Build env_dicts from config_file, args, and envs
+    env_dicts = _build_env_dicts(config_file=config_file, args=args, envs=envs)
 
     config_params = DeploymentConfigParameters(
         name=name,
@@ -152,38 +204,54 @@ def create_deployment(
             sys.exit(1)
 
 
-def _get_urls(deployment: Deployment) -> List[str]:
-    """Get URLs from deployment."""
-    latest = deployment._client.v2.get_deployment(deployment.name, deployment.cluster)
-    urls = latest.urls if hasattr(latest, "urls") else None
-    return urls if urls is not None else []
-
+@inject
+def update_deployment(
+    name: str,
+    config_file: Optional[TextIO] = None,
+    args: Optional[List[str]] = None,
+    envs: Optional[List[str]] = None,
+    _cloud_client: BentoCloudClient = Provide[BentoMLContainer.bentocloud_client],
+) -> Deployment:
+    """Update an existing deployment on Dynamo Cloud.
 
-def _display_deployment_info(spinner: Spinner, deployment: Deployment) -> None:
-    """Helper function to display deployment status and URLs consistently."""
-    # Get status directly from schema and escape any Rich markup
-    status = deployment._schema.status if deployment._schema.status else "unknown"
-    # Escape any characters that are interpreted as markup
-    reformatted_status = status.replace("[", "\\[")
-    spinner.log(f"[bold]Status:[/] {reformatted_status}")
+    Args:
+        name: The name of the deployment to update
+        config_file: Optional configuration file for the update
+        args: Optional extra arguments for config
+        envs: Optional list of environment variables (KEY=VALUE)
 
-    # Get URLs directly from schema
-    spinner.log("[bold]Ingress URLs:[/]")
+    Returns:
+        Deployment: The updated deployment object
+    """
+    # Build env_dicts from config_file, args, and envs
+    env_dicts = _build_env_dicts(config_file=config_file, args=args, envs=envs)
+    config_params = DeploymentConfigParameters(
+        name=name,
+        envs=env_dicts,
+        cli=True,
+    )
     try:
-        # Get latest deployment info for URLs
-        urls = _get_urls(deployment)
-        if urls:
-            for url in urls:
-                spinner.log(f"  - {url}")
-        else:
-            spinner.log("    No URLs available")
-    except Exception:
-        # If refresh fails, fall back to existing URLs
-        if deployment._urls:
-            for url in deployment._urls:
-                spinner.log(f"  - {url}")
-        else:
-            spinner.log("    No URLs available")
+        config_params.verify(create=False)
+    except BentoMLException as e:
+        print(f"Error: {str(e)}")
+        sys.exit(1)
+    with Spinner(console=console) as spinner:
+        try:
+            spinner.update(f'Updating deployment "{name}" on Dynamo Cloud...')
+            deployment = _cloud_client.deployment.update(
+                deployment_config_params=config_params
+            )
+            spinner.log(
+                f':white_check_mark: Updated deployment "{deployment.name}" in cluster "{deployment.cluster}"'
+            )
+            spinner.log(
+                "[yellow]Update submitted. It may take a short time for the new pods to become active. Please wait a bit before accessing the deployment to ensure your changes are live.[/yellow]"
+            )
+            _display_deployment_info(spinner, deployment)
+            return deployment
+        except BentoMLException as e:
+            spinner.log(f"[red]:x: Error:[/] Failed to update deployment: {str(e)}")
+            sys.exit(1)
 
 
 @inject
@@ -343,8 +411,8 @@ def get(
     get_deployment(name, cluster=cluster)
 
 
-@app.command()
-def list(
+@app.command("list")
+def list_deployments_command(
     cluster: Optional[str] = typer.Option(None, "--cluster", help="Cluster name"),
     search: Optional[str] = typer.Option(None, "--search", help="Search query"),
     dev: bool = typer.Option(False, "--dev", help="List development deployments"),
@@ -367,6 +435,33 @@ def list(
     list_deployments(cluster=cluster, search=search, dev=dev, q=query)
 
 
+@app.command()
+def update(
+    name: str = typer.Argument(..., help="Deployment name to update"),
+    config_file: Optional[typer.FileText] = typer.Option(
+        None, "--config-file", "-f", help="Configuration file path"
+    ),
+    envs: Optional[List[str]] = typer.Option(
+        None,
+        "--env",
+        help="Environment variable(s) to set (format: KEY=VALUE). Note: These environment variables will be set on ALL services in your Dynamo pipeline.",
+    ),
+    endpoint: str = typer.Option(
+        ..., "--endpoint", "-e", help="Dynamo Cloud endpoint", envvar="DYNAMO_CLOUD"
+    ),
+) -> None:
+    """Update an existing deployment on Dynamo Cloud.
+
+    Update a deployment using parameters or a config yaml file.
+    """
+    login_to_cloud(endpoint)
+    update_deployment(
+        name=name,
+        config_file=config_file,
+        envs=envs,
+    )
+
+
 @app.command()
 def delete(
     name: str = typer.Argument(..., help="Deployment name"),

docs/guides/dynamo_deploy/operator_deployment.md

@@ -104,7 +104,7 @@ Deploy your service using the Dynamo deployment command:
 export DEPLOYMENT_NAME=hello-world
 
 # Create the deployment
-dynamo deployment create $DYNAMO_TAG --no-wait -n $DEPLOYMENT_NAME
+dynamo deployment create $DYNAMO_TAG -n $DEPLOYMENT_NAME
 ```
 
 #### Managing Deployments
@@ -124,6 +124,12 @@ To get detailed information about a specific deployment:
 dynamo deployment get $DEPLOYMENT_NAME
 ```
 
+To update a specific deployment:
+
+```bash
+dynamo deployment update $DEPLOYMENT_NAME [--config-file FILENAME] [--env ENV_VAR]
+```
+
 To remove a deployment and all its associated resources:
 
 ```bash