fix(cli): defer parent agent env validation

libs/cli/langgraph_cli/__init__.py

@@ -1 +1 @@
-__version__ = "0.4.32.dev0"
+__version__ = "0.4.32"

libs/cli/langgraph_cli/deploy.py

@@ -12,6 +12,7 @@ from collections.abc import Callable, Mapping, Sequence
 from contextlib import contextmanager
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
+from functools import partial
 from typing import Protocol, TypeVar
 
 import click
@@ -805,46 +806,6 @@ def _find_deployment(
     selector: ByName | ByAgent,
     *,
     not_found_message: str,
-<<<<<<< HEAD
-    agent: dict[str, str] | None = None,
-) -> tuple[str | None, bool, int]:
-    """Resolve an existing deployment by ID or exact name match."""
-    needs_creation = False
-    if deployment_id:
-        _log_deploy_step(step, f"Using deployment {deployment_id}")
-        _call_host_backend_with_optional_tenant(
-            client, lambda c: c.get_deployment(deployment_id)
-        )
-        return deployment_id, needs_creation, step + 1
-
-    if agent is not None:
-        _log_deploy_step(
-            step, f"Looking up agent '{agent['agent_id']}' in {agent['environment']}"
-        )
-        existing = _call_host_backend_with_optional_tenant(
-            client,
-            lambda c: c.list_deployments(
-                agent_id=agent["agent_id"], agent_environment=agent["environment"]
-            ),
-        )
-        found_id = next(
-            (
-                dep["id"]
-                for dep in existing.get("resources", [])
-                if not dep.get("is_preview")
-            ),
-            None,
-        )
-    else:
-        _log_deploy_step(step, f"Looking up deployment '{name}'")
-        found_id = _call_host_backend_with_optional_tenant(
-            client, lambda c: find_deployment_id_by_name(c, name)
-        )
-    em = _get_emitter()
-    if found_id:
-        deployment_id = str(found_id)
-        em.info(f"Found existing deployment (ID: {deployment_id})")
-=======
 ) -> tuple[ExistingDeployment | None, int]:
     if isinstance(selector, ByAgent):
         _log_deploy_step(
@@ -872,7 +833,6 @@ def _find_deployment(
             ),
             None,
         )
->>>>>>> origin
     else:
         _log_deploy_step(step, f"Looking up deployment '{selector.name}'")
         found = _call_host_backend_with_optional_tenant(
@@ -897,22 +857,12 @@ def _create_deployment(
     step: int,
     *,
     name: str | None,
-<<<<<<< HEAD
-    deployment_type: str,
-    source: str,
-    config_rel: str | None = None,
-    secrets: list[dict[str, str]] | None = None,
-    agent: dict[str, str] | None = None,
-) -> tuple[str, int]:
-    """Create a deployment and return its ID and next step number."""
-=======
     source: str,
     source_config: dict[str, object],
     source_revision_config: dict[str, object],
     secrets: list[dict[str, str]],
     agent: dict[str, str] | None = None,
 ) -> tuple[CreatedDeployment, int]:
