diff --git a/libs/cli/langgraph_cli/deploy.py b/libs/cli/langgraph_cli/deploy.py index dfdb9edf9..d052af0a5 100644 --- a/libs/cli/langgraph_cli/deploy.py +++ b/libs/cli/langgraph_cli/deploy.py @@ -639,23 +639,15 @@ def _create_deployment( source: str, config_rel: str | None = None, secrets: list[dict[str, str]] | None = None, - image_uri: str | None = None, - tracked_packages: list[str] | None = None, ) -> tuple[str, int]: """Create a deployment and return its ID and next step number.""" _log_deploy_step(step, f"Creating deployment '{name}'") - extra = ( - {"image_uri": image_uri, "tracked_packages": tracked_packages} - if image_uri - else {} - ) created = client.create_deployment( name=name, deployment_type=deployment_type, source=source, config_path=config_rel, secrets=secrets, - **extra, ) created_id = created.get("id") if isinstance(created, dict) else None if not isinstance(created_id, str) or not created_id: @@ -674,6 +666,7 @@ def _smith_dashboard_base_url(host_url: str | None) -> str: return "https://smith.langchain.com" parsed = urlparse(host_url) hostname = parsed.hostname or "" + # Self-hosted: host_url is :///api-host — return just the root path = parsed.path.rstrip("/") if path == "/api-host" or path.endswith("/api-host"): return f"{parsed.scheme}://{parsed.netloc}" @@ -953,29 +946,43 @@ def _build_image_tagged( verbose: bool, ) -> None: """Build a Docker image to *tag*, using buildx on non-x86_64 hosts to target linux/amd64.""" - needs_buildx = platform.machine() != "x86_64" - build_flags = ["--platform", "linux/amd64", "--load"] if needs_buildx else [] - if needs_buildx and not verbose: - build_flags.append("--progress=quiet") - with Progress(message="Building...", elapsed=not verbose): - build_docker_image( - runner, - lambda _msg: None, - config, - config_json, - base_image, - api_version, - pull, - tag, - docker_build_args, - install_command, - build_command, - docker_command=("docker", "buildx", "build") - if needs_buildx - else ("docker", "build"), - extra_flags=build_flags, - verbose=verbose, - ) + if platform.machine() != "x86_64": + build_flags: list[str] = ["--platform", "linux/amd64", "--load"] + if not verbose: + build_flags.append("--progress=quiet") + with Progress(message="Building...", elapsed=not verbose): + build_docker_image( + runner, + lambda _msg: None, + config, + config_json, + base_image, + api_version, + pull, + tag, + docker_build_args, + install_command, + build_command, + docker_command=("docker", "buildx", "build"), + extra_flags=build_flags, + verbose=verbose, + ) + else: + with Progress(message="Building...", elapsed=not verbose): + build_docker_image( + runner, + lambda _msg: None, + config, + config_json, + base_image, + api_version, + pull, + tag, + docker_build_args, + install_command, + build_command, + verbose=verbose, + ) def _run_local_build( @@ -1145,13 +1152,14 @@ def _run_local_build( ) -def _push_external_image( +def _run_external_deploy( *, + client: HostBackendClient, + deployment_id: str, step: int, config: pathlib.Path, config_json: dict, - push_to: str, - prebuilt_image: str | None, + image_uri: str, verbose: bool, pull: bool, api_version: str | None, @@ -1159,40 +1167,55 @@ def _push_external_image( install_command: str | None, build_command: str | None, docker_build_args: Sequence[str], -) -> str: - """Build or retag an image and push using existing Docker credentials.""" + secrets: list[dict[str, str]], + tracked_packages: list[str] | None, +) -> "BuildResult": + """Build image, push using existing Docker credentials, and update the deployment.""" with Runner() as runner: - if prebuilt_image: - _log_deploy_step(step, f"Validating image {prebuilt_image}") - _validate_prebuilt_image(runner, prebuilt_image, verbose=verbose) - runner.run( - subp_exec("docker", "tag", prebuilt_image, push_to, verbose=verbose) - ) - else: - _log_deploy_step(step, f"Building image {push_to}") - _build_image_tagged( - runner, - config, - config_json, - base_image, - api_version, - pull, - push_to, - docker_build_args, - install_command, - build_command, - verbose=verbose, - ) - _log_deploy_step(step + 1, f"Pushing image {push_to}") - with Progress(message="Pushing...", elapsed=not verbose): - runner.run(subp_exec("docker", "push", push_to, verbose=verbose)) - return _resolve_pushed_image_digest( + _log_deploy_step(step, f"Building image {image_uri}") + _build_image_tagged( runner, - remote_image=push_to, + config, + config_json, + base_image, + api_version, + pull, + image_uri, + docker_build_args, + install_command, + build_command, + verbose=verbose, + ) + step += 1 + + _log_deploy_step(step, f"Pushing image {image_uri}") + with Progress(message="Pushing...", elapsed=not verbose): + runner.run(subp_exec("docker", "push", image_uri, verbose=verbose)) + step += 1 + + resolved_image = _resolve_pushed_image_digest( + runner, + remote_image=image_uri, docker_config_dir=None, verbose=verbose, ) + _log_deploy_step(step, f"Updating deployment {deployment_id}") + updated = client.update_deployment_external( + deployment_id, + resolved_image, + secrets=secrets, + tracked_packages=tracked_packages, + ) + + return BuildResult( + updated=updated if isinstance(updated, dict) else {}, + progress_message="Deploying...", + timeout_seconds=300, + poll_interval_seconds=1, + no_result_message="Deployment updated", + ) + def _run_remote_build( *, @@ -1329,20 +1352,30 @@ def _create_host_backend_client( tenant_id = env_vars.get("LANGSMITH_TENANT_ID") or os.environ.get( "LANGSMITH_TENANT_ID" ) - if not host_url: - from urllib.parse import urlparse - - endpoint = env_vars.get("LANGSMITH_ENDPOINT") or os.environ.get( + # If no explicit host URL was provided, check LANGSMITH_ENDPOINT as a + # fallback so self-hosted customers don't need to know about LANGGRAPH_HOST_URL. + # Self-hosted control plane always lives at /api-host. + _cloud_default = "https://api.host.langchain.com" + _cloud_endpoints = { + "https://api.smith.langchain.com", + "https://api.langchain.com", + } + resolved_host = host_url + if not resolved_host or resolved_host == _cloud_default: + langsmith_endpoint = env_vars.get("LANGSMITH_ENDPOINT") or os.environ.get( "LANGSMITH_ENDPOINT" ) - parsed = urlparse(endpoint or "https://api.smith.langchain.com") - if parsed.hostname in ("api.smith.langchain.com", "api.langchain.com"): - host_url = "https://api.host.langchain.com" - elif parsed.hostname == "eu.api.smith.langchain.com": - host_url = "https://eu.api.host.langchain.com" + if ( + langsmith_endpoint + and langsmith_endpoint.rstrip("/") not in _cloud_endpoints + ): + from urllib.parse import urlparse as _urlparse + + _p = _urlparse(langsmith_endpoint) + resolved_host = f"{_p.scheme}://{_p.netloc}/api-host" else: - host_url = f"{parsed.scheme}://{parsed.netloc}/api-host" - return HostBackendClient(host_url, resolved_api_key, tenant_id=tenant_id) + resolved_host = _cloud_default + return HostBackendClient(resolved_host, resolved_api_key, tenant_id=tenant_id) def _call_host_backend_with_optional_tenant( @@ -1421,7 +1454,7 @@ OPT_HOST_DEPLOYMENT_NAME = click.option( OPT_HOST_URL = click.option( "--host-url", envvar="LANGGRAPH_HOST_URL", - default=None, + default="https://api.host.langchain.com", hidden=True, ) @@ -1561,11 +1594,13 @@ def _deploy_base_options( ), ), click.option( - "--push-to", + "--image-uri", help=( - "Push to this customer-managed registry URI for self-hosted/hybrid " - "deployments using existing Docker credentials. Builds the project " - "or retags the local image supplied with --image." + "Image URI to build, push, and deploy " + "(e.g. 123456789.dkr.ecr.us-east-1.amazonaws.com/repo:tag). " + "Builds the project, pushes using existing Docker credentials, " + "and triggers a deployment revision. " + "Required for self-hosted deployments." ), ), click.option( @@ -1669,7 +1704,7 @@ def _deploy_cmd( name: str | None, image_name: str | None, image: str | None, - push_to: str | None, + image_uri: str | None, tag: str, base_image: str | None, install_command: str | None, @@ -1716,10 +1751,12 @@ def _deploy_cmd( secrets = _secrets_from_env(_env_without_deployment_name(env_vars)) - if push_to and remote_build_flag is True: - raise click.UsageError("--push-to cannot be combined with --remote.") + if image_uri and remote_build_flag is True: + raise click.UsageError("--image-uri cannot be combined with --remote.") + if image_uri and image: + raise click.UsageError("--image-uri cannot be combined with --image.") - use_external_docker = push_to is not None + use_external_docker = image_uri is not None if use_external_docker: use_remote_build = False @@ -1745,35 +1782,45 @@ def _deploy_cmd( name, not_found_message=( "No deployment found. Will create." - if use_remote_build + if (use_remote_build or use_external_docker) else "No deployment found. Will create after build." ), ) - if needs_creation and not use_external_docker: + if needs_creation: + if use_external_docker: + source = "external_docker" + elif use_remote_build: + source = "internal_source" + else: + source = "internal_docker" deployment_id, step = _create_deployment( client, step, name=name, deployment_type=deployment_type, - source="internal_source" if use_remote_build else "internal_docker", + source=source, secrets=secrets, ) - if not deployment_id and not use_external_docker: + if not deployment_id: raise click.ClickException("Failed to determine deployment ID") + # Validate that an existing deployment is compatible with --image-uri. + # update_deployment_external sends an external_docker revision; applying + # it to a deployment created with a different source mode may be rejected + # by the backend or silently produce an inconsistent revision. if use_external_docker and not needs_creation: existing = _call_host_backend_with_optional_tenant( client, lambda c: c.get_deployment(deployment_id) ) existing_source = existing.get("source") if isinstance(existing, dict) else None - if existing_source != "external_docker": + if existing_source and existing_source != "external_docker": raise click.UsageError( f"Deployment {deployment_id} uses a different build mode and " - f"cannot be updated with --push-to. To use --push-to, omit " + f"cannot be updated with --image-uri. To use --image-uri, omit " f"--deployment-id to create a new deployment, or remove " - f"--push-to to continue using the current build mode." + f"--image-uri to continue using the current build mode." ) # Scan local sources for tracked packages so the new revision carries @@ -1787,12 +1834,13 @@ def _deploy_cmd( # -- 3. Build (divergent path) -- if use_external_docker: - resolved_image = _push_external_image( + build_result = _run_external_deploy( + client=client, + deployment_id=deployment_id, step=step, config=config, config_json=config_json, - push_to=push_to, - prebuilt_image=image, + image_uri=image_uri, verbose=verbose, pull=pull, api_version=api_version, @@ -1800,32 +1848,8 @@ def _deploy_cmd( install_command=install_command, build_command=build_command, docker_build_args=docker_build_args, - ) - if needs_creation: - deployment_id, step = _create_deployment( - client, - step + 2, - name=name, - deployment_type=deployment_type, - source="external_docker", - secrets=secrets, - image_uri=resolved_image, - tracked_packages=tracked_packages, - ) - updated = client.get_deployment(deployment_id) - else: - updated = client.update_deployment_external( - deployment_id, - resolved_image, - secrets=secrets, - tracked_packages=tracked_packages, - ) - build_result = BuildResult( - updated=updated, - progress_message="Deploying...", - timeout_seconds=300, - poll_interval_seconds=1, - no_result_message="Deployment updated", + secrets=secrets, + tracked_packages=tracked_packages, ) elif use_remote_build: build_result = _run_remote_build( diff --git a/libs/cli/langgraph_cli/host_backend.py b/libs/cli/langgraph_cli/host_backend.py index 09c6f8360..77e4e89aa 100644 --- a/libs/cli/langgraph_cli/host_backend.py +++ b/libs/cli/langgraph_cli/host_backend.py @@ -81,8 +81,6 @@ class HostBackendClient: source: str, config_path: str | None = None, secrets: list[dict[str, str]] | None = None, - image_uri: str | None = None, - tracked_packages: list[str] | None = None, ) -> dict[str, Any]: """Create a deployment.""" payload: dict[str, Any] = { @@ -91,11 +89,6 @@ class HostBackendClient: "source_config": {"deployment_type": deployment_type}, "source_revision_config": {}, } - if source == "external_docker": - payload["source_config"] = {} - payload["source_revision_config"] = {"image_uri": image_uri} - if tracked_packages: - payload["tracked_packages"] = tracked_packages if source == "internal_source" and config_path: payload["source_revision_config"]["langgraph_config_path"] = config_path if secrets is not None: diff --git a/libs/cli/tests/unit_tests/cli/test_cli.py b/libs/cli/tests/unit_tests/cli/test_cli.py index 40ae36df2..1d9a69c55 100644 --- a/libs/cli/tests/unit_tests/cli/test_cli.py +++ b/libs/cli/tests/unit_tests/cli/test_cli.py @@ -6,10 +6,8 @@ import tempfile import textwrap from contextlib import contextmanager from pathlib import Path -from unittest.mock import MagicMock import click -import pytest from click.testing import CliRunner import langgraph_cli.deploy as deploy_module @@ -1358,89 +1356,47 @@ def test_prepare_args_and_stdin_distributed_mode() -> None: assert "executor_entrypoint.sh" in actual_stdin -@pytest.mark.parametrize( - "source,image", - [ - (None, None), - (None, "local:latest"), - ("external_docker", None), - ("external_docker", "local:latest"), - ("internal_docker", None), - ], -) -def test_deploy_push_to(monkeypatch, tmp_path, source, image): +def test_deploy_image_uri_rejects_incompatible_source(monkeypatch, tmp_path) -> None: + """--image-uri raises UsageError when applied to a non-external_docker deployment.""" + # --no-input sets the module-level _no_input global; ensure it's restored. monkeypatch.setattr(deploy_module, "_no_input", False) - monkeypatch.setattr(deploy_module, "_emitter", None) + config = tmp_path / "langgraph.json" config.write_text('{"graphs": {"agent": "agent.py:graph"}, "dependencies": ["."]}') - events = [] - client = MagicMock(base_url="https://smith.example.com/api-host") - client.get_deployment.return_value = {"id": "dep-123", "source": source} - client.list_deployments.return_value = {"deployments": []} - client.create_deployment.side_effect = lambda **kw: ( - events.append(("create", kw)) or {"id": "dep-123"} + + class FakeClient: + def __init__(self, host_url: str, api_key: str, tenant_id: str | None = None): + self.base_url = host_url + + def get_deployment(self, deployment_id: str): + return { + "id": deployment_id, + "name": "test-deploy", + "source": "internal_docker", + "tenant_id": "tenant-1", + } + + monkeypatch.setattr(deploy_module, "HostBackendClient", FakeClient) + + runner = CliRunner() + result = runner.invoke( + cli, + [ + "deploy", + "--api-key", + "test-key", + "--host-url", + "https://api.example.com", + "--deployment-id", + "dep-123", + "--image-uri", + "registry.example.com/app:latest", + "--config", + str(config), + "--no-input", + ], ) - client.update_deployment_external.side_effect = lambda *a, **kw: ( - events.append(("update", a)) or {} - ) - monkeypatch.setattr(deploy_module, "HostBackendClient", lambda *a, **kw: client) - monkeypatch.setattr( - deploy_module, - "_build_image_tagged", - lambda *a, **kw: events.append(("build", a[6])), - ) - monkeypatch.setattr( - deploy_module, - "_validate_prebuilt_image", - lambda *a, **kw: events.append(("validate", a[1])), - ) - monkeypatch.setattr(deploy_module, "subp_exec", lambda *a, **kw: a) - runner = MagicMock() - runner.__enter__.return_value = runner - runner.run.side_effect = lambda args: events.append(args) - monkeypatch.setattr(deploy_module, "Runner", lambda: runner) - digest = "registry.example.com/app@sha256:abc123" - monkeypatch.setattr( - deploy_module, - "_resolve_pushed_image_digest", - lambda *a, **kw: events.append(("digest",)) or digest, - ) - args = [ - "deploy", - "--api-key", - "key", - "--push-to", - "registry.example.com/app:latest", - "--config", - str(config), - "--no-input", - "--no-wait", - ] - args += ["--deployment-id", "dep-123"] if source else ["--name", "app"] - if image: - args += ["--image", image] - result = CliRunner().invoke(cli, args) - if source == "internal_docker": - assert result.exit_code != 0 - assert "cannot be updated with --push-to" in result.output - assert events == [] - return - assert result.exit_code == 0, result.output - destination = "registry.example.com/app:latest" - expected = ( - [("validate", image), ("docker", "tag", image, destination)] - if image - else [("build", destination)] - ) - assert events[: len(expected) + 2] == expected + [ - ("docker", "push", destination), - ("digest",), - ] - if source: - assert events[-1] == ("update", ("dep-123", digest)) - client.create_deployment.assert_not_called() - else: - assert events[-1][0] == "create" - assert events[-1][1]["image_uri"] == digest - assert events[-1][1]["source"] == "external_docker" - client.update_deployment_external.assert_not_called() + + assert result.exit_code != 0 + assert "different build mode" in result.output + assert "cannot be updated with --image-uri" in result.output diff --git a/libs/cli/tests/unit_tests/test_deploy_helpers.py b/libs/cli/tests/unit_tests/test_deploy_helpers.py index 37b3ff7de..d93eb0d25 100644 --- a/libs/cli/tests/unit_tests/test_deploy_helpers.py +++ b/libs/cli/tests/unit_tests/test_deploy_helpers.py @@ -543,47 +543,53 @@ class TestCreateHostBackendClientNoInput: assert client is not None -@pytest.mark.parametrize( - "endpoint,host,expected", - [ - (None, None, "https://api.host.langchain.com"), - ("https://api.smith.langchain.com/", None, "https://api.host.langchain.com"), - ("https://api.langchain.com", None, "https://api.host.langchain.com"), - ( - "https://eu.api.smith.langchain.com", - None, - "https://eu.api.host.langchain.com", - ), - ( - "https://smith.example.com/api/v1", - None, - "https://smith.example.com/api-host", - ), - ( - "https://smith.example.com", - "https://api.host.langchain.com", - "https://api.host.langchain.com", - ), - ( - "https://smith.example.com", - "https://custom.host.com", - "https://custom.host.com", - ), - ], -) -@pytest.mark.parametrize("from_env", [True, False]) -def test_endpoint_fallback(monkeypatch, endpoint, host, expected, from_env): - monkeypatch.delenv("LANGSMITH_ENDPOINT", raising=False) - env_vars = {} - if endpoint: - if from_env: - monkeypatch.setenv("LANGSMITH_ENDPOINT", endpoint) - else: - env_vars["LANGSMITH_ENDPOINT"] = endpoint - client = _create_host_backend_client( - host_url=host, api_key="key", env_vars=env_vars - ) - assert client.base_url == expected +class TestCreateHostBackendClientEndpointFallback: + def test_langsmith_endpoint_env_var_used_as_fallback(self, monkeypatch): + monkeypatch.setenv("LANGSMITH_API_KEY", "lsv2_test") + monkeypatch.setenv("LANGSMITH_ENDPOINT", "https://smith.example.com/api/v1") + monkeypatch.delenv("LANGGRAPH_HOST_URL", raising=False) + client = _create_host_backend_client(host_url=None, api_key=None, env_vars={}) + assert client.base_url == "https://smith.example.com/api-host" + + def test_langsmith_endpoint_from_env_vars_dict(self, monkeypatch): + monkeypatch.setenv("LANGSMITH_API_KEY", "lsv2_test") + monkeypatch.delenv("LANGSMITH_ENDPOINT", raising=False) + client = _create_host_backend_client( + host_url=None, + api_key=None, + env_vars={"LANGSMITH_ENDPOINT": "https://smith.example.com/api/v1"}, + ) + assert client.base_url == "https://smith.example.com/api-host" + + def test_cloud_langsmith_endpoint_not_used_as_self_hosted(self, monkeypatch): + monkeypatch.setenv("LANGSMITH_API_KEY", "lsv2_test") + monkeypatch.setenv("LANGSMITH_ENDPOINT", "https://api.smith.langchain.com") + client = _create_host_backend_client(host_url=None, api_key=None, env_vars={}) + assert client.base_url == "https://api.host.langchain.com" + + def test_langchain_api_endpoint_not_used_as_self_hosted(self, monkeypatch): + monkeypatch.setenv("LANGSMITH_API_KEY", "lsv2_test") + monkeypatch.setenv("LANGSMITH_ENDPOINT", "https://api.langchain.com") + client = _create_host_backend_client(host_url=None, api_key=None, env_vars={}) + assert client.base_url == "https://api.host.langchain.com" + + def test_explicit_host_url_takes_precedence_over_langsmith_endpoint( + self, monkeypatch + ): + monkeypatch.setenv("LANGSMITH_API_KEY", "lsv2_test") + monkeypatch.setenv("LANGSMITH_ENDPOINT", "https://smith.example.com/api/v1") + client = _create_host_backend_client( + host_url="https://custom.host.com", + api_key=None, + env_vars={}, + ) + assert client.base_url == "https://custom.host.com" + + def test_no_endpoint_falls_back_to_cloud_default(self, monkeypatch): + monkeypatch.setenv("LANGSMITH_API_KEY", "lsv2_test") + monkeypatch.delenv("LANGSMITH_ENDPOINT", raising=False) + client = _create_host_backend_client(host_url=None, api_key=None, env_vars={}) + assert client.base_url == "https://api.host.langchain.com" class TestSmithDashboardBaseUrl: @@ -641,16 +647,23 @@ class TestSmithDashboardBaseUrl: == "https://smith.langchain.com" ) - @pytest.mark.parametrize( - "url,expected", - [ - ("https://smith.example.com/api-host", "https://smith.example.com"), - ("https://smith.example.com/api-host/", "https://smith.example.com"), - ("http://localhost:8080/api-host", "http://localhost:8080"), - ], - ) - def test_self_hosted(self, url, expected): - assert _smith_dashboard_base_url(url) == expected + def test_self_hosted_api_host_suffix(self): + assert ( + _smith_dashboard_base_url("https://smith.example.com/api-host") + == "https://smith.example.com" + ) + + def test_self_hosted_api_host_trailing_slash(self): + assert ( + _smith_dashboard_base_url("https://smith.example.com/api-host/") + == "https://smith.example.com" + ) + + def test_self_hosted_localhost_api_host(self): + assert ( + _smith_dashboard_base_url("http://localhost:8080/api-host") + == "http://localhost:8080" + ) class TestResolvePushedImageDigest: diff --git a/libs/cli/tests/unit_tests/test_host_backend.py b/libs/cli/tests/unit_tests/test_host_backend.py index ad4916431..d39e503ef 100644 --- a/libs/cli/tests/unit_tests/test_host_backend.py +++ b/libs/cli/tests/unit_tests/test_host_backend.py @@ -178,32 +178,45 @@ def test_update_deployment_no_secrets(client): assert result == {"ok": True} -@pytest.mark.parametrize("create", [True, False]) -def test_external_deployment_payload(create): +def test_update_deployment_external(): captured: dict = {} c = _capturing_client(captured) - image = "registry.example.com/app@sha256:abc123" - kwargs = { - "secrets": [{"name": "KEY", "value": "value"}], - "tracked_packages": ["google-adk:1.0.0"], - } - if create: - c.create_deployment("app", "dev", "external_docker", image_uri=image, **kwargs) - else: - c.update_deployment_external("dep-123", image, **kwargs) + result = c.update_deployment_external( + "dep-123", "registry.example.com/app@sha256:abc123" + ) + assert result == {"ok": True} body = json.loads(captured["body"]) - assert body == { - **( - {"name": "app", "source": "external_docker", "source_config": {}} - if create - else {} - ), - "source_revision_config": {"image_uri": image}, - **kwargs, - } + assert "revision_source" not in body + assert body["source_revision_config"]["image_uri"] == ( + "registry.example.com/app@sha256:abc123" + ) + + +def test_update_deployment_external_forwards_tracked_packages(): + captured: dict = {} + c = _capturing_client(captured) + c.update_deployment_external( + "dep-123", + "registry.example.com/app:latest", + tracked_packages=["google-adk:1.0.0"], + ) + body = json.loads(captured["body"]) + assert body["tracked_packages"] == ["google-adk:1.0.0"] + assert "revision_source" not in body + + +def test_update_deployment_external_omits_tracked_packages_when_absent(): + captured: dict = {} + c = _capturing_client(captured) + c.update_deployment_external("dep-123", "registry.example.com/app:latest") + body = json.loads(captured["body"]) + assert "tracked_packages" not in body + assert "revision_source" not in body def test_request_builds_full_url_with_path_prefix(): + """base_url with a path prefix (/api-host) must not be dropped when paths start with /.""" + def handler(req: httpx.Request) -> httpx.Response: assert str(req.url) == "https://smith.example.com/api-host/v2/deployments" return httpx.Response(200, json={"ok": True})