From 422ed92aef23ece58f8093b59f21d87a1b9ac5a3 Mon Sep 17 00:00:00 2001 From: vbarda Date: Tue, 20 Aug 2024 15:34:38 -0400 Subject: [PATCH 1/8] sdk-py: add sync client --- libs/sdk-py/langgraph_sdk/client/__init__.py | 3 + .../{client.py => client/async_client.py} | 45 +- .../langgraph_sdk/client/sync_client.py | 1656 +++++++++++++++++ libs/sdk-py/langgraph_sdk/schema.py | 7 +- libs/sdk-py/langgraph_sdk/utils.py | 18 + 5 files changed, 1695 insertions(+), 34 deletions(-) create mode 100644 libs/sdk-py/langgraph_sdk/client/__init__.py rename libs/sdk-py/langgraph_sdk/{client.py => client/async_client.py} (98%) create mode 100644 libs/sdk-py/langgraph_sdk/client/sync_client.py create mode 100644 libs/sdk-py/langgraph_sdk/utils.py diff --git a/libs/sdk-py/langgraph_sdk/client/__init__.py b/libs/sdk-py/langgraph_sdk/client/__init__.py new file mode 100644 index 000000000..741fc51d1 --- /dev/null +++ b/libs/sdk-py/langgraph_sdk/client/__init__.py @@ -0,0 +1,3 @@ +from langgraph_sdk.client.async_client import get_client + +__all__ = ["get_client"] diff --git a/libs/sdk-py/langgraph_sdk/client.py b/libs/sdk-py/langgraph_sdk/client/async_client.py similarity index 98% rename from libs/sdk-py/langgraph_sdk/client.py rename to libs/sdk-py/langgraph_sdk/client/async_client.py index 13d5560db..7e5c872fd 100644 --- a/libs/sdk-py/langgraph_sdk/client.py +++ b/libs/sdk-py/langgraph_sdk/client/async_client.py @@ -2,14 +2,12 @@ from __future__ import annotations import asyncio import logging -import os import sys from typing import ( Any, AsyncIterator, Dict, List, - NamedTuple, Optional, Union, overload, @@ -32,17 +30,19 @@ from langgraph_sdk.schema import ( Run, RunCreate, StreamMode, + StreamPart, Thread, ThreadState, ThreadStatus, ) +from langgraph_sdk.utils import get_api_key logger = logging.getLogger(__name__) def get_client( *, url: Optional[str] = None, api_key: Optional[str] = None -) -> LangGraphClient: +) -> AsyncLangGraphClient: """Get a LangGraphClient instance. Args: @@ -68,7 +68,7 @@ def get_client( headers = { "User-Agent": f"langgraph-sdk-py/{langgraph_sdk.__version__}", } - api_key = _get_api_key(api_key) + api_key = get_api_key(api_key) if api_key: headers["x-api-key"] = api_key client = httpx.AsyncClient( @@ -77,24 +77,19 @@ def get_client( timeout=httpx.Timeout(connect=5, read=60, write=60, pool=5), headers=headers, ) - return LangGraphClient(client) + return AsyncLangGraphClient(client) -class StreamPart(NamedTuple): - event: str - data: dict - - -class LangGraphClient: +class AsyncLangGraphClient: def __init__(self, client: httpx.AsyncClient) -> None: - self.http = HttpClient(client) + self.http = AsyncHttpClient(client) self.assistants = AssistantsClient(self.http) self.threads = ThreadsClient(self.http) self.runs = RunsClient(self.http) self.crons = CronClient(self.http) -class HttpClient: +class AsyncHttpClient: def __init__(self, client: httpx.AsyncClient) -> None: self.client = client @@ -231,7 +226,7 @@ async def decode_json(r: httpx.Response) -> Any: class AssistantsClient: - def __init__(self, http: HttpClient) -> None: + def __init__(self, http: AsyncHttpClient) -> None: self.http = http async def get(self, assistant_id: str) -> Assistant: @@ -561,7 +556,7 @@ class AssistantsClient: class ThreadsClient: - def __init__(self, http: HttpClient) -> None: + def __init__(self, http: AsyncHttpClient) -> None: self.http = http async def get(self, thread_id: str) -> Thread: @@ -944,7 +939,7 @@ class ThreadsClient: class RunsClient: - def __init__(self, http: HttpClient) -> None: + def __init__(self, http: AsyncHttpClient) -> None: self.http = http @overload @@ -1471,7 +1466,7 @@ class RunsClient: class CronClient: - def __init__(self, http_client: HttpClient) -> None: + def __init__(self, http_client: AsyncHttpClient) -> None: self.http = http_client async def create_for_thread( @@ -1679,19 +1674,3 @@ class CronClient: } payload = {k: v for k, v in payload.items() if v is not None} return await self.http.post("/runs/crons/search", json=payload) - - -def _get_api_key(api_key: Optional[str] = None) -> Optional[str]: - """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 diff --git a/libs/sdk-py/langgraph_sdk/client/sync_client.py b/libs/sdk-py/langgraph_sdk/client/sync_client.py new file mode 100644 index 000000000..9e867cb97 --- /dev/null +++ b/libs/sdk-py/langgraph_sdk/client/sync_client.py @@ -0,0 +1,1656 @@ +from __future__ import annotations + +import logging +import sys +from typing import ( + Any, + Dict, + Iterator, + List, + Optional, + Union, + overload, +) + +import httpx +import httpx_sse +import orjson +from httpx._types import QueryParamTypes + +import langgraph_sdk +from langgraph_sdk.schema import ( + Assistant, + Config, + Cron, + GraphSchema, + Metadata, + MultitaskStrategy, + OnConflictBehavior, + Run, + RunCreate, + StreamMode, + StreamPart, + Thread, + ThreadState, + ThreadStatus, +) +from langgraph_sdk.utils import get_api_key + +logger = logging.getLogger(__name__) + + +def get_client( + *, url: Optional[str] = None, api_key: Optional[str] = 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 + """ + + if url is None: + url = "http://localhost:8123" + transport = httpx.HTTPTransport(retries=5) + headers = { + "User-Agent": f"langgraph-sdk-py/{langgraph_sdk.__version__}", + } + api_key = get_api_key(api_key) + if api_key: + headers["x-api-key"] = api_key + client = httpx.Client( + base_url=url, + transport=transport, + timeout=httpx.Timeout(connect=5, read=60, write=60, pool=5), + headers=headers, + ) + return LangGraphClient(client) + + +class LangGraphClient: + def __init__(self, client: httpx.Client) -> 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) + + +class HttpClient: + def __init__(self, client: httpx.Client) -> None: + self.client = client + + def get(self, path: str, *, params: Optional[QueryParamTypes] = None) -> Any: + """Make a GET request.""" + r = self.client.get(path, params=params) + 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: Optional[dict]) -> Any: + """Make a POST request.""" + if json is not None: + headers, content = encode_json(json) + else: + headers, content = {}, b"" + r = self.client.post(path, headers=headers, content=content) + 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) -> Any: + """Make a PUT request.""" + headers, content = encode_json(json) + r = self.client.put(path, headers=headers, content=content) + 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) -> Any: + """Make a PATCH request.""" + headers, content = encode_json(json) + r = self.client.patch(path, headers=headers, content=content) + 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) -> None: + """Make a DELETE request.""" + r = self.client.delete(path) + 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: Optional[dict] = None + ) -> Iterator[StreamPart]: + """Stream the results of a request using SSE.""" + headers, content = encode_json(json) + with httpx_sse.connect_sse( + self.client, method, path, headers=headers, content=content + ) as sse: + try: + sse.response.raise_for_status() + except httpx.HTTPStatusError as e: + body = (sse.response.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 + for event in sse.iter_sse(): + yield StreamPart( + event.event, orjson.loads(event.data) if event.data else None + ) + + +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") + + +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 AssistantsClient: + def __init__(self, http: HttpClient) -> None: + self.http = http + + def get(self, assistant_id: str) -> Assistant: + """Get an assistant by ID. + + Args: + assistant_id: The ID of the assistant to get. + + Returns: + Assistant: Assistant Object. + + Example Usage: + + assistant = client.assistants.get( + assistant_id="my_assistant_id" + ) + print(assistant) + + ---------------------------------------------------- + + { + '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}") + + def get_graph(self, assistant_id: str) -> 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. + + Returns: + Graph: The graph information for the assistant in JSON format. + + Example Usage: + + 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") + + def get_schemas(self, assistant_id: str) -> GraphSchema: + """Get the schemas of an assistant by ID. + + Args: + assistant_id: The ID of the assistant to get the schema of. + + Returns: + GraphSchema: The graph schema for the assistant. + + Example Usage: + + schema = client.assistants.get_schemas( + assistant_id="my_assistant_id" + ) + print(schema) + + ---------------------------------------------------------------------------------------------------------------------------- + + { + '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") + + def create( + self, + graph_id: Optional[str], + config: Optional[Config] = None, + *, + metadata: Metadata = None, + assistant_id: Optional[str] = None, + if_exists: Optional[OnConflictBehavior] = 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). + + Returns: + Assistant: The created assistant. + + Example Usage: + + assistant = client.assistants.create( + graph_id="agent", + config={"configurable": {"model_name": "openai"}}, + metadata={"number":1}, + assistant_id="my-assistant-id", + if_exists="do_nothing" + ) + """ # 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 + return self.http.post("/assistants", json=payload) + + def update( + self, + assistant_id: str, + *, + graph_id: Optional[str] = None, + config: Optional[Config] = None, + metadata: Metadata = 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 add to assistant. + + Returns: + Assistant: The updated assistant. + + Example Usage: + + 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 + return self.http.patch( + f"/assistants/{assistant_id}", + json=payload, + ) + + def delete( + self, + assistant_id: str, + ) -> None: + """Delete an assistant. + + Args: + assistant_id: The assistant ID to delete. + + Returns: + None + + Example Usage: + + client.assistants.delete( + assistant_id="my_assistant_id" + ) + + """ # noqa: E501 + self.http.delete(f"/assistants/{assistant_id}") + + def search( + self, + *, + metadata: Metadata = None, + graph_id: Optional[str] = None, + limit: int = 10, + offset: int = 0, + ) -> 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. + + Returns: + list[Assistant]: A list of assistants. + + Example Usage: + + 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 + return self.http.post( + "/assistants/search", + json=payload, + ) + + +class ThreadsClient: + def __init__(self, http: HttpClient) -> None: + self.http = http + + def get(self, thread_id: str) -> Thread: + """Get a thread by ID. + + Args: + thread_id: The ID of the thread to get. + + Returns: + Thread: Thread object. + + Example Usage: + + thread = client.threads.get( + thread_id="my_thread_id" + ) + print(thread) + + ----------------------------------------------------- + + { + '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}") + + def create( + self, + *, + metadata: Metadata = None, + thread_id: Optional[str] = None, + if_exists: Optional[OnConflictBehavior] = 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). + + Returns: + Thread: The created thread. + + Example Usage: + + 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: + payload["metadata"] = metadata + if if_exists: + payload["if_exists"] = if_exists + return self.http.post("/threads", json=payload) + + def update(self, thread_id: str, *, metadata: dict[str, Any]) -> Thread: + """Update a thread. + + Args: + thread_id: ID of thread to update. + metadata: Metadata to add/update to thread. + + Returns: + Thread: The created thread. + + Example Usage: + + 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}) + + def delete(self, thread_id: str) -> None: + """Delete a thread. + + Args: + thread_id: The ID of the thread to delete. + + Returns: + None + + Example Usage: + + client.threads.delete( + thread_id="my_thread_id" + ) + + """ # noqa: E501 + self.http.delete(f"/threads/{thread_id}") + + def search( + self, + *, + metadata: Metadata = None, + status: Optional[ThreadStatus] = None, + limit: int = 10, + offset: int = 0, + ) -> list[Thread]: + """Search for threads. + + Args: + metadata: Thread metadata to search for. + status: Status to search for. + Must be one of 'idle', 'busy', or 'interrupted'. + limit: Limit on number of threads to return. + offset: Offset in threads table to start search from. + + Returns: + list[Thread]: List of the threads matching the search parameters. + + Example Usage: + + 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 status: + payload["status"] = status + return self.http.post( + "/threads/search", + json=payload, + ) + + def copy(self, thread_id: str) -> None: + """Copy a thread. + + Args: + thread_id: The ID of the thread to copy. + + Returns: + None + + Example Usage: + + client.threads.copy( + thread_id="my_thread_id" + ) + + """ # noqa: E501 + return self.http.post(f"/threads/{thread_id}/copy", json=None) + + def get_state( + self, thread_id: str, checkpoint_id: Optional[str] = None + ) -> ThreadState: + """Get the state of a thread. + + Args: + thread_id: The ID of the thread to get the state of. + checkpoint_id: The ID of the checkpoint to get the state of. + + Returns: + ThreadState: the thread of the state. + + Example Usage: + + thread_state = client.threads.get_state( + thread_id="my_thread_id", + checkpoint_id="my_checkpoint_id" + ) + print(thread_state) + + ---------------------------------------------------------------------------------------------------------------------------------------------------------------------- + + { + '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': [], + 'config': + { + 'configurable': + { + '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': + { + 'configurable': + { + 'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2', + 'checkpoint_ns': '', + 'checkpoint_id': '1ef4a9b8-d80d-6fa7-8000-9300467fad0f' + } + } + } + + """ # noqa: E501 + if checkpoint_id: + return self.http.get(f"/threads/{thread_id}/state/{checkpoint_id}") + else: + return self.http.get(f"/threads/{thread_id}/state") + + def update_state( + self, + thread_id: str, + values: dict, + *, + as_node: Optional[str] = None, + checkpoint_id: Optional[str] = None, + ) -> None: + """Update the state of a thread. + + Args: + thread_id: The ID of the thread to update. + values: The values to update to the state. + as_node: Update the state as if this node had just executed. + + checkpoint_id: The ID of the checkpoint to get the state of. + + Returns: + None + + Example Usage: + + client.threads.get_state( + thread_id="my_thread_id", + values={"messages":[{"role": "user", "content": "hello!"}]}, + as_node="my_node", + checkpoint_id="my_checkpoint_id" + ) + + """ # noqa: E501 + payload: Dict[str, Any] = { + "values": values, + } + if checkpoint_id: + payload["checkpoint_id"] = checkpoint_id + if as_node: + payload["as_node"] = as_node + return self.http.post(f"/threads/{thread_id}/state", json=payload) + + def patch_state( + self, + thread_id: Union[str, Config], + metadata: dict, + ) -> None: + """Patch the state of a thread. + + Args: + thread_id: The ID of the thread to get the state of. + metadata: The metadata to assign to the state. + + Returns: + None + + Example Usage: + + client.threads.patch_state( + thread_id="my_thread_id", + metadata={"name":"new_name"}, + ) + + """ # noqa: E501 + if isinstance(thread_id, dict): + thread_id_: str = thread_id["configurable"]["thread_id"] + else: + thread_id_ = thread_id + return self.http.patch( + f"/threads/{thread_id_}/state", + json={"metadata": metadata}, + ) + + def get_history( + self, + thread_id: str, + limit: int = 10, + before: Optional[str] = None, + metadata: Optional[dict] = None, + ) -> list[ThreadState]: + """Get the state history of a thread. + + Args: + thread_id: The ID of the thread to get the state of. + limit: The maximum number of results to return. + before: Thread timestamp to get history before. + metadata: The metadata of the thread history to get. + + Returns: + list[ThreadState]: the state history of the thread. + + Example Usage: + + 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 + return self.http.post(f"/threads/{thread_id}/history", json=payload) + + +class RunsClient: + def __init__(self, http: HttpClient) -> None: + self.http = http + + @overload + def stream( + self, + thread_id: str, + assistant_id: str, + *, + input: Optional[dict] = None, + stream_mode: Union[StreamMode, list[StreamMode]] = "values", + metadata: Optional[dict] = None, + config: Optional[Config] = None, + checkpoint_id: Optional[str] = None, + interrupt_before: Optional[list[str]] = None, + interrupt_after: Optional[list[str]] = None, + feedback_keys: Optional[list[str]] = None, + multitask_strategy: Optional[MultitaskStrategy] = None, + ) -> Iterator[StreamPart]: + ... + + @overload + def stream( + self, + thread_id: None, + assistant_id: str, + *, + input: Optional[dict] = None, + stream_mode: Union[StreamMode, list[StreamMode]] = "values", + metadata: Optional[dict] = None, + config: Optional[Config] = None, + interrupt_before: Optional[list[str]] = None, + interrupt_after: Optional[list[str]] = None, + feedback_keys: Optional[list[str]] = None, + ) -> Iterator[StreamPart]: + ... + + def stream( + self, + thread_id: Optional[str], + assistant_id: str, + *, + input: Optional[dict] = None, + stream_mode: Union[StreamMode, list[StreamMode]] = "values", + metadata: Optional[dict] = None, + config: Optional[Config] = None, + checkpoint_id: Optional[str] = None, + interrupt_before: Optional[list[str]] = None, + interrupt_after: Optional[list[str]] = None, + feedback_keys: Optional[list[str]] = None, + webhook: Optional[str] = None, + multitask_strategy: Optional[MultitaskStrategy] = 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. + stream_mode: The stream mode(s) to use. + metadata: Metadata to assign to the run. + config: The configuration for the assistant. + checkpoint_id: The checkpoint to start streaming from. + 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. + 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'. + + Returns: + Iterator[StreamPart]: Asynchronous iterator of stream results. + + Example Usage: + + 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"}}, + checkpoint_id="my_checkpoint", + 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) + + ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ + + 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, + "config": config, + "metadata": metadata, + "stream_mode": stream_mode, + "assistant_id": assistant_id, + "interrupt_before": interrupt_before, + "interrupt_after": interrupt_after, + "feedback_keys": feedback_keys, + "webhook": webhook, + "checkpoint_id": checkpoint_id, + "multitask_strategy": multitask_strategy, + } + endpoint = ( + f"/threads/{thread_id}/runs/stream" + if thread_id is not None + else "/runs/stream" + ) + return self.http.stream( + endpoint, "POST", json={k: v for k, v in payload.items() if v is not None} + ) + + @overload + def create( + self, + thread_id: None, + assistant_id: str, + *, + input: Optional[dict] = None, + metadata: Optional[dict] = None, + config: Optional[Config] = None, + interrupt_before: Optional[list[str]] = None, + interrupt_after: Optional[list[str]] = None, + webhook: Optional[str] = None, + ) -> Run: + ... + + @overload + def create( + self, + thread_id: str, + assistant_id: str, + *, + input: Optional[dict] = None, + metadata: Optional[dict] = None, + config: Optional[Config] = None, + checkpoint_id: Optional[str] = None, + interrupt_before: Optional[list[str]] = None, + interrupt_after: Optional[list[str]] = None, + webhook: Optional[str] = None, + multitask_strategy: Optional[MultitaskStrategy] = None, + ) -> Run: + ... + + def create( + self, + thread_id: Optional[str], + assistant_id: str, + *, + input: Optional[dict] = None, + metadata: Optional[dict] = None, + config: Optional[Config] = None, + checkpoint_id: Optional[str] = None, + interrupt_before: Optional[list[str]] = None, + interrupt_after: Optional[list[str]] = None, + webhook: Optional[str] = None, + multitask_strategy: Optional[MultitaskStrategy] = 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. + metadata: Metadata to assign to the run. + config: The configuration for the assistant. + checkpoint_id: The checkpoint to start streaming from. + 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'. + + Returns: + Run: The created background run. + + Example Usage: + + 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"}}, + checkpoint_id="my_checkpoint", + 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) + + -------------------------------------------------------------------------------- + + { + '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, + "config": config, + "metadata": metadata, + "assistant_id": assistant_id, + "interrupt_before": interrupt_before, + "interrupt_after": interrupt_after, + "webhook": webhook, + "checkpoint_id": checkpoint_id, + "multitask_strategy": multitask_strategy, + } + payload = {k: v for k, v in payload.items() if v is not None} + if thread_id: + return self.http.post(f"/threads/{thread_id}/runs", json=payload) + else: + return self.http.post("/runs", json=payload) + + def create_batch(self, payloads: list[RunCreate]) -> list[Run]: + """Create a batch of 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) + + @overload + def wait( + self, + thread_id: str, + assistant_id: str, + *, + input: Optional[dict] = None, + metadata: Optional[dict] = None, + config: Optional[Config] = None, + checkpoint_id: Optional[str] = None, + interrupt_before: Optional[list[str]] = None, + interrupt_after: Optional[list[str]] = None, + multitask_strategy: Optional[MultitaskStrategy] = None, + ) -> Union[list[dict], dict[str, Any]]: + ... + + @overload + def wait( + self, + thread_id: None, + assistant_id: str, + *, + input: Optional[dict] = None, + metadata: Optional[dict] = None, + config: Optional[Config] = None, + interrupt_before: Optional[list[str]] = None, + interrupt_after: Optional[list[str]] = None, + ) -> Union[list[dict], dict[str, Any]]: + ... + + def wait( + self, + thread_id: Optional[str], + assistant_id: str, + *, + input: Optional[dict] = None, + metadata: Optional[dict] = None, + config: Optional[Config] = None, + checkpoint_id: Optional[str] = None, + interrupt_before: Optional[list[str]] = None, + interrupt_after: Optional[list[str]] = None, + webhook: Optional[str] = None, + multitask_strategy: Optional[MultitaskStrategy] = None, + ) -> Union[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. + metadata: Metadata to assign to the run. + config: The configuration for the assistant. + checkpoint_id: The checkpoint to start streaming from. + 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'. + + Returns: + Union[list[dict], dict[str, Any]]: The output of the run. + + Example Usage: + + 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"}}, + checkpoint_id="my_checkpoint", + 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) + + ------------------------------------------------------------------------------------------------------------------------------------------- + + { + '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, + "config": config, + "metadata": metadata, + "assistant_id": assistant_id, + "interrupt_before": interrupt_before, + "interrupt_after": interrupt_after, + "webhook": webhook, + "checkpoint_id": checkpoint_id, + "multitask_strategy": multitask_strategy, + } + 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} + ) + + def list(self, thread_id: str, *, limit: int = 10, offset: int = 0) -> 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. + + Returns: + List[Run]: The runs for the thread. + + Example Usage: + + client.runs.delete( + thread_id="thread_id_to_delete", + limit=5, + offset=5, + ) + + """ # noqa: E501 + return self.http.get(f"/threads/{thread_id}/runs?limit={limit}&offset={offset}") + + def get(self, thread_id: str, run_id: str) -> Run: + """Get a run. + + Args: + thread_id: The thread ID to get. + run_id: The run ID to get. + + Returns: + Run: Run object. + + Example Usage: + + 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}") + + def cancel(self, thread_id: str, run_id: str, *, wait: bool = False) -> None: + """Get a run. + + Args: + thread_id: The thread ID to cancel. + run_id: The run ID to cancek. + wait: Whether to wait until run has completed. + + Returns: + None + + Example Usage: + + client.runs.cancel( + thread_id="thread_id_to_cancel", + run_id="run_id_to_cancel", + wait=True + ) + + """ # noqa: E501 + return self.http.post( + f"/threads/{thread_id}/runs/{run_id}/cancel?wait={1 if wait else 0}", + json=None, + ) + + def join(self, thread_id: str, run_id: str) -> None: + """Block until a run is done. + + Args: + thread_id: The thread ID to join. + run_id: The run ID to join. + + Returns: + None + + Example Usage: + + 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") + + def delete(self, thread_id: str, run_id: str) -> None: + """Delete a run. + + Args: + thread_id: The thread ID to delete. + run_id: The run ID to delete. + + Returns: + None + + Example Usage: + + 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}") + + +class CronClient: + def __init__(self, http_client: HttpClient) -> None: + self.http = http_client + + def create_for_thread( + self, + thread_id: str, + assistant_id: str, + *, + schedule: str, + input: Optional[dict] = None, + metadata: Optional[dict] = None, + config: Optional[Config] = None, + interrupt_before: Optional[list[str]] = None, + interrupt_after: Optional[list[str]] = None, + webhook: Optional[str] = None, + multitask_strategy: Optional[str] = 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. + 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'. + + Returns: + Run: The cron run. + + Example Usage: + + 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, + "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 self.http.post(f"/threads/{thread_id}/runs/crons", json=payload) + + def create( + self, + assistant_id: str, + *, + schedule: str, + input: Optional[dict] = None, + metadata: Optional[dict] = None, + config: Optional[Config] = None, + interrupt_before: Optional[list[str]] = None, + interrupt_after: Optional[list[str]] = None, + webhook: Optional[str] = None, + multitask_strategy: Optional[str] = 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. + 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'. + + Returns: + Run: The cron run. + + Example Usage: + + 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, + "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 self.http.post("/runs/crons", json=payload) + + def delete(self, cron_id: str) -> None: + """Delete a cron. + + Args: + cron_id: The cron ID to delete. + + Returns: + None + + Example Usage: + + client.crons.delete( + cron_id="cron_to_delete" + ) + + """ # noqa: E501 + self.http.delete(f"/runs/crons/{cron_id}") + + def search( + self, + *, + assistant_id: Optional[str] = None, + thread_id: Optional[str] = None, + limit: int = 10, + offset: int = 0, + ) -> 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. + + Returns: + list[Cron]: The list of cron jobs returned by the search, + + Example Usage: + + cron_jobs = client.crons.search( + assistant_id="my_assistant_id", + thread_id="my_thread_id", + limit=5, + offset=5, + ) + print(cron_jobs) + + ---------------------------------------------------------- + + [ + { + '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, + } + payload = {k: v for k, v in payload.items() if v is not None} + return self.http.post("/runs/crons/search", json=payload) diff --git a/libs/sdk-py/langgraph_sdk/schema.py b/libs/sdk-py/langgraph_sdk/schema.py index a0fac51a0..7d97d4de6 100644 --- a/libs/sdk-py/langgraph_sdk/schema.py +++ b/libs/sdk-py/langgraph_sdk/schema.py @@ -1,5 +1,5 @@ from datetime import datetime -from typing import Any, Literal, Optional, Sequence, TypedDict, Union +from typing import Any, Literal, NamedTuple, Optional, Sequence, TypedDict, Union Metadata = Optional[dict[str, Any]] @@ -148,3 +148,8 @@ class RunCreate(TypedDict): interrupt_after: Optional[list[str]] webhook: Optional[str] multitask_strategy: Optional[MultitaskStrategy] + + +class StreamPart(NamedTuple): + event: str + data: dict diff --git a/libs/sdk-py/langgraph_sdk/utils.py b/libs/sdk-py/langgraph_sdk/utils.py new file mode 100644 index 000000000..254d636a5 --- /dev/null +++ b/libs/sdk-py/langgraph_sdk/utils.py @@ -0,0 +1,18 @@ +import os +from typing import Optional + + +def get_api_key(api_key: Optional[str] = None) -> Optional[str]: + """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 From a161a9ca97bda4bad36df32203d289c2d0f8f664 Mon Sep 17 00:00:00 2001 From: vbarda Date: Tue, 20 Aug 2024 15:44:16 -0400 Subject: [PATCH 2/8] more explicit naming --- libs/sdk-py/langgraph_sdk/client/async_client.py | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/libs/sdk-py/langgraph_sdk/client/async_client.py b/libs/sdk-py/langgraph_sdk/client/async_client.py index 7e5c872fd..57ea318d2 100644 --- a/libs/sdk-py/langgraph_sdk/client/async_client.py +++ b/libs/sdk-py/langgraph_sdk/client/async_client.py @@ -83,10 +83,10 @@ def get_client( class AsyncLangGraphClient: def __init__(self, client: httpx.AsyncClient) -> None: self.http = AsyncHttpClient(client) - self.assistants = AssistantsClient(self.http) - self.threads = ThreadsClient(self.http) - self.runs = RunsClient(self.http) - self.crons = CronClient(self.http) + self.assistants = AsyncAssistantsClient(self.http) + self.threads = AsyncThreadsClient(self.http) + self.runs = AsyncRunsClient(self.http) + self.crons = AsyncCronClient(self.http) class AsyncHttpClient: @@ -225,7 +225,7 @@ async def decode_json(r: httpx.Response) -> Any: ) -class AssistantsClient: +class AsyncAssistantsClient: def __init__(self, http: AsyncHttpClient) -> None: self.http = http @@ -555,7 +555,7 @@ class AssistantsClient: ) -class ThreadsClient: +class AsyncThreadsClient: def __init__(self, http: AsyncHttpClient) -> None: self.http = http @@ -938,7 +938,7 @@ class ThreadsClient: return await self.http.post(f"/threads/{thread_id}/history", json=payload) -class RunsClient: +class AsyncRunsClient: def __init__(self, http: AsyncHttpClient) -> None: self.http = http @@ -1465,7 +1465,7 @@ class RunsClient: await self.http.delete(f"/threads/{thread_id}/runs/{run_id}") -class CronClient: +class AsyncCronClient: def __init__(self, http_client: AsyncHttpClient) -> None: self.http = http_client From a73484d336178d81491e061191bc44a9139c784b Mon Sep 17 00:00:00 2001 From: vbarda Date: Fri, 6 Sep 2024 15:33:29 -0400 Subject: [PATCH 3/8] rename back --- .../langgraph_sdk/client/async_client.py | 36 ++++++++++--------- .../langgraph_sdk/client/sync_client.py | 31 ++++++++-------- 2 files changed, 35 insertions(+), 32 deletions(-) diff --git a/libs/sdk-py/langgraph_sdk/client/async_client.py b/libs/sdk-py/langgraph_sdk/client/async_client.py index bd5e5ce63..76d61ae39 100644 --- a/libs/sdk-py/langgraph_sdk/client/async_client.py +++ b/libs/sdk-py/langgraph_sdk/client/async_client.py @@ -46,7 +46,7 @@ def get_client( url: Optional[str] = None, api_key: Optional[str] = None, headers: Optional[dict[str, str]] = None, -) -> AsyncLangGraphClient: +) -> LangGraphClient: """Get a LangGraphClient instance. Args: @@ -68,27 +68,29 @@ def get_client( transport = httpx.ASGITransport(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(connect=5, read=60, write=60, pool=5), headers=get_headers(api_key, headers), ) - return AsyncLangGraphClient(client) + return LangGraphClient(client) -class AsyncLangGraphClient: +class LangGraphClient: def __init__(self, client: httpx.AsyncClient) -> None: - self.http = AsyncHttpClient(client) - self.assistants = AsyncAssistantsClient(self.http) - self.threads = AsyncThreadsClient(self.http) - self.runs = AsyncRunsClient(self.http) - self.crons = AsyncCronClient(self.http) + self.http = HttpClient(client) + self.assistants = AssistantsClient(self.http) + self.threads = ThreadsClient(self.http) + self.runs = RunsClient(self.http) + self.crons = CronClient(self.http) -class AsyncHttpClient: +class HttpClient: def __init__(self, client: httpx.AsyncClient) -> None: self.client = client @@ -213,8 +215,8 @@ async def decode_json(r: httpx.Response) -> Any: ) -class AsyncAssistantsClient: - def __init__(self, http: AsyncHttpClient) -> None: +class AssistantsClient: + def __init__(self, http: HttpClient) -> None: self.http = http async def get(self, assistant_id: str) -> Assistant: @@ -543,8 +545,8 @@ class AsyncAssistantsClient: ) -class AsyncThreadsClient: - def __init__(self, http: AsyncHttpClient) -> None: +class ThreadsClient: + def __init__(self, http: HttpClient) -> None: self.http = http async def get(self, thread_id: str) -> Thread: @@ -928,8 +930,8 @@ class AsyncThreadsClient: return await self.http.post(f"/threads/{thread_id}/history", json=payload) -class AsyncRunsClient: - def __init__(self, http: AsyncHttpClient) -> None: +class RunsClient: + def __init__(self, http: HttpClient) -> None: self.http = http @overload @@ -1498,8 +1500,8 @@ class AsyncRunsClient: await self.http.delete(f"/threads/{thread_id}/runs/{run_id}") -class AsyncCronClient: - def __init__(self, http_client: AsyncHttpClient) -> None: +class CronClient: + def __init__(self, http_client: HttpClient) -> None: self.http = http_client async def create_for_thread( diff --git a/libs/sdk-py/langgraph_sdk/client/sync_client.py b/libs/sdk-py/langgraph_sdk/client/sync_client.py index 096f5ba56..4927e1cc0 100644 --- a/libs/sdk-py/langgraph_sdk/client/sync_client.py +++ b/libs/sdk-py/langgraph_sdk/client/sync_client.py @@ -41,7 +41,7 @@ logger = logging.getLogger(__name__) def get_client( *, url: Optional[str] = None, api_key: Optional[str] = None -) -> LangGraphClient: +) -> SyncLangGraphClient: """Get a LangGraphClient instance. Args: @@ -56,6 +56,7 @@ def get_client( if url is None: url = "http://localhost:8123" + transport = httpx.HTTPTransport(retries=5) headers = { "User-Agent": f"langgraph-sdk-py/{langgraph_sdk.__version__}", @@ -66,19 +67,19 @@ def get_client( timeout=httpx.Timeout(connect=5, read=60, write=60, pool=5), headers=get_headers(api_key, headers), ) - return LangGraphClient(client) + return SyncLangGraphClient(client) -class LangGraphClient: +class SyncLangGraphClient: def __init__(self, client: httpx.Client) -> None: - self.http = HttpClient(client) - self.assistants = AssistantsClient(self.http) - self.threads = ThreadsClient(self.http) - self.runs = RunsClient(self.http) + self.http = SyncHttpClient(client) + self.assistants = SyncAssistantsClient(self.http) + self.threads = SyncThreadsClient(self.http) + self.runs = SyncRunsClient(self.http) self.crons = CronClient(self.http) -class HttpClient: +class SyncHttpClient: def __init__(self, client: httpx.Client) -> None: self.client = client @@ -197,8 +198,8 @@ def decode_json(r: httpx.Response) -> Any: return orjson.loads(body if body else None) -class AssistantsClient: - def __init__(self, http: HttpClient) -> None: +class SyncAssistantsClient: + def __init__(self, http: SyncHttpClient) -> None: self.http = http def get(self, assistant_id: str) -> Assistant: @@ -527,8 +528,8 @@ class AssistantsClient: ) -class ThreadsClient: - def __init__(self, http: HttpClient) -> None: +class SyncThreadsClient: + def __init__(self, http: SyncHttpClient) -> None: self.http = http def get(self, thread_id: str) -> Thread: @@ -908,8 +909,8 @@ class ThreadsClient: return self.http.post(f"/threads/{thread_id}/history", json=payload) -class RunsClient: - def __init__(self, http: HttpClient) -> None: +class SyncRunsClient: + def __init__(self, http: SyncHttpClient) -> None: self.http = http @overload @@ -1432,7 +1433,7 @@ class RunsClient: class CronClient: - def __init__(self, http_client: HttpClient) -> None: + def __init__(self, http_client: SyncHttpClient) -> None: self.http = http_client def create_for_thread( From 7233b06c2bd72f3b2ae2b6d44154a48a9336c674 Mon Sep 17 00:00:00 2001 From: vbarda Date: Fri, 6 Sep 2024 16:07:14 -0400 Subject: [PATCH 4/8] update sync --- .../langgraph_sdk/client/async_client.py | 21 ++-- .../langgraph_sdk/client/sync_client.py | 105 ++++++++++++++---- 2 files changed, 96 insertions(+), 30 deletions(-) diff --git a/libs/sdk-py/langgraph_sdk/client/async_client.py b/libs/sdk-py/langgraph_sdk/client/async_client.py index 76d61ae39..1ed1656f7 100644 --- a/libs/sdk-py/langgraph_sdk/client/async_client.py +++ b/libs/sdk-py/langgraph_sdk/client/async_client.py @@ -666,6 +666,7 @@ class ThreadsClient: Args: metadata: Thread metadata to search for. + values: Thread values to search for. status: Status to search for. Must be one of 'idle', 'busy', or 'interrupted'. limit: Limit on number of threads to return. @@ -968,8 +969,8 @@ class RunsClient: interrupt_after: Optional[list[str]] = None, feedback_keys: Optional[list[str]] = None, on_disconnect: Optional[DisconnectMode] = None, - webhook: Optional[str] = None, on_completion: Optional[OnCompletionBehavior] = None, + webhook: Optional[str] = None, ) -> AsyncIterator[StreamPart]: ... @@ -987,9 +988,9 @@ class RunsClient: interrupt_after: Optional[list[str]] = None, feedback_keys: Optional[list[str]] = None, on_disconnect: Optional[DisconnectMode] = None, + on_completion: Optional[OnCompletionBehavior] = None, webhook: Optional[str] = None, multitask_strategy: Optional[MultitaskStrategy] = None, - on_completion: Optional[OnCompletionBehavior] = None, ) -> AsyncIterator[StreamPart]: """Create a run and stream the results. @@ -1008,11 +1009,13 @@ class RunsClient: 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'. - on_disconnect: The disconnect mode to use. - Must be one of 'cancel' or 'continue'. Returns: AsyncIterator[StreamPart]: Asynchronous iterator of stream results. @@ -1131,6 +1134,8 @@ class RunsClient: 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'. Returns: Run: The created background run. @@ -1275,8 +1280,8 @@ class RunsClient: interrupt_after: Optional[list[str]] = None, webhook: Optional[str] = None, on_disconnect: Optional[DisconnectMode] = None, - multitask_strategy: Optional[MultitaskStrategy] = None, on_completion: Optional[OnCompletionBehavior] = None, + multitask_strategy: Optional[MultitaskStrategy] = None, ) -> Union[list[dict], dict[str, Any]]: """Create a run, wait until it finishes and return the final state. @@ -1292,10 +1297,12 @@ class RunsClient: 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_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'. Returns: Union[list[dict], dict[str, Any]]: The output of the run. diff --git a/libs/sdk-py/langgraph_sdk/client/sync_client.py b/libs/sdk-py/langgraph_sdk/client/sync_client.py index 4927e1cc0..a7563f01e 100644 --- a/libs/sdk-py/langgraph_sdk/client/sync_client.py +++ b/libs/sdk-py/langgraph_sdk/client/sync_client.py @@ -17,14 +17,15 @@ import httpx_sse import orjson from httpx._types import QueryParamTypes -import langgraph_sdk from langgraph_sdk.schema import ( Assistant, Config, Cron, + DisconnectMode, GraphSchema, - Metadata, + Json, MultitaskStrategy, + OnCompletionBehavior, OnConflictBehavior, Run, RunCreate, @@ -40,7 +41,10 @@ logger = logging.getLogger(__name__) def get_client( - *, url: Optional[str] = None, api_key: Optional[str] = None + *, + url: Optional[str] = None, + api_key: Optional[str] = None, + headers: Optional[dict[str, str]] = None, ) -> SyncLangGraphClient: """Get a LangGraphClient instance. @@ -52,15 +56,13 @@ def get_client( 2. LANGGRAPH_API_KEY 3. LANGSMITH_API_KEY 4. LANGCHAIN_API_KEY + headers: Optional custom headers """ if url is None: url = "http://localhost:8123" transport = httpx.HTTPTransport(retries=5) - headers = { - "User-Agent": f"langgraph-sdk-py/{langgraph_sdk.__version__}", - } client = httpx.Client( base_url=url, transport=transport, @@ -89,7 +91,7 @@ class SyncHttpClient: try: r.raise_for_status() except httpx.HTTPStatusError as e: - body = (r.read()).decode() + body = r.read().decode() if sys.version_info >= (3, 11): e.add_note(body) else: @@ -107,7 +109,7 @@ class SyncHttpClient: try: r.raise_for_status() except httpx.HTTPStatusError as e: - body = (r.read()).decode() + body = r.read().decode() if sys.version_info >= (3, 11): e.add_note(body) else: @@ -122,7 +124,7 @@ class SyncHttpClient: try: r.raise_for_status() except httpx.HTTPStatusError as e: - body = (r.read()).decode() + body = r.read().decode() if sys.version_info >= (3, 11): e.add_note(body) else: @@ -137,7 +139,7 @@ class SyncHttpClient: try: r.raise_for_status() except httpx.HTTPStatusError as e: - body = (r.read()).decode() + body = r.read().decode() if sys.version_info >= (3, 11): e.add_note(body) else: @@ -151,7 +153,7 @@ class SyncHttpClient: try: r.raise_for_status() except httpx.HTTPStatusError as e: - body = (r.read()).decode() + body = r.read().decode() if sys.version_info >= (3, 11): e.add_note(body) else: @@ -169,7 +171,7 @@ class SyncHttpClient: try: sse.response.raise_for_status() except httpx.HTTPStatusError as e: - body = (sse.response.read()).decode() + body = sse.response.read().decode() if sys.version_info >= (3, 11): e.add_note(body) else: @@ -378,7 +380,7 @@ class SyncAssistantsClient: graph_id: Optional[str], config: Optional[Config] = None, *, - metadata: Metadata = None, + metadata: Json = None, assistant_id: Optional[str] = None, if_exists: Optional[OnConflictBehavior] = None, ) -> Assistant: @@ -426,7 +428,7 @@ class SyncAssistantsClient: *, graph_id: Optional[str] = None, config: Optional[Config] = None, - metadata: Metadata = None, + metadata: Json = None, ) -> Assistant: """Update an assistant. @@ -488,7 +490,7 @@ class SyncAssistantsClient: def search( self, *, - metadata: Metadata = None, + metadata: Json = None, graph_id: Optional[str] = None, limit: int = 10, offset: int = 0, @@ -564,7 +566,7 @@ class SyncThreadsClient: def create( self, *, - metadata: Metadata = None, + metadata: Json = None, thread_id: Optional[str] = None, if_exists: Optional[OnConflictBehavior] = None, ) -> Thread: @@ -637,7 +639,8 @@ class SyncThreadsClient: def search( self, *, - metadata: Metadata = None, + metadata: Json = None, + values: Json = None, status: Optional[ThreadStatus] = None, limit: int = 10, offset: int = 0, @@ -646,6 +649,7 @@ class SyncThreadsClient: Args: metadata: Thread metadata to search for. + values: Thread values to search for. status: Status to search for. Must be one of 'idle', 'busy', or 'interrupted'. limit: Limit on number of threads to return. @@ -670,6 +674,8 @@ class SyncThreadsClient: } if metadata: payload["metadata"] = metadata + if values: + payload["values"] = values if status: payload["status"] = status return self.http.post( @@ -816,14 +822,14 @@ class SyncThreadsClient: values: The values to update to the state. as_node: Update the state as if this node had just executed. - checkpoint_id: The ID of the checkpoint to get the state of. + checkpoint_id: The ID of the checkpoint to update the state of. Returns: None Example Usage: - client.threads.get_state( + client.threads.update_state( thread_id="my_thread_id", values={"messages":[{"role": "user", "content": "hello!"}]}, as_node="my_node", @@ -927,6 +933,8 @@ class SyncRunsClient: interrupt_before: Optional[list[str]] = None, interrupt_after: Optional[list[str]] = None, feedback_keys: Optional[list[str]] = None, + on_disconnect: Optional[DisconnectMode] = None, + webhook: Optional[str] = None, multitask_strategy: Optional[MultitaskStrategy] = None, ) -> Iterator[StreamPart]: ... @@ -944,6 +952,9 @@ class SyncRunsClient: interrupt_before: Optional[list[str]] = None, interrupt_after: Optional[list[str]] = None, feedback_keys: Optional[list[str]] = None, + on_disconnect: Optional[DisconnectMode] = None, + on_completion: Optional[OnCompletionBehavior] = None, + webhook: Optional[str] = None, ) -> Iterator[StreamPart]: ... @@ -960,6 +971,8 @@ class SyncRunsClient: interrupt_before: Optional[list[str]] = None, interrupt_after: Optional[list[str]] = None, feedback_keys: Optional[list[str]] = None, + on_disconnect: Optional[DisconnectMode] = None, + on_completion: Optional[OnCompletionBehavior] = None, webhook: Optional[str] = None, multitask_strategy: Optional[MultitaskStrategy] = None, ) -> Iterator[StreamPart]: @@ -980,6 +993,10 @@ class SyncRunsClient: 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'. @@ -1025,6 +1042,8 @@ class SyncRunsClient: "webhook": webhook, "checkpoint_id": checkpoint_id, "multitask_strategy": multitask_strategy, + "on_disconnect": on_disconnect, + "on_completion": on_completion, } endpoint = ( f"/threads/{thread_id}/runs/stream" @@ -1047,6 +1066,7 @@ class SyncRunsClient: interrupt_before: Optional[list[str]] = None, interrupt_after: Optional[list[str]] = None, webhook: Optional[str] = None, + on_completion: Optional[OnCompletionBehavior] = None, ) -> Run: ... @@ -1080,6 +1100,7 @@ class SyncRunsClient: interrupt_after: Optional[list[str]] = None, webhook: Optional[str] = None, multitask_strategy: Optional[MultitaskStrategy] = None, + on_completion: Optional[OnCompletionBehavior] = None, ) -> Run: """Create a background run. @@ -1099,6 +1120,8 @@ class SyncRunsClient: 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'. Returns: Run: The created background run. @@ -1178,6 +1201,7 @@ class SyncRunsClient: "webhook": webhook, "checkpoint_id": checkpoint_id, "multitask_strategy": multitask_strategy, + "on_completion": on_completion, } payload = {k: v for k, v in payload.items() if v is not None} if thread_id: @@ -1206,6 +1230,8 @@ class SyncRunsClient: checkpoint_id: Optional[str] = None, interrupt_before: Optional[list[str]] = None, interrupt_after: Optional[list[str]] = None, + webhook: Optional[str] = None, + on_disconnect: Optional[DisconnectMode] = None, multitask_strategy: Optional[MultitaskStrategy] = None, ) -> Union[list[dict], dict[str, Any]]: ... @@ -1221,6 +1247,9 @@ class SyncRunsClient: config: Optional[Config] = None, interrupt_before: Optional[list[str]] = None, interrupt_after: Optional[list[str]] = None, + webhook: Optional[str] = None, + on_disconnect: Optional[DisconnectMode] = None, + on_completion: Optional[OnCompletionBehavior] = None, ) -> Union[list[dict], dict[str, Any]]: ... @@ -1236,6 +1265,8 @@ class SyncRunsClient: interrupt_before: Optional[list[str]] = None, interrupt_after: Optional[list[str]] = None, webhook: Optional[str] = None, + on_disconnect: Optional[DisconnectMode] = None, + on_completion: Optional[OnCompletionBehavior] = None, multitask_strategy: Optional[MultitaskStrategy] = None, ) -> Union[list[dict], dict[str, Any]]: """Create a run, wait until it finishes and return the final state. @@ -1250,12 +1281,16 @@ class SyncRunsClient: config: The configuration for the assistant. checkpoint_id: The checkpoint to start streaming from. 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_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'. Returns: Union[list[dict], dict[str, Any]]: The output of the run. @@ -1315,6 +1350,8 @@ class SyncRunsClient: "webhook": webhook, "checkpoint_id": checkpoint_id, "multitask_strategy": multitask_strategy, + "on_disconnect": on_disconnect, + "on_completion": on_completion, } endpoint = ( f"/threads/{thread_id}/runs/wait" if thread_id is not None else "/runs/wait" @@ -1391,8 +1428,8 @@ class SyncRunsClient: json=None, ) - def join(self, thread_id: str, run_id: str) -> None: - """Block until a run is done. + def join(self, thread_id: str, run_id: str) -> dict: + """Block until a run is done. Returns the final state of the thread. Args: thread_id: The thread ID to join. @@ -1411,6 +1448,28 @@ class SyncRunsClient: """ # noqa: E501 return self.http.get(f"/threads/{thread_id}/runs/{run_id}/join") + def join_stream(self, thread_id: str, run_id: str) -> 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. + + Returns: + None + + Example Usage: + + await client.runs.join( + thread_id="thread_id_to_join", + run_id="run_id_to_join" + ) + + """ # noqa: E501 + return self.http.stream(f"/threads/{thread_id}/runs/{run_id}/stream", "GET") + def delete(self, thread_id: str, run_id: str) -> None: """Delete a run. From 6fd6ef43f6cdd7e3b47deea4032f5ca72f5daf1a Mon Sep 17 00:00:00 2001 From: vbarda Date: Fri, 6 Sep 2024 16:28:05 -0400 Subject: [PATCH 5/8] lint --- .../langgraph_sdk/client/async_client.py | 18 ++++++------------ .../sdk-py/langgraph_sdk/client/sync_client.py | 18 ++++++------------ 2 files changed, 12 insertions(+), 24 deletions(-) diff --git a/libs/sdk-py/langgraph_sdk/client/async_client.py b/libs/sdk-py/langgraph_sdk/client/async_client.py index c1710e32e..c9eb25cf0 100644 --- a/libs/sdk-py/langgraph_sdk/client/async_client.py +++ b/libs/sdk-py/langgraph_sdk/client/async_client.py @@ -952,8 +952,7 @@ class RunsClient: on_disconnect: Optional[DisconnectMode] = None, webhook: Optional[str] = None, multitask_strategy: Optional[MultitaskStrategy] = None, - ) -> AsyncIterator[StreamPart]: - ... + ) -> AsyncIterator[StreamPart]: ... @overload def stream( @@ -971,8 +970,7 @@ class RunsClient: on_disconnect: Optional[DisconnectMode] = None, on_completion: Optional[OnCompletionBehavior] = None, webhook: Optional[str] = None, - ) -> AsyncIterator[StreamPart]: - ... + ) -> AsyncIterator[StreamPart]: ... def stream( self, @@ -1083,8 +1081,7 @@ class RunsClient: interrupt_after: Optional[list[str]] = None, webhook: Optional[str] = None, on_completion: Optional[OnCompletionBehavior] = None, - ) -> Run: - ... + ) -> Run: ... @overload async def create( @@ -1100,8 +1097,7 @@ class RunsClient: interrupt_after: Optional[list[str]] = None, webhook: Optional[str] = None, multitask_strategy: Optional[MultitaskStrategy] = None, - ) -> Run: - ... + ) -> Run: ... async def create( self, @@ -1247,8 +1243,7 @@ class RunsClient: webhook: Optional[str] = None, on_disconnect: Optional[DisconnectMode] = None, multitask_strategy: Optional[MultitaskStrategy] = None, - ) -> Union[list[dict], dict[str, Any]]: - ... + ) -> Union[list[dict], dict[str, Any]]: ... @overload async def wait( @@ -1264,8 +1259,7 @@ class RunsClient: webhook: Optional[str] = None, on_disconnect: Optional[DisconnectMode] = None, on_completion: Optional[OnCompletionBehavior] = None, - ) -> Union[list[dict], dict[str, Any]]: - ... + ) -> Union[list[dict], dict[str, Any]]: ... async def wait( self, diff --git a/libs/sdk-py/langgraph_sdk/client/sync_client.py b/libs/sdk-py/langgraph_sdk/client/sync_client.py index a7563f01e..f6c75af6d 100644 --- a/libs/sdk-py/langgraph_sdk/client/sync_client.py +++ b/libs/sdk-py/langgraph_sdk/client/sync_client.py @@ -936,8 +936,7 @@ class SyncRunsClient: on_disconnect: Optional[DisconnectMode] = None, webhook: Optional[str] = None, multitask_strategy: Optional[MultitaskStrategy] = None, - ) -> Iterator[StreamPart]: - ... + ) -> Iterator[StreamPart]: ... @overload def stream( @@ -955,8 +954,7 @@ class SyncRunsClient: on_disconnect: Optional[DisconnectMode] = None, on_completion: Optional[OnCompletionBehavior] = None, webhook: Optional[str] = None, - ) -> Iterator[StreamPart]: - ... + ) -> Iterator[StreamPart]: ... def stream( self, @@ -1067,8 +1065,7 @@ class SyncRunsClient: interrupt_after: Optional[list[str]] = None, webhook: Optional[str] = None, on_completion: Optional[OnCompletionBehavior] = None, - ) -> Run: - ... + ) -> Run: ... @overload def create( @@ -1084,8 +1081,7 @@ class SyncRunsClient: interrupt_after: Optional[list[str]] = None, webhook: Optional[str] = None, multitask_strategy: Optional[MultitaskStrategy] = None, - ) -> Run: - ... + ) -> Run: ... def create( self, @@ -1233,8 +1229,7 @@ class SyncRunsClient: webhook: Optional[str] = None, on_disconnect: Optional[DisconnectMode] = None, multitask_strategy: Optional[MultitaskStrategy] = None, - ) -> Union[list[dict], dict[str, Any]]: - ... + ) -> Union[list[dict], dict[str, Any]]: ... @overload def wait( @@ -1250,8 +1245,7 @@ class SyncRunsClient: webhook: Optional[str] = None, on_disconnect: Optional[DisconnectMode] = None, on_completion: Optional[OnCompletionBehavior] = None, - ) -> Union[list[dict], dict[str, Any]]: - ... + ) -> Union[list[dict], dict[str, Any]]: ... def wait( self, From 142db1f0205af1dce95e1f5c72032784220173d5 Mon Sep 17 00:00:00 2001 From: vbarda Date: Fri, 6 Sep 2024 16:37:56 -0400 Subject: [PATCH 6/8] fix links in the reference --- docs/docs/cloud/reference/sdk/python_sdk_ref.md | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/docs/docs/cloud/reference/sdk/python_sdk_ref.md b/docs/docs/cloud/reference/sdk/python_sdk_ref.md index 9c362ddea..e44485675 100644 --- a/docs/docs/cloud/reference/sdk/python_sdk_ref.md +++ b/docs/docs/cloud/reference/sdk/python_sdk_ref.md @@ -16,14 +16,14 @@ client = get_client(url="http://localhost:8123") assistants = await client.assistants.get(assistant_id="some_uuid") ``` -::: langgraph_sdk.client.get_client +::: langgraph_sdk.client.async_client.get_client handler: python ## LangGraphClient `LangGraphClient` is the top-level client for accessing `AssistantsClient`, `ThreadsClient`, `RunsClient`, and `CronClient`. -::: langgraph_sdk.client.LangGraphClient +::: langgraph_sdk.client.async_client.LangGraphClient handler: python ## AssistantsClient @@ -36,7 +36,7 @@ client = get_client(url="http://localhost:8123") await client.assistants.() ``` -::: langgraph_sdk.client.AssistantsClient +::: langgraph_sdk.client.async_client.AssistantsClient handler: python ## ThreadsClient @@ -49,7 +49,7 @@ client = get_client(url="http://localhost:8123") await client.threads.() ``` -::: langgraph_sdk.client.ThreadsClient +::: langgraph_sdk.client.async_client.ThreadsClient handler: python ## RunsClient @@ -62,7 +62,7 @@ client = get_client(url="http://localhost:8123") await client.runs.() ``` -::: langgraph_sdk.client.RunsClient +::: langgraph_sdk.client.async_client.RunsClient handler: python ## CronClient @@ -75,5 +75,5 @@ client = get_client(url="http://localhost:8123") await client.crons.() ``` -::: langgraph_sdk.client.CronClient +::: langgraph_sdk.client.async_client.CronClient handler: python From 2d741c7cbf05eda42f9094a568f28e4435c519f7 Mon Sep 17 00:00:00 2001 From: vbarda Date: Mon, 23 Sep 2024 17:15:02 -0400 Subject: [PATCH 7/8] update --- .../langgraph_sdk/client/async_client.py | 1 + .../langgraph_sdk/client/sync_client.py | 249 ++++++++++++------ 2 files changed, 165 insertions(+), 85 deletions(-) diff --git a/libs/sdk-py/langgraph_sdk/client/async_client.py b/libs/sdk-py/langgraph_sdk/client/async_client.py index f4257130e..2c4083b37 100644 --- a/libs/sdk-py/langgraph_sdk/client/async_client.py +++ b/libs/sdk-py/langgraph_sdk/client/async_client.py @@ -463,6 +463,7 @@ class AssistantsClient: 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 add to assistant. + name: The new name for the assistant. Returns: Assistant: The updated assistant. diff --git a/libs/sdk-py/langgraph_sdk/client/sync_client.py b/libs/sdk-py/langgraph_sdk/client/sync_client.py index f6c75af6d..f41a04c2e 100644 --- a/libs/sdk-py/langgraph_sdk/client/sync_client.py +++ b/libs/sdk-py/langgraph_sdk/client/sync_client.py @@ -19,6 +19,8 @@ from httpx._types import QueryParamTypes from langgraph_sdk.schema import ( Assistant, + AssistantVersion, + Checkpoint, Config, Cron, DisconnectMode, @@ -383,6 +385,7 @@ class SyncAssistantsClient: metadata: Json = None, assistant_id: Optional[str] = None, if_exists: Optional[OnConflictBehavior] = None, + name: Optional[str] = None, ) -> Assistant: """Create a new assistant. @@ -395,6 +398,7 @@ class SyncAssistantsClient: 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. Returns: Assistant: The created assistant. @@ -406,7 +410,8 @@ class SyncAssistantsClient: config={"configurable": {"model_name": "openai"}}, metadata={"number":1}, assistant_id="my-assistant-id", - if_exists="do_nothing" + if_exists="do_nothing", + name="my_name" ) """ # noqa: E501 payload: Dict[str, Any] = { @@ -420,6 +425,8 @@ class SyncAssistantsClient: payload["assistant_id"] = assistant_id if if_exists: payload["if_exists"] = if_exists + if name: + payload["name"] = name return self.http.post("/assistants", json=payload) def update( @@ -429,6 +436,7 @@ class SyncAssistantsClient: graph_id: Optional[str] = None, config: Optional[Config] = None, metadata: Json = None, + name: Optional[str] = None, ) -> Assistant: """Update an assistant. @@ -440,6 +448,7 @@ class SyncAssistantsClient: 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 add to assistant. + name: The new name for the assistant. Returns: Assistant: The updated assistant. @@ -461,6 +470,8 @@ class SyncAssistantsClient: payload["config"] = config if metadata: payload["metadata"] = metadata + if name: + payload["name"] = name return self.http.patch( f"/assistants/{assistant_id}", json=payload, @@ -529,6 +540,60 @@ class SyncAssistantsClient: json=payload, ) + def get_versions( + self, + assistant_id: str, + metadata: Json = None, + limit: int = 10, + offset: int = 0, + ) -> list[AssistantVersion]: + """List all versions of an assistant. + + Args: + assistant_id: The assistant ID to delete. + + Returns: + list[Assistant]: A list of assistants. + + Example Usage: + + 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 self.http.post(f"/assistants/{assistant_id}/versions", json=payload) + + def set_latest(self, assistant_id: str, version: int) -> Assistant: + """Change the version of an assistant. + + Args: + assistant_id: The assistant ID to delete. + version: The version to change to. + + Returns: + Assistant: Assistant Object. + + Example Usage: + + new_version_assistant = await 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) + class SyncThreadsClient: def __init__(self, http: SyncHttpClient) -> None: @@ -702,13 +767,19 @@ class SyncThreadsClient: return self.http.post(f"/threads/{thread_id}/copy", json=None) def get_state( - self, thread_id: str, checkpoint_id: Optional[str] = None + self, + thread_id: str, + checkpoint: Optional[Checkpoint] = None, + checkpoint_id: Optional[str] = None, # deprecated + *, + subgraphs: bool = False, ) -> ThreadState: """Get the state of a thread. Args: thread_id: The ID of the thread to get the state of. - checkpoint_id: The ID of the checkpoint to get the state of. + checkpoint: The checkpoint to get the state of. + subgraphs: Include subgraphs in the state. Returns: ThreadState: the thread of the state. @@ -750,15 +821,12 @@ class SyncThreadsClient: ] }, 'next': [], - 'config': + 'checkpoint': { - 'configurable': - { - 'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2', - 'checkpoint_ns': '', - 'checkpoint_id': '1ef4a9b8-e6fb-67b1-8001-abd5184439d1' - } - }, + 'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2', + 'checkpoint_ns': '', + 'checkpoint_id': '1ef4a9b8-e6fb-67b1-8001-abd5184439d1' + } 'metadata': { 'step': 1, @@ -768,20 +836,20 @@ class SyncThreadsClient: { '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': [] - } - ] + '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, @@ -792,20 +860,28 @@ class SyncThreadsClient: 'created_at': '2024-07-25T15:35:44.184703+00:00', 'parent_config': { - 'configurable': - { - 'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2', - 'checkpoint_ns': '', - 'checkpoint_id': '1ef4a9b8-d80d-6fa7-8000-9300467fad0f' - } + 'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2', + 'checkpoint_ns': '', + 'checkpoint_id': '1ef4a9b8-d80d-6fa7-8000-9300467fad0f' } } """ # noqa: E501 - if checkpoint_id: - return self.http.get(f"/threads/{thread_id}/state/{checkpoint_id}") + if checkpoint: + return self.http.post( + f"/threads/{thread_id}/state/checkpoint", + json={"checkpoint": checkpoint, "subgraphs": subgraphs}, + ) + elif checkpoint_id: + return self.http.get( + f"/threads/{thread_id}/state/{checkpoint_id}", + params={"subgraphs": subgraphs}, + ) else: - return self.http.get(f"/threads/{thread_id}/state") + return self.http.get( + f"/threads/{thread_id}/state", + params={"subgraphs": subgraphs}, + ) def update_state( self, @@ -813,7 +889,8 @@ class SyncThreadsClient: values: dict, *, as_node: Optional[str] = None, - checkpoint_id: Optional[str] = None, + checkpoint: Optional[Checkpoint] = None, + checkpoint_id: Optional[str] = None, # deprecated ) -> None: """Update the state of a thread. @@ -821,19 +898,17 @@ class SyncThreadsClient: thread_id: The ID of the thread to update. values: The values to update to the state. as_node: Update the state as if this node had just executed. - - checkpoint_id: The ID of the checkpoint to update the state of. + checkpoint: The checkpoint to update the state of. Returns: None Example Usage: - client.threads.update_state( + await client.threads.update_state( thread_id="my_thread_id", values={"messages":[{"role": "user", "content": "hello!"}]}, as_node="my_node", - checkpoint_id="my_checkpoint_id" ) """ # noqa: E501 @@ -842,41 +917,12 @@ class SyncThreadsClient: } 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) - def patch_state( - self, - thread_id: Union[str, Config], - metadata: dict, - ) -> None: - """Patch the state of a thread. - - Args: - thread_id: The ID of the thread to get the state of. - metadata: The metadata to assign to the state. - - Returns: - None - - Example Usage: - - client.threads.patch_state( - thread_id="my_thread_id", - metadata={"name":"new_name"}, - ) - - """ # noqa: E501 - if isinstance(thread_id, dict): - thread_id_: str = thread_id["configurable"]["thread_id"] - else: - thread_id_ = thread_id - return self.http.patch( - f"/threads/{thread_id_}/state", - json={"metadata": metadata}, - ) - def get_history( self, thread_id: str, @@ -927,8 +973,10 @@ class SyncRunsClient: *, input: Optional[dict] = None, stream_mode: Union[StreamMode, list[StreamMode]] = "values", + stream_subgraphs: bool = False, metadata: Optional[dict] = None, config: Optional[Config] = None, + checkpoint: Optional[Checkpoint] = None, checkpoint_id: Optional[str] = None, interrupt_before: Optional[list[str]] = None, interrupt_after: Optional[list[str]] = None, @@ -936,6 +984,7 @@ class SyncRunsClient: on_disconnect: Optional[DisconnectMode] = None, webhook: Optional[str] = None, multitask_strategy: Optional[MultitaskStrategy] = None, + after_seconds: Optional[int] = None, ) -> Iterator[StreamPart]: ... @overload @@ -946,6 +995,7 @@ class SyncRunsClient: *, input: Optional[dict] = None, stream_mode: Union[StreamMode, list[StreamMode]] = "values", + stream_subgraphs: bool = False, metadata: Optional[dict] = None, config: Optional[Config] = None, interrupt_before: Optional[list[str]] = None, @@ -954,6 +1004,7 @@ class SyncRunsClient: on_disconnect: Optional[DisconnectMode] = None, on_completion: Optional[OnCompletionBehavior] = None, webhook: Optional[str] = None, + after_seconds: Optional[int] = None, ) -> Iterator[StreamPart]: ... def stream( @@ -963,8 +1014,10 @@ class SyncRunsClient: *, input: Optional[dict] = None, stream_mode: Union[StreamMode, list[StreamMode]] = "values", + stream_subgraphs: bool = False, metadata: Optional[dict] = None, config: Optional[Config] = None, + checkpoint: Optional[Checkpoint] = None, checkpoint_id: Optional[str] = None, interrupt_before: Optional[list[str]] = None, interrupt_after: Optional[list[str]] = None, @@ -973,6 +1026,7 @@ class SyncRunsClient: on_completion: Optional[OnCompletionBehavior] = None, webhook: Optional[str] = None, multitask_strategy: Optional[MultitaskStrategy] = None, + after_seconds: Optional[int] = None, ) -> Iterator[StreamPart]: """Create a run and stream the results. @@ -983,13 +1037,12 @@ class SyncRunsClient: If using graph name, will default to first assistant created from that graph. input: The input to the graph. stream_mode: The stream mode(s) to use. + stream_subgraphs: Whether to stream output from subgraphs. metadata: Metadata to assign to the run. config: The configuration for the assistant. - checkpoint_id: The checkpoint to start streaming from. + checkpoint: The checkpoint to resume from. 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'. @@ -998,9 +1051,11 @@ class SyncRunsClient: 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'. + after_seconds: The number of seconds to wait before starting the run. + Use to schedule future runs. Returns: - Iterator[StreamPart]: Asynchronous iterator of stream results. + Iterator[StreamPart]: Iterator of stream results. Example Usage: @@ -1011,7 +1066,6 @@ class SyncRunsClient: stream_mode=["values","debug"], metadata={"name":"my_run"}, config={"configurable": {"model_name": "anthropic"}}, - checkpoint_id="my_checkpoint", 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"], @@ -1033,15 +1087,18 @@ class SyncRunsClient: "config": config, "metadata": metadata, "stream_mode": stream_mode, + "stream_subgraphs": stream_subgraphs, "assistant_id": assistant_id, "interrupt_before": interrupt_before, "interrupt_after": interrupt_after, "feedback_keys": feedback_keys, "webhook": webhook, + "checkpoint": checkpoint, "checkpoint_id": checkpoint_id, "multitask_strategy": multitask_strategy, "on_disconnect": on_disconnect, "on_completion": on_completion, + "after_seconds": after_seconds, } endpoint = ( f"/threads/{thread_id}/runs/stream" @@ -1059,12 +1116,15 @@ class SyncRunsClient: assistant_id: str, *, input: Optional[dict] = None, + stream_mode: Union[StreamMode, list[StreamMode]] = "values", + stream_subgraphs: bool = False, metadata: Optional[dict] = None, config: Optional[Config] = None, interrupt_before: Optional[list[str]] = None, interrupt_after: Optional[list[str]] = None, webhook: Optional[str] = None, on_completion: Optional[OnCompletionBehavior] = None, + after_seconds: Optional[int] = None, ) -> Run: ... @overload @@ -1074,13 +1134,17 @@ class SyncRunsClient: assistant_id: str, *, input: Optional[dict] = None, + stream_mode: Union[StreamMode, list[StreamMode]] = "values", + stream_subgraphs: bool = False, metadata: Optional[dict] = None, config: Optional[Config] = None, + checkpoint: Optional[Checkpoint] = None, checkpoint_id: Optional[str] = None, interrupt_before: Optional[list[str]] = None, interrupt_after: Optional[list[str]] = None, webhook: Optional[str] = None, multitask_strategy: Optional[MultitaskStrategy] = None, + after_seconds: Optional[int] = None, ) -> Run: ... def create( @@ -1089,14 +1153,18 @@ class SyncRunsClient: assistant_id: str, *, input: Optional[dict] = None, + stream_mode: Union[StreamMode, list[StreamMode]] = "values", + stream_subgraphs: bool = False, metadata: Optional[dict] = None, config: Optional[Config] = None, + checkpoint: Optional[Checkpoint] = None, checkpoint_id: Optional[str] = None, interrupt_before: Optional[list[str]] = None, interrupt_after: Optional[list[str]] = None, webhook: Optional[str] = None, multitask_strategy: Optional[MultitaskStrategy] = None, on_completion: Optional[OnCompletionBehavior] = None, + after_seconds: Optional[int] = None, ) -> Run: """Create a background run. @@ -1106,18 +1174,20 @@ class SyncRunsClient: 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. + stream_mode: The stream mode(s) to use. + stream_subgraphs: Whether to stream output from subgraphs. metadata: Metadata to assign to the run. config: The configuration for the assistant. - checkpoint_id: The checkpoint to start streaming from. + checkpoint: The checkpoint to resume from. 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'. + after_seconds: The number of seconds to wait before starting the run. + Use to schedule future runs. Returns: Run: The created background run. @@ -1130,7 +1200,6 @@ class SyncRunsClient: input={"messages": [{"role": "user", "content": "hello!"}]}, metadata={"name":"my_run"}, config={"configurable": {"model_name": "openai"}}, - checkpoint_id="my_checkpoint", 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", @@ -1189,15 +1258,19 @@ class SyncRunsClient: """ # noqa: E501 payload = { "input": input, + "stream_mode": stream_mode, + "stream_subgraphs": stream_subgraphs, "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, "on_completion": on_completion, + "after_seconds": after_seconds, } payload = {k: v for k, v in payload.items() if v is not None} if thread_id: @@ -1206,7 +1279,7 @@ class SyncRunsClient: return self.http.post("/runs", json=payload) def create_batch(self, payloads: list[RunCreate]) -> list[Run]: - """Create a batch of background runs.""" + """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} @@ -1223,12 +1296,14 @@ class SyncRunsClient: input: Optional[dict] = None, metadata: Optional[dict] = None, config: Optional[Config] = None, + checkpoint: Optional[Checkpoint] = None, checkpoint_id: Optional[str] = None, interrupt_before: Optional[list[str]] = None, interrupt_after: Optional[list[str]] = None, webhook: Optional[str] = None, on_disconnect: Optional[DisconnectMode] = None, multitask_strategy: Optional[MultitaskStrategy] = None, + after_seconds: Optional[int] = None, ) -> Union[list[dict], dict[str, Any]]: ... @overload @@ -1245,6 +1320,7 @@ class SyncRunsClient: webhook: Optional[str] = None, on_disconnect: Optional[DisconnectMode] = None, on_completion: Optional[OnCompletionBehavior] = None, + after_seconds: Optional[int] = None, ) -> Union[list[dict], dict[str, Any]]: ... def wait( @@ -1255,6 +1331,7 @@ class SyncRunsClient: input: Optional[dict] = None, metadata: Optional[dict] = None, config: Optional[Config] = None, + checkpoint: Optional[Checkpoint] = None, checkpoint_id: Optional[str] = None, interrupt_before: Optional[list[str]] = None, interrupt_after: Optional[list[str]] = None, @@ -1262,6 +1339,7 @@ class SyncRunsClient: on_disconnect: Optional[DisconnectMode] = None, on_completion: Optional[OnCompletionBehavior] = None, multitask_strategy: Optional[MultitaskStrategy] = None, + after_seconds: Optional[int] = None, ) -> Union[list[dict], dict[str, Any]]: """Create a run, wait until it finishes and return the final state. @@ -1273,18 +1351,18 @@ class SyncRunsClient: input: The input to the graph. metadata: Metadata to assign to the run. config: The configuration for the assistant. - checkpoint_id: The checkpoint to start streaming from. + checkpoint: The checkpoint to resume from. 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_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'. + after_seconds: The number of seconds to wait before starting the run. + Use to schedule future runs. Returns: Union[list[dict], dict[str, Any]]: The output of the run. @@ -1297,7 +1375,6 @@ class SyncRunsClient: input={"messages": [{"role": "user", "content": "how are you?"}]}, metadata={"name":"my_run"}, config={"configurable": {"model_name": "anthropic"}}, - checkpoint_id="my_checkpoint", 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", @@ -1342,10 +1419,12 @@ class SyncRunsClient: "interrupt_before": interrupt_before, "interrupt_after": interrupt_after, "webhook": webhook, + "checkpoint": checkpoint, "checkpoint_id": checkpoint_id, "multitask_strategy": multitask_strategy, "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" @@ -1456,7 +1535,7 @@ class SyncRunsClient: Example Usage: - await client.runs.join( + client.runs.join_stream( thread_id="thread_id_to_join", run_id="run_id_to_join" ) From 985f1b1e261acc620ae86814b778ee2aa3176d1a Mon Sep 17 00:00:00 2001 From: vbarda Date: Mon, 23 Sep 2024 19:41:08 -0400 Subject: [PATCH 8/8] code review --- libs/sdk-py/langgraph_sdk/client/__init__.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/libs/sdk-py/langgraph_sdk/client/__init__.py b/libs/sdk-py/langgraph_sdk/client/__init__.py index 741fc51d1..008ca2714 100644 --- a/libs/sdk-py/langgraph_sdk/client/__init__.py +++ b/libs/sdk-py/langgraph_sdk/client/__init__.py @@ -1,3 +1,3 @@ -from langgraph_sdk.client.async_client import get_client +from langgraph_sdk.client.async_client import LangGraphClient, get_client -__all__ = ["get_client"] +__all__ = ["get_client", "LangGraphClient"]