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/generate_schema.py

@@ -23,7 +23,6 @@ from langgraph_cli.schemas import (
     GraphDef,
     HttpConfig,
     IndexConfig,
-    PriorityConfig,
     SecurityConfig,
     SerdeConfig,
     StoreConfig,
@@ -114,7 +113,6 @@ def add_descriptions_to_schema(schema, cls):
                 StoreConfig,
                 IndexConfig,
                 AuthConfig,
-                PriorityConfig,
                 SecurityConfig,
                 HttpConfig,
                 CorsConfig,

libs/cli/langgraph_cli/cli.py

@@ -847,7 +847,6 @@ def dev(
         store=config_json.get("store"),
         wait_for_client=wait_for_client,
         auth=config_json.get("auth"),
-        priority=config_json.get("priority"),
         http=config_json.get("http"),
         ui=config_json.get("ui"),
         ui_config=config_json.get("ui_config"),

libs/cli/langgraph_cli/config.py

@@ -442,7 +442,6 @@ def validate_config(
         "store": config.get("store"),
         "auth": config.get("auth"),
         "encryption": config.get("encryption"),
-        "priority": config.get("priority"),
         "http": config.get("http"),
         # Pass through webhooks config so it can be injected into the image
         "webhooks": config.get("webhooks"),
@@ -607,19 +606,6 @@ def validate_config(
                     f"Invalid encryption.path format: '{encryption_conf['path']}'. "
                     "Must be in format './path/to/file.py:attribute_name'"
                 )
-    # Validate priority config
-    if priority_conf := config.get("priority"):
-        if "path" in priority_conf:
-            if ":" not in priority_conf["path"]:
-                raise ValueError(
-                    f"Invalid priority.path format: '{priority_conf['path']}'. "
-                    "Must be in format './path/to/file.py:attribute_name'"
-                )
-            if _is_node_graph(priority_conf["path"]):
-                raise ValueError(
-                    "JavaScript priority callbacks are not supported: "
-                    f"'{priority_conf['path']}'. Use a Python file or module instead."
-                )
     if http_conf := config.get("http"):
         if "app" in http_conf:
             if ":" not in http_conf["app"]:
@@ -663,7 +649,6 @@ _KNOWN_CONFIG_KEYS = {
     "store",
     "auth",
     "encryption",
-    "priority",
     "http",
     "webhooks",
     "checkpointer",
@@ -1095,49 +1080,6 @@ def _update_encryption_path(
     )
 
 
-def _update_priority_path(
-    config_path: pathlib.Path, config: Config, local_deps: LocalDeps
-) -> None:
-    """Update priority.path to use Docker container paths."""
-    priority_conf = config.get("priority")
-    if not priority_conf or not (path_str := priority_conf.get("path")):
-        return
-
-    module_str, sep, attr_str = path_str.partition(":")
-    if not sep or not module_str.startswith("."):
-        return  # Already validated or absolute path
-
-    resolved = config_path.parent / module_str
-    if not resolved.exists():
-        raise FileNotFoundError(
-            f"Priority file not found: {resolved} (from {path_str})"
-        )
-    if not resolved.is_file():
-        raise IsADirectoryError(f"Priority path must be a file: {resolved}")
-
-    # Check faux packages first (they take precedence)
-    for faux_path, (_, destpath) in local_deps.faux_pkgs.items():
-        if resolved.is_relative_to(faux_path):
-            new_path = f"{destpath}/{resolved.relative_to(faux_path)}:{attr_str}"
-            priority_conf["path"] = new_path
-            return
-
-    # Check real packages
-    for real_path in local_deps.real_pkgs:
-        if resolved.is_relative_to(real_path):
-            new_path = (
-                f"/deps/{real_path.name}/{resolved.relative_to(real_path)}:{attr_str}"
-            )
-            priority_conf["path"] = new_path
-            return
-
-    raise ValueError(
-        f"Priority file '{resolved}' not covered by dependencies.\n"
-        "Add its parent directory to the 'dependencies' array in your config.\n"
-        f"Current dependencies: {config['dependencies']}"
-    )
-
-
 def _update_checkpointer_path(
     config_path: pathlib.Path, config: Config, local_deps: LocalDeps
 ) -> None:
@@ -1271,9 +1213,6 @@ def _build_runtime_env_vars(config: Config) -> list[str]:
     if (encryption_config := config.get("encryption")) is not None:
         env_vars.append(f"ENV LANGGRAPH_ENCRYPTION='{json.dumps(encryption_config)}'")
 
-    if (priority_config := config.get("priority")) is not None:
-        env_vars.append(f"ENV LANGGRAPH_PRIORITY='{json.dumps(priority_config)}'")
-
     if (http_config := config.get("http")) is not None:
         env_vars.append(f"ENV LANGGRAPH_HTTP='{json.dumps(http_config)}'")
 
@@ -1441,8 +1380,6 @@ def python_config_to_docker(
     _update_auth_path(config_path, config, local_deps)
     # Rewrite encryption path, so it points to the correct location in the Docker container
     _update_encryption_path(config_path, config, local_deps)
-    # Rewrite priority path, so it points to the correct location in the Docker container
-    _update_priority_path(config_path, config, local_deps)
     # Rewrite checkpointer path, so it points to the correct location in the Docker container
     _update_checkpointer_path(config_path, config, local_deps)
     # Rewrite HTTP app path, so it points to the correct location in the Docker container

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/langgraph_cli/schemas.py

@@ -373,27 +373,6 @@ class EncryptionConfig(TypedDict, total=False):
     """
 
 
-class PriorityConfig(TypedDict, total=False):
-    """Configuration for custom run priority logic.
-
-    Allows you to decide each thread's priority settings with a Python function
-    that the server calls at the first run on a thread with no settings, which
-    can come long after the thread was created."""
-
-    path: str
-    """Required. Path to an instance of the Priority() class that registers an assign function.
-
-    Format: "path/to/file.py:my_priority"
-
-    Example:
-        {
-            "priority": {
-                "path": "./priority.py:my_priority"
-            }
-        }
-    """
-
-
 class CorsConfig(TypedDict, total=False):
     """Specifies Cross-Origin Resource Sharing (CORS) rules for your server.
 
@@ -774,13 +753,6 @@ class Config(TypedDict, total=False):
     Allows you to implement custom encryption for sensitive data stored in the database.
     """
 
-    priority: PriorityConfig | None
-    """Optional. Custom run priority config, including the path to your Python priority logic.
-
-    Allows you to decide each thread's priority settings at the first run on a thread with no settings,
-    which can come long after the thread was created.
-    """
-
     http: HttpConfig | None
     """Optional. Configuration for the built-in HTTP server, controlling which custom routes are exposed
     and how cross-origin requests are handled.
@@ -812,7 +784,6 @@ __all__ = [
     "CheckpointerConfig",
     "AuthConfig",
     "EncryptionConfig",
-    "PriorityConfig",
     "HttpConfig",
     "MiddlewareOrders",
     "Distros",

libs/cli/langgraph_cli/uv_lock.py

@@ -910,7 +910,6 @@ def python_config_to_docker_uv_lock(
     for section, key in [
         ("auth", "path"),
         ("encryption", "path"),
-        ("priority", "path"),
         ("checkpointer", "path"),
         ("http", "app"),
     ]:

libs/cli/schemas/schema.json

@@ -196,17 +196,6 @@
                 }
               ]
             },
-            "priority": {
-              "anyOf": [
-                {
-                  "$ref": "#/$defs/PriorityConfig"
-                },
-                {
-                  "type": "null"
-                }
-              ],
-              "description": "Optional. Custom run priority config, including the path to your Python priority logic.\n\nAllows you to decide each thread's priority settings at the first run on a thread with no settings,\nwhich can come long after the thread was created.\n"
-            },
             "source": {
               "anyOf": [
                 {
@@ -449,17 +438,6 @@
                 }
               ]
             },
-            "priority": {
-              "anyOf": [
-                {
-                  "$ref": "#/$defs/PriorityConfig"
-                },
-                {
-                  "type": "null"
-                }
-              ],
-              "description": "Optional. Custom run priority config, including the path to your Python priority logic.\n\nAllows you to decide each thread's priority settings at the first run on a thread with no settings,\nwhich can come long after the thread was created.\n"
-            },
             "source": {
               "$ref": "#/$defs/UvSource"
             },
@@ -688,17 +666,6 @@
                 }
               ]
             },
-            "priority": {
-              "anyOf": [
-                {
-                  "$ref": "#/$defs/PriorityConfig"
-                },
-                {
-                  "type": "null"
-                }
-              ],
-              "description": "Optional. Custom run priority config, including the path to your Python priority logic.\n\nAllows you to decide each thread's priority settings at the first run on a thread with no settings,\nwhich can come long after the thread was created.\n"
-            },
             "source": {
               "anyOf": [
                 {
@@ -1203,18 +1170,6 @@
       },
       "required": []
     },
-    "PriorityConfig": {
-      "title": "PriorityConfig",
-      "description": "Configuration for custom run priority logic.\n\nAllows you to decide each thread's priority settings with a Python function\nthat the server calls at the first run on a thread with no settings, which\ncan come long after the thread was created.",
-      "type": "object",
-      "properties": {
-        "path": {
-          "type": "string",
-          "description": "Required. Path to an instance of the Priority() class that registers an assign function.\n"
-        }
-      },
-      "required": []
-    },
     "UvSource": {
       "title": "UvSource",
       "description": "Deployment source rooted at a uv project or workspace.",

libs/cli/schemas/schema.v0.json

@@ -196,17 +196,6 @@
                 }
               ]
             },
