done

- [ ] **Add tests and docs**: If you're adding a new integration, you
must include:
1. A test for the integration, preferably unit tests that do not rely on
network access,
2. An example notebook showing its use. It lives in
`docs/docs/integrations` directory.

- [ ] **Lint and test**: Run `make format`, `make lint` and `make test`
from the root of the package(s) you've modified. We will not consider a
PR unless these three are passing in CI. See [contribution
guidelines](https://docs.langchain.com/oss/python/contributing/overview)
for more.

Additional guidelines:

- Make sure optional dependencies are imported within a function.
- Please do not add dependencies to `pyproject.toml` files (even
optional ones) unless they are **required** for unit tests.
- Most PRs should not touch more than one package.
- Changes should be backwards compatible.

libs/cli/langgraph_cli/api_version.py

@@ -0,0 +1,95 @@
+"""Resolve the LangGraph API version from CLI flags and langgraph.json."""
+
+import json
+import pathlib
+import re
+import urllib.request
+
+import click
+
+VERSION_MARKER_REPO = "langchain/langgraph-published-version-marker"
+_PATCH_VERSION_RE = re.compile(r"^\d+\.\d+\.\d+$")
+_SEMVER_RE = re.compile(r"^\d+\.\d+\.\d+")
+
+
+def resolve_langgraph_api_version(
+    config_path: pathlib.Path,
+    api_version_cli_param: str | None,
+) -> str:
+    """Resolve the API version from the CLI flag and/or langgraph.json.
+
+    Returns the resolved patch-level version string. When neither source
+    provides a version, the latest published version is fetched from Docker Hub.
+
+    Raises `click.ClickException` when both sources specify a version, or
+    when a version cannot be resolved via Docker Hub.
+    """
+    api_version_langgraph_json = _read_api_version_from_config(config_path)
+
+    if api_version_cli_param and api_version_langgraph_json:
+        raise click.ClickException(
+            "API version specified in both --api-version CLI flag "
+            f"({api_version_cli_param!r}) and langgraph.json "
+            f"({api_version_langgraph_json!r}). Please use only one."
+        )
+
+    preferred_api_version = api_version_cli_param or api_version_langgraph_json
+
+    if preferred_api_version and _PATCH_VERSION_RE.match(preferred_api_version):
+        return preferred_api_version
+
+    version_prefix = preferred_api_version or ""
+    if version_prefix:
+        click.secho(
+            f"Resolving API version matching {version_prefix!r} from Docker Hub...",
+            fg="cyan",
+        )
+    else:
+        click.secho(
+            "Resolving latest API version from Docker Hub...",
+            fg="cyan",
+        )
+    resolved = _fetch_matching_version(version_prefix)
+    click.secho(f"Resolved API version: {resolved}", fg="cyan")
+    return resolved
+
+
+def _read_api_version_from_config(config_path: pathlib.Path) -> str | None:
+    """Read the `api_version` field from langgraph.json (if present)."""
+    try:
+        with open(config_path) as f:
+            raw_config = json.load(f)
+    except (OSError, json.JSONDecodeError):
+        return None
+    return raw_config.get("api_version")
+
+
+def _fetch_matching_version(version_prefix: str = "") -> str:
+    """Query Docker Hub for the latest patch version matching *version_prefix*.
+
+    When *version_prefix* is empty, returns the latest published version.
+    """
+    url = f"https://hub.docker.com/v2/repositories/{VERSION_MARKER_REPO}/tags/?page_size=10"
+    if version_prefix:
+        url += f"&name={version_prefix}"
+    try:
+        with urllib.request.urlopen(url, timeout=10) as resp:
+            data = json.loads(resp.read())
+    except Exception as exc:
+        raise click.ClickException(
+            f"Failed to fetch API version from {VERSION_MARKER_REPO}: {exc}\n"
+            "You can specify an exact version with --api-version (e.g. 0.7.67)."
+        ) from exc
+
+    for tag in data.get("results", []):
+        name = tag.get("name", "")
+        if _SEMVER_RE.match(name):
+            return name
+
+    if version_prefix:
+        msg = f"Could not find a version matching {version_prefix!r} in {VERSION_MARKER_REPO}."
+    else:
+        msg = f"Could not find a published version in {VERSION_MARKER_REPO}."
+    raise click.ClickException(
+        f"{msg}\nYou can specify an exact version with --api-version (e.g. 0.7.67)."
+    )

libs/cli/langgraph_cli/cli.py

@@ -13,6 +13,7 @@ import tempfile
 import time
 from collections.abc import Callable, Sequence
 from contextlib import contextmanager
+from typing import Any
 
 import click
 import click.exceptions
@@ -22,9 +23,11 @@ from dotenv import dotenv_values
 import langgraph_cli.config
 import langgraph_cli.docker
 from langgraph_cli.analytics import log_command
+from langgraph_cli.api_version import resolve_langgraph_api_version
 from langgraph_cli.config import Config
 from langgraph_cli.constants import DEFAULT_CONFIG, DEFAULT_PORT
 from langgraph_cli.docker import DockerCapabilities
+from langgraph_cli.engine_runtime_mode import resolve_engine_runtime_mode
 from langgraph_cli.exec import Runner, subp_exec
 from langgraph_cli.host_backend import HostBackendClient, HostBackendError
 from langgraph_cli.progress import Progress
@@ -290,8 +293,8 @@ OPT_API_VERSION = click.option(
 OPT_ENGINE_RUNTIME_MODE = click.option(
     "--engine-runtime-mode",
     type=click.Choice(["combined_queue_worker", "distributed"]),
-    default="combined_queue_worker",
+    default=None,
-    help="Runtime mode. 'distributed' uses separate executor and orchestrator containers.",
+    help="Runtime mode. 'distributed' uses separate executor and orchestrator containers. Defaults to distributed.",
 )
 
 
@@ -347,10 +350,14 @@ def up(
     debugger_base_url: str | None,
     postgres_uri: str | None,
     api_version: str | None,
-    engine_runtime_mode: str,
+    engine_runtime_mode: str | None,
     image: str | None,
     base_image: str | None,
 ):
+    api_version = resolve_langgraph_api_version(config, api_version)
+    engine_runtime_mode = resolve_engine_runtime_mode(
+        config, api_version, engine_runtime_mode
+    )
     click.secho("Starting LangGraph API server...", fg="green")
     click.secho(
         """For local dev, requires env var LANGSMITH_API_KEY with access to LangSmith Deployment.
@@ -550,12 +557,16 @@ def build(
     docker_build_args: Sequence[str],
     base_image: str | None,
     api_version: str | None,
-    engine_runtime_mode: str,
+    engine_runtime_mode: str | None,
     pull: bool,
     tag: str,
     install_command: str | None,
     build_command: str | None,
 ):
+    api_version = resolve_langgraph_api_version(config, api_version)
+    engine_runtime_mode = resolve_engine_runtime_mode(
+        config, api_version, engine_runtime_mode
+    )
     if install_command and langgraph_cli.config.has_disallowed_build_command_content(
         install_command
     ):
@@ -688,6 +699,8 @@ def deploy(
     no_wait: bool,
     docker_build_args: Sequence[str],
 ):
+    api_version = resolve_langgraph_api_version(config, api_version)
+    engine_runtime_mode = resolve_engine_runtime_mode(config, api_version, None)
     click.secho(
         "Note: 'langgraph deploy' is in beta. Expect frequent updates and improvements.",
         fg="yellow",
@@ -858,13 +871,16 @@ def deploy(
 
         if needs_creation:
             log_step(f"{step}. Creating deployment '{name}'")
-            payload = {
+            payload: dict[str, Any] = {
                 "name": name,
                 "source": "internal_docker",
                 "source_config": {"deployment_type": deployment_type},
                 "source_revision_config": {},
                 "secrets": secrets,
+                "engine_runtime_mode": engine_runtime_mode,
             }
+            if api_version:
+                payload["deployed_api_version"] = api_version
             created = client.create_deployment(payload)
             created_id = created.get("id") if isinstance(created, dict) else None
             if not isinstance(created_id, str) or not created_id:
@@ -973,7 +989,13 @@ def deploy(
 
         # -- Step: Update deployment --
         log_step(f"{step}. Updating deployment {deployment_id}")
-        updated = client.update_deployment(deployment_id, remote_image, secrets=secrets)
+        updated = client.update_deployment(
+            deployment_id,
+            remote_image,
+            secrets=secrets,
+            engine_runtime_mode=engine_runtime_mode,
+            deployed_api_version=api_version,
+        )
         tenant_id = updated.get("tenant_id") if isinstance(updated, dict) else None
         if tenant_id:
             status_url = (
@@ -1161,8 +1183,12 @@ def dockerfile(
     add_docker_compose: bool,
     base_image: str | None = None,
     api_version: str | None = None,
-    engine_runtime_mode: str = "combined_queue_worker",
+    engine_runtime_mode: str | None = None,
 ) -> None:
+    api_version = resolve_langgraph_api_version(config, api_version)
+    engine_runtime_mode = resolve_engine_runtime_mode(
+        config, api_version, engine_runtime_mode
+    )
     save_path = pathlib.Path(save_path).absolute()
     secho(f"🔍 Validating configuration at path: {config}", fg="yellow")
     config_json = langgraph_cli.config.validate_config_file(config)
@@ -1511,6 +1537,7 @@ def prepare(
     """Prepare the arguments and stdin for running the LangGraph API server."""
     config_json = langgraph_cli.config.validate_config_file(config_path)
     warn_non_wolfi_distro(config_json)
+
     # pull latest images
     if pull:
         runner.run(
@@ -1535,6 +1562,14 @@ def prepare(
                     verbose=verbose,
                 )
             )
+            runner.run(
+                subp_exec(
+                    "docker",
+                    "pull",
+                    f"langchain/langgraph-orchestrator-licensed:{api_version}",
+                    verbose=verbose,
+                )
+            )
 
     args, stdin = prepare_args_and_stdin(
         capabilities=capabilities,

libs/cli/langgraph_cli/config.py

@@ -1269,6 +1269,11 @@ def docker_tag(
     version_distro_tag = f"{version}{distro_tag}"
 
     # Prepend API version if provided
+    # Strip an existing tag from base_image so we don't produce two colons
+    # (e.g. "langchain/langgraph-server:0.2" → "langchain/langgraph-server").
+    if ":" in base_image:
+        base_image = base_image.rsplit(":", 1)[0]
+
     if api_version:
         full_tag = f"{api_version}-{language}{version_distro_tag}"
     elif "/langgraph-server" in base_image and version_distro_tag not in base_image:
@@ -1424,7 +1429,7 @@ def config_to_compose(
 
             postgres_uri = "postgres://postgres:postgres@langgraph-postgres:5432/postgres?sslmode=disable"
             result += f"""    langgraph-orchestrator:
-        image: langchain/langgraph-orchestrator-licensed:latest
+        image: langchain/langgraph-orchestrator-licensed:{api_version}
         depends_on:
             langgraph-api:
                 condition: service_healthy

libs/cli/langgraph_cli/engine_runtime_mode.py

@@ -0,0 +1,69 @@
+"""Resolve the LangGraph engine runtime mode."""
+
+import json
+import pathlib
+
+import click
+
+_DISTRIBUTED_MIN_VERSION = (0, 7, 68)
+
+
+def resolve_engine_runtime_mode(
+    config_path: pathlib.Path,
+    api_version: str,
+    engine_runtime_mode_cli_param: str | None,
+) -> str:
+    """Resolve the engine runtime mode.
+
+    *api_version* must already be resolved to a patch-level semver string.
+
+    Returns ``"distributed"`` or ``"combined_queue_worker"``.
+
+    Raises `click.ClickException` when distributed mode is requested but
+    not supported (JavaScript project or api_version <= 0.7.67).
+    """
+    requires_combined = _requires_combined(config_path, api_version)
+
+    if engine_runtime_mode_cli_param == "distributed":
+        if requires_combined:
+            reasons = _constraint_reasons(config_path, api_version)
+            raise click.ClickException(
+                f"Distributed runtime is not supported for {' and '.join(reasons)}."
+            )
+        return "distributed"
+
+    if engine_runtime_mode_cli_param == "combined_queue_worker":
+        return "combined_queue_worker"
+
+    # No explicit choice → default to distributed
+    return "distributed"
+
+
+def _requires_combined(config_path: pathlib.Path, api_version: str) -> bool:
+    return _is_javascript_project(config_path) or _version_too_old(api_version)
+
+
+def _version_too_old(api_version: str) -> bool:
+    try:
+        parts = tuple(int(x) for x in api_version.split("."))
+    except (ValueError, AttributeError):
+        return True
+    return parts < _DISTRIBUTED_MIN_VERSION
+
+
+def _is_javascript_project(config_path: pathlib.Path) -> bool:
+    try:
+        with open(config_path) as f:
+            cfg = json.load(f)
+    except (OSError, json.JSONDecodeError):
+        return False
+    return bool(cfg.get("node_version")) and not cfg.get("python_version")
+
+
+def _constraint_reasons(config_path: pathlib.Path, api_version: str) -> list[str]:
+    reasons: list[str] = []
+    if _is_javascript_project(config_path):
+        reasons.append("JavaScript projects")
+    if _version_too_old(api_version):
+        reasons.append(f"API version {api_version} (<= 0.7.67)")
+    return reasons

libs/cli/langgraph_cli/host_backend.py

@@ -82,12 +82,18 @@ class HostBackendClient:
         deployment_id: str,
         image_uri: str,
         secrets: list[dict[str, str]] | None = None,
+        engine_runtime_mode: str | None = None,
+        deployed_api_version: str | None = None,
     ) -> dict[str, Any]:
         payload: dict[str, Any] = {
             "source_revision_config": {"image_uri": image_uri},
         }
         if secrets is not None:
             payload["secrets"] = secrets
+        if engine_runtime_mode is not None:
+            payload["engine_runtime_mode"] = engine_runtime_mode
+        if deployed_api_version is not None:
+            payload["deployed_api_version"] = deployed_api_version
         return self._request(
             "PATCH",
             f"/v2/deployments/{deployment_id}",

libs/cli/tests/unit_tests/cli/test_cli.py

@@ -382,9 +382,10 @@ def test_dockerfile_command_with_base_image() -> None:
         assert save_path.exists()
         with open(save_path) as f:
             dockerfile = f.read()
-            assert re.match("FROM langchain/langgraph-server:0.2-py3.*", dockerfile), (
+            assert re.match(
-                "\n".join(dockerfile.splitlines()[:3])
+                r"FROM langchain/langgraph-server:\d+\.\d+\.\d+-py3\..*",
-            )
+                dockerfile,
+            ), "\n".join(dockerfile.splitlines()[:3])
 
 
 def test_dockerfile_command_with_docker_compose() -> None:
@@ -567,6 +568,8 @@ def test_build_generate_proper_build_context():
                     "test-image",
                     "--config",
                     str(temp_dir / "config.json"),
+                    "--engine-runtime-mode",
+                    "combined_queue_worker",
                 ],
                 catch_exceptions=True,
             )
@@ -602,6 +605,8 @@ def test_dockerfile_command_with_api_version() -> None:
                 str(temp_dir / "config.json"),
                 "--api-version",
                 "0.2.74",
+                "--engine-runtime-mode",
+                "combined_queue_worker",
             ],
         )
 
