tessl/pypi-langgraph-sdk

Python SDK for interacting with the LangGraph Platform REST API to build and manage AI assistants and conversational workflows

Overview

Eval results

Files

Client Management

Name: tessl/pypi-langgraph-sdk
Author: tessl

Core client creation and HTTP operations for connecting to LangGraph servers with automatic discovery, custom authentication, and connection management.

Capabilities

Client Factory Functions

Create async and sync clients for interacting with LangGraph servers, with automatic local server discovery and configurable connection settings.

from collections.abc import Mapping
from typing import Union, Optional
import httpx

# Type aliases
TimeoutTypes = Union[
    None,
    float,
    tuple[Optional[float], Optional[float]],
    tuple[Optional[float], Optional[float], Optional[float], Optional[float]],
    httpx.Timeout,
]

def get_client(
    *,
    url: str | None = None,
    api_key: str | None = None,
    headers: Mapping[str, str] | None = None,
    timeout: TimeoutTypes | None = None,
) -> LangGraphClient:
    """
    Create an async LangGraph client.

    Args:
        url: Base URL of the LangGraph API.
            – If `None`, the client first attempts an in-process connection via ASGI transport.
              If that fails, it falls back to `http://localhost:8123`.
        api_key: API key for authentication. If omitted, the client reads from environment
            variables in the following order:
              1. Function argument
              2. `LANGGRAPH_API_KEY`
              3. `LANGSMITH_API_KEY`
              4. `LANGCHAIN_API_KEY`
        headers: Additional HTTP headers to include in requests. Merged with authentication headers.
        timeout: HTTP timeout configuration. May be:
              – `httpx.Timeout` instance
              – float (total seconds)
              – tuple of timeouts (connect, read, write, pool)

    Returns:
        LangGraphClient: The top-level asynchronous client for accessing AssistantsClient,
        ThreadsClient, RunsClient, and CronClient.
    """

def get_sync_client(
    *,
    url: str | None = None,
    api_key: str | None = None,
    headers: Mapping[str, str] | None = None,
    timeout: TimeoutTypes | None = None,
) -> SyncLangGraphClient:
    """
    Create a sync LangGraph client.

    Args:
        url: The URL of the LangGraph API.
        api_key: The API key. If not provided, it will be read from the environment.
            Precedence:
                1. explicit argument
                2. LANGGRAPH_API_KEY
                3. LANGSMITH_API_KEY
                4. LANGCHAIN_API_KEY
        headers: Optional custom headers
        timeout: Optional timeout configuration for the HTTP client.
            Accepts an httpx.Timeout instance, a float (seconds), or a tuple of timeouts.
            Tuple format is (connect, read, write, pool)

    Returns:
        SyncLangGraphClient: The top-level synchronous client for accessing AssistantsClient,
        ThreadsClient, RunsClient, and CronClient.
    """

Main Client Classes

Primary client interfaces providing access to all LangGraph Platform resources through specialized resource clients.

class LangGraphClient:
    """
    Async client for LangGraph Platform API.

    Attributes:
    - assistants: AssistantsClient for managing AI assistants
    - threads: ThreadsClient for managing conversation threads
    - runs: RunsClient for executing workflows
    - crons: CronClient for scheduled tasks
    - store: StoreClient for persistent storage
    """
    assistants: AssistantsClient
    threads: ThreadsClient
    runs: RunsClient
    crons: CronClient
    store: StoreClient

    async def aclose(self) -> None:
        """Close the HTTP client and cleanup resources."""

class SyncLangGraphClient:
    """
    Sync client for LangGraph Platform API.

    Attributes:
    - assistants: SyncAssistantsClient for managing AI assistants
    - threads: SyncThreadsClient for managing conversation threads
    - runs: SyncRunsClient for executing workflows
    - crons: SyncCronClient for scheduled tasks
    - store: SyncStoreClient for persistent storage
    """
    assistants: SyncAssistantsClient
    threads: SyncThreadsClient
    runs: SyncRunsClient
    crons: SyncCronClient
    store: SyncStoreClient

    def close(self) -> None:
        """Close the HTTP client and cleanup resources."""

HTTP Client Operations

Low-level HTTP operations for direct API access when needed, supporting all standard HTTP methods with streaming capabilities.

from collections.abc import AsyncIterator, Iterator, Mapping, Callable
from typing import Any
from langgraph_sdk.schema import QueryParamTypes, StreamPart
import httpx

