codekingpro/portable-devtools
114k
1"""Synchronous client for managing assistants in LangGraph."""2 3from __future__ import annotations4 5from collections.abc import Mapping6from typing import Any, Literal, cast, overload7 8import httpx9 10from langgraph_sdk._sync.http import SyncHttpClient11from langgraph_sdk.schema import (12 Assistant,13 AssistantSelectField,14 AssistantSortBy,15 AssistantsSearchResponse,16 AssistantVersion,17 Config,18 Context,19 GraphSchema,20 Json,21 OnConflictBehavior,22 QueryParamTypes,23 SortOrder,24 Subgraphs,25)26 27 28class SyncAssistantsClient:29 """Client for managing assistants in LangGraph synchronously.30 31 This class provides methods to interact with assistants, which are versioned configurations of your graph.32 33 ???+ example "Example"34 35 ```python36 client = get_sync_client(url="http://localhost:2024")37 assistant = client.assistants.get("assistant_id_123")38 ```39 """40 41 def __init__(self, http: SyncHttpClient) -> None:42 self.http = http43 44 def get(45 self,46 assistant_id: str,47 *,48 headers: Mapping[str, str] | None = None,49 params: QueryParamTypes | None = None,50 ) -> Assistant:51 """Get an assistant by ID.52 53 Args:54 assistant_id: The ID of the assistant to get OR the name of the graph (to use the default assistant).55 headers: Optional custom headers to include with the request.56 params: Optional query parameters to include with the request.57 58 Returns:59 `Assistant` Object.60 61 ???+ example "Example Usage"62 63 ```python64 assistant = client.assistants.get(65 assistant_id="my_assistant_id"66 )67 print(assistant)68 ```69 70 ```shell71 ----------------------------------------------------72 73 {74 'assistant_id': 'my_assistant_id',75 'graph_id': 'agent',76 'created_at': '2024-06-25T17:10:33.109781+00:00',77 'updated_at': '2024-06-25T17:10:33.109781+00:00',78 'config': {},79 'context': {},80 'metadata': {'created_by': 'system'}81 }82 ```83 84 """85 return self.http.get(86 f"/assistants/{assistant_id}", headers=headers, params=params87 )88 89 def get_graph(90 self,91 assistant_id: str,92 *,93 xray: int | bool = False,94 headers: Mapping[str, str] | None = None,95 params: QueryParamTypes | None = None,96 ) -> dict[str, list[dict[str, Any]]]:97 """Get the graph of an assistant by ID.98 99 Args:100 assistant_id: The ID of the assistant to get the graph of.101 xray: Include graph representation of subgraphs. If an integer value is provided, only subgraphs with a depth less than or equal to the value will be included.102 headers: Optional custom headers to include with the request.103 params: Optional query parameters to include with the request.104 105 Returns:106 The graph information for the assistant in JSON format.107 108 ???+ example "Example Usage"109 110 ```python111 client = get_sync_client(url="http://localhost:2024")112 graph_info = client.assistants.get_graph(113 assistant_id="my_assistant_id"114 )115 print(graph_info)116 117 --------------------------------------------------------------------------------------------------------------------------118 119 {120 'nodes':121 [122 {'id': '__start__', 'type': 'schema', 'data': '__start__'},123 {'id': '__end__', 'type': 'schema', 'data': '__end__'},124 {'id': 'agent','type': 'runnable','data': {'id': ['langgraph', 'utils', 'RunnableCallable'],'name': 'agent'}},125 ],126 'edges':127 [128 {'source': '__start__', 'target': 'agent'},129 {'source': 'agent','target': '__end__'}130 ]131 }132 ```133 134 """135 query_params = {"xray": xray}136 if params:137 query_params.update(params)138 return self.http.get(139 f"/assistants/{assistant_id}/graph", params=query_params, headers=headers140 )141 142 def get_schemas(143 self,144 assistant_id: str,145 *,146 headers: Mapping[str, str] | None = None,147 params: QueryParamTypes | None = None,148 ) -> GraphSchema:149 """Get the schemas of an assistant by ID.150 151 Args:152 assistant_id: The ID of the assistant to get the schema of.153 headers: Optional custom headers to include with the request.154 params: Optional query parameters to include with the request.155 156 Returns:157 GraphSchema: The graph schema for the assistant.158 159 ???+ example "Example Usage"160 161 ```python162 client = get_sync_client(url="http://localhost:2024")163 schema = client.assistants.get_schemas(164 assistant_id="my_assistant_id"165 )166 print(schema)167 ```168 ```shell169 ----------------------------------------------------------------------------------------------------------------------------170 171 {172 'graph_id': 'agent',173 'state_schema':174 {175 'title': 'LangGraphInput',176 '$ref': '#/definitions/AgentState',177 'definitions':178 {179 'BaseMessage':180 {181 'title': 'BaseMessage',182 'description': 'Base abstract Message class. Messages are the inputs and outputs of ChatModels.',183 'type': 'object',184 'properties':185 {186 'content':187 {188 'title': 'Content',189 'anyOf': [190 {'type': 'string'},191 {'type': 'array','items': {'anyOf': [{'type': 'string'}, {'type': 'object'}]}}192 ]193 },194 'additional_kwargs':195 {196 'title': 'Additional Kwargs',197 'type': 'object'198 },199 'response_metadata':200 {201 'title': 'Response Metadata',202 'type': 'object'203 },204 'type':205 {206 'title': 'Type',207 'type': 'string'208 },209 'name':210 {211 'title': 'Name',212 'type': 'string'213 },214 'id':215 {216 'title': 'Id',217 'type': 'string'218 }219 },220 'required': ['content', 'type']221 },222 'AgentState':223 {224 'title': 'AgentState',225 'type': 'object',226 'properties':227 {228 'messages':229 {230 'title': 'Messages',231 'type': 'array',232 'items': {'$ref': '#/definitions/BaseMessage'}233 }234 },235 'required': ['messages']236 }237 }238 },239 'config_schema':240 {241 'title': 'Configurable',242 'type': 'object',243 'properties':244 {245 'model_name':246 {247 'title': 'Model Name',248 'enum': ['anthropic', 'openai'],249 'type': 'string'250 }251 }252 },253 'context_schema':254 {255 'title': 'Context',256 'type': 'object',257 'properties':258 {259 'model_name':260 {261 'title': 'Model Name',262 'enum': ['anthropic', 'openai'],263 'type': 'string'264 }265 }266 }267 }268 ```269 270 """271 return self.http.get(272 f"/assistants/{assistant_id}/schemas", headers=headers, params=params273 )274 275 def get_subgraphs(276 self,277 assistant_id: str,278 namespace: str | None = None,279 recurse: bool = False,280 *,281 headers: Mapping[str, str] | None = None,282 params: QueryParamTypes | None = None,283 ) -> Subgraphs:284 """Get the schemas of an assistant by ID.285 286 Args:287 assistant_id: The ID of the assistant to get the schema of.288 headers: Optional custom headers to include with the request.289 params: Optional query parameters to include with the request.290 291 Returns:292 Subgraphs: The graph schema for the assistant.293 294 """295 get_params = {"recurse": recurse}296 if params:297 get_params = {**get_params, **dict(params)}298 if namespace is not None:299 return self.http.get(300 f"/assistants/{assistant_id}/subgraphs/{namespace}",301 params=get_params,302 headers=headers,303 )304 else:305 return self.http.get(306 f"/assistants/{assistant_id}/subgraphs",307 params=get_params,308 headers=headers,309 )310 311 def create(312 self,313 graph_id: str | None,314 config: Config | None = None,315 *,316 context: Context | None = None,317 metadata: Json = None,318 assistant_id: str | None = None,319 if_exists: OnConflictBehavior | None = None,320 name: str | None = None,321 headers: Mapping[str, str] | None = None,322 description: str | None = None,323 params: QueryParamTypes | None = None,324 ) -> Assistant:325 """Create a new assistant.326 327 Useful when graph is configurable and you want to create different assistants based on different configurations.328 329 Args:330 graph_id: The ID of the graph the assistant should use. The graph ID is normally set in your langgraph.json configuration.331 config: Configuration to use for the graph.332 context: Static context to add to the assistant.333 !!! version-added "Added in version 0.6.0"334 metadata: Metadata to add to assistant.335 assistant_id: Assistant ID to use, will default to a random UUID if not provided.336 if_exists: How to handle duplicate creation. Defaults to 'raise' under the hood.337 Must be either 'raise' (raise error if duplicate), or 'do_nothing' (return existing assistant).338 name: The name of the assistant. Defaults to 'Untitled' under the hood.339 headers: Optional custom headers to include with the request.340 description: Optional description of the assistant.341 The description field is available for langgraph-api server version>=0.0.45342 params: Optional query parameters to include with the request.343 344 Returns:345 The created assistant.346 347 ???+ example "Example Usage"348 349 ```python350 client = get_sync_client(url="http://localhost:2024")351 assistant = client.assistants.create(352 graph_id="agent",353 context={"model_name": "openai"},354 metadata={"number":1},355 assistant_id="my-assistant-id",356 if_exists="do_nothing",357 name="my_name"358 )359 ```360 """361 payload: dict[str, Any] = {362 "graph_id": graph_id,363 }364 if config:365 payload["config"] = config366 if context:367 payload["context"] = context368 if metadata:369 payload["metadata"] = metadata370 if assistant_id:371 payload["assistant_id"] = assistant_id372 if if_exists:373 payload["if_exists"] = if_exists374 if name:375 payload["name"] = name376 if description:377 payload["description"] = description378 return self.http.post(379 "/assistants", json=payload, headers=headers, params=params380 )381 382 def update(383 self,384 assistant_id: str,385 *,386 graph_id: str | None = None,387 config: Config | None = None,388 context: Context | None = None,389 metadata: Json = None,390 name: str | None = None,391 headers: Mapping[str, str] | None = None,392 description: str | None = None,393 params: QueryParamTypes | None = None,394 ) -> Assistant:395 """Update an assistant.396 397 Use this to point to a different graph, update the configuration, or change the metadata of an assistant.398 399 Args:400 assistant_id: Assistant to update.401 graph_id: The ID of the graph the assistant should use.402 The graph ID is normally set in your langgraph.json configuration. If `None`, assistant will keep pointing to same graph.403 config: Configuration to use for the graph.404 context: Static context to add to the assistant.405 !!! version-added "Added in version 0.6.0"406 metadata: Metadata to merge with existing assistant metadata.407 name: The new name for the assistant.408 headers: Optional custom headers to include with the request.409 description: Optional description of the assistant.410 The description field is available for langgraph-api server version>=0.0.45411 412 Returns:413 The updated assistant.414 415 ???+ example "Example Usage"416 417 ```python418 client = get_sync_client(url="http://localhost:2024")419 assistant = client.assistants.update(420 assistant_id='e280dad7-8618-443f-87f1-8e41841c180f',421 graph_id="other-graph",422 context={"model_name": "anthropic"},423 metadata={"number":2}424 )425 ```426 """427 payload: dict[str, Any] = {}428 if graph_id:429 payload["graph_id"] = graph_id430 if config is not None:431 payload["config"] = config432 if context is not None:433 payload["context"] = context434 if metadata:435 payload["metadata"] = metadata436 if name:437 payload["name"] = name438 if description:439 payload["description"] = description440 return self.http.patch(441 f"/assistants/{assistant_id}",442 json=payload,443 headers=headers,444 params=params,445 )446 447 def delete(448 self,449 assistant_id: str,450 *,451 delete_threads: bool = False,452 headers: Mapping[str, str] | None = None,453 params: QueryParamTypes | None = None,454 ) -> None:455 """Delete an assistant.456 457 Args:458 assistant_id: The assistant ID to delete.459 delete_threads: If true, delete all threads with `metadata.assistant_id`460 matching this assistant, along with runs and checkpoints belonging to461 those threads.462 headers: Optional custom headers to include with the request.463 params: Optional query parameters to include with the request.464 465 Returns:466 `None`467 468 ???+ example "Example Usage"469 470 ```python471 client = get_sync_client(url="http://localhost:2024")472 client.assistants.delete(473 assistant_id="my_assistant_id"474 )475 ```476 477 """478 query_params: dict[str, Any] = {}479 if delete_threads:480 query_params["delete_threads"] = True481 if params:482 query_params.update(params)483 self.http.delete(484 f"/assistants/{assistant_id}",485 headers=headers,486 params=query_params or None,487 )488 489 @overload490 def search(491 self,492 *,493 metadata: Json = None,494 graph_id: str | None = None,495 name: str | None = None,496 limit: int = 10,497 offset: int = 0,498 sort_by: AssistantSortBy | None = None,499 sort_order: SortOrder | None = None,500 select: list[AssistantSelectField] | None = None,501 response_format: Literal["object"],502 headers: Mapping[str, str] | None = None,503 params: QueryParamTypes | None = None,504 ) -> AssistantsSearchResponse: ...505 506 @overload507 def search(508 self,509 *,510 metadata: Json = None,511 graph_id: str | None = None,512 name: str | None = None,513 limit: int = 10,514 offset: int = 0,515 sort_by: AssistantSortBy | None = None,516 sort_order: SortOrder | None = None,517 select: list[AssistantSelectField] | None = None,518 response_format: Literal["array"] = "array",519 headers: Mapping[str, str] | None = None,520 params: QueryParamTypes | None = None,521 ) -> list[Assistant]: ...522 523 def search(524 self,525 *,526 metadata: Json = None,527 graph_id: str | None = None,528 name: str | None = None,529 limit: int = 10,530 offset: int = 0,531 sort_by: AssistantSortBy | None = None,532 sort_order: SortOrder | None = None,533 select: list[AssistantSelectField] | None = None,534 response_format: Literal["array", "object"] = "array",535 headers: Mapping[str, str] | None = None,536 params: QueryParamTypes | None = None,537 ) -> AssistantsSearchResponse | list[Assistant]:538 """Search for assistants.539 540 Args:541 metadata: Metadata to filter by. Exact match filter for each KV pair.542 graph_id: The ID of the graph to filter by.543 The graph ID is normally set in your langgraph.json configuration.544 name: The name of the assistant to filter by.545 The filtering logic will match assistants where 'name' is a substring (case insensitive) of the assistant name.546 limit: The maximum number of results to return.547 offset: The number of results to skip.548 sort_by: The field to sort by.549 sort_order: The order to sort by.550 select: Specific assistant fields to include in the response.551 response_format: Controls the response shape. Use `"array"` (default)552 to return a bare list of assistants, or `"object"` to return553 a mapping containing assistants plus pagination metadata.554 Defaults to "array", though this default will be changed to "object" in a future release.555 headers: Optional custom headers to include with the request.556 557 Returns:558 A list of assistants (when `response_format="array"`) or a mapping559 with the assistants and the next pagination cursor (when560 `response_format="object"`).561 562 ???+ example "Example Usage"563 564 ```python565 client = get_sync_client(url="http://localhost:2024")566 response = client.assistants.search(567 metadata = {"name":"my_name"},568 graph_id="my_graph_id",569 limit=5,570 offset=5,571 response_format="object",572 )573 assistants = response["assistants"]574 next_cursor = response["next"]575 ```576 """577 if response_format not in ("array", "object"):578 raise ValueError("response_format must be 'array' or 'object'")579 payload: dict[str, Any] = {580 "limit": limit,581 "offset": offset,582 }583 if metadata:584 payload["metadata"] = metadata585 if graph_id:586 payload["graph_id"] = graph_id587 if name:588 payload["name"] = name589 if sort_by:590 payload["sort_by"] = sort_by591 if sort_order:592 payload["sort_order"] = sort_order593 if select:594 payload["select"] = select595 next_cursor: str | None = None596 597 def capture_pagination(response: httpx.Response) -> None:598 nonlocal next_cursor599 next_cursor = response.headers.get("X-Pagination-Next")600 601 assistants = cast(602 list[Assistant],603 self.http.post(604 "/assistants/search",605 json=payload,606 headers=headers,607 params=params,608 on_response=capture_pagination if response_format == "object" else None,609 ),610 )611 if response_format == "object":612 return {"assistants": assistants, "next": next_cursor}613 return assistants614 615 def count(616 self,617 *,618 metadata: Json = None,619 graph_id: str | None = None,620 name: str | None = None,621 headers: Mapping[str, str] | None = None,622 params: QueryParamTypes | None = None,623 ) -> int:624 """Count assistants matching filters.625 626 Args:627 metadata: Metadata to filter by. Exact match for each key/value.628 graph_id: Optional graph id to filter by.629 name: Optional name to filter by.630 The filtering logic will match assistants where 'name' is a substring (case insensitive) of the assistant name.631 headers: Optional custom headers to include with the request.632 params: Optional query parameters to include with the request.633 634 Returns:635 int: Number of assistants matching the criteria.636 """637 payload: dict[str, Any] = {}638 if metadata:639 payload["metadata"] = metadata640 if graph_id:641 payload["graph_id"] = graph_id642 if name:643 payload["name"] = name644 return self.http.post(645 "/assistants/count", json=payload, headers=headers, params=params646 )647 648 def get_versions(649 self,650 assistant_id: str,651 metadata: Json = None,652 limit: int = 10,653 offset: int = 0,654 *,655 headers: Mapping[str, str] | None = None,656 params: QueryParamTypes | None = None,657 ) -> list[AssistantVersion]:658 """List all versions of an assistant.659 660 Args:661 assistant_id: The assistant ID to get versions for.662 metadata: Metadata to filter versions by. Exact match filter for each KV pair.663 limit: The maximum number of versions to return.664 offset: The number of versions to skip.665 headers: Optional custom headers to include with the request.666 667 Returns:668 A list of assistants.669 670 ???+ example "Example Usage"671 672 ```python673 client = get_sync_client(url="http://localhost:2024")674 assistant_versions = client.assistants.get_versions(675 assistant_id="my_assistant_id"676 )677 ```678 679 """680 681 payload: dict[str, Any] = {682 "limit": limit,683 "offset": offset,684 }685 if metadata:686 payload["metadata"] = metadata687 return self.http.post(688 f"/assistants/{assistant_id}/versions",689 json=payload,690 headers=headers,691 params=params,692 )693 694 def set_latest(695 self,696 assistant_id: str,697 version: int,698 *,699 headers: Mapping[str, str] | None = None,700 params: QueryParamTypes | None = None,701 ) -> Assistant:702 """Change the version of an assistant.703 704 Args:705 assistant_id: The assistant ID to delete.706 version: The version to change to.707 headers: Optional custom headers to include with the request.708 709 Returns:710 `Assistant` Object.711 712 ???+ example "Example Usage"713 714 ```python715 client = get_sync_client(url="http://localhost:2024")716 new_version_assistant = client.assistants.set_latest(717 assistant_id="my_assistant_id",718 version=3719 )720 ```721 722 """723 724 payload: dict[str, Any] = {"version": version}725 726 return self.http.post(727 f"/assistants/{assistant_id}/latest",728 json=payload,729 headers=headers,730 params=params,731 )732