@@ -718,6 +723,8 @@ def test_build_command_with_api_version() -> None:
                     str(temp_dir / "config.json"),
                     "--api-version",
                     "0.2.74",
+                    "--engine-runtime-mode",
+                    "combined_queue_worker",
                     "--no-pull",  # Avoid pulling non-existent images
                 ],
                 catch_exceptions=True,
@@ -856,7 +863,10 @@ def test_dockerfile_command_distributed_mode() -> None:
         assert save_path.exists()
         with open(save_path) as f:
             dockerfile = f.read()
-            assert "FROM langchain/langgraph-executor:3.11" in dockerfile
+            assert re.search(
+                r"FROM langchain/langgraph-executor:\d+\.\d+\.\d+-py3\.11",
+                dockerfile,
+            ), dockerfile.splitlines()[0]
 
 
 def test_dockerfile_command_combined_mode() -> None:
@@ -889,7 +899,10 @@ def test_dockerfile_command_combined_mode() -> None:
         assert save_path.exists()
         with open(save_path) as f:
             dockerfile = f.read()
-            assert "FROM langchain/langgraph-api:3.11" in dockerfile
+            assert re.search(
+                r"FROM langchain/langgraph-api:\d+\.\d+\.\d+-py3\.11",
+                dockerfile,
+            ), dockerfile.splitlines()[0]
 
 
 def test_dockerfile_command_distributed_with_explicit_base_image() -> None:
