import os
import time
import requests
from dotenv import load_dotenv
load_dotenv()
# 必需的环境变量
CONTROL_PLANE_HOST = os.getenv("CONTROL_PLANE_HOST")
LANGSMITH_API_KEY = os.getenv("LANGSMITH_API_KEY")
WORKSPACE_ID = os.getenv("WORKSPACE_ID")
INTEGRATION_ID = os.getenv("INTEGRATION_ID")
MAX_WAIT_TIME = 1800 # 30 分钟
def get_headers() -> dict:
"""返回控制平面 API 请求的通用请求头。"""
return {
"X-Api-Key": LANGSMITH_API_KEY,
"X-Tenant-Id": WORKSPACE_ID,
}
def create_deployment() -> str:
"""创建部署。返回部署 ID。"""
headers = get_headers()
headers["Content-Type"] = "application/json"
deployment_name = "my_deployment"
request_body = {
"name": deployment_name,
"source": "github",
"source_config": {
"integration_id": INTEGRATION_ID,
"repo_url": "https://github.com/langchain-ai/langgraph-example",
"deployment_type": "dev",
"build_on_push": False,
"custom_url": None,
"resource_spec": None,
},
"source_revision_config": {
"repo_ref": "main",
"langgraph_config_path": "langgraph.json",
"image_uri": None,
},
"secrets": [
{
"name": "OPENAI_API_KEY",
"value": "test_openai_api_key",
},
{
"name": "ANTHROPIC_API_KEY",
"value": "test_anthropic_api_key",
},
{
"name": "TAVILY_API_KEY",
"value": "test_tavily_api_key",
},
],
}
response = requests.post(
url=f"{CONTROL_PLANE_HOST}/v2/deployments",
headers=headers,
json=request_body,
)
if response.status_code != 201:
raise Exception(f"创建部署失败: {response.text}")
deployment_id = response.json()["id"]
print(f"已创建部署 {deployment_name} ({deployment_id})")
return deployment_id
def get_deployment(deployment_id: str) -> dict:
"""获取部署信息。"""
response = requests.get(
url=f"{CONTROL_PLANE_HOST}/v2/deployments/{deployment_id}",
headers=get_headers(),
)
if response.status_code != 200:
raise Exception(f"获取部署 ID {deployment_id} 失败: {response.text}")
return response.json()
def list_revisions(deployment_id: str) -> list[dict]:
"""列出修订版本。
返回的列表按 created_at 降序排序（最新的在前）。
"""
response = requests.get(
url=f"{CONTROL_PLANE_HOST}/v2/deployments/{deployment_id}/revisions",
headers=get_headers(),
)
if response.status_code != 200:
raise Exception(
f"列出部署 ID {deployment_id} 的修订版本失败: {response.text}"
)
return response.json()
def get_revision(
deployment_id: str,
revision_id: str,
) -> dict:
"""获取修订版本信息。"""
response = requests.get(
url=f"{CONTROL_PLANE_HOST}/v2/deployments/{deployment_id}/revisions/{revision_id}",
headers=get_headers(),
)
if response.status_code != 200:
raise Exception(f"获取修订版本 ID {revision_id} 失败: {response.text}")
return response.json()
def patch_deployment(deployment_id: str) -> None:
"""更新部署。"""
headers = get_headers()
headers["Content-Type"] = "application/json"
# 这会创建一个新的修订版本，因为包含了 source_revision_config
response = requests.patch(
url=f"{CONTROL_PLANE_HOST}/v2/deployments/{deployment_id}",
headers=headers,
json={
"source_config": {
"build_on_push": True,
},
"source_revision_config": {
"repo_ref": "main",
"langgraph_config_path": "langgraph.json",
},
},
)
if response.status_code != 200:
raise Exception(f"更新部署失败: {response.text}")
print(f"已更新部署 ID {deployment_id}")
def wait_for_deployment(deployment_id: str, revision_id: str) -> None:
"""等待修订版本状态变为 DEPLOYED。"""
start_time = time.time()
revision, status = None, None
while time.time() - start_time < MAX_WAIT_TIME:
revision = get_revision(deployment_id, revision_id)
status = revision["status"]
if status == "DEPLOYED":
break
elif "FAILED" in status:
raise Exception(f"修订版本 ID {revision_id} 失败: {revision}")
print(f"等待修订版本 ID {revision_id} 变为 DEPLOYED...")
time.sleep(60)
if status != "DEPLOYED":
raise Exception(
f"等待修订版本 ID {revision_id} 变为 DEPLOYED 超时: {revision}"
)
def delete_deployment(deployment_id: str) -> None:
"""删除部署。"""
response = requests.delete(
url=f"{CONTROL_PLANE_HOST}/v2/deployments/{deployment_id}",
headers=get_headers(),
)
if response.status_code != 204:
raise Exception(
f"删除部署 ID {deployment_id} 失败: {response.text}"
)
print(f"部署 ID {deployment_id} 已删除")
if __name__ == "__main__":
# 创建部署并获取最新修订版本
deployment_id = create_deployment()
revisions = list_revisions(deployment_id)
latest_revision = revisions["resources"][0]
latest_revision_id = latest_revision["id"]
# 等待最新修订版本变为 DEPLOYED
wait_for_deployment(deployment_id, latest_revision_id)
# 更新部署并获取最新修订版本
patch_deployment(deployment_id)
revisions = list_revisions(deployment_id)
latest_revision = revisions["resources"][0]
latest_revision_id = latest_revision["id"]
# 等待最新修订版本变为 DEPLOYED
wait_for_deployment(deployment_id, latest_revision_id)
# 删除部署
delete_deployment(deployment_id)
