feat(cli): send the upload headers the control plane returns

The CLI tells the control plane it sends `upload_headers` with its source
upload. When the response includes them, the upload sends exactly those
headers plus Content-Length; otherwise it keeps the previous defaults. Some
storage needs this: Azure Blob Storage rejects a PUT without
`x-ms-blob-type: BlockBlob`.

libs/cli/langgraph_cli/deploy.py

@@ -1106,12 +1106,24 @@ class _ProgressReader:
         return self._file_size
 
 
-def _upload_to_gcs(signed_url: str, file_path: str, file_size: int) -> None:
-    """Upload tarball to GCS via signed PUT URL with progress display."""
+def _upload_to_gcs(
+    signed_url: str,
+    file_path: str,
+    file_size: int,
+    upload_headers: dict[str, str] | None,
+) -> None:
+    """Upload tarball via signed PUT URL with progress display."""
     import urllib.error
     import urllib.request
 
     em = _get_emitter()
+    # Control planes that return no `upload_headers` sign for the Cloud Storage
+    # size range. Content-Type is a fallback: without one, urllib sends a form
+    # content type, which breaks signatures that cover it.
+    headers = {
+        "Content-Type": "application/gzip",
+        **(upload_headers or {"X-Goog-Content-Length-Range": "0,209715200"}),
+    }
 
     with open(file_path, "rb") as f:
         reader = _ProgressReader(f, file_size, em)
@@ -1119,11 +1131,7 @@ def _upload_to_gcs(signed_url: str, file_path: str, file_size: int) -> None:
             signed_url,
             data=reader,
             method="PUT",
-            headers={
-                "Content-Type": "application/gzip",
-                "Content-Length": str(file_size),
-                "X-Goog-Content-Length-Range": "0,209715200",
-            },
+            headers={**headers, "Content-Length": str(file_size)},
         )
         try:
             urllib.request.urlopen(req, timeout=_UPLOAD_TIMEOUT_SECONDS)
@@ -1443,7 +1451,9 @@ def _run_remote_build(
         step += 1
 
         _log_deploy_step(step, "Uploading source")
-        _upload_to_gcs(signed_url, archive_path, file_size)
+        _upload_to_gcs(
+            signed_url, archive_path, file_size, upload_data.get("upload_headers")
+        )
     step += 1
 
     _log_deploy_step(step, "Triggering remote build")

libs/cli/langgraph_cli/host_backend.py

@@ -258,10 +258,13 @@ class HostBackendClient:
         )
 
     def request_upload_url(self, deployment_id: str) -> dict[str, Any]:
-        """Get a signed GCS URL for uploading the source tarball."""
+        """Get a signed URL for uploading the source tarball."""
+        # Tells the control plane this client sends the returned
+        # `upload_headers` with its upload, which some storage requires.
         return self._request(
             "POST",
             f"/v2/deployments/{deployment_id}/upload-url",
+            {"supports_upload_headers": True},
         )
 
     def update_deployment(

libs/cli/tests/unit_tests/cli/test_deploy_command.py

@@ -96,6 +96,7 @@ class ControlPlaneDouble:
     existing_deployments: list[dict] = field(default_factory=list)
     push_token_status: int = 200
     push_token_extra: dict = field(default_factory=dict)
+    upload_url_extra: dict = field(default_factory=dict)
     create_error: str | None = None
     listeners: list[dict] = field(default_factory=list)
     listeners_by_id: dict[str, dict] = field(default_factory=dict)
@@ -159,7 +160,12 @@ class ControlPlaneDouble:
             )
         if path.endswith("/upload-url"):
             return httpx.Response(
-                200, json={"upload_url": SIGNED_UPLOAD_URL, "object_path": OBJECT_PATH}
+                200,
+                json={
+                    "upload_url": SIGNED_UPLOAD_URL,
+                    "object_path": OBJECT_PATH,
+                    **self.upload_url_extra,
+                },
             )
         if method == "PATCH":
             return httpx.Response(200, json={"tenant_id": "tenant-1"})
@@ -276,7 +282,7 @@ class DeployProject:
     control_plane: ControlPlaneDouble
     docker: DockerDouble
     timeline: list[str]