-            "priority": {
-              "anyOf": [
-                {
-                  "$ref": "#/$defs/PriorityConfig"
-                },
-                {
-                  "type": "null"
-                }
-              ],
-              "description": "Optional. Custom run priority config, including the path to your Python priority logic.\n\nAllows you to decide each thread's priority settings at the first run on a thread with no settings,\nwhich can come long after the thread was created.\n"
-            },
             "source": {
               "anyOf": [
                 {
@@ -449,17 +438,6 @@
                 }
               ]
             },
-            "priority": {
-              "anyOf": [
-                {
-                  "$ref": "#/$defs/PriorityConfig"
-                },
-                {
-                  "type": "null"
-                }
-              ],
-              "description": "Optional. Custom run priority config, including the path to your Python priority logic.\n\nAllows you to decide each thread's priority settings at the first run on a thread with no settings,\nwhich can come long after the thread was created.\n"
-            },
             "source": {
               "$ref": "#/$defs/UvSource"
             },
@@ -688,17 +666,6 @@
                 }
               ]
             },
-            "priority": {
-              "anyOf": [
-                {
-                  "$ref": "#/$defs/PriorityConfig"
-                },
-                {
-                  "type": "null"
-                }
-              ],
-              "description": "Optional. Custom run priority config, including the path to your Python priority logic.\n\nAllows you to decide each thread's priority settings at the first run on a thread with no settings,\nwhich can come long after the thread was created.\n"
-            },
             "source": {
               "anyOf": [
                 {
@@ -1203,18 +1170,6 @@
       },
       "required": []
     },
-    "PriorityConfig": {
-      "title": "PriorityConfig",
-      "description": "Configuration for custom run priority logic.\n\nAllows you to decide each thread's priority settings with a Python function\nthat the server calls at the first run on a thread with no settings, which\ncan come long after the thread was created.",
-      "type": "object",
-      "properties": {
-        "path": {
-          "type": "string",
-          "description": "Required. Path to an instance of the Priority() class that registers an assign function.\n"
-        }
-      },
-      "required": []
-    },
     "UvSource": {
       "title": "UvSource",
       "description": "Deployment source rooted at a uv project or workspace.",

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

@@ -2,12 +2,10 @@ import json
 import pathlib
 import re
 import shutil
-import sys
 import tempfile
 import textwrap
 from contextlib import contextmanager
 from pathlib import Path
-from unittest.mock import MagicMock
 
 import click
 from click.testing import CliRunner
@@ -373,32 +371,6 @@ def test_dev_command_requires_ssl_certfile_and_keyfile_together(tmp_path) -> Non
     assert "Both --ssl-certfile and --ssl-keyfile must be provided" in result.output
 
 
-def test_dev_command_passes_priority_to_the_server(tmp_path, monkeypatch) -> None:
-    config_path = tmp_path / "langgraph.json"
-    config_path.write_text(
-        json.dumps(
-            {
-                "dependencies": ["."],
-                "graphs": {"agent": "./agent.py:graph"},
-                "priority": {"path": "./priority.py:priority"},
-            }
-        ),
-        encoding="utf-8",
-    )
-    # The test environment doesn't install langgraph-api, so stand in for its server.
-    server = MagicMock()
-    monkeypatch.setitem(sys.modules, "langgraph_api.cli", server)
-    # dev appends the working directory and its dependencies to sys.path.
-    monkeypatch.setattr(sys, "path", list(sys.path))
-
-    result = CliRunner().invoke(cli, ["dev", "--config", str(config_path)])
-
-    assert result.exit_code == 0, result.output
-    assert server.run_server.call_args.kwargs["priority"] == {
-        "path": "./priority.py:priority"
-    }
-
-
 def test_deploy_list_command(monkeypatch) -> None:
     runner = CliRunner()
     captured: dict[str, str] = {}

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_config.py

@@ -16,7 +16,6 @@ from langgraph_cli.config import (
     config_to_docker,
     default_base_image,
     docker_tag,
-    get_unknown_keys,
     has_disallowed_build_command_content,
     validate_config,
     validate_config_file,
@@ -156,7 +155,6 @@ def test_validate_config():
         "store": None,
         "auth": None,
         "encryption": None,
-        "priority": None,
         "webhooks": None,
         "checkpointer": None,
         "http": None,
@@ -186,7 +184,6 @@ def test_validate_config():
         "store": None,
         "auth": None,
         "encryption": None,
-        "priority": None,
         "webhooks": None,
         "checkpointer": None,
         "http": None,
@@ -1292,109 +1289,6 @@ def test_config_to_docker_python_encryption_formatted():
     )
 
 
-@pytest.mark.parametrize(
-    "path", ["./priority.py:priority", "my_package.priority:priority"]
-)
-def test_config_to_docker_python_priority(path: str):
-    # Test that priority config is kept by validation and isn't an unknown key
-    raw_config = {
-        "python_version": "3.11",
-        "graphs": {"agent": "./agent.py:graph"},
-        "dependencies": ["."],
-        "priority": {"path": path},
-    }
-    validated = validate_config(raw_config)
-
-    assert validated["priority"] == {"path": path}
-    assert get_unknown_keys(raw_config) == []
-
-
-def test_config_to_docker_python_priority_bad_path():
-    # Test that invalid priority path format raises ValueError
-    with pytest.raises(ValueError, match="Invalid priority.path format"):
-        validate_config(
-            {
-                "python_version": "3.11",
-                "graphs": {"agent": "./agent.py:graph"},
-                "dependencies": ["."],
-                "priority": {"path": "./priority.py"},  # Missing :attribute
-            }
-        )
-
-
-@pytest.mark.parametrize("ext", [".ts", ".mts", ".cts", ".js", ".mjs", ".cjs"])
-def test_config_to_docker_python_priority_js_path(ext: str):
-    # Test that a JavaScript priority path is rejected
-    with pytest.raises(ValueError, match="JavaScript priority callbacks"):
-        validate_config(
-            {
-                "python_version": "3.11",
-                "graphs": {"agent": "./agent.py:graph"},
-                "dependencies": ["."],
-                "priority": {"path": f"./priority{ext}:priority"},
-            }
-        )
-
-
-def test_config_to_docker_nodejs_priority():
-    # Test that a Node.js-only project can name a Python priority file
-    actual_docker_stdin, _ = config_to_docker(
-        PATH_TO_CONFIG,
-        validate_config(
-            {
-                "node_version": "20",
-                "graphs": {"agent": "./graphs/agent.js:graph"},
-                "priority": {"path": "./priority.py:priority"},
-            }
-        ),
-        base_image="langchain/langgraphjs-api",
-    )
-    assert (
-        """ENV LANGGRAPH_PRIORITY='{"path": "./priority.py:priority"}'"""
-        in actual_docker_stdin
-    )
-
-
-def test_config_to_docker_python_priority_formatted():
-    # Test that priority config is written to the Docker output with its container path
-    actual_docker_stdin, _ = config_to_docker(
-        PATH_TO_CONFIG,
-        validate_config(
-            {
-                "python_version": "3.11",
-                "dependencies": ["."],
-                "graphs": {"agent": "./graphs/agent.py:graph"},
-                "priority": {"path": "./agent.py:my_priority"},
-            }
-        ),
-        base_image="langchain/langgraph-api",
-    )
-    assert (
-        "ENV LANGGRAPH_PRIORITY="
-        """'{"path": "/deps/outer-unit_tests/unit_tests/agent.py:my_priority"}'"""
-        in actual_docker_stdin
-    )
-
-
-def test_config_to_docker_python_priority_not_in_dependencies():
-    # Test that a priority file outside the dependencies raises ValueError
-    with pytest.raises(
-        ValueError, match="Priority file .* not covered by dependencies"
-    ):
-        config_to_docker(
-            PATH_TO_CONFIG,
-            validate_config(
-                {
-                    "python_version": "3.11",
-                    "dependencies": ["./graphs"],
-                    "graphs": {"agent": "./graphs/agent.py:graph"},
-                    "priority": {"path": "./agent.py:my_priority"},
-                }
-            ),
-            base_image="langchain/langgraph-api",
-        )
-
-
 def test_config_to_docker_nodejs_internal_docker_tag():
     graphs = {"agent": "./graphs/agent.js:graph"}
     actual_docker_stdin, additional_contexts = config_to_docker(
@@ -1837,38 +1731,6 @@ def test_config_to_docker_uv_lock_honors_root_workspace_sources():
         )
 
 
-def test_config_to_docker_uv_lock_rewrites_priority_path():
-    with tempfile.TemporaryDirectory() as tmpdir:
-        tmpdir_path = pathlib.Path(tmpdir)
-        project_root, config_path = _write_uv_lock_workspace(
-            tmpdir_path,
-            agent_sources="[tool.uv.sources]\nshared = { workspace = true }",
-        )
-        (project_root / "libs" / "shared" / "src" / "shared" / "priority.py").touch()
-
-        config = validate_config(
-            {
-                "python_version": "3.11",
-                "graphs": {
-                    "agent": "../../apps/agent/src/agent/graph.py:graph",
-                },
-                "source": {"kind": "uv", "root": "../..", "package": "agent"},
-                "priority": {
-                    "path": "../../libs/shared/src/shared/priority.py:priority"
-                },
-            }
-        )
-        docker, _ = config_to_docker(
-            config_path, config, base_image="langchain/langgraph-api:0.2.47"
-        )
-
-        assert (
-            "ENV LANGGRAPH_PRIORITY="
-            """'{"path": "/deps/workspace/libs/shared/src/shared/priority.py:priority"}'"""
-            in docker
-        )
-
-
 def test_config_to_docker_uv_lock_ignores_unrelated_workspace_package_sources():
     with tempfile.TemporaryDirectory() as tmpdir:
         tmpdir_path = pathlib.Path(tmpdir)

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

libs/langgraph/uv.lock

@@ -1919,7 +1919,7 @@ requires-dist = [
     { name = "langchain-core", specifier = ">=1.4.0,<2" },
     { name = "langchain-protocol", specifier = ">=0.0.15" },
     { name = "orjson", specifier = ">=3.11.5" },
-    { name = "websockets", specifier = ">=14" },
+    { name = "websockets", specifier = ">=14,<17" },
 ]
 
 [package.metadata.requires-dev]

libs/prebuilt/uv.lock

@@ -614,7 +614,7 @@ requires-dist = [
     { name = "langchain-core", specifier = ">=1.4.0,<2" },
     { name = "langchain-protocol", specifier = ">=0.0.15" },
     { name = "orjson", specifier = ">=3.11.5" },
-    { name = "websockets", specifier = ">=14" },
+    { name = "websockets", specifier = ">=14,<17" },
 ]
 
 [package.metadata.requires-dev]

libs/sdk-py/CHANGELOG.md

@@ -34,14 +34,6 @@
 - **`thread.output`** — awaitable that resolves to the terminal thread state
   `values` dict after the run lifecycle completes.
 
-- **`Priority` (beta)** — register an async function with `@priority.assign`
-  and name the object in the `priority` section of `langgraph.json`. The
-  server calls the function with a `PriorityContext` at the first run on a
-  thread with no settings and stores the `PrioritySettings` it returns
-  (`priority`, `fairness_group`, `fairness_weight`) on the thread. Requires
-  `langgraph-api` 0.18.0 or later and a `langgraph-cli` that supports the
-  section; older versions of either ignore it.
-
 ### Changed
 
 - `client.threads.stream()` now accepts `transport="sse"` (default) or

libs/sdk-py/langgraph_sdk/__init__.py

@@ -2,8 +2,6 @@ from langgraph_sdk.auth import Auth
 from langgraph_sdk.client import get_client, get_sync_client
 from langgraph_sdk.encryption import Encryption
 from langgraph_sdk.encryption.types import DecryptResult, EncryptionContext
-from langgraph_sdk.priority import Priority
-from langgraph_sdk.priority.types import PriorityContext, PrioritySettings
 
 __version__ = "0.4.8"
 
@@ -12,9 +10,6 @@ __all__ = [
     "DecryptResult",
     "Encryption",
     "EncryptionContext",
-    "Priority",
-    "PriorityContext",
-    "PrioritySettings",
     "get_client",
     "get_sync_client",
 ]

libs/sdk-py/langgraph_sdk/priority/__init__.py

@@ -1,155 +0,0 @@
-"""Custom run priority for LangGraph.
-
-.. warning::
-    This API is in beta and may change in future versions.
-
-This module lets developers decide the priority settings of each thread with
-an async function that the server calls at the first run on a thread with no
-settings. The function is registered on an object that `langgraph.json` names,
-and runs server-side.
-"""
-
-from __future__ import annotations
-
-import functools
-import inspect
-import typing
-import warnings
-from collections.abc import Awaitable, Callable
-
-from langgraph_sdk.encryption import DuplicateHandlerError, LangGraphBetaWarning
-from langgraph_sdk.priority import types
-
-_AssignHandler = Callable[
-    [types.PriorityContext], Awaitable[types.PrioritySettings | None]
-]
-_AssignHandlerT = typing.TypeVar("_AssignHandlerT", bound=_AssignHandler)
-
-
-@functools.lru_cache(maxsize=1)
-def _warn_priority_beta() -> None:
-    warnings.warn(
-        "The Priority API is in beta and may change in future versions.",
-        LangGraphBetaWarning,
-        stacklevel=4,
-    )
-
-
-def _validate_handler(fn: typing.Callable) -> None:
-    """Validate that the assign handler is async and takes one parameter.
-
-    Raises:
-        TypeError: If the handler is not an async function or has wrong parameter count
-    """
-    if not inspect.iscoroutinefunction(fn):
-        raise TypeError(
-            f"Priority assign handler must be an async function, got {type(fn)}"
-        )
-
-    sig = inspect.signature(fn)
-    params = [
-        p
-        for p in sig.parameters.values()
-        if p.kind in (p.POSITIONAL_ONLY, p.POSITIONAL_OR_KEYWORD)
-    ]
-    if len(params) != 1:
-        raise TypeError(
-            "Priority assign handler must accept exactly 1 parameter (ctx), "
-            f"got {len(params)}"
-        )
-
-
-class Priority:
-    """Decide the priority settings of your LangGraph application's threads.
-
-    .. warning::
-        This API is in beta and may change in future versions.
-
-    To use, create a separate Python file and add the path to the file to your
-    LangGraph API configuration file (`langgraph.json`). Within that file, create
-    an instance of the Priority class and register an assign function with
-    `@priority.assign`.
-
-    Example `langgraph.json` file:
-
-    ```json
-    {
-      "dependencies": ["."],
-      "graphs": {
-        "agent": "./my_agent/agent.py:graph"
-      },
-      "env": ".env",
-      "priority": {
-        "path": "./priority.py:priority"
-      }
-    }
-    ```
-
-    The server calls the assign function at the first run on a thread with no
-    settings, which can come long after the thread was created, and stores the
-    settings it returns on the thread. Later runs on the thread use the stored
-    settings. The function can be called more than once for one thread, for
-    example after it raises, and only one result is stored, so keep it free of
-    side effects.
-
-    ???+ example "Basic Usage"
-
-        ```python
-        from langgraph_sdk import Priority, PriorityContext, PrioritySettings
-
-        priority = Priority()
-
-        @priority.assign
-        async def assign(ctx: PriorityContext) -> PrioritySettings | None:
-            return {
-                "priority": 3 if "premium" in ctx.permissions else 0,
-                # Stored unencrypted, so an opaque ID rather than personal data.
-                "fairness_group": ctx.user["org_id"],
-            }
-        ```
-    """
-
-    __slots__ = ("_assign_handler",)
-
-    types = types
-    """Reference to priority type definitions.
-
-    Provides access to the type definitions used by the priority system:
-    PriorityContext and PrioritySettings.
-    """
-
-    def __init__(self) -> None:
-        """Initialize the Priority instance."""
-        _warn_priority_beta()
-        self._assign_handler: _AssignHandler | None = None
-
-    def assign(self, fn: _AssignHandlerT) -> _AssignHandlerT:
-        """Register the function that assigns a thread's priority settings.
-
-        The function receives a `PriorityContext` and returns `PrioritySettings`,
-        or `None` for the defaults.
-
-        Example:
-            ```python
-            @priority.assign
-            async def assign(ctx: PriorityContext) -> PrioritySettings | None:
-                if ctx.cron_id is not None:
-                    return {"priority": -2}
-                return None
-            ```
-
-        Args:
-            fn: The assign handler function
-
-        Returns:
-            The registered handler function
-
-        Raises:
-            DuplicateHandlerError: If an assign handler is already registered
-            TypeError: If the handler has an invalid signature
-        """
-        if self._assign_handler is not None:
-            raise DuplicateHandlerError("Priority assign handler already registered")
-        _validate_handler(fn)
-        self._assign_handler = fn
-        return fn

libs/sdk-py/langgraph_sdk/priority/types.py

@@ -1,67 +0,0 @@
-"""Priority types for LangGraph.
-
-This module defines the context the server passes to the function registered
-with `@priority.assign`, and the settings that function returns.
-"""
-
-from __future__ import annotations
-
-import typing
-from collections.abc import Sequence
-from dataclasses import dataclass
-
-from langgraph_sdk.auth.types import BaseUser
-
-
-@dataclass(frozen=True, slots=True)
-class PriorityContext:
-    """Context passed to the `@priority.assign` function at the first run on a thread with no settings."""
-
-    user: BaseUser
-    """The user from the deployment's auth, custom auth or LangSmith auth. Without auth, `user.identity` is empty."""
-
-    permissions: Sequence[str]
-    """The user's permissions from the same auth. Without auth, it's empty."""
-
-    thread_id: str | None
-    """The thread's ID, or `None` when the run creates its thread without being given an ID."""
-
-    thread_metadata: dict[str, typing.Any]
-    """The thread's metadata."""
-
-    assistant_id: str
-    """The ID of the assistant the run uses."""
-
-    graph_id: str
-    """The ID of the graph the assistant runs."""
-
-    assistant_metadata: dict[str, typing.Any]
-    """The assistant's metadata."""
-
-    run_metadata: dict[str, typing.Any]
-    """The run's metadata."""
-
-    cron_id: str | None
-    """The ID of the cron that started the run, or `None` when no cron started it."""
-
-
-class PrioritySettings(typing.TypedDict, total=False):
-    """Priority settings returned by the `@priority.assign` function.
-
-    Every key is optional: a missing key or a key set to `None` takes its
-    default, and returning `None` instead of a dict gives every key its default.
-    A value outside its range, or an empty `fairness_group`, is an error.
-    """
-
-    priority: int | None
-    """An integer from -5 to 5, where 5 runs first and -5 runs last. Defaults to 0."""
-
-    fairness_group: str | None
-    """The thread's fairness group, a non-empty string. Defaults to `"default"`.
-
-    The server stores it unencrypted, even when the deployment uses custom
-    encryption, so use an opaque ID rather than personal data.
-    """
-
-    fairness_weight: int | None
-    """The weight of the thread's fairness group, an integer from 1 to 100. Defaults to 1."""

libs/sdk-py/pyproject.toml

@@ -16,7 +16,7 @@ dependencies = [
     "orjson>=3.11.5",
     "langchain-protocol>=0.0.15",
     "langchain-core>=1.4.0,<2",
-    "websockets>=14",
+    "websockets>=14,<17",
 ]
 
 [tool.hatch.version]

libs/sdk-py/tests/test_priority.py

@@ -1,128 +0,0 @@
-import dataclasses
-
-import pytest
-
-import langgraph_sdk
-from langgraph_sdk import Priority, PriorityContext, PrioritySettings
-from langgraph_sdk.encryption import DuplicateHandlerError, LangGraphBetaWarning
-from langgraph_sdk.priority import _warn_priority_beta
-
-
-class _User:
-    is_authenticated = True
-    display_name = "user-1"
-    identity = "user-1"
-    permissions: tuple[str, ...] = ()
-
-    def __getitem__(self, key):
-        return getattr(self, key)
-
-    def __contains__(self, key):
-        return hasattr(self, key)
-
-    def __iter__(self):
-        return iter(("identity", "display_name", "is_authenticated", "permissions"))
-
-
-def _context(thread_id: str | None) -> PriorityContext:
-    return PriorityContext(
-        user=_User(),
-        permissions=[],
-        thread_id=thread_id,
-        thread_metadata={},
-        assistant_id="assistant-1",
-        graph_id="agent",
-        assistant_metadata={"tier": "gold"},
-        run_metadata={},
-        cron_id=None,
-    )
-
-
-async def _no_params() -> PrioritySettings | None:
-    return None
-
-
-async def _two_params(_ctx: PriorityContext, _extra: object) -> PrioritySettings | None:
-    return None
-
-
-def test_priority_and_its_types_are_exported():
-    assert {"Priority", "PriorityContext", "PrioritySettings"} <= set(
-        langgraph_sdk.__all__
-    )
-    assert Priority.types.PriorityContext is PriorityContext
-    assert Priority.types.PrioritySettings is PrioritySettings
-
-
-def test_construction_warns_that_priority_is_beta():
-    # The warning fires once per process, so an earlier test may have used it up.
-    _warn_priority_beta.cache_clear()
-    with pytest.warns(LangGraphBetaWarning, match="Priority API is in beta"):
-        Priority()
-
-
-def test_assign_registers_and_returns_the_function():
-    priority = Priority()
-
-    async def assign(_ctx: PriorityContext) -> PrioritySettings | None:
-        return {"priority": 3, "fairness_group": "org-42"}
-
-    assert priority.assign(assign) is assign
-    assert priority._assign_handler is assign
-
-
-def test_assign_rejects_a_sync_function():
-    priority = Priority()
-
-    def assign(_ctx: PriorityContext) -> PrioritySettings | None:
-        return None
-
-    with pytest.raises(TypeError, match="must be an async function"):
-        priority.assign(assign)  # ty: ignore[invalid-argument-type]
-    assert priority._assign_handler is None
-
-
-@pytest.mark.parametrize("fn", [_no_params, _two_params])
-def test_assign_rejects_the_wrong_number_of_parameters(fn):
-    priority = Priority()
-
-    with pytest.raises(TypeError, match="must accept exactly 1 parameter"):
-        priority.assign(fn)
-    assert priority._assign_handler is None
-
-
-def test_assign_rejects_a_second_function():
-    priority = Priority()
-
-    @priority.assign
-    async def first(_ctx: PriorityContext) -> PrioritySettings | None:
-        return None
-
-    with pytest.raises(DuplicateHandlerError):
-
-        @priority.assign
-        async def second(_ctx: PriorityContext) -> PrioritySettings | None:
-            return None
-
-    assert priority._assign_handler is first
-
-
-def test_context_accepts_no_thread_id():
-    # `make lint` type-checks this call too, so `thread_id` must accept None there.
-    assert _context(thread_id=None).thread_id is None
-
-
-def test_context_is_frozen():
-    ctx = _context(thread_id="thread-1")
-
-    with pytest.raises(dataclasses.FrozenInstanceError):
-        ctx.thread_id = "thread-2"  # ty: ignore[invalid-assignment]
-
-
-def test_every_settings_key_is_optional():
-    assert PrioritySettings.__optional_keys__ == {
-        "priority",
-        "fairness_group",
-        "fairness_weight",
-    }
-    assert PrioritySettings.__required_keys__ == frozenset()

libs/sdk-py/uv.lock

@@ -525,7 +525,7 @@ requires-dist = [
     { name = "langchain-core", specifier = ">=1.4.0,<2" },
     { name = "langchain-protocol", specifier = ">=0.0.15" },
     { name = "orjson", specifier = ">=3.11.5" },
-    { name = "websockets", specifier = ">=14" },
+    { name = "websockets", specifier = ">=14,<17" },
 ]
 
 [package.metadata.requires-dev]