diff --git a/client_backup.py b/client_backup.py new file mode 100644 index 000000000..00ace666c --- /dev/null +++ b/client_backup.py @@ -0,0 +1,5600 @@ +"""The LangGraph client implementations connect to the LangGraph API. + +This module provides both asynchronous ([get_client(url="http://localhost:2024"))](#get_client) or [LangGraphClient](#LangGraphClient)) +and synchronous ([get_sync_client(url="http://localhost:2024"))](#get_sync_client) or [SyncLanggraphClient](#SyncLanggraphClient)) +clients to interacting with the LangGraph API's core resources such as +Assistants, Threads, Runs, and Cron jobs, as well as its persistent +document Store. +""" # noqa: E501 + +from __future__ import annotations + +import asyncio +import functools +import logging +import os +import re +import sys +from collections.abc import AsyncIterator, Iterator, Sequence +from typing import ( + Any, + Callable, + Literal, + Optional, + Union, + overload, +) + +import httpx +import orjson +from httpx._types import QueryParamTypes + +import langgraph_sdk +from langgraph_sdk.schema import ( + All, + Assistant, + AssistantSortBy, + AssistantVersion, + CancelAction, + Checkpoint, + Command, + Config, + Cron, + CronSortBy, + DisconnectMode, + GraphSchema, + IfNotExists, + Item, + Json, + ListNamespaceResponse, + MultitaskStrategy, + OnCompletionBehavior, + OnConflictBehavior, + Run, + RunCreate, + RunCreateMetadata, + RunStatus, + SearchItemsResponse, + SortOrder, + StreamMode, + StreamPart, + Subgraphs, + Thread, + ThreadSortBy, + ThreadState, + ThreadStatus, + ThreadUpdateStateResponse, +) +from langgraph_sdk.sse import SSEDecoder, aiter_lines_raw, iter_lines_raw + +logger = logging.getLogger(__name__) + + +RESERVED_HEADERS = ("x-api-key",) + + +def _get_api_key(api_key: str | None = None) -> str | None: + """Get the API key from the environment. + Precedence: + 1. explicit argument + 2. LANGGRAPH_API_KEY + 3. LANGSMITH_API_KEY + 4. LANGCHAIN_API_KEY + """ + if api_key: + return api_key + for prefix in ["LANGGRAPH", "LANGSMITH", "LANGCHAIN"]: + if env := os.getenv(f"{prefix}_API_KEY"): + return env.strip().strip('"').strip("'") + return None # type: ignore + + +def _get_headers( + api_key: str | None, custom_headers: dict[str, str] | None +) -> dict[str, str]: + """Combine api_key and custom user-provided headers.""" + custom_headers = custom_headers or {} + for header in RESERVED_HEADERS: + if header in custom_headers: + raise ValueError(f"Cannot set reserved header '{header}'") + + headers = { + "User-Agent": f"langgraph-sdk-py/{langgraph_sdk.__version__}", + **custom_headers, + } + api_key = _get_api_key(api_key) + if api_key: + headers["x-api-key"] = api_key + + return headers + + +def _orjson_default(obj: Any) -> Any: + if hasattr(obj, "model_dump") and callable(obj.model_dump): + return obj.model_dump() + elif hasattr(obj, "dict") and callable(obj.dict): + return obj.dict() + elif isinstance(obj, (set, frozenset)): + return list(obj) + else: + raise TypeError(f"Object of type {type(obj)} is not JSON serializable") + + +# Compiled regex pattern for extracting run metadata from Content-Location header +_RUN_METADATA_PATTERN = re.compile( + r"(\/threads\/(?P.+))?\/runs\/(?P.+)" +) + + +def _get_run_metadata_from_response( + response: httpx.Response, +) -> RunCreateMetadata | None: + """Extract run metadata from the response headers.""" + if (content_location := response.headers.get("Content-Location")) and ( + match := _RUN_METADATA_PATTERN.search(content_location) + ): + return RunCreateMetadata( + run_id=match.group("run_id"), + thread_id=match.group("thread_id") or None, + ) + + return None + + +def get_client( + *, + url: str | None = None, + api_key: str | None = None, + headers: dict[str, str] | None = None, + timeout: TimeoutTypes | None = None, +) -> LangGraphClient: + """Get a LangGraphClient instance. + + 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) + If not provided, defaults to connect=5s, read=300s, write=300s, and pool=5s. + + Returns: + LangGraphClient: The top-level client for accessing AssistantsClient, + ThreadsClient, RunsClient, and CronClient. + + ???+ example "Example" + + ```python + from langgraph_sdk import get_client + + # get top-level LangGraphClient + client = get_client(url="http://localhost:8123") + + # example usage: client..() + assistants = await client.assistants.get(assistant_id="some_uuid") + ``` + """ + + transport: httpx.AsyncBaseTransport | None = None + if url is None: + if os.environ.get("__LANGGRAPH_DEFER_LOOPBACK_TRANSPORT") == "true": + transport = get_asgi_transport()(app=None, root_path="/noauth") + _registered_transports.append(transport) + url = "http://api" + else: + try: + from langgraph_api.server import app # type: ignore + + url = "http://api" + + transport = get_asgi_transport()(app, root_path="/noauth") + except Exception: + url = "http://localhost:8123" + + if transport is None: + transport = httpx.AsyncHTTPTransport(retries=5) + client = httpx.AsyncClient( + base_url=url, + transport=transport, + timeout=( + httpx.Timeout(timeout) + if timeout is not None + else httpx.Timeout(connect=5, read=300, write=300, pool=5) + ), + headers=_get_headers(api_key, headers), + ) + return LangGraphClient(client) + + +class LangGraphClient: + """Top-level client for LangGraph API. + + Attributes: + assistants: Manages versioned configuration for your graphs. + threads: Handles (potentially) multi-turn interactions, such as conversational threads. + runs: Controls individual invocations of the graph. + crons: Manages scheduled operations. + store: Interfaces with persistent, shared data storage. + """ + + def __init__(self, client: httpx.AsyncClient) -> None: + self.http = HttpClient(client) + self.assistants = AssistantsClient(self.http) + self.threads = ThreadsClient(self.http) + self.runs = RunsClient(self.http) + self.crons = CronClient(self.http) + self.store = StoreClient(self.http) + + +class HttpClient: + """Handle async requests to the LangGraph API. + + Adds additional error messaging & content handling above the + provided httpx client. + + Attributes: + client (httpx.AsyncClient): Underlying HTTPX async client. + """ + + def __init__(self, client: httpx.AsyncClient) -> None: + self.client = client + + async def get( + self, + path: str, + *, + params: QueryParamTypes | None = None, + headers: dict[str, str] | None = None, + on_response: Callable[[httpx.Response], None] | None = None, + ) -> Any: + """Send a GET request.""" + r = await self.client.get(path, params=params, headers=headers) + if on_response: + on_response(r) + try: + r.raise_for_status() + except httpx.HTTPStatusError as e: + body = (await r.aread()).decode() + if sys.version_info >= (3, 11): + e.add_note(body) + else: + logger.error(f"Error from langgraph-api: {body}", exc_info=e) + raise e + return await _adecode_json(r) + + async def post( + self, + path: str, + *, + json: dict | None, + headers: dict[str, str] | None = None, + on_response: Callable[[httpx.Response], None] | None = None, + ) -> Any: + """Send a POST request.""" + if json is not None: + request_headers, content = await _aencode_json(json) + else: + request_headers, content = {}, b"" + # Merge headers, with runtime headers taking precedence + if headers: + request_headers.update(headers) + r = await self.client.post(path, headers=request_headers, content=content) + if on_response: + on_response(r) + try: + r.raise_for_status() + except httpx.HTTPStatusError as e: + body = (await r.aread()).decode() + if sys.version_info >= (3, 11): + e.add_note(body) + else: + logger.error(f"Error from langgraph-api: {body}", exc_info=e) + raise e + return await _adecode_json(r) + + async def put( + self, + path: str, + *, + json: dict, + headers: dict[str, str] | None = None, + on_response: Callable[[httpx.Response], None] | None = None, + ) -> Any: + """Send a PUT request.""" + request_headers, content = await _aencode_json(json) + if headers: + request_headers.update(headers) + r = await self.client.put(path, headers=request_headers, content=content) + if on_response: + on_response(r) + try: + r.raise_for_status() + except httpx.HTTPStatusError as e: + body = (await r.aread()).decode() + if sys.version_info >= (3, 11): + e.add_note(body) + else: + logger.error(f"Error from langgraph-api: {body}", exc_info=e) + raise e + return await _adecode_json(r) + + async def patch( + self, + path: str, + *, + json: dict, + headers: dict[str, str] | None = None, + on_response: Callable[[httpx.Response], None] | None = None, + ) -> Any: + """Send a PATCH request.""" + request_headers, content = await _aencode_json(json) + if headers: + request_headers.update(headers) + r = await self.client.patch(path, headers=request_headers, content=content) + if on_response: + on_response(r) + try: + r.raise_for_status() + except httpx.HTTPStatusError as e: + body = (await r.aread()).decode() + if sys.version_info >= (3, 11): + e.add_note(body) + else: + logger.error(f"Error from langgraph-api: {body}", exc_info=e) + raise e + return await _adecode_json(r) + + async def delete( + self, + path: str, + *, + json: Any | None = None, + headers: dict[str, str] | None = None, + on_response: Callable[[httpx.Response], None] | None = None, + ) -> None: + """Send a DELETE request.""" + r = await self.client.request("DELETE", path, json=json, headers=headers) + if on_response: + on_response(r) + try: + r.raise_for_status() + except httpx.HTTPStatusError as e: + body = (await r.aread()).decode() + if sys.version_info >= (3, 11): + e.add_note(body) + else: + logger.error(f"Error from langgraph-api: {body}", exc_info=e) + raise e + + async def stream( + self, + path: str, + method: str, + *, + json: dict | None = None, + params: QueryParamTypes | None = None, + headers: dict[str, str] | None = None, + on_response: Callable[[httpx.Response], None] | None = None, + ) -> AsyncIterator[StreamPart]: + """Stream results using SSE.""" + request_headers, content = await _aencode_json(json) + request_headers["Accept"] = "text/event-stream" + request_headers["Cache-Control"] = "no-store" + # Add runtime headers with precedence + if headers: + request_headers.update(headers) + + async with self.client.stream( + method, path, headers=request_headers, content=content, params=params + ) as res: + if on_response: + on_response(res) + # check status + try: + res.raise_for_status() + except httpx.HTTPStatusError as e: + body = (await res.aread()).decode() + if sys.version_info >= (3, 11): + e.add_note(body) + else: + logger.error(f"Error from langgraph-api: {body}", exc_info=e) + raise e + # check content type + content_type = res.headers.get("content-type", "").partition(";")[0] + if "text/event-stream" not in content_type: + raise httpx.TransportError( + "Expected response header Content-Type to contain 'text/event-stream', " + f"got {content_type!r}" + ) + # parse SSE + decoder = SSEDecoder() + async for line in aiter_lines_raw(res): + sse = decoder.decode(line=line.rstrip(b"\n")) + if sse is not None: + yield sse + + +async def _aencode_json(json: Any) -> tuple[dict[str, str], bytes]: + if json is None: + return {}, None + body = await asyncio.get_running_loop().run_in_executor( + None, + orjson.dumps, + json, + _orjson_default, + orjson.OPT_SERIALIZE_NUMPY | orjson.OPT_NON_STR_KEYS, + ) + content_length = str(len(body)) + content_type = "application/json" + headers = {"Content-Length": content_length, "Content-Type": content_type} + return headers, body + + +async def _adecode_json(r: httpx.Response) -> Any: + body = await r.aread() + return ( + await asyncio.get_running_loop().run_in_executor(None, orjson.loads, body) + if body + else None + ) + + +class AssistantsClient: + """Client for managing assistants in LangGraph. + + This class provides methods to interact with assistants, + which are versioned configurations of your graph. + + ???+ example "Example" + + ```python + client = get_client(url="http://localhost:2024") + assistant = await client.assistants.get("assistant_id_123") + ``` + """ + + def __init__(self, http: HttpClient) -> None: + self.http = http + + async def get( + self, assistant_id: str, *, headers: dict[str, str] | 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. + + Returns: + Assistant: Assistant Object. + + ???+ example "Example Usage" + + ```python + assistant = await client.assistants.get( + assistant_id="my_assistant_id" + ) + print(assistant) + ``` + + ```shell + ---------------------------------------------------- + + { + 'assistant_id': 'my_assistant_id', + 'graph_id': 'agent', + 'created_at': '2024-06-25T17:10:33.109781+00:00', + 'updated_at': '2024-06-25T17:10:33.109781+00:00', + 'config': {}, + 'metadata': {'created_by': 'system'}, + 'version': 1, + 'name': 'my_assistant' + } + ``` + """ # noqa: E501 + return await self.http.get(f"/assistants/{assistant_id}", headers=headers) + + async def get_graph( + self, + assistant_id: str, + *, + xray: int | bool = False, + headers: dict[str, str] | 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. + + Returns: + Graph: The graph information for the assistant in JSON format. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + graph_info = await client.assistants.get_graph( + assistant_id="my_assistant_id" + ) + print(graph_info) + ``` + + ```shell + + -------------------------------------------------------------------------------------------------------------------------- + + { + 'nodes': + [ + {'id': '__start__', 'type': 'schema', 'data': '__start__'}, + {'id': '__end__', 'type': 'schema', 'data': '__end__'}, + {'id': 'agent','type': 'runnable','data': {'id': ['langgraph', 'utils', 'RunnableCallable'],'name': 'agent'}}, + ], + 'edges': + [ + {'source': '__start__', 'target': 'agent'}, + {'source': 'agent','target': '__end__'} + ] + } + ``` + + + """ # noqa: E501 + return await self.http.get( + f"/assistants/{assistant_id}/graph", params={"xray": xray}, headers=headers + ) + + async def get_schemas( + self, assistant_id: str, *, headers: dict[str, str] | None = None + ) -> GraphSchema: + """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. + + Returns: + GraphSchema: The graph schema for the assistant. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + schema = await client.assistants.get_schemas( + assistant_id="my_assistant_id" + ) + print(schema) + ``` + + ```shell + + ---------------------------------------------------------------------------------------------------------------------------- + + { + 'graph_id': 'agent', + 'state_schema': + { + 'title': 'LangGraphInput', + '$ref': '#/definitions/AgentState', + 'definitions': + { + 'BaseMessage': + { + 'title': 'BaseMessage', + 'description': 'Base abstract Message class. Messages are the inputs and outputs of ChatModels.', + 'type': 'object', + 'properties': + { + 'content': + { + 'title': 'Content', + 'anyOf': [ + {'type': 'string'}, + {'type': 'array','items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]}} + ] + }, + 'additional_kwargs': + { + 'title': 'Additional Kwargs', + 'type': 'object' + }, + 'response_metadata': + { + 'title': 'Response Metadata', + 'type': 'object' + }, + 'type': + { + 'title': 'Type', + 'type': 'string' + }, + 'name': + { + 'title': 'Name', + 'type': 'string' + }, + 'id': + { + 'title': 'Id', + 'type': 'string' + } + }, + 'required': ['content', 'type'] + }, + 'AgentState': + { + 'title': 'AgentState', + 'type': 'object', + 'properties': + { + 'messages': + { + 'title': 'Messages', + 'type': 'array', + 'items': {'$ref': '#/definitions/BaseMessage'} + } + }, + 'required': ['messages'] + } + } + }, + 'config_schema': + { + 'title': 'Configurable', + 'type': 'object', + 'properties': + { + 'model_name': + { + 'title': 'Model Name', + 'enum': ['anthropic', 'openai'], + 'type': 'string' + } + } + } + } + ``` + + """ # noqa: E501 + return await self.http.get( + f"/assistants/{assistant_id}/schemas", headers=headers + ) + + async def get_subgraphs( + self, + assistant_id: str, + namespace: str | None = None, + recurse: bool = False, + *, + headers: dict[str, str] | None = None, + ) -> Subgraphs: + """Get the schemas of an assistant by ID. + + Args: + assistant_id: The ID of the assistant to get the schema of. + namespace: Optional namespace to filter by. + recurse: Whether to recursively get subgraphs. + headers: Optional custom headers to include with the request. + + Returns: + Subgraphs: The graph schema for the assistant. + + """ # noqa: E501 + if namespace is not None: + return await self.http.get( + f"/assistants/{assistant_id}/subgraphs/{namespace}", + params={"recurse": recurse}, + headers=headers, + ) + else: + return await self.http.get( + f"/assistants/{assistant_id}/subgraphs", + params={"recurse": recurse}, + headers=headers, + ) + + async def create( + self, + graph_id: str | None, + config: Config | None = None, + *, + metadata: Json = None, + assistant_id: str | None = None, + if_exists: OnConflictBehavior | None = None, + name: str | None = None, + headers: dict[str, str] | None = None, + description: str | None = None, + ) -> Assistant: + """Create a new assistant. + + Useful when graph is configurable and you want to create different assistants based on different configurations. + + Args: + graph_id: The ID of the graph the assistant should use. The graph ID is normally set in your langgraph.json configuration. + config: Configuration to use for the graph. + metadata: Metadata to add to assistant. + assistant_id: Assistant ID to use, will default to a random UUID if not provided. + if_exists: How to handle duplicate creation. Defaults to 'raise' under the hood. + Must be either 'raise' (raise error if duplicate), or 'do_nothing' (return existing assistant). + name: The name of the assistant. Defaults to 'Untitled' under the hood. + headers: Optional custom headers to include with the request. + description: Optional description of the assistant. + The description field is available for langgraph-api server version>=0.0.45 + + Returns: + Assistant: The created assistant. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + assistant = await client.assistants.create( + graph_id="agent", + config={"configurable": {"model_name": "openai"}}, + metadata={"number":1}, + assistant_id="my-assistant-id", + if_exists="do_nothing", + name="my_name" + ) + ``` + """ # noqa: E501 + payload: dict[str, Any] = { + "graph_id": graph_id, + } + if config: + payload["config"] = config + if metadata: + payload["metadata"] = metadata + if assistant_id: + payload["assistant_id"] = assistant_id + if if_exists: + payload["if_exists"] = if_exists + if name: + payload["name"] = name + if description: + payload["description"] = description + return await self.http.post("/assistants", json=payload, headers=headers) + + async def update( + self, + assistant_id: str, + *, + graph_id: str | None = None, + config: Config | None = None, + metadata: Json = None, + name: str | None = None, + headers: dict[str, str] | None = None, + description: str | None = None, + ) -> Assistant: + """Update an assistant. + + Use this to point to a different graph, update the configuration, or change the metadata of an assistant. + + Args: + assistant_id: Assistant to update. + graph_id: The ID of the graph the assistant should use. + The graph ID is normally set in your langgraph.json configuration. If None, assistant will keep pointing to same graph. + config: Configuration to use for the graph. + metadata: Metadata to merge with existing assistant metadata. + name: The new name for the assistant. + headers: Optional custom headers to include with the request. + description: Optional description of the assistant. + The description field is available for langgraph-api server version>=0.0.45 + + Returns: + Assistant: The updated assistant. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + assistant = await client.assistants.update( + assistant_id='e280dad7-8618-443f-87f1-8e41841c180f', + graph_id="other-graph", + config={"configurable": {"model_name": "anthropic"}}, + metadata={"number":2} + ) + ``` + + """ # noqa: E501 + payload: dict[str, Any] = {} + if graph_id: + payload["graph_id"] = graph_id + if config: + payload["config"] = config + if metadata: + payload["metadata"] = metadata + if name: + payload["name"] = name + if description: + payload["description"] = description + return await self.http.patch( + f"/assistants/{assistant_id}", + json=payload, + headers=headers, + ) + + async def delete( + self, + assistant_id: str, + *, + headers: dict[str, str] | None = None, + ) -> None: + """Delete an assistant. + + Args: + assistant_id: The assistant ID to delete. + headers: Optional custom headers to include with the request. + + Returns: + None + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + await client.assistants.delete( + assistant_id="my_assistant_id" + ) + ``` + + """ # noqa: E501 + await self.http.delete(f"/assistants/{assistant_id}", headers=headers) + + async def search( + self, + *, + metadata: Json = None, + graph_id: str | None = None, + limit: int = 10, + offset: int = 0, + sort_by: AssistantSortBy | None = None, + sort_order: SortOrder | None = None, + headers: dict[str, str] | None = None, + ) -> list[Assistant]: + """Search for assistants. + + Args: + metadata: Metadata to filter by. Exact match filter for each KV pair. + graph_id: The ID of the graph to filter by. + The graph ID is normally set in your langgraph.json configuration. + limit: The maximum number of results to return. + offset: The number of results to skip. + sort_by: The field to sort by. + sort_order: The order to sort by. + headers: Optional custom headers to include with the request. + + Returns: + list[Assistant]: A list of assistants. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + assistants = await client.assistants.search( + metadata = {"name":"my_name"}, + graph_id="my_graph_id", + limit=5, + offset=5 + ) + ``` + """ + payload: dict[str, Any] = { + "limit": limit, + "offset": offset, + } + if metadata: + payload["metadata"] = metadata + if graph_id: + payload["graph_id"] = graph_id + if sort_by: + payload["sort_by"] = sort_by + if sort_order: + payload["sort_order"] = sort_order + return await self.http.post( + "/assistants/search", + json=payload, + headers=headers, + ) + + async def get_versions( + self, + assistant_id: str, + metadata: Json = None, + limit: int = 10, + offset: int = 0, + *, + headers: dict[str, str] | 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. + + Returns: + list[AssistantVersion]: A list of assistant versions. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + assistant_versions = await client.assistants.get_versions( + assistant_id="my_assistant_id" + ) + ``` + """ # noqa: E501 + + payload: dict[str, Any] = { + "limit": limit, + "offset": offset, + } + if metadata: + payload["metadata"] = metadata + return await self.http.post( + f"/assistants/{assistant_id}/versions", json=payload, headers=headers + ) + + async def set_latest( + self, + assistant_id: str, + version: int, + *, + headers: dict[str, str] | None = None, + ) -> Assistant: + """Change the version of an assistant. + + Args: + assistant_id: The assistant ID to delete. + version: The version to change to. + headers: Optional custom headers to include with the request. + + Returns: + Assistant: Assistant Object. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + new_version_assistant = await client.assistants.set_latest( + assistant_id="my_assistant_id", + version=3 + ) + ``` + + """ # noqa: E501 + + payload: dict[str, Any] = {"version": version} + + return await self.http.post( + f"/assistants/{assistant_id}/latest", json=payload, headers=headers + ) + + +class ThreadsClient: + """Client for managing threads in LangGraph. + + A thread maintains the state of a graph across multiple interactions/invocations (aka runs). + It accumulates and persists the graph's state, allowing for continuity between separate + invocations of the graph. + + ???+ example "Example" + + ```python + client = get_client(url="http://localhost:2024")) + new_thread = await client.threads.create(metadata={"user_id": "123"}) + ``` + """ + + def __init__(self, http: HttpClient) -> None: + self.http = http + + async def get( + self, thread_id: str, *, headers: dict[str, str] | None = None + ) -> Thread: + """Get a thread by ID. + + Args: + thread_id: The ID of the thread to get. + headers: Optional custom headers to include with the request. + + Returns: + Thread: Thread object. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + thread = await client.threads.get( + thread_id="my_thread_id" + ) + print(thread) + ``` + + ```shell + ----------------------------------------------------- + + { + 'thread_id': 'my_thread_id', + 'created_at': '2024-07-18T18:35:15.540834+00:00', + 'updated_at': '2024-07-18T18:35:15.540834+00:00', + 'metadata': {'graph_id': 'agent'} + } + ``` + + """ # noqa: E501 + + return await self.http.get(f"/threads/{thread_id}", headers=headers) + + async def create( + self, + *, + metadata: Json = None, + thread_id: str | None = None, + if_exists: OnConflictBehavior | None = None, + supersteps: Sequence[dict[str, Sequence[dict[str, Any]]]] | None = None, + graph_id: str | None = None, + headers: dict[str, str] | None = None, + ) -> Thread: + """Create a new thread. + + Args: + metadata: Metadata to add to thread. + thread_id: ID of thread. + If None, ID will be a randomly generated UUID. + if_exists: How to handle duplicate creation. Defaults to 'raise' under the hood. + Must be either 'raise' (raise error if duplicate), or 'do_nothing' (return existing thread). + supersteps: Apply a list of supersteps when creating a thread, each containing a sequence of updates. + Each update has `values` or `command` and `as_node`. Used for copying a thread between deployments. + graph_id: Optional graph ID to associate with the thread. + headers: Optional custom headers to include with the request. + + Returns: + Thread: The created thread. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + thread = await client.threads.create( + metadata={"number":1}, + thread_id="my-thread-id", + if_exists="raise" + ) + ``` + """ # noqa: E501 + payload: dict[str, Any] = {} + if thread_id: + payload["thread_id"] = thread_id + if metadata or graph_id: + payload["metadata"] = { + **(metadata or {}), + **({"graph_id": graph_id} if graph_id else {}), + } + if if_exists: + payload["if_exists"] = if_exists + if supersteps: + payload["supersteps"] = [ + { + "updates": [ + { + "values": u["values"], + "command": u.get("command"), + "as_node": u["as_node"], + } + for u in s["updates"] + ] + } + for s in supersteps + ] + + return await self.http.post("/threads", json=payload, headers=headers) + + async def update( + self, + thread_id: str, + *, + metadata: dict[str, Any], + headers: dict[str, str] | None = None, + ) -> Thread: + """Update a thread. + + Args: + thread_id: ID of thread to update. + metadata: Metadata to merge with existing thread metadata. + headers: Optional custom headers to include with the request. + + Returns: + Thread: The created thread. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + thread = await client.threads.update( + thread_id="my-thread-id", + metadata={"number":1}, + ) + ``` + """ # noqa: E501 + return await self.http.patch( + f"/threads/{thread_id}", json={"metadata": metadata}, headers=headers + ) + + async def delete( + self, thread_id: str, *, headers: dict[str, str] | None = None + ) -> None: + """Delete a thread. + + Args: + thread_id: The ID of the thread to delete. + headers: Optional custom headers to include with the request. + + Returns: + None + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost2024) + await client.threads.delete( + thread_id="my_thread_id" + ) + ``` + + """ # noqa: E501 + await self.http.delete(f"/threads/{thread_id}", headers=headers) + + async def search( + self, + *, + metadata: Json = None, + values: Json = None, + status: ThreadStatus | None = None, + limit: int = 10, + offset: int = 0, + sort_by: ThreadSortBy | None = None, + sort_order: SortOrder | None = None, + headers: dict[str, str] | None = None, + ) -> list[Thread]: + """Search for threads. + + Args: + metadata: Thread metadata to filter on. + values: State values to filter on. + status: Thread status to filter on. + Must be one of 'idle', 'busy', 'interrupted' or 'error'. + limit: Limit on number of threads to return. + offset: Offset in threads table to start search from. + sort_by: Sort by field. + sort_order: Sort order. + headers: Optional custom headers to include with the request. + + Returns: + list[Thread]: List of the threads matching the search parameters. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + threads = await client.threads.search( + metadata={"number":1}, + status="interrupted", + limit=15, + offset=5 + ) + ``` + + """ # noqa: E501 + payload: dict[str, Any] = { + "limit": limit, + "offset": offset, + } + if metadata: + payload["metadata"] = metadata + if values: + payload["values"] = values + if status: + payload["status"] = status + if sort_by: + payload["sort_by"] = sort_by + if sort_order: + payload["sort_order"] = sort_order + return await self.http.post( + "/threads/search", + json=payload, + headers=headers, + ) + + async def copy( + self, thread_id: str, *, headers: dict[str, str] | None = None + ) -> None: + """Copy a thread. + + Args: + thread_id: The ID of the thread to copy. + headers: Optional custom headers to include with the request. + + Returns: + None + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024) + await client.threads.copy( + thread_id="my_thread_id" + ) + ``` + + """ # noqa: E501 + return await self.http.post( + f"/threads/{thread_id}/copy", json=None, headers=headers + ) + + async def get_state( + self, + thread_id: str, + checkpoint: Checkpoint | None = None, + checkpoint_id: str | None = None, # deprecated + *, + subgraphs: bool = False, + headers: dict[str, str] | None = None, + ) -> ThreadState: + """Get the state of a thread. + + Args: + thread_id: The ID of the thread to get the state of. + checkpoint: The checkpoint to get the state of. + checkpoint_id: (deprecated) The checkpoint ID to get the state of. + subgraphs: Include subgraphs states. + headers: Optional custom headers to include with the request. + + Returns: + ThreadState: the thread of the state. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024) + thread_state = await client.threads.get_state( + thread_id="my_thread_id", + checkpoint_id="my_checkpoint_id" + ) + print(thread_state) + ``` + + ```shell + ---------------------------------------------------------------------------------------------------------------------------------------------------------------------- + + { + 'values': { + 'messages': [ + { + 'content': 'how are you?', + 'additional_kwargs': {}, + 'response_metadata': {}, + 'type': 'human', + 'name': None, + 'id': 'fe0a5778-cfe9-42ee-b807-0adaa1873c10', + 'example': False + }, + { + 'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.", + 'additional_kwargs': {}, + 'response_metadata': {}, + 'type': 'ai', + 'name': None, + 'id': 'run-159b782c-b679-4830-83c6-cef87798fe8b', + 'example': False, + 'tool_calls': [], + 'invalid_tool_calls': [], + 'usage_metadata': None + } + ] + }, + 'next': [], + 'checkpoint': + { + 'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2', + 'checkpoint_ns': '', + 'checkpoint_id': '1ef4a9b8-e6fb-67b1-8001-abd5184439d1' + } + 'metadata': + { + 'step': 1, + 'run_id': '1ef4a9b8-d7da-679a-a45a-872054341df2', + 'source': 'loop', + 'writes': + { + 'agent': + { + 'messages': [ + { + 'id': 'run-159b782c-b679-4830-83c6-cef87798fe8b', + 'name': None, + 'type': 'ai', + 'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.", + 'example': False, + 'tool_calls': [], + 'usage_metadata': None, + 'additional_kwargs': {}, + 'response_metadata': {}, + 'invalid_tool_calls': [] + } + ] + } + }, + 'user_id': None, + 'graph_id': 'agent', + 'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2', + 'created_by': 'system', + 'assistant_id': 'fe096781-5601-53d2-b2f6-0d3403f7e9ca'}, + 'created_at': '2024-07-25T15:35:44.184703+00:00', + 'parent_config': + { + 'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2', + 'checkpoint_ns': '', + 'checkpoint_id': '1ef4a9b8-d80d-6fa7-8000-9300467fad0f' + } + } + ``` + """ # noqa: E501 + if checkpoint: + return await self.http.post( + f"/threads/{thread_id}/state/checkpoint", + json={"checkpoint": checkpoint, "subgraphs": subgraphs}, + headers=headers, + ) + elif checkpoint_id: + return await self.http.get( + f"/threads/{thread_id}/state/{checkpoint_id}", + params={"subgraphs": subgraphs}, + headers=headers, + ) + else: + return await self.http.get( + f"/threads/{thread_id}/state", + params={"subgraphs": subgraphs}, + headers=headers, + ) + + async def update_state( + self, + thread_id: str, + values: dict | Sequence[dict] | None, + *, + as_node: str | None = None, + checkpoint: Checkpoint | None = None, + checkpoint_id: str | None = None, # deprecated + headers: dict[str, str] | None = None, + ) -> ThreadUpdateStateResponse: + """Update the state of a thread. + + Args: + thread_id: The ID of the thread to update. + values: The values to update the state with. + as_node: Update the state as if this node had just executed. + checkpoint: The checkpoint to update the state of. + checkpoint_id: (deprecated) The checkpoint ID to update the state of. + headers: Optional custom headers to include with the request. + + Returns: + ThreadUpdateStateResponse: Response after updating a thread's state. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024) + response = await client.threads.update_state( + thread_id="my_thread_id", + values={"messages":[{"role": "user", "content": "hello!"}]}, + as_node="my_node", + ) + print(response) + ``` + ```shell + + ---------------------------------------------------------------------------------------------------------------------------------------------------------------------- + + { + 'checkpoint': { + 'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2', + 'checkpoint_ns': '', + 'checkpoint_id': '1ef4a9b8-e6fb-67b1-8001-abd5184439d1', + 'checkpoint_map': {} + } + } + ``` + """ # noqa: E501 + payload: dict[str, Any] = { + "values": values, + } + if checkpoint_id: + payload["checkpoint_id"] = checkpoint_id + if checkpoint: + payload["checkpoint"] = checkpoint + if as_node: + payload["as_node"] = as_node + return await self.http.post( + f"/threads/{thread_id}/state", json=payload, headers=headers + ) + + async def get_history( + self, + thread_id: str, + *, + limit: int = 10, + before: str | Checkpoint | None = None, + metadata: dict | None = None, + checkpoint: Checkpoint | None = None, + headers: dict[str, str] | None = None, + ) -> list[ThreadState]: + """Get the state history of a thread. + + Args: + thread_id: The ID of the thread to get the state history for. + checkpoint: Return states for this subgraph. If empty defaults to root. + limit: The maximum number of states to return. + before: Return states before this checkpoint. + metadata: Filter states by metadata key-value pairs. + headers: Optional custom headers to include with the request. + + Returns: + list[ThreadState]: the state history of the thread. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024) + thread_state = await client.threads.get_history( + thread_id="my_thread_id", + limit=5, + ) + ``` + + """ # noqa: E501 + payload: dict[str, Any] = { + "limit": limit, + } + if before: + payload["before"] = before + if metadata: + payload["metadata"] = metadata + if checkpoint: + payload["checkpoint"] = checkpoint + return await self.http.post( + f"/threads/{thread_id}/history", json=payload, headers=headers + ) + + +class RunsClient: + """Client for managing runs in LangGraph. + + A run is a single assistant invocation with optional input, config, and metadata. + This client manages runs, which can be stateful (on threads) or stateless. + + ???+ example "Example" + + ```python + client = get_client(url="http://localhost:2024") + run = await client.runs.create(assistant_id="asst_123", thread_id="thread_456", input={"query": "Hello"}) + ``` + """ + + def __init__(self, http: HttpClient) -> None: + self.http = http + + @overload + def stream( + self, + thread_id: str, + assistant_id: str, + *, + input: dict | None = None, + command: Command | None = None, + stream_mode: StreamMode | Sequence[StreamMode] = "values", + stream_subgraphs: bool = False, + stream_resumable: bool = False, + metadata: dict | None = None, + config: Config | None = None, + checkpoint: Checkpoint | None = None, + checkpoint_id: str | None = None, + checkpoint_during: bool | None = None, + interrupt_before: All | Sequence[str] | None = None, + interrupt_after: All | Sequence[str] | None = None, + feedback_keys: Sequence[str] | None = None, + on_disconnect: DisconnectMode | None = None, + webhook: str | None = None, + multitask_strategy: MultitaskStrategy | None = None, + if_not_exists: IfNotExists | None = None, + after_seconds: int | None = None, + headers: dict[str, str] | None = None, + on_run_created: Callable[[RunCreateMetadata], None] | None = None, + ) -> AsyncIterator[StreamPart]: ... + + @overload + def stream( + self, + thread_id: None, + assistant_id: str, + *, + input: dict | None = None, + command: Command | None = None, + stream_mode: StreamMode | Sequence[StreamMode] = "values", + stream_subgraphs: bool = False, + stream_resumable: bool = False, + metadata: dict | None = None, + config: Config | None = None, + checkpoint_during: bool | None = None, + interrupt_before: All | Sequence[str] | None = None, + interrupt_after: All | Sequence[str] | None = None, + feedback_keys: Sequence[str] | None = None, + on_disconnect: DisconnectMode | None = None, + on_completion: OnCompletionBehavior | None = None, + if_not_exists: IfNotExists | None = None, + webhook: str | None = None, + after_seconds: int | None = None, + headers: dict[str, str] | None = None, + on_run_created: Callable[[RunCreateMetadata], None] | None = None, + ) -> AsyncIterator[StreamPart]: ... + + def stream( + self, + thread_id: str | None, + assistant_id: str, + *, + input: dict | None = None, + command: Command | None = None, + stream_mode: StreamMode | Sequence[StreamMode] = "values", + stream_subgraphs: bool = False, + stream_resumable: bool = False, + metadata: dict | None = None, + config: Config | None = None, + checkpoint: Checkpoint | None = None, + checkpoint_id: str | None = None, + checkpoint_during: bool | None = None, + interrupt_before: All | Sequence[str] | None = None, + interrupt_after: All | Sequence[str] | None = None, + feedback_keys: Sequence[str] | None = None, + on_disconnect: DisconnectMode | None = None, + on_completion: OnCompletionBehavior | None = None, + webhook: str | None = None, + multitask_strategy: MultitaskStrategy | None = None, + if_not_exists: IfNotExists | None = None, + after_seconds: int | None = None, + headers: dict[str, str] | None = None, + on_run_created: Callable[[RunCreateMetadata], None] | None = None, + ) -> AsyncIterator[StreamPart]: + """Create a run and stream the results. + + Args: + thread_id: the thread ID to assign to the thread. + If None will create a stateless run. + assistant_id: The assistant ID or graph name to stream from. + If using graph name, will default to first assistant created from that graph. + input: The input to the graph. + command: A command to execute. Cannot be combined with input. + stream_mode: The stream mode(s) to use. + stream_subgraphs: Whether to stream output from subgraphs. + stream_resumable: Whether the stream is considered resumable. + If true, the stream can be resumed and replayed in its entirety even after disconnection. + metadata: Metadata to assign to the run. + config: The configuration for the assistant. + checkpoint: The checkpoint to resume from. + checkpoint_during: Whether to checkpoint during the run (or only at the end/interruption). + interrupt_before: Nodes to interrupt immediately before they get executed. + interrupt_after: Nodes to Nodes to interrupt immediately after they get executed. + feedback_keys: Feedback keys to assign to run. + on_disconnect: The disconnect mode to use. + Must be one of 'cancel' or 'continue'. + on_completion: Whether to delete or keep the thread created for a stateless run. + Must be one of 'delete' or 'keep'. + webhook: Webhook to call after LangGraph API call is done. + multitask_strategy: Multitask strategy to use. + Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + if_not_exists: How to handle missing thread. Defaults to 'reject'. + Must be either 'reject' (raise error if missing), or 'create' (create new thread). + after_seconds: The number of seconds to wait before starting the run. + Use to schedule future runs. + on_run_created: Callback when a run is created. + + Returns: + AsyncIterator[StreamPart]: Asynchronous iterator of stream results. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024) + async for chunk in client.runs.stream( + thread_id=None, + assistant_id="agent", + input={"messages": [{"role": "user", "content": "how are you?"}]}, + stream_mode=["values","debug"], + metadata={"name":"my_run"}, + config={"configurable": {"model_name": "anthropic"}}, + interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"], + interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"], + feedback_keys=["my_feedback_key_1","my_feedback_key_2"], + webhook="https://my.fake.webhook.com", + multitask_strategy="interrupt" + ): + print(chunk) + ``` + + ```shell + + ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ + + StreamPart(event='metadata', data={'run_id': '1ef4a9b8-d7da-679a-a45a-872054341df2'}) + StreamPart(event='values', data={'messages': [{'content': 'how are you?', 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'human', 'name': None, 'id': 'fe0a5778-cfe9-42ee-b807-0adaa1873c10', 'example': False}]}) + StreamPart(event='values', data={'messages': [{'content': 'how are you?', 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'human', 'name': None, 'id': 'fe0a5778-cfe9-42ee-b807-0adaa1873c10', 'example': False}, {'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.", 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'ai', 'name': None, 'id': 'run-159b782c-b679-4830-83c6-cef87798fe8b', 'example': False, 'tool_calls': [], 'invalid_tool_calls': [], 'usage_metadata': None}]}) + StreamPart(event='end', data=None) + ``` + + """ # noqa: E501 + payload = { + "input": input, + "command": ( + {k: v for k, v in command.items() if v is not None} if command else None + ), + "config": config, + "metadata": metadata, + "stream_mode": stream_mode, + "stream_subgraphs": stream_subgraphs, + "stream_resumable": stream_resumable, + "assistant_id": assistant_id, + "interrupt_before": interrupt_before, + "interrupt_after": interrupt_after, + "feedback_keys": feedback_keys, + "webhook": webhook, + "checkpoint": checkpoint, + "checkpoint_id": checkpoint_id, + "checkpoint_during": checkpoint_during, + "multitask_strategy": multitask_strategy, + "if_not_exists": if_not_exists, + "on_disconnect": on_disconnect, + "on_completion": on_completion, + "after_seconds": after_seconds, + } + endpoint = ( + f"/threads/{thread_id}/runs/stream" + if thread_id is not None + else "/runs/stream" + ) + + def on_response(res: httpx.Response): + """Callback function to handle the response.""" + if on_run_created and (metadata := _get_run_metadata_from_response(res)): + on_run_created(metadata) + + return self.http.stream( + endpoint, + "POST", + json={k: v for k, v in payload.items() if v is not None}, + headers=headers, + on_response=on_response if on_run_created else None, + ) + + @overload + async def create( + self, + thread_id: None, + assistant_id: str, + *, + input: dict | None = None, + command: Command | None = None, + stream_mode: StreamMode | Sequence[StreamMode] = "values", + stream_subgraphs: bool = False, + stream_resumable: bool = False, + metadata: dict | None = None, + checkpoint_during: bool | None = None, + config: Config | None = None, + interrupt_before: All | Sequence[str] | None = None, + interrupt_after: All | Sequence[str] | None = None, + webhook: str | None = None, + on_completion: OnCompletionBehavior | None = None, + if_not_exists: IfNotExists | None = None, + after_seconds: int | None = None, + headers: dict[str, str] | None = None, + on_run_created: Callable[[RunCreateMetadata], None] | None = None, + ) -> Run: ... + + @overload + async def create( + self, + thread_id: str, + assistant_id: str, + *, + input: dict | None = None, + command: Command | None = None, + stream_mode: StreamMode | Sequence[StreamMode] = "values", + stream_subgraphs: bool = False, + stream_resumable: bool = False, + metadata: dict | None = None, + config: Config | None = None, + checkpoint: Checkpoint | None = None, + checkpoint_id: str | None = None, + checkpoint_during: bool | None = None, + interrupt_before: All | Sequence[str] | None = None, + interrupt_after: All | Sequence[str] | None = None, + webhook: str | None = None, + multitask_strategy: MultitaskStrategy | None = None, + if_not_exists: IfNotExists | None = None, + after_seconds: int | None = None, + headers: dict[str, str] | None = None, + on_run_created: Callable[[RunCreateMetadata], None] | None = None, + ) -> Run: ... + + async def create( + self, + thread_id: str | None, + assistant_id: str, + *, + input: dict | None = None, + command: Command | None = None, + stream_mode: StreamMode | Sequence[StreamMode] = "values", + stream_subgraphs: bool = False, + stream_resumable: bool = False, + metadata: dict | None = None, + config: Config | None = None, + checkpoint: Checkpoint | None = None, + checkpoint_id: str | None = None, + checkpoint_during: bool | None = None, + interrupt_before: All | Sequence[str] | None = None, + interrupt_after: All | Sequence[str] | None = None, + webhook: str | None = None, + multitask_strategy: MultitaskStrategy | None = None, + if_not_exists: IfNotExists | None = None, + on_completion: OnCompletionBehavior | None = None, + after_seconds: int | None = None, + headers: dict[str, str] | None = None, + on_run_created: Callable[[RunCreateMetadata], None] | None = None, + ) -> Run: + """Create a background run. + + Args: + thread_id: the thread ID to assign to the thread. + If None will create a stateless run. + assistant_id: The assistant ID or graph name to stream from. + If using graph name, will default to first assistant created from that graph. + input: The input to the graph. + command: A command to execute. Cannot be combined with input. + stream_mode: The stream mode(s) to use. + stream_subgraphs: Whether to stream output from subgraphs. + stream_resumable: Whether the stream is considered resumable. + If true, the stream can be resumed and replayed in its entirety even after disconnection. + metadata: Metadata to assign to the run. + config: The configuration for the assistant. + checkpoint: The checkpoint to resume from. + checkpoint_during: Whether to checkpoint during the run (or only at the end/interruption). + interrupt_before: Nodes to interrupt immediately before they get executed. + interrupt_after: Nodes to Nodes to interrupt immediately after they get executed. + webhook: Webhook to call after LangGraph API call is done. + multitask_strategy: Multitask strategy to use. + Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + on_completion: Whether to delete or keep the thread created for a stateless run. + Must be one of 'delete' or 'keep'. + if_not_exists: How to handle missing thread. Defaults to 'reject'. + Must be either 'reject' (raise error if missing), or 'create' (create new thread). + after_seconds: The number of seconds to wait before starting the run. + Use to schedule future runs. + headers: Optional custom headers to include with the request. + on_run_created: Optional callback to call when a run is created. + + Returns: + Run: The created background run. + + ???+ example "Example Usage" + + ```python + + background_run = await client.runs.create( + thread_id="my_thread_id", + assistant_id="my_assistant_id", + input={"messages": [{"role": "user", "content": "hello!"}]}, + metadata={"name":"my_run"}, + config={"configurable": {"model_name": "openai"}}, + interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"], + interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"], + webhook="https://my.fake.webhook.com", + multitask_strategy="interrupt" + ) + print(background_run) + ``` + + ```shell + -------------------------------------------------------------------------------- + + { + 'run_id': 'my_run_id', + 'thread_id': 'my_thread_id', + 'assistant_id': 'my_assistant_id', + 'created_at': '2024-07-25T15:35:42.598503+00:00', + 'updated_at': '2024-07-25T15:35:42.598503+00:00', + 'metadata': {}, + 'status': 'pending', + 'kwargs': + { + 'input': + { + 'messages': [ + { + 'role': 'user', + 'content': 'how are you?' + } + ] + }, + 'config': + { + 'metadata': + { + 'created_by': 'system' + }, + 'configurable': + { + 'run_id': 'my_run_id', + 'user_id': None, + 'graph_id': 'agent', + 'thread_id': 'my_thread_id', + 'checkpoint_id': None, + 'model_name': "openai", + 'assistant_id': 'my_assistant_id' + } + }, + 'webhook': "https://my.fake.webhook.com", + 'temporary': False, + 'stream_mode': ['values'], + 'feedback_keys': None, + 'interrupt_after': ["node_to_stop_after_1","node_to_stop_after_2"], + 'interrupt_before': ["node_to_stop_before_1","node_to_stop_before_2"] + }, + 'multitask_strategy': 'interrupt' + } + ``` + """ # noqa: E501 + payload = { + "input": input, + "command": ( + {k: v for k, v in command.items() if v is not None} if command else None + ), + "stream_mode": stream_mode, + "stream_subgraphs": stream_subgraphs, + "stream_resumable": stream_resumable, + "config": config, + "metadata": metadata, + "assistant_id": assistant_id, + "interrupt_before": interrupt_before, + "interrupt_after": interrupt_after, + "webhook": webhook, + "checkpoint": checkpoint, + "checkpoint_id": checkpoint_id, + "checkpoint_during": checkpoint_during, + "multitask_strategy": multitask_strategy, + "if_not_exists": if_not_exists, + "on_completion": on_completion, + "after_seconds": after_seconds, + } + payload = {k: v for k, v in payload.items() if v is not None} + + def on_response(res: httpx.Response): + """Callback function to handle the response.""" + if on_run_created and (metadata := _get_run_metadata_from_response(res)): + on_run_created(metadata) + + return await self.http.post( + f"/threads/{thread_id}/runs" if thread_id else "/runs", + json=payload, + headers=headers, + on_response=on_response if on_run_created else None, + ) + + async def create_batch(self, payloads: list[RunCreate]) -> list[Run]: + """Create a batch of stateless background runs.""" + + def filter_payload(payload: RunCreate): + return {k: v for k, v in payload.items() if v is not None} + + payloads = [filter_payload(payload) for payload in payloads] + return await self.http.post("/runs/batch", json=payloads) + + @overload + async def wait( + self, + thread_id: str, + assistant_id: str, + *, + input: dict | None = None, + command: Command | None = None, + metadata: dict | None = None, + config: Config | None = None, + checkpoint: Checkpoint | None = None, + checkpoint_id: str | None = None, + checkpoint_during: bool | None = None, + interrupt_before: All | Sequence[str] | None = None, + interrupt_after: All | Sequence[str] | None = None, + webhook: str | None = None, + on_disconnect: DisconnectMode | None = None, + multitask_strategy: MultitaskStrategy | None = None, + if_not_exists: IfNotExists | None = None, + after_seconds: int | None = None, + raise_error: bool = True, + headers: dict[str, str] | None = None, + on_run_created: Callable[[RunCreateMetadata], None] | None = None, + ) -> list[dict] | dict[str, Any]: ... + + @overload + async def wait( + self, + thread_id: None, + assistant_id: str, + *, + input: dict | None = None, + command: Command | None = None, + metadata: dict | None = None, + config: Config | None = None, + checkpoint_during: bool | None = None, + interrupt_before: All | Sequence[str] | None = None, + interrupt_after: All | Sequence[str] | None = None, + webhook: str | None = None, + on_disconnect: DisconnectMode | None = None, + on_completion: OnCompletionBehavior | None = None, + if_not_exists: IfNotExists | None = None, + after_seconds: int | None = None, + raise_error: bool = True, + headers: dict[str, str] | None = None, + on_run_created: Callable[[RunCreateMetadata], None] | None = None, + ) -> list[dict] | dict[str, Any]: ... + + async def wait( + self, + thread_id: str | None, + assistant_id: str, + *, + input: dict | None = None, + command: Command | None = None, + metadata: dict | None = None, + config: Config | None = None, + checkpoint: Checkpoint | None = None, + checkpoint_id: str | None = None, + checkpoint_during: bool | None = None, + interrupt_before: All | Sequence[str] | None = None, + interrupt_after: All | Sequence[str] | None = None, + webhook: str | None = None, + on_disconnect: DisconnectMode | None = None, + on_completion: OnCompletionBehavior | None = None, + multitask_strategy: MultitaskStrategy | None = None, + if_not_exists: IfNotExists | None = None, + after_seconds: int | None = None, + raise_error: bool = True, + headers: dict[str, str] | None = None, + on_run_created: Callable[[RunCreateMetadata], None] | None = None, + ) -> list[dict] | dict[str, Any]: + """Create a run, wait until it finishes and return the final state. + + Args: + thread_id: the thread ID to create the run on. + If None will create a stateless run. + assistant_id: The assistant ID or graph name to run. + If using graph name, will default to first assistant created from that graph. + input: The input to the graph. + command: A command to execute. Cannot be combined with input. + metadata: Metadata to assign to the run. + config: The configuration for the assistant. + checkpoint: The checkpoint to resume from. + checkpoint_during: Whether to checkpoint during the run (or only at the end/interruption). + interrupt_before: Nodes to interrupt immediately before they get executed. + interrupt_after: Nodes to Nodes to interrupt immediately after they get executed. + webhook: Webhook to call after LangGraph API call is done. + on_disconnect: The disconnect mode to use. + Must be one of 'cancel' or 'continue'. + on_completion: Whether to delete or keep the thread created for a stateless run. + Must be one of 'delete' or 'keep'. + multitask_strategy: Multitask strategy to use. + Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + if_not_exists: How to handle missing thread. Defaults to 'reject'. + Must be either 'reject' (raise error if missing), or 'create' (create new thread). + after_seconds: The number of seconds to wait before starting the run. + Use to schedule future runs. + headers: Optional custom headers to include with the request. + on_run_created: Optional callback to call when a run is created. + + Returns: + Union[list[dict], dict[str, Any]]: The output of the run. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + final_state_of_run = await client.runs.wait( + thread_id=None, + assistant_id="agent", + input={"messages": [{"role": "user", "content": "how are you?"}]}, + metadata={"name":"my_run"}, + config={"configurable": {"model_name": "anthropic"}}, + interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"], + interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"], + webhook="https://my.fake.webhook.com", + multitask_strategy="interrupt" + ) + print(final_state_of_run) + ``` + + ```shell + ------------------------------------------------------------------------------------------------------------------------------------------- + + { + 'messages': [ + { + 'content': 'how are you?', + 'additional_kwargs': {}, + 'response_metadata': {}, + 'type': 'human', + 'name': None, + 'id': 'f51a862c-62fe-4866-863b-b0863e8ad78a', + 'example': False + }, + { + 'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.", + 'additional_kwargs': {}, + 'response_metadata': {}, + 'type': 'ai', + 'name': None, + 'id': 'run-bf1cd3c6-768f-4c16-b62d-ba6f17ad8b36', + 'example': False, + 'tool_calls': [], + 'invalid_tool_calls': [], + 'usage_metadata': None + } + ] + } + ``` + + """ # noqa: E501 + payload = { + "input": input, + "command": ( + {k: v for k, v in command.items() if v is not None} if command else None + ), + "config": config, + "metadata": metadata, + "assistant_id": assistant_id, + "interrupt_before": interrupt_before, + "interrupt_after": interrupt_after, + "webhook": webhook, + "checkpoint": checkpoint, + "checkpoint_id": checkpoint_id, + "multitask_strategy": multitask_strategy, + "checkpoint_during": checkpoint_during, + "if_not_exists": if_not_exists, + "on_disconnect": on_disconnect, + "on_completion": on_completion, + "after_seconds": after_seconds, + } + endpoint = ( + f"/threads/{thread_id}/runs/wait" if thread_id is not None else "/runs/wait" + ) + + def on_response(res: httpx.Response): + """Callback function to handle the response.""" + if on_run_created and (metadata := _get_run_metadata_from_response(res)): + on_run_created(metadata) + + response = await self.http.post( + endpoint, + json={k: v for k, v in payload.items() if v is not None}, + headers=headers, + on_response=on_response if on_run_created else None, + ) + if ( + raise_error + and isinstance(response, dict) + and "__error__" in response + and isinstance(response["__error__"], dict) + ): + raise Exception( + f"{response['__error__'].get('error')}: {response['__error__'].get('message')}" + ) + return response + + async def list( + self, + thread_id: str, + *, + limit: int = 10, + offset: int = 0, + status: RunStatus | None = None, + headers: dict[str, str] | None = None, + ) -> list[Run]: + """List runs. + + Args: + thread_id: The thread ID to list runs for. + limit: The maximum number of results to return. + offset: The number of results to skip. + status: The status of the run to filter by. + headers: Optional custom headers to include with the request. + + Returns: + list[Run]: The runs for the thread. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + await client.runs.list( + thread_id="thread_id", + limit=5, + offset=5, + ) + ``` + + """ # noqa: E501 + params = { + "limit": limit, + "offset": offset, + } + if status is not None: + params["status"] = status + return await self.http.get( + f"/threads/{thread_id}/runs", params=params, headers=headers + ) + + async def get( + self, thread_id: str, run_id: str, *, headers: dict[str, str] | None = None + ) -> Run: + """Get a run. + + Args: + thread_id: The thread ID to get. + run_id: The run ID to get. + headers: Optional custom headers to include with the request. + + Returns: + Run: Run object. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + run = await client.runs.get( + thread_id="thread_id_to_delete", + run_id="run_id_to_delete", + ) + ``` + + """ # noqa: E501 + + return await self.http.get( + f"/threads/{thread_id}/runs/{run_id}", headers=headers + ) + + async def cancel( + self, + thread_id: str, + run_id: str, + *, + wait: bool = False, + action: CancelAction = "interrupt", + headers: dict[str, str] | None = None, + ) -> None: + """Get a run. + + Args: + thread_id: The thread ID to cancel. + run_id: The run ID to cancel. + wait: Whether to wait until run has completed. + action: Action to take when cancelling the run. Possible values + are `interrupt` or `rollback`. Default is `interrupt`. + headers: Optional custom headers to include with the request. + + Returns: + None + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + await client.runs.cancel( + thread_id="thread_id_to_cancel", + run_id="run_id_to_cancel", + wait=True, + action="interrupt" + ) + ``` + + """ # noqa: E501 + return await self.http.post( + f"/threads/{thread_id}/runs/{run_id}/cancel?wait={1 if wait else 0}&action={action}", + json=None, + headers=headers, + ) + + async def join( + self, thread_id: str, run_id: str, *, headers: dict[str, str] | None = None + ) -> dict: + """Block until a run is done. Returns the final state of the thread. + + Args: + thread_id: The thread ID to join. + run_id: The run ID to join. + headers: Optional custom headers to include with the request. + + Returns: + None + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + result =await client.runs.join( + thread_id="thread_id_to_join", + run_id="run_id_to_join" + ) + ``` + + """ # noqa: E501 + return await self.http.get( + f"/threads/{thread_id}/runs/{run_id}/join", headers=headers + ) + + def join_stream( + self, + thread_id: str, + run_id: str, + *, + cancel_on_disconnect: bool = False, + stream_mode: StreamMode | Sequence[StreamMode] | None = None, + headers: dict[str, str] | None = None, + last_event_id: str | None = None, + ) -> AsyncIterator[StreamPart]: + """Stream output from a run in real-time, until the run is done. + Output is not buffered, so any output produced before this call will + not be received here. + + Args: + thread_id: The thread ID to join. + run_id: The run ID to join. + cancel_on_disconnect: Whether to cancel the run when the stream is disconnected. + stream_mode: The stream mode(s) to use. Must be a subset of the stream modes passed + when creating the run. Background runs default to having the union of all + stream modes. + headers: Optional custom headers to include with the request. + + Returns: + None + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + async for part in client.runs.join_stream( + thread_id="thread_id_to_join", + run_id="run_id_to_join", + stream_mode=["values", "debug"] + ): + print(part) + ``` + + """ # noqa: E501 + return self.http.stream( + f"/threads/{thread_id}/runs/{run_id}/stream", + "GET", + params={ + "cancel_on_disconnect": cancel_on_disconnect, + "stream_mode": stream_mode, + }, + headers={ + **({"Last-Event-ID": last_event_id} if last_event_id else {}), + **(headers or {}), + } + or None, + ) + + async def delete( + self, thread_id: str, run_id: str, *, headers: dict[str, str] | None = None + ) -> None: + """Delete a run. + + Args: + thread_id: The thread ID to delete. + run_id: The run ID to delete. + headers: Optional custom headers to include with the request. + + Returns: + None + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + await client.runs.delete( + thread_id="thread_id_to_delete", + run_id="run_id_to_delete" + ) + ``` + + """ # noqa: E501 + await self.http.delete(f"/threads/{thread_id}/runs/{run_id}", headers=headers) + + +class CronClient: + """Client for managing recurrent runs (cron jobs) in LangGraph. + + A run is a single invocation of an assistant with optional input and config. + This client allows scheduling recurring runs to occur automatically. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024")) + cron_job = await client.crons.create_for_thread( + thread_id="thread_123", + assistant_id="asst_456", + schedule="0 9 * * *", + input={"message": "Daily update"} + ) + ``` + + !!! note "Feature Availability" + The crons client functionality is not supported on all licenses. + Please check the relevant license documentation for the most up-to-date + details on feature availability. + """ + + def __init__(self, http_client: HttpClient) -> None: + self.http = http_client + + async def create_for_thread( + self, + thread_id: str, + assistant_id: str, + *, + schedule: str, + input: dict | None = None, + metadata: dict | None = None, + config: Config | None = None, + checkpoint_during: bool | None = None, + interrupt_before: All | list[str] | None = None, + interrupt_after: All | list[str] | None = None, + webhook: str | None = None, + multitask_strategy: str | None = None, + headers: dict[str, str] | None = None, + ) -> Run: + """Create a cron job for a thread. + + Args: + thread_id: the thread ID to run the cron job on. + assistant_id: The assistant ID or graph name to use for the cron job. + If using graph name, will default to first assistant created from that graph. + schedule: The cron schedule to execute this job on. + input: The input to the graph. + metadata: Metadata to assign to the cron job runs. + config: The configuration for the assistant. + checkpoint_during: Whether to checkpoint during the run (or only at the end/interruption). + interrupt_before: Nodes to interrupt immediately before they get executed. + + interrupt_after: Nodes to Nodes to interrupt immediately after they get executed. + + webhook: Webhook to call after LangGraph API call is done. + multitask_strategy: Multitask strategy to use. + Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + headers: Optional custom headers to include with the request. + + Returns: + Run: The cron run. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + cron_run = await client.crons.create_for_thread( + thread_id="my-thread-id", + assistant_id="agent", + schedule="27 15 * * *", + input={"messages": [{"role": "user", "content": "hello!"}]}, + metadata={"name":"my_run"}, + config={"configurable": {"model_name": "openai"}}, + interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"], + interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"], + webhook="https://my.fake.webhook.com", + multitask_strategy="interrupt" + ) + ``` + """ # noqa: E501 + payload = { + "schedule": schedule, + "input": input, + "config": config, + "metadata": metadata, + "assistant_id": assistant_id, + "checkpoint_during": checkpoint_during, + "interrupt_before": interrupt_before, + "interrupt_after": interrupt_after, + "webhook": webhook, + } + if multitask_strategy: + payload["multitask_strategy"] = multitask_strategy + payload = {k: v for k, v in payload.items() if v is not None} + return await self.http.post( + f"/threads/{thread_id}/runs/crons", json=payload, headers=headers + ) + + async def create( + self, + assistant_id: str, + *, + schedule: str, + input: dict | None = None, + metadata: dict | None = None, + config: Config | None = None, + checkpoint_during: bool | None = None, + interrupt_before: All | list[str] | None = None, + interrupt_after: All | list[str] | None = None, + webhook: str | None = None, + multitask_strategy: str | None = None, + headers: dict[str, str] | None = None, + ) -> Run: + """Create a cron run. + + Args: + assistant_id: The assistant ID or graph name to use for the cron job. + If using graph name, will default to first assistant created from that graph. + schedule: The cron schedule to execute this job on. + input: The input to the graph. + metadata: Metadata to assign to the cron job runs. + config: The configuration for the assistant. + checkpoint_during: Whether to checkpoint during the run (or only at the end/interruption). + interrupt_before: Nodes to interrupt immediately before they get executed. + interrupt_after: Nodes to Nodes to interrupt immediately after they get executed. + webhook: Webhook to call after LangGraph API call is done. + multitask_strategy: Multitask strategy to use. + Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + headers: Optional custom headers to include with the request. + + Returns: + Run: The cron run. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + cron_run = client.crons.create( + assistant_id="agent", + schedule="27 15 * * *", + input={"messages": [{"role": "user", "content": "hello!"}]}, + metadata={"name":"my_run"}, + config={"configurable": {"model_name": "openai"}}, + interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"], + interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"], + webhook="https://my.fake.webhook.com", + multitask_strategy="interrupt" + ) + ``` + + """ # noqa: E501 + payload = { + "schedule": schedule, + "input": input, + "config": config, + "metadata": metadata, + "assistant_id": assistant_id, + "checkpoint_during": checkpoint_during, + "interrupt_before": interrupt_before, + "interrupt_after": interrupt_after, + "webhook": webhook, + } + if multitask_strategy: + payload["multitask_strategy"] = multitask_strategy + payload = {k: v for k, v in payload.items() if v is not None} + return await self.http.post("/runs/crons", json=payload, headers=headers) + + async def delete( + self, + cron_id: str, + headers: dict[str, str] | None = None, + ) -> None: + """Delete a cron. + + Args: + cron_id: The cron ID to delete. + headers: Optional custom headers to include with the request. + + Returns: + None + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + await client.crons.delete( + cron_id="cron_to_delete" + ) + ``` + + """ # noqa: E501 + await self.http.delete(f"/runs/crons/{cron_id}", headers=headers) + + async def search( + self, + *, + assistant_id: str | None = None, + thread_id: str | None = None, + limit: int = 10, + offset: int = 0, + sort_by: CronSortBy | None = None, + sort_order: SortOrder | None = None, + headers: dict[str, str] | None = None, + ) -> list[Cron]: + """Get a list of cron jobs. + + Args: + assistant_id: The assistant ID or graph name to search for. + thread_id: the thread ID to search for. + limit: The maximum number of results to return. + offset: The number of results to skip. + headers: Optional custom headers to include with the request. + + Returns: + list[Cron]: The list of cron jobs returned by the search, + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + cron_jobs = await client.crons.search( + assistant_id="my_assistant_id", + thread_id="my_thread_id", + limit=5, + offset=5, + ) + print(cron_jobs) + ``` + ```shell + + ---------------------------------------------------------- + + [ + { + 'cron_id': '1ef3cefa-4c09-6926-96d0-3dc97fd5e39b', + 'assistant_id': 'my_assistant_id', + 'thread_id': 'my_thread_id', + 'user_id': None, + 'payload': + { + 'input': {'start_time': ''}, + 'schedule': '4 * * * *', + 'assistant_id': 'my_assistant_id' + }, + 'schedule': '4 * * * *', + 'next_run_date': '2024-07-25T17:04:00+00:00', + 'end_time': None, + 'created_at': '2024-07-08T06:02:23.073257+00:00', + 'updated_at': '2024-07-08T06:02:23.073257+00:00' + } + ] + ``` + + """ # noqa: E501 + payload = { + "assistant_id": assistant_id, + "thread_id": thread_id, + "limit": limit, + "offset": offset, + } + if sort_by: + payload["sort_by"] = sort_by + if sort_order: + payload["sort_order"] = sort_order + payload = {k: v for k, v in payload.items() if v is not None} + return await self.http.post("/runs/crons/search", json=payload, headers=headers) + + +class StoreClient: + """Client for interacting with the graph's shared storage. + + The Store provides a key-value storage system for persisting data across graph executions, + allowing for stateful operations and data sharing across threads. + + ???+ example "Example" + + ```python + client = get_client(url="http://localhost:2024") + await client.store.put_item(["users", "user123"], "mem-123451342", {"name": "Alice", "score": 100}) + ``` + """ + + def __init__(self, http: HttpClient) -> None: + self.http = http + + async def put_item( + self, + namespace: Sequence[str], + /, + key: str, + value: dict[str, Any], + index: Literal[False] | list[str] | None = None, + ttl: int | None = None, + headers: dict[str, str] | None = None, + ) -> None: + """Store or update an item. + + Args: + namespace: A list of strings representing the namespace path. + key: The unique identifier for the item within the namespace. + value: A dictionary containing the item's data. + index: Controls search indexing - None (use defaults), False (disable), or list of field paths to index. + ttl: Optional time-to-live in minutes for the item, or None for no expiration. + headers: Optional custom headers to include with the request. + + Returns: + None + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + await client.store.put_item( + ["documents", "user123"], + key="item456", + value={"title": "My Document", "content": "Hello World"} + ) + ``` + """ + for label in namespace: + if "." in label: + raise ValueError( + f"Invalid namespace label '{label}'. Namespace labels cannot contain periods ('.')." + ) + payload = { + "namespace": namespace, + "key": key, + "value": value, + "index": index, + "ttl": ttl, + } + await self.http.put( + "/store/items", json=_provided_vals(payload), headers=headers + ) + + async def get_item( + self, + namespace: Sequence[str], + /, + key: str, + *, + refresh_ttl: bool | None = None, + headers: dict[str, str] | None = None, + ) -> Item: + """Retrieve a single item. + + Args: + key: The unique identifier for the item. + namespace: Optional list of strings representing the namespace path. + refresh_ttl: Whether to refresh the TTL on this read operation. If None, uses the store's default behavior. + + Returns: + Item: The retrieved item. + headers: Optional custom headers to include with the request. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + item = await client.store.get_item( + ["documents", "user123"], + key="item456", + ) + print(item) + ``` + ```shell + + ---------------------------------------------------------------- + + { + 'namespace': ['documents', 'user123'], + 'key': 'item456', + 'value': {'title': 'My Document', 'content': 'Hello World'}, + 'created_at': '2024-07-30T12:00:00Z', + 'updated_at': '2024-07-30T12:00:00Z' + } + ``` + """ + for label in namespace: + if "." in label: + raise ValueError( + f"Invalid namespace label '{label}'. Namespace labels cannot contain periods ('.')." + ) + params = {"namespace": ".".join(namespace), "key": key} + if refresh_ttl is not None: + params["refresh_ttl"] = refresh_ttl + return await self.http.get("/store/items", params=params, headers=headers) + + async def delete_item( + self, + namespace: Sequence[str], + /, + key: str, + headers: dict[str, str] | None = None, + ) -> None: + """Delete an item. + + Args: + key: The unique identifier for the item. + namespace: Optional list of strings representing the namespace path. + headers: Optional custom headers to include with the request. + + Returns: + None + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + await client.store.delete_item( + ["documents", "user123"], + key="item456", + ) + ``` + """ + await self.http.delete( + "/store/items", + json={"namespace": namespace, "key": key}, + headers=headers, + ) + + async def search_items( + self, + namespace_prefix: Sequence[str], + /, + filter: dict[str, Any] | None = None, + limit: int = 10, + offset: int = 0, + query: str | None = None, + refresh_ttl: bool | None = None, + headers: dict[str, str] | None = None, + ) -> SearchItemsResponse: + """Search for items within a namespace prefix. + + Args: + namespace_prefix: List of strings representing the namespace prefix. + filter: Optional dictionary of key-value pairs to filter results. + 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 for natural language search. + refresh_ttl: Whether to refresh the TTL on items returned by this search. If None, uses the store's default behavior. + headers: Optional custom headers to include with the request. + + Returns: + list[Item]: A list of items matching the search criteria. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + items = await client.store.search_items( + ["documents"], + filter={"author": "John Doe"}, + limit=5, + offset=0 + ) + print(items) + ``` + ```shell + + ---------------------------------------------------------------- + + { + "items": [ + { + "namespace": ["documents", "user123"], + "key": "item789", + "value": { + "title": "Another Document", + "author": "John Doe" + }, + "created_at": "2024-07-30T12:00:00Z", + "updated_at": "2024-07-30T12:00:00Z" + }, + # ... additional items ... + ] + } + ``` + """ + payload = { + "namespace_prefix": namespace_prefix, + "filter": filter, + "limit": limit, + "offset": offset, + "query": query, + "refresh_ttl": refresh_ttl, + } + + return await self.http.post( + "/store/items/search", + json=_provided_vals(payload), + headers=headers, + ) + + async def list_namespaces( + self, + prefix: list[str] | None = None, + suffix: list[str] | None = None, + max_depth: int | None = None, + limit: int = 100, + offset: int = 0, + headers: dict[str, str] | 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. + + Returns: + list[list[str]]: A list of namespaces matching the criteria. + + ???+ example "Example Usage" + + ```python + client = get_client(url="http://localhost:2024") + namespaces = await client.store.list_namespaces( + prefix=["documents"], + max_depth=3, + limit=10, + offset=0 + ) + print(namespaces) + + ---------------------------------------------------------------- + + [ + ["documents", "user123", "reports"], + ["documents", "user456", "invoices"], + ... + ] + ``` + """ + payload = { + "prefix": prefix, + "suffix": suffix, + "max_depth": max_depth, + "limit": limit, + "offset": offset, + } + return await self.http.post( + "/store/namespaces", + json=_provided_vals(payload), + headers=headers, + ) + + +def get_sync_client( + *, + url: str | None = None, + api_key: str | None = None, + headers: dict[str, str] | None = None, + timeout: TimeoutTypes | None = None, +) -> SyncLangGraphClient: + """Get a synchronous LangGraphClient instance. + + 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) + If not provided, defaults to connect=5s, read=300s, write=300s, and pool=5s. + Returns: + SyncLangGraphClient: The top-level synchronous client for accessing AssistantsClient, + ThreadsClient, RunsClient, and CronClient. + + ???+ example "Example" + + ```python + from langgraph_sdk import get_sync_client + + # get top-level synchronous LangGraphClient + client = get_sync_client(url="http://localhost:8123") + + # example usage: client..() + assistant = client.assistants.get(assistant_id="some_uuid") + ``` + """ + + if url is None: + url = "http://localhost:8123" + + transport = httpx.HTTPTransport(retries=5) + client = httpx.Client( + base_url=url, + transport=transport, + timeout=( + httpx.Timeout(timeout) + if timeout is not None + else httpx.Timeout(connect=5, read=300, write=300, pool=5) + ), + headers=_get_headers(api_key, headers), + ) + return SyncLangGraphClient(client) + + +class SyncLangGraphClient: + """Synchronous client for interacting with the LangGraph API. + + This class provides synchronous access to LangGraph API endpoints for managing + assistants, threads, runs, cron jobs, and data storage. + + ???+ example "Example" + + ```python + client = get_sync_client(url="http://localhost:2024") + assistant = client.assistants.get("asst_123") + ``` + """ + + def __init__(self, client: httpx.Client) -> None: + self.http = SyncHttpClient(client) + self.assistants = SyncAssistantsClient(self.http) + self.threads = SyncThreadsClient(self.http) + self.runs = SyncRunsClient(self.http) + self.crons = SyncCronClient(self.http) + self.store = SyncStoreClient(self.http) + + +class SyncHttpClient: + """Handle synchronous requests to the LangGraph API. + + Provides error messaging and content handling enhancements above the + underlying httpx client, mirroring the interface of [HttpClient](#HttpClient) + but for sync usage. + + Attributes: + client (httpx.Client): Underlying HTTPX sync client. + """ + + def __init__(self, client: httpx.Client) -> None: + self.client = client + + def get( + self, + path: str, + *, + params: QueryParamTypes | None = None, + headers: dict[str, str] | None = None, + on_response: Callable[[httpx.Response], None] | None = None, + ) -> Any: + """Send a GET request.""" + r = self.client.get(path, params=params, headers=headers) + if on_response: + on_response(r) + try: + r.raise_for_status() + except httpx.HTTPStatusError as e: + body = r.read().decode() + if sys.version_info >= (3, 11): + e.add_note(body) + else: + logger.error(f"Error from langgraph-api: {body}", exc_info=e) + raise e + return _decode_json(r) + + def post( + self, + path: str, + *, + json: dict | None, + headers: dict[str, str] | None = None, + on_response: Callable[[httpx.Response], None] | None = None, + ) -> Any: + """Send a POST request.""" + if json is not None: + request_headers, content = _encode_json(json) + else: + request_headers, content = {}, b"" + if headers: + request_headers.update(headers) + r = self.client.post(path, headers=request_headers, content=content) + if on_response: + on_response(r) + try: + r.raise_for_status() + except httpx.HTTPStatusError as e: + body = r.read().decode() + if sys.version_info >= (3, 11): + e.add_note(body) + else: + logger.error(f"Error from langgraph-api: {body}", exc_info=e) + raise e + return _decode_json(r) + + def put( + self, + path: str, + *, + json: dict, + headers: dict[str, str] | None = None, + on_response: Callable[[httpx.Response], None] | None = None, + ) -> Any: + """Send a PUT request.""" + request_headers, content = _encode_json(json) + if headers: + request_headers.update(headers) + + r = self.client.put(path, headers=request_headers, content=content) + if on_response: + on_response(r) + try: + r.raise_for_status() + except httpx.HTTPStatusError as e: + body = r.read().decode() + if sys.version_info >= (3, 11): + e.add_note(body) + else: + logger.error(f"Error from langgraph-api: {body}", exc_info=e) + raise e + return _decode_json(r) + + def patch( + self, + path: str, + *, + json: dict, + headers: dict[str, str] | None = None, + on_response: Callable[[httpx.Response], None] | None = None, + ) -> Any: + """Send a PATCH request.""" + request_headers, content = _encode_json(json) + if headers: + request_headers.update(headers) + r = self.client.patch(path, headers=request_headers, content=content) + if on_response: + on_response(r) + try: + r.raise_for_status() + except httpx.HTTPStatusError as e: + body = r.read().decode() + if sys.version_info >= (3, 11): + e.add_note(body) + else: + logger.error(f"Error from langgraph-api: {body}", exc_info=e) + raise e + return _decode_json(r) + + def delete( + self, + path: str, + *, + json: Any | None = None, + headers: dict[str, str] | None = None, + on_response: Callable[[httpx.Response], None] | None = None, + ) -> None: + """Send a DELETE request.""" + r = self.client.request("DELETE", path, json=json, headers=headers) + if on_response: + on_response(r) + try: + r.raise_for_status() + except httpx.HTTPStatusError as e: + body = r.read().decode() + if sys.version_info >= (3, 11): + e.add_note(body) + else: + logger.error(f"Error from langgraph-api: {body}", exc_info=e) + raise e + + def stream( + self, + path: str, + method: str, + *, + json: dict | None = None, + params: QueryParamTypes | None = None, + headers: dict[str, str] | None = None, + on_response: Callable[[httpx.Response], None] | None = None, + ) -> Iterator[StreamPart]: + """Stream the results of a request using SSE.""" + request_headers, content = _encode_json(json) + request_headers["Accept"] = "text/event-stream" + request_headers["Cache-Control"] = "no-store" + if headers: + request_headers.update(headers) + with self.client.stream( + method, path, headers=request_headers, content=content, params=params + ) as res: + if on_response: + on_response(res) + # check status + try: + res.raise_for_status() + except httpx.HTTPStatusError as e: + body = (res.read()).decode() + if sys.version_info >= (3, 11): + e.add_note(body) + else: + logger.error(f"Error from langgraph-api: {body}", exc_info=e) + raise e + # check content type + content_type = res.headers.get("content-type", "").partition(";")[0] + if "text/event-stream" not in content_type: + raise httpx.TransportError( + "Expected response header Content-Type to contain 'text/event-stream', " + f"got {content_type!r}" + ) + # parse SSE + decoder = SSEDecoder() + for line in iter_lines_raw(res): + sse = decoder.decode(line.rstrip(b"\n")) + if sse is not None: + yield sse + + +def _encode_json(json: Any) -> tuple[dict[str, str], bytes]: + body = orjson.dumps( + json, + _orjson_default, + orjson.OPT_SERIALIZE_NUMPY | orjson.OPT_NON_STR_KEYS, + ) + content_length = str(len(body)) + content_type = "application/json" + headers = {"Content-Length": content_length, "Content-Type": content_type} + return headers, body + + +def _decode_json(r: httpx.Response) -> Any: + body = r.read() + return orjson.loads(body) if body else None + + +class SyncAssistantsClient: + """Client for managing assistants in LangGraph synchronously. + + This class provides methods to interact with assistants, which are versioned configurations of your graph. + + ???+ example "Examples" + + ```python + client = get_sync_client(url="http://localhost:2024") + assistant = client.assistants.get("assistant_id_123") + ``` + """ + + def __init__(self, http: SyncHttpClient) -> None: + self.http = http + + def get( + self, + assistant_id: str, + *, + headers: dict[str, str] | None = None, + ) -> Assistant: + """Get an assistant by ID. + + Args: + assistant_id: The ID of the assistant to get OR the name of the graph (to use the default assistant). + headers: Optional custom headers to include with the request. + + Returns: + Assistant: Assistant Object. + + ???+ example "Example Usage" + + ```python + assistant = client.assistants.get( + assistant_id="my_assistant_id" + ) + print(assistant) + ``` + + ```shell + ---------------------------------------------------- + + { + 'assistant_id': 'my_assistant_id', + 'graph_id': 'agent', + 'created_at': '2024-06-25T17:10:33.109781+00:00', + 'updated_at': '2024-06-25T17:10:33.109781+00:00', + 'config': {}, + 'metadata': {'created_by': 'system'} + } + ``` + + """ # noqa: E501 + return self.http.get(f"/assistants/{assistant_id}", headers=headers) + + def get_graph( + self, + assistant_id: str, + *, + xray: int | bool = False, + headers: dict[str, str] | 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. + + Returns: + Graph: The graph information for the assistant in JSON format. + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:2024") + graph_info = client.assistants.get_graph( + assistant_id="my_assistant_id" + ) + print(graph_info) + + -------------------------------------------------------------------------------------------------------------------------- + + { + 'nodes': + [ + {'id': '__start__', 'type': 'schema', 'data': '__start__'}, + {'id': '__end__', 'type': 'schema', 'data': '__end__'}, + {'id': 'agent','type': 'runnable','data': {'id': ['langgraph', 'utils', 'RunnableCallable'],'name': 'agent'}}, + ], + 'edges': + [ + {'source': '__start__', 'target': 'agent'}, + {'source': 'agent','target': '__end__'} + ] + } + ``` + + """ # noqa: E501 + return self.http.get( + f"/assistants/{assistant_id}/graph", params={"xray": xray}, headers=headers + ) + + def get_schemas( + self, + assistant_id: str, + *, + headers: dict[str, str] | None = None, + ) -> GraphSchema: + """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. + + Returns: + GraphSchema: The graph schema for the assistant. + + ???+ example " Example Usage" + + ```python + client = get_sync_client(url="http://localhost:2024") + schema = client.assistants.get_schemas( + assistant_id="my_assistant_id" + ) + print(schema) + ``` + ```shell + ---------------------------------------------------------------------------------------------------------------------------- + + { + 'graph_id': 'agent', + 'state_schema': + { + 'title': 'LangGraphInput', + '$ref': '#/definitions/AgentState', + 'definitions': + { + 'BaseMessage': + { + 'title': 'BaseMessage', + 'description': 'Base abstract Message class. Messages are the inputs and outputs of ChatModels.', + 'type': 'object', + 'properties': + { + 'content': + { + 'title': 'Content', + 'anyOf': [ + {'type': 'string'}, + {'type': 'array','items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]}} + ] + }, + 'additional_kwargs': + { + 'title': 'Additional Kwargs', + 'type': 'object' + }, + 'response_metadata': + { + 'title': 'Response Metadata', + 'type': 'object' + }, + 'type': + { + 'title': 'Type', + 'type': 'string' + }, + 'name': + { + 'title': 'Name', + 'type': 'string' + }, + 'id': + { + 'title': 'Id', + 'type': 'string' + } + }, + 'required': ['content', 'type'] + }, + 'AgentState': + { + 'title': 'AgentState', + 'type': 'object', + 'properties': + { + 'messages': + { + 'title': 'Messages', + 'type': 'array', + 'items': {'$ref': '#/definitions/BaseMessage'} + } + }, + 'required': ['messages'] + } + } + }, + 'config_schema': + { + 'title': 'Configurable', + 'type': 'object', + 'properties': + { + 'model_name': + { + 'title': 'Model Name', + 'enum': ['anthropic', 'openai'], + 'type': 'string' + } + } + } + } + ``` + + """ # noqa: E501 + return self.http.get(f"/assistants/{assistant_id}/schemas", headers=headers) + + def get_subgraphs( + self, + assistant_id: str, + namespace: str | None = None, + recurse: bool = False, + *, + headers: dict[str, str] | None = None, + ) -> Subgraphs: + """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. + + Returns: + Subgraphs: The graph schema for the assistant. + + """ # noqa: E501 + if namespace is not None: + return self.http.get( + f"/assistants/{assistant_id}/subgraphs/{namespace}", + params={"recurse": recurse}, + headers=headers, + ) + else: + return self.http.get( + f"/assistants/{assistant_id}/subgraphs", + params={"recurse": recurse}, + headers=headers, + ) + + def create( + self, + graph_id: str | None, + config: Config | None = None, + *, + metadata: Json = None, + assistant_id: str | None = None, + if_exists: OnConflictBehavior | None = None, + name: str | None = None, + headers: dict[str, str] | None = None, + description: str | None = None, + ) -> Assistant: + """Create a new assistant. + + Useful when graph is configurable and you want to create different assistants based on different configurations. + + Args: + graph_id: The ID of the graph the assistant should use. The graph ID is normally set in your langgraph.json configuration. + config: Configuration to use for the graph. + metadata: Metadata to add to assistant. + assistant_id: Assistant ID to use, will default to a random UUID if not provided. + if_exists: How to handle duplicate creation. Defaults to 'raise' under the hood. + Must be either 'raise' (raise error if duplicate), or 'do_nothing' (return existing assistant). + name: The name of the assistant. Defaults to 'Untitled' under the hood. + headers: Optional custom headers to include with the request. + description: Optional description of the assistant. + The description field is available for langgraph-api server version>=0.0.45 + + Returns: + Assistant: The created assistant. + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:2024") + assistant = client.assistants.create( + graph_id="agent", + config={"configurable": {"model_name": "openai"}}, + metadata={"number":1}, + assistant_id="my-assistant-id", + if_exists="do_nothing", + name="my_name" + ) + ``` + """ # noqa: E501 + payload: dict[str, Any] = { + "graph_id": graph_id, + } + if config: + payload["config"] = config + if metadata: + payload["metadata"] = metadata + if assistant_id: + payload["assistant_id"] = assistant_id + if if_exists: + payload["if_exists"] = if_exists + if name: + payload["name"] = name + if description: + payload["description"] = description + return self.http.post("/assistants", json=payload, headers=headers) + + def update( + self, + assistant_id: str, + *, + graph_id: str | None = None, + config: Config | None = None, + metadata: Json = None, + name: str | None = None, + headers: dict[str, str] | None = None, + description: str | None = None, + ) -> Assistant: + """Update an assistant. + + Use this to point to a different graph, update the configuration, or change the metadata of an assistant. + + Args: + assistant_id: Assistant to update. + graph_id: The ID of the graph the assistant should use. + The graph ID is normally set in your langgraph.json configuration. If None, assistant will keep pointing to same graph. + config: Configuration to use for the graph. + metadata: Metadata to merge with existing assistant metadata. + name: The new name for the assistant. + headers: Optional custom headers to include with the request. + description: Optional description of the assistant. + The description field is available for langgraph-api server version>=0.0.45 + + Returns: + Assistant: The updated assistant. + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:2024") + assistant = client.assistants.update( + assistant_id='e280dad7-8618-443f-87f1-8e41841c180f', + graph_id="other-graph", + config={"configurable": {"model_name": "anthropic"}}, + metadata={"number":2} + ) + ``` + """ # noqa: E501 + payload: dict[str, Any] = {} + if graph_id: + payload["graph_id"] = graph_id + if config: + payload["config"] = config + if metadata: + payload["metadata"] = metadata + if name: + payload["name"] = name + if description: + payload["description"] = description + return self.http.patch( + f"/assistants/{assistant_id}", + json=payload, + headers=headers, + ) + + def delete( + self, + assistant_id: str, + *, + headers: dict[str, str] | None = None, + ) -> None: + """Delete an assistant. + + Args: + assistant_id: The assistant ID to delete. + headers: Optional custom headers to include with the request. + + Returns: + None + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:2024") + client.assistants.delete( + assistant_id="my_assistant_id" + ) + ``` + + """ # noqa: E501 + self.http.delete(f"/assistants/{assistant_id}", headers=headers) + + def search( + self, + *, + metadata: Json = None, + graph_id: str | None = None, + limit: int = 10, + offset: int = 0, + sort_by: AssistantSortBy | None = None, + sort_order: SortOrder | None = None, + headers: dict[str, str] | None = None, + ) -> list[Assistant]: + """Search for assistants. + + Args: + metadata: Metadata to filter by. Exact match filter for each KV pair. + graph_id: The ID of the graph to filter by. + The graph ID is normally set in your langgraph.json configuration. + limit: The maximum number of results to return. + offset: The number of results to skip. + headers: Optional custom headers to include with the request. + + Returns: + list[Assistant]: A list of assistants. + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:2024") + assistants = client.assistants.search( + metadata = {"name":"my_name"}, + graph_id="my_graph_id", + limit=5, + offset=5 + ) + ``` + """ + payload: dict[str, Any] = { + "limit": limit, + "offset": offset, + } + if metadata: + payload["metadata"] = metadata + if graph_id: + payload["graph_id"] = graph_id + if sort_by: + payload["sort_by"] = sort_by + if sort_order: + payload["sort_order"] = sort_order + return self.http.post( + "/assistants/search", + json=payload, + headers=headers, + ) + + def get_versions( + self, + assistant_id: str, + metadata: Json = None, + limit: int = 10, + offset: int = 0, + *, + headers: dict[str, str] | 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. + + Returns: + list[Assistant]: A list of assistants. + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:2024") + assistant_versions = client.assistants.get_versions( + assistant_id="my_assistant_id" + ) + ``` + + """ # noqa: E501 + + payload: dict[str, Any] = { + "limit": limit, + "offset": offset, + } + if metadata: + payload["metadata"] = metadata + return self.http.post( + f"/assistants/{assistant_id}/versions", json=payload, headers=headers + ) + + def set_latest( + self, + assistant_id: str, + version: int, + *, + headers: dict[str, str] | None = None, + ) -> Assistant: + """Change the version of an assistant. + + Args: + assistant_id: The assistant ID to delete. + version: The version to change to. + headers: Optional custom headers to include with the request. + + Returns: + Assistant: Assistant Object. + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:2024") + new_version_assistant = client.assistants.set_latest( + assistant_id="my_assistant_id", + version=3 + ) + ``` + + """ # noqa: E501 + + payload: dict[str, Any] = {"version": version} + + return self.http.post( + f"/assistants/{assistant_id}/latest", json=payload, headers=headers + ) + + +class SyncThreadsClient: + """Synchronous client for managing threads in LangGraph. + + This class provides methods to create, retrieve, and manage threads, + which represent conversations or stateful interactions. + + ???+ example "Example" + + ```python + client = get_sync_client(url="http://localhost:2024") + thread = client.threads.create(metadata={"user_id": "123"}) + ``` + """ + + def __init__(self, http: SyncHttpClient) -> None: + self.http = http + + def get( + self, + thread_id: str, + *, + headers: dict[str, str] | None = None, + ) -> Thread: + """Get a thread by ID. + + Args: + thread_id: The ID of the thread to get. + headers: Optional custom headers to include with the request. + + Returns: + Thread: Thread object. + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:2024") + thread = client.threads.get( + thread_id="my_thread_id" + ) + print(thread) + ``` + ```shell + ----------------------------------------------------- + + { + 'thread_id': 'my_thread_id', + 'created_at': '2024-07-18T18:35:15.540834+00:00', + 'updated_at': '2024-07-18T18:35:15.540834+00:00', + 'metadata': {'graph_id': 'agent'} + } + ``` + + """ # noqa: E501 + + return self.http.get(f"/threads/{thread_id}", headers=headers) + + def create( + self, + *, + metadata: Json = None, + thread_id: str | None = None, + if_exists: OnConflictBehavior | None = None, + supersteps: Sequence[dict[str, Sequence[dict[str, Any]]]] | None = None, + graph_id: str | None = None, + headers: dict[str, str] | None = None, + ) -> Thread: + """Create a new thread. + + Args: + metadata: Metadata to add to thread. + thread_id: ID of thread. + If None, ID will be a randomly generated UUID. + if_exists: How to handle duplicate creation. Defaults to 'raise' under the hood. + Must be either 'raise' (raise error if duplicate), or 'do_nothing' (return existing thread). + supersteps: Apply a list of supersteps when creating a thread, each containing a sequence of updates. + Each update has `values` or `command` and `as_node`. Used for copying a thread between deployments. + graph_id: Optional graph ID to associate with the thread. + headers: Optional custom headers to include with the request. + + Returns: + Thread: The created thread. + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:2024") + thread = client.threads.create( + metadata={"number":1}, + thread_id="my-thread-id", + if_exists="raise" + ) + ``` + ) + """ # noqa: E501 + payload: dict[str, Any] = {} + if thread_id: + payload["thread_id"] = thread_id + if metadata or graph_id: + payload["metadata"] = { + **(metadata or {}), + **({"graph_id": graph_id} if graph_id else {}), + } + if if_exists: + payload["if_exists"] = if_exists + if supersteps: + payload["supersteps"] = [ + { + "updates": [ + { + "values": u["values"], + "command": u.get("command"), + "as_node": u["as_node"], + } + for u in s["updates"] + ] + } + for s in supersteps + ] + + return self.http.post("/threads", json=payload, headers=headers) + + def update( + self, + thread_id: str, + *, + metadata: dict[str, Any], + headers: dict[str, str] | None = None, + ) -> Thread: + """Update a thread. + + Args: + thread_id: ID of thread to update. + metadata: Metadata to merge with existing thread metadata. + headers: Optional custom headers to include with the request. + + Returns: + Thread: The created thread. + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:2024") + thread = client.threads.update( + thread_id="my-thread-id", + metadata={"number":1}, + ) + ``` + """ # noqa: E501 + return self.http.patch( + f"/threads/{thread_id}", json={"metadata": metadata}, headers=headers + ) + + def delete( + self, + thread_id: str, + *, + headers: dict[str, str] | None = None, + ) -> None: + """Delete a thread. + + Args: + thread_id: The ID of the thread to delete. + headers: Optional custom headers to include with the request. + + Returns: + None + + ???+ example "Example Usage" + + ```python + client.threads.delete( + thread_id="my_thread_id" + ) + ``` + + """ # noqa: E501 + self.http.delete(f"/threads/{thread_id}", headers=headers) + + def search( + self, + *, + metadata: Json = None, + values: Json = None, + status: ThreadStatus | None = None, + limit: int = 10, + offset: int = 0, + sort_by: ThreadSortBy | None = None, + sort_order: SortOrder | None = None, + headers: dict[str, str] | None = None, + ) -> list[Thread]: + """Search for threads. + + Args: + metadata: Thread metadata to filter on. + values: State values to filter on. + status: Thread status to filter on. + Must be one of 'idle', 'busy', 'interrupted' or 'error'. + limit: Limit on number of threads to return. + offset: Offset in threads table to start search from. + headers: Optional custom headers to include with the request. + + Returns: + list[Thread]: List of the threads matching the search parameters. + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:2024") + threads = client.threads.search( + metadata={"number":1}, + status="interrupted", + limit=15, + offset=5 + ) + ``` + """ # noqa: E501 + payload: dict[str, Any] = { + "limit": limit, + "offset": offset, + } + if metadata: + payload["metadata"] = metadata + if values: + payload["values"] = values + if status: + payload["status"] = status + if sort_by: + payload["sort_by"] = sort_by + if sort_order: + payload["sort_order"] = sort_order + return self.http.post("/threads/search", json=payload, headers=headers) + + def copy( + self, + thread_id: str, + *, + headers: dict[str, str] | None = None, + ) -> None: + """Copy a thread. + + Args: + thread_id: The ID of the thread to copy. + headers: Optional custom headers to include with the request. + + Returns: + None + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:2024") + client.threads.copy( + thread_id="my_thread_id" + ) + ``` + + """ # noqa: E501 + return self.http.post(f"/threads/{thread_id}/copy", json=None, headers=headers) + + def get_state( + self, + thread_id: str, + checkpoint: Checkpoint | None = None, + checkpoint_id: str | None = None, # deprecated + *, + subgraphs: bool = False, + headers: dict[str, str] | None = None, + ) -> ThreadState: + """Get the state of a thread. + + Args: + thread_id: The ID of the thread to get the state of. + checkpoint: The checkpoint to get the state of. + subgraphs: Include subgraphs states. + headers: Optional custom headers to include with the request. + + Returns: + ThreadState: the thread of the state. + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:2024") + thread_state = client.threads.get_state( + thread_id="my_thread_id", + checkpoint_id="my_checkpoint_id" + ) + print(thread_state) + ``` + + ```shell + ---------------------------------------------------------------------------------------------------------------------------------------------------------------------- + + { + 'values': { + 'messages': [ + { + 'content': 'how are you?', + 'additional_kwargs': {}, + 'response_metadata': {}, + 'type': 'human', + 'name': None, + 'id': 'fe0a5778-cfe9-42ee-b807-0adaa1873c10', + 'example': False + }, + { + 'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.", + 'additional_kwargs': {}, + 'response_metadata': {}, + 'type': 'ai', + 'name': None, + 'id': 'run-159b782c-b679-4830-83c6-cef87798fe8b', + 'example': False, + 'tool_calls': [], + 'invalid_tool_calls': [], + 'usage_metadata': None + } + ] + }, + 'next': [], + 'checkpoint': + { + 'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2', + 'checkpoint_ns': '', + 'checkpoint_id': '1ef4a9b8-e6fb-67b1-8001-abd5184439d1' + } + 'metadata': + { + 'step': 1, + 'run_id': '1ef4a9b8-d7da-679a-a45a-872054341df2', + 'source': 'loop', + 'writes': + { + 'agent': + { + 'messages': [ + { + 'id': 'run-159b782c-b679-4830-83c6-cef87798fe8b', + 'name': None, + 'type': 'ai', + 'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.", + 'example': False, + 'tool_calls': [], + 'usage_metadata': None, + 'additional_kwargs': {}, + 'response_metadata': {}, + 'invalid_tool_calls': [] + } + ] + } + }, + 'user_id': None, + 'graph_id': 'agent', + 'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2', + 'created_by': 'system', + 'assistant_id': 'fe096781-5601-53d2-b2f6-0d3403f7e9ca'}, + 'created_at': '2024-07-25T15:35:44.184703+00:00', + 'parent_config': + { + 'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2', + 'checkpoint_ns': '', + 'checkpoint_id': '1ef4a9b8-d80d-6fa7-8000-9300467fad0f' + } + } + ``` + + """ # noqa: E501 + if checkpoint: + return self.http.post( + f"/threads/{thread_id}/state/checkpoint", + json={"checkpoint": checkpoint, "subgraphs": subgraphs}, + headers=headers, + ) + elif checkpoint_id: + return self.http.get( + f"/threads/{thread_id}/state/{checkpoint_id}", + params={"subgraphs": subgraphs}, + headers=headers, + ) + else: + return self.http.get( + f"/threads/{thread_id}/state", + params={"subgraphs": subgraphs}, + headers=headers, + ) + + def update_state( + self, + thread_id: str, + values: dict | Sequence[dict] | None, + *, + as_node: str | None = None, + checkpoint: Checkpoint | None = None, + checkpoint_id: str | None = None, # deprecated + headers: dict[str, str] | None = None, + ) -> ThreadUpdateStateResponse: + """Update the state of a thread. + + Args: + thread_id: The ID of the thread to update. + values: The values to update the state with. + as_node: Update the state as if this node had just executed. + checkpoint: The checkpoint to update the state of. + headers: Optional custom headers to include with the request. + + Returns: + ThreadUpdateStateResponse: Response after updating a thread's state. + + ???+ example "Example Usage" + + ```python + + response = await client.threads.update_state( + thread_id="my_thread_id", + values={"messages":[{"role": "user", "content": "hello!"}]}, + as_node="my_node", + ) + print(response) + + ---------------------------------------------------------------------------------------------------------------------------------------------------------------------- + + { + 'checkpoint': { + 'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2', + 'checkpoint_ns': '', + 'checkpoint_id': '1ef4a9b8-e6fb-67b1-8001-abd5184439d1', + 'checkpoint_map': {} + } + } + ``` + + """ # noqa: E501 + payload: dict[str, Any] = { + "values": values, + } + if checkpoint_id: + payload["checkpoint_id"] = checkpoint_id + if checkpoint: + payload["checkpoint"] = checkpoint + if as_node: + payload["as_node"] = as_node + return self.http.post( + f"/threads/{thread_id}/state", json=payload, headers=headers + ) + + def get_history( + self, + thread_id: str, + *, + limit: int = 10, + before: str | Checkpoint | None = None, + metadata: dict | None = None, + checkpoint: Checkpoint | None = None, + headers: dict[str, str] | None = None, + ) -> list[ThreadState]: + """Get the state history of a thread. + + Args: + thread_id: The ID of the thread to get the state history for. + checkpoint: Return states for this subgraph. If empty defaults to root. + limit: The maximum number of states to return. + before: Return states before this checkpoint. + metadata: Filter states by metadata key-value pairs. + headers: Optional custom headers to include with the request. + + Returns: + list[ThreadState]: the state history of the thread. + + ???+ example "Example Usage" + + ```python + + thread_state = client.threads.get_history( + thread_id="my_thread_id", + limit=5, + before="my_timestamp", + metadata={"name":"my_name"} + ) + ``` + + """ # noqa: E501 + payload: dict[str, Any] = { + "limit": limit, + } + if before: + payload["before"] = before + if metadata: + payload["metadata"] = metadata + if checkpoint: + payload["checkpoint"] = checkpoint + return self.http.post( + f"/threads/{thread_id}/history", json=payload, headers=headers + ) + + +class SyncRunsClient: + """Synchronous client for managing runs in LangGraph. + + This class provides methods to create, retrieve, and manage runs, which represent + individual executions of graphs. + + ???+ example "Example" + + ```python + client = get_sync_client(url="http://localhost:2024") + run = client.runs.create(thread_id="thread_123", assistant_id="asst_456") + ``` + """ + + def __init__(self, http: SyncHttpClient) -> None: + self.http = http + + @overload + def stream( + self, + thread_id: str, + assistant_id: str, + *, + input: dict | None = None, + command: Command | None = None, + stream_mode: StreamMode | Sequence[StreamMode] = "values", + stream_subgraphs: bool = False, + metadata: dict | None = None, + config: Config | None = None, + checkpoint: Checkpoint | None = None, + checkpoint_id: str | None = None, + checkpoint_during: bool | None = None, + interrupt_before: All | Sequence[str] | None = None, + interrupt_after: All | Sequence[str] | None = None, + feedback_keys: Sequence[str] | None = None, + on_disconnect: DisconnectMode | None = None, + webhook: str | None = None, + multitask_strategy: MultitaskStrategy | None = None, + if_not_exists: IfNotExists | None = None, + after_seconds: int | None = None, + headers: dict[str, str] | None = None, + on_run_created: Callable[[RunCreateMetadata], None] | None = None, + ) -> Iterator[StreamPart]: ... + + @overload + def stream( + self, + thread_id: None, + assistant_id: str, + *, + input: dict | None = None, + command: Command | None = None, + stream_mode: StreamMode | Sequence[StreamMode] = "values", + stream_subgraphs: bool = False, + stream_resumable: bool = False, + metadata: dict | None = None, + config: Config | None = None, + checkpoint_during: bool | None = None, + interrupt_before: All | Sequence[str] | None = None, + interrupt_after: All | Sequence[str] | None = None, + feedback_keys: Sequence[str] | None = None, + on_disconnect: DisconnectMode | None = None, + on_completion: OnCompletionBehavior | None = None, + if_not_exists: IfNotExists | None = None, + webhook: str | None = None, + after_seconds: int | None = None, + headers: dict[str, str] | None = None, + on_run_created: Callable[[RunCreateMetadata], None] | None = None, + ) -> Iterator[StreamPart]: ... + + def stream( + self, + thread_id: str | None, + assistant_id: str, + *, + input: dict | None = None, + command: Command | None = None, + stream_mode: StreamMode | Sequence[StreamMode] = "values", + stream_subgraphs: bool = False, + stream_resumable: bool = False, + metadata: dict | None = None, + config: Config | None = None, + checkpoint: Checkpoint | None = None, + checkpoint_id: str | None = None, + checkpoint_during: bool | None = None, + interrupt_before: All | Sequence[str] | None = None, + interrupt_after: All | Sequence[str] | None = None, + feedback_keys: Sequence[str] | None = None, + on_disconnect: DisconnectMode | None = None, + on_completion: OnCompletionBehavior | None = None, + webhook: str | None = None, + multitask_strategy: MultitaskStrategy | None = None, + if_not_exists: IfNotExists | None = None, + after_seconds: int | None = None, + headers: dict[str, str] | None = None, + on_run_created: Callable[[RunCreateMetadata], None] | None = None, + ) -> Iterator[StreamPart]: + """Create a run and stream the results. + + Args: + thread_id: the thread ID to assign to the thread. + If None will create a stateless run. + assistant_id: The assistant ID or graph name to stream from. + If using graph name, will default to first assistant created from that graph. + input: The input to the graph. + command: The command to execute. + stream_mode: The stream mode(s) to use. + stream_subgraphs: Whether to stream output from subgraphs. + stream_resumable: Whether the stream is considered resumable. + If true, the stream can be resumed and replayed in its entirety even after disconnection. + metadata: Metadata to assign to the run. + config: The configuration for the assistant. + checkpoint: The checkpoint to resume from. + checkpoint_during: Whether to checkpoint during the run (or only at the end/interruption). + interrupt_before: Nodes to interrupt immediately before they get executed. + interrupt_after: Nodes to Nodes to interrupt immediately after they get executed. + feedback_keys: Feedback keys to assign to run. + on_disconnect: The disconnect mode to use. + Must be one of 'cancel' or 'continue'. + on_completion: Whether to delete or keep the thread created for a stateless run. + Must be one of 'delete' or 'keep'. + webhook: Webhook to call after LangGraph API call is done. + multitask_strategy: Multitask strategy to use. + Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + if_not_exists: How to handle missing thread. Defaults to 'reject'. + Must be either 'reject' (raise error if missing), or 'create' (create new thread). + after_seconds: The number of seconds to wait before starting the run. + Use to schedule future runs. + headers: Optional custom headers to include with the request. + on_run_created: Optional callback to call when a run is created. + + Returns: + Iterator[StreamPart]: Iterator of stream results. + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:2024") + async for chunk in client.runs.stream( + thread_id=None, + assistant_id="agent", + input={"messages": [{"role": "user", "content": "how are you?"}]}, + stream_mode=["values","debug"], + metadata={"name":"my_run"}, + config={"configurable": {"model_name": "anthropic"}}, + interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"], + interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"], + feedback_keys=["my_feedback_key_1","my_feedback_key_2"], + webhook="https://my.fake.webhook.com", + multitask_strategy="interrupt" + ): + print(chunk) + ``` + ```shell + ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ + + StreamPart(event='metadata', data={'run_id': '1ef4a9b8-d7da-679a-a45a-872054341df2'}) + StreamPart(event='values', data={'messages': [{'content': 'how are you?', 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'human', 'name': None, 'id': 'fe0a5778-cfe9-42ee-b807-0adaa1873c10', 'example': False}]}) + StreamPart(event='values', data={'messages': [{'content': 'how are you?', 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'human', 'name': None, 'id': 'fe0a5778-cfe9-42ee-b807-0adaa1873c10', 'example': False}, {'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.", 'additional_kwargs': {}, 'response_metadata': {}, 'type': 'ai', 'name': None, 'id': 'run-159b782c-b679-4830-83c6-cef87798fe8b', 'example': False, 'tool_calls': [], 'invalid_tool_calls': [], 'usage_metadata': None}]}) + StreamPart(event='end', data=None) + ``` + """ # noqa: E501 + payload = { + "input": input, + "command": ( + {k: v for k, v in command.items() if v is not None} if command else None + ), + "config": config, + "metadata": metadata, + "stream_mode": stream_mode, + "stream_subgraphs": stream_subgraphs, + "stream_resumable": stream_resumable, + "assistant_id": assistant_id, + "interrupt_before": interrupt_before, + "interrupt_after": interrupt_after, + "feedback_keys": feedback_keys, + "webhook": webhook, + "checkpoint": checkpoint, + "checkpoint_id": checkpoint_id, + "checkpoint_during": checkpoint_during, + "multitask_strategy": multitask_strategy, + "if_not_exists": if_not_exists, + "on_disconnect": on_disconnect, + "on_completion": on_completion, + "after_seconds": after_seconds, + } + endpoint = ( + f"/threads/{thread_id}/runs/stream" + if thread_id is not None + else "/runs/stream" + ) + + def on_response(res: httpx.Response): + """Callback function to handle the response.""" + if on_run_created and (metadata := _get_run_metadata_from_response(res)): + on_run_created(metadata) + + return self.http.stream( + endpoint, + "POST", + json={k: v for k, v in payload.items() if v is not None}, + headers=headers, + on_response=on_response if on_run_created else None, + ) + + @overload + def create( + self, + thread_id: None, + assistant_id: str, + *, + input: dict | None = None, + command: Command | None = None, + stream_mode: StreamMode | Sequence[StreamMode] = "values", + stream_subgraphs: bool = False, + stream_resumable: bool = False, + metadata: dict | None = None, + config: Config | None = None, + checkpoint_during: bool | None = None, + interrupt_before: All | Sequence[str] | None = None, + interrupt_after: All | Sequence[str] | None = None, + webhook: str | None = None, + on_completion: OnCompletionBehavior | None = None, + if_not_exists: IfNotExists | None = None, + after_seconds: int | None = None, + headers: dict[str, str] | None = None, + on_run_created: Callable[[RunCreateMetadata], None] | None = None, + ) -> Run: ... + + @overload + def create( + self, + thread_id: str, + assistant_id: str, + *, + input: dict | None = None, + command: Command | None = None, + stream_mode: StreamMode | Sequence[StreamMode] = "values", + stream_subgraphs: bool = False, + stream_resumable: bool = False, + metadata: dict | None = None, + config: Config | None = None, + checkpoint: Checkpoint | None = None, + checkpoint_id: str | None = None, + checkpoint_during: bool | None = None, + interrupt_before: All | Sequence[str] | None = None, + interrupt_after: All | Sequence[str] | None = None, + webhook: str | None = None, + multitask_strategy: MultitaskStrategy | None = None, + if_not_exists: IfNotExists | None = None, + after_seconds: int | None = None, + headers: dict[str, str] | None = None, + on_run_created: Callable[[RunCreateMetadata], None] | None = None, + ) -> Run: ... + + def create( + self, + thread_id: str | None, + assistant_id: str, + *, + input: dict | None = None, + command: Command | None = None, + stream_mode: StreamMode | Sequence[StreamMode] = "values", + stream_subgraphs: bool = False, + stream_resumable: bool = False, + metadata: dict | None = None, + config: Config | None = None, + checkpoint: Checkpoint | None = None, + checkpoint_id: str | None = None, + checkpoint_during: bool | None = None, + interrupt_before: All | Sequence[str] | None = None, + interrupt_after: All | Sequence[str] | None = None, + webhook: str | None = None, + multitask_strategy: MultitaskStrategy | None = None, + on_completion: OnCompletionBehavior | None = None, + if_not_exists: IfNotExists | None = None, + after_seconds: int | None = None, + headers: dict[str, str] | None = None, + on_run_created: Callable[[RunCreateMetadata], None] | None = None, + ) -> Run: + """Create a background run. + + Args: + thread_id: the thread ID to assign to the thread. + If None will create a stateless run. + assistant_id: The assistant ID or graph name to stream from. + If using graph name, will default to first assistant created from that graph. + input: The input to the graph. + command: The command to execute. + stream_mode: The stream mode(s) to use. + stream_subgraphs: Whether to stream output from subgraphs. + stream_resumable: Whether the stream is considered resumable. + If true, the stream can be resumed and replayed in its entirety even after disconnection. + metadata: Metadata to assign to the run. + config: The configuration for the assistant. + checkpoint: The checkpoint to resume from. + checkpoint_during: Whether to checkpoint during the run (or only at the end/interruption). + interrupt_before: Nodes to interrupt immediately before they get executed. + interrupt_after: Nodes to Nodes to interrupt immediately after they get executed. + webhook: Webhook to call after LangGraph API call is done. + multitask_strategy: Multitask strategy to use. + Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + on_completion: Whether to delete or keep the thread created for a stateless run. + Must be one of 'delete' or 'keep'. + if_not_exists: How to handle missing thread. Defaults to 'reject'. + Must be either 'reject' (raise error if missing), or 'create' (create new thread). + after_seconds: The number of seconds to wait before starting the run. + Use to schedule future runs. + headers: Optional custom headers to include with the request. + on_run_created: Optional callback to call when a run is created. + + Returns: + Run: The created background run. + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:2024") + background_run = client.runs.create( + thread_id="my_thread_id", + assistant_id="my_assistant_id", + input={"messages": [{"role": "user", "content": "hello!"}]}, + metadata={"name":"my_run"}, + config={"configurable": {"model_name": "openai"}}, + interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"], + interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"], + webhook="https://my.fake.webhook.com", + multitask_strategy="interrupt" + ) + print(background_run) + ``` + + ```shell + -------------------------------------------------------------------------------- + + { + 'run_id': 'my_run_id', + 'thread_id': 'my_thread_id', + 'assistant_id': 'my_assistant_id', + 'created_at': '2024-07-25T15:35:42.598503+00:00', + 'updated_at': '2024-07-25T15:35:42.598503+00:00', + 'metadata': {}, + 'status': 'pending', + 'kwargs': + { + 'input': + { + 'messages': [ + { + 'role': 'user', + 'content': 'how are you?' + } + ] + }, + 'config': + { + 'metadata': + { + 'created_by': 'system' + }, + 'configurable': + { + 'run_id': 'my_run_id', + 'user_id': None, + 'graph_id': 'agent', + 'thread_id': 'my_thread_id', + 'checkpoint_id': None, + 'model_name': "openai", + 'assistant_id': 'my_assistant_id' + } + }, + 'webhook': "https://my.fake.webhook.com", + 'temporary': False, + 'stream_mode': ['values'], + 'feedback_keys': None, + 'interrupt_after': ["node_to_stop_after_1","node_to_stop_after_2"], + 'interrupt_before': ["node_to_stop_before_1","node_to_stop_before_2"] + }, + 'multitask_strategy': 'interrupt' + } + ``` + """ # noqa: E501 + payload = { + "input": input, + "command": ( + {k: v for k, v in command.items() if v is not None} if command else None + ), + "stream_mode": stream_mode, + "stream_subgraphs": stream_subgraphs, + "stream_resumable": stream_resumable, + "config": config, + "metadata": metadata, + "assistant_id": assistant_id, + "interrupt_before": interrupt_before, + "interrupt_after": interrupt_after, + "webhook": webhook, + "checkpoint": checkpoint, + "checkpoint_id": checkpoint_id, + "checkpoint_during": checkpoint_during, + "multitask_strategy": multitask_strategy, + "if_not_exists": if_not_exists, + "on_completion": on_completion, + "after_seconds": after_seconds, + } + payload = {k: v for k, v in payload.items() if v is not None} + + def on_response(res: httpx.Response): + """Callback function to handle the response.""" + if on_run_created and (metadata := _get_run_metadata_from_response(res)): + on_run_created(metadata) + + return self.http.post( + f"/threads/{thread_id}/runs" if thread_id else "/runs", + json=payload, + headers=headers, + on_response=on_response if on_run_created else None, + ) + + def create_batch( + self, payloads: list[RunCreate], *, headers: dict[str, str] | None = None + ) -> list[Run]: + """Create a batch of stateless background runs.""" + + def filter_payload(payload: RunCreate): + return {k: v for k, v in payload.items() if v is not None} + + payloads = [filter_payload(payload) for payload in payloads] + return self.http.post("/runs/batch", json=payloads, headers=headers) + + @overload + def wait( + self, + thread_id: str, + assistant_id: str, + *, + input: dict | None = None, + command: Command | None = None, + metadata: dict | None = None, + config: Config | None = None, + checkpoint: Checkpoint | None = None, + checkpoint_id: str | None = None, + checkpoint_during: bool | None = None, + interrupt_before: All | Sequence[str] | None = None, + interrupt_after: All | Sequence[str] | None = None, + webhook: str | None = None, + on_disconnect: DisconnectMode | None = None, + multitask_strategy: MultitaskStrategy | None = None, + if_not_exists: IfNotExists | None = None, + after_seconds: int | None = None, + headers: dict[str, str] | None = None, + on_run_created: Callable[[RunCreateMetadata], None] | None = None, + ) -> list[dict] | dict[str, Any]: ... + + @overload + def wait( + self, + thread_id: None, + assistant_id: str, + *, + input: dict | None = None, + command: Command | None = None, + metadata: dict | None = None, + config: Config | None = None, + checkpoint_during: bool | None = None, + interrupt_before: All | Sequence[str] | None = None, + interrupt_after: All | Sequence[str] | None = None, + webhook: str | None = None, + on_disconnect: DisconnectMode | None = None, + on_completion: OnCompletionBehavior | None = None, + if_not_exists: IfNotExists | None = None, + after_seconds: int | None = None, + headers: dict[str, str] | None = None, + on_run_created: Callable[[RunCreateMetadata], None] | None = None, + ) -> list[dict] | dict[str, Any]: ... + + def wait( + self, + thread_id: str | None, + assistant_id: str, + *, + input: dict | None = None, + command: Command | None = None, + metadata: dict | None = None, + config: Config | None = None, + checkpoint_during: bool | None = None, + checkpoint: Checkpoint | None = None, + checkpoint_id: str | None = None, + interrupt_before: All | Sequence[str] | None = None, + interrupt_after: All | Sequence[str] | None = None, + webhook: str | None = None, + on_disconnect: DisconnectMode | None = None, + on_completion: OnCompletionBehavior | None = None, + multitask_strategy: MultitaskStrategy | None = None, + if_not_exists: IfNotExists | None = None, + after_seconds: int | None = None, + headers: dict[str, str] | None = None, + on_run_created: Callable[[RunCreateMetadata], None] | None = None, + ) -> list[dict] | dict[str, Any]: + """Create a run, wait until it finishes and return the final state. + + Args: + thread_id: the thread ID to create the run on. + If None will create a stateless run. + assistant_id: The assistant ID or graph name to run. + If using graph name, will default to first assistant created from that graph. + input: The input to the graph. + command: The command to execute. + metadata: Metadata to assign to the run. + config: The configuration for the assistant. + checkpoint: The checkpoint to resume from. + checkpoint_during: Whether to checkpoint during the run (or only at the end/interruption). + interrupt_before: Nodes to interrupt immediately before they get executed. + interrupt_after: Nodes to Nodes to interrupt immediately after they get executed. + webhook: Webhook to call after LangGraph API call is done. + on_disconnect: The disconnect mode to use. + Must be one of 'cancel' or 'continue'. + on_completion: Whether to delete or keep the thread created for a stateless run. + Must be one of 'delete' or 'keep'. + multitask_strategy: Multitask strategy to use. + Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + if_not_exists: How to handle missing thread. Defaults to 'reject'. + Must be either 'reject' (raise error if missing), or 'create' (create new thread). + after_seconds: The number of seconds to wait before starting the run. + Use to schedule future runs. + headers: Optional custom headers to include with the request. + on_run_created: Optional callback to call when a run is created. + + Returns: + Union[list[dict], dict[str, Any]]: The output of the run. + + ???+ example "Example Usage" + + ```python + + final_state_of_run = client.runs.wait( + thread_id=None, + assistant_id="agent", + input={"messages": [{"role": "user", "content": "how are you?"}]}, + metadata={"name":"my_run"}, + config={"configurable": {"model_name": "anthropic"}}, + interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"], + interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"], + webhook="https://my.fake.webhook.com", + multitask_strategy="interrupt" + ) + print(final_state_of_run) + ``` + + ```shell + + ------------------------------------------------------------------------------------------------------------------------------------------- + + { + 'messages': [ + { + 'content': 'how are you?', + 'additional_kwargs': {}, + 'response_metadata': {}, + 'type': 'human', + 'name': None, + 'id': 'f51a862c-62fe-4866-863b-b0863e8ad78a', + 'example': False + }, + { + 'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.", + 'additional_kwargs': {}, + 'response_metadata': {}, + 'type': 'ai', + 'name': None, + 'id': 'run-bf1cd3c6-768f-4c16-b62d-ba6f17ad8b36', + 'example': False, + 'tool_calls': [], + 'invalid_tool_calls': [], + 'usage_metadata': None + } + ] + } + ``` + + """ # noqa: E501 + payload = { + "input": input, + "command": ( + {k: v for k, v in command.items() if v is not None} if command else None + ), + "config": config, + "metadata": metadata, + "assistant_id": assistant_id, + "interrupt_before": interrupt_before, + "interrupt_after": interrupt_after, + "webhook": webhook, + "checkpoint": checkpoint, + "checkpoint_id": checkpoint_id, + "multitask_strategy": multitask_strategy, + "if_not_exists": if_not_exists, + "on_disconnect": on_disconnect, + "checkpoint_during": checkpoint_during, + "on_completion": on_completion, + "after_seconds": after_seconds, + } + + def on_response(res: httpx.Response): + """Callback function to handle the response.""" + if on_run_created and (metadata := _get_run_metadata_from_response(res)): + on_run_created(metadata) + + endpoint = ( + f"/threads/{thread_id}/runs/wait" if thread_id is not None else "/runs/wait" + ) + return self.http.post( + endpoint, + json={k: v for k, v in payload.items() if v is not None}, + headers=headers, + on_response=on_response if on_run_created else None, + ) + + def list( + self, + thread_id: str, + *, + limit: int = 10, + offset: int = 0, + headers: dict[str, str] | None = None, + ) -> list[Run]: + """List runs. + + Args: + thread_id: The thread ID to list runs for. + limit: The maximum number of results to return. + offset: The number of results to skip. + headers: Optional custom headers to include with the request. + + Returns: + list[Run]: The runs for the thread. + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:2024") + client.runs.list( + thread_id="thread_id", + limit=5, + offset=5, + ) + ``` + + """ # noqa: E501 + return self.http.get( + f"/threads/{thread_id}/runs?limit={limit}&offset={offset}", headers=headers + ) + + def get( + self, + thread_id: str, + run_id: str, + *, + headers: dict[str, str] | None = None, + ) -> Run: + """Get a run. + + Args: + thread_id: The thread ID to get. + run_id: The run ID to get. + headers: Optional custom headers to include with the request. + + Returns: + Run: Run object. + + ???+ example "Example Usage" + + ```python + + run = client.runs.get( + thread_id="thread_id_to_delete", + run_id="run_id_to_delete", + ) + ``` + """ # noqa: E501 + + return self.http.get(f"/threads/{thread_id}/runs/{run_id}", headers=headers) + + def cancel( + self, + thread_id: str, + run_id: str, + *, + wait: bool = False, + action: CancelAction = "interrupt", + headers: dict[str, str] | None = None, + ) -> None: + """Get a run. + + Args: + thread_id: The thread ID to cancel. + run_id: The run ID to cancel. + wait: Whether to wait until run has completed. + action: Action to take when cancelling the run. Possible values + are `interrupt` or `rollback`. Default is `interrupt`. + headers: Optional custom headers to include with the request. + + Returns: + None + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:2024") + client.runs.cancel( + thread_id="thread_id_to_cancel", + run_id="run_id_to_cancel", + wait=True, + action="interrupt" + ) + ``` + + """ # noqa: E501 + return self.http.post( + f"/threads/{thread_id}/runs/{run_id}/cancel?wait={1 if wait else 0}&action={action}", + json=None, + headers=headers, + ) + + def join( + self, + thread_id: str, + run_id: str, + *, + headers: dict[str, str] | None = None, + ) -> dict: + """Block until a run is done. Returns the final state of the thread. + + Args: + thread_id: The thread ID to join. + run_id: The run ID to join. + headers: Optional custom headers to include with the request. + + Returns: + None + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:2024") + client.runs.join( + thread_id="thread_id_to_join", + run_id="run_id_to_join" + ) + ``` + + """ # noqa: E501 + return self.http.get( + f"/threads/{thread_id}/runs/{run_id}/join", headers=headers + ) + + def join_stream( + self, + thread_id: str, + run_id: str, + *, + stream_mode: StreamMode | Sequence[StreamMode] | None = None, + cancel_on_disconnect: bool = False, + headers: dict[str, str] | None = None, + last_event_id: str | None = None, + ) -> Iterator[StreamPart]: + """Stream output from a run in real-time, until the run is done. + Output is not buffered, so any output produced before this call will + not be received here. + + Args: + thread_id: The thread ID to join. + run_id: The run ID to join. + stream_mode: The stream mode(s) to use. Must be a subset of the stream modes passed + when creating the run. Background runs default to having the union of all + stream modes. + cancel_on_disconnect: Whether to cancel the run when the stream is disconnected. + headers: Optional custom headers to include with the request. + + Returns: + None + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:2024") + client.runs.join_stream( + thread_id="thread_id_to_join", + run_id="run_id_to_join", + stream_mode=["values", "debug"] + ) + ``` + + """ # noqa: E501 + return self.http.stream( + f"/threads/{thread_id}/runs/{run_id}/stream", + "GET", + params={ + "stream_mode": stream_mode, + "cancel_on_disconnect": cancel_on_disconnect, + }, + headers={ + **({"Last-Event-ID": last_event_id} if last_event_id else {}), + **(headers or {}), + } + or None, + ) + + def delete( + self, + thread_id: str, + run_id: str, + *, + headers: dict[str, str] | None = None, + ) -> None: + """Delete a run. + + Args: + thread_id: The thread ID to delete. + run_id: The run ID to delete. + headers: Optional custom headers to include with the request. + + Returns: + None + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:2024") + client.runs.delete( + thread_id="thread_id_to_delete", + run_id="run_id_to_delete" + ) + ``` + + """ # noqa: E501 + self.http.delete(f"/threads/{thread_id}/runs/{run_id}", headers=headers) + + +class SyncCronClient: + """Synchronous client for managing cron jobs in LangGraph. + + This class provides methods to create and manage scheduled tasks (cron jobs) for automated graph executions. + + ???+ example "Example" + + ```python + client = get_sync_client(url="http://localhost:8123") + cron_job = client.crons.create_for_thread(thread_id="thread_123", assistant_id="asst_456", schedule="0 * * * *") + ``` + + !!! note "Feature Availability" + The crons client functionality is not supported on all licenses. + Please check the relevant license documentation for the most up-to-date + details on feature availability. + """ + + def __init__(self, http_client: SyncHttpClient) -> None: + self.http = http_client + + def create_for_thread( + self, + thread_id: str, + assistant_id: str, + *, + schedule: str, + input: dict | None = None, + metadata: dict | None = None, + checkpoint_during: bool | None = None, + config: Config | None = None, + interrupt_before: All | list[str] | None = None, + interrupt_after: All | list[str] | None = None, + webhook: str | None = None, + multitask_strategy: str | None = None, + headers: dict[str, str] | None = None, + ) -> Run: + """Create a cron job for a thread. + + Args: + thread_id: the thread ID to run the cron job on. + assistant_id: The assistant ID or graph name to use for the cron job. + If using graph name, will default to first assistant created from that graph. + schedule: The cron schedule to execute this job on. + input: The input to the graph. + metadata: Metadata to assign to the cron job runs. + config: The configuration for the assistant. + checkpoint_during: Whether to checkpoint during the run (or only at the end/interruption). + interrupt_before: Nodes to interrupt immediately before they get executed. + interrupt_after: Nodes to Nodes to interrupt immediately after they get executed. + webhook: Webhook to call after LangGraph API call is done. + multitask_strategy: Multitask strategy to use. + Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + headers: Optional custom headers to include with the request. + + Returns: + Run: The cron run. + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:8123") + cron_run = client.crons.create_for_thread( + thread_id="my-thread-id", + assistant_id="agent", + schedule="27 15 * * *", + input={"messages": [{"role": "user", "content": "hello!"}]}, + metadata={"name":"my_run"}, + config={"configurable": {"model_name": "openai"}}, + interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"], + interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"], + webhook="https://my.fake.webhook.com", + multitask_strategy="interrupt" + ) + ``` + """ # noqa: E501 + payload = { + "schedule": schedule, + "input": input, + "config": config, + "metadata": metadata, + "assistant_id": assistant_id, + "interrupt_before": interrupt_before, + "interrupt_after": interrupt_after, + "checkpoint_during": checkpoint_during, + "webhook": webhook, + "multitask_strategy": multitask_strategy, + } + payload = {k: v for k, v in payload.items() if v is not None} + return self.http.post( + f"/threads/{thread_id}/runs/crons", json=payload, headers=headers + ) + + def create( + self, + assistant_id: str, + *, + schedule: str, + input: dict | None = None, + metadata: dict | None = None, + config: Config | None = None, + checkpoint_during: bool | None = None, + interrupt_before: All | list[str] | None = None, + interrupt_after: All | list[str] | None = None, + webhook: str | None = None, + multitask_strategy: str | None = None, + headers: dict[str, str] | None = None, + ) -> Run: + """Create a cron run. + + Args: + assistant_id: The assistant ID or graph name to use for the cron job. + If using graph name, will default to first assistant created from that graph. + schedule: The cron schedule to execute this job on. + input: The input to the graph. + metadata: Metadata to assign to the cron job runs. + config: The configuration for the assistant. + checkpoint_during: Whether to checkpoint during the run (or only at the end/interruption). + interrupt_before: Nodes to interrupt immediately before they get executed. + interrupt_after: Nodes to Nodes to interrupt immediately after they get executed. + webhook: Webhook to call after LangGraph API call is done. + multitask_strategy: Multitask strategy to use. + Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + headers: Optional custom headers to include with the request. + + Returns: + Run: The cron run. + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:8123") + cron_run = client.crons.create( + assistant_id="agent", + schedule="27 15 * * *", + input={"messages": [{"role": "user", "content": "hello!"}]}, + metadata={"name":"my_run"}, + config={"configurable": {"model_name": "openai"}}, + checkpoint_during=True, + interrupt_before=["node_to_stop_before_1","node_to_stop_before_2"], + interrupt_after=["node_to_stop_after_1","node_to_stop_after_2"], + webhook="https://my.fake.webhook.com", + multitask_strategy="interrupt" + ) + ``` + + """ # noqa: E501 + payload = { + "schedule": schedule, + "input": input, + "config": config, + "metadata": metadata, + "assistant_id": assistant_id, + "interrupt_before": interrupt_before, + "interrupt_after": interrupt_after, + "webhook": webhook, + "checkpoint_during": checkpoint_during, + "multitask_strategy": multitask_strategy, + } + payload = {k: v for k, v in payload.items() if v is not None} + return self.http.post("/runs/crons", json=payload, headers=headers) + + def delete( + self, + cron_id: str, + *, + headers: dict[str, str] | None = None, + ) -> None: + """Delete a cron. + + Args: + cron_id: The cron ID to delete. + headers: Optional custom headers to include with the request. + + Returns: + None + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:8123") + client.crons.delete( + cron_id="cron_to_delete" + ) + ``` + + """ # noqa: E501 + self.http.delete(f"/runs/crons/{cron_id}", headers=headers) + + def search( + self, + *, + assistant_id: str | None = None, + thread_id: str | None = None, + limit: int = 10, + offset: int = 0, + sort_by: CronSortBy | None = None, + sort_order: SortOrder | None = None, + headers: dict[str, str] | None = None, + ) -> list[Cron]: + """Get a list of cron jobs. + + Args: + assistant_id: The assistant ID or graph name to search for. + thread_id: the thread ID to search for. + limit: The maximum number of results to return. + offset: The number of results to skip. + headers: Optional custom headers to include with the request. + + Returns: + list[Cron]: The list of cron jobs returned by the search, + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:8123") + cron_jobs = client.crons.search( + assistant_id="my_assistant_id", + thread_id="my_thread_id", + limit=5, + offset=5, + ) + print(cron_jobs) + ``` + + ```shell + ---------------------------------------------------------- + + [ + { + 'cron_id': '1ef3cefa-4c09-6926-96d0-3dc97fd5e39b', + 'assistant_id': 'my_assistant_id', + 'thread_id': 'my_thread_id', + 'user_id': None, + 'payload': + { + 'input': {'start_time': ''}, + 'schedule': '4 * * * *', + 'assistant_id': 'my_assistant_id' + }, + 'schedule': '4 * * * *', + 'next_run_date': '2024-07-25T17:04:00+00:00', + 'end_time': None, + 'created_at': '2024-07-08T06:02:23.073257+00:00', + 'updated_at': '2024-07-08T06:02:23.073257+00:00' + } + ] + ``` + """ # noqa: E501 + payload = { + "assistant_id": assistant_id, + "thread_id": thread_id, + "limit": limit, + "offset": offset, + } + if sort_by: + payload["sort_by"] = sort_by + if sort_order: + payload["sort_order"] = sort_order + payload = {k: v for k, v in payload.items() if v is not None} + return self.http.post("/runs/crons/search", json=payload, headers=headers) + + +class SyncStoreClient: + """A client for synchronous operations on a key-value store. + + Provides methods to interact with a remote key-value store, allowing + storage and retrieval of items within namespaced hierarchies. + + ???+ example "Example" + + ```python + client = get_sync_client(url="http://localhost:2024")) + client.store.put_item(["users", "profiles"], "user123", {"name": "Alice", "age": 30}) + ``` + """ + + def __init__(self, http: SyncHttpClient) -> None: + self.http = http + + def put_item( + self, + namespace: Sequence[str], + /, + key: str, + value: dict[str, Any], + index: Literal[False] | list[str] | None = None, + ttl: int | None = None, + headers: dict[str, str] | None = None, + ) -> None: + """Store or update an item. + + Args: + namespace: A list of strings representing the namespace path. + key: The unique identifier for the item within the namespace. + value: A dictionary containing the item's data. + index: Controls search indexing - None (use defaults), False (disable), or list of field paths to index. + ttl: Optional time-to-live in minutes for the item, or None for no expiration. + headers: Optional custom headers to include with the request. + + Returns: + None + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:8123") + client.store.put_item( + ["documents", "user123"], + key="item456", + value={"title": "My Document", "content": "Hello World"} + ) + ``` + """ + for label in namespace: + if "." in label: + raise ValueError( + f"Invalid namespace label '{label}'. Namespace labels cannot contain periods ('.')." + ) + payload = { + "namespace": namespace, + "key": key, + "value": value, + "index": index, + "ttl": ttl, + } + self.http.put("/store/items", json=_provided_vals(payload), headers=headers) + + def get_item( + self, + namespace: Sequence[str], + /, + key: str, + *, + refresh_ttl: bool | None = None, + headers: dict[str, str] | None = None, + ) -> Item: + """Retrieve a single item. + + Args: + key: The unique identifier for the item. + namespace: Optional list of strings representing the namespace path. + refresh_ttl: Whether to refresh the TTL on this read operation. If None, uses the store's default behavior. + headers: Optional custom headers to include with the request. + + Returns: + Item: The retrieved item. + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:8123") + item = client.store.get_item( + ["documents", "user123"], + key="item456", + ) + print(item) + ``` + + ```shell + ---------------------------------------------------------------- + + { + 'namespace': ['documents', 'user123'], + 'key': 'item456', + 'value': {'title': 'My Document', 'content': 'Hello World'}, + 'created_at': '2024-07-30T12:00:00Z', + 'updated_at': '2024-07-30T12:00:00Z' + } + ``` + """ + for label in namespace: + if "." in label: + raise ValueError( + f"Invalid namespace label '{label}'. Namespace labels cannot contain periods ('.')." + ) + + params = {"key": key, "namespace": ".".join(namespace)} + if refresh_ttl is not None: + params["refresh_ttl"] = refresh_ttl + return self.http.get("/store/items", params=params, headers=headers) + + def delete_item( + self, + namespace: Sequence[str], + /, + key: str, + headers: dict[str, str] | None = None, + ) -> None: + """Delete an item. + + Args: + key: The unique identifier for the item. + namespace: Optional list of strings representing the namespace path. + headers: Optional custom headers to include with the request. + + Returns: + None + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:8123") + client.store.delete_item( + ["documents", "user123"], + key="item456", + ) + ``` + """ + self.http.delete( + "/store/items", json={"key": key, "namespace": namespace}, headers=headers + ) + + def search_items( + self, + namespace_prefix: Sequence[str], + /, + filter: dict[str, Any] | None = None, + limit: int = 10, + offset: int = 0, + query: str | None = None, + refresh_ttl: bool | None = None, + headers: dict[str, str] | None = None, + ) -> SearchItemsResponse: + """Search for items within a namespace prefix. + + Args: + namespace_prefix: List of strings representing the namespace prefix. + filter: Optional dictionary of key-value pairs to filter results. + 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 for natural language search. + refresh_ttl: Whether to refresh the TTL on items returned by this search. If None, uses the store's default behavior. + headers: Optional custom headers to include with the request. + + Returns: + list[Item]: A list of items matching the search criteria. + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:8123") + items = client.store.search_items( + ["documents"], + filter={"author": "John Doe"}, + limit=5, + offset=0 + ) + print(items) + ``` + ```shell + ---------------------------------------------------------------- + + { + "items": [ + { + "namespace": ["documents", "user123"], + "key": "item789", + "value": { + "title": "Another Document", + "author": "John Doe" + }, + "created_at": "2024-07-30T12:00:00Z", + "updated_at": "2024-07-30T12:00:00Z" + }, + # ... additional items ... + ] + } + ``` + """ + payload = { + "namespace_prefix": namespace_prefix, + "filter": filter, + "limit": limit, + "offset": offset, + "query": query, + "refresh_ttl": refresh_ttl, + } + return self.http.post( + "/store/items/search", json=_provided_vals(payload), headers=headers + ) + + def list_namespaces( + self, + prefix: list[str] | None = None, + suffix: list[str] | None = None, + max_depth: int | None = None, + limit: int = 100, + offset: int = 0, + headers: dict[str, str] | 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. + + Returns: + list[list[str]]: A list of namespaces matching the criteria. + + ???+ example "Example Usage" + + ```python + client = get_sync_client(url="http://localhost:8123") + namespaces = client.store.list_namespaces( + prefix=["documents"], + max_depth=3, + limit=10, + offset=0 + ) + print(namespaces) + ``` + + ```shell + ---------------------------------------------------------------- + + [ + ["documents", "user123", "reports"], + ["documents", "user456", "invoices"], + ... + ] + ``` + """ + payload = { + "prefix": prefix, + "suffix": suffix, + "max_depth": max_depth, + "limit": limit, + "offset": offset, + } + return self.http.post( + "/store/namespaces", json=_provided_vals(payload), headers=headers + ) + + +def _provided_vals(d: dict): + return {k: v for k, v in d.items() if v is not None} + + +_registered_transports: list[httpx.ASGITransport] = [] + + +# Do not move; this is used in the server. +def configure_loopback_transports(app: Any) -> None: + for transport in _registered_transports: + transport.app = app + + +@functools.lru_cache(maxsize=1) +def get_asgi_transport() -> type[httpx.ASGITransport]: + try: + from langgraph_api import asgi_transport + + return asgi_transport.ASGITransport + except ImportError: + # Older versions of the server + return httpx.ASGITransport + + +TimeoutTypes = Union[ + None, + float, + tuple[Optional[float], Optional[float]], + tuple[Optional[float], Optional[float], Optional[float], Optional[float]], + httpx.Timeout, +]