Sweep: Update the docstrings and comments in sdks/python/julep/managers/memory.py to fix any issues and mismatch between the comments present and surrounding code

[creatorrr]

See the rest of the python files in sdks/python/julep/ directory for context. Make sure that every comment matches the logic in the associated code. Overtime, comments may have drifted and accidentally not kept up with the code changes. Be concise and add new comments ONLY when necessary.

Checklist

• Modify sdks/python/julep/managers/memory.py ✓ 963a591 Edit
• Running GitHub Actions for sdks/python/julep/managers/memory.py ✓ Edit

[sweep-ai]

🚀 Here's the PR! #271

See Sweep's progress at the progress dashboard!

💎 Sweep Pro: I'm using GPT-4. You have unlimited GPT-4 tickets. (tracking ID: 3b314d5f66)

---

Tip

I can email you next time I complete a pull request if you set up your email here!

---

Actions (click)

• ↻ Restart Sweep

---

Step 1: 🔎 Searching

I found the following snippets in your repository. I will now analyze these snippets and come up with a plan.

Some code snippets I think are relevant in decreasing order of relevance (click to expand). If some file is missing from here, you can mention the path in the ticket description.

julep/sdks/python/julep/managers/memory.py

Lines 1 to 190 in 26e57ac

	from uuid import UUID
	from typing import Optional
	from beartype import beartype
	from beartype.typing import Awaitable, List, Union
	from ..api.types import (
	Memory,
	GetAgentMemoriesResponse,
	)
	from .base import BaseManager
	from .utils import is_valid_uuid4
	class BaseMemoriesManager(BaseManager):
	"""
	A base manager class for handling agent memories.
	This manager provides an interface to interact with agent memories, allowing
	for listing and other operations to manage an agent's memories.
	Methods:
	_list(agent_id, query, types=None, user_id=None, limit=None, offset=None):
	Retrieves a list of memories for a given agent.
	Args:
	agent_id (str): A valid UUID v4 string identifying the agent.
	query (str): The query string to search memories.
	types (Optional[Union[str, List[str]]]): The type(s) of memories to retrieve.
	user_id (Optional[str]): The user identifier associated with the memories.
	limit (Optional[int]): The maximum number of memories to retrieve.
	offset (Optional[int]): The number of initial memories to skip in the result set.
	Returns:
	Union[GetAgentMemoriesResponse, Awaitable[GetAgentMemoriesResponse]]:
	A synchronous or asynchronous response object containing the list of agent memories.
	Raises:
	AssertionError: If `agent_id` is not a valid UUID v4.
	"""
	def _list(
	self,
	agent_id: str,
	query: str,
	types: Optional[Union[str, List[str]]] = None,
	user_id: Optional[str] = None,
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	) -> Union[GetAgentMemoriesResponse, Awaitable[GetAgentMemoriesResponse]]:
	"""
	List memories from a given agent based on a query and further filtering options.
	Args:
	agent_id (str): A valid UUID v4 representing the agent ID.
	query (str): Query string to filter memories.
	types (Optional[Union[str, List[str]]], optional): The types of memories to filter.
	user_id (Optional[str], optional): The user ID to filter memories.
	limit (Optional[int], optional): The maximum number of memories to return.
	offset (Optional[int], optional): The number of memories to skip before starting to collect the result set.
	Returns:
	Union[GetAgentMemoriesResponse, Awaitable[GetAgentMemoriesResponse]]: Returns a synchronous or asynchronous response with the agent memories.
	Raises:
	AssertionError: If `agent_id` is not a valid UUID v4.
	"""
	assert is_valid_uuid4(agent_id), "agent_id must be a valid UUID v4"
	return self.api_client.get_agent_memories(
	agent_id=agent_id,
	query=query,
	user_id=user_id,
	limit=limit,
	offset=offset,
	)
	class MemoriesManager(BaseMemoriesManager):
	"""
	A class for managing memory entities associated with agents.
	This class inherits from `BaseMemoriesManager` and provides
	an interface to list memory entities for a given agent.
	Attributes:
	Inherited from `BaseMemoriesManager`.
	Methods:
	list: Retrieves a list of memory entities based on query parameters.
	"""
	@beartype
	def list(
	self,
	*,
	agent_id: Union[str, UUID],
	query: str,
	types: Optional[Union[str, List[str]]] = None,
	user_id: Optional[str] = None,
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	) -> List[Memory]:
	"""
	List memories meeting specified criteria.
	This function fetches a list of Memory objects based on various filters and parameters such as agent_id, query, types, user_id, limit, and offset.
	Args:
	agent_id (Union[str, UUID]): The unique identifier for the agent.
	query (str): The search term used to filter memories.
	types (Optional[Union[str, List[str]]], optional): The types of memories to retrieve. Can be a single type as a string or a list of types. Default is None, which does not filter by type.
	user_id (Optional[str], optional): The unique identifier for the user. If provided, only memories associated with this user will be retrieved. Default is None.
	limit (Optional[int], optional): The maximum number of memories to return. Default is None, which means no limit.
	offset (Optional[int], optional): The number of memories to skip before starting to return the results. Default is None.
	Returns:
	List[Memory]: A list of Memory objects that match the given criteria.
	Note:
	The `@beartype` decorator is used to ensure that arguments conform to the expected types at runtime.
	"""
	return self._list(
	agent_id=agent_id,
	query=query,
	types=types,
	user_id=user_id,
	limit=limit,
	offset=offset,
	).items
	class AsyncMemoriesManager(BaseMemoriesManager):
	"""
	Asynchronously lists memories based on various filter parameters.
	Args:
	agent_id (Union[str, UUID]): The unique identifier of the agent.
	query (str): The search query string to filter memories.
	types (Optional[Union[str, List[str]]], optional): The types of memories to filter by. Defaults to None.
	user_id (Optional[str], optional): The unique identifier of the user. Defaults to None.
	limit (Optional[int], optional): The maximum number of memories to return. Defaults to None.
	offset (Optional[int], optional): The number of memories to skip before starting to collect the result set. Defaults to None.
	Returns:
	List[Memory]: A list of Memory objects that match the given filters.
	Raises:
	ValidationError: If the input validation fails.
	DatabaseError: If there is a problem accessing the database.
	"""
	@beartype
	async def list(
	self,
	*,
	agent_id: Union[str, UUID],
	query: str,
	types: Optional[Union[str, List[str]]] = None,
	user_id: Optional[str] = None,
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	) -> List[Memory]:
	"""
	Asynchronously list memories based on query parameters.
	Args:
	agent_id (Union[str, UUID]): The ID of the agent to list memories for.
	query (str): The query string to filter memories.
	types (Optional[Union[str, List[str]]], optional): The types of memories to retrieve. Defaults to None.
	user_id (Optional[str], optional): The ID of the user to list memories for. Defaults to None.
	limit (Optional[int], optional): The maximum number of memories to return. Defaults to None.
	offset (Optional[int], optional): The offset to start listing memories from. Defaults to None.
	Returns:
	List[Memory]: A list of Memory objects that match the query.
	Note:
	`@beartype` decorator is used for runtime type checking.
	"""
	return (
	await self._list(
	agent_id=agent_id,
	query=query,
	types=types,
	user_id=user_id,
	limit=limit,
	offset=offset,
	)

julep/sdks/python/julep/managers/doc.py

Lines 1 to 470 in 26e57ac

	import json
	from typing import Optional, TypedDict
	from uuid import UUID
	from beartype import beartype
	from beartype.typing import Any, Awaitable, Dict, List, Union
	from ..api.types import (
	CreateDoc,
	Doc,
	ResourceCreatedResponse,
	GetAgentDocsResponse,
	)
	from .utils import rewrap_in_class
	from .base import BaseManager
	from .utils import is_valid_uuid4
	from .types import DocDict
	class DocsCreateArgs(TypedDict):
	agent_id: Optional[Union[str, UUID]]
	user_id: Optional[Union[str, UUID]]
	doc: DocDict
	metadata: Dict[str, Any] = {}
	class BaseDocsManager(BaseManager):
	"""
	Manages documents for agents or users by providing methods to get, create, and delete docsrmation.
	The class utilizes an API client to interact with a back-end service that handles the document management operations.
	Typical usage example:
	docs_manager = BaseDocsManager(api_client)
	agent_docs = docs_manager._list(agent_id="some-agent-uuid")
	user_docs = docs_manager._list(user_id="some-user-uuid")
	created_doc = docs_manager._create(agent_id="some-agent-uuid", doc={"key": "value"})
	docs_manager._delete(user_id="some-user-uuid", doc_id="some-doc-uuid")
	Attributes:
	api_client: A client instance used to make API calls to the document management system.
	Methods:
	_list(agent_id: Optional[Union[str, UUID]], user_id: Optional[Union[str, UUID]],
	limit: Optional[int]=None, offset: Optional[int]=None) -> Union[GetAgentDocsResponse, Awaitable[GetAgentDocsResponse]]
	Retrieves docsrmation for either an agent or user.
	Must provide exactly one valid UUID v4 for either `agent_id` or `user_id`.
	_create(agent_id: Optional[Union[str, UUID]], user_id: Optional[Union[str, UUID]], doc: DocDict) -> Union[ResourceCreatedResponse, Awaitable[ResourceCreatedResponse]]
	Creates docsrmation for either an agent or user.
	Must provide exactly one valid UUID v4 for either `agent_id` or `user_id`.
	The `doc` parameter contains the document information to be created.
	_delete(agent_id: Optional[Union[str, UUID]], user_id: Optional[Union[str, UUID]], doc_id: Union[str, UUID]):
	Deletes docsrmation for either an agent or user.
	Must provide exactly one valid UUID v4 for either `agent_id` or `user_id`, and a valid UUID for `doc_id`.
	"""
	def _list(
	self,
	agent_id: Optional[Union[str, UUID]],
	user_id: Optional[Union[str, UUID]],
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	metadata_filter: Dict[str, Any] = {},
	) -> Union[GetAgentDocsResponse, Awaitable[GetAgentDocsResponse]]:
	"""
	Retrieve docsrmation for an agent or user based on their ID.
	This internal method fetches docsrmation for either an agent or a user,
	but not both. If both or neither `agent_id` and `user_id` are provided, it will
	assert an error.
	Args:
	agent_id (Optional[Union[str, UUID]]): The UUID v4 of the agent for whom docs is requested, exclusive with `user_id`.
	user_id (Optional[Union[str, UUID]]): The UUID v4 of the user for whom docs is requested, exclusive with `agent_id`.
	limit (Optional[int]): The maximum number of records to return. Defaults to None.
	offset (Optional[int]): The number of records to skip before starting to collect the response set. Defaults to None.
	Returns:
	Union[GetAgentDocsResponse, Awaitable[GetAgentDocsResponse]]: The response object containing docsrmation about the agent or user, or a promise of such an object if the call is asynchronous.
	Raises:
	AssertionError: If both `agent_id` and `user_id` are provided or neither is provided, or if the provided IDs are not valid UUID v4.
	"""
	metadata_filter_string = json.dumps(metadata_filter)
	assert (
	(agent_id and is_valid_uuid4(agent_id))
	or (user_id and is_valid_uuid4(user_id))
	and not (agent_id and user_id)
	), "One and only one of user_id or agent_id must be given and must be valid UUID v4"
	if agent_id is not None:
	return self.api_client.get_agent_docs(
	agent_id=agent_id,
	limit=limit,
	offset=offset,
	metadata_filter=metadata_filter_string,
	)
	if user_id is not None:
	return self.api_client.get_user_docs(
	user_id=user_id,
	limit=limit,
	offset=offset,
	metadata_filter=metadata_filter_string,
	)
	def _create(
	self,
	doc: DocDict,
	agent_id: Optional[Union[str, UUID]] = None,
	user_id: Optional[Union[str, UUID]] = None,
	metadata: Dict[str, Any] = {},
	) -> Union[ResourceCreatedResponse, Awaitable[ResourceCreatedResponse]]:
	"""
	Create a new resource with docsrmation for either an agent or a user, but not both.
	This function asserts that exactly one of `agent_id` or `user_id` is provided and is a valid UUID v4.
	It then creates the appropriate docsrmation based on which ID was provided.
	Args:
	agent_id (Optional[Union[str, UUID]]): The UUID of the agent or None.
	user_id (Optional[Union[str, UUID]]): The UUID of the user or None.
	doc (DocDict): A dictionary containing the document data for the resource being created.
	Returns:
	Union[ResourceCreatedResponse, Awaitable[ResourceCreatedResponse]]: The response after creating the resource, which could be immediate or an awaitable for asynchronous execution.
	Raises:
	AssertionError: If both `agent_id` and `user_id` are provided, neither are provided, or if the provided IDs are not valid UUID v4 strings.
	Note:
	One and only one of `agent_id` or `user_id` must be provided and must be a valid UUID v4.
	The `DocDict` type should be a dictionary compatible with the `CreateDoc` schema.
	"""
	assert (
	(agent_id and is_valid_uuid4(agent_id))
	or (user_id and is_valid_uuid4(user_id))
	and not (agent_id and user_id)
	), "One and only one of user_id or agent_id must be given and must be valid UUID v4"
	doc: CreateDoc = CreateDoc(**doc)
	if agent_id is not None:
	return self.api_client.create_agent_doc(
	agent_id=agent_id,
	request=doc,
	)
	if user_id is not None:
	return self.api_client.create_user_doc(
	user_id=user_id,
	request=doc,
	)
	def _delete(
	self,
	agent_id: Optional[Union[str, UUID]],
	user_id: Optional[Union[str, UUID]],
	doc_id: Union[str, UUID],
	):
	"""
	Delete docs based on either an agent_id or a user_id.
	This method selects the appropriate deletion operation (agent or user) based on whether an `agent_id` or `user_id` is provided. Only one of these ID types should be valid and provided.
	Args:
	agent_id (Optional[Union[str, UUID]]): A unique identifier of an agent. Either a string or UUID v4, but not both `agent_id` and `user_id`.
	user_id (Optional[Union[str, UUID]]): A unique identifier of a user. Either a string or UUID v4, but not both `agent_id` and `user_id`.
	doc_id (Union[str, UUID]): A unique identifier for docsrmation to be deleted, as a string or UUID v4.
	Returns:
	The result of the API deletion request. This can be the response object from the client's delete operation.
	Raises:
	AssertionError: If neither `agent_id` nor `user_id` is valid, if both are provided simultaneously, or if negation logic for valid UUIDs fails.
	Other exceptions related to the `api_client` operations could potentially be raised and depend on its implementation.
	"""
	assert (
	(agent_id and is_valid_uuid4(agent_id))
	or (user_id and is_valid_uuid4(user_id))
	and not (agent_id and user_id)
	), "One and only one of user_id or agent_id must be given and must be valid UUID v4"
	if agent_id is not None:
	return self.api_client.delete_agent_doc(
	agent_id=agent_id,
	doc_id=doc_id,
	)
	if user_id is not None:
	return self.api_client.delete_user_doc(
	user_id=user_id,
	doc_id=doc_id,
	)
	class DocsManager(BaseDocsManager):
	"""
	A class responsible for managing documents.
	This class provides methods for retrieving, creating, and deleting documents. It uses a base document management system to perform operations.
	Attributes:
	None specific to this class, as all are inherited from BaseDocsManager.
	Methods:
	get:
	Retrieves a list of documents according to specified filters.
	Args:
	agent_id (Optional[Union[str, UUID]]): The agent's identifier to filter documents by, if any.
	user_id (Optional[Union[str, UUID]]): The user's identifier to filter documents by, if any.
	limit (Optional[int]): The maximum number of documents to be retrieved.
	offset (Optional[int]): The number of documents to skip before starting to collect the document output list.
	Returns:
	List[Doc]: A list of documents matching the provided filters.
	create:
	Creates a new document.
	Args:
	agent_id (Optional[Union[str, UUID]]): The agent's identifier associated with the document, if any.
	user_id (Optional[Union[str, UUID]]): The user's identifier associated with the document, if any.
	doc (DocDict): The document to be created represented as a dictionary of document metadata.
	Returns:
	ResourceCreatedResponse: An object representing the creation response, typically containing the ID of the created document.
	delete:
	Deletes a document by its document identifier.
	Args:
	doc_id (Union[str, UUID]): The identifier of the document to be deleted.
	agent_id (Optional[Union[str, UUID]]): The agent's identifier associated with the document, if any.
	user_id (Optional[Union[str, UUID]]): The user's identifier associated with the document, if any.
	Returns:
	None, but the method may raise exceptions on failure.
	"""
	@beartype
	def list(
	self,
	*,
	agent_id: Optional[Union[str, UUID]] = None,
	user_id: Optional[Union[str, UUID]] = None,
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	metadata_filter: Dict[str, Any] = {},
	) -> List[Doc]:
	"""
	Retrieve a list of documents based on specified criteria.
	This method supports filtering the documents by agent_id or user_id, and also supports pagination through the limit and offset parameters.
	Args:
	agent_id (Optional[Union[str, UUID]]): The unique identifier for the agent. Can be a string or a UUID object. Default is None, which means no filtering by agent_id is applied.
	user_id (Optional[Union[str, UUID]]): The unique identifier for the user. Can be a string or a UUID object. Default is None, which means no filtering by user_id is applied.
	limit (Optional[int]): The maximum number of documents to retrieve. Default is None, which means no limit is applied.
	offset (Optional[int]): The number of documents to skip before starting to collect the document list. Default is None, which means no offset is applied.
	Returns:
	List[Doc]: A list of documents that match the provided criteria.
	Note:
	The `@beartype` decorator is used to ensure that the input arguments are of the expected types. If an argument is passed that does not match the expected type, a type error will be raised.
	"""
	return self._list(
	agent_id=agent_id,
	user_id=user_id,
	limit=limit,
	offset=offset,
	metadata_filter=metadata_filter,
	).items
	@beartype
	@rewrap_in_class(Doc)
	def create(self, **kwargs: DocsCreateArgs) -> Doc:
	"""
	Create a new resource with the specified document.
	This method wraps a call to an internal '_create' method, passing along any
	specified agent or user identifiers, along with the document data.
	Args:
	agent_id (Optional[Union[str, UUID]]): The agent identifier associated with the resource creation.
	user_id (Optional[Union[str, UUID]]): The user identifier associated with the resource creation.
	doc (DocDict): A dictionary containing the document data.
	Returns:
	ResourceCreatedResponse: An object representing the response for the resource creation operation.
	Raises:
	BeartypeException: If any input parameters are of incorrect type, due to type enforcement by the @beartype decorator.
	"""
	result = self._create(**kwargs)
	return result
	@beartype
	def delete(
	self,
	*,
	doc_id: Union[str, UUID],
	agent_id: Optional[Union[str, UUID]] = None,
	user_id: Optional[Union[str, UUID]] = None,
	):
	"""
	Deletes a document by its identifier.
	This function wraps the internal _delete method, providing an interface to delete documents by their ID while optionally specifying the agent ID and user ID.
	Args:
	doc_id (Union[str, UUID]): The unique identifier of the document to be deleted.
	agent_id (Optional[Union[str, UUID]]): The unique identifier of the agent performing the delete operation, if any.
	user_id (Optional[Union[str, UUID]]): The unique identifier of the user performing the delete operation, if any.
	Returns:
	The return type depends on the implementation of the `_delete` method.
	Raises:
	The exceptions raised depend on the implementation of the `_delete` method.
	"""
	return self._delete(
	agent_id=agent_id,
	user_id=user_id,
	doc_id=doc_id,
	)
	class AsyncDocsManager(BaseDocsManager):
	"""
	A class for managing asynchronous operations on documents.
	Inherits from BaseDocsManager to provide async document retrieval, creation, and deletion.
	Attributes:
	Inherited from BaseDocsManager.
	Methods:
	async list(self, *, agent_id: Optional[Union[str, UUID]] = None, user_id: Optional[Union[str, UUID]] = None, limit: Optional[int] = None, offset: Optional[int] = None) -> List[Doc]:
	Asynchronously get a list of documents, with optional filtering based on agent_id, user_id, and pagination options limit and offset.
	Args:
	agent_id (Optional[Union[str, UUID]]): The agent's identifier to filter documents.
	user_id (Optional[Union[str, UUID]]): The user's identifier to filter documents.
	limit (Optional[int]): The maximum number of documents to return.
	offset (Optional[int]): The offset from where to start returning documents.
	Returns:
	List[Doc]: A list of documents.
	async create(self, *, agent_id: Optional[Union[str, UUID]] = None, user_id: Optional[Union[str, UUID]] = None, doc: DocDict) -> ResourceCreatedResponse:
	Asynchronously create a new document with the given document information, and optional agent_id and user_id.
	Args:
	agent_id (Optional[Union[str, UUID]]): The agent's identifier associated with the document.
	user_id (Optional[Union[str, UUID]]): The user's identifier associated with the document.
	doc (DocDict): The document data to be created.
	Returns:
	ResourceCreatedResponse: A response object indicating successful creation of the document.
	async delete(self, *, doc_id: Union[str, UUID], agent_id: Optional[Union[str, UUID]] = None, user_id: Optional[Union[str, UUID]] = None):
	Asynchronously delete a document by its id, with optional association to a specific agent_id or user_id.
	Args:
	doc_id (Union[str, UUID]): The unique identifier of the document to be deleted.
	agent_id (Optional[Union[str, UUID]]): The agent's identifier associated with the document, if applicable.
	user_id (Optional[Union[str, UUID]]): The user's identifier associated with the document, if applicable.
	Note:
	The `@beartype` decorator is being used to perform runtime type checking on the function arguments.
	"""
	@beartype
	async def list(
	self,
	*,
	agent_id: Optional[Union[str, UUID]] = None,
	user_id: Optional[Union[str, UUID]] = None,
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	metadata_filter: Dict[str, Any] = {},
	) -> List[Doc]:
	"""
	Asynchronously get a list of documents.
	This function fetches documents based on the provided filtering criteria such as `agent_id`, `user_id`,
	and supports pagination through `limit` and `offset`.
	Args:
	agent_id (Optional[Union[str, UUID]]): The ID of the agent to filter documents by. Default is None.
	user_id (Optional[Union[str, UUID]]): The ID of the user to filter documents by. Default is None.
	limit (Optional[int]): The maximum number of documents to return. Default is None.
	offset (Optional[int]): The offset from where to start the document retrieval. Default is None.
	Returns:
	List[Doc]: A list of document objects.
	Note:
	The `@beartype` decorator is used to ensure that arguments conform to the expected types.
	Raises:
	BeartypeDecorHintPepParamException: If any of the parameters do not adhere to the declared types.
	"""
	return (
	await self._list(
	agent_id=agent_id,
	user_id=user_id,
	limit=limit,
	offset=offset,
	metadata_filter=metadata_filter,
	)
	).items
	@beartype
	@rewrap_in_class(Doc)
	async def create(self, **kwargs: DocsCreateArgs) -> Doc:
	"""
	Create a new resource asynchronously.
	Args:
	agent_id (Optional[Union[str, UUID]]): The ID of the agent. Default is None.
	user_id (Optional[Union[str, UUID]]): The ID of the user. Default is None.
	doc (DocDict): A dictionary containing document data.
	Returns:
	ResourceCreatedResponse: An object representing the response for a resource created.
	Raises:
	BeartypeException: If any of the input arguments do not match their expected types. This is implicitly raised due to the use of the beartype decorator.
	"""
	result = await self._create(**kwargs)
	return result
	@beartype
	async def delete(
	self,
	*,
	doc_id: Union[str, UUID],
	agent_id: Optional[Union[str, UUID]] = None,
	user_id: Optional[Union[str, UUID]] = None,
	):
	"""
	Asynchronously deletes a document by its ID.
	This function is a coroutine and must be awaited.
	Args:
	doc_id (Union[str, UUID]): The unique identifier of the document to delete.
	agent_id (Optional[Union[str, UUID]]): The unique identifier of the agent, if any.
	user_id (Optional[Union[str, UUID]]): The unique identifier of the user, if any.
	Returns:
	Coroutine[Any]: A coroutine that, when awaited, returns the result of the document deletion process.
	Note:
	The `@beartype` decorator is used to enforce type checking on the function arguments.
	"""
	# Assert either user_id or agent_id is provided
	if not agent_id and not user_id:
	raise ValueError("Either agent_id or user_id must be provided.")
	return await self._delete(
	agent_id=agent_id,
	user_id=user_id,
	doc_id=doc_id,

julep/sdks/python/julep/managers/agent.py

Lines 1 to 687 in 26e57ac

	import json
	from typing import Optional, TypedDict
	from uuid import UUID
	from beartype import beartype
	from beartype.typing import Any, Awaitable, Dict, List, Literal, Union
	from ..api.types import (
	Agent,
	AgentDefaultSettings,
	CreateDoc,
	CreateToolRequest,
	ResourceCreatedResponse,
	ListAgentsResponse,
	ResourceUpdatedResponse,
	)
	from .utils import rewrap_in_class
	from .base import BaseManager
	from .utils import is_valid_uuid4, NotSet
	from .types import (
	ToolDict,
	FunctionDefDict,
	DefaultSettingsDict,
	DocDict,
	)
	###########
	## TYPES ##
	###########
	ModelName = Literal[
	"julep-ai/samantha-1",
	"julep-ai/samantha-1-turbo",
	]
	class AgentCreateArgs(TypedDict):
	name: str
	about: Optional[str]
	instructions: Optional[List[str]]
	tools: List[ToolDict] = []
	functions: List[FunctionDefDict] = []
	default_settings: DefaultSettingsDict = {}
	model: ModelName = "julep-ai/samantha-1-turbo"
	docs: List[DocDict] = []
	metadata: Dict[str, Any] = {}
	class AgentUpdateArgs(TypedDict):
	about: Optional[str] = None
	instructions: Optional[List[str]] = None
	name: Optional[str] = None
	model: Optional[str] = None
	default_settings: Optional[DefaultSettingsDict] = None
	metadata: Optional[Dict[str, Any]] = None
	overwrite: bool = False
	class BaseAgentsManager(BaseManager):
	"""
	A class responsible for managing agent entities.
	This manager handles CRUD operations for agents including retrieving, creating, listing, deleting, and updating agents using an API client.
	Attributes:
	api_client (ApiClientType): The client responsible for API interactions.
	Methods:
	_get(self, id: Union[str, UUID]) -> Union[Agent, Awaitable[Agent]]:
	Retrieves a single agent by its UUID.
	Args:
	id (Union[str, UUID]): The UUID of the agent to retrieve.
	Returns:
	The agent object or an awaitable that resolves to the agent object.
	_create(self, name: str, about: str, instructions: List[str], tools: List[ToolDict] = [], functions: List[FunctionDefDict] = [], default_settings: DefaultSettingsDict = {}, model: ModelName = 'julep-ai/samantha-1-turbo', docs: List[DocDict] = [], metadata: Dict[str, Any] = {}) -> Union[ResourceCreatedResponse, Awaitable[ResourceCreatedResponse]]:
	Creates an agent with the given specifications.
	Args:
	name (str): The name of the new agent.
	about (str): Description about the new agent.
	instructions (List[str]): List of instructions or instruction dictionaries for the new agent.
	tools (List[ToolDict], optional): List of tool dictionaries. Defaults to an empty list.
	functions (List[FunctionDefDict], optional): List of function definition dictionaries. Defaults to an empty list.
	default_settings (DefaultSettingsDict, optional): Dictionary of default settings for the new agent. Defaults to an empty dictionary.
	model (ModelName, optional): The model name for the new agent. Defaults to 'julep-ai/samantha-1-turbo'.
	docs (List[DocDict], optional): List of document dictionaries for the new agent. Defaults to an empty list.
	metadata (Dict[str, Any], optional): Dictionary of metadata for the new agent. Defaults to an empty dictionary.
	Returns:
	The response indicating creation or an awaitable that resolves to the creation response.
	_list_items(self, limit: Optional[int] = None, offset: Optional[int] = None) -> Union[ListAgentsResponse, Awaitable[ListAgentsResponse]]:
	Lists agents with pagination support.
	Args:
	limit (Optional[int], optional): The maximum number of agents to list. Defaults to None.
	offset (Optional[int], optional): The number of agents to skip (for pagination). Defaults to None.
	Returns:
	The list of agents or an awaitable that resolves to the list of agents.
	_delete(self, agent_id: Union[str, UUID]) -> Union[None, Awaitable[None]]:
	Deletes an agent with the specified UUID.
	Args:
	agent_id (Union[str, UUID]): The UUID of the agent to delete.
	Returns:
	None or an awaitable that resolves to None.
	_update(self, agent_id: Union[str, UUID], about: Optional[str] = None, instructions: Optional[List[str]] = None, name: Optional[str] = None, model: Optional[str] = None, default_settings: Optional[DefaultSettingsDict] = None, metadata: Dict[str, Any] = {}) -> Union[ResourceUpdatedResponse, Awaitable[ResourceUpdatedResponse]]:
	Updates the specified fields of an agent.
	Args:
	agent_id (Union[str, UUID]): The UUID of the agent to update.
	about (Optional[str], optional): The new description about the agent.
	instructions (Optional[List[str]], optional): The new list of instructions or instruction dictionaries.
	name (Optional[str], optional): The new name for the agent.
	model (Optional[str], optional): The new model name for the agent.
	default_settings (Optional[DefaultSettingsDict], optional): The new default settings dictionary for the agent.
	metadata (Dict[str, Any])
	Returns:
	The response indicating successful update or an awaitable that resolves to the update response.
	"""
	def _get(self, id: Union[str, UUID]) -> Union[Agent, Awaitable[Agent]]:
	"""
	Retrieves an agent based on the provided identifier.
	Args:
	id (Union[str, UUID]): The identifier of the agent, which can be a string or UUID object.
	Returns:
	Union[Agent, Awaitable[Agent]]: The agent object or an awaitable yielding the agent object, depending on the API client.
	Raises:
	AssertionError: If the provided id is not a valid UUID v4.
	"""
	assert is_valid_uuid4(id), "id must be a valid UUID v4"
	return self.api_client.get_agent(agent_id=id)
	def _create(
	self,
	name: str,
	about: str = "",
	instructions: List[str] = [],
	tools: List[ToolDict] = [],
	functions: List[FunctionDefDict] = [],
	default_settings: DefaultSettingsDict = {},
	model: ModelName = "julep-ai/samantha-1-turbo",
	docs: List[DocDict] = [],
	metadata: Dict[str, Any] = {},
	) -> Union[ResourceCreatedResponse, Awaitable[ResourceCreatedResponse]]:
	# Cast instructions to a list of Instruction objects
	"""
	Create a new agent with the specified configuration.
	Args:
	name (str): Name of the agent.
	about (str): Information about the agent.
	instructions (List[str]): List of instructions as either string or dictionaries for the agent.
	tools (List[ToolDict], optional): List of tool configurations for the agent. Defaults to an empty list.
	functions (List[FunctionDefDict], optional): List of function definitions for the agent. Defaults to an empty list.
	default_settings (DefaultSettingsDict, optional): Dictionary of default settings for the agent. Defaults to an empty dict.
	model (ModelName, optional): The model name identifier. Defaults to 'julep-ai/samantha-1-turbo'.
	docs (List[DocDict], optional): List of document configurations for the agent. Defaults to an empty list.
	metadata (Dict[str, Any])
	Returns:
	Union[ResourceCreatedResponse, Awaitable[ResourceCreatedResponse]]: The response object indicating the resource has been created or a future of the response object if the creation is being awaited.
	Raises:
	AssertionError: If both functions and tools are provided.
	Notes:
	The `_create` method is meant to be used internally and should be considered private.
	It assumes the input data for instructions, tools, and docs will have the proper format,
	and items in the 'instructions' list will be converted to Instruction instances.
	"""
	# Ensure that only functions or tools are provided
	assert not (functions and tools), "Only functions or tools can be provided"
	# Cast functions/tools to a list of CreateToolRequest objects
	tools: List[CreateToolRequest] = (
	[
	CreateToolRequest(type="function", function=function)
	for function in functions
	]
	if functions
	else [CreateToolRequest(**tool) for tool in tools]
	)
	# Cast default_settings to AgentDefaultSettings
	default_settings: AgentDefaultSettings = AgentDefaultSettings(
	**default_settings
	)
	# Cast docs to a list of CreateDoc objects
	docs: List[CreateDoc] = [CreateDoc(**doc) for doc in docs]
	return self.api_client.create_agent(
	name=name,
	about=about,
	instructions=instructions,
	tools=tools,
	default_settings=default_settings,
	model=model,
	docs=docs,
	metadata=metadata,
	)
	def _list_items(
	self,
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	metadata_filter: str = "{}",
	) -> Union[ListAgentsResponse, Awaitable[ListAgentsResponse]]:
	"""
	Lists items with optional pagination.
	This method wraps the `list_agents` API call and includes optional limit and offset parameters for pagination.
	Args:
	limit (Optional[int], optional): The maximum number of items to return. Defaults to None, which means no limit.
	offset (Optional[int], optional): The index of the first item to return. Defaults to None, which means no offset.
	Returns:
	Union[ListAgentsResponse, Awaitable[ListAgentsResponse]]: A ListAgentsResponse object, or an awaitable that resolves to a ListAgentsResponse object.
	"""
	return self.api_client.list_agents(
	limit=limit,
	offset=offset,
	metadata_filter=metadata_filter,
	)
	def _delete(self, agent_id: Union[str, UUID]) -> Union[None, Awaitable[None]]:
	"""
	Delete an agent by its ID.
	Args:
	agent_id (Union[str, UUID]): The UUID v4 of the agent to be deleted.
	Returns:
	Union[None, Awaitable[None]]: A future that resolves to None if the
	operation is asynchronous, or None immediately if the operation is
	synchronous.
	Raises:
	AssertionError: If `agent_id` is not a valid UUID v4.
	"""
	assert is_valid_uuid4(agent_id), "id must be a valid UUID v4"
	return self.api_client.delete_agent(agent_id=agent_id)
	def _update(
	self,
	agent_id: Union[str, UUID],
	about: Optional[str] = NotSet,
	instructions: List[str] = NotSet,
	name: Optional[str] = NotSet,
	model: Optional[str] = NotSet,
	default_settings: Optional[DefaultSettingsDict] = NotSet,
	metadata: Dict[str, Any] = NotSet,
	overwrite: bool = False,
	) -> Union[ResourceUpdatedResponse, Awaitable[ResourceUpdatedResponse]]:
	"""
	Update the agent's properties.
	Args:
	agent_id (Union[str, UUID]): The unique identifier for the agent, which can be a string or UUID object.
	about (Optional[str], optional): A brief description of the agent. Defaults to None.
	instructions (Optional[List[str]], optional): A list of either strings or instruction dictionaries that will be converted into Instruction objects. Defaults to None.
	name (Optional[str], optional): The name of the agent. Defaults to None.
	model (Optional[str], optional): The model identifier for the agent. Defaults to None.
	default_settings (Optional[DefaultSettingsDict], optional): A dictionary of default settings to apply to the agent. Defaults to None.
	metadata (Dict[str, Any])
	overwrite (bool, optional): Whether to overwrite the existing agent settings. Defaults to False.
	Returns:
	Union[ResourceUpdatedResponse, Awaitable[ResourceUpdatedResponse]]: An object representing the response for the resource updated, which can also be an awaitable in asynchronous contexts.
	Raises:
	AssertionError: If the provided agent_id is not a valid UUID v4.
	Note:
	This method asserts that the agent_id must be a valid UUID v4. The instructions and default_settings, if provided, are converted into their respective object types before making the update API call.
	"""
	assert is_valid_uuid4(agent_id), "id must be a valid UUID v4"
	if default_settings is not NotSet:
	default_settings: AgentDefaultSettings = AgentDefaultSettings(
	**default_settings
	)
	updateFn = (
	self.api_client.update_agent if overwrite else self.api_client.patch_agent
	)
	update_payload = dict(
	agent_id=agent_id,
	about=about,
	instructions=instructions,
	name=name,
	model=model,
	default_settings=default_settings,
	metadata=metadata,
	)
	update_payload = {k: v for k, v in update_payload.items() if v is not NotSet}
	return updateFn(**update_payload)
	class AgentsManager(BaseAgentsManager):
	"""
	A class for managing agents, inheriting from `BaseAgentsManager`.
	This class provides functionalities to interact with and manage agents, including creating, retrieving, listing, updating, and deleting agents. It utilizes type annotations to ensure type correctness at runtime using the `beartype` decorator.
	Methods:
	get(id: Union[str, UUID]) -> Agent:
	Retrieves an agent by its unique identifier.
	Args:
	id (Union[str, UUID]): The unique identifier of the agent, which can be either a string or a UUID.
	Returns:
	Agent: The agent with the corresponding identifier.
	create(*, name: str, about: str, instructions: List[str], tools: List[ToolDict]=[], functions: List[FunctionDefDict]=[], default_settings: DefaultSettingsDict={}, model: ModelName='julep-ai/samantha-1-turbo', docs: List[DocDict]=[]) -> ResourceCreatedResponse:
	Creates a new agent with the provided details.
	Args:
	name (str): The name of the agent.
	about (str): A description of the agent.
	instructions (List[str]): A list of instructions or dictionaries defining instructions.
	tools (List[ToolDict], optional): A list of dictionaries defining tools. Defaults to an empty list.
	functions (List[FunctionDefDict], optional): A list of dictionaries defining functions. Defaults to an empty list.
	default_settings (DefaultSettingsDict, optional): A dictionary of default settings. Defaults to an empty dictionary.
	model (ModelName, optional): The model name to be used. Defaults to 'julep-ai/samantha-1-turbo'.
	docs (List[DocDict], optional): A list of dictionaries defining documentation. Defaults to an empty list.
	metadata (Dict[str, Any])
	Returns:
	ResourceCreatedResponse: The response indicating the resource (agent) was successfully created.
	list(*, limit: Optional[int]=None, offset: Optional[int]=None) -> List[Agent]:
	Lists all agents with pagination support.
	Args:
	limit (Optional[int], optional): The maximum number of agents to retrieve. Defaults to None, meaning no limit.
	offset (Optional[int], optional): The number of agents to skip (for pagination). Defaults to None.
	Returns:
	List[Agent]: A list of agents, considering the pagination parameters.
	delete(agent_id: Union[str, UUID]):
	Deletes an agent by its unique identifier.
	Args:
	agent_id (Union[str, UUID]): The unique identifier of the agent to be deleted.
	update(*, agent_id: Union[str, UUID], about: Optional[str]=None, instructions: Optional[List[str]]=None, name: Optional[str]=None, model: Optional[str]=None, default_settings: Optional[DefaultSettingsDict]=None) -> ResourceUpdatedResponse:
	Updates an existing agent with new details.
	Args:
	agent_id (Union[str, UUID]): The unique identifier of the agent to be updated.
	about (Optional[str], optional): A new description of the agent. Defaults to None (no change).
	instructions (Optional[List[str]], optional): A new list of instructions or dictionaries defining instructions. Defaults to None (no change).
	name (Optional[str], optional): A new name for the agent. Defaults to None (no change).
	model (Optional[str], optional): A new model name to be used. Defaults to None (no change).
	default_settings (Optional[DefaultSettingsDict], optional): A new dictionary of default settings. Defaults to None (no change).
	metadata (Dict[str, Any])
	Returns:
	ResourceUpdatedResponse: The response indicating the resource (agent) was successfully updated.
	"""
	@beartype
	def get(self, id: Union[str, UUID]) -> Agent:
	"""
	Retrieve an Agent object by its identifier.
	Args:
	id (Union[str, UUID]): The unique identifier of the Agent to be retrieved.
	Returns:
	Agent: An instance of the Agent with the specified ID.
	Raises:
	BeartypeException: If the type of `id` is neither a string nor a UUID.
	Any exception raised by the `_get` method.
	"""
	return self._get(id=id)
	@beartype
	@rewrap_in_class(Agent)
	def create(self, **kwargs: AgentCreateArgs) -> Agent:
	"""
	Creates a new resource with the specified details.
	Args:
	name (str): The name of the resource.
	about (str): A description of the resource.
	instructions (List[str]): A list of instructions or dictionaries with instruction details.
	tools (List[ToolDict], optional): A list of dictionaries with tool details. Defaults to an empty list.
	functions (List[FunctionDefDict], optional): A list of dictionaries with function definition details. Defaults to an empty list.
	default_settings (DefaultSettingsDict, optional): A dictionary with default settings. Defaults to an empty dictionary.
	model (ModelName, optional): The name of the model to use. Defaults to 'julep-ai/samantha-1-turbo'.
	docs (List[DocDict], optional): A list of dictionaries with documentation details. Defaults to an empty list.
	metadata (Dict[str, Any])
	Returns:
	Agent: An instance of the Agent with the specified details
	Note:
	This function is decorated with `@beartype`, which will perform runtime type checking on the arguments.
	"""
	result = self._create(**kwargs)
	return result
	@beartype
	def list(
	self,
	*,
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	metadata_filter: Dict[str, Any] = {},
	) -> List[Agent]:
	"""
	List the Agent objects, possibly with pagination.
	Args:
	limit (Optional[int], optional): The maximum number of Agent objects to return.
	Defaults to None, meaning no limit is applied.
	offset (Optional[int], optional): The number of initial Agent objects to skip before
	starting to collect the return list. Defaults to None,
	meaning no offset is applied.
	Returns:
	List[Agent]: A list of Agent objects.
	Raises:
	BeartypeDecorHintPepParamViolation: If the function is called with incorrect types
	for the `limit` or `offset` parameters.
	"""
	metadata_filter_string = json.dumps(metadata_filter)
	return self._list_items(
	limit=limit,
	offset=offset,
	metadata_filter=metadata_filter_string,
	).items
	@beartype
	def delete(self, agent_id: Union[str, UUID]):
	"""
	Delete the agent with the specified ID.
	Args:
	agent_id (Union[str, UUID]): The identifier of the agent to be deleted.
	Returns:
	The return type depends on the implementation of the `_delete` method. This will typically be `None`
	if the deletion is successful, or an error may be raised if the deletion fails.
	Note:
	The `@beartype` decorator is used to enforce type checking of the `agent_id` parameter.
	"""
	return self._delete(agent_id=agent_id)
	@beartype
	@rewrap_in_class(Agent)
	def update(self, *, agent_id: Union[str, UUID], **kwargs: AgentUpdateArgs) -> Agent:
	"""
	Update the properties of a resource.
	This function updates various attributes of an existing resource based on the provided keyword arguments. All updates are optional and are applied only if the corresponding argument is given.
	Args:
	agent_id (Union[str, UUID]): The identifier of the agent, either as a string or a UUID object.
	about (Optional[str], optional): A brief description of the agent. Defaults to None.
	instructions (Optional[List[str]], optional): A list of instructions or instruction dictionaries to update the agent with. Defaults to None.
	name (Optional[str], optional): The new name to assign to the agent. Defaults to None.
	model (Optional[str], optional): The model identifier to associate with the agent. Defaults to None.
	default_settings (Optional[DefaultSettingsDict], optional): A dictionary of default settings to apply to the agent. Defaults to None.
	metadata (Dict[str, Any])
	overwrite (bool, optional): Whether to overwrite the existing agent settings. Defaults to False.
	Returns:
	ResourceUpdatedResponse: An object representing the response to the update request.
	Note:
	This method is decorated with `beartype`, which means it enforces type annotations at runtime.
	"""
	result = self._update(agent_id=agent_id, **kwargs)
	return result
	class AsyncAgentsManager(BaseAgentsManager):
	"""
	A class for managing asynchronous agent operations.
	This class provides asynchronous methods for creating, retrieving, updating,
	listing, and deleting agents. It is a subclass of BaseAgentsManager, which
	defines the underlying functionality and structure that this class utilizes.
	Attributes:
	None explicitly listed, as they are inherited from the `BaseAgentsManager` class.
	Methods:
	get:
	Retrieves a single agent by its ID.
	Args:
	id (Union[UUID, str]): The unique identifier of the agent to retrieve.
	Returns:
	Agent: The requested agent.
	create:
	Creates a new agent with the provided specifications.
	Args:
	name (str): The name of the agent to create.
	about (str): A description of the agent.
	instructions (List[str]): The instructions for operating the agent.
	tools (List[ToolDict], optional): An optional list of tools for the agent.
	functions (List[FunctionDefDict], optional): An optional list of functions the agent can perform.
	default_settings (DefaultSettingsDict, optional): Optional default settings for the agent.
	model (ModelName, optional): The model name to associate with the agent, defaults to 'julep-ai/samantha-1-turbo'.
	docs (List[DocDict], optional): An optional list of documents associated with the agent.
	metadata (Dict[str, Any])
	Returns:
	ResourceCreatedResponse: A response indicating the agent was created successfully.
	list:
	Lists agents with optional pagination.
	Args:
	limit (Optional[int], optional): The maximum number of agents to retrieve.
	offset (Optional[int], optional): The number of agents to skip before starting to collect the results.
	Returns:
	List[Agent]: A list of agents.
	delete:
	Deletes an agent by its ID.
	Args:
	agent_id (Union[str, UUID]): The unique identifier of the agent to delete.
	Returns:
	The response from the delete operation (specific return type may vary).
	update:
	Updates the specified fields of an agent by its ID.
	Args:
	agent_id (Union[str, UUID]): The unique identifier of the agent to update.
	about (Optional[str], optional): An optional new description for the agent.
	instructions (Optional[List[str]], optional): Optional new instructions for the agent.
	name (Optional[str], optional): An optional new name for the agent.
	model (Optional[str], optional): Optional new model associated with the agent.
	default_settings (Optional[DefaultSettingsDict], optional): Optional new default settings for the agent.
	metadata (Dict[str, Any])
	Returns:
	ResourceUpdatedResponse: A response indicating the agent was updated successfully.
	"""
	@beartype
	async def get(self, id: Union[UUID, str]) -> Agent:
	"""
	Asynchronously retrieve an Agent object by its ID.
	The `id` parameter can be either a UUID or a string representation of a UUID.
	Args:
	id (Union[UUID, str]): The unique identifier of the Agent to retrieve.
	Returns:
	Agent: The Agent object associated with the given id.
	Raises:
	Beartype exceptions: If the input id does not conform to the specified types.
	Other exceptions: Depending on the implementation of the `_get` method.
	"""
	return await self._get(id=id)
	@beartype
	@rewrap_in_class(Agent)
	async def create(self, **kwargs: AgentCreateArgs) -> Agent:
	"""
	Create a new resource asynchronously with specified details.
	This function is decorated with `beartype` to ensure that arguments conform to specified types.
	Args:
	name (str): The name of the resource to create.
	about (str): Information or description about the resource.
	instructions (List[str]): A list of strings or dictionaries detailing the instructions for the resource.
	tools (List[ToolDict], optional): A list of dictionaries representing the tools associated with the resource. Defaults to an empty list.
	functions (List[FunctionDefDict], optional): A list of dictionaries defining functions that can be performed with the resource. Defaults to an empty list.
	default_settings (DefaultSettingsDict, optional): A dictionary with default settings for the resource. Defaults to an empty dictionary.
	model (ModelName, optional): The model identifier to use for the resource. Defaults to 'julep-ai/samantha-1-turbo'.
	docs (List[DocDict], optional): A list of dictionaries containing documentation for the resource. Defaults to an empty list.
	metadata (Dict[str, Any])
	Returns:
	Agent: An instance of the Agent with the specified details
	Raises:
	The exceptions that may be raised are not specified in the signature and depend on the implementation of the _create method.
	"""
	result = await self._create(**kwargs)
	return result
	@beartype
	async def list(
	self,
	*,
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	metadata_filter: Dict[str, Any] = {},
	) -> List[Agent]:
	"""
	Asynchronously lists agents with optional limit and offset.
	This method wraps the call to a private method '_list_items' which performs the actual listing
	of agent items. It uses the 'beartype' decorator for runtime type checking.
	Args:
	limit (Optional[int], optional): The maximum number of agent items to return. Defaults to None, which means no limit.
	offset (Optional[int], optional): The offset from where to start the listing. Defaults to None, which means start from the beginning.
	Returns:
	List[Agent]: A list of agent items collected based on the provided 'limit' and 'offset' parameters.
	"""
	metadata_filter_string = json.dumps(metadata_filter)
	return (
	await self._list_items(
	limit=limit,
	offset=offset,
	metadata_filter=metadata_filter_string,
	)
	).items
	@beartype
	async def delete(self, agent_id: Union[str, UUID]):
	"""
	Asynchronously deletes an agent given its identifier.
	This function is decorated with @beartype to ensure type checking of the input argument at runtime.
	Args:
	agent_id (Union[str, UUID]): The identifier of the agent to be deleted. Can be a string or a UUID object.
	Returns:
	The result of the asynchronous deletion operation, which is implementation-dependent.
	"""
	return await self._delete(agent_id=agent_id)
	@beartype
	@rewrap_in_class(Agent)
	async def update(
	self, *, agent_id: Union[str, UUID], **kwargs: AgentUpdateArgs
	) -> Agent:
	"""
	Asynchronously update an agent's details.
	This function is decorated with `beartype` to enforce the type checking of parameters. It updates the properties of the agent identified by `agent_id`.
	Args:
	agent_id (Union[str, UUID]): Unique identifier for the agent. It can be a string or a UUID object.
	about (Optional[str]): Additional information about the agent. Default is None.
	instructions (Optional[List[str]]): A list of instructions or instruction dictionaries. Default is None.
	name (Optional[str]): The name of the agent. Default is None.
	model (Optional[str]): The model identifier or name. Default is None.
	default_settings (Optional[DefaultSettingsDict]): Dictionary with default settings for the agent. Default is None.
	metadata (Dict[str, Any])
	overwrite (bool): Whether to overwrite the existing agent settings. Default is False.
	Returns:
	ResourceUpdatedResponse: An object containing the details of the update response.
	"""
	result = await self._update(agent_id=agent_id, **kwargs)

julep/sdks/python/julep/managers/user.py

Lines 1 to 507 in 26e57ac

	import json
	from uuid import UUID
	from typing import Optional, TypedDict
	from beartype import beartype
	from beartype.typing import Any, Awaitable, Dict, List, Union
	from .utils import rewrap_in_class, NotSet
	from ..api.types import (
	User,
	CreateDoc,
	ResourceCreatedResponse,
	ResourceUpdatedResponse,
	ListUsersResponse,
	)
	from .base import BaseManager
	from .utils import is_valid_uuid4
	from .types import DocDict
	class UserCreateArgs(TypedDict):
	name: str
	about: str
	docs: List[str] = []
	metadata: Dict[str, Any] = {}
	class UserUpdateArgs(TypedDict):
	name: Optional[str] = None
	metadata: Optional[Dict[str, Any]] = None
	about: Optional[str] = None
	overwrite: bool = False
	class BaseUsersManager(BaseManager):
	"""
	A manager class for handling user-related operations through an API client.
	This class provides high-level methods to interact with user records,
	such as retrieving a user by ID, creating a new user, listing all users,
	deleting a user, and updating a user's details.
	Attributes:
	api_client: The API client instance to communicate with the user service.
	Methods:
	_get: Retrieve a user by a unique UUID.
	_create: Create a new user with the specified name and about fields, and optionally additional documents.
	_list_items: List users with optional pagination through limit and offset parameters.
	_delete: Delete a user by UUID.
	_update: Update user details such as the 'about' description and name by UUID.
	Raises:
	AssertionError: If the provided ID for the user operations is not a valid UUID v4.
	"""
	def _get(self, id: Union[str, UUID]) -> Union[User, Awaitable[User]]:
	"""
	Get the user by their ID.
	This method is intended to retrieve a User object or an Awaitable User object by the user's unique identifier. The identifier can be a string representation of a UUID or a UUID object itself.
	Args:
	id (Union[str, UUID]): The unique identifier of the user to retrieve. It must be a valid UUID v4.
	Returns:
	Union[User, Awaitable[User]]: The retrieved User instance or an Awaitable that resolves to a User instance.
	Raises:
	AssertionError: If the `id` is not a valid UUID v4.
	Note:
	The leading underscore in the method name suggests that this method is intended for internal use and should not be a part of the public interface of the class.
	"""
	assert is_valid_uuid4(id), "id must be a valid UUID v4"
	return self.api_client.get_user(user_id=id)
	def _create(
	self,
	name: str,
	about: str,
	docs: List[DocDict] = [],
	metadata: Dict[str, Any] = {},
	) -> Union[ResourceCreatedResponse, Awaitable[ResourceCreatedResponse]]:
	# Cast docs to a list of CreateDoc objects
	"""
	Create a new resource with the given name and about information, optionally including additional docs.
	This internal method allows for creating a new resource with optional docsrmation.
	Args:
	name (str): The name of the new resource.
	about (str): A brief description about the new resource.
	docs (List[DocDict], optional): A list of dictionaries with documentation-related information. Each dictionary
	must conform to the structure expected by CreateDoc. Defaults to an empty list.
	metadata (Dict[str, Any])
	Returns:
	Union[ResourceCreatedResponse, Awaitable[ResourceCreatedResponse]]: The response indicating the resource has been
	created successfully. It can be either a synchronous ResourceCreatedResponse or an asynchronous Awaitable object
	containing a ResourceCreatedResponse.
	Note:
	This method is an internal API implementation detail and should not be used directly outside the defining class
	or module.
	Side effects:
	Modifies internal state by converting each doc dict to an instance of CreateDoc and uses the
	internal API client to create a new user resource.
	"""
	docs: List[CreateDoc] = [CreateDoc(**doc) for doc in docs]
	return self.api_client.create_user(
	name=name,
	about=about,
	docs=docs,
	metadata=metadata,
	)
	def _list_items(
	self,
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	metadata_filter: str = "{}",
	) -> Union[ListUsersResponse, Awaitable[ListUsersResponse]]:
	"""
	Fetch a list of users, with optional pagination parameters.
	Args:
	limit (Optional[int], optional): The maximum number of users to return. Defaults to None.
	offset (Optional[int], optional): The offset from the start of the list to begin returning users. Defaults to None.
	Returns:
	Union[ListUsersResponse, Awaitable[ListUsersResponse]]: An instance of ListUsersResponse,
	or an awaitable that will resolve to it, depending on the API client implementation.
	"""
	return self.api_client.list_users(
	limit=limit,
	offset=offset,
	metadata_filter=metadata_filter,
	)
	def _delete(self, user_id: Union[str, UUID]) -> Union[None, Awaitable[None]]:
	"""
	Delete a user given their user ID.
	Args:
	user_id (Union[str, UUID]): The identifier of the user. Must be a valid UUID version 4.
	Returns:
	Union[None, Awaitable[None]]: None if the deletion is synchronous, or an Awaitable
	that resolves to None if the deletion is asynchronous.
	Raises:
	AssertionError: If the user_id is not a valid UUID v4.
	"""
	assert is_valid_uuid4(user_id), "id must be a valid UUID v4"
	return self.api_client.delete_user(user_id=user_id)
	def _update(
	self,
	user_id: Union[str, UUID],
	about: Optional[str] = NotSet,
	name: Optional[str] = NotSet,
	metadata: Dict[str, Any] = NotSet,
	overwrite: bool = False,
	) -> Union[ResourceUpdatedResponse, Awaitable[ResourceUpdatedResponse]]:
	"""
	Update user details for a given user ID.
	This method updates the 'about' and/or 'name' fields for the user identified by user_id.
	Args:
	user_id (Union[str, UUID]): The ID of the user to be updated. Must be a valid UUID v4.
	about (Optional[str], optional): The new information about the user. Defaults to None.
	name (Optional[str], optional): The new name for the user. Defaults to None.
	metadata (Dict[str, Any])
	overwrite (bool, optional): Whether to overwrite the existing user data. Defaults to False.
	Returns:
	Union[ResourceUpdatedResponse, Awaitable[ResourceUpdatedResponse]]: The response indicating successful update or an Awaitable that resolves to such a response.
	Raises:
	AssertionError: If `user_id` is not a valid UUID v4.
	"""
	assert is_valid_uuid4(user_id), "id must be a valid UUID v4"
	updateFn = (
	self.api_client.update_user if overwrite else self.api_client.patch_user
	)
	update_data = dict(
	user_id=user_id,
	about=about,
	name=name,
	metadata=metadata,
	)
	update_data = {k: v for k, v in update_data.items() if v is not NotSet}
	if not update_data:
	raise ValueError("No fields to update")
	return updateFn(**update_data)
	class UsersManager(BaseUsersManager):
	"""
	A class responsible for managing users in a system.
	This class is a specialized version of the BaseUsersManager and provides
	methods for retrieving, creating, listing, deleting, and updating users within
	the system.
	Methods:
	get(id: Union[str, UUID]) -> User:
	Retrieves a user based on their unique identifier (either a string or UUID).
	create(*, name: str, about: str, docs: List[DocDict]=[]) -> ResourceCreatedResponse:
	Creates a new user with the specified name, description about the user,
	and an optional list of documents associated with the user.
	list(*, limit: Optional[int]=None, offset: Optional[int]=None) -> List[User]:
	Lists users in the system, with optional limit and offset for pagination.
	delete(*, user_id: Union[str, UUID]) -> None:
	Deletes a user from the system based on their unique identifier.
	update(*, user_id: Union[str, UUID], about: Optional[str]=None, name: Optional[str]=None) -> ResourceUpdatedResponse:
	Updates an existing user's information with optional new about and name
	fields.
	"""
	@beartype
	def get(self, id: Union[str, UUID]) -> User:
	"""
	Retrieve a User object by its identifier.
	The method supports retrieval by both string representations of a UUID and
	UUID objects directly.
	Args:
	id (Union[str, UUID]): The identifier of the User, can be a string or UUID.
	Returns:
	User: The User object associated with the provided id.
	Raises:
	ValueError: If 'id' is neither a string nor a UUID.
	NotFoundError: If a User with the given 'id' does not exist.
	"""
	return self._get(id=id)
	@beartype
	@rewrap_in_class(User)
	def create(self, **kwargs: UserCreateArgs) -> User:
	"""
	Create a new resource with the specified name, about text, and associated docs.
	Args:
	name (str): The name of the resource to create.
	about (str): A brief description of the resource.
	docs (List[DocDict], optional): A list of dictionaries representing the documents associated with the resource. Defaults to an empty list.
	Returns:
	ResourceCreatedResponse: An object representing the response received upon the successful creation of the resource.
	Note:
	Using mutable types like list as default argument values in Python is risky because if the list is modified,
	those changes will persist across subsequent calls to the function which use the default value. It is
	generally safer to use `None` as a default value and then set the default inside the function if needed.
	Raises:
	BeartypeException: If the input types do not match the specified function annotations.
	"""
	result = self._create(**kwargs)
	return result
	@beartype
	def list(
	self,
	*,
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	metadata_filter: Dict[str, Any] = {},
	) -> List[User]:
	"""
	Lists the users optionally applying limit and offset.
	Args:
	limit (Optional[int], optional): The maximum number of users to return.
	None means no limit. Defaults to None.
	offset (Optional[int], optional): The index of the first user to return.
	None means start from the beginning. Defaults to None.
	Returns:
	List[User]: A list of user objects.
	Raises:
	BeartypeException: If the type of `limit` or `offset` is not as expected.
	"""
	metadata_filter_string = json.dumps(metadata_filter)
	return self._list_items(
	limit=limit,
	offset=offset,
	metadata_filter=metadata_filter_string,
	).items
	@beartype
	def delete(
	self,
	user_id: Union[str, UUID],
	) -> None:
	"""
	Deletes a user based on the provided user ID.
	Args:
	user_id (Union[str, UUID]): Unique identifier of the user.
	Returns:
	None
	Note:
	This function is type-checked with `beartype` to ensure that the `user_id`
	parameter matches either a string or a UUID type.
	Raises:
	The specific exceptions raised depend on the implementation of the `_delete`
	method this function proxies to.
	"""
	return self._delete(
	user_id=user_id,
	)
	@beartype
	@rewrap_in_class(User)
	def update(self, *, user_id: Union[str, UUID], **kwargs: UserUpdateArgs) -> User:
	"""
	Update user information.
	This method updates user details such as the `about` text and user's `name` for a given `user_id`.
	Args:
	user_id (Union[str, UUID]): The unique identifier for the user, which can be a string or a UUID object.
	about(Optional[str], optional): The descriptive information about the user. Defaults to None, indicating that `about` should not be updated if not provided.
	name(Optional[str], optional): The name of the user. Defaults to None, indicating that `name` should not be updated if not provided.
	overwrite(bool, optional): Whether to overwrite the existing user data. Defaults to False.
	Returns:
	ResourceUpdatedResponse: An object indicating the outcome of the update operation, which typically includes the status of the operation and possibly the updated resource data.
	"""
	result = self._update(user_id=user_id, **kwargs)
	return result
	class AsyncUsersManager(BaseUsersManager):
	"""
	A class that provides asynchronous management of users extending BaseUsersManager.
	Attributes are inherited from BaseUsersManager.
	Methods:
	get (Union[UUID, str]) -> User:
	Asynchronously retrieves a user by their unique identifier (either a UUID or a string).
	create (*, name: str, about: str, docs: List[DocDict]=[]) -> ResourceCreatedResponse:
	Asynchronously creates a new user with the given name, about description, and optional list of documents.
	list (*, limit: Optional[int]=None, offset: Optional[int]=None) -> List[User]:
	Asynchronously retrieves a list of users with an optional limit and offset for pagination.
	delete (*, user_id: Union[str, UUID]) -> None:
	Asynchronously deletes a user identified by their unique ID (either a string or a UUID).
	update (*, user_id: Union[str, UUID], about: Optional[str]=None, name: Optional[str]=None) -> ResourceUpdatedResponse:
	Asynchronously updates a user's information identified by their unique ID with optional new about description and name.
	Note:
	The beartype decorator is used for runtime type checking of the parameters passed to the methods.
	"""
	@beartype
	async def get(self, id: Union[UUID, str]) -> User:
	"""
	Fetch a User object asynchronously by its identifier.
	This method retrieves a User object from some data storage asynchronously based on the provided identifier, which can be either a UUID or a string.
	Args:
	id (Union[UUID, str]): The unique identifier of the User to be retrieved.
	Returns:
	User: An instance of the User class corresponding to the given id.
	Raises:
	Exception: If the retrieval fails or the user cannot be found.
	"""
	return await self._get(id=id)
	@beartype
	@rewrap_in_class(User)
	async def create(self, **kwargs: UserCreateArgs) -> User:
	"""
	Asynchronously create a new resource with the provided name, description, and documents.
	This function is decorated with `@beartype` to ensure type checking at runtime.
	Args:
	name (str): The name of the resource to be created.
	about (str): Brief information about the resource.
	docs (List[DocDict], optional): A list of document dictionaries with structure defined by DocDict type.
	Returns:
	ResourceCreatedResponse: An instance representing the response after resource creation.
	Raises:
	BeartypeException: If any of the parameters do not match their annotated types.
	"""
	result = await self._create(**kwargs)
	return result
	@beartype
	async def list(
	self,
	*,
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	metadata_filter: Dict[str, Any] = {},
	) -> List[User]:
	"""
	Asynchronously lists users with optional limits and offsets.
	This function applies the `beartype` decorator for runtime type checking.
	Args:
	limit (Optional[int], optional): The maximum number of users to be returned. Defaults to None, which means no limit.
	offset (Optional[int], optional): The number to offset the list of returned users by. Defaults to None, which means no offset.
	Returns:
	List[User]: A list of user objects.
	Raises:
	TypeError: If any input arguments are not of the expected type.
	Any other exception that might be raised during the retrieval of users from the data source.
	Note:
	The actual exception raised by the `beartype` decorator or during the users' retrieval will depend on the implementation detail of the `self._list_items` method and the `beartype` configuration.
	"""
	metadata_filter_string = json.dumps(metadata_filter)
	return (
	await self._list_items(
	limit,
	offset,
	metadata_filter=metadata_filter_string,
	)
	).items
	@beartype
	async def delete(
	self,
	user_id: Union[str, UUID],
	) -> None:
	"""
	Asynchronously deletes a user by their user ID.
	Args:
	user_id (Union[str, UUID]): The unique identifier of the user to delete, which can be a string or a UUID.
	Returns:
	None: This function does not return anything.
	Notes:
	- The function is decorated with `@beartype` for runtime type checking.
	- This function is a coroutine, it should be called with `await`.
	Raises:
	- The raised exceptions depend on the implementation of the underlying `_delete` coroutine.
	"""
	return await self._delete(
	user_id=user_id,
	)
	@beartype
	@rewrap_in_class(User)
	async def update(
	self, *, user_id: Union[str, UUID], **kwargs: UserUpdateArgs
	) -> User:
	"""
	Asynchronously updates user details.
	This function updates user details such as 'about' and 'name'. It uses type annotations to enforce the types of the parameters and an asynchronous call to '_update' method to perform the actual update operation.
	Args:
	user_id (Union[str, UUID]): The unique identifier of the user, which can be either a string or a UUID.
	about (Optional[str], optional): A description of the user. Default is None, indicating no update.
	name (Optional[str], optional): The name of the user. Default is None, indicating no update.
	Returns:
	ResourceUpdatedResponse: An object representing the update status.
	Note:
	This function is decorated with @beartype to perform runtime type checking.
	"""
	result = await self._update(user_id=user_id, **kwargs)

julep/sdks/python/julep/client.py

Lines 1 to 266 in 26e57ac

	from typing import Dict, Optional
	from urllib.parse import urlparse
	from beartype import beartype
	import httpx
	from openai import AsyncOpenAI, OpenAI
	from openai.resources.chat.chat import AsyncChat, Chat
	from openai.resources.completions import AsyncCompletions, Completions
	# Note: This is just here because fern generates docs where it asks to:
	# `from julep_ai.client import AsyncJulepApi, JulepApi`
	from .api.client import AsyncJulepApi, JulepApi # noqa: F401
	from .env import JULEP_API_KEY, JULEP_API_URL
	from .managers.agent import AgentsManager, AsyncAgentsManager
	from .managers.user import UsersManager, AsyncUsersManager
	from .managers.doc import DocsManager, AsyncDocsManager
	from .managers.memory import MemoriesManager, AsyncMemoriesManager
	from .managers.session import SessionsManager, AsyncSessionsManager
	from .managers.tool import ToolsManager, AsyncToolsManager
	from .utils.openai_patch import (
	patch_chat_acreate,
	patch_chat_create,
	patch_completions_acreate,
	patch_completions_create,
	)
	# See Note above
	__all__ = ["AsyncJulepApi", "JulepApi", "Client", "AsyncClient"]
	def get_base_url(url):
	return f"{urlparse(url).scheme}://{urlparse(url).netloc}"
	class Client:
	"""
	A class that encapsulates managers for different aspects of a system and provides an interface for interacting with an API.
	This class initializes and makes use of various manager classes to handle agents, users, sessions, documents, memories, and tools. It requires an API key and a base URL to initialize the API client that the managers will use.
	Attributes:
	agents (AgentsManager): A manager instance for handling agents.
	users (UsersManager): A manager instance for handling users.
	sessions (SessionsManager): A manager instance for handling sessions.
	docs (DocsManager): A manager instance for handling documents.
	memories (MemoriesManager): A manager instance for handling memories.
	tools (ToolsManager): A manager instance for handling tools.
	chat (Chat): A chat manager instance for handling chat interactions (based on OpenAI client).
	completions (Completions): A manager instance for handling completions (based on OpenAI client).
	Args:
	api_key (Optional[str]): The API key needed to authenticate with the API. Defaults to the JULEP_API_KEY environment variable.
	base_url (Optional[str]): The base URL for the API endpoints. Defaults to the JULEP_API_URL environment variable.
	*args: Variable length argument list.
	**kwargs: Arbitrary keyword arguments.
	Raises:
	AssertionError: If either `api_key` or `base_url` is not provided and not set as an environment variable.
	Note:
	`beartype` decorator is expected to ensure type checking on the parameters during runtime. The constants `JULEP_API_KEY` and `JULEP_API_URL` should be predefined and represent default values for the API key and base URL, respectively, which can be overridden by providing a value at instantiation.
	"""
	agents: AgentsManager
	users: UsersManager
	sessions: SessionsManager
	docs: DocsManager
	memories: MemoriesManager
	tools: ToolsManager
	chat: Chat
	completions: Completions
	@beartype
	def __init__(
	self,
	api_key: Optional[str] = JULEP_API_KEY,
	base_url: Optional[str] = JULEP_API_URL,
	timeout: int = 300,
	additional_headers: Dict[str, str] = {},
	_httpx_client: Optional[httpx.Client] = None,
	*args,
	**kwargs,
	):
	"""
	Initialize a new client object with the given API key and base URL.
	Args:
	api_key (Optional[str]): The API key for authentication. If not provided,
	the default is taken from the environment variable JULEP_API_KEY.
	base_url (Optional[str]): The base URL for the API endpoints. If not provided,
	the default is taken from the environment variable JULEP_API_URL.
	*args: Variable length argument list.
	**kwargs: Arbitrary keyword arguments.
	Raises:
	AssertionError: If either `api_key` or `base_url` is None, indicating they have
	not been provided and are not set as environment variables.
	Note:
	This constructor populates the client with different resource managers like
	agents, users, sessions, docs, memories, and tools, by initializing respective
	manager classes with the provided API client configuration.
	"""
	assert (
	api_key is not None
	), "api_key must be provided or set as env var JULEP_API_KEY"
	assert (
	base_url is not None
	), "base_url must be provided or set as env var JULEP_API_URL"
	# Create an httpz client that follows redirects and has a timeout
	httpx_client = _httpx_client or httpx.Client(
	timeout=timeout,
	headers=additional_headers,
	follow_redirects=True,
	)
	self._api_client = JulepApi(
	api_key=f"Bearer {api_key}",
	base_url=base_url,
	httpx_client=httpx_client,
	*args,
	**kwargs,
	)
	self.agents = AgentsManager(api_client=self._api_client)
	self.users = UsersManager(api_client=self._api_client)
	self.sessions = SessionsManager(api_client=self._api_client)
	self.docs = DocsManager(api_client=self._api_client)
	self.memories = MemoriesManager(api_client=self._api_client)
	self.tools = ToolsManager(api_client=self._api_client)
	# Set up the OpenAI client
	openai_base_url = f"{get_base_url(base_url)}/v1"
	self._openai_client = OpenAI(
	api_key=api_key,
	base_url=openai_base_url,
	*args,
	**kwargs,
	)
	# Patch the OpenAI client to pass non-openai params
	patch_chat_create(self._openai_client)
	patch_completions_create(self._openai_client)
	self.chat = self._openai_client.chat
	self.completions = self._openai_client.completions
	class AsyncClient:
	"""
	A class representing an asynchronous client for interacting with various managers.
	This class initializes asynchronous managers for agents, users, sessions, documents, memories,
	and tools. It requires an API key and a base URL to establish a connection with the backend
	service. If these are not explicitly provided, it looks for them in the environment variables.
	Attributes:
	agents (AsyncAgentsManager): Manager for handling agent-related interactions.
	users (AsyncUsersManager): Manager for handling user-related interactions.
	sessions (AsyncSessionsManager): Manager for handling session-related interactions.
	docs (AsyncDocsManager): Manager for handling document-related interactions.
	memories (AsyncMemoriesManager): Manager for handling memory-related interactions.
	tools (AsyncToolsManager): Manager for handling tool-related interactions.
	chat (AsyncChat): A chat manager instance for handling chat interactions (based on OpenAI client).
	completions (AsyncCompletions): A manager instance for handling completions (based on OpenAI client).
	Raises:
	AssertionError: If `api_key` or `base_url` is not provided and also not set as an
	environment variable.
	Note:
	The `api_key` and `base_url` can either be passed explicitly or set as environment
	variables `JULEP_API_KEY` and `JULEP_API_URL`, respectively.
	Args:
	api_key (Optional[str]): The API key required to authenticate with the service.
	Defaults to the value of the `JULEP_API_KEY` environment variable.
	base_url (Optional[str]): The base URL of the API service.
	Defaults to the value of the `JULEP_API_URL` environment variable.
	*args: Variable length argument list.
	**kwargs: Arbitrary keyword arguments.
	"""
	agents: AsyncAgentsManager
	users: AsyncUsersManager
	sessions: AsyncSessionsManager
	docs: AsyncDocsManager
	memories: AsyncMemoriesManager
	tools: AsyncToolsManager
	chat: AsyncChat
	completions: AsyncCompletions
	@beartype
	def __init__(
	self,
	api_key: Optional[str] = JULEP_API_KEY,
	base_url: Optional[str] = JULEP_API_URL,
	*args,
	**kwargs,
	):
	"""
	Initialize the client with the provided API key and base URL.
	This constructor sets up an asynchronous client for various service managers
	like agents, users, sessions, documents, memories, and tools.
	The 'beartype' decorator is used for runtime type checking on the given arguments.
	The `api_key` and `base_url` must either be provided as arguments or set as
	environment variables (`JULEP_API_KEY`, `JULEP_API_URL` respectively).
	Args:
	api_key (Optional[str]): The API key for authentication. If not provided,
	it should be set as the environment variable 'JULEP_API_KEY'.
	base_url (Optional[str]): The base URL of the API. If not provided,
	it should be set as the environment variable 'JULEP_API_URL'.
	*args: Variable length argument list to be passed to the `AsyncJulepApi`.
	**kwargs: Arbitrary keyword arguments to be passed to the `AsyncJulepApi`.
	Raises:
	AssertionError: If `api_key` or `base_url` is not supplied either
	through function arguments or environment variables.
	Note:
	The actual function signature does not explicitly show the environment variables
	(`JULEP_API_KEY`, `JULEP_API_URL`) as default values for `api_key` and `base_url`
	due to the limitations in using environment variables as default function argument values.
	Instead, they must be handled within the body of the function.
	"""
	assert (
	api_key is not None
	), "api_key must be provided or set as env var JULEP_API_KEY"
	assert (
	base_url is not None
	), "base_url must be provided or set as env var JULEP_API_URL"
	self._api_client = AsyncJulepApi(
	api_key=f"Bearer {api_key}", base_url=base_url, *args, **kwargs
	)
	self.agents = AsyncAgentsManager(api_client=self._api_client)
	self.users = AsyncUsersManager(api_client=self._api_client)
	self.sessions = AsyncSessionsManager(api_client=self._api_client)
	self.docs = AsyncDocsManager(api_client=self._api_client)
	self.memories = AsyncMemoriesManager(api_client=self._api_client)
	self.tools = AsyncToolsManager(api_client=self._api_client)
	# Set up the OpenAI client
	openai_base_url = f"{get_base_url(base_url)}/v1"
	self._openai_client = AsyncOpenAI(
	api_key=api_key,
	base_url=openai_base_url,
	*args,
	**kwargs,
	)
	# Patch the OpenAI client to pass non-openai params
	patch_chat_acreate(self._openai_client)
	patch_completions_acreate(self._openai_client)

julep/sdks/python/julep/client.py

Lines 267 to 268 in 26e57ac

	self.chat = self._openai_client.chat

julep/sdks/python/julep/client.py

Lines 267 to 268 in 26e57ac

	self.chat = self._openai_client.chat

julep/sdks/python/julep/client.py

Lines 267 to 268 in 26e57ac

	self.chat = self._openai_client.chat

julep/sdks/python/julep/managers/base.py

Lines 1 to 28 in 26e57ac

	from typing import Union
	from ..client import JulepApi, AsyncJulepApi
	# Purpose: Base class for all managers
	class BaseManager:
	"""
	A class that serves as a base manager for working with different API clients.
	Attributes:
	api_client (Union[JulepApi, AsyncJulepApi]): A client instance for communicating with an API.
	It can either be an instance of JulepApi for synchronous operations or
	AsyncJulepApi for asynchronous operations.
	Args:
	api_client (Union[JulepApi, AsyncJulepApi]): The API client that is used for making API calls.
	"""
	api_client: Union[JulepApi, AsyncJulepApi]
	def __init__(self, api_client: Union[JulepApi, AsyncJulepApi]):
	"""
	Constructor for the class that initializes it with an API client.
	Args:
	api_client (Union[JulepApi, AsyncJulepApi]): An instance of JulepApi or AsyncJulepApi to be used by this class.
	"""

julep/sdks/python/julep/managers/tool.py

Lines 1 to 496 in 26e57ac

	from uuid import UUID
	from typing import Optional
	from beartype import beartype
	from beartype.typing import Awaitable, List, Union
	from ..api.types import (
	ResourceCreatedResponse,
	FunctionDef,
	GetAgentToolsResponse,
	CreateToolRequest,
	ResourceUpdatedResponse,
	Tool,
	)
	from .base import BaseManager
	from .utils import is_valid_uuid4
	from .types import ToolDict, FunctionDefDict
	class BaseToolsManager(BaseManager):
	"""
	A class to manage base tools by interacting with an API client.
	This class provides an interface for creating, reading, updating, and deleting tools, where each tool is associated with an agent.
	Attributes:
	api_client: The API client utilized to send requests to the service.
	Methods:
	_get(self, agent_id: Union[str, UUID], limit: Optional[int]=None, offset: Optional[int]=None) -> Union[GetAgentToolsResponse, Awaitable[GetAgentToolsResponse]]:
	Retrieve a list of tools associated with the given agent using the API client.
	_create(self, agent_id: Union[str, UUID], tool: ToolDict) -> Union[ResourceCreatedResponse, Awaitable[ResourceCreatedResponse]]:
	Create a new tool associated with the given agent using the API client.
	_update(self, agent_id: Union[str, UUID], tool_id: Union[str, UUID], function: FunctionDefDict) -> Union[ResourceUpdatedResponse, Awaitable[ResourceUpdatedResponse]]:
	Update the definition of an existing tool associated with the given agent using the API client.
	_delete(self, agent_id: Union[str, UUID], tool_id: Union[str, UUID]):
	Delete a tool associated with the given agent using the API client.
	Note:
	All methods assert the validity of UUIDs for agent_id and tool_id when applicable.
	The _get, _create, and _update methods may return either synchronous or asynchronous responses, indicated by the return type Union[..., Awaitable[...]].
	"""
	def _get(
	self,
	agent_id: Union[str, UUID],
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	) -> Union[GetAgentToolsResponse, Awaitable[GetAgentToolsResponse]]:
	"""
	Retrieve tools associated with the given agent.
	This is a private method that fetches tools for the provided agent ID, with optional
	limit and offset parameters for pagination.
	Args:
	agent_id (Union[str, UUID]): The unique identifier for the agent, which can be a
	string or UUID object.
	limit (Optional[int], optional): The maximum number of tools to retrieve. Defaults to None.
	offset (Optional[int], optional): The number of tools to skip before starting to collect
	the result set. Defaults to None.
	Returns:
	Union[GetAgentToolsResponse, Awaitable[GetAgentToolsResponse]]: The response object which
	contains the list of tools, or an awaitable response object for asynchronous operations.
	Raises:
	AssertionError: If the agent_id is not a valid UUID v4.
	"""
	assert is_valid_uuid4(agent_id), "agent_id must be a valid UUID v4"
	return self.api_client.get_agent_tools(
	agent_id=agent_id, limit=limit, offset=offset
	)
	def _create(
	self,
	agent_id: Union[str, UUID],
	tool: ToolDict,
	) -> Union[ResourceCreatedResponse, Awaitable[ResourceCreatedResponse]]:
	"""
	Create a new tool associated with a given agent.
	Args:
	agent_id (Union[str, UUID]): The ID of the agent for which to create the tool. Must be a valid UUID v4.
	tool (ToolDict): A dictionary with tool data conforming to the CreateToolRequest structure.
	Returns:
	Union[ResourceCreatedResponse, Awaitable[ResourceCreatedResponse]]: The response object indicating the newly created tool,
	either as a direct response or as an awaitable if it's an asynchronous operation.
	"""
	assert is_valid_uuid4(agent_id), "agent_id must be a valid UUID v4"
	tool: CreateToolRequest = CreateToolRequest(**tool)
	return self.api_client.create_agent_tool(
	agent_id=agent_id,
	request=tool,
	)
	def _update(
	self,
	agent_id: Union[str, UUID],
	tool_id: Union[str, UUID],
	function: FunctionDefDict,
	overwrite: bool = False,
	) -> Union[ResourceUpdatedResponse, Awaitable[ResourceUpdatedResponse]]:
	"""
	Update the tool definition for a given agent.
	Args:
	agent_id (Union[str, UUID]): The unique identifier for the agent, either in string or UUID format.
	tool_id (Union[str, UUID]): The unique identifier for the tool, either in string or UUID format.
	function (FunctionDefDict): A dictionary containing the function definition that conforms with the required API schema.
	overwrite (bool): A flag to indicate whether to overwrite the existing function definition. Defaults to False.
	Returns:
	Union[ResourceUpdatedResponse, Awaitable[ResourceUpdatedResponse]]: The updated resource response sync or async.
	Raises:
	AssertionError: If the `agent_id` or `tool_id` is not a valid UUID v4.
	"""
	assert is_valid_uuid4(agent_id) and is_valid_uuid4(
	tool_id
	), "agent_id and tool_id must be a valid UUID v4"
	function: FunctionDef = FunctionDef(**function)
	updateFn = (
	self.api_client.update_agent_tool
	if overwrite
	else self.api_client.patch_agent_tool
	)
	return updateFn(
	agent_id=agent_id,
	tool_id=tool_id,
	function=function,
	)
	def _delete(
	self,
	agent_id: Union[str, UUID],
	tool_id: Union[str, UUID],
	):
	"""
	Delete a tool associated with an agent.
	Args:
	agent_id (Union[str, UUID]): The UUID of the agent.
	tool_id (Union[str, UUID]): The UUID of the tool to be deleted.
	Returns:
	The response from the API client's delete_agent_tool method.
	Raises:
	AssertionError: If either `agent_id` or `tool_id` is not a valid UUID v4.
	"""
	assert is_valid_uuid4(agent_id) and is_valid_uuid4(
	tool_id
	), "agent_id and tool_id must be a valid UUID v4"
	return self.api_client.delete_agent_tool(
	agent_id=agent_id,
	tool_id=tool_id,
	)
	class ToolsManager(BaseToolsManager):
	"""
	A manager class for handling tools related to agents.
	This class provides an interface to manage tools, including their retrieval, creation,
	deletion, and updating. It extends the functionalities of the BaseToolsManager.
	Methods:
	get:
	Retrieves a list of tools for the given agent.
	Args:
	agent_id (Union[str, UUID]): The identifier of the agent whose tools are being retrieved.
	limit (Optional[int], optional): The maximum number of tools to retrieve.
	offset (Optional[int], optional): The index from which to start the retrieval.
	Returns:
	List[Tool]: A list of Tool objects.
	create:
	Creates a new tool for the given agent.
	Args:
	agent_id (Union[str, UUID]): The identifier of the agent for whom the tool is being created.
	tool (ToolDict): A dictionary of tool attributes.
	Returns:
	ResourceCreatedResponse: An object indicating the successful creation of the tool.
	delete:
	Deletes a tool for the given agent.
	Args:
	agent_id (Union[str, UUID]): The identifier of the agent whose tool is being deleted.
	tool_id (Union[str, UUID]): The unique identifier of the tool to be deleted.
	update:
	Updates the definition of an existing tool for the given agent.
	Args:
	agent_id (Union[str, UUID]): The identifier of the agent whose tool is being updated.
	tool_id (Union[str, UUID]): The unique identifier of the tool to be updated.
	function (FunctionDefDict): A dictionary representing the definition of the tool to be updated.
	Returns:
	ResourceUpdatedResponse: An object indicating the successful update of the tool.
	Inherits:
	BaseToolsManager: A base class that defines the basic operations for tool management.
	"""
	@beartype
	def get(
	self,
	*,
	agent_id: Union[str, UUID],
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	) -> List[Tool]:
	"""
	Retrieve a list of Tool objects for the specified agent.
	This method wraps the internal _get method and returns the 'items' property
	from the result, which contains a list of Tool instances.
	Args:
	agent_id (Union[str, UUID]): The ID of the agent to retrieve Tool objects for.
	limit (Optional[int]): The maximum number of Tool objects to return. Defaults to None, implying no limit.
	offset (Optional[int]): The number to skip before starting to collect the result set. Defaults to None, implying no offset.
	Returns:
	List[Tool]: A list of Tool instances associated with the specified agent ID.
	Raises:
	BeartypeException: If the input argument types do not match the expected types.
	"""
	return self._get(
	agent_id=agent_id,
	limit=limit,
	offset=offset,
	).items
	@beartype
	def create(
	self,
	*,
	agent_id: Union[str, UUID],
	tool: ToolDict,
	) -> ResourceCreatedResponse:
	"""
	Create a new resource with the provided agent identifier and tool information.
	This method wraps the internal `_create` method to construct and return a ResourceCreatedResponse.
	Args:
	agent_id (Union[str, UUID]): The unique identifier for the agent. Can be a string or a UUID object.
	tool (ToolDict): A dictionary containing tool-specific configuration or information.
	Returns:
	ResourceCreatedResponse: An object representing the successfully created resource, including metadata like creation timestamps and resource identifiers.
	Raises:
	TypeError: If the `agent_id` or `tool` arguments are not of the expected type.
	ValueError: If any values within the `tool` dictionary are invalid or out of accepted range.
	RuntimeError: If the creation process encounters an unexpected error.
	Note:
	The `@beartype` decorator is used to enforce type checking of the arguments at runtime.
	Example usage:
	>>> response = instance.create(agent_id="123e4567-e89b-12d3-a456-426614174000", tool={"type": "screwdriver", "size": "M4"})
	>>> print(response)
	ResourceCreatedResponse(resource_id='abcde-12345', created_at='2021-01-01T12:00:00Z')
	"""
	return self._create(
	agent_id=agent_id,
	tool=tool,
	)
	@beartype
	def delete(
	self,
	*,
	agent_id: Union[str, UUID],
	tool_id: Union[str, UUID],
	):
	"""
	Deletes an agent's access to a specific tool.
	Args:
	agent_id (Union[str, UUID]): The unique identifier of the agent.
	tool_id (Union[str, UUID]): The unique identifier of the tool.
	Returns:
	The result of the delete operation, the specific type of which may depend on the implementation of `_delete`.
	Raises:
	Beartype exceptions: If `agent_id` or `tool_id` are of the wrong type.
	"""
	return self._delete(
	agent_id=agent_id,
	tool_id=tool_id,
	)
	@beartype
	def update(
	self,
	*,
	agent_id: Union[str, UUID],
	tool_id: Union[str, UUID],
	function: FunctionDefDict,
	overwrite: bool = False,
	) -> ResourceUpdatedResponse:
	"""
	Update a specific tool definition for an agent.
	Args:
	agent_id (Union[str, UUID]): The unique identifier of the agent.
	tool_id (Union[str, UUID]): The unique identifier of the tool to be updated.
	function (FunctionDefDict): A dictionary containing the new definition of the tool.
	overwrite (bool): A flag indicating whether to overwrite the existing definition.
	Returns:
	ResourceUpdatedResponse: An object representing the update operation response.
	Note:
	This function is decorated with `beartype` which ensures that the arguments provided
	match the expected types at runtime.
	Raises:
	BeartypeDecorHintPepParamException: If the type of any parameter does not match the expected type.
	Any exceptions that `self._update` method might raise.
	"""
	return self._update(
	agent_id=agent_id,
	tool_id=tool_id,
	function=function,
	overwrite=overwrite,
	)
	class AsyncToolsManager(BaseToolsManager):
	"""
	A manager for asynchronous tools handling.
	This class provides async methods to manage tools, allowing create, retrieve, update, and delete operations.
	Methods:
	get: Asynchronously retrieves tools associated with an agent.
	create: Asynchronously creates a new tool associated with an agent.
	delete: Asynchronously deletes a tool associated with an agent.
	update: Asynchronously updates a tool associated with an agent.
	Attributes:
	Inherited from BaseToolsManager.
	"""
	@beartype
	async def get(
	self,
	*,
	agent_id: Union[str, UUID],
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	) -> List[Tool]:
	"""
	Asynchronously get a list of Tool objects based on provided filters.
	This method fetches Tool objects with the option to limit the results and
	offset them, to allow for pagination.
	Args:
	agent_id (Union[str, UUID]): The unique identifier of the agent for which to fetch tools.
	limit (Optional[int], optional): The maximum number of tools to return. Defaults to None, which implies no limit.
	offset (Optional[int], optional): The number of tools to skip before starting to return the tools. Defaults to None, which means start from the beginning.
	Returns:
	List[Tool]: A list of Tool objects that match the criteria.
	Note:
	This function is decorated with beartype, which assures that the input
	arguments adhere to the specified types at runtime. It is an asynchronous
	function and should be called with `await`.
	"""
	return (
	await self._get(
	agent_id=agent_id,
	limit=limit,
	offset=offset,
	)
	).items
	@beartype
	async def create(
	self,
	*,
	agent_id: Union[str, UUID],
	tool: ToolDict,
	) -> ResourceCreatedResponse:
	"""
	Create a new resource asynchronously.
	This method creates a new resource using the provided `agent_id` and `tool` information.
	Args:
	agent_id (Union[str, UUID]): The identifier of the agent, which can be a string or a UUID object.
	tool (ToolDict): A dictionary containing tool-specific data.
	Returns:
	ResourceCreatedResponse: An object representing the response received after creating the resource.
	Raises:
	BeartypeException: If the type of any argument does not match the expected type.
	"""
	return await self._create(
	agent_id=agent_id,
	tool=tool,
	)
	@beartype
	async def delete(
	self,
	*,
	agent_id: Union[str, UUID],
	tool_id: Union[str, UUID],
	):
	"""
	Asynchronously delete a specified agent-tool association.
	This function is a high-level asynchronous API that delegates to a
	private coroutinal method `_delete` to perform the actual deletion based on
	the provided `agent_id` and `tool_id`.
	Args:
	agent_id (Union[str, UUID]): The unique identifier of the agent.
	tool_id (Union[str, UUID]): The unique identifier of the tool.
	Returns:
	The return type is not explicitly stated in the function signature.
	Typically, the returned value would be documented here, but you may need
	to investigate the implementation of `_delete` to determine what it
	returns.
	Raises:
	The exceptions that this function might raise are not stated in the
	snippet provided. If the private method `_delete` raises any exceptions,
	they should be documented here. Depending on the implementation, common
	exceptions might include `ValueError` if identifiers are invalid or
	`RuntimeError` if deletion fails.
	"""
	return await self._delete(
	agent_id=agent_id,
	tool_id=tool_id,
	)
	@beartype
	async def update(
	self,
	*,
	agent_id: Union[str, UUID],
	tool_id: Union[str, UUID],
	function: FunctionDefDict,
	) -> ResourceUpdatedResponse:
	"""
	Asynchronously updates a resource identified by the agent_id and tool_id with a new definition.
	This function is type-enforced using the `beartype` decorator.
	Args:
	agent_id (Union[str, UUID]): The unique identifier for the agent.
	tool_id (Union[str, UUID]): The unique identifier for the tool.
	function (FunctionDefDict): A dictionary containing the function definition which needs to be updated.
	Returns:
	ResourceUpdatedResponse: An object representing the response received after updating the resource.
	Raises:
	This will depend on the implementation of the `_update` method and any exceptions that it raises.
	"""
	return await self._update(
	agent_id=agent_id,
	tool_id=tool_id,
	function=function,

julep/sdks/python/julep/managers/session.py

Lines 1 to 800 in 26e57ac

	import json
	from typing import Optional, TypedDict
	from uuid import UUID
	from beartype import beartype
	from beartype.typing import Any, Awaitable, Dict, List, Union
	from ..api.types import (
	ResourceCreatedResponse,
	ResourceUpdatedResponse,
	Session,
	ListSessionsResponse,
	ChatSettingsStop,
	ChatSettingsResponseFormat,
	InputChatMlMessage,
	ChatMlMessage,
	ToolChoiceOption,
	Tool,
	ChatResponse,
	GetSuggestionsResponse,
	GetHistoryResponse,
	Suggestion,
	)
	from .utils import rewrap_in_class
	from .base import BaseManager
	from .types import (
	ChatSettingsResponseFormatDict,
	InputChatMlMessageDict,
	ToolDict,
	)
	from .utils import is_valid_uuid4
	class SessionCreateArgs(TypedDict):
	user_id: Optional[Union[str, UUID]]
	agent_id: Union[str, UUID]
	situation: Optional[str] = None
	metadata: Dict[str, Any] = {}
	class SessionUpdateArgs(TypedDict):
	session_id: Union[str, UUID]
	situation: Optional[str] = None
	metadata: Optional[Dict[str, Any]] = None
	overwrite: bool = False
	class BaseSessionsManager(BaseManager):
	"""
	A class to manage sessions using base API client methods.
	This manager handles CRUD operations and additional actions on the session data,
	such as chatting and retrieving history or suggestions.
	Attributes:
	api_client: The client used for communicating with an API.
	Methods:
	_get(id):
	Retrieve a specific session by its identifier.
	Args:
	id (Union[str, UUID]): The unique identifier for the session.
	Returns:
	Union[Session, Awaitable[Session]]: The session object or an awaitable yielding it.
	_create(user_id, agent_id, situation):
	Create a new session with specified user and agent identifiers.
	Args:
	agent_id (Union[str, UUID]): The unique identifier for the agent.
	user_id (Optional[Union[str, UUID]]): The unique identifier for the user.
	situation (Optional[str]): An optional description of the situation for the session.
	Returns:
	Union[ResourceCreatedResponse, Awaitable[ResourceCreatedResponse]]: The response for the created session or an awaitable yielding it.
	_list_items(limit, offset):
	List multiple session items with optional pagination.
	Args:
	limit (Optional[int]): The limit on the number of items to be retrieved.
	offset (Optional[int]): The number of items to be skipped before starting to collect the result set.
	Returns:
	Union[ListSessionsResponse, Awaitable[ListSessionsResponse]]: The list of sessions or an awaitable yielding it.
	_delete(session_id):
	Delete a session by its identifier.
	Args:
	session_id (Union[str, UUID]): The unique identifier for the session to be deleted.
	Returns:
	Union[None, Awaitable[None]]: None or an awaitable yielding None if the operation is successful.
	_update(session_id, situation):
	Update the situation for an existing session.
	Args:
	session_id (Union[str, UUID]): The unique identifier for the session to be updated.
	situation (str): The new situation description for the session.
	Returns:
	Union[ResourceUpdatedResponse, Awaitable[ResourceUpdatedResponse]]: The response for the updated session or an awaitable yielding it.
	_chat(session_id, messages, ...):
	Send chat messages and get responses during a session.
	Args:
	session_id (str): The unique identifier for the session.
	messages (List[InputChatMlMessage]): The list of input chat messages to be sent.
	tools (Optional[List[Tool]]): ...
	tool_choice (Optional[ToolChoiceOption]): ...
	...: Other optional parameters for chat settings and modifiers.
	Returns:
	Union[ChatResponse, Awaitable[ChatResponse]]: The chat response for the session or an awaitable yielding it.
	_suggestions(session_id, limit, offset):
	Get suggestions for a session.
	Args:
	session_id (Union[str, UUID]): The unique identifier for the session.
	limit (Optional[int]): The limit on the number of suggestions to be retrieved.
	offset (Optional[int]): The number of suggestions to be skipped before starting to collect the result set.
	Returns:
	Union[GetSuggestionsResponse, Awaitable[GetSuggestionsResponse]]: The suggestions response for the session or an awaitable yielding it.
	_history(session_id, limit, offset):
	Get the history for a session.
	Args:
	session_id (Union[str, UUID]): The unique identifier for the session.
	limit (Optional[int]): The limit on the number of history entries to be retrieved.
	offset (Optional[int]): The number of history entries to be skipped before starting to collect the result set.
	Returns:
	Union[GetHistoryResponse, Awaitable[GetHistoryResponse]]: The history response for the session or an awaitable yielding it.
	_delete_history(session_id):
	Delete the history of a session.
	Args:
	session_id (Union[str, UUID]): The unique identifier for the session.
	Returns:
	Union[None, Awaitable[None]]: None or an awaitable yielding None if the operation is successful.
	"""
	def _get(self, id: Union[str, UUID]) -> Union[Session, Awaitable[Session]]:
	"""
	Get a session by its ID.
	Args:
	id (Union[str, UUID]): A string or UUID representing the session ID.
	Returns:
	Union[Session, Awaitable[Session]]: The session object associated with the given ID, which can be either a `Session` instance or an `Awaitable` that resolves to a `Session`.
	Raises:
	AssertionError: If the id is not a valid UUID v4.
	"""
	assert is_valid_uuid4(id), "id must be a valid UUID v4"
	return self.api_client.get_session(session_id=id)
	def _create(
	self,
	agent_id: Union[str, UUID],
	user_id: Optional[Union[str, UUID]] = None,
	situation: Optional[str] = None,
	metadata: Dict[str, Any] = {},
	) -> Union[ResourceCreatedResponse, Awaitable[ResourceCreatedResponse]]:
	# Cast instructions to a list of Instruction objects
	"""
	Creates a session for a specified user and agent.
	This internal method is responsible for creating a session using the API client. It validates that both the user and agent IDs are valid UUID v4 strings before proceeding with session creation.
	Args:
	agent_id (Union[str, UUID]): The agent's identifier which could be a string or a UUID object.
	user_id (Optional[Union[str, UUID]]): The user's identifier which could be a string or a UUID object.
	situation (Optional[str], optional): An optional description of the situation.
	metadata (Dict[str, Any])
	Returns:
	Union[ResourceCreatedResponse, Awaitable[ResourceCreatedResponse]]: The response from the API client upon successful session creation, which can be a synchronous `ResourceCreatedResponse` or an asynchronous `Awaitable` of it.
	Raises:
	AssertionError: If either `user_id` or `agent_id` is not a valid UUID v4.
	"""
	assert is_valid_uuid4(agent_id), "agent_id must be a valid UUID v4"
	if user_id is not None:
	assert is_valid_uuid4(user_id), "user_id must be a valid UUID v4"
	return self.api_client.create_session(
	user_id=user_id,
	agent_id=agent_id,
	situation=situation,
	metadata=metadata,
	)
	def _list_items(
	self,
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	metadata_filter: str = "{}",
	) -> Union[ListSessionsResponse, Awaitable[ListSessionsResponse]]:
	"""
	List items with optional pagination.
	Args:
	limit (Optional[int]): The maximum number of items to return. Defaults to None.
	offset (Optional[int]): The number of items to skip before starting to collect the result set. Defaults to None.
	Returns:
	Union[ListSessionsResponse, Awaitable[ListSessionsResponse]]: The response object containing the list of items or an awaitable response object if called asynchronously.
	Note:
	The '_list_items' function is assumed to be a method of a class that has an 'api_client' attribute capable of listing sessions.
	"""
	return self.api_client.list_sessions(
	limit=limit,
	offset=offset,
	metadata_filter=metadata_filter,
	)
	def _delete(self, session_id: Union[str, UUID]) -> Union[None, Awaitable[None]]:
	"""
	Delete a session given its session ID.
	This is an internal method that asserts the provided session_id is a valid UUID v4
	before making the delete request through the API client.
	Args:
	session_id (Union[str, UUID]): The session identifier, which should be a valid UUID v4.
	Returns:
	Union[None, Awaitable[None]]: The result of the delete operation, which can be either
	None or an Awaitable that resolves to None, depending on whether this is a synchronous
	or asynchronous call.
	Raises:
	AssertionError: If the `session_id` is not a valid UUID v4.
	"""
	assert is_valid_uuid4(session_id), "id must be a valid UUID v4"
	return self.api_client.delete_session(session_id=session_id)
	def _update(
	self,
	session_id: Union[str, UUID],
	situation: Optional[str] = None,
	metadata: Optional[Dict[str, Any]] = None,
	overwrite: bool = False,
	) -> Union[ResourceUpdatedResponse, Awaitable[ResourceUpdatedResponse]]:
	"""
	Update a session with a given situation.
	Args:
	session_id (Union[str, UUID]): The session identifier, which can be a string-formatted UUID or an actual UUID object.
	situation (str): A string describing the current situation.
	overwrite (bool, optional): Whether to overwrite the existing situation. Defaults to False.
	Returns:
	Union[ResourceUpdatedResponse, Awaitable[ResourceUpdatedResponse]]: The response from the update operation, which can be either synchronous or asynchronous.
	Raises:
	AssertionError: If `session_id` is not a valid UUID v4.
	"""
	assert is_valid_uuid4(session_id), "id must be a valid UUID v4"
	updateFn = (
	self.api_client.update_session
	if overwrite
	else self.api_client.patch_session
	)
	return updateFn(
	session_id=session_id,
	situation=situation,
	metadata=metadata,
	)
	def _chat(
	self,
	*,
	session_id: str,
	messages: List[Union[InputChatMlMessageDict, InputChatMlMessage]],
	tools: Optional[List[Union[ToolDict, Tool]]] = None,
	tool_choice: Optional[ToolChoiceOption] = None,
	frequency_penalty: Optional[float] = None,
	length_penalty: Optional[float] = None,
	logit_bias: Optional[Dict[str, Optional[int]]] = None,
	max_tokens: Optional[int] = None,
	presence_penalty: Optional[float] = None,
	repetition_penalty: Optional[float] = None,
	response_format: Optional[
	Union[ChatSettingsResponseFormatDict, ChatSettingsResponseFormat]
	] = None,
	seed: Optional[int] = None,
	stop: Optional[ChatSettingsStop] = None,
	stream: Optional[bool] = None,
	temperature: Optional[float] = None,
	top_p: Optional[float] = None,
	recall: Optional[bool] = None,
	remember: Optional[bool] = None,
	) -> Union[ChatResponse, Awaitable[ChatResponse]]:
	"""
	Conducts a chat conversation with an AI model using specific parameters.
	Args:
	session_id (str): A unique identifier for the chat session.
	messages (List[InputChatMlMessage]): A list of input messages for the AI to respond to.
	tools (Optional[List[Tool]]): A list of tools to be used during the chat session.
	tool_choice (Optional[ToolChoiceOption]): A method for choosing which tools to apply.
	frequency_penalty (Optional[float]): A modifier to decrease the likelihood of frequency-based repetitions.
	length_penalty (Optional[float]): A modifier to control the length of the generated responses.
	logit_bias (Optional[Dict[str, Optional[int]]]): Adjustments to the likelihood of specific tokens appearing.
	max_tokens (Optional[int]): The maximum number of tokens to generate in the output.
	presence_penalty (Optional[float]): A modifier to control for new concepts' appearance.
	repetition_penalty (Optional[float]): A modifier to discourage repetitive responses.
	response_format (Optional[ChatSettingsResponseFormat]): The format in which the response is to be delivered.
	seed (Optional[int]): An integer to seed the random number generator for reproducibility.
	stop (Optional[ChatSettingsStop]): Tokens at which to stop generating further tokens.
	stream (Optional[bool]): Whether to stream the response or deliver it when it's complete.
	temperature (Optional[float]): A value to control the randomness of the output.
	top_p (Optional[float]): A value to control the nucleus sampling, i.e., the cumulative probability cutoff.
	recall (Optional[bool]): A flag to control the recall capability of the AI model.
	remember (Optional[bool]): A flag to control the persistence of the chat history in the AI's memory.
	Returns:
	Union[ChatResponse, Awaitable[ChatResponse]]: The response from the AI given the input messages and parameters. This could be a synchronous `ChatResponse` object or an asynchronous `Awaitable[ChatResponse]` if the `stream` parameter is True.
	Note:
	The precise types of some arguments, like `Tool`, `ToolChoiceOption`, `ChatSettingsResponseFormat`, and `ChatSettingsStop`, are not defined within the given context. It's assumed that these types have been defined elsewhere in the code base.
	Raises:
	It is not specified what exceptions this function might raise. Typically, one would expect potential exceptions to be associated with the underlying API client's `chat` method failure modes, such as network issues, invalid parameters, etc.
	"""
	return self.api_client.chat(
	session_id=session_id,
	messages=messages,
	tools=tools,
	tool_choice=tool_choice,
	frequency_penalty=frequency_penalty,
	length_penalty=length_penalty,
	logit_bias=logit_bias,
	max_tokens=max_tokens,
	presence_penalty=presence_penalty,
	repetition_penalty=repetition_penalty,
	response_format=response_format,
	seed=seed,
	stop=stop,
	stream=stream,
	temperature=temperature,
	top_p=top_p,
	recall=recall,
	remember=remember,
	)
	def _suggestions(
	self,
	session_id: Union[str, UUID],
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	) -> Union[GetSuggestionsResponse, Awaitable[GetSuggestionsResponse]]:
	"""
	Retrieve a list of suggestions for a given session.
	Args:
	session_id (Union[str, UUID]): The ID of the session for which to get suggestions.
	limit (Optional[int], optional): The maximum number of suggestions to retrieve. Defaults to None.
	offset (Optional[int], optional): The offset from where to start retrieving suggestions. Defaults to None.
	Returns:
	Union[GetSuggestionsResponse, Awaitable[GetSuggestionsResponse]]: The response containing the list of suggestions synchronously or asynchronously, depending on the API client.
	"""
	return self.api_client.get_suggestions(
	session_id=session_id,
	limit=limit,
	offset=offset,
	)
	def _history(
	self,
	session_id: Union[str, UUID],
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	) -> Union[GetHistoryResponse, Awaitable[GetHistoryResponse]]:
	"""
	Retrieve a session's history with optional pagination controls.
	Args:
	session_id (Union[str, UUID]): Unique identifier for the session
	whose history is being queried. Can be a string or a UUID object.
	limit (Optional[int], optional): The maximum number of history
	entries to retrieve. Defaults to None, which uses the API's default setting.
	offset (Optional[int], optional): The number of initial history
	entries to skip. Defaults to None, which means no offset is applied.
	Returns:
	Union[GetHistoryResponse, Awaitable[GetHistoryResponse]]:
	The history response object, which may be either synchronous or
	asynchronous (awaitable), depending on the API client configuration.
	"""
	return self.api_client.get_history(
	session_id=session_id,
	limit=limit,
	offset=offset,
	)
	def _delete_history(
	self,
	session_id: Union[str, UUID],
	) -> Union[None, Awaitable[None]]:
	"""
	Delete the history of a session.
	Args:
	session_id (Union[str, UUID]): The unique identifier for the session.
	Returns:
	Union[None, Awaitable[None]]: The result of the delete operation, which can be either
	None or an Awaitable that resolves to None, depending on whether this is a synchronous
	or asynchronous call.
	Raises:
	AssertionError: If the `session_id` is not a valid UUID v4.
	"""
	assert is_valid_uuid4(session_id), "id must be a valid UUID v4"
	return self.api_client.delete_session_history(session_id=session_id)
	class SessionsManager(BaseSessionsManager):
	"""
	A class responsible for managing session interactions.
	This class extends `BaseSessionsManager` and provides methods to get, create,
	list, delete, and update sessions, as well as to initiate a chat within a session,
	request suggestions, and access session history.
	Methods:
	get (id: Union[str, UUID]) -> Session:
	Retrieves a session by its identifier.
	create (
	*,
	user_id: Union[str, UUID],
	agent_id: Union[str, UUID],
	situation: Optional[str]=None
	) -> ResourceCreatedResponse:
	Creates a new session given a user ID and an agent ID, and optionally
	a description of the situation.
	list (
	*,
	limit: Optional[int]=None,
	offset: Optional[int]=None
	) -> List[Session]:
	Lists sessions with optional pagination via limit and offset.
	delete (session_id: Union[str, UUID]):
	Deletes a session identified by the given session ID.
	update (
	*,
	session_id: Union[str, UUID],
	situation: str
	) -> ResourceUpdatedResponse:
	Updates the situation of a specific session by its ID.
	chat (
	*args
	see full method signature for detailed arguments
	) -> ChatResponse:
	Initiates a chat in the given session with messages and various settings,
	including tools, penalties, biases, tokens, response format, etc.
	suggestions (
	*,
	session_id: Union[str, UUID],
	limit: Optional[int]=None,
	offset: Optional[int]=None
	) -> List[Suggestion]:
	Retrieves a list of suggestions for a given session, supported by
	optional pagination parameters.
	history (
	*,
	session_id: Union[str, UUID],
	limit: Optional[int]=None,
	offset: Optional[int]=None
	) -> List[ChatMlMessage]:
	Retrieves the chat history for a given session, supported by
	optional pagination parameters.
	delete_history (session_id: Union[str, UUID]) -> None:
	Each method is decorated with `@beartype` for runtime type enforcement.
	"""
	@beartype
	def get(self, id: Union[str, UUID]) -> Session:
	"""
	Retrieve a Session object based on a given identifier.
	Args:
	id (Union[str, UUID]): The identifier of the session, which can be either a string or a UUID.
	Returns:
	Session: The session object associated with the given id.
	Raises:
	TypeError: If the id is neither a string nor a UUID.
	KeyError: If the session with the given id does not exist.
	"""
	return self._get(id=id)
	@beartype
	@rewrap_in_class(Session)
	def create(self, **kwargs: SessionCreateArgs) -> Session:
	"""
	Create a new resource with a user ID and an agent ID, optionally including a situation description.
	Args:
	user_id (Union[str, UUID]): The unique identifier for the user.
	agent_id (Union[str, UUID]): The unique identifier for the agent.
	situation (Optional[str]): An optional description of the situation.
	Returns:
	Session: The created Session object.
	Raises:
	BeartypeException: If the provided `user_id` or `agent_id` do not match the required type.
	Any other exception that `_create` might raise.
	"""
	result = self._create(**kwargs)
	return result
	@beartype
	def list(
	self,
	*,
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	metadata_filter: Dict[str, Any] = {},
	) -> List[Session]:
	"""
	Retrieve a list of Session objects with optional pagination.
	Args:
	limit (Optional[int]): The maximum number of Session objects to return.
	Defaults to None, which indicates no limit.
	offset (Optional[int]): The number of items to skip before starting to return the results.
	Defaults to None, which indicates no offset.
	Returns:
	List[Session]: A list of Session objects meeting the criteria.
	Raises:
	BeartypeException: If the input arguments do not match their annotated types.
	"""
	metadata_filter_string = json.dumps(metadata_filter)
	return self._list_items(
	limit=limit,
	offset=offset,
	metadata_filter=metadata_filter_string,
	).items
	@beartype
	def delete(self, session_id: Union[str, UUID]):
	"""
	Deletes a session based on its session ID.
	Args:
	session_id (Union[str, UUID]): The session ID to be deleted, which can be a string or a UUID object.
	Returns:
	The result from the internal `_delete` method call.
	Raises:
	The specific exceptions that `self._delete` might raise, which should be documented in its docstring.
	"""
	return self._delete(session_id=session_id)
	@beartype
	@rewrap_in_class(Session)
	def update(
	self,
	**kwargs: SessionUpdateArgs,
	) -> Session:
	"""
	Updates the state of a resource based on a given situation.
	This function is type-checked using beartype to ensure that the `session_id` parameter
	is either a string or a UUID and that the `situation` parameter is a string. It delegates
	the actual update process to an internal method '_update'.
	Args:
	session_id (Union[str, UUID]): The session identifier, which can be a UUID or a
	string that uniquely identifies the session.
	situation (str): A string that represents the new situation for the resource update.
	overwrite (bool, optional): A flag to indicate whether to overwrite the existing
	Returns:
	Session: The updated Session object.
	Note:
	The `@beartype` decorator is used for runtime type checking of the function arguments.
	"""
	result = self._update(**kwargs)
	return result
	@beartype
	def chat(
	self,
	*,
	session_id: str,
	messages: List[Union[InputChatMlMessageDict, InputChatMlMessage]],
	tools: Optional[List[Union[ToolDict, Tool]]] = None,
	tool_choice: Optional[ToolChoiceOption] = None,
	frequency_penalty: Optional[float] = None,
	length_penalty: Optional[float] = None,
	logit_bias: Optional[Dict[str, Optional[int]]] = None,
	max_tokens: Optional[int] = None,
	presence_penalty: Optional[float] = None,
	repetition_penalty: Optional[float] = None,
	response_format: Optional[
	Union[ChatSettingsResponseFormatDict, ChatSettingsResponseFormat]
	] = None,
	seed: Optional[int] = None,
	stop: Optional[ChatSettingsStop] = None,
	stream: Optional[bool] = None,
	temperature: Optional[float] = None,
	top_p: Optional[float] = None,
	recall: Optional[bool] = None,
	remember: Optional[bool] = None,
	) -> ChatResponse:
	"""
	Initiate a chat session with the provided inputs and configurations.
	Args:
	session_id (str): Unique identifier for the chat session.
	messages (List[InputChatMlMessage]): List of messages to send in the chat session.
	tools (Optional[List[Tool]], optional): List of tools to be used in the session. Defaults to None.
	tool_choice (Optional[ToolChoiceOption], optional): The choice of tool to optimize response for. Defaults to None.
	frequency_penalty (Optional[float], optional): Penalty for frequent tokens to control repetition. Defaults to None.
	length_penalty (Optional[float], optional): Penalty for longer responses to control verbosity. Defaults to None.
	logit_bias (Optional[Dict[str, Optional[int]]], optional): Bias for or against specific tokens. Defaults to None.
	max_tokens (Optional[int], optional): Maximum number of tokens to generate in the response. Defaults to None.
	presence_penalty (Optional[float], optional): Penalty for new tokens to control topic introduction. Defaults to None.
	repetition_penalty (Optional[float], optional): Penalty to discourage repetition. Defaults to None.
	response_format (Optional[ChatSettingsResponseFormat], optional): Format of the response. Defaults to None.
	seed (Optional[int], optional): Random seed for deterministic responses. Defaults to None.
	stop (Optional[ChatSettingsStop], optional): Sequence at which to stop generating further tokens. Defaults to None.
	stream (Optional[bool], optional): Whether to stream responses or not. Defaults to None.
	temperature (Optional[float], optional): Sampling temperature for randomness in the response. Defaults to None.
	top_p (Optional[float], optional): Nucleus sampling parameter to control diversity. Defaults to None.
	recall (Optional[bool], optional): Whether to allow recalling previous parts of the chat. Defaults to None.
	remember (Optional[bool], optional): Whether to allow the model to remember previous chats. Defaults to None.
	Returns:
	ChatResponse: The response object after processing chat messages.
	Note:
	The 'beartype' decorator is used for runtime type checking.
	"""
	return self._chat(
	session_id=session_id,
	messages=messages,
	tools=tools,
	tool_choice=tool_choice,
	frequency_penalty=frequency_penalty,
	length_penalty=length_penalty,
	logit_bias=logit_bias,
	max_tokens=max_tokens,
	presence_penalty=presence_penalty,
	repetition_penalty=repetition_penalty,
	response_format=response_format,
	seed=seed,
	stop=stop,
	stream=stream,
	temperature=temperature,
	top_p=top_p,
	recall=recall,
	remember=remember,
	)
	@beartype
	def suggestions(
	self,
	*,
	session_id: Union[str, UUID],
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	) -> List[Suggestion]:
	"""
	Provides a list of suggestion objects based on the given session ID.
	This method retrieves suggestions and is decorated with `beartype` for runtime type
	checking of the passed arguments.
	Args:
	session_id (Union[str, UUID]): The ID of the session to retrieve suggestions for.
	limit (Optional[int], optional): The maximum number of suggestions to return.
	Defaults to None, which means no limit.
	offset (Optional[int], optional): The number to offset the list of returned
	suggestions by. Defaults to None, which means no offset.
	Returns:
	List[Suggestion]: A list of suggestion objects.
	Raises:
	Any exceptions that `_suggestions` might raise, for example, if the method is unable
	to retrieve suggestions based on the provided session ID.
	"""
	return self._suggestions(
	session_id=session_id,
	limit=limit,
	offset=offset,
	).items
	@beartype
	def history(
	self,
	*,
	session_id: Union[str, UUID],
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	) -> List[ChatMlMessage]:
	"""
	Retrieve a history of ChatMl messages for a given session.
	This method uses the private method `_history` to fetch the message history and returns a list of ChatMlMessage objects.
	Args:
	session_id (Union[str, UUID]): The session identifier to fetch the chat history for.
	limit (Optional[int], optional): The maximum number of messages to return. If None, no limit is applied. Defaults to None.
	offset (Optional[int], optional): The offset from where to start fetching messages. If None, no offset is applied. Defaults to None.
	Returns:
	List[ChatMlMessage]: A list of ChatMlMessage objects representing the history of messages for the session.
	"""
	return self._history(
	session_id=session_id,
	limit=limit,
	offset=offset,
	).items
	@beartype
	def delete_history(self, session_id: Union[str, UUID]) -> None:
	"""
	Delete the history of a session.
	Args:
	session_id (Union[str, UUID]): The unique identifier for the session.
	Returns:
	None: The result of the delete operation.
	Raises:
	AssertionError: If the `session_id` is not a valid UUID v4.
	"""
	return self._delete_history(session_id=session_id)
	class AsyncSessionsManager(BaseSessionsManager):
	"""
	A class for managing asynchronous sessions.
	This class handles operations related to creating, retrieving, updating,
	deleting, and interacting with sessions asynchronously. It extends the
	functionality of BaseSessionsManager with asynchronous behavior.
	Attributes:
	Inherits attributes from the BaseSessionsManager class.
	Methods:
	async get(id: Union[UUID, str]) -> Session:
	Retrieves a session by its ID.
	async create(*, user_id: Union[str, UUID], agent_id: Union[str, UUID], situation: Optional[str]=None) -> ResourceCreatedResponse:
	Creates a new session with the specified user and agent IDs, and an optional situation.
	async list(*, limit: Optional[int]=None, offset: Optional[int]=None) -> List[Session]:
	Lists sessions with an optional limit and offset for pagination.
	async delete(session_id: Union[str, UUID]):
	Deletes a session by its ID.
	async update(*, session_id: Union[str, UUID], situation: str) -> ResourceUpdatedResponse:
	Updates the situation for a session by its ID.
	async chat(*, session_id: str, messages: List[InputChatMlMessage], tools: Optional[List[Tool]]=None, tool_choice: Optional[ToolChoiceOption]=None, frequency_penalty: Optional[float]=None, length_penalty: Optional[float]=None, logit_bias: Optional[Dict[str, Optional[int]]]=None, max_tokens: Optional[int]=None, presence_penalty: Optional[float]=None, repetition_penalty: Optional[float]=None, response_format: Optional[ChatSettingsResponseFormat]=None, seed: Optional[int]=None, stop: Optional[ChatSettingsStop]=None, stream: Optional[bool]=None, temperature: Optional[float]=None, top_p: Optional[float]=None, recall: Optional[bool]=None, remember: Optional[bool]=None) -> ChatResponse:

julep/sdks/python/julep/managers/user.py

Lines 1 to 507 in 26e57ac

	import json
	from uuid import UUID
	from typing import Optional, TypedDict
	from beartype import beartype
	from beartype.typing import Any, Awaitable, Dict, List, Union
	from .utils import rewrap_in_class, NotSet
	from ..api.types import (
	User,
	CreateDoc,
	ResourceCreatedResponse,
	ResourceUpdatedResponse,
	ListUsersResponse,
	)
	from .base import BaseManager
	from .utils import is_valid_uuid4
	from .types import DocDict
	class UserCreateArgs(TypedDict):
	name: str
	about: str
	docs: List[str] = []
	metadata: Dict[str, Any] = {}
	class UserUpdateArgs(TypedDict):
	name: Optional[str] = None
	metadata: Optional[Dict[str, Any]] = None
	about: Optional[str] = None
	overwrite: bool = False
	class BaseUsersManager(BaseManager):
	"""
	A manager class for handling user-related operations through an API client.
	This class provides high-level methods to interact with user records,
	such as retrieving a user by ID, creating a new user, listing all users,
	deleting a user, and updating a user's details.
	Attributes:
	api_client: The API client instance to communicate with the user service.
	Methods:
	_get: Retrieve a user by a unique UUID.
	_create: Create a new user with the specified name and about fields, and optionally additional documents.
	_list_items: List users with optional pagination through limit and offset parameters.
	_delete: Delete a user by UUID.
	_update: Update user details such as the 'about' description and name by UUID.
	Raises:
	AssertionError: If the provided ID for the user operations is not a valid UUID v4.
	"""
	def _get(self, id: Union[str, UUID]) -> Union[User, Awaitable[User]]:
	"""
	Get the user by their ID.
	This method is intended to retrieve a User object or an Awaitable User object by the user's unique identifier. The identifier can be a string representation of a UUID or a UUID object itself.
	Args:
	id (Union[str, UUID]): The unique identifier of the user to retrieve. It must be a valid UUID v4.
	Returns:
	Union[User, Awaitable[User]]: The retrieved User instance or an Awaitable that resolves to a User instance.
	Raises:
	AssertionError: If the `id` is not a valid UUID v4.
	Note:
	The leading underscore in the method name suggests that this method is intended for internal use and should not be a part of the public interface of the class.
	"""
	assert is_valid_uuid4(id), "id must be a valid UUID v4"
	return self.api_client.get_user(user_id=id)
	def _create(
	self,
	name: str,
	about: str,
	docs: List[DocDict] = [],
	metadata: Dict[str, Any] = {},
	) -> Union[ResourceCreatedResponse, Awaitable[ResourceCreatedResponse]]:
	# Cast docs to a list of CreateDoc objects
	"""
	Create a new resource with the given name and about information, optionally including additional docs.
	This internal method allows for creating a new resource with optional docsrmation.
	Args:
	name (str): The name of the new resource.
	about (str): A brief description about the new resource.
	docs (List[DocDict], optional): A list of dictionaries with documentation-related information. Each dictionary
	must conform to the structure expected by CreateDoc. Defaults to an empty list.
	metadata (Dict[str, Any])
	Returns:
	Union[ResourceCreatedResponse, Awaitable[ResourceCreatedResponse]]: The response indicating the resource has been
	created successfully. It can be either a synchronous ResourceCreatedResponse or an asynchronous Awaitable object
	containing a ResourceCreatedResponse.
	Note:
	This method is an internal API implementation detail and should not be used directly outside the defining class
	or module.
	Side effects:
	Modifies internal state by converting each doc dict to an instance of CreateDoc and uses the
	internal API client to create a new user resource.
	"""
	docs: List[CreateDoc] = [CreateDoc(**doc) for doc in docs]
	return self.api_client.create_user(
	name=name,
	about=about,
	docs=docs,
	metadata=metadata,
	)
	def _list_items(
	self,
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	metadata_filter: str = "{}",
	) -> Union[ListUsersResponse, Awaitable[ListUsersResponse]]:
	"""
	Fetch a list of users, with optional pagination parameters.
	Args:
	limit (Optional[int], optional): The maximum number of users to return. Defaults to None.
	offset (Optional[int], optional): The offset from the start of the list to begin returning users. Defaults to None.
	Returns:
	Union[ListUsersResponse, Awaitable[ListUsersResponse]]: An instance of ListUsersResponse,
	or an awaitable that will resolve to it, depending on the API client implementation.
	"""
	return self.api_client.list_users(
	limit=limit,
	offset=offset,
	metadata_filter=metadata_filter,
	)
	def _delete(self, user_id: Union[str, UUID]) -> Union[None, Awaitable[None]]:
	"""
	Delete a user given their user ID.
	Args:
	user_id (Union[str, UUID]): The identifier of the user. Must be a valid UUID version 4.
	Returns:
	Union[None, Awaitable[None]]: None if the deletion is synchronous, or an Awaitable
	that resolves to None if the deletion is asynchronous.
	Raises:
	AssertionError: If the user_id is not a valid UUID v4.
	"""
	assert is_valid_uuid4(user_id), "id must be a valid UUID v4"
	return self.api_client.delete_user(user_id=user_id)
	def _update(
	self,
	user_id: Union[str, UUID],
	about: Optional[str] = NotSet,
	name: Optional[str] = NotSet,
	metadata: Dict[str, Any] = NotSet,
	overwrite: bool = False,
	) -> Union[ResourceUpdatedResponse, Awaitable[ResourceUpdatedResponse]]:
	"""
	Update user details for a given user ID.
	This method updates the 'about' and/or 'name' fields for the user identified by user_id.
	Args:
	user_id (Union[str, UUID]): The ID of the user to be updated. Must be a valid UUID v4.
	about (Optional[str], optional): The new information about the user. Defaults to None.
	name (Optional[str], optional): The new name for the user. Defaults to None.
	metadata (Dict[str, Any])
	overwrite (bool, optional): Whether to overwrite the existing user data. Defaults to False.
	Returns:
	Union[ResourceUpdatedResponse, Awaitable[ResourceUpdatedResponse]]: The response indicating successful update or an Awaitable that resolves to such a response.
	Raises:
	AssertionError: If `user_id` is not a valid UUID v4.
	"""
	assert is_valid_uuid4(user_id), "id must be a valid UUID v4"
	updateFn = (
	self.api_client.update_user if overwrite else self.api_client.patch_user
	)
	update_data = dict(
	user_id=user_id,
	about=about,
	name=name,
	metadata=metadata,
	)
	update_data = {k: v for k, v in update_data.items() if v is not NotSet}
	if not update_data:
	raise ValueError("No fields to update")
	return updateFn(**update_data)
	class UsersManager(BaseUsersManager):
	"""
	A class responsible for managing users in a system.
	This class is a specialized version of the BaseUsersManager and provides
	methods for retrieving, creating, listing, deleting, and updating users within
	the system.
	Methods:
	get(id: Union[str, UUID]) -> User:
	Retrieves a user based on their unique identifier (either a string or UUID).
	create(*, name: str, about: str, docs: List[DocDict]=[]) -> ResourceCreatedResponse:
	Creates a new user with the specified name, description about the user,
	and an optional list of documents associated with the user.
	list(*, limit: Optional[int]=None, offset: Optional[int]=None) -> List[User]:
	Lists users in the system, with optional limit and offset for pagination.
	delete(*, user_id: Union[str, UUID]) -> None:
	Deletes a user from the system based on their unique identifier.
	update(*, user_id: Union[str, UUID], about: Optional[str]=None, name: Optional[str]=None) -> ResourceUpdatedResponse:
	Updates an existing user's information with optional new about and name
	fields.
	"""
	@beartype
	def get(self, id: Union[str, UUID]) -> User:
	"""
	Retrieve a User object by its identifier.
	The method supports retrieval by both string representations of a UUID and
	UUID objects directly.
	Args:
	id (Union[str, UUID]): The identifier of the User, can be a string or UUID.
	Returns:
	User: The User object associated with the provided id.
	Raises:
	ValueError: If 'id' is neither a string nor a UUID.
	NotFoundError: If a User with the given 'id' does not exist.
	"""
	return self._get(id=id)
	@beartype
	@rewrap_in_class(User)
	def create(self, **kwargs: UserCreateArgs) -> User:
	"""
	Create a new resource with the specified name, about text, and associated docs.
	Args:
	name (str): The name of the resource to create.
	about (str): A brief description of the resource.
	docs (List[DocDict], optional): A list of dictionaries representing the documents associated with the resource. Defaults to an empty list.
	Returns:
	ResourceCreatedResponse: An object representing the response received upon the successful creation of the resource.
	Note:
	Using mutable types like list as default argument values in Python is risky because if the list is modified,
	those changes will persist across subsequent calls to the function which use the default value. It is
	generally safer to use `None` as a default value and then set the default inside the function if needed.
	Raises:
	BeartypeException: If the input types do not match the specified function annotations.
	"""
	result = self._create(**kwargs)
	return result
	@beartype
	def list(
	self,
	*,
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	metadata_filter: Dict[str, Any] = {},
	) -> List[User]:
	"""
	Lists the users optionally applying limit and offset.
	Args:
	limit (Optional[int], optional): The maximum number of users to return.
	None means no limit. Defaults to None.
	offset (Optional[int], optional): The index of the first user to return.
	None means start from the beginning. Defaults to None.
	Returns:
	List[User]: A list of user objects.
	Raises:
	BeartypeException: If the type of `limit` or `offset` is not as expected.
	"""
	metadata_filter_string = json.dumps(metadata_filter)
	return self._list_items(
	limit=limit,
	offset=offset,
	metadata_filter=metadata_filter_string,
	).items
	@beartype
	def delete(
	self,
	user_id: Union[str, UUID],
	) -> None:
	"""
	Deletes a user based on the provided user ID.
	Args:
	user_id (Union[str, UUID]): Unique identifier of the user.
	Returns:
	None
	Note:
	This function is type-checked with `beartype` to ensure that the `user_id`
	parameter matches either a string or a UUID type.
	Raises:
	The specific exceptions raised depend on the implementation of the `_delete`
	method this function proxies to.
	"""
	return self._delete(
	user_id=user_id,
	)
	@beartype
	@rewrap_in_class(User)
	def update(self, *, user_id: Union[str, UUID], **kwargs: UserUpdateArgs) -> User:
	"""
	Update user information.
	This method updates user details such as the `about` text and user's `name` for a given `user_id`.
	Args:
	user_id (Union[str, UUID]): The unique identifier for the user, which can be a string or a UUID object.
	about(Optional[str], optional): The descriptive information about the user. Defaults to None, indicating that `about` should not be updated if not provided.
	name(Optional[str], optional): The name of the user. Defaults to None, indicating that `name` should not be updated if not provided.
	overwrite(bool, optional): Whether to overwrite the existing user data. Defaults to False.
	Returns:
	ResourceUpdatedResponse: An object indicating the outcome of the update operation, which typically includes the status of the operation and possibly the updated resource data.
	"""
	result = self._update(user_id=user_id, **kwargs)
	return result
	class AsyncUsersManager(BaseUsersManager):
	"""
	A class that provides asynchronous management of users extending BaseUsersManager.
	Attributes are inherited from BaseUsersManager.
	Methods:
	get (Union[UUID, str]) -> User:
	Asynchronously retrieves a user by their unique identifier (either a UUID or a string).
	create (*, name: str, about: str, docs: List[DocDict]=[]) -> ResourceCreatedResponse:
	Asynchronously creates a new user with the given name, about description, and optional list of documents.
	list (*, limit: Optional[int]=None, offset: Optional[int]=None) -> List[User]:
	Asynchronously retrieves a list of users with an optional limit and offset for pagination.
	delete (*, user_id: Union[str, UUID]) -> None:
	Asynchronously deletes a user identified by their unique ID (either a string or a UUID).
	update (*, user_id: Union[str, UUID], about: Optional[str]=None, name: Optional[str]=None) -> ResourceUpdatedResponse:
	Asynchronously updates a user's information identified by their unique ID with optional new about description and name.
	Note:
	The beartype decorator is used for runtime type checking of the parameters passed to the methods.
	"""
	@beartype
	async def get(self, id: Union[UUID, str]) -> User:
	"""
	Fetch a User object asynchronously by its identifier.
	This method retrieves a User object from some data storage asynchronously based on the provided identifier, which can be either a UUID or a string.
	Args:
	id (Union[UUID, str]): The unique identifier of the User to be retrieved.
	Returns:
	User: An instance of the User class corresponding to the given id.
	Raises:
	Exception: If the retrieval fails or the user cannot be found.
	"""
	return await self._get(id=id)
	@beartype
	@rewrap_in_class(User)
	async def create(self, **kwargs: UserCreateArgs) -> User:
	"""
	Asynchronously create a new resource with the provided name, description, and documents.
	This function is decorated with `@beartype` to ensure type checking at runtime.
	Args:
	name (str): The name of the resource to be created.
	about (str): Brief information about the resource.
	docs (List[DocDict], optional): A list of document dictionaries with structure defined by DocDict type.
	Returns:
	ResourceCreatedResponse: An instance representing the response after resource creation.
	Raises:
	BeartypeException: If any of the parameters do not match their annotated types.
	"""
	result = await self._create(**kwargs)
	return result
	@beartype
	async def list(
	self,
	*,
	limit: Optional[int] = None,
	offset: Optional[int] = None,
	metadata_filter: Dict[str, Any] = {},
	) -> List[User]:
	"""
	Asynchronously lists users with optional limits and offsets.
	This function applies the `beartype` decorator for runtime type checking.
	Args:
	limit (Optional[int], optional): The maximum number of users to be returned. Defaults to None, which means no limit.
	offset (Optional[int], optional): The number to offset the list of returned users by. Defaults to None, which means no offset.
	Returns:
	List[User]: A list of user objects.
	Raises:
	TypeError: If any input arguments are not of the expected type.
	Any other exception that might be raised during the retrieval of users from the data source.
	Note:
	The actual exception raised by the `beartype` decorator or during the users' retrieval will depend on the implementation detail of the `self._list_items` method and the `beartype` configuration.
	"""
	metadata_filter_string = json.dumps(metadata_filter)
	return (
	await self._list_items(
	limit,
	offset,
	metadata_filter=metadata_filter_string,
	)
	).items
	@beartype
	async def delete(
	self,
	user_id: Union[str, UUID],
	) -> None:
	"""
	Asynchronously deletes a user by their user ID.
	Args:
	user_id (Union[str, UUID]): The unique identifier of the user to delete, which can be a string or a UUID.
	Returns:
	None: This function does not return anything.
	Notes:
	- The function is decorated with `@beartype` for runtime type checking.
	- This function is a coroutine, it should be called with `await`.
	Raises:
	- The raised exceptions depend on the implementation of the underlying `_delete` coroutine.
	"""
	return await self._delete(
	user_id=user_id,
	)
	@beartype
	@rewrap_in_class(User)
	async def update(
	self, *, user_id: Union[str, UUID], **kwargs: UserUpdateArgs
	) -> User:
	"""
	Asynchronously updates user details.
	This function updates user details such as 'about' and 'name'. It uses type annotations to enforce the types of the parameters and an asynchronous call to '_update' method to perform the actual update operation.
	Args:
	user_id (Union[str, UUID]): The unique identifier of the user, which can be either a string or a UUID.
	about (Optional[str], optional): A description of the user. Default is None, indicating no update.
	name (Optional[str], optional): The name of the user. Default is None, indicating no update.
	Returns:
	ResourceUpdatedResponse: An object representing the update status.
	Note:
	This function is decorated with @beartype to perform runtime type checking.
	"""
	result = await self._update(user_id=user_id, **kwargs)

---

Step 2: ⌨️ Coding

• Modify sdks/python/julep/managers/memory.py ✓ 963a591 Edit

Modify sdks/python/julep/managers/memory.py with contents:
• Review the entire `memory.py` file to identify any discrepancies between the comments/docstrings and the actual code logic.
• Update the docstrings for the `BaseMemoriesManager` and `MemoriesManager` classes to accurately describe their purpose, methods, parameters, and return types. Ensure that the descriptions are concise and reflect the current functionality.
• For each method within these classes, ensure that the comments accurately describe what the method does, its parameters, any side effects, and its return values. This includes methods like `_list`, `list`, and any other methods present in the file.
• Remove any outdated comments that no longer apply to the current codebase or have become irrelevant due to code changes.
• Add new comments only where necessary to clarify complex logic or to explain the purpose and expected behavior of a method or a block of code that might not be immediately obvious to someone unfamiliar with the code.
• Ensure consistency in the documentation style throughout the file, following the existing conventions used in the `julep` SDK for docstrings and comments.
• Verify that all technical terms and entity names used in the comments and docstrings are correctly spelled and accurately reflect the entities they refer to.

--- 
+++ 
@@ -17,8 +17,8 @@
     """
     A base manager class for handling agent memories.
 
-        This manager provides an interface to interact with agent memories, allowing
-        for listing and other operations to manage an agent's memories.
+        This manager provides an interface to interact with agent memories, facilitating
+        operations such as listing and retrieving memories based on various criteria.
 
         Methods:
             _list(agent_id, query, types=None, user_id=None, limit=None, offset=None):
@@ -81,8 +81,8 @@
     """
     A class for managing memory entities associated with agents.
 
-        This class inherits from `BaseMemoriesManager` and provides
-        an interface to list memory entities for a given agent.
+        Inherits from `BaseMemoriesManager` and extends its functionality to specifically
+        manage and retrieve memory entities for agents based on query parameters.
 
         Attributes:
             Inherited from `BaseMemoriesManager`.

• Running GitHub Actions for sdks/python/julep/managers/memory.py ✓ Edit

Check sdks/python/julep/managers/memory.py with contents:

Ran GitHub Actions for 963a5916b82ff1e685c3a94223ad677f573ca4da:

---

Step 3: 🔁 Code Review

I have finished reviewing the code for completeness. I did not find errors for sweep/update_the_docstrings_and_comments_in_sd_ad5f0.

---

🎉 Latest improvements to Sweep:

• New dashboard launched for real-time tracking of Sweep issues, covering all stages from search to coding.
• Integration of OpenAI's latest Assistant API for more efficient and reliable code planning and editing, improving speed by 3x.
• Use the GitHub issues extension for creating Sweep issues directly from your editor.

💡 To recreate the pull request edit the issue title or description.
^{Something wrong? Let us know.}

This is an automated message generated by Sweep AI.