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

Persistent Storage

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

Cross-thread persistent memory system for storing and retrieving documents, configuration, and application data with namespacing, search capabilities, and flexible data organization.

Capabilities

Item Storage Operations

Store, retrieve, update, and delete items in the persistent storage system with flexible key-value organization.

from collections.abc import Mapping, Sequence
from typing import Any, Literal
from langgraph_sdk.schema import Item, QueryParamTypes

# Via client.store
async def put_item(
    namespace: Sequence[str],
    /,
    key: str,
    value: Mapping[str, Any],
    index: Literal[False] | list[str] | None = None,
    ttl: int | None = None,
    headers: Mapping[str, str] | None = None,
    params: QueryParamTypes | None = None,
) -> None:
    """
    Store an item.

    Args:
        namespace: The namespace to store the item in.
        key: The unique identifier for the item.
        value: The value to store.
        index: Whether to index the item. If a list is provided, only the specified fields will be indexed.
        ttl: Time-to-live for the item in seconds.
        headers: Optional custom headers to include with the request.
        params: Optional query parameters to include with the request.
    """

async def get_item(
    namespace: Sequence[str],
    /,
    key: str,
    *,
    refresh_ttl: bool | None = None,
    headers: Mapping[str, str] | None = None,
    params: QueryParamTypes | None = None,
) -> Item:
    """
    Retrieve a single item.

    Args:
        namespace: The namespace to retrieve the item from.
        key: The unique identifier for the item.
        refresh_ttl: Whether to refresh the item's TTL.
        headers: Optional custom headers to include with the request.
        params: Optional query parameters to include with the request.

    Returns:
        Item: The retrieved item.
    """

async def delete_item(
    namespace: Sequence[str],
    /,
    key: str,
    headers: Mapping[str, str] | None = None,
    params: QueryParamTypes | None = None,
) -> None:
    """
    Delete an item.

    Args:
        namespace: The namespace to delete the item from.
        key: The unique identifier for the item.
        headers: Optional custom headers to include with the request.
        params: Optional query parameters to include with the request.
    """

Search & Discovery

Search for items across namespaces with text queries, filtering, and pagination.

from langgraph_sdk.schema import SearchItemsResponse, ListNamespaceResponse

async def search_items(
    namespace_prefix: Sequence[str],
    /,
    filter: Mapping[str, Any] | None = None,
    limit: int = 10,
    offset: int = 0,
    query: str | None = None,
    refresh_ttl: bool | None = None,
    headers: Mapping[str, str] | None = None,
    params: QueryParamTypes | None = None,
) -> SearchItemsResponse:
    """
    Search for items in the store.

    Args:
        namespace_prefix: The namespace prefix to search under.
        filter: Filtering criteria for the search.
        limit: Maximum number of items to return (default is 10).
        offset: Number of items to skip before returning results (default is 0).
        query: Optional query string for full-text search.
        refresh_ttl: Whether to refresh the TTL of found items.
        headers: Optional custom headers to include with the request.
        params: Optional query parameters to include with the request.

    Returns:
        SearchItemsResponse: The search results.
    """

async def list_namespaces(
    *,
    prefix: Sequence[str] | None = None,
    suffix: Sequence[str] | None = None,
    max_depth: int | None = None,
    limit: int = 100,
    offset: int = 0,
    headers: Mapping[str, str] | None = None,
    params: QueryParamTypes | None = None,
) -> ListNamespaceResponse:
    """
    List namespaces with optional match conditions.

    Args:
        prefix: Optional list of strings representing the prefix to filter namespaces.
        suffix: Optional list of strings representing the suffix to filter namespaces.
        max_depth: Optional integer specifying the maximum depth of namespaces to return.
        limit: Maximum number of namespaces to return (default is 100).
        offset: Number of namespaces to skip before returning results (default is 0).
        headers: Optional custom headers to include with the request.
        params: Optional query parameters to include with the request.

    Returns:
        ListNamespaceResponse: The list of namespaces.
    """

Types

class Item(TypedDict):
    """Stored item with metadata."""
    namespace: list[str]
    key: str
    value: dict
    created_at: str
    updated_at: str
    index: dict

class SearchItem(TypedDict):
    """Search result item with relevance score."""
    namespace: list[str]
    key: str
    value: dict
    created_at: str
    updated_at: str
    index: dict
    score: float

class SearchItemsResponse(TypedDict):
    """Search results with pagination."""
    items: list[SearchItem]
    total: int
    limit: int
    offset: int

class ListNamespaceResponse(TypedDict):
    """Namespace listing response."""
    namespaces: list[list[str]]
    total: int
    limit: int
    offset: int

Usage Examples

Basic Item Operations

# Store user preferences
user_prefs = {
    "theme": "dark",
    "language": "en",
    "notifications": True,
    "timezone": "America/New_York"
}

await client.store.put_item(
    value=user_prefs,
    namespace=["users", "user-123", "preferences"],
    key="ui_settings"
)

# Retrieve user preferences
prefs = await client.store.get_item(
    namespace=["users", "user-123", "preferences"],
    key="ui_settings"
)

print(f"User theme: {prefs['value']['theme']}")

# Update preferences (store overwrites)
updated_prefs = prefs["value"].copy()
updated_prefs["theme"] = "light"

