mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-08-20 14:42:28 +02:00
David Asamuandhari-dhanushkodi
72b765ef7f
- Added functions to parse .env files and resolve environment variables from configuration. - Introduced a context manager for temporary Docker configuration using authentication tokens. - Updated Docker capabilities check to include optional requirements for Docker Compose and Buildx. - Enhanced the HostBackendClient with new methods for listing and retrieving deployments and revisions. - Improved the Progress class to display elapsed time during tasks.
113 lines
3.6 KiB
Python
113 lines
3.6 KiB
Python
"""HTTP client for LangGraph host backend deployments."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
from typing import Any
|
|
|
|
import click
|
|
|
|
|
|
class HostBackendError(click.ClickException):
|
|
"""Raised when the host backend returns an error response."""
|
|
|
|
|
|
class HostBackendClient:
|
|
"""Minimal JSON HTTP client for the host backend deployment service."""
|
|
|
|
def __init__(self, base_url: str, api_key: str):
|
|
if not base_url:
|
|
raise click.UsageError("Host backend URL is required")
|
|
base_url = base_url.rstrip("/")
|
|
self._base_url = base_url
|
|
self._api_key = api_key
|
|
|
|
def _request(
|
|
self, method: str, path: str, payload: dict[str, Any] | None = None
|
|
) -> Any:
|
|
url = f"{self._base_url}{path}"
|
|
data: bytes | None
|
|
if payload is not None:
|
|
data = json.dumps(payload).encode("utf-8")
|
|
else:
|
|
data = None
|
|
headers: dict[str, str] = {
|
|
"X-Api-Key": self._api_key,
|
|
"Accept": "application/json",
|
|
}
|
|
if data is not None:
|
|
headers["Content-Type"] = "application/json"
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
body = resp.read()
|
|
except urllib.error.HTTPError as err:
|
|
detail = err.read().decode("utf-8", errors="ignore")
|
|
message = detail or err.reason
|
|
raise HostBackendError(
|
|
f"{method} {path} failed with status {err.code}: {message}"
|
|
) from None
|
|
except urllib.error.URLError as err:
|
|
raise HostBackendError(str(err.reason)) from None
|
|
|
|
if not body:
|
|
return None
|
|
try:
|
|
return json.loads(body)
|
|
except json.JSONDecodeError as err:
|
|
raise HostBackendError(
|
|
f"Failed to decode response from {path}: {err.msg}"
|
|
) from None
|
|
|
|
def create_deployment(self, payload: dict[str, Any]) -> dict[str, Any]:
|
|
return self._request("POST", "/v2/deployments", payload)
|
|
|
|
def list_deployments(self, name_contains: str) -> dict[str, Any]:
|
|
encoded = urllib.parse.quote(name_contains, safe="")
|
|
return self._request("GET", f"/v2/deployments?name_contains={encoded}")
|
|
|
|
def get_deployment(self, deployment_id: str) -> dict[str, Any]:
|
|
return self._request("GET", f"/v2/deployments/{deployment_id}")
|
|
|
|
def request_push_token(self, deployment_id: str) -> dict[str, Any]:
|
|
return self._request(
|
|
"POST",
|
|
f"/v2/deployments/{deployment_id}/push-token",
|
|
)
|
|
|
|
def update_deployment(
|
|
self,
|
|
deployment_id: str,
|
|
image_uri: str,
|
|
secrets: list[dict[str, str]] | None = None,
|
|
) -> dict[str, Any]:
|
|
payload: dict[str, Any] = {
|
|
"source_revision_config": {"image_uri": image_uri},
|
|
}
|
|
if secrets is not None:
|
|
payload["secrets"] = secrets
|
|
return self._request(
|
|
"PATCH",
|
|
f"/v2/deployments/{deployment_id}",
|
|
payload,
|
|
)
|
|
|
|
def list_revisions(
|
|
self, deployment_id: str, limit: int = 1
|
|
) -> dict[str, Any]:
|
|
return self._request(
|
|
"GET",
|
|
f"/v2/deployments/{deployment_id}/revisions?limit={limit}",
|
|
)
|
|
|
|
def get_revision(
|
|
self, deployment_id: str, revision_id: str
|
|
) -> dict[str, Any]:
|
|
return self._request(
|
|
"GET",
|
|
f"/v2/deployments/{deployment_id}/revisions/{revision_id}",
|
|
)
|