feat: topology arg for langgraph up

libs/cli/langgraph_cli/cli.py

@@ -158,6 +158,25 @@ OPT_API_VERSION = click.option(
     help="API server version to use for the base image. If unspecified, the latest version will be used.",
 )
 
+OPT_TOPOLOGY = click.option(
+    "--topology",
+    type=click.Choice(["single", "split"]),
+    default="split",
+    show_default=True,
+    help="""Deployment topology for the LangGraph services.
+
+    \b
+    - 'single': Run API server and queue worker in the same container (legacy behavior)
+    - 'split': Run API server and queue worker in separate containers (recommended)
+    Other topologies (e.g., 'distributed') coming soon.
+
+    \b
+    Example:
+        langgraph up --topology single
+    \b
+    """,
+)
+
 
 @click.group()
 @click.version_option(version=__version__, prog_name="LangGraph CLI")
@@ -176,6 +195,7 @@ def cli():
 @OPT_WATCH
 @OPT_POSTGRES_URI
 @OPT_API_VERSION
+@OPT_TOPOLOGY
 @click.option(
     "--image",
     type=str,
@@ -210,6 +230,7 @@ def up(
     debugger_base_url: str | None,
     postgres_uri: str | None,
     api_version: str | None,
+    topology: str,
     image: str | None,
     base_image: str | None,
 ):
@@ -233,6 +254,7 @@ For production use, requires a license key in env var LANGGRAPH_CLOUD_LICENSE_KE
             debugger_base_url=debugger_base_url,
             postgres_uri=postgres_uri,
             api_version=api_version,
+            topology=topology,
             image=image,
             base_image=base_image,
         )
@@ -557,6 +579,7 @@ def dockerfile(
                 capabilities,
                 port=8123,
                 base_image=base_image,
+                topology="split",
             )
             # Add .env file to the docker-compose.yml for the langgraph-api service
             compose_dict["services"]["langgraph-api"]["env_file"] = [".env"]
