codekingpro/portable-devtools

1"""Sync LangGraph client."""2 3from __future__ import annotations4 5from collections.abc import Mapping6from types import TracebackType7 8import httpx9 10from langgraph_sdk._shared.types import TimeoutTypes11from langgraph_sdk._shared.utilities import NOT_PROVIDED, _get_headers12from langgraph_sdk._sync.assistants import SyncAssistantsClient13from langgraph_sdk._sync.cron import SyncCronClient14from langgraph_sdk._sync.http import SyncHttpClient15from langgraph_sdk._sync.runs import SyncRunsClient16from langgraph_sdk._sync.store import SyncStoreClient17from langgraph_sdk._sync.threads import SyncThreadsClient18 19 20def get_sync_client(21    *,22    url: str | None = None,23    api_key: str | None = NOT_PROVIDED,24    headers: Mapping[str, str] | None = None,25    timeout: TimeoutTypes | None = None,26) -> SyncLangGraphClient:27    """Get a synchronous LangGraphClient instance.28 29    Args:30        url: The URL of the LangGraph API.31        api_key: API key for authentication. Can be:32            - A string: use this exact API key33            - `None`: explicitly skip loading from environment variables34            - Not provided (default): auto-load from environment in this order:35                1. `LANGGRAPH_API_KEY`36                2. `LANGSMITH_API_KEY`37                3. `LANGCHAIN_API_KEY`38        headers: Optional custom headers39        timeout: Optional timeout configuration for the HTTP client.40            Accepts an httpx.Timeout instance, a float (seconds), or a tuple of timeouts.41            Tuple format is (connect, read, write, pool)42            If not provided, defaults to connect=5s, read=300s, write=300s, and pool=5s.43    Returns:44        SyncLangGraphClient: The top-level synchronous client for accessing AssistantsClient,45        ThreadsClient, RunsClient, and CronClient.46 47    ???+ example "Example"48 49        ```python50        from langgraph_sdk import get_sync_client51 52        # get top-level synchronous LangGraphClient53        client = get_sync_client(url="http://localhost:8123")54 55        # example usage: client.<model>.<method_name>()56        assistant = client.assistants.get(assistant_id="some_uuid")57        ```58 59    ???+ example "Skip auto-loading API key from environment:"60 61        ```python62        from langgraph_sdk import get_sync_client63 64        # Don't load API key from environment variables65        client = get_sync_client(66            url="http://localhost:8123",67            api_key=None68        )69        ```70    """71 72    if url is None:73        url = "http://localhost:8123"74 75    transport = httpx.HTTPTransport(retries=5)76    client = httpx.Client(77        base_url=url,78        transport=transport,79        timeout=(80            httpx.Timeout(timeout)  # type: ignore[arg-type]81            if timeout is not None82            else httpx.Timeout(connect=5, read=300, write=300, pool=5)83        ),84        headers=_get_headers(api_key, headers),85    )86    return SyncLangGraphClient(client)87 88 89class SyncLangGraphClient:90    """Synchronous client for interacting with the LangGraph API.91 92    This class provides synchronous access to LangGraph API endpoints for managing93    assistants, threads, runs, cron jobs, and data storage.94 95    ???+ example "Example"96 97        ```python98        client = get_sync_client(url="http://localhost:2024")99        assistant = client.assistants.get("asst_123")100        ```101    """102 103    def __init__(self, client: httpx.Client) -> None:104        self.http = SyncHttpClient(client)105        self.assistants = SyncAssistantsClient(self.http)106        self.threads = SyncThreadsClient(self.http)107        self.runs = SyncRunsClient(self.http)108        self.crons = SyncCronClient(self.http)109        self.store = SyncStoreClient(self.http)110 111    def __enter__(self) -> SyncLangGraphClient:112        """Enter the sync context manager."""113        return self114 115    def __exit__(116        self,117        exc_type: type[BaseException] | None,118        exc_val: BaseException | None,119        exc_tb: TracebackType | None,120    ) -> None:121        """Exit the sync context manager."""122        self.close()123 124    def close(self) -> None:125        """Close the underlying HTTP client."""126        if hasattr(self, "http"):127            self.http.client.close()128