LangSmith Deployment용 컨트롤 플레인 API 레퍼런스

컨트롤 플레인 API는 LangSmith Deployment의 일부예요. 이 API를 사용하면 Agent Server 배포를 프로그래매틱하게 만들고, 관리하고, 자동화할 수 있답니다. 예를 들어 커스텀 CI/CD 워크플로우의 일부로 활용할 수 있어요.

출처: 문서

본문

컨트롤 플레인 API는 LangSmith Deployment의 일부입니다. 컨트롤 플레인 API로 Agent Server 배포를 프로그래매틱하게 만들고, 관리하고, 자동화할 수 있습니다 — 예를 들어 커스텀 CI/CD 워크플로우의 일부로 말이죠.

전체 API 레퍼런스는 사이드바의 Control Plane API 섹션에서 찾아보거나, 엔드포인트 그룹을 참조하세요:

Integrations (v1): GitHub 통합 및 저장소 목록
Deployments (v2): Agent Server 배포 생성, 관리, 업데이트
Listeners (v2): 자체 호스팅 엔터프라이즈 조직을 위한 Listener 리소스
Auth Service (v2): OAuth 제공자 구성 및 인증 흐름

호스트

Cloud 데이터 지역의 컨트롤 플레인 호스트:

지역	URL
GCP US	`https://api.smith.langchain.com`
GCP EU	`https://eu.api.smith.langchain.com`
GCP APAC	`https://apac.api.smith.langchain.com`
AWS US	`https://aws.api.smith.langchain.com`

참고: LangSmith의 자체 호스팅 배포는 컨트롤 플레인에 대한 커스텀 호스트를 갖습니다. 컨트롤 플레인 API는 /api-host 경로에서 접근할 수 있습니다. 예를 들어 http(s)://<host>/api-host/v2/deployments입니다. 자세한 내용은 자체 호스팅 사용 가이드를 참고하세요.

인증

컨트롤 플레인 API로 인증하려면 X-Api-Key 헤더를 유효한 LangSmith API 키로 설정하고, X-Tenant-Id 헤더를 대상으로 할 유효한 워크스페이스 ID로 설정하세요.

예시 curl 명령:

curl --request GET \
  --url http://localhost:8124/v2/deployments \
  --header 'X-Api-Key: LANGSM...KEY'
  --header 'X-Tenant-Id': WORKSPACE_ID'

버전 관리

각 엔드포인트 경로에는 버전이 접두사로 붙습니다 (예: v1, v2).

빠른 시작

POST /v2/deployments를 호출해 새 Deployment를 만듭니다. 응답 본문에는 Deployment ID(id)와 최신(그리고 첫 번째) 리비전의 ID(latest_revision_id)가 포함됩니다.
GET /v2/deployments/{deployment_id}를 호출해 Deployment를 가져옵니다. URL의 deployment_id를 Deployment ID(id) 값으로 설정합니다.
GET /v2/deployments/{deployment_id}/revisions/{latest_revision_id}를 호출해 리비전 status가 DEPLOYED가 될 때까지 폴링합니다.
PATCH /v2/deployments/{deployment_id}를 호출해 배포를 업데이트합니다.

예시 코드

다음은 컨트롤 플레인 API를 오케스트레이션해 배포를 만들고, 업데이트하고, 삭제하는 방법을 보여주는 예시 Python 코드입니다.

import os
import time

import requests
from dotenv import load_dotenv


load_dotenv()

# required environment variables
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 mins


def get_headers() -> dict:
    """Return common headers for requests to the control plane API."""
    return {
        "X-Api-Key": LANGSMITH_API_KEY,
        "X-Tenant-Id": WORKSPACE_ID,
    }


def create_deployment() -> str:
    """Create deployment. Return deployment 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": "serverless",
            "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"Failed to create deployment: {response.text}")

    deployment_id = response.json()["id"]
    print(f"Created deployment {deployment_name} ({deployment_id})")
    return deployment_id


def get_deployment(deployment_id: str) -> dict:
    """Get deployment."""
    response = requests.get(
        url=f"{CONTROL_PLANE_HOST}/v2/deployments/{deployment_id}",
        headers=get_headers(),
    )

    if response.status_code != 200:
        raise Exception(f"Failed to get deployment ID {deployment_id}: {response.text}")

    return response.json()


def list_revisions(deployment_id: str) -> list[dict]:
    """List revisions.

    Return list is sorted by created_at in descending order (latest first).
    """
    response = requests.get(
        url=f"{CONTROL_PLANE_HOST}/v2/deployments/{deployment_id}/revisions",
        headers=get_headers(),
    )

    if response.status_code != 200:
        raise Exception(
            f"Failed to list revisions for deployment ID {deployment_id}: {response.text}"
        )

    return response.json()


def get_revision(
    deployment_id: str,
    revision_id: str,
) -> dict:
    """Get revision."""
    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"Failed to get revision ID {revision_id}: {response.text}")

    return response.json()


def patch_deployment(deployment_id: str) -> None:
    """Patch deployment."""
    headers = get_headers()
    headers["Content-Type"] = "application/json"

    # This creates a new revision because source_revision_config is included
    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"Failed to patch deployment: {response.text}")

    print(f"Patched deployment ID {deployment_id}")


def wait_for_deployment(deployment_id: str, revision_id: str) -> None:
    """Wait for revision status to be 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"Revision ID {revision_id} failed: {revision}")

        print(f"Waiting for revision ID {revision_id} to be DEPLOYED...")
        time.sleep(60)

    if status != "DEPLOYED":
        raise Exception(
            f"Timeout waiting for revision ID {revision_id} to be DEPLOYED: {revision}"
        )


def delete_deployment(deployment_id: str) -> None:
    """Delete deployment."""
    response = requests.delete(
        url=f"{CONTROL_PLANE_HOST}/v2/deployments/{deployment_id}",
        headers=get_headers(),
    )

    if response.status_code != 204:
        raise Exception(
            f"Failed to delete deployment ID {deployment_id}: {response.text}"
        )

    print(f"Deployment ID {deployment_id} deleted")


if __name__ == "__main__":
    # create deployment and get the latest revision
    deployment_id = create_deployment()
    revisions = list_revisions(deployment_id)
    latest_revision = revisions["resources"][0]
    latest_revision_id = latest_revision["id"]

    # wait for latest revision to be DEPLOYED
    wait_for_deployment(deployment_id, latest_revision_id)

    # patch the deployment and get the latest revision
    patch_deployment(deployment_id)
    revisions = list_revisions(deployment_id)
    latest_revision = revisions["resources"][0]
    latest_revision_id = latest_revision["id"]

    # wait for latest revision to be DEPLOYED
    wait_for_deployment(deployment_id, latest_revision_id)

    # delete the deployment
    delete_deployment(deployment_id)

더 알아보기

LangSmith Deployment의 전체 개념은 Deployment 문서를 참고하세요.
Agent Server에 대한 자세한 내용은 Agent Server 문서를 확인해 보세요.