@@ -924,7 +937,10 @@ def test_dockerfile_command_distributed_with_explicit_base_image() -> None:
         assert save_path.exists()
         with open(save_path) as f:
             dockerfile = f.read()
-            assert "FROM my-custom-executor:latest" in dockerfile
+            assert re.search(
+                r"FROM my-custom-executor:\d+\.\d+\.\d+-py3\.11",
+                dockerfile,
+            ), dockerfile.splitlines()[0]
 
 
 def test_prepare_args_and_stdin_distributed_mode() -> None:
@@ -943,18 +959,42 @@ def test_prepare_args_and_stdin_distributed_mode() -> None:
         port=port,
         watch=False,
         engine_runtime_mode="distributed",
+        api_version="0.7.67",
     )
 
-    # API service should use langgraph-api base image
+    # API service should use langgraph-api base image with pinned version
-    assert "FROM langchain/langgraph-api:" in actual_stdin
+    assert "FROM langchain/langgraph-api:0.7.67-py3.11" in actual_stdin
 
     # Distributed mode sets N_JOBS_PER_WORKER=0 on the API service
     assert 'N_JOBS_PER_WORKER: "0"' in actual_stdin
 
-    # Orchestrator service present
+    # Orchestrator service present with pinned version
     assert "langgraph-orchestrator:" in actual_stdin
