codekingpro/portable-devtools

1"""Async LangGraph client."""2 3from __future__ import annotations4 5import logging6import os7from collections.abc import Mapping8from types import TracebackType9 10import httpx11 12from langgraph_sdk._async.assistants import AssistantsClient13from langgraph_sdk._async.cron import CronClient14from langgraph_sdk._async.http import HttpClient15from langgraph_sdk._async.runs import RunsClient16from langgraph_sdk._async.store import StoreClient17from langgraph_sdk._async.threads import ThreadsClient18from langgraph_sdk._shared.types import TimeoutTypes19from langgraph_sdk._shared.utilities import (20    NOT_PROVIDED,21    _get_headers,22    _registered_transports,23    get_asgi_transport,24)25 26logger = logging.getLogger(__name__)27 28 29def get_client(30    *,31    url: str | None = None,32    api_key: str | None = NOT_PROVIDED,33    headers: Mapping[str, str] | None = None,34    timeout: TimeoutTypes | None = None,35) -> LangGraphClient:36    """Create and configure a LangGraphClient.37 38    The client provides programmatic access to LangSmith Deployment. It supports39    both remote servers and local in-process connections (when running inside a LangGraph server).40 41    Args:42        url:43            Base URL of the LangGraph API.44            - If `None`, the client first attempts an in-process connection via ASGI transport.45              If that fails, it defers registration until after app initialization. This46              only works if the client is used from within the Agent server.47        api_key:48            API key for authentication. Can be:49              - A string: use this exact API key50              - `None`: explicitly skip loading from environment variables51              - Not provided (default): auto-load from environment in this order:52                1. `LANGGRAPH_API_KEY`53                2. `LANGSMITH_API_KEY`54                3. `LANGCHAIN_API_KEY`55        headers:56            Additional HTTP headers to include in requests. Merged with authentication headers.57        timeout:58            HTTP timeout configuration. May be:59              - `httpx.Timeout` instance60              - float (total seconds)61              - tuple `(connect, read, write, pool)` in seconds62            Defaults: connect=5, read=300, write=300, pool=5.63 64    Returns:65        LangGraphClient:66            A top-level client exposing sub-clients for assistants, threads,67            runs, and cron operations.68 69    ???+ example "Connect to a remote server:"70 71        ```python72        from langgraph_sdk import get_client73 74        # get top-level LangGraphClient75        client = get_client(url="http://localhost:8123")76 77        # example usage: client.<model>.<method_name>()78        assistants = await client.assistants.get(assistant_id="some_uuid")79        ```80 81    ???+ example "Connect in-process to a running LangGraph server:"82 83        ```python84        from langgraph_sdk import get_client85 86        client = get_client(url=None)87 88        async def my_node(...):89            subagent_result = await client.runs.wait(90                thread_id=None,91                assistant_id="agent",92                input={"messages": [{"role": "user", "content": "Foo"}]},93            )94        ```95 96    ???+ example "Skip auto-loading API key from environment:"97 98        ```python99        from langgraph_sdk import get_client100 101        # Don't load API key from environment variables102        client = get_client(103            url="http://localhost:8123",104            api_key=None105        )106        ```107    """108 109    transport: httpx.AsyncBaseTransport | None = None110    if url is None:111        url = "http://api"112        if os.environ.get("__LANGGRAPH_DEFER_LOOPBACK_TRANSPORT") == "true":113            transport = get_asgi_transport()(app=None, root_path="/noauth")  # type: ignore[invalid-argument-type]114            _registered_transports.append(transport)115        else:116            try:117                from langgraph_api.server import app  # type: ignore118 119                transport = get_asgi_transport()(app, root_path="/noauth")120            except Exception:121                logger.debug(122                    "Failed to connect to in-process LangGraph server. Deferring configuration.",123                    exc_info=True,124                )125                transport = get_asgi_transport()(app=None, root_path="/noauth")  # type: ignore[invalid-argument-type]126                _registered_transports.append(transport)127 128    if transport is None:129        transport = httpx.AsyncHTTPTransport(retries=5)130    client = httpx.AsyncClient(131        base_url=url,132        transport=transport,133        timeout=(134            httpx.Timeout(timeout)  # type: ignore[arg-type]135            if timeout is not None136            else httpx.Timeout(connect=5, read=300, write=300, pool=5)137        ),138        headers=_get_headers(api_key, headers),139    )140    return LangGraphClient(client)141 142 143class LangGraphClient:144    """Top-level client for LangGraph API.145 146    Attributes:147        assistants: Manages versioned configuration for your graphs.148        threads: Handles (potentially) multi-turn interactions, such as conversational threads.149        runs: Controls individual invocations of the graph.150        crons: Manages scheduled operations.151        store: Interfaces with persistent, shared data storage.152    """153 154    def __init__(self, client: httpx.AsyncClient) -> None:155        self.http = HttpClient(client)156        self.assistants = AssistantsClient(self.http)157        self.threads = ThreadsClient(self.http)158        self.runs = RunsClient(self.http)159        self.crons = CronClient(self.http)160        self.store = StoreClient(self.http)161 162    async def __aenter__(self) -> LangGraphClient:163        """Enter the async context manager."""164        return self165 166    async def __aexit__(167        self,168        exc_type: type[BaseException] | None,169        exc_val: BaseException | None,170        exc_tb: TracebackType | None,171    ) -> None:172        """Exit the async context manager."""173        await self.aclose()174 175    async def aclose(self) -> None:176        """Close the underlying HTTP client."""177        if hasattr(self, "http"):178            await self.http.client.aclose()179