-    uploads: list[tuple[str, str, int]]
+    uploads: list[tuple[str, str, int, dict[str, str] | None]]
 
     def run(self, *args: str, host_url: str = CONTROL_PLANE_URL) -> Result:
         return CliRunner().invoke(
@@ -313,16 +319,21 @@ def deploy_project(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> DeployPro
     timeline: list[str] = []
     control_plane = ControlPlaneDouble(timeline)
     docker = DockerDouble(timeline)
-    uploads: list[tuple[str, str, int]] = []
+    uploads: list[tuple[str, str, int, dict[str, str] | None]] = []
 
     @contextmanager
     def fake_create_archive(config_path: Path, config: dict) -> Iterator[tuple]:
         timeline.append("create_archive")
         yield ARCHIVE
 
-    def fake_upload(signed_url: str, file_path: str, file_size: int) -> None:
+    def fake_upload(
+        signed_url: str,
+        file_path: str,
+        file_size: int,
+        upload_headers: dict[str, str] | None,
+    ) -> None:
         timeline.append("upload_archive")
-        uploads.append((signed_url, file_path, file_size))
+        uploads.append((signed_url, file_path, file_size, upload_headers))
 
     monkeypatch.setattr(deploy_module, "_no_input", False)
     monkeypatch.setattr(deploy_module, "_emitter", None)
@@ -699,6 +710,9 @@ def test_deployment_not_created_by_the_cli_gets_an_actionable_error(
 def test_remote_build_creates_an_internal_source_deployment_and_uploads_the_archive(
     deploy_project: DeployProject,
 ) -> None:
+    upload_headers = {"Content-Type": "application/gzip", "x-ms-blob-type": "BlockBlob"}
+    deploy_project.control_plane.upload_url_extra = {"upload_headers": upload_headers}
+
     result = deploy_project.run("--remote", "--install-command", "yarn install")
 
     assert result.exit_code == 0, result.output
@@ -713,7 +727,12 @@ def test_remote_build_creates_an_internal_source_deployment_and_uploads_the_arch
     assert deploy_project.control_plane.bodies[CREATE_DEPLOYMENT]["source"] == (
         "internal_source"
     )
-    assert deploy_project.uploads == [(SIGNED_UPLOAD_URL, ARCHIVE[0], ARCHIVE[1])]
+    assert deploy_project.control_plane.bodies[_upload_url(CREATED_ID)] == {
+        "supports_upload_headers": True
+    }
+    assert deploy_project.uploads == [
+        (SIGNED_UPLOAD_URL, ARCHIVE[0], ARCHIVE[1], upload_headers)
+    ]
     assert deploy_project.control_plane.bodies[_patch(CREATED_ID)] == {
         "revision_source": "internal_source",
         "source_revision_config": {

libs/cli/tests/unit_tests/test_deploy_helpers.py

@@ -1414,3 +1414,59 @@ class TestPlacementOnAKnownListener:
     def test_refuses_and_names_the_namespaces(self, request_, listener, message):
         with pytest.raises(click.UsageError, match=message):
             request_.on(listener)
+
+
+DEFAULT_UPLOAD_HEADERS = {
+    "content-type": "application/gzip",
+    "content-length": "7",
+    "x-goog-content-length-range": "0,209715200",
+}
+
+
+@pytest.mark.parametrize(
+    ("upload_headers", "expected"),
+    [
+        pytest.param(None, DEFAULT_UPLOAD_HEADERS, id="missing_sends_defaults"),
+        pytest.param({}, DEFAULT_UPLOAD_HEADERS, id="empty_sends_defaults"),
+        pytest.param(
+            {
+                "content-type": "application/x-gzip",
+                "x-ms-blob-type": "BlockBlob",
+                "content-length": "1",
+            },
+            {
+                "content-type": "application/x-gzip",
+                "x-ms-blob-type": "BlockBlob",
+                "content-length": "7",
+            },
+            id="returned_headers_replace_defaults",
+        ),
+        pytest.param(
+            {"x-ms-blob-type": "BlockBlob"},
+            {
+                "content-type": "application/gzip",
+                "x-ms-blob-type": "BlockBlob",
+                "content-length": "7",
+            },
+            id="returned_headers_keep_a_content_type",
+        ),
+    ],
+)
+def test_source_upload_sends_the_headers_the_signed_url_requires(
+    tmp_path, mocker, upload_headers, expected
+):
+    archive = tmp_path / "source.tar.gz"
+    archive.write_bytes(b"tarball")
+    sent = []
+    mocker.patch(
+        "langgraph_cli.deploy._get_emitter", return_value=MagicMock(json_mode=True)
+    )
+    mocker.patch(
+        "urllib.request.urlopen", side_effect=lambda req, timeout: sent.append(req)
+    )
+
+    deploy_mod._upload_to_gcs(
+        "https://storage.example/signed", str(archive), 7, upload_headers
+    )
+
+    assert {name.lower(): value for name, value in sent[0].header_items()} == expected