+    assert "langchain/langgraph-orchestrator-licensed:0.7.67" in actual_stdin
 
     # Executor service present with correct base image
     assert "langgraph-executor:" in actual_stdin
-    assert "FROM langchain/langgraph-executor:" in actual_stdin
+    assert "FROM langchain/langgraph-executor:0.7.67-py3.11" in actual_stdin
     assert "executor_entrypoint.sh" in actual_stdin
+
+
+def test_prepare_args_and_stdin_distributed_with_api_version() -> None:
+    """All 3 images should use the same api_version in distributed mode."""
+    config_path = pathlib.Path(__file__).parent / "langgraph.json"
+    config = validate_config(
+        Config(dependencies=["."], graphs={"agent": "agent.py:graph"})
+    )
+    actual_args, actual_stdin = prepare_args_and_stdin(
+        capabilities=DEFAULT_DOCKER_CAPABILITIES,
+        config_path=config_path,
+        config=config,
+        docker_compose=None,
+        port=8000,
+        watch=False,
+        engine_runtime_mode="distributed",
+        api_version="0.7.67",
+    )
+
+    assert "FROM langchain/langgraph-api:0.7.67-py3.11" in actual_stdin
+    assert "FROM langchain/langgraph-executor:0.7.67-py3.11" in actual_stdin
+    assert "langchain/langgraph-orchestrator-licensed:0.7.67" in actual_stdin

