codekingpro/portable-devtools
114k
1"""Async client for managing threads in LangGraph."""2 3from __future__ import annotations4 5from collections.abc import AsyncIterator, Mapping, Sequence6from typing import Any, Literal, overload7 8from langgraph_sdk._async.http import HttpClient9from langgraph_sdk.schema import (10 Checkpoint,11 Json,12 OnConflictBehavior,13 PruneStrategy,14 QueryParamTypes,15 SortOrder,16 StreamPart,17 Thread,18 ThreadSelectField,19 ThreadSortBy,20 ThreadState,21 ThreadStatus,22 ThreadStreamMode,23 ThreadUpdateStateResponse,24)25 26 27class ThreadsClient:28 """Client for managing threads in LangGraph.29 30 A thread maintains the state of a graph across multiple interactions/invocations (aka runs).31 It accumulates and persists the graph's state, allowing for continuity between separate32 invocations of the graph.33 34 ???+ example "Example"35 36 ```python37 client = get_client(url="http://localhost:2024"))38 new_thread = await client.threads.create(metadata={"user_id": "123"})39 ```40 """41 42 def __init__(self, http: HttpClient) -> None:43 self.http = http44 45 async def get(46 self,47 thread_id: str,48 *,49 include: Sequence[str] | None = None,50 headers: Mapping[str, str] | None = None,51 params: QueryParamTypes | None = None,52 ) -> Thread:53 """Get a thread by ID.54 55 Args:56 thread_id: The ID of the thread to get.57 include: Additional fields to include in the response.58 Supported values: `"ttl"`.59 headers: Optional custom headers to include with the request.60 params: Optional query parameters to include with the request.61 62 Returns:63 Thread object.64 65 ???+ example "Example Usage"66 67 ```python68 client = get_client(url="http://localhost:2024")69 thread = await client.threads.get(70 thread_id="my_thread_id"71 )72 print(thread)73 ```74 75 ```shell76 -----------------------------------------------------77 78 {79 'thread_id': 'my_thread_id',80 'created_at': '2024-07-18T18:35:15.540834+00:00',81 'updated_at': '2024-07-18T18:35:15.540834+00:00',82 'metadata': {'graph_id': 'agent'}83 }84 ```85 86 """87 query_params: dict[str, Any] = {}88 if include:89 query_params["include"] = ",".join(include)90 if params:91 query_params.update(params)92 return await self.http.get(93 f"/threads/{thread_id}",94 headers=headers,95 params=query_params or None,96 )97 98 async def create(99 self,100 *,101 metadata: Json = None,102 thread_id: str | None = None,103 if_exists: OnConflictBehavior | None = None,104 supersteps: Sequence[dict[str, Sequence[dict[str, Any]]]] | None = None,105 graph_id: str | None = None,106 ttl: int | Mapping[str, Any] | None = None,107 headers: Mapping[str, str] | None = None,108 params: QueryParamTypes | None = None,109 ) -> Thread:110 """Create a new thread.111 112 Args:113 metadata: Metadata to add to thread.114 thread_id: ID of thread.115 If `None`, ID will be a randomly generated UUID.116 if_exists: How to handle duplicate creation. Defaults to 'raise' under the hood.117 Must be either 'raise' (raise error if duplicate), or 'do_nothing' (return existing thread).118 supersteps: Apply a list of supersteps when creating a thread, each containing a sequence of updates.119 Each update has `values` or `command` and `as_node`. Used for copying a thread between deployments.120 graph_id: Optional graph ID to associate with the thread.121 ttl: Optional time-to-live in minutes for the thread. You can pass an122 integer (minutes) or a mapping with keys `ttl` and optional123 `strategy` (defaults to "delete").124 headers: Optional custom headers to include with the request.125 params: Optional query parameters to include with the request.126 127 Returns:128 The created thread.129 130 ???+ example "Example Usage"131 132 ```python133 client = get_client(url="http://localhost:2024")134 thread = await client.threads.create(135 metadata={"number":1},136 thread_id="my-thread-id",137 if_exists="raise"138 )139 ```140 """141 payload: dict[str, Any] = {}142 if thread_id:143 payload["thread_id"] = thread_id144 if metadata or graph_id:145 payload["metadata"] = {146 **(metadata or {}),147 **({"graph_id": graph_id} if graph_id else {}),148 }149 if if_exists:150 payload["if_exists"] = if_exists151 if supersteps:152 payload["supersteps"] = [153 {154 "updates": [155 {156 "values": u["values"],157 "command": u.get("command"),158 "as_node": u["as_node"],159 }160 for u in s["updates"]161 ]162 }163 for s in supersteps164 ]165 if ttl is not None:166 if isinstance(ttl, (int, float)):167 payload["ttl"] = {"ttl": ttl, "strategy": "delete"}168 else:169 payload["ttl"] = ttl170 171 return await self.http.post(172 "/threads", json=payload, headers=headers, params=params173 )174 175 @overload176 async def update(177 self,178 thread_id: str,179 *,180 metadata: Mapping[str, Any],181 ttl: int | Mapping[str, Any] | None = None,182 return_minimal: Literal[False] = False,183 headers: Mapping[str, str] | None = None,184 params: QueryParamTypes | None = None,185 ) -> Thread: ...186 187 @overload188 async def update(189 self,190 thread_id: str,191 *,192 metadata: Mapping[str, Any],193 ttl: int | Mapping[str, Any] | None = None,194 return_minimal: Literal[True],195 headers: Mapping[str, str] | None = None,196 params: QueryParamTypes | None = None,197 ) -> None: ...198 199 @overload200 async def update(201 self,202 thread_id: str,203 *,204 metadata: Mapping[str, Any],205 ttl: int | Mapping[str, Any] | None = None,206 return_minimal: bool,207 headers: Mapping[str, str] | None = None,208 params: QueryParamTypes | None = None,209 ) -> Thread | None: ...210 211 async def update(212 self,213 thread_id: str,214 *,215 metadata: Mapping[str, Any],216 ttl: int | Mapping[str, Any] | None = None,217 return_minimal: bool = False,218 headers: Mapping[str, str] | None = None,219 params: QueryParamTypes | None = None,220 ) -> Thread | None:221 """Update a thread.222 223 Args:224 thread_id: ID of thread to update.225 metadata: Metadata to merge with existing thread metadata.226 ttl: Optional time-to-live in minutes for the thread. You can pass an227 integer (minutes) or a mapping with keys `ttl` and optional228 `strategy` (defaults to "delete").229 return_minimal: If `True`, request a 204 response with no body.230 headers: Optional custom headers to include with the request.231 params: Optional query parameters to include with the request.232 233 Returns:234 The updated thread, or `None` when `return_minimal=True`.235 236 ???+ example "Example Usage"237 238 ```python239 client = get_client(url="http://localhost:2024")240 thread = await client.threads.update(241 thread_id="my-thread-id",242 metadata={"number":1},243 ttl=43_200,244 )245 ```246 """247 payload: dict[str, Any] = {"metadata": metadata}248 if ttl is not None:249 if isinstance(ttl, (int, float)):250 payload["ttl"] = {"ttl": ttl, "strategy": "delete"}251 else:252 payload["ttl"] = ttl253 request_headers = dict(headers or {})254 if return_minimal:255 request_headers["Prefer"] = "return=minimal"256 return await self.http.patch(257 f"/threads/{thread_id}",258 json=payload,259 headers=request_headers or None,260 params=params,261 )262 263 async def delete(264 self,265 thread_id: str,266 *,267 headers: Mapping[str, str] | None = None,268 params: QueryParamTypes | None = None,269 ) -> None:270 """Delete a thread.271 272 Args:273 thread_id: The ID of the thread to delete.274 headers: Optional custom headers to include with the request.275 params: Optional query parameters to include with the request.276 277 Returns:278 `None`279 280 ???+ example "Example Usage"281 282 ```python283 client = get_client(url="http://localhost2024)284 await client.threads.delete(285 thread_id="my_thread_id"286 )287 ```288 289 """290 await self.http.delete(f"/threads/{thread_id}", headers=headers, params=params)291 292 async def search(293 self,294 *,295 metadata: Json = None,296 values: Json = None,297 ids: Sequence[str] | None = None,298 status: ThreadStatus | None = None,299 limit: int = 10,300 offset: int = 0,301 sort_by: ThreadSortBy | None = None,302 sort_order: SortOrder | None = None,303 select: list[ThreadSelectField] | None = None,304 extract: dict[str, str] | None = None,305 headers: Mapping[str, str] | None = None,306 params: QueryParamTypes | None = None,307 ) -> list[Thread]:308 """Search for threads.309 310 Args:311 metadata: Thread metadata to filter on.312 values: State values to filter on.313 ids: List of thread IDs to filter by.314 status: Thread status to filter on.315 Must be one of 'idle', 'busy', 'interrupted' or 'error'.316 limit: Limit on number of threads to return.317 offset: Offset in threads table to start search from.318 sort_by: Sort by field.319 sort_order: Sort order.320 select: List of fields to include in the response.321 extract: Dictionary mapping aliases to JSONB paths to extract322 from thread data. Paths use dot notation for nested keys and323 bracket notation for array indices (e.g.,324 `{"last_msg": "values.messages[-1]"}`). Extracted values are325 returned in an `extracted` field on each thread. Maximum 10326 paths per request.327 headers: Optional custom headers to include with the request.328 params: Optional query parameters to include with the request.329 330 Returns:331 List of the threads matching the search parameters.332 333 ???+ example "Example Usage"334 335 ```python336 client = get_client(url="http://localhost:2024")337 threads = await client.threads.search(338 metadata={"number":1},339 status="interrupted",340 limit=15,341 offset=5342 )343 ```344 345 """346 payload: dict[str, Any] = {347 "limit": limit,348 "offset": offset,349 }350 if metadata:351 payload["metadata"] = metadata352 if values:353 payload["values"] = values354 if ids:355 payload["ids"] = ids356 if status:357 payload["status"] = status358 if sort_by:359 payload["sort_by"] = sort_by360 if sort_order:361 payload["sort_order"] = sort_order362 if select:363 payload["select"] = select364 if extract:365 payload["extract"] = extract366 return await self.http.post(367 "/threads/search",368 json=payload,369 headers=headers,370 params=params,371 )372 373 async def count(374 self,375 *,376 metadata: Json = None,377 values: Json = None,378 status: ThreadStatus | None = None,379 headers: Mapping[str, str] | None = None,380 params: QueryParamTypes | None = None,381 ) -> int:382 """Count threads matching filters.383 384 Args:385 metadata: Thread metadata to filter on.386 values: State values to filter on.387 status: Thread status to filter on.388 headers: Optional custom headers to include with the request.389 params: Optional query parameters to include with the request.390 391 Returns:392 int: Number of threads matching the criteria.393 """394 payload: dict[str, Any] = {}395 if metadata:396 payload["metadata"] = metadata397 if values:398 payload["values"] = values399 if status:400 payload["status"] = status401 return await self.http.post(402 "/threads/count", json=payload, headers=headers, params=params403 )404 405 async def copy(406 self,407 thread_id: str,408 *,409 headers: Mapping[str, str] | None = None,410 params: QueryParamTypes | None = None,411 ) -> None:412 """Copy a thread.413 414 Args:415 thread_id: The ID of the thread to copy.416 headers: Optional custom headers to include with the request.417 params: Optional query parameters to include with the request.418 419 Returns:420 `None`421 422 ???+ example "Example Usage"423 424 ```python425 client = get_client(url="http://localhost:2024)426 await client.threads.copy(427 thread_id="my_thread_id"428 )429 ```430 431 """432 return await self.http.post(433 f"/threads/{thread_id}/copy", json=None, headers=headers, params=params434 )435 436 async def prune(437 self,438 thread_ids: Sequence[str],439 *,440 strategy: PruneStrategy = "delete",441 headers: Mapping[str, str] | None = None,442 params: QueryParamTypes | None = None,443 ) -> dict[str, Any]:444 """Prune threads by ID.445 446 Args:447 thread_ids: List of thread IDs to prune.448 strategy: The prune strategy. `"delete"` removes threads entirely.449 `"keep_latest"` prunes old checkpoints but keeps threads and their450 latest state. Defaults to `"delete"`.451 headers: Optional custom headers to include with the request.452 params: Optional query parameters to include with the request.453 454 Returns:455 A dict containing `pruned_count` (number of threads pruned).456 457 ???+ example "Example Usage"458 459 ```python460 client = get_client(url="http://localhost:2024")461 result = await client.threads.prune(462 thread_ids=["thread_1", "thread_2"],463 )464 print(result) # {'pruned_count': 2}465 ```466 467 """468 payload: dict[str, Any] = {469 "thread_ids": thread_ids,470 }471 if strategy != "delete":472 payload["strategy"] = strategy473 return await self.http.post(474 "/threads/prune", json=payload, headers=headers, params=params475 )476 477 async def get_state(478 self,479 thread_id: str,480 checkpoint: Checkpoint | None = None,481 checkpoint_id: str | None = None, # deprecated482 *,483 subgraphs: bool = False,484 headers: Mapping[str, str] | None = None,485 params: QueryParamTypes | None = None,486 ) -> ThreadState:487 """Get the state of a thread.488 489 Args:490 thread_id: The ID of the thread to get the state of.491 checkpoint: The checkpoint to get the state of.492 checkpoint_id: (deprecated) The checkpoint ID to get the state of.493 subgraphs: Include subgraphs states.494 headers: Optional custom headers to include with the request.495 params: Optional query parameters to include with the request.496 497 Returns:498 The thread of the state.499 500 ???+ example "Example Usage"501 502 ```python503 client = get_client(url="http://localhost:2024)504 thread_state = await client.threads.get_state(505 thread_id="my_thread_id",506 checkpoint_id="my_checkpoint_id"507 )508 print(thread_state)509 ```510 511 ```shell512 ----------------------------------------------------------------------------------------------------------------------------------------------------------------------513 514 {515 'values': {516 'messages': [517 {518 'content': 'how are you?',519 'additional_kwargs': {},520 'response_metadata': {},521 'type': 'human',522 'name': None,523 'id': 'fe0a5778-cfe9-42ee-b807-0adaa1873c10',524 'example': False525 },526 {527 'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.",528 'additional_kwargs': {},529 'response_metadata': {},530 'type': 'ai',531 'name': None,532 'id': 'run-159b782c-b679-4830-83c6-cef87798fe8b',533 'example': False,534 'tool_calls': [],535 'invalid_tool_calls': [],536 'usage_metadata': None537 }538 ]539 },540 'next': [],541 'checkpoint':542 {543 'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2',544 'checkpoint_ns': '',545 'checkpoint_id': '1ef4a9b8-e6fb-67b1-8001-abd5184439d1'546 }547 'metadata':548 {549 'step': 1,550 'run_id': '1ef4a9b8-d7da-679a-a45a-872054341df2',551 'source': 'loop',552 'writes':553 {554 'agent':555 {556 'messages': [557 {558 'id': 'run-159b782c-b679-4830-83c6-cef87798fe8b',559 'name': None,560 'type': 'ai',561 'content': "I'm doing well, thanks for asking! I'm an AI assistant created by Anthropic to be helpful, honest, and harmless.",562 'example': False,563 'tool_calls': [],564 'usage_metadata': None,565 'additional_kwargs': {},566 'response_metadata': {},567 'invalid_tool_calls': []568 }569 ]570 }571 },572 'user_id': None,573 'graph_id': 'agent',574 'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2',575 'created_by': 'system',576 'assistant_id': 'fe096781-5601-53d2-b2f6-0d3403f7e9ca'},577 'created_at': '2024-07-25T15:35:44.184703+00:00',578 'parent_config':579 {580 'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2',581 'checkpoint_ns': '',582 'checkpoint_id': '1ef4a9b8-d80d-6fa7-8000-9300467fad0f'583 }584 }585 ```586 """587 if checkpoint:588 return await self.http.post(589 f"/threads/{thread_id}/state/checkpoint",590 json={"checkpoint": checkpoint, "subgraphs": subgraphs},591 headers=headers,592 params=params,593 )594 elif checkpoint_id:595 get_params = {"subgraphs": subgraphs}596 if params:597 get_params = {**get_params, **dict(params)}598 return await self.http.get(599 f"/threads/{thread_id}/state/{checkpoint_id}",600 params=get_params,601 headers=headers,602 )603 else:604 get_params = {"subgraphs": subgraphs}605 if params:606 get_params = {**get_params, **dict(params)}607 return await self.http.get(608 f"/threads/{thread_id}/state",609 params=get_params,610 headers=headers,611 )612 613 async def update_state(614 self,615 thread_id: str,616 values: dict[str, Any] | Sequence[dict] | None,617 *,618 as_node: str | None = None,619 checkpoint: Checkpoint | None = None,620 checkpoint_id: str | None = None, # deprecated621 headers: Mapping[str, str] | None = None,622 params: QueryParamTypes | None = None,623 ) -> ThreadUpdateStateResponse:624 """Update the state of a thread.625 626 Args:627 thread_id: The ID of the thread to update.628 values: The values to update the state with.629 as_node: Update the state as if this node had just executed.630 checkpoint: The checkpoint to update the state of.631 checkpoint_id: (deprecated) The checkpoint ID to update the state of.632 headers: Optional custom headers to include with the request.633 params: Optional query parameters to include with the request.634 635 Returns:636 Response after updating a thread's state.637 638 ???+ example "Example Usage"639 640 ```python641 client = get_client(url="http://localhost:2024)642 response = await client.threads.update_state(643 thread_id="my_thread_id",644 values={"messages":[{"role": "user", "content": "hello!"}]},645 as_node="my_node",646 )647 print(response)648 ```649 ```shell650 651 ----------------------------------------------------------------------------------------------------------------------------------------------------------------------652 653 {654 'checkpoint': {655 'thread_id': 'e2496803-ecd5-4e0c-a779-3226296181c2',656 'checkpoint_ns': '',657 'checkpoint_id': '1ef4a9b8-e6fb-67b1-8001-abd5184439d1',658 'checkpoint_map': {}659 }660 }661 ```662 """663 payload: dict[str, Any] = {664 "values": values,665 }666 if checkpoint_id:667 payload["checkpoint_id"] = checkpoint_id668 if checkpoint:669 payload["checkpoint"] = checkpoint670 if as_node:671 payload["as_node"] = as_node672 return await self.http.post(673 f"/threads/{thread_id}/state", json=payload, headers=headers, params=params674 )675 676 async def get_history(677 self,678 thread_id: str,679 *,680 limit: int = 10,681 before: str | Checkpoint | None = None,682 metadata: Mapping[str, Any] | None = None,683 checkpoint: Checkpoint | None = None,684 headers: Mapping[str, str] | None = None,685 params: QueryParamTypes | None = None,686 ) -> list[ThreadState]:687 """Get the state history of a thread.688 689 Args:690 thread_id: The ID of the thread to get the state history for.691 checkpoint: Return states for this subgraph. If empty defaults to root.692 limit: The maximum number of states to return.693 before: Return states before this checkpoint.694 metadata: Filter states by metadata key-value pairs.695 headers: Optional custom headers to include with the request.696 params: Optional query parameters to include with the request.697 698 Returns:699 The state history of the thread.700 701 ???+ example "Example Usage"702 703 ```python704 client = get_client(url="http://localhost:2024)705 thread_state = await client.threads.get_history(706 thread_id="my_thread_id",707 limit=5,708 )709 ```710 711 """712 payload: dict[str, Any] = {713 "limit": limit,714 }715 if before:716 payload["before"] = before717 if metadata:718 payload["metadata"] = metadata719 if checkpoint:720 payload["checkpoint"] = checkpoint721 return await self.http.post(722 f"/threads/{thread_id}/history",723 json=payload,724 headers=headers,725 params=params,726 )727 728 async def join_stream(729 self,730 thread_id: str,731 *,732 last_event_id: str | None = None,733 stream_mode: ThreadStreamMode | Sequence[ThreadStreamMode] = "run_modes",734 headers: Mapping[str, str] | None = None,735 params: QueryParamTypes | None = None,736 ) -> AsyncIterator[StreamPart]:737 """Get a stream of events for a thread.738 739 Args:740 thread_id: The ID of the thread to get the stream for.741 last_event_id: The ID of the last event to get.742 headers: Optional custom headers to include with the request.743 params: Optional query parameters to include with the request.744 745 Returns:746 An iterator of stream parts.747 748 ???+ example "Example Usage"749 750 ```python751 752 for chunk in client.threads.join_stream(753 thread_id="my_thread_id",754 last_event_id="my_event_id",755 ):756 print(chunk)757 ```758 759 """760 query_params = {761 "stream_mode": stream_mode,762 }763 if params:764 query_params.update(params)765 return self.http.stream(766 f"/threads/{thread_id}/stream",767 "GET",768 headers={769 **({"Last-Event-ID": last_event_id} if last_event_id else {}),770 **(headers or {}),771 },772 params=query_params,773 )774