@@ -570,6 +593,17 @@ def dockerfile(
                 compose_dict["services"]["langgraph-api"]["build"]["args"] = {
                     "BASE_IMAGE": base_image
                 }
+            # Also configure the worker service with the same build context
+            if "langgraph-worker" in compose_dict["services"]:
+                compose_dict["services"]["langgraph-worker"]["env_file"] = [".env"]
+                compose_dict["services"]["langgraph-worker"]["build"] = {
+                    "context": ".",
+                    "dockerfile": save_path.name,
+                }
+                if base_image:
+                    compose_dict["services"]["langgraph-worker"]["build"]["args"] = {
+                        "BASE_IMAGE": base_image
+                    }
             f.write(langgraph_cli.docker.dict_to_yaml(compose_dict))
             secho("✅ Created: docker-compose.yml", fg="green")
 
@@ -793,6 +827,8 @@ def prepare_args_and_stdin(
     debugger_base_url: str | None = None,
     postgres_uri: str | None = None,
     api_version: str | None = None,
+    # Deployment topology: "single" (combined) or "split" (separate API/worker)
+    topology: str = "split",
     # Like "my-tag" (if you already built it locally)
     image: str | None = None,
     # Like "langchain/langgraphjs-api" or "langchain/langgraph-api
@@ -809,6 +845,7 @@ def prepare_args_and_stdin(
         image=image,  # Pass image to compose YAML generator
         base_image=base_image,
         api_version=api_version,
+        topology=topology,
     )
     args = [
         "--project-directory",
@@ -826,6 +863,8 @@ def prepare_args_and_stdin(
         base_image=langgraph_cli.config.default_base_image(config),
         api_version=api_version,
         image=image,
+        topology=topology,
+        postgres_uri=postgres_uri,
     )
     return args, stdin
 
@@ -844,6 +883,7 @@ def prepare(
     debugger_base_url: str | None = None,
     postgres_uri: str | None = None,
     api_version: str | None = None,
+    topology: str = "split",
     image: str | None = None,
     base_image: str | None = None,
 ) -> tuple[list[str], str]:
@@ -872,6 +912,7 @@ def prepare(
         debugger_base_url=debugger_base_url or f"http://127.0.0.1:{port}",
         postgres_uri=postgres_uri,
         api_version=api_version,
+        topology=topology,
         image=image,
         base_image=base_image,
     )

libs/cli/langgraph_cli/config.py

@@ -1242,6 +1242,8 @@ def config_to_compose(
     api_version: str | None = None,
     image: str | None = None,
     watch: bool = False,
+    topology: str = "split",
+    postgres_uri: str | None = None,
 ) -> str:
     base_image = base_image or default_base_image(config)
 
@@ -1268,11 +1270,20 @@ def config_to_compose(
     else:
         watch_str = ""
     if image:
-        return f"""
+        api_config = f"""
 {textwrap.indent(env_vars_str, "            ")}
         {env_file_str}
         {watch_str}
 """
+        # For split topology with pre-built image, worker also needs env vars
+        if topology == "split":
+            worker_config = f"""
+    langgraph-worker:
+{textwrap.indent(env_vars_str, "            ")}
+        {env_file_str}
+"""
+            return api_config + worker_config
+        return api_config
 
     else:
         dockerfile, additional_contexts = config_to_docker(
@@ -1292,7 +1303,7 @@ def config_to_compose(
             additional_contexts:
 {additional_contexts_str}"""
 
-        return f"""
+        api_config = f"""
 {textwrap.indent(env_vars_str, "            ")}
         {env_file_str}
         pull_policy: build
@@ -1302,3 +1313,43 @@ def config_to_compose(
 {textwrap.indent(dockerfile, "                ")}
         {watch_str}
 """
+        # For split topology, add complete worker service definition with build config
+        # (worker service is NOT added in docker.py when building from Dockerfile)
+        if topology == "split":
+            # Import here to avoid circular dependency
+            from langgraph_cli.docker import DEFAULT_POSTGRES_URI
+
+            # Determine if using internal postgres (include_db) based on postgres_uri
+            include_db = postgres_uri is None
+            effective_postgres_uri = postgres_uri or DEFAULT_POSTGRES_URI
+
+            # Build depends_on based on whether we have internal postgres
+            if include_db:
+                worker_depends_on = """depends_on:
+            langgraph-redis:
+                condition: service_healthy
+            langgraph-postgres:
+                condition: service_healthy"""
+            else:
+                worker_depends_on = """depends_on:
+            langgraph-redis:
+                condition: service_healthy"""
+
+            worker_config = f"""
+    langgraph-worker:
+        {worker_depends_on}
+        environment:
+            REDIS_URI: redis://langgraph-redis:6379
+            POSTGRES_URI: {effective_postgres_uri}
+{textwrap.indent(env_vars_str, "            ")}
+        {env_file_str}
+        entrypoint:
+            - /storage/queue_entrypoint.sh
+        pull_policy: build
+        build:
+            context: .{additional_contexts_str}
+            dockerfile_inline: |
+{textwrap.indent(dockerfile, "                ")}
+"""
+            return api_config + worker_config
+        return api_config

libs/cli/langgraph_cli/docker.py

@@ -20,6 +20,7 @@ class Version(NamedTuple):
 
 
 DockerComposeType = Literal["plugin", "standalone"]
+TopologyType = Literal["single", "split"]
 
 
 class DockerCapabilities(NamedTuple):
@@ -149,6 +150,8 @@ def compose_as_dict(
     base_image: str | None = None,
     # API version of the base image
     api_version: str | None = None,
+    # Deployment topology: "single" (combined) or "split" (separate API/worker)
+    topology: TopologyType = "split",
 ) -> dict:
     """Create a docker compose file as a dictionary in YML style."""
     if postgres_uri is None:
@@ -207,15 +210,20 @@ def compose_as_dict(
         )["langgraph-debugger"]
 
     # Add langgraph-api service
+    api_environment = {
+        "REDIS_URI": "redis://langgraph-redis:6379",
+        "POSTGRES_URI": postgres_uri,
+    }
+    # In split mode, disable queue processing in the API server
+    if topology == "split":
+        api_environment["N_JOBS_PER_WORKER"] = "0"
+
     services["langgraph-api"] = {
         "ports": [f'"{port}:8000"'],
         "depends_on": {
             "langgraph-redis": {"condition": "service_healthy"},
         },
-        "environment": {
+        "environment": api_environment,
-            "REDIS_URI": "redis://langgraph-redis:6379",
-            "POSTGRES_URI": postgres_uri,
-        },
     }
     if image:
         services["langgraph-api"]["image"] = image
@@ -235,6 +243,37 @@ def compose_as_dict(
             "start_period": "10s",
         }
 
+    # Add langgraph-worker service in split mode ONLY when using pre-built image
+    # When building from Dockerfile, the worker service is added in config_to_compose
+    # to avoid duplicate service key issues with YAML concatenation
+    if topology == "split" and image:
+        services["langgraph-worker"] = {
+            "depends_on": {
+                "langgraph-redis": {"condition": "service_healthy"},
+            },
+            "environment": {
+                "REDIS_URI": "redis://langgraph-redis:6379",
+                "POSTGRES_URI": postgres_uri,
+            },
+            "entrypoint": ["/storage/queue_entrypoint.sh"],
+            "image": image,
+        }
+
+        # If Postgres is included, add it to the dependencies of langgraph-worker
+        if include_db:
+            services["langgraph-worker"]["depends_on"]["langgraph-postgres"] = {
+                "condition": "service_healthy"
+            }
+
+        # Additional healthcheck for langgraph-worker if supported
+        if capabilities.healthcheck_start_interval:
+            services["langgraph-worker"]["healthcheck"] = {
+                "test": "python /api/healthcheck.py",
+                "interval": "60s",
+                "start_interval": "1s",
+                "start_period": "10s",
+            }
+
     # Final compose dictionary with volumes included if needed
     compose_dict = {}
     if include_db:
@@ -255,6 +294,8 @@ def compose(
     image: str | None = None,
     base_image: str | None = None,
     api_version: str | None = None,
+    # Deployment topology: "single" (combined) or "split" (separate API/worker)
+    topology: TopologyType = "split",
 ) -> str:
     """Create a docker compose file as a string."""
     compose_content = compose_as_dict(
@@ -266,6 +307,7 @@ def compose(
         image=image,
         base_image=base_image,
         api_version=api_version,
+        topology=topology,
     )
     compose_str = dict_to_yaml(compose_content)
     return compose_str

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

@@ -66,6 +66,7 @@ def test_prepare_args_and_stdin() -> None:
         debugger_port=debugger_port,
         debugger_base_url=debugger_graph_url,
         watch=True,
+        topology="single",
     )
 
     expected_args = [
@@ -189,6 +190,7 @@ def test_prepare_args_and_stdin_with_image() -> None:
         debugger_base_url=debugger_graph_url,
         watch=True,
         image="my-cool-image",
+        topology="single",
     )
 
     expected_args = [

libs/cli/tests/unit_tests/test_config.py

@@ -1147,6 +1147,7 @@ def test_config_to_compose_simple_config():
         PATH_TO_CONFIG,
         validate_config({"dependencies": ["."], "graphs": graphs}),
         "langchain/langgraph-api",
+        topology="single",
     )
     assert (
         clean_empty_lines(actual_compose_stdin).strip()
@@ -1195,6 +1196,7 @@ def test_config_to_compose_env_vars():
             }
         ),
         "langchain/langgraph-api-custom",
+        topology="single",
     )
     assert clean_empty_lines(actual_compose_stdin) == expected_compose_stdin
 
@@ -1233,6 +1235,7 @@ def test_config_to_compose_env_file():
         PATH_TO_CONFIG,
         validate_config({"dependencies": ["."], "graphs": graphs, "env": ".env"}),
         "langchain/langgraph-api",
+        topology="single",
     )
     assert clean_empty_lines(actual_compose_stdin) == expected_compose_stdin
 
@@ -1279,6 +1282,7 @@ def test_config_to_compose_watch():
         validate_config({"dependencies": ["."], "graphs": graphs}),
         "langchain/langgraph-api",
         watch=True,
+        topology="single",
     )
     assert clean_empty_lines(actual_compose_stdin) == expected_compose_stdin
 
@@ -1326,6 +1330,7 @@ def test_config_to_compose_end_to_end():
         validate_config({"dependencies": ["."], "graphs": graphs, "env": ".env"}),
         "langchain/langgraph-api",
         watch=True,
+        topology="single",
     )
     assert clean_empty_lines(actual_compose_stdin) == expected_compose_stdin
 

libs/cli/tests/unit_tests/test_docker.py

@@ -20,7 +20,10 @@ def test_compose_with_no_debugger_and_custom_db():
     port = 8123
     custom_postgres_uri = "custom_postgres_uri"
     actual_compose_str = compose(
-        DEFAULT_DOCKER_CAPABILITIES, port=port, postgres_uri=custom_postgres_uri
+        DEFAULT_DOCKER_CAPABILITIES,
+        port=port,
+        postgres_uri=custom_postgres_uri,
+        topology="single",
     )
     expected_compose_str = f"""services:
     langgraph-redis:
@@ -49,6 +52,7 @@ def test_compose_with_no_debugger_and_custom_db_with_healthcheck():
         DEFAULT_DOCKER_CAPABILITIES._replace(healthcheck_start_interval=True),
         port=port,
         postgres_uri=custom_postgres_uri,
+        topology="single",
     )
     expected_compose_str = f"""services:
     langgraph-redis:
@@ -82,6 +86,7 @@ def test_compose_with_debugger_and_custom_db():
         DEFAULT_DOCKER_CAPABILITIES,
         port=port,
         postgres_uri=custom_postgres_uri,
+        topology="single",
     )
     expected_compose_str = f"""services:
     langgraph-redis:
@@ -105,7 +110,9 @@ def test_compose_with_debugger_and_custom_db():
 
 def test_compose_with_debugger_and_default_db():
     port = 8123
-    actual_compose_str = compose(DEFAULT_DOCKER_CAPABILITIES, port=port)
+    actual_compose_str = compose(
+        DEFAULT_DOCKER_CAPABILITIES, port=port, topology="single"
+    )
     expected_compose_str = f"""volumes:
     langgraph-data:
         driver: local
@@ -157,7 +164,10 @@ def test_compose_with_api_version():
     api_version = "0.2.74"
 
     actual_compose_str = compose(
-        DEFAULT_DOCKER_CAPABILITIES, port=port, api_version=api_version
+        DEFAULT_DOCKER_CAPABILITIES,
+        port=port,
+        api_version=api_version,
+        topology="single",
     )
 
     # The compose function should generate a compose file that doesn't directly
@@ -219,6 +229,7 @@ def test_compose_with_api_version_and_base_image():
         port=port,
         api_version=api_version,
         base_image=base_image,
+        topology="single",
     )
 
     # Similar to the previous test - the compose function doesn't directly embed
@@ -280,6 +291,7 @@ def test_compose_with_api_version_and_custom_postgres():
         port=port,
         api_version=api_version,
         postgres_uri=custom_postgres_uri,
+        topology="single",
     )
 
     expected_compose_str = f"""services:
@@ -313,6 +325,7 @@ def test_compose_with_api_version_and_debugger():
         port=port,
         api_version=api_version,
         debugger_port=debugger_port,
+        topology="single",
     )
 
     expected_compose_str = f"""volumes:
@@ -368,6 +381,101 @@ services:
     assert clean_empty_lines(actual_compose_str) == expected_compose_str
 
 
+def test_compose_with_single_topology():
+    """Test compose function with single topology (legacy behavior)."""
+    port = 8123
+    custom_postgres_uri = "custom_postgres_uri"
+    actual_compose_str = compose(
+        DEFAULT_DOCKER_CAPABILITIES,
+        port=port,
+        postgres_uri=custom_postgres_uri,
+        topology="single",
+    )
+    # Assert langgraph-worker service is NOT present
+    assert "langgraph-worker" not in actual_compose_str
+    assert "/storage/queue_entrypoint.sh" not in actual_compose_str
+    # Assert N_JOBS_PER_WORKER is NOT set (queue runs in API container)
+    assert "N_JOBS_PER_WORKER" not in actual_compose_str
+
+
+def test_compose_with_split_topology():
+    """Test compose function with split topology when using pre-built image."""
+    port = 8123
+    custom_postgres_uri = "custom_postgres_uri"
+    image = "my-custom-image:latest"
+    actual_compose_str = compose(
+        DEFAULT_DOCKER_CAPABILITIES,
+        port=port,
+        postgres_uri=custom_postgres_uri,
+        topology="split",
+        image=image,
+    )
+    # Assert langgraph-worker service is present when using image
+    assert "langgraph-worker" in actual_compose_str
+    assert "/storage/queue_entrypoint.sh" in actual_compose_str
+    # Assert N_JOBS_PER_WORKER is set to 0 on API service
+    assert "N_JOBS_PER_WORKER" in actual_compose_str
+
+
+def test_compose_split_topology_is_default():
+    """Test that split topology is the default (N_JOBS_PER_WORKER set on api)."""
+    port = 8123
+    custom_postgres_uri = "custom_postgres_uri"
+    # Call without specifying topology
+    actual_compose_str = compose(
+        DEFAULT_DOCKER_CAPABILITIES,
+        port=port,
+        postgres_uri=custom_postgres_uri,
+    )
+    # In split mode without image, worker is NOT in compose (added in config_to_compose)
+    # But N_JOBS_PER_WORKER should be set on api service
+    assert "N_JOBS_PER_WORKER" in actual_compose_str
+
+
+def test_compose_split_topology_with_image():
+    """Test that worker service uses the same image as api in split mode."""
+    port = 8123
+    image = "my-custom-image:latest"
+    actual_compose_str = compose(
+        DEFAULT_DOCKER_CAPABILITIES,
+        port=port,
+        image=image,
+        topology="split",
+    )
+    # Both services should reference the same image
+    assert actual_compose_str.count(f"image: {image}") == 2
+
+
+def test_compose_split_topology_with_postgres():
+    """Test split topology includes postgres service (worker added when image provided)."""
+    port = 8123
+    image = "my-custom-image:latest"
+    actual_compose_str = compose(
+        DEFAULT_DOCKER_CAPABILITIES,
+        port=port,
+        topology="split",
+        image=image,
+    )
+    # Worker should be present when image is provided
+    assert "langgraph-worker" in actual_compose_str
+    # The compose should include langgraph-postgres service
+    assert "langgraph-postgres" in actual_compose_str
+
+
+def test_compose_split_topology_with_healthcheck():
+    """Test split topology with healthcheck capabilities when using image."""
+    port = 8123
+    image = "my-custom-image:latest"
+    actual_compose_str = compose(
+        DEFAULT_DOCKER_CAPABILITIES._replace(healthcheck_start_interval=True),
+        port=port,
+        topology="split",
+        image=image,
+    )
+    # Both api and worker should have healthchecks
+    assert actual_compose_str.count("python /api/healthcheck.py") == 2
+
+
 @pytest.mark.parametrize(
     "input_str,expected",
     [