libs/cli/tests/unit_tests/test_api_version.py

@@ -0,0 +1,156 @@
+import io
+import json
+import pathlib
+import urllib.error
+import urllib.request
+
+import click
+import pytest
+
+from langgraph_cli.api_version import (
+    _fetch_matching_version,
+    resolve_langgraph_api_version,
+)
+
+
+@pytest.fixture()
+def config_dir(tmp_path: pathlib.Path) -> pathlib.Path:
+    return tmp_path
+
+
+def _write_config(
+    config_dir: pathlib.Path, api_version: str | None = None
+) -> pathlib.Path:
+    cfg: dict = {"dependencies": ["."], "graphs": {"agent": "agent.py:graph"}}
+    if api_version is not None:
+        cfg["api_version"] = api_version
+    path = config_dir / "langgraph.json"
+    path.write_text(json.dumps(cfg))
+    return path
+
+
+class TestResolveLanggraphApiVersion:
+    def test_exact_patch_from_cli(self, config_dir: pathlib.Path) -> None:
+        path = _write_config(config_dir)
+        assert resolve_langgraph_api_version(path, "0.7.67") == "0.7.67"
+
+    def test_exact_patch_from_json(self, config_dir: pathlib.Path) -> None:
+        path = _write_config(config_dir, api_version="0.7.67")
+        assert resolve_langgraph_api_version(path, None) == "0.7.67"
+
+    def test_neither_source_fetches_latest(
+        self, config_dir: pathlib.Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        path = _write_config(config_dir)
+
+        fake_body = json.dumps(
+            {"results": [{"name": "latest"}, {"name": "0.9.2"}]}
+        ).encode()
+
+        def mock_urlopen(url, *, timeout=None):
+            assert "name=" not in url
+            return io.BytesIO(fake_body)
+
+        monkeypatch.setattr(urllib.request, "urlopen", mock_urlopen)
+        assert resolve_langgraph_api_version(path, None) == "0.9.2"
+
+    def test_both_sources_raises(self, config_dir: pathlib.Path) -> None:
+        path = _write_config(config_dir, api_version="0.7.67")
+        with pytest.raises(click.ClickException, match="both"):
+            resolve_langgraph_api_version(path, "0.8.0")
+
+    def test_partial_version_resolves_from_dockerhub(
+        self, config_dir: pathlib.Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        path = _write_config(config_dir, api_version="0.7")
+
+        fake_body = json.dumps(
+            {"results": [{"name": "latest"}, {"name": "0.7.67"}]}
+        ).encode()
+
+        def mock_urlopen(url, *, timeout=None):
+            assert "name=0.7" in url
+            return io.BytesIO(fake_body)
+
+        monkeypatch.setattr(urllib.request, "urlopen", mock_urlopen)
+        assert resolve_langgraph_api_version(path, None) == "0.7.67"
+
+    def test_partial_cli_version_resolves_from_dockerhub(
+        self, config_dir: pathlib.Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        path = _write_config(config_dir)
+
+        fake_body = json.dumps(
+            {"results": [{"name": "0.8.1"}, {"name": "0.8.0"}]}
+        ).encode()
+
+        def mock_urlopen(url, *, timeout=None):
+            assert "name=0.8" in url
+            return io.BytesIO(fake_body)
+
+        monkeypatch.setattr(urllib.request, "urlopen", mock_urlopen)
+        assert resolve_langgraph_api_version(path, "0.8") == "0.8.1"
+
+    def test_missing_config_file_fetches_latest(
+        self, config_dir: pathlib.Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        path = config_dir / "nonexistent.json"
+
+        fake_body = json.dumps({"results": [{"name": "0.9.2"}]}).encode()
+
+        def mock_urlopen(url, *, timeout=None):
+            return io.BytesIO(fake_body)
+
+        monkeypatch.setattr(urllib.request, "urlopen", mock_urlopen)
+        assert resolve_langgraph_api_version(path, None) == "0.9.2"
+
+    def test_missing_config_file_with_cli_version(
+        self, config_dir: pathlib.Path
+    ) -> None:
+        path = config_dir / "nonexistent.json"
+        assert resolve_langgraph_api_version(path, "0.7.67") == "0.7.67"
+
+
+class TestFetchMatchingVersion:
+    def test_empty_prefix_returns_latest(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        fake_body = json.dumps(
+            {"results": [{"name": "latest"}, {"name": "0.9.2"}, {"name": "abc123"}]}
+        ).encode()
+
+        def mock_urlopen(url, *, timeout=None):
+            assert "name=" not in url
+            return io.BytesIO(fake_body)
+
+        monkeypatch.setattr(urllib.request, "urlopen", mock_urlopen)
+        assert _fetch_matching_version() == "0.9.2"
+
+    def test_returns_first_semver(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        fake_body = json.dumps(
+            {"results": [{"name": "latest"}, {"name": "abc1234"}, {"name": "0.7.67"}]}
+        ).encode()
+
+        def mock_urlopen(url, *, timeout=None):
+            return io.BytesIO(fake_body)
+
+        monkeypatch.setattr(urllib.request, "urlopen", mock_urlopen)
+        assert _fetch_matching_version("0.7") == "0.7.67"
+
+    def test_no_semver_raises(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        fake_body = json.dumps(
+            {"results": [{"name": "latest"}, {"name": "abc1234"}]}
+        ).encode()
+
+        def mock_urlopen(url, *, timeout=None):
+            return io.BytesIO(fake_body)
+
+        monkeypatch.setattr(urllib.request, "urlopen", mock_urlopen)
+        with pytest.raises(click.ClickException, match="Could not find a version"):
+            _fetch_matching_version("0.7")
+
+    def test_network_error_raises(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        def mock_urlopen(url, *, timeout=None):
+            raise urllib.error.URLError("connection refused")
+
+        monkeypatch.setattr(urllib.request, "urlopen", mock_urlopen)
+        with pytest.raises(click.ClickException, match="Failed to fetch"):
+            _fetch_matching_version("0.7")

libs/cli/tests/unit_tests/test_config.py

@@ -1772,19 +1772,23 @@ def test_config_to_compose_distributed_mode():
         validate_config({"dependencies": ["."], "graphs": graphs}),
         "langchain/langgraph-api",
         engine_runtime_mode="distributed",
+        api_version="0.7.67",
     )
 
     # API service uses langchain/langgraph-api base image
-    assert "FROM langchain/langgraph-api:3.11" in actual_compose_stdin
+    assert "FROM langchain/langgraph-api:0.7.67-py3.11" in actual_compose_stdin
 
-    # Orchestrator service is present
+    # Orchestrator service is present with pinned version
     assert "langgraph-orchestrator:" in actual_compose_stdin
+    assert "langchain/langgraph-orchestrator-licensed:0.7.67" in actual_compose_stdin
     assert "EXECUTOR_TARGET: langgraph-executor:8188" in actual_compose_stdin
 
     # Executor service is present with correct base image
     assert "langgraph-executor:" in actual_compose_stdin
-    assert "FROM langchain/langgraph-executor:3.11" in actual_compose_stdin
+    assert "FROM langchain/langgraph-executor:0.7.67-py3.11" in actual_compose_stdin
-    assert 'entrypoint: ["sh", "/storage/executor_entrypoint.sh"]' in actual_compose_stdin
+    assert (
+        'entrypoint: ["sh", "/storage/executor_entrypoint.sh"]' in actual_compose_stdin
+    )
 
     # Executor has required environment variables
     assert "EXECUTOR_GRPC_PORT:" in actual_compose_stdin
@@ -1802,6 +1806,7 @@ def test_config_to_compose_distributed_mode_with_env_file():
         validate_config({"dependencies": ["."], "graphs": graphs, "env": ".env"}),
         "langchain/langgraph-api",
         engine_runtime_mode="distributed",
+        api_version="0.7.67",
     )
 
     # env_file should appear multiple times: API, orchestrator, executor
@@ -1820,6 +1825,7 @@ def test_config_to_compose_distributed_mode_generates_two_dockerfiles():
         validate_config({"dependencies": ["."], "graphs": graphs}),
         "langchain/langgraph-api",
         engine_runtime_mode="distributed",
+        api_version="0.7.67",
     )
 
     # Should contain two different FROM lines
@@ -1829,8 +1835,8 @@ def test_config_to_compose_distributed_mode_generates_two_dockerfiles():
         if line.strip().startswith("FROM ")
     ]
     assert len(from_lines) == 2
-    assert "FROM langchain/langgraph-api:3.11" in from_lines[0]
+    assert "FROM langchain/langgraph-api:0.7.67-py3.11" in from_lines[0]
-    assert "FROM langchain/langgraph-executor:3.11" in from_lines[1]
+    assert "FROM langchain/langgraph-executor:0.7.67-py3.11" in from_lines[1]
 
 
 def test_config_to_compose_combined_mode_no_orchestrator():
@@ -1869,6 +1875,7 @@ def test_config_to_compose_distributed_executor_gets_correct_paths():
         validate_config({"dependencies": ["."], "graphs": graphs}),
         "langchain/langgraph-api",
         engine_runtime_mode="distributed",
+        api_version="0.7.67",
     )
 
     # Both API and executor Dockerfiles should contain valid LANGSERVE_GRAPHS
@@ -1884,6 +1891,34 @@ def test_config_to_compose_distributed_executor_gets_correct_paths():
     )
 
 
+def test_config_to_compose_distributed_orchestrator_uses_api_version():
+    """Orchestrator image tag should use the api_version when provided."""
+    graphs = {"agent": "./agent.py:graph"}
+    actual = config_to_compose(
+        PATH_TO_CONFIG,
+        validate_config({"dependencies": ["."], "graphs": graphs}),
+        "langchain/langgraph-api",
+        engine_runtime_mode="distributed",
+        api_version="0.7.67",
+    )
+    assert "langchain/langgraph-orchestrator-licensed:0.7.67" in actual
+
+
+def test_config_to_compose_distributed_all_images_same_version():
+    """All 3 images (api, executor, orchestrator) should use the same api_version."""
+    graphs = {"agent": "./agent.py:graph"}
+    actual = config_to_compose(
+        PATH_TO_CONFIG,
+        validate_config({"dependencies": ["."], "graphs": graphs}),
+        "langchain/langgraph-api",
+        engine_runtime_mode="distributed",
+        api_version="0.7.67",
+    )
+    assert "FROM langchain/langgraph-api:0.7.67-py3.11" in actual
+    assert "FROM langchain/langgraph-executor:0.7.67-py3.11" in actual
+    assert "langchain/langgraph-orchestrator-licensed:0.7.67" in actual
+
+
 class TestHasDisallowedBuildCommandContent:
     """Tests for has_disallowed_build_command_content."""
 

libs/cli/tests/unit_tests/test_engine_runtime_mode.py

@@ -0,0 +1,99 @@
+import json
+import pathlib
+
+import click
+import pytest
+
+from langgraph_cli.engine_runtime_mode import resolve_engine_runtime_mode
+
+
+def _write_config(
+    tmp_path: pathlib.Path,
+    *,
+    python_version: str | None = "3.11",
+    node_version: str | None = None,
+) -> pathlib.Path:
+    cfg: dict = {"dependencies": ["."], "graphs": {"agent": "agent.py:graph"}}
+    if python_version is not None:
+        cfg["python_version"] = python_version
+    if node_version is not None:
+        cfg["node_version"] = node_version
+    path = tmp_path / "langgraph.json"
+    path.write_text(json.dumps(cfg))
+    return path
+
+
+class TestResolveEngineRuntimeMode:
+    # -- cli_param == "distributed" -------------------------------------------
+
+    def test_distributed_explicit_new_version(self, tmp_path: pathlib.Path) -> None:
+        path = _write_config(tmp_path)
+        assert (
+            resolve_engine_runtime_mode(path, "0.7.68", "distributed") == "distributed"
+        )
+
+    def test_distributed_explicit_old_version_raises(
+        self, tmp_path: pathlib.Path
+    ) -> None:
+        path = _write_config(tmp_path)
+        with pytest.raises(click.ClickException, match="0.7.67"):
+            resolve_engine_runtime_mode(path, "0.7.67", "distributed")
+
+    def test_distributed_explicit_js_raises(self, tmp_path: pathlib.Path) -> None:
+        path = _write_config(tmp_path, python_version=None, node_version="20")
+        with pytest.raises(click.ClickException, match="JavaScript"):
+            resolve_engine_runtime_mode(path, "0.8.0", "distributed")
+
+    def test_distributed_explicit_js_and_old_version_raises(
+        self, tmp_path: pathlib.Path
+    ) -> None:
+        path = _write_config(tmp_path, python_version=None, node_version="20")
+        with pytest.raises(click.ClickException, match="JavaScript.*0.7.60"):
+            resolve_engine_runtime_mode(path, "0.7.60", "distributed")
+
+    # -- cli_param == "combined_queue_worker" ----------------------------------
+
+    def test_combined_explicit(self, tmp_path: pathlib.Path) -> None:
+        path = _write_config(tmp_path)
+        assert (
+            resolve_engine_runtime_mode(path, "0.8.0", "combined_queue_worker")
+            == "combined_queue_worker"
+        )
+
+    def test_combined_explicit_old_version(self, tmp_path: pathlib.Path) -> None:
+        path = _write_config(tmp_path)
+        assert (
+            resolve_engine_runtime_mode(path, "0.7.67", "combined_queue_worker")
+            == "combined_queue_worker"
+        )
+
+    # -- cli_param is None (default) -------------------------------------------
+
+    def test_default_is_distributed(self, tmp_path: pathlib.Path) -> None:
+        path = _write_config(tmp_path)
+        assert resolve_engine_runtime_mode(path, "0.8.0", None) == "distributed"
+
+    def test_default_old_version(self, tmp_path: pathlib.Path) -> None:
+        path = _write_config(tmp_path)
+        assert resolve_engine_runtime_mode(path, "0.7.67", None) == "distributed"
+
+    def test_default_js(self, tmp_path: pathlib.Path) -> None:
+        path = _write_config(tmp_path, python_version=None, node_version="20")
+        assert resolve_engine_runtime_mode(path, "0.8.0", None) == "distributed"
+
+    # -- edge: version boundary ------------------------------------------------
+
+    def test_version_boundary_0_7_67_blocks_distributed(
+        self, tmp_path: pathlib.Path
+    ) -> None:
+        path = _write_config(tmp_path)
+        with pytest.raises(click.ClickException):
+            resolve_engine_runtime_mode(path, "0.7.67", "distributed")
+
+    def test_version_boundary_0_7_68_allows_distributed(
+        self, tmp_path: pathlib.Path
+    ) -> None:
+        path = _write_config(tmp_path)
+        assert (
+            resolve_engine_runtime_mode(path, "0.7.68", "distributed") == "distributed"
+        )

libs/cli/tests/unit_tests/test_host_backend.py

@@ -152,6 +152,77 @@ def test_update_deployment_no_secrets(client):
     assert result == {"ok": True}
 
 
+def test_update_deployment_with_engine_runtime_mode():
+    """Verify engine_runtime_mode is included in the PATCH payload."""
+    import json
+
+    def handler(req: httpx.Request) -> httpx.Response:
+        body = json.loads(req.content)
+        assert body["engine_runtime_mode"] == "distributed"
+        assert body["source_revision_config"]["image_uri"] == "img:v1"
+        return httpx.Response(200, json={"id": "dep-1"})
+
+    c = HostBackendClient("https://api.example.com", "test-key")
+    c._client = httpx.Client(
+        base_url="https://api.example.com",
+        transport=httpx.MockTransport(handler),
+        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
+        timeout=30,
+    )
+    result = c.update_deployment("dep-1", "img:v1", engine_runtime_mode="distributed")
+    assert result == {"id": "dep-1"}
+
+
+def test_update_deployment_with_deployed_api_version():
+    """Verify deployed_api_version is included in the PATCH payload."""
+    import json
+
+    def handler(req: httpx.Request) -> httpx.Response:
+        body = json.loads(req.content)
+        assert body["deployed_api_version"] == "0.3.5"
+        assert body["engine_runtime_mode"] == "distributed"
+        assert body["source_revision_config"]["image_uri"] == "img:v2"
+        assert body["secrets"] == [{"name": "K", "value": "V"}]
+        return httpx.Response(200, json={"id": "dep-2"})
+
+    c = HostBackendClient("https://api.example.com", "test-key")
+    c._client = httpx.Client(
+        base_url="https://api.example.com",
+        transport=httpx.MockTransport(handler),
+        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
+        timeout=30,
+    )
+    result = c.update_deployment(
+        "dep-2",
+        "img:v2",
+        secrets=[{"name": "K", "value": "V"}],
+        engine_runtime_mode="distributed",
+        deployed_api_version="0.3.5",
+    )
+    assert result == {"id": "dep-2"}
+
+
+def test_update_deployment_omits_none_fields():
+    """Verify None values for optional fields are not sent in payload."""
+    import json
+
+    def handler(req: httpx.Request) -> httpx.Response:
+        body = json.loads(req.content)
+        assert "engine_runtime_mode" not in body
+        assert "deployed_api_version" not in body
+        assert "secrets" not in body
+        return httpx.Response(200, json={"ok": True})
+
+    c = HostBackendClient("https://api.example.com", "test-key")
+    c._client = httpx.Client(
+        base_url="https://api.example.com",
+        transport=httpx.MockTransport(handler),
+        headers={"X-Api-Key": "test-key", "Accept": "application/json"},
+        timeout=30,
+    )
+    c.update_deployment("dep-3", "img:v1")
+
+
 def test_list_revisions(client):
     result = client.list_revisions("dep-123", limit=5)
     assert result == {"ok": True}