class HttpClient:
    """Low-level HTTP client for direct API access."""

    async def get(
        self,
        path: str,
        *,
        params: QueryParamTypes | None = None,
        headers: Mapping[str, str] | None = None,
        on_response: Callable[[httpx.Response], None] | None = None,
    ) -> Any:
        """Send a GET request."""

    async def post(
        self,
        path: str,
        *,
        json: dict[str, Any] | list | None,
        params: QueryParamTypes | None = None,
        headers: Mapping[str, str] | None = None,
        on_response: Callable[[httpx.Response], None] | None = None,
    ) -> Any:
        """Send a POST request."""

    async def put(
        self,
        path: str,
        *,
        json: dict,
        params: QueryParamTypes | None = None,
        headers: Mapping[str, str] | None = None,
        on_response: Callable[[httpx.Response], None] | None = None,
    ) -> Any:
        """Send a PUT request."""

    async def patch(
        self,
        path: str,
        *,
        json: dict,
        params: QueryParamTypes | None = None,
        headers: Mapping[str, str] | None = None,
        on_response: Callable[[httpx.Response], None] | None = None,
    ) -> Any:
        """Send a PATCH request."""

    async def delete(
        self,
        path: str,
        *,
        json: Any | None = None,
        params: QueryParamTypes | None = None,
        headers: Mapping[str, str] | None = None,
        on_response: Callable[[httpx.Response], None] | None = None,
    ) -> None:
        """Send a DELETE request."""

    async def stream(
        self,
        path: str,
        method: str,
        *,
        json: dict[str, Any] | None = None,
        params: QueryParamTypes | None = None,
        headers: Mapping[str, str] | None = None,
        on_response: Callable[[httpx.Response], None] | None = None,
    ) -> AsyncIterator[StreamPart]:
        """Stream results using SSE."""

class SyncHttpClient:
    """Sync HTTP client with identical methods but without async/await."""

    def get(
        self,
        path: str,
        *,
        params: QueryParamTypes | None = None,
        headers: Mapping[str, str] | None = None,
        on_response: Callable[[httpx.Response], None] | None = None,
    ) -> Any: ...

    def post(
        self,
        path: str,
        *,
        json: dict[str, Any] | list | None,
        params: QueryParamTypes | None = None,
        headers: Mapping[str, str] | None = None,
        on_response: Callable[[httpx.Response], None] | None = None,
    ) -> Any: ...

    def put(
        self,
        path: str,
        *,
        json: dict,
        params: QueryParamTypes | None = None,
        headers: Mapping[str, str] | None = None,
        on_response: Callable[[httpx.Response], None] | None = None,
    ) -> Any: ...

    def patch(
        self,
        path: str,
        *,
        json: dict,
        params: QueryParamTypes | None = None,
        headers: Mapping[str, str] | None = None,
        on_response: Callable[[httpx.Response], None] | None = None,
    ) -> Any: ...

    def delete(
        self,
        path: str,
        *,
        json: Any | None = None,
        params: QueryParamTypes | None = None,
        headers: Mapping[str, str] | None = None,
        on_response: Callable[[httpx.Response], None] | None = None,
    ) -> None: ...

    def stream(
        self,
        path: str,
        method: str,
        *,
        json: dict[str, Any] | None = None,
        params: QueryParamTypes | None = None,
        headers: Mapping[str, str] | None = None,
        on_response: Callable[[httpx.Response], None] | None = None,
    ) -> Iterator[StreamPart]: ...

Usage Examples

Basic Client Setup

from langgraph_sdk import get_client, get_sync_client

# Async client with auto-discovery
async_client = await get_client()

# Async client with specific server
async_client = await get_client(url="https://api.langgraph.com", api_key="your-api-key")

# Sync client
sync_client = get_sync_client()

# With custom headers
client = await get_client(
    headers={"User-Agent": "MyApp/1.0"},
    timeout=60.0
)

Resource Access

# Access different resource managers
assistants = await client.assistants.search()
threads = await client.threads.search()
runs = await client.runs.list(thread_id="thread-123")

# Cleanup when done
await client.aclose()

Direct HTTP Operations

# Low-level API access
response = await client.http.get("/assistants", params={"limit": 20})
new_thread = await client.http.post("/threads", json={"metadata": {"user": "john"}})

# Streaming
async for chunk in client.http.stream("/runs/stream", method="POST", json=run_data):
    print(chunk)

Install with Tessl CLI