WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-09-20 08:37:59 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor(cli): make the control plane client source-agnostic

Browse Source
This commit is contained in:
Hugo Durand
2026-09-18 13:44:04 -04:00
parent 6063a65bbe
commit 8d3ad6474e
3 changed files with 107 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/cli/langgraph_cli/host_backend.py
+10 -34
View File
@@ -153,21 +153,19 @@ class HostBackendClient:
def create_deployment(
self,
*,
name: str,
deployment_type: str,
source: str,
config_path: str | None = None,
source_config: dict[str, Any],
source_revision_config: dict[str, Any],
secrets: list[dict[str, str]] | None = None,
) -> dict[str, Any]:
"""Create a deployment."""
payload: dict[str, Any] = {
"name": name,
"source": source,
"source_config": {"deployment_type": deployment_type},
"source_revision_config": {},
"source_config": source_config,
"source_revision_config": source_revision_config,
}
if source == "internal_source" and config_path:
payload["source_revision_config"]["langgraph_config_path"] = config_path
if secrets is not None:
payload["secrets"] = secrets
return self._request("POST", "/v2/deployments", payload)
@@ -202,43 +200,21 @@ class HostBackendClient:
self,
deployment_id: str,
image_uri: str,
*,
revision_source: str | None,
secrets: list[dict[str, str]] | None = None,
tracked_packages: list[str] | None = None,
) -> dict[str, Any]:
payload: dict[str, Any] = {
"revision_source": "internal_docker",
"source_revision_config": {"image_uri": image_uri},
}
if revision_source is not None:
payload["revision_source"] = revision_source
if tracked_packages:
payload["tracked_packages"] = tracked_packages
if secrets is not None:
payload["secrets"] = secrets
return self._request(
"PATCH",
f"/v2/deployments/{deployment_id}",
payload,
)
def update_deployment_external(
self,
deployment_id: str,
image_uri: str,
secrets: list[dict[str, str]] | None = None,
tracked_packages: list[str] | None = None,
) -> dict[str, Any]:
"""Update a deployment with a pre-built external image."""
payload: dict[str, Any] = {
"source_revision_config": {"image_uri": image_uri},
}
if tracked_packages:
payload["tracked_packages"] = tracked_packages
if secrets is not None:
payload["secrets"] = secrets
return self._request(
"PATCH",
f"/v2/deployments/{deployment_id}",
payload,
)
return self._request("PATCH", f"/v2/deployments/{deployment_id}", payload)
def update_deployment_internal_source(
self,