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

Assistant Management

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

Create, configure, and manage AI assistants based on registered graphs. Assistants serve as the execution engines for conversational workflows, with support for versioning, configuration, and metadata management.

Capabilities

Assistant CRUD Operations

Core operations for creating, reading, updating, and deleting assistants with comprehensive configuration and metadata support.

from collections.abc import Mapping
from langgraph_sdk.schema import (
    Assistant, AssistantSelectField, Config, Context, Json,
    OnConflictBehavior, QueryParamTypes
)

# Via client.assistants
async def get(
    assistant_id: str,
    *,
    headers: Mapping[str, str] | None = None,
    params: QueryParamTypes | None = None,
) -> Assistant:
    """
    Get an assistant by ID.

    Args:
        assistant_id: The ID of the assistant to get.
        headers: Optional custom headers to include with the request.
        params: Optional query parameters to include with the request.

    Returns:
        Assistant: Assistant Object.
    """

async def create(
    graph_id: str | None,
    config: Config | None = None,
    *,
    context: Context | None = None,
    metadata: Json = None,
    assistant_id: str | None = None,
    if_exists: OnConflictBehavior | None = None,
    name: str | None = None,
    headers: Mapping[str, str] | None = None,
    params: QueryParamTypes | None = None,
) -> Assistant:
    """
    Create a new assistant.

    Args:
        graph_id: The ID of the graph to create an assistant for.
        config: Configuration to merge with the assistant.
        context: Context to apply when running the assistant.
        metadata: Metadata to merge with the assistant.
        assistant_id: ID of the assistant. If None, ID will be a UUID.
        if_exists: How to handle duplicate creates. Defaults to "create".
        name: The name of the assistant. Defaults to Untitled.
        headers: Optional custom headers to include with the request.
        params: Optional query parameters to include with the request.

    Returns:
        Assistant: The created assistant.
    """

async def update(
    assistant_id: str,
    *,
    graph_id: str | None = None,
    config: Config | None = None,
    context: Context | None = None,
    metadata: Json = None,
    name: str | None = None,
    headers: Mapping[str, str] | None = None,
    description: str | None = None,
    params: QueryParamTypes | None = None,
) -> Assistant:
    """
    Update an assistant.

    Args:
        assistant_id: The ID of the assistant to update.
        graph_id: The ID of the graph to assign to the assistant.
        config: Configuration to merge with the assistant.
        context: Context to apply when running the assistant.
        metadata: Metadata to merge with the assistant.
        name: The name of the assistant.
        headers: Optional custom headers to include with the request.
        description: The description of the assistant.
        params: Optional query parameters to include with the request.

    Returns:
        Assistant: The updated assistant.
    """

async def delete(
    assistant_id: str,
    *,
    headers: Mapping[str, str] | None = None,
    params: QueryParamTypes | None = None,
) -> None:
    """
    Delete an assistant.

    Args:
        assistant_id: The assistant ID to delete.
        headers: Optional custom headers to include with the request.
        params: Optional query parameters to include with the request.
    """

Assistant Discovery & Search

Search and enumerate assistants with filtering, pagination, and sorting capabilities.

from langgraph_sdk.schema import (
    Assistant, AssistantSelectField, AssistantSortBy, SortOrder, Json, QueryParamTypes
)

async def search(
    *,
    metadata: Json = None,
    graph_id: str | None = None,
    limit: int = 10,
    offset: int = 0,
    sort_by: AssistantSortBy | None = None,
    sort_order: SortOrder | None = None,
    select: list[AssistantSelectField] | None = None,
    headers: Mapping[str, str] | None = None,
    params: QueryParamTypes | None = None,
) -> list[Assistant]:
    """
    List assistants from the store.

    Args:
        metadata: Metadata to filter by. Exact match filter for each KV pair.
        graph_id: ID of a graph to filter by.
        limit: Limit the number of assistants to return.
        offset: Offset to start from.
        sort_by: Field to sort by. Options are "created_at" and "updated_at".
        sort_order: Order to sort by. Options are "asc" and "desc".
        select: Fields to include in the response.
        headers: Optional custom headers to include with the request.
        params: Optional query parameters to include with the request.

    Returns:
        list[Assistant]: List of assistants.
    """

async def count(
    *,
    metadata: Json = None,
    graph_id: str | None = None,
    headers: Mapping[str, str] | None = None,
    params: QueryParamTypes | None = None,
) -> int:
    """
    Count assistants.

    Args:
        metadata: Metadata to filter by. Exact match filter for each KV pair.
        graph_id: ID of a graph to filter by.
        headers: Optional custom headers to include with the request.
        params: Optional query parameters to include with the request.

    Returns:
        int: Number of assistants that match the query.
    """

Graph Schema & Structure

Retrieve graph definitions, schemas, and subgraph structures associated with assistants.

from typing import Any
from langgraph_sdk.schema import Subgraphs, QueryParamTypes