->>>>>>> origin
     _log_deploy_step(
         step,
         f"Creating deployment for agent '{agent['agent_id']}' in {agent['environment']}"
@@ -922,15 +872,9 @@ def _create_deployment(
     try:
         created = client.create_deployment(
             name=name,
-<<<<<<< HEAD
-            deployment_type=deployment_type,
-            source=source,
-            config_path=config_rel,
-=======
             source=source,
             source_config=source_config,
             source_revision_config=source_revision_config,
->>>>>>> origin
             secrets=secrets,
             agent=agent,
         )
@@ -947,36 +891,9 @@ def _create_deployment(
             "POST /v2/deployments succeeded but response missing a valid 'id'"
         )
     if agent is not None:
-<<<<<<< HEAD
-        _get_emitter().info(f"Deployment name: {created['name']}")
-    _get_emitter().info(f"Deployment ID: {created_id}", deployment_id=created_id)
-    return created_id, step + 1
-
-
-def _smith_dashboard_base_url(host_url: str | None) -> str:
-    """Derive the LangSmith dashboard base URL from the API host URL."""
-    from urllib.parse import urlparse
-
-    if not host_url:
-        return "https://smith.langchain.com"
-    parsed = urlparse(host_url)
-    hostname = parsed.hostname or ""
-    if hostname in ("localhost", "127.0.0.1"):
-        return host_url.rstrip("/")
-
-    for api_host_suffix in ("api.host.langchain.com", "api.smith.langchain.com"):
-        if hostname == api_host_suffix:
-            return "https://smith.langchain.com"
-        if hostname.endswith(f".{api_host_suffix}"):
-            prefix = hostname[: -(len(api_host_suffix) + 1)]
-            return f"https://{prefix}.smith.langchain.com"
-
-    return "https://smith.langchain.com"
-=======
         _get_emitter().info(f"Deployment name: {created.get('name')}")
     _get_emitter().info(f"Deployment ID: {created_id}", deployment_id=created_id)
     return CreatedDeployment(created_id, created), step + 1
->>>>>>> origin
 
 
 def _get_deployment_status_url(
@@ -2001,7 +1918,8 @@ OPT_AGENT_ID = click.option(
     help="Logical agent ID (requires agent mode enabled for the tenant).",
 )
 
-OPT_AGENT_ENVIRONMENT = click.option(
+OPT_AGENT_ENVIRONMENT = partial(
+    click.option,
     "--agent-environment",
     "environment",
     envvar="LANGSMITH_AGENT_ENVIRONMENT",
@@ -2109,7 +2027,9 @@ def _deploy_base_options(
             OPT_HOST_API_KEY,
             OPT_HOST_DEPLOYMENT_NAME,
             OPT_AGENT_ID,
-            OPT_AGENT_ENVIRONMENT,
+            OPT_AGENT_ENVIRONMENT()
+            if include_docker_args
+            else OPT_AGENT_ENVIRONMENT(type=str),
             click.option(
                 "--deployment-id",
                 help=(
@@ -2256,6 +2176,12 @@ def deploy(ctx: click.Context, **_: object):
     # otherwise, we return None here and click will proceed to actually run the subcommand (list or delete)
     if ctx.invoked_subcommand is not None:
         return
+    environment_param = next(
+        param for param in _deploy_cmd.params if param.name == "environment"
+    )
+    ctx.params["environment"] = environment_param.type_cast_value(
+        ctx, ctx.params["environment"]
+    )
     if (
         ctx.params.get("agent_id") is not None
         or ctx.params.get("environment") is not None
@@ -2310,15 +2236,6 @@ def _deploy_cmd(
     validate_deploy_commands(install_command, build_command)
     agent = None
     if agent_id is not None or environment is not None:
-<<<<<<< HEAD
-        if not agent_id or not agent_id.strip() or not environment:
-            raise click.UsageError(
-                "--agent-id and --environment are required together."
-            )
-        if name is not None or deployment_id is not None:
-            raise click.UsageError(
-                "--agent-id and --environment cannot be combined with --name or --deployment-id."
-=======
         em.note("Note: --agent-id and --agent-environment flags are in private beta")
         if not agent_id or not agent_id.strip() or not environment:
             raise click.UsageError(
@@ -2327,7 +2244,6 @@ def _deploy_cmd(
         if name is not None or deployment_id is not None:
             raise click.UsageError(
                 "--agent-id and --agent-environment cannot be combined with --name or --deployment-id."
->>>>>>> origin
             )
         agent = {"agent_id": agent_id, "environment": environment}
     if not config.exists():
@@ -2375,41 +2291,6 @@ def _deploy_cmd(
     )
 
     client = _create_host_backend_client(host_url, api_key, env_vars=env_vars)
-<<<<<<< HEAD
-    step = 1
-
-    deployment_id, needs_creation, step = _resolve_deployment(
-        client,
-        step,
-        deployment_id,
-        name,
-        not_found_message=(
-            "No deployment found. Will create."
-            if use_remote_build
-            else "No deployment found. Will create after build."
-        ),
-        agent=agent,
-    )
-
-    if needs_creation:
-        deployment_id, step = _create_deployment(
-            client,
-            step,
-            name=name,
-            deployment_type=deployment_type,
-            source="internal_source" if use_remote_build else "internal_docker",
-            secrets=secrets,
-            agent=agent,
-        )
-
-    if not deployment_id:
-        raise click.ClickException("Failed to determine deployment ID")
-
-    # Scan local sources for tracked packages so the new revision carries
-    # the same metadata GitHub-backed deploys produce. Failures must never
-    # block a deploy.
-=======
->>>>>>> origin
     try:
         tracked_packages = find_tracked_packages(config, config_json) or None
     except Exception as exc:
@@ -2500,7 +2381,7 @@ def _deploy_cmd(
 @OPT_HOST_API_KEY
 @OPT_HOST_URL
 @OPT_AGENT_ID
-@OPT_AGENT_ENVIRONMENT
+@OPT_AGENT_ENVIRONMENT()
 @click.option(
     "--name-contains",
     default="",
@@ -2514,14 +2395,11 @@ def deploy_list(
     agent_id: str | None,
     environment: str | None,
 ) -> None:
-<<<<<<< HEAD
-=======
     if agent_id is not None or environment is not None:
         click.secho(
             "Note: --agent-id and --agent-environment flags are in private beta",
             fg="yellow",
         )
->>>>>>> origin
     if agent_id is not None and not agent_id.strip():
         raise click.UsageError("--agent-id must not be empty.")
     filters = {}
@@ -2533,15 +2411,6 @@ def deploy_list(
     deployments = _call_host_backend_with_optional_tenant(
         client,
         lambda c: c.list_deployments(name_contains=name_contains, **filters),
-<<<<<<< HEAD
-    )
-    resources = response.get("resources") if isinstance(response, dict) else None
-    deployments = (
-        [item for item in resources if isinstance(item, dict)]
-        if isinstance(resources, list)
-        else []
-=======
->>>>>>> origin
     )
     if not deployments:
         click.echo("No deployments found.")