WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-09-25 19:15:11 +02:00
Code Issues Packages Projects Releases Wiki Activity

langgaph cli

Browse Source
This commit is contained in:
syachamaneni-lc
2026-09-22 11:32:33 -07:00
parent 49cce0ca85
commit ab851822c5
4 changed files with 221 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/cli/langgraph_cli/deploy.py
+101 -22
View File
@@ -606,6 +606,7 @@ def _resolve_deployment(
name: str | None,
*,
not_found_message: str,
agent: dict[str, str] | None = None,
) -> tuple[str | None, bool, int]:
"""Resolve an existing deployment by ID or exact name match."""
needs_creation = False
@@ -616,10 +617,29 @@ def _resolve_deployment(
)
return deployment_id, needs_creation, step + 1
_log_deploy_step(step, f"Looking up deployment '{name}'")
found_id = _call_host_backend_with_optional_tenant(
client, lambda c: find_deployment_id_by_name(c, name)
)
if agent is not None:
_log_deploy_step(
step, f"Looking up agent '{agent['agent_id']}' in {agent['environment']}"
)
existing = _call_host_backend_with_optional_tenant(
client,
lambda c: c.list_deployments(
agent_id=agent["agent_id"], agent_environment=agent["environment"]
),
)
found_id = next(
(
dep["id"]
for dep in existing.get("resources", [])
if not dep.get("is_preview")
),
None,
)
else:
_log_deploy_step(step, f"Looking up deployment '{name}'")
found_id = _call_host_backend_with_optional_tenant(
client, lambda c: find_deployment_id_by_name(c, name)
)
em = _get_emitter()
if found_id:
deployment_id = str(found_id)
@@ -634,26 +654,43 @@ def _create_deployment(
client: HostBackendClient,
step: int,
*,
name: str,
name: str | None,
deployment_type: str,
source: str,
config_rel: str | None = None,
secrets: list[dict[str, str]] | None = None,
agent: dict[str, 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}'")
created = client.create_deployment(
name=name,
deployment_type=deployment_type,
source=source,
config_path=config_rel,
secrets=secrets,
_log_deploy_step(
step,
f"Creating deployment for agent '{agent['agent_id']}' in {agent['environment']}"
if agent is not None
else f"Creating deployment '{name}'",
)
try:
created = client.create_deployment(
name=name,
deployment_type=deployment_type,
source=source,
config_path=config_rel,
secrets=secrets,
agent=agent,
)
except HostBackendError as err:
if agent is not None and err.status_code == 409:
raise HostBackendError(
"This agent already has a deployment in this environment.",
status_code=409,
) from None
raise
created_id = created.get("id") if isinstance(created, dict) else None
if not isinstance(created_id, str) or not created_id:
raise HostBackendError(
"POST /v2/deployments succeeded but response missing a valid 'id'"
)
if agent is not None:
_get_emitter().info(f"Deployment name: {created['name']}")
_get_emitter().info(f"Deployment ID: {created_id}", deployment_id=created_id)
return created_id, step + 1
@@ -669,12 +706,12 @@ def _smith_dashboard_base_url(host_url: str | None) -> str:
if hostname in ("localhost", "127.0.0.1"):
return host_url.rstrip("/")
api_host_suffix = "api.host.langchain.com"
if hostname == api_host_suffix:
return "https://smith.langchain.com"
if hostname.endswith(f".{api_host_suffix}"):
prefix = hostname[: -(len(api_host_suffix) + 1)]
return f"https://{prefix}.smith.langchain.com"
for api_host_suffix in ("api.host.langchain.com", "api.smith.langchain.com"):
if hostname == api_host_suffix:
return "https://smith.langchain.com"
if hostname.endswith(f".{api_host_suffix}"):
prefix = hostname[: -(len(api_host_suffix) + 1)]
return f"https://{prefix}.smith.langchain.com"
return "https://smith.langchain.com"
@@ -1344,6 +1381,16 @@ OPT_HOST_URL = click.option(
hidden=True,
)
OPT_AGENT_ID = click.option(
"--agent-id", help="Logical agent ID (requires agent mode enabled for the tenant)."
)
OPT_AGENT_ENVIRONMENT = click.option(
"--environment",
type=click.Choice(["development", "staging", "production"]),
help="Agent environment (requires agent mode enabled for the tenant).",
)
OPT_VERBOSE = click.option(
"--verbose",
is_flag=True,
@@ -1442,6 +1489,8 @@ def _deploy_base_options(
decorators = [
OPT_HOST_API_KEY,
OPT_HOST_DEPLOYMENT_NAME,
OPT_AGENT_ID,
OPT_AGENT_ENVIRONMENT,
click.option(
"--deployment-id",
help=(
@@ -1578,6 +1627,8 @@ def _deploy_cmd(
deployment_id: str | None,
deployment_type: str,
name: str | None,
agent_id: str | None,
environment: str | None,
image_name: str | None,
image: str | None,
tag: str,
@@ -1603,6 +1654,17 @@ def _deploy_cmd(
# -- 1. Preflight --
validate_deploy_commands(install_command, build_command)
agent = None
if agent_id is not None or environment is not None:
if not agent_id or not agent_id.strip() or not environment:
raise click.UsageError(
"--agent-id and --environment are required together."
)
if name is not None or deployment_id is not None:
raise click.UsageError(
"--agent-id and --environment cannot be combined with --name or --deployment-id."
)
agent = {"agent_id": agent_id, "environment": environment}
if not config.exists():
message = (
"We couldn't find a langgraph.json file. Run `langgraph deploy` from "
@@ -1618,9 +1680,9 @@ def _deploy_cmd(
env_vars = _parse_env_from_config(config_json, config)
if not deployment_id and not name:
if not agent and not deployment_id and not name:
name = env_vars.get(_DEPLOYMENT_NAME_ENV)
if not deployment_id and not name:
if not agent and not deployment_id and not name:
default_name = normalize_name(pathlib.Path.cwd().name)
if no_input:
name = default_name
@@ -1661,6 +1723,7 @@ def _deploy_cmd(
if use_remote_build
else "No deployment found. Will create after build."
),
agent=agent,
)
if needs_creation:
@@ -1671,6 +1734,7 @@ def _deploy_cmd(
deployment_type=deployment_type,
source="internal_source" if use_remote_build else "internal_docker",
secrets=secrets,
agent=agent,
)
if not deployment_id:
@@ -1784,17 +1848,32 @@ def _deploy_cmd(
@OPT_HOST_API_KEY
@OPT_HOST_URL
@OPT_AGENT_ID
@OPT_AGENT_ENVIRONMENT
@click.option(
"--name-contains",
default="",
help="Only show deployments whose names contain this value.",
)
@deploy.command("list", help="[Beta] List LangSmith Deployments.")
def deploy_list(api_key: str | None, host_url: str | None, name_contains: str) -> None:
def deploy_list(
api_key: str | None,
host_url: str | None,
name_contains: str,
agent_id: str | None,
environment: str | None,
) -> None:
if agent_id is not None and not agent_id.strip():
raise click.UsageError("--agent-id must not be empty.")
filters = {}
if agent_id is not None:
filters["agent_id"] = agent_id
if environment is not None:
filters["agent_environment"] = environment
client = _create_host_backend_client(host_url, api_key)
response = _call_host_backend_with_optional_tenant(
client,
lambda c: c.list_deployments(name_contains=name_contains),
lambda c: c.list_deployments(name_contains=name_contains, **filters),
)
resources = response.get("resources") if isinstance(response, dict) else None
deployments = (
libs/cli/langgraph_cli/host_backend.py
+19 -4
View File
@@ -72,30 +72,45 @@ class HostBackendClient:
def create_deployment(
self,
name: str,
name: str | None,
deployment_type: str,
source: str,
config_path: str | None = None,
secrets: list[dict[str, str]] | None = None,
agent: 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": {},
}
if agent is not None:
payload["agent"] = agent
else:
payload["name"] = name
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)
def list_deployments(self, name_contains: str = "") -> dict[str, Any]:
def list_deployments(
self,
name_contains: str = "",
*,
agent_id: str | None = None,
agent_environment: str | None = None,
) -> dict[str, Any]:
params = {"name_contains": name_contains}
if agent_id is not None:
params["agent_id"] = agent_id
if agent_environment is not None:
params["agent_environment"] = agent_environment
return self._request(
"GET",
"/v2/deployments",
params={"name_contains": name_contains},
params=params,
)
def get_deployment(self, deployment_id: str) -> dict[str, Any]:
libs/cli/tests/unit_tests/test_deploy_agents.py
+95
View File
@@ -0,0 +1,95 @@
import json
from unittest.mock import Mock
import httpx
import pytest
from click.testing import CliRunner
import langgraph_cli.deploy as deploy
from langgraph_cli.cli import cli
from langgraph_cli.host_backend import HostBackendClient
@pytest.fixture
def deployment_api(monkeypatch, tmp_path):
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("LANGSMITH_DEPLOYMENT_NAME", raising=False)
monkeypatch.setattr(deploy, "_emitter", None)
monkeypatch.setattr(deploy, "_no_input", False)
(tmp_path / "langgraph.json").write_text(
json.dumps({"dependencies": ["."], "graphs": {"agent": "./agent.py:graph"}})
)
(tmp_path / ".env").write_text("LANGSMITH_DEPLOYMENT_NAME=legacy\n")
requests = []
state = {"enabled": True, "resources": []}
def handler(request):
requests.append(request)
assert request.url.path == "/v2/deployments"
if request.method == "GET":
if not state["enabled"] and (
"agent_id" in request.url.params
or "agent_environment" in request.url.params
):
return httpx.Response(
400, text="Agent filters are not available for this tenant."
)
return httpx.Response(200, json={"resources": state["resources"]})
assert request.method == "POST"
return httpx.Response(200, json={"id": "runtime-id", "name": "server-name"})
client = HostBackendClient("https://api.example.com", "test-key")
client._client.close()
client._client = httpx.Client(
base_url="https://api.example.com",
transport=httpx.MockTransport(handler),
headers={"X-Api-Key": "test-key"},
)
monkeypatch.setattr(deploy, "_create_host_backend_client", lambda *a, **kw: client)
monkeypatch.setattr(deploy, "find_tracked_packages", lambda *a: [])
remote_build = Mock(return_value=deploy.BuildResult())
monkeypatch.setattr(deploy, "_run_remote_build", remote_build)
monkeypatch.setattr(deploy, "_resolve_build_mode", lambda flag, **kw: (flag, None))
yield state, requests, remote_build
client._client.close()
AGENT_ARGS = [
"deploy",
"--agent-id",
"customer-support",
"--environment",
"staging",
"--remote",
"--no-wait",
"--no-input",
]
def test_agent_create(deployment_api, tmp_path):
_, requests, build = deployment_api
result = CliRunner().invoke(cli, AGENT_ARGS)
assert result.exit_code == 0, result.output
assert dict(requests[0].url.params) == {
"name_contains": "",
"agent_id": "customer-support",
"agent_environment": "staging",
}
payload = json.loads(requests[1].content)
assert payload["agent"] == {
"agent_id": "customer-support",
"environment": "staging",
}
assert "name" not in payload
assert build.call_args.kwargs["deployment_id"] == "runtime-id"
assert "server-name" in result.output
assert (tmp_path / ".env").read_text() == "LANGSMITH_DEPLOYMENT_NAME=legacy\n"
def test_agent_update(deployment_api):
state, requests, build = deployment_api
state["resources"] = [{"id": "existing-id", "is_preview": False}]
result = CliRunner().invoke(cli, AGENT_ARGS)
assert result.exit_code == 0, result.output
assert len(requests) == 1
assert build.call_args.kwargs["deployment_id"] == "existing-id"
libs/cli/tests/unit_tests/test_deploy_helpers.py
+6
View File
@@ -541,6 +541,12 @@ class TestCreateHostBackendClientNoInput:
class TestSmithDashboardBaseUrl:
def test_dev_smith_api_host_url(self):
assert (
_smith_dashboard_base_url("https://dev.api.smith.langchain.com/")
== "https://dev.smith.langchain.com"
)
def test_none_returns_default(self):
assert _smith_dashboard_base_url(None) == "https://smith.langchain.com"