async def get_graph(
    assistant_id: str,
    *,
    xray: int | bool = False,
    headers: Mapping[str, str] | None = None,
    params: QueryParamTypes | None = None,
) -> dict[str, list[dict[str, Any]]]:
    """
    Get the graph of an assistant by ID.

    Args:
        assistant_id: The ID of the assistant to get the graph of.
        xray: Include graph representation of subgraphs. If an integer value is provided,
              only subgraphs with a depth less than or equal to the value will be included.
        headers: Optional custom headers to include with the request.
        params: Optional query parameters to include with the request.

    Returns:
        Graph: The graph information for the assistant in JSON format.
    """

async def get_schemas(
    assistant_id: str,
    *,
    headers: Mapping[str, str] | None = None,
    params: QueryParamTypes | None = None,
) -> dict[str, Any]:
    """
    Get the schemas of an assistant by ID.

    Args:
        assistant_id: The ID of the assistant to get the schema of.
        headers: Optional custom headers to include with the request.
        params: Optional query parameters to include with the request.

    Returns:
        Graph: The graph schema for the assistant in JSON format.
    """

async def get_subgraphs(
    assistant_id: str,
    namespace: str | None = None,
    recurse: bool = False,
    *,
    headers: Mapping[str, str] | None = None,
    params: QueryParamTypes | None = None,
) -> Subgraphs:
    """
    Get the subgraphs of an assistant by ID.

    Args:
        assistant_id: The ID of the assistant to get the subgraphs of.
        namespace: Optional namespace to filter by.
        recurse: Whether to recursively get subgraphs.
        headers: Optional custom headers to include with the request.
        params: Optional query parameters to include with the request.

    Returns:
        Subgraphs: A mapping of subgraph names to their definitions.
    """

Version Management

Manage assistant versions and control which version is active for execution.

from langgraph_sdk.schema import AssistantVersion, Json, QueryParamTypes

async def get_versions(
    assistant_id: str,
    metadata: Json = None,
    limit: int = 10,
    offset: int = 0,
    *,
    headers: Mapping[str, str] | None = None,
    params: QueryParamTypes | None = None,
) -> list[AssistantVersion]:
    """
    List all versions of an assistant.

    Args:
        assistant_id: The assistant ID to get versions for.
        metadata: Metadata to filter versions by. Exact match filter for each KV pair.
        limit: The maximum number of versions to return.
        offset: The number of versions to skip.
        headers: Optional custom headers to include with the request.
        params: Optional query parameters to include with the request.

    Returns:
        list[AssistantVersion]: A list of assistant versions.
    """

async def set_latest(
    assistant_id: str,
    version: int,
    *,
    headers: Mapping[str, str] | None = None,
    params: QueryParamTypes | None = None,
) -> AssistantVersion:
    """
    Set the latest version of an assistant.

    Args:
        assistant_id: The assistant ID to set the version for.
        version: The version to set as the latest.
        headers: Optional custom headers to include with the request.
        params: Optional query parameters to include with the request.

    Returns:
        AssistantVersion: The updated assistant version.
    """

Types

class Assistant(TypedDict):
    """Assistant definition and configuration."""
    assistant_id: str
    graph_id: str
    config: Config
    created_at: str
    updated_at: str
    metadata: dict
    version: int

class AssistantVersion(TypedDict):
    """Versioned assistant information."""
    assistant_id: str
    version: int
    config: Config
    created_at: str
    metadata: dict

class GraphSchema(TypedDict):
    """Graph structure definition."""
    graph_id: str
    input_schema: dict
    output_schema: dict
    state_schema: dict
    config_schema: dict

AssistantSelectField = Literal[
    "assistant_id", "graph_id", "config", "created_at",
    "updated_at", "metadata", "version"
]

AssistantSortBy = Literal["created_at", "updated_at", "assistant_id"]

IfNotExists = Literal["create", "reject"]

Subgraphs = dict[str, GraphSchema]

Usage Examples

Creating and Managing Assistants

# Create an assistant from a registered graph
assistant = await client.assistants.create(
    graph_id="my-chatbot-graph",
    config={"max_turns": 10, "temperature": 0.7},
    metadata={"environment": "production", "owner": "team-ai"}
)

# Get assistant details
assistant = await client.assistants.get("assistant-123")

# Update assistant configuration
updated = await client.assistants.update(
    "assistant-123",
    config={"max_turns": 20},
    metadata={"last_updated_by": "admin"}
)

Searching and Discovery

# Search assistants by metadata
production_assistants = await client.assistants.search(
    metadata={"environment": "production"},
    limit=50
)

# Get assistants for a specific graph
graph_assistants = await client.assistants.search(
    graph_id="my-graph",
    sort_by="updated_at",
    order_by="desc"
)

# Count total assistants
total = await client.assistants.count()

Graph Inspection

# Get graph structure
graph = await client.assistants.get_graph("assistant-123")
print(f"Graph has {len(graph['nodes'])} nodes")

# Get input/output schemas
schemas = await client.assistants.get_schemas("assistant-123")
input_schema = schemas["input_schema"]

# Get subgraphs
subgraphs = await client.assistants.get_subgraphs("assistant-123")

Version Control

# Get all versions of an assistant
versions = await client.assistants.get_versions("assistant-123")

# Set specific version as active
await client.assistants.set_latest("assistant-123", version=5)

Install with Tessl CLI