await client.store.put_item(
    value=updated_prefs,
    namespace=["users", "user-123", "preferences"],
    key="ui_settings"
)

Document Storage

# Store conversation history
conversation = {
    "messages": [
        {"role": "human", "content": "Hello", "timestamp": "2023-12-01T10:00:00Z"},
        {"role": "assistant", "content": "Hi there!", "timestamp": "2023-12-01T10:00:01Z"}
    ],
    "summary": "Greeting exchange",
    "participants": ["user-123", "assistant-456"]
}

await client.store.put_item(
    value=conversation,
    namespace=["conversations", "user-123"],
    key="conv-2023-12-01",
    index={
        "participants": ["user-123", "assistant-456"],
        "date": "2023-12-01",
        "message_count": 2
    }
)

# Store configuration
app_config = {
    "api_endpoints": {
        "auth": "https://auth.api.com",
        "data": "https://data.api.com"
    },
    "features": {
        "analytics": True,
        "caching": True,
        "rate_limiting": {"rpm": 1000}
    },
    "version": "1.2.0"
}

await client.store.put_item(
    value=app_config,
    namespace=["application", "config"],
    key="production"
)

Hierarchical Organization

# Organize data hierarchically
namespaces = [
    # User data
    ["users", "user-123", "profile"],
    ["users", "user-123", "preferences"],
    ["users", "user-123", "sessions"],

    # Application data
    ["application", "config", "production"],
    ["application", "config", "staging"],
    ["application", "templates", "emails"],

    # Analytics data
    ["analytics", "2023", "12", "daily"],
    ["analytics", "2023", "12", "weekly"],
]

# Store items in different namespaces
for namespace in namespaces:
    await client.store.put_item(
        value={"created": "2023-12-01", "type": namespace[-1]},
        namespace=namespace,
        key="metadata"
    )

Search Operations

# Search user conversations
user_conversations = await client.store.search_items(
    namespace_prefix=["conversations", "user-123"],
    query="greeting",
    limit=20
)

print(f"Found {user_conversations['total']} conversations")
for item in user_conversations["items"]:
    print(f"  {item['key']}: {item['value']['summary']}")

# Search with filters
recent_conversations = await client.store.search_items(
    namespace_prefix=["conversations"],
    filter={"date": "2023-12-01", "message_count": {"$gt": 5}},
    limit=50
)

# Search application configs
configs = await client.store.search_items(
    namespace_prefix=["application", "config"],
    query="production OR staging"
)

Namespace Management

# List all user namespaces
user_namespaces = await client.store.list_namespaces(
    prefix=["users"],
    max_depth=3
)

print("User namespaces:")
for ns in user_namespaces["namespaces"]:
    print(f"  {'/'.join(ns)}")

# List application namespaces
app_namespaces = await client.store.list_namespaces(
    prefix=["application"]
)

# Get all top-level namespaces
all_namespaces = await client.store.list_namespaces(max_depth=1)

Session & Cache Management

# Store session data
session_data = {
    "user_id": "user-123",
    "login_time": "2023-12-01T10:00:00Z",
    "permissions": ["read", "write"],
    "temporary_data": {"cart_items": [], "form_state": {}}
}

await client.store.put_item(
    value=session_data,
    namespace=["sessions", "active"],
    key="session-abc123",
    index={"user_id": "user-123", "active": True}
)

# Cache frequently accessed data
cache_data = {
    "computed_result": [1, 2, 3, 4, 5],
    "computation_time": 0.5,
    "expires_at": "2023-12-01T11:00:00Z"
}

await client.store.put_item(
    value=cache_data,
    namespace=["cache", "computations"],
    key="fibonacci-100",
    index={"expires_at": "2023-12-01T11:00:00Z"}
)

Thread-Shared Memory

# Store shared state across multiple threads
shared_state = {
    "active_users": ["user-123", "user-456"],
    "global_config": {"maintenance_mode": False},
    "counters": {"api_calls": 1000, "errors": 5}
}

await client.store.put_item(
    value=shared_state,
    namespace=["global", "runtime"],
    key="current_state"
)

# Access from any thread
current_state = await client.store.get_item(
    namespace=["global", "runtime"],
    key="current_state"
)

# Update counters atomically (read-modify-write pattern)
state = current_state["value"]
state["counters"]["api_calls"] += 1

await client.store.put_item(
    value=state,
    namespace=["global", "runtime"],
    key="current_state"
)

Cleanup Operations

# Find and delete expired items
expired_items = await client.store.search_items(
    namespace_prefix=["cache"],
    filter={"expires_at": {"$lt": "2023-12-01T10:00:00Z"}}
)

for item in expired_items["items"]:
    await client.store.delete_item(
        namespace=item["namespace"],
        key=item["key"]
    )

# Delete user data
user_namespaces = await client.store.list_namespaces(
    prefix=["users", "user-123"]
)

# Delete all items in user namespaces (requires iteration)
for namespace in user_namespaces["namespaces"]:
    # Note: You would need to search and delete individual items
    # as there's no bulk delete operation shown in the API
    items = await client.store.search_items(
        namespace_prefix=namespace,
        limit=1000
    )

    for item in items["items"]:
        await client.store.delete_item(
            namespace=item["namespace"],
            key=item["key"]
        )

Install with Tessl CLI