diff --git a/.gitignore b/.gitignore index fd7e8ca..45e08bb 100644 --- a/.gitignore +++ b/.gitignore @@ -20,3 +20,6 @@ coverage.xml .coverage .coverage.* .run + +# Ignore dist files +dist/ diff --git a/memorylake/mem0/__init__.py b/memorylake/mem0/__init__.py new file mode 100644 index 0000000..05f3dbd --- /dev/null +++ b/memorylake/mem0/__init__.py @@ -0,0 +1,2 @@ +from memorylake.mem0.extend.main import AsyncMemoryLakeClient as AsyncMemoryClient +from memorylake.mem0.extend.main import MemoryLakeClient as MemoryClient # noqa diff --git a/memorylake/mem0/client/__init__.py b/memorylake/mem0/client/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/memorylake/mem0/client/main.py b/memorylake/mem0/client/main.py new file mode 100644 index 0000000..3e31ac9 --- /dev/null +++ b/memorylake/mem0/client/main.py @@ -0,0 +1,1747 @@ +import hashlib +import logging +import os +import warnings +from typing import Any, NoReturn, Optional + +import httpx + +from memorylake.mem0.client.project import AsyncProject, Project +from memorylake.mem0.client.utils import api_error_handler, safe_cast + +# Exception classes are referenced in docstrings only +from memorylake.mem0.memory.setup import get_user_id, setup_config +from memorylake.mem0.memory.telemetry import capture_client_event + +logger = logging.getLogger(__name__) + +warnings.filterwarnings("default", category=DeprecationWarning) + +# Setup user config +setup_config() + + +class MemoryClient: + """Client for interacting with the Mem0 API. + + This class provides methods to create, retrieve, search, and delete + memories using the Mem0 API. + + Attributes: + api_key (str): The API key for authenticating with the Mem0 API. + host (str): The base URL for the Mem0 API. + client (httpx.Client): The HTTP client used for making API requests. + org_id (str, optional): Organization ID. + project_id (str, optional): Project ID. + user_id (str): Unique identifier for the user. + """ + + api_key: Optional[str] + host: str + org_id: Optional[str] + project_id: Optional[str] + user_id: str + client: httpx.Client + user_email: Optional[str] + project: Project + + def __init__( + self, + api_key: Optional[str] = None, + host: Optional[str] = None, + org_id: Optional[str] = None, + project_id: Optional[str] = None, + client: Optional[httpx.Client] = None, + ): + """Initialize the MemoryClient. + + Args: + api_key: The API key for authenticating with the Mem0 API. If not + provided, it will attempt to use the MEM0_API_KEY + environment variable. + host: The base URL for the Mem0 API. Defaults to + "https://api.mem0.ai". + org_id: The ID of the organization. + project_id: The ID of the project. + client: A custom httpx.Client instance. If provided, it will be + used instead of creating a new one. Note that base_url and + headers will be set/overridden as needed. + + Raises: + ValueError: If no API key is provided or found in the environment. + """ + self.api_key = api_key or os.getenv("MEM0_API_KEY") + self.host = host or "https://api.mem0.ai" + self.org_id = org_id + self.project_id = project_id + self.user_id = get_user_id() + + if not self.api_key: + raise ValueError("Mem0 API Key not provided. Please provide an API Key.") + + # Create MD5 hash of API key for user_id + self.user_id = hashlib.md5(self.api_key.encode()).hexdigest() + + if client is not None: + self.client = client + # Ensure the client has the correct base_url and headers + self.client.base_url = httpx.URL(self.host) + self.client.headers.update( + { + "Authorization": f"Token {self.api_key}", + "Mem0-User-ID": self.user_id, + } + ) + else: + self.client = httpx.Client( + base_url=self.host, + headers={ + "Authorization": f"Token {self.api_key}", + "Mem0-User-ID": self.user_id, + }, + timeout=300, + ) + self.user_email = self._validate_api_key() + + # Initialize project manager + self.project = Project( + client=self.client, + org_id=self.org_id, + project_id=self.project_id, + user_email=self.user_email, + ) + + capture_client_event("client.init", self, {"sync_type": "sync"}) + + def _validate_api_key(self) -> Optional[str]: + """Validate the API key by making a test request.""" + try: + params = self._prepare_params() + response = self.client.get("/v1/ping/", params=params) + data = response.json() + + response.raise_for_status() + + if data.get("org_id") and data.get("project_id"): + self.org_id = data.get("org_id") + self.project_id = data.get("project_id") + + return data.get("user_email") + + except httpx.HTTPStatusError as e: + try: + error_data = e.response.json() + error_message = error_data.get("detail", str(e)) + except Exception: + error_message = str(e) + raise ValueError(f"Error: {error_message}") + + @api_error_handler + def add(self, messages: Any, **kwargs: Any) -> dict[str, Any]: + """Add a new memory. + + Args: + messages: A list of message dictionaries, a single message dictionary, + or a string. If a string is provided, it will be converted to + a user message. + **kwargs: Additional parameters such as user_id, agent_id, app_id, + metadata, filters, async_mode. + + Returns: + A dictionary containing the API response in v1.1 format. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + """ + # Handle different message input formats (align with OSS behavior) + if isinstance(messages, str): + messages = [{"role": "user", "content": messages}] + elif isinstance(messages, dict): + messages = [messages] + elif not isinstance(messages, list): + raise ValueError( + f"messages must be str, dict, or list[dict], got {type(messages).__name__}" + ) + + kwargs = self._prepare_params(kwargs) + + # Set async_mode to True by default, but allow user override + if "async_mode" not in kwargs: + kwargs["async_mode"] = True + + # Force v1.1 format for all add operations + kwargs["output_format"] = "v1.1" + payload = self._prepare_payload(safe_cast(list[dict[str, str]], messages), kwargs) + response = self.client.post("/v1/memories/", json=payload) + response.raise_for_status() + if "metadata" in kwargs: + del kwargs["metadata"] + capture_client_event("client.add", self, {"keys": list(kwargs.keys()), "sync_type": "sync"}) + return response.json() + + @api_error_handler + def get(self, memory_id: str) -> dict[str, Any]: + """Retrieve a specific memory by ID. + + Args: + memory_id: The ID of the memory to retrieve. + + Returns: + A dictionary containing the memory data. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + """ + params = self._prepare_params() + response = self.client.get(f"/v1/memories/{memory_id}/", params=params) + response.raise_for_status() + capture_client_event("client.get", self, {"memory_id": memory_id, "sync_type": "sync"}) + return response.json() + + @api_error_handler + def get_all(self, **kwargs: Any) -> dict[str, Any]: + """Retrieve all memories, with optional filtering. + + Args: + **kwargs: Optional parameters for filtering (user_id, agent_id, + app_id, top_k, page, page_size). + + Returns: + A dictionary containing memories in v1.1 format: {"results": [...]} + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + """ + params = self._prepare_params(kwargs) + params.pop("async_mode", None) + + if "page" in params and "page_size" in params: + query_params = { + "page": params.pop("page"), + "page_size": params.pop("page_size"), + } + response = self.client.post("/v2/memories/", json=params, params=query_params) + else: + response = self.client.post("/v2/memories/", json=params) + response.raise_for_status() + if "metadata" in kwargs: + del kwargs["metadata"] + capture_client_event( + "client.get_all", + self, + { + "api_version": "v2", + "keys": list(kwargs.keys()), + "sync_type": "sync", + }, + ) + result = response.json() + + # Ensure v1.1 format (wrap raw list if needed) + if isinstance(result, list): + return {"results": result} + return result + + @api_error_handler + def search(self, query: str, **kwargs: Any) -> dict[str, Any]: + """Search memories based on a query. + + Args: + query: The search query string. + **kwargs: Additional parameters such as user_id, agent_id, app_id, + top_k, filters. + + Returns: + A dictionary containing search results in v1.1 format: {"results": [...]} + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + """ + payload = {"query": query} + params = self._prepare_params(kwargs) + params.pop("async_mode", None) + + payload.update(params) + + response = self.client.post("/v2/memories/search/", json=payload) + response.raise_for_status() + if "metadata" in kwargs: + del kwargs["metadata"] + capture_client_event( + "client.search", + self, + { + "api_version": "v2", + "keys": list(kwargs.keys()), + "sync_type": "sync", + }, + ) + result = response.json() + + # Ensure v1.1 format (wrap raw list if needed) + if isinstance(result, list): + return {"results": result} + return result + + @api_error_handler + def update( + self, + memory_id: str, + text: Optional[str] = None, + metadata: Optional[dict[str, Any]] = None, + ) -> dict[str, Any]: + """ + Update a memory by ID. + + Args: + memory_id (str): Memory ID. + text (str, optional): New content to update the memory with. + metadata (dict, optional): Metadata to update in the memory. + + Returns: + dict[str, Any]: The response from the server. + + Example: + >>> client.update(memory_id="mem_123", text="Likes to play tennis on weekends") + """ + if text is None and metadata is None: + raise ValueError("Either text or metadata must be provided for update.") + + payload: dict[str, Any] = {} + if text is not None: + payload["text"] = text + if metadata is not None: + payload["metadata"] = metadata + + capture_client_event("client.update", self, {"memory_id": memory_id, "sync_type": "sync"}) + params = self._prepare_params() + response = self.client.put(f"/v1/memories/{memory_id}/", json=payload, params=params) + response.raise_for_status() + return response.json() + + @api_error_handler + def delete(self, memory_id: str) -> dict[str, Any]: + """Delete a specific memory by ID. + + Args: + memory_id: The ID of the memory to delete. + + Returns: + A dictionary containing the API response. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + """ + params = self._prepare_params() + response = self.client.delete(f"/v1/memories/{memory_id}/", params=params) + response.raise_for_status() + capture_client_event("client.delete", self, {"memory_id": memory_id, "sync_type": "sync"}) + return response.json() + + @api_error_handler + def delete_all(self, **kwargs: Any) -> dict[str, str]: + """Delete all memories, with optional filtering. + + Args: + **kwargs: Optional parameters for filtering (user_id, agent_id, + app_id). + + Returns: + A dictionary containing the API response. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + """ + params = self._prepare_params(kwargs) + response = self.client.delete("/v1/memories/", params=params) + response.raise_for_status() + capture_client_event( + "client.delete_all", + self, + {"keys": list(kwargs.keys()), "sync_type": "sync"}, + ) + return response.json() + + @api_error_handler + def history(self, memory_id: str) -> list[dict[str, Any]]: + """Retrieve the history of a specific memory. + + Args: + memory_id: The ID of the memory to retrieve history for. + + Returns: + A list of dictionaries containing the memory history. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + """ + params = self._prepare_params() + response = self.client.get(f"/v1/memories/{memory_id}/history/", params=params) + response.raise_for_status() + capture_client_event("client.history", self, {"memory_id": memory_id, "sync_type": "sync"}) + return response.json() + + @api_error_handler + def users(self) -> dict[str, Any]: + """Get all users, agents, and sessions for which memories exist.""" + params = self._prepare_params() + response = self.client.get("/v1/entities/", params=params) + response.raise_for_status() + capture_client_event("client.users", self, {"sync_type": "sync"}) + return response.json() + + @api_error_handler + def delete_users( + self, + user_id: Optional[str] = None, + agent_id: Optional[str] = None, + app_id: Optional[str] = None, + run_id: Optional[str] = None, + ) -> dict[str, str]: + """Delete specific entities or all entities if no filters provided. + + Args: + user_id: Optional user ID to delete specific user + agent_id: Optional agent ID to delete specific agent + app_id: Optional app ID to delete specific app + run_id: Optional run ID to delete specific run + + Returns: + Dict with success message + + Raises: + ValueError: If specified entity not found + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + MemoryNotFoundError: If the entity doesn't exist. + NetworkError: If network connectivity issues occur. + """ + + if user_id: + to_delete = [{"type": "user", "name": user_id}] + elif agent_id: + to_delete = [{"type": "agent", "name": agent_id}] + elif app_id: + to_delete = [{"type": "app", "name": app_id}] + elif run_id: + to_delete = [{"type": "run", "name": run_id}] + else: + entities = self.users() + # Filter entities based on provided IDs using list comprehension + to_delete = [{"type": entity["type"], "name": entity["name"]} for entity in entities["results"]] + + params = self._prepare_params() + + if not to_delete: + raise ValueError("No entities to delete") + + # Delete entities and check response immediately + for entity in to_delete: + response = self.client.delete(f"/v2/entities/{entity['type']}/{entity['name']}/", params=params) + response.raise_for_status() + + capture_client_event( + "client.delete_users", + self, + { + "user_id": user_id, + "agent_id": agent_id, + "app_id": app_id, + "run_id": run_id, + "sync_type": "sync", + }, + ) + return { + "message": "Entity deleted successfully." + if (user_id or agent_id or app_id or run_id) + else "All users, agents, apps and runs deleted." + } + + @api_error_handler + def reset(self) -> dict[str, str]: + """Reset the client by deleting all users and memories. + + This method deletes all users, agents, sessions, and memories + associated with the client. + + Returns: + dict[str, str]: Message client reset successful. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + """ + self.delete_users() + + capture_client_event("client.reset", self, {"sync_type": "sync"}) + return {"message": "Client reset successful. All users and memories deleted."} + + @api_error_handler + def batch_update(self, memories: list[dict[str, Any]]) -> dict[str, Any]: + """Batch update memories. + + Args: + memories: List of memory dictionaries to update. Each dictionary must contain: + - memory_id (str): ID of the memory to update + - text (str, optional): New text content for the memory + - metadata (dict, optional): New metadata for the memory + + Returns: + dict[str, Any]: The response from the server. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + """ + response = self.client.put("/v1/batch/", json={"memories": memories}) + response.raise_for_status() + + capture_client_event("client.batch_update", self, {"sync_type": "sync"}) + return response.json() + + @api_error_handler + def batch_delete(self, memories: list[dict[str, Any]]) -> dict[str, Any]: + """Batch delete memories. + + Args: + memories: List of memory dictionaries to delete. Each dictionary + must contain: + - memory_id (str): ID of the memory to delete + + Returns: + str: Message indicating the success of the batch deletion. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + """ + response = self.client.request("DELETE", "/v1/batch/", json={"memories": memories}) + response.raise_for_status() + + capture_client_event("client.batch_delete", self, {"sync_type": "sync"}) + return response.json() + + @api_error_handler + def create_memory_export(self, schema: str, **kwargs: Any) -> dict[str, Any]: + """Create a memory export with the provided schema. + + Args: + schema: JSON schema defining the export structure + **kwargs: Optional filters like user_id, run_id, etc. + + Returns: + Dict containing export request ID and status message + """ + response = self.client.post( + "/v1/exports/", + json={"schema": schema, **self._prepare_params(kwargs)}, + ) + response.raise_for_status() + capture_client_event( + "client.create_memory_export", + self, + { + "schema": schema, + "keys": list(kwargs.keys()), + "sync_type": "sync", + }, + ) + return response.json() + + @api_error_handler + def get_memory_export(self, **kwargs: Any) -> dict[str, Any]: + """Get a memory export. + + Args: + **kwargs: Filters like user_id to get specific export + + Returns: + Dict containing the exported data + """ + response = self.client.post("/v1/exports/get/", json=self._prepare_params(kwargs)) + response.raise_for_status() + capture_client_event( + "client.get_memory_export", + self, + {"keys": list(kwargs.keys()), "sync_type": "sync"}, + ) + return response.json() + + @api_error_handler + def get_summary(self, filters: Optional[dict[str, Any]] = None) -> dict[str, Any]: + """Get the summary of a memory export. + + Args: + filters: Optional filters to apply to the summary request + + Returns: + Dict containing the export status and summary data + """ + + response = self.client.post("/v1/summary/", json=self._prepare_params({"filters": filters})) + response.raise_for_status() + capture_client_event("client.get_summary", self, {"sync_type": "sync"}) + return response.json() + + @api_error_handler + def get_project(self, fields: Optional[list[str]] = None) -> dict[str, Any]: + """Get instructions or categories for the current project. + + Args: + fields: List of fields to retrieve + + Returns: + Dictionary containing the requested fields. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + ValueError: If org_id or project_id are not set. + """ + logger.warning( + "get_project() method is going to be deprecated in version v1.0 of the package. Please use the client.project.get() method instead." + ) + if not (self.org_id and self.project_id): + raise ValueError("org_id and project_id must be set to access instructions or categories") + + params = self._prepare_params({"fields": fields}) + response = self.client.get( + f"/api/v1/orgs/organizations/{self.org_id}/projects/{self.project_id}/", + params=params, + ) + response.raise_for_status() + capture_client_event( + "client.get_project_details", + self, + {"fields": fields, "sync_type": "sync"}, + ) + return response.json() + + @api_error_handler + def update_project( + self, + custom_instructions: Optional[str] = None, + custom_categories: Optional[list[str]] = None, + retrieval_criteria: Optional[list[dict[str, Any]]] = None, + enable_graph: Optional[bool] = None, + version: Optional[str] = None, + ) -> dict[str, Any]: + """Update the project settings. + + Args: + custom_instructions: New instructions for the project + custom_categories: New categories for the project + retrieval_criteria: New retrieval criteria for the project + enable_graph: Enable or disable the graph for the project + version: Version of the project + + Returns: + Dictionary containing the API response. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + ValueError: If org_id or project_id are not set. + """ + logger.warning( + "update_project() method is going to be deprecated in version v1.0 of the package. Please use the client.project.update() method instead." + ) + if not (self.org_id and self.project_id): + raise ValueError("org_id and project_id must be set to update instructions or categories") + + if ( + custom_instructions is None + and custom_categories is None + and retrieval_criteria is None + and enable_graph is None + and version is None + ): + raise ValueError( + "Currently we only support updating custom_instructions or " + + "custom_categories or retrieval_criteria, so you must provide at least one of them" + ) + + payload = self._prepare_params( + { + "custom_instructions": custom_instructions, + "custom_categories": custom_categories, + "retrieval_criteria": retrieval_criteria, + "enable_graph": enable_graph, + "version": version, + } + ) + response = self.client.patch( + f"/api/v1/orgs/organizations/{self.org_id}/projects/{self.project_id}/", + json=payload, + ) + response.raise_for_status() + capture_client_event( + "client.update_project", + self, + { + "custom_instructions": custom_instructions, + "custom_categories": custom_categories, + "retrieval_criteria": retrieval_criteria, + "enable_graph": enable_graph, + "version": version, + "sync_type": "sync", + }, + ) + return response.json() + + def chat(self) -> NoReturn: + """Start a chat with the Mem0 AI. (Not implemented) + + Raises: + NotImplementedError: This method is not implemented yet. + """ + raise NotImplementedError("Chat is not implemented yet") + + @api_error_handler + def get_webhooks(self, project_id: str) -> dict[str, Any]: + """Get webhooks configuration for the project. + + Args: + project_id: The ID of the project to get webhooks for. + + Returns: + Dictionary containing webhook details. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + ValueError: If project_id is not set. + """ + + response = self.client.get(f"api/v1/webhooks/projects/{project_id}/") + response.raise_for_status() + capture_client_event("client.get_webhook", self, {"sync_type": "sync"}) + return response.json() + + @api_error_handler + def create_webhook(self, url: str, name: str, project_id: str, event_types: list[str]) -> dict[str, Any]: + """Create a webhook for the current project. + + Args: + url: The URL to send the webhook to. + name: The name of the webhook. + event_types: List of event types to trigger the webhook for. + + Returns: + Dictionary containing the created webhook details. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + ValueError: If project_id is not set. + """ + + payload = {"url": url, "name": name, "event_types": event_types} + response = self.client.post(f"api/v1/webhooks/projects/{project_id}/", json=payload) + response.raise_for_status() + capture_client_event("client.create_webhook", self, {"sync_type": "sync"}) + return response.json() + + @api_error_handler + def update_webhook( + self, + webhook_id: int, + name: Optional[str] = None, + url: Optional[str] = None, + event_types: Optional[list[str]] = None, + ) -> dict[str, Any]: + """Update a webhook configuration. + + Args: + webhook_id: ID of the webhook to update + name: Optional new name for the webhook + url: Optional new URL for the webhook + event_types: Optional list of event types to trigger the webhook for. + + Returns: + Dictionary containing the updated webhook details. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + """ + + payload = {k: v for k, v in {"name": name, "url": url, "event_types": event_types}.items() if v is not None} + response = self.client.put(f"api/v1/webhooks/{webhook_id}/", json=payload) + response.raise_for_status() + capture_client_event("client.update_webhook", self, {"webhook_id": webhook_id, "sync_type": "sync"}) + return response.json() + + @api_error_handler + def delete_webhook(self, webhook_id: int) -> dict[str, str]: + """Delete a webhook configuration. + + Args: + webhook_id: ID of the webhook to delete + + Returns: + Dictionary containing success message. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + """ + + response = self.client.delete(f"api/v1/webhooks/{webhook_id}/") + response.raise_for_status() + capture_client_event( + "client.delete_webhook", + self, + {"webhook_id": webhook_id, "sync_type": "sync"}, + ) + return response.json() + + @api_error_handler + def feedback( + self, + memory_id: str, + feedback: Optional[str] = None, + feedback_reason: Optional[str] = None, + ) -> dict[str, str]: + valid_feedback_values = {"POSITIVE", "NEGATIVE", "VERY_NEGATIVE"} + + feedback = feedback.upper() if feedback else None + if feedback is not None and feedback not in valid_feedback_values: + raise ValueError(f"feedback must be one of {', '.join(valid_feedback_values)} or None") + + data = { + "memory_id": memory_id, + "feedback": feedback, + "feedback_reason": feedback_reason, + } + + response = self.client.post("/v1/feedback/", json=data) + response.raise_for_status() + capture_client_event("client.feedback", self, {**data, "sync_type": "sync"}) + return response.json() + + def _prepare_payload(self, messages: list[dict[str, str]], kwargs: dict[str, Any]) -> dict[str, Any]: + """Prepare the payload for API requests. + + Args: + messages: The messages to include in the payload. + kwargs: Additional keyword arguments to include in the payload. + + Returns: + A dictionary containing the prepared payload. + """ + payload: dict[str, Any] = {} + payload["messages"] = messages + + payload.update({k: v for k, v in kwargs.items() if v is not None}) + return payload + + def _prepare_params(self, kwargs: Optional[dict[str, Any]] = None) -> dict[str, Any]: + """Prepare query parameters for API requests. + + Args: + kwargs: Keyword arguments to include in the parameters. + + Returns: + A dictionary containing the prepared parameters. + + Raises: + ValueError: If either org_id or project_id is provided but not both. + """ + + if kwargs is None: + kwargs = {} + + # Add org_id and project_id if both are available + if self.org_id and self.project_id: + kwargs["org_id"] = self.org_id + kwargs["project_id"] = self.project_id + elif self.org_id or self.project_id: + raise ValueError("Please provide both org_id and project_id") + + return {k: v for k, v in kwargs.items() if v is not None} + + +class AsyncMemoryClient: + """Asynchronous client for interacting with the Mem0 API. + + This class provides asynchronous versions of all MemoryClient methods. + It uses httpx.AsyncClient for making non-blocking API requests. + """ + + api_key: Optional[str] + host: str + org_id: Optional[str] + project_id: Optional[str] + user_id: str + async_client: httpx.AsyncClient + user_email: Optional[str] + project: AsyncProject + + def __init__( + self, + api_key: Optional[str] = None, + host: Optional[str] = None, + org_id: Optional[str] = None, + project_id: Optional[str] = None, + client: Optional[httpx.AsyncClient] = None, + ): + """Initialize the AsyncMemoryClient. + + Args: + api_key: The API key for authenticating with the Mem0 API. If not + provided, it will attempt to use the MEM0_API_KEY + environment variable. + host: The base URL for the Mem0 API. Defaults to + "https://api.mem0.ai". + org_id: The ID of the organization. + project_id: The ID of the project. + client: A custom httpx.AsyncClient instance. If provided, it will + be used instead of creating a new one. Note that base_url + and headers will be set/overridden as needed. + + Raises: + ValueError: If no API key is provided or found in the environment. + """ + self.api_key = api_key or os.getenv("MEM0_API_KEY") + self.host = host or "https://api.mem0.ai" + self.org_id = org_id + self.project_id = project_id + self.user_id = get_user_id() + + if not self.api_key: + raise ValueError("Mem0 API Key not provided. Please provide an API Key.") + + # Create MD5 hash of API key for user_id + self.user_id = hashlib.md5(self.api_key.encode()).hexdigest() + + if client is not None: + self.async_client = client + # Ensure the client has the correct base_url and headers + self.async_client.base_url = httpx.URL(self.host) + self.async_client.headers.update( + { + "Authorization": f"Token {self.api_key}", + "Mem0-User-ID": self.user_id, + } + ) + else: + self.async_client = httpx.AsyncClient( + base_url=self.host, + headers={ + "Authorization": f"Token {self.api_key}", + "Mem0-User-ID": self.user_id, + }, + timeout=300, + ) + + self.user_email = self._validate_api_key() + + # Initialize project manager + self.project = AsyncProject( + client=self.async_client, + org_id=self.org_id, + project_id=self.project_id, + user_email=self.user_email, + ) + + capture_client_event("client.init", self, {"sync_type": "async"}) + + def _validate_api_key(self) -> Optional[str]: + """Validate the API key by making a test request.""" + try: + params = self._prepare_params() + response = httpx.get( + f"{self.host}/v1/ping/", + headers={ + "Authorization": f"Token {self.api_key}", + "Mem0-User-ID": self.user_id, + }, + params=params, + ) + data = response.json() + + response.raise_for_status() + + if data.get("org_id") and data.get("project_id"): + self.org_id = data.get("org_id") + self.project_id = data.get("project_id") + + return data.get("user_email") + + except httpx.HTTPStatusError as e: + try: + error_data = e.response.json() + error_message = error_data.get("detail", str(e)) + except Exception: + error_message = str(e) + raise ValueError(f"Error: {error_message}") + + def _prepare_payload(self, messages: list[dict[str, str]], kwargs: dict[str, Any]) -> dict[str, Any]: + """Prepare the payload for API requests. + + Args: + messages: The messages to include in the payload. + kwargs: Additional keyword arguments to include in the payload. + + Returns: + A dictionary containing the prepared payload. + """ + payload: dict[str, Any] = {} + payload["messages"] = messages + + payload.update({k: v for k, v in kwargs.items() if v is not None}) + return payload + + def _prepare_params(self, kwargs: Optional[dict[str, Any]] = None) -> dict[str, Any]: + """Prepare query parameters for API requests. + + Args: + kwargs: Keyword arguments to include in the parameters. + + Returns: + A dictionary containing the prepared parameters. + + Raises: + ValueError: If either org_id or project_id is provided but not both. + """ + + if kwargs is None: + kwargs = {} + + # Add org_id and project_id if both are available + if self.org_id and self.project_id: + kwargs["org_id"] = self.org_id + kwargs["project_id"] = self.project_id + elif self.org_id or self.project_id: + raise ValueError("Please provide both org_id and project_id") + + return {k: v for k, v in kwargs.items() if v is not None} + + async def __aenter__(self) -> "AsyncMemoryClient": + return self + + async def __aexit__( + self, + exc_type: Optional[type[BaseException]], + exc_val: Optional[BaseException], + exc_tb: Optional[Any], + ) -> None: + await self.async_client.aclose() + + @api_error_handler + async def add(self, messages: Any, **kwargs: Any) -> dict[str, Any]: + # Handle different message input formats (align with OSS behavior) + if isinstance(messages, str): + messages = [{"role": "user", "content": messages}] + elif isinstance(messages, dict): + messages = [messages] + elif not isinstance(messages, list): + raise ValueError( + f"messages must be str, dict, or list[dict], got {type(messages).__name__}" + ) + + kwargs = self._prepare_params(kwargs) + + # Set async_mode to True by default, but allow user override + if "async_mode" not in kwargs: + kwargs["async_mode"] = True + + # Force v1.1 format for all add operations + kwargs["output_format"] = "v1.1" + payload = self._prepare_payload(safe_cast(list[dict[str, str]], messages), kwargs) + response = await self.async_client.post("/v1/memories/", json=payload) + response.raise_for_status() + if "metadata" in kwargs: + del kwargs["metadata"] + capture_client_event("client.add", self, {"keys": list(kwargs.keys()), "sync_type": "async"}) + return response.json() + + @api_error_handler + async def get(self, memory_id: str) -> dict[str, Any]: + params = self._prepare_params() + response = await self.async_client.get(f"/v1/memories/{memory_id}/", params=params) + response.raise_for_status() + capture_client_event("client.get", self, {"memory_id": memory_id, "sync_type": "async"}) + return response.json() + + @api_error_handler + async def get_all(self, **kwargs: Any) -> dict[str, Any]: + params = self._prepare_params(kwargs) + params.pop("async_mode", None) + + if "page" in params and "page_size" in params: + query_params = { + "page": params.pop("page"), + "page_size": params.pop("page_size"), + } + response = await self.async_client.post("/v2/memories/", json=params, params=query_params) + else: + response = await self.async_client.post("/v2/memories/", json=params) + response.raise_for_status() + if "metadata" in kwargs: + del kwargs["metadata"] + capture_client_event( + "client.get_all", + self, + { + "api_version": "v2", + "keys": list(kwargs.keys()), + "sync_type": "async", + }, + ) + result = response.json() + + # Ensure v1.1 format (wrap raw list if needed) + if isinstance(result, list): + return {"results": result} + return result + + @api_error_handler + async def search(self, query: str, **kwargs: Any) -> dict[str, Any]: + payload = {"query": query} + params = self._prepare_params(kwargs) + params.pop("async_mode", None) + + payload.update(params) + + response = await self.async_client.post("/v2/memories/search/", json=payload) + response.raise_for_status() + if "metadata" in kwargs: + del kwargs["metadata"] + capture_client_event( + "client.search", + self, + { + "api_version": "v2", + "keys": list(kwargs.keys()), + "sync_type": "async", + }, + ) + result = response.json() + + # Ensure v1.1 format (wrap raw list if needed) + if isinstance(result, list): + return {"results": result} + return result + + @api_error_handler + async def update( + self, memory_id: str, text: Optional[str] = None, metadata: Optional[dict[str, Any]] = None + ) -> dict[str, Any]: + """ + Update a memory by ID asynchronously. + + Args: + memory_id (str): Memory ID. + text (str, optional): New content to update the memory with. + metadata (dict, optional): Metadata to update in the memory. + + Returns: + dict[str, Any]: The response from the server. + + Example: + >>> await client.update(memory_id="mem_123", text="Likes to play tennis on weekends") + """ + if text is None and metadata is None: + raise ValueError("Either text or metadata must be provided for update.") + + payload: dict[str, Any] = {} + if text is not None: + payload["text"] = text + if metadata is not None: + payload["metadata"] = metadata + + capture_client_event("client.update", self, {"memory_id": memory_id, "sync_type": "async"}) + params = self._prepare_params() + response = await self.async_client.put(f"/v1/memories/{memory_id}/", json=payload, params=params) + response.raise_for_status() + return response.json() + + @api_error_handler + async def delete(self, memory_id: str) -> dict[str, Any]: + """Delete a specific memory by ID. + + Args: + memory_id: The ID of the memory to delete. + + Returns: + A dictionary containing the API response. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + """ + params = self._prepare_params() + response = await self.async_client.delete(f"/v1/memories/{memory_id}/", params=params) + response.raise_for_status() + capture_client_event("client.delete", self, {"memory_id": memory_id, "sync_type": "async"}) + return response.json() + + @api_error_handler + async def delete_all(self, **kwargs: Any) -> dict[str, str]: + """Delete all memories, with optional filtering. + + Args: + **kwargs: Optional parameters for filtering (user_id, agent_id, app_id). + + Returns: + A dictionary containing the API response. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + """ + params = self._prepare_params(kwargs) + response = await self.async_client.delete("/v1/memories/", params=params) + response.raise_for_status() + capture_client_event("client.delete_all", self, {"keys": list(kwargs.keys()), "sync_type": "async"}) + return response.json() + + @api_error_handler + async def history(self, memory_id: str) -> list[dict[str, Any]]: + """Retrieve the history of a specific memory. + + Args: + memory_id: The ID of the memory to retrieve history for. + + Returns: + A list of dictionaries containing the memory history. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + """ + params = self._prepare_params() + response = await self.async_client.get(f"/v1/memories/{memory_id}/history/", params=params) + response.raise_for_status() + capture_client_event("client.history", self, {"memory_id": memory_id, "sync_type": "async"}) + return response.json() + + @api_error_handler + async def users(self) -> dict[str, Any]: + """Get all users, agents, and sessions for which memories exist.""" + params = self._prepare_params() + response = await self.async_client.get("/v1/entities/", params=params) + response.raise_for_status() + capture_client_event("client.users", self, {"sync_type": "async"}) + return response.json() + + @api_error_handler + async def delete_users( + self, + user_id: Optional[str] = None, + agent_id: Optional[str] = None, + app_id: Optional[str] = None, + run_id: Optional[str] = None, + ) -> dict[str, str]: + """Delete specific entities or all entities if no filters provided. + + Args: + user_id: Optional user ID to delete specific user + agent_id: Optional agent ID to delete specific agent + app_id: Optional app ID to delete specific app + run_id: Optional run ID to delete specific run + + Returns: + Dict with success message + + Raises: + ValueError: If specified entity not found + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + MemoryNotFoundError: If the entity doesn't exist. + NetworkError: If network connectivity issues occur. + """ + + if user_id: + to_delete = [{"type": "user", "name": user_id}] + elif agent_id: + to_delete = [{"type": "agent", "name": agent_id}] + elif app_id: + to_delete = [{"type": "app", "name": app_id}] + elif run_id: + to_delete = [{"type": "run", "name": run_id}] + else: + entities = await self.users() + # Filter entities based on provided IDs using list comprehension + to_delete = [{"type": entity["type"], "name": entity["name"]} for entity in entities["results"]] + + params = self._prepare_params() + + if not to_delete: + raise ValueError("No entities to delete") + + # Delete entities and check response immediately + for entity in to_delete: + response = await self.async_client.delete(f"/v2/entities/{entity['type']}/{entity['name']}/", params=params) + response.raise_for_status() + + capture_client_event( + "client.delete_users", + self, + { + "user_id": user_id, + "agent_id": agent_id, + "app_id": app_id, + "run_id": run_id, + "sync_type": "async", + }, + ) + return { + "message": "Entity deleted successfully." + if (user_id or agent_id or app_id or run_id) + else "All users, agents, apps and runs deleted." + } + + @api_error_handler + async def reset(self) -> dict[str, str]: + """Reset the client by deleting all users and memories. + + This method deletes all users, agents, sessions, and memories + associated with the client. + + Returns: + dict[str, str]: Message client reset successful. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + """ + await self.delete_users() + capture_client_event("client.reset", self, {"sync_type": "async"}) + return {"message": "Client reset successful. All users and memories deleted."} + + @api_error_handler + async def batch_update(self, memories: list[dict[str, Any]]) -> dict[str, Any]: + """Batch update memories. + + Args: + memories: List of memory dictionaries to update. Each dictionary must contain: + - memory_id (str): ID of the memory to update + - text (str, optional): New text content for the memory + - metadata (dict, optional): New metadata for the memory + + Returns: + dict[str, Any]: The response from the server. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + """ + response = await self.async_client.put("/v1/batch/", json={"memories": memories}) + response.raise_for_status() + + capture_client_event("client.batch_update", self, {"sync_type": "async"}) + return response.json() + + @api_error_handler + async def batch_delete(self, memories: list[dict[str, Any]]) -> dict[str, Any]: + """Batch delete memories. + + Args: + memories: List of memory dictionaries to delete. Each dictionary + must contain: + - memory_id (str): ID of the memory to delete + + Returns: + str: Message indicating the success of the batch deletion. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + """ + response = await self.async_client.request("DELETE", "/v1/batch/", json={"memories": memories}) + response.raise_for_status() + + capture_client_event("client.batch_delete", self, {"sync_type": "async"}) + return response.json() + + @api_error_handler + async def create_memory_export(self, schema: str, **kwargs: Any) -> dict[str, Any]: + """Create a memory export with the provided schema. + + Args: + schema: JSON schema defining the export structure + **kwargs: Optional filters like user_id, run_id, etc. + + Returns: + Dict containing export request ID and status message + """ + response = await self.async_client.post("/v1/exports/", json={"schema": schema, **self._prepare_params(kwargs)}) + response.raise_for_status() + capture_client_event( + "client.create_memory_export", self, {"schema": schema, "keys": list(kwargs.keys()), "sync_type": "async"} + ) + return response.json() + + @api_error_handler + async def get_memory_export(self, **kwargs: Any) -> dict[str, Any]: + """Get a memory export. + + Args: + **kwargs: Filters like user_id to get specific export + + Returns: + Dict containing the exported data + """ + response = await self.async_client.post("/v1/exports/get/", json=self._prepare_params(kwargs)) + response.raise_for_status() + capture_client_event("client.get_memory_export", self, {"keys": list(kwargs.keys()), "sync_type": "async"}) + return response.json() + + @api_error_handler + async def get_summary(self, filters: Optional[dict[str, Any]] = None) -> dict[str, Any]: + """Get the summary of a memory export. + + Args: + filters: Optional filters to apply to the summary request + + Returns: + Dict containing the export status and summary data + """ + + response = await self.async_client.post("/v1/summary/", json=self._prepare_params({"filters": filters})) + response.raise_for_status() + capture_client_event("client.get_summary", self, {"sync_type": "async"}) + return response.json() + + @api_error_handler + async def get_project(self, fields: Optional[list[str]] = None) -> dict[str, Any]: + """Get instructions or categories for the current project. + + Args: + fields: List of fields to retrieve + + Returns: + Dictionary containing the requested fields. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + ValueError: If org_id or project_id are not set. + """ + logger.warning( + "get_project() method is going to be deprecated in version v1.0 of the package. Please use the client.project.get() method instead." + ) + if not (self.org_id and self.project_id): + raise ValueError("org_id and project_id must be set to access instructions or categories") + + params = self._prepare_params({"fields": fields}) + response = await self.async_client.get( + f"/api/v1/orgs/organizations/{self.org_id}/projects/{self.project_id}/", + params=params, + ) + response.raise_for_status() + capture_client_event("client.get_project", self, {"fields": fields, "sync_type": "async"}) + return response.json() + + @api_error_handler + async def update_project( + self, + custom_instructions: Optional[str] = None, + custom_categories: Optional[list[str]] = None, + retrieval_criteria: Optional[list[dict[str, Any]]] = None, + enable_graph: Optional[bool] = None, + version: Optional[str] = None, + ) -> dict[str, Any]: + """Update the project settings. + + Args: + custom_instructions: New instructions for the project + custom_categories: New categories for the project + retrieval_criteria: New retrieval criteria for the project + enable_graph: Enable or disable the graph for the project + version: Version of the project + + Returns: + Dictionary containing the API response. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + ValueError: If org_id or project_id are not set. + """ + logger.warning( + "update_project() method is going to be deprecated in version v1.0 of the package. Please use the client.project.update() method instead." + ) + if not (self.org_id and self.project_id): + raise ValueError("org_id and project_id must be set to update instructions or categories") + + if ( + custom_instructions is None + and custom_categories is None + and retrieval_criteria is None + and enable_graph is None + and version is None + ): + raise ValueError( + "Currently we only support updating custom_instructions or custom_categories or retrieval_criteria, so you must provide at least one of them" + ) + + payload = self._prepare_params( + { + "custom_instructions": custom_instructions, + "custom_categories": custom_categories, + "retrieval_criteria": retrieval_criteria, + "enable_graph": enable_graph, + "version": version, + } + ) + response = await self.async_client.patch( + f"/api/v1/orgs/organizations/{self.org_id}/projects/{self.project_id}/", + json=payload, + ) + response.raise_for_status() + capture_client_event( + "client.update_project", + self, + { + "custom_instructions": custom_instructions, + "custom_categories": custom_categories, + "retrieval_criteria": retrieval_criteria, + "enable_graph": enable_graph, + "version": version, + "sync_type": "async", + }, + ) + return response.json() + + async def chat(self) -> NoReturn: + """Start a chat with the Mem0 AI. (Not implemented) + + Raises: + NotImplementedError: This method is not implemented yet. + """ + raise NotImplementedError("Chat is not implemented yet") + + @api_error_handler + async def get_webhooks(self, project_id: str) -> dict[str, Any]: + """Get webhooks configuration for the project. + + Args: + project_id: The ID of the project to get webhooks for. + + Returns: + Dictionary containing webhook details. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + ValueError: If project_id is not set. + """ + + response = await self.async_client.get(f"api/v1/webhooks/projects/{project_id}/") + response.raise_for_status() + capture_client_event("client.get_webhook", self, {"sync_type": "async"}) + return response.json() + + @api_error_handler + async def create_webhook(self, url: str, name: str, project_id: str, event_types: list[str]) -> dict[str, Any]: + """Create a webhook for the current project. + + Args: + url: The URL to send the webhook to. + name: The name of the webhook. + event_types: List of event types to trigger the webhook for. + + Returns: + Dictionary containing the created webhook details. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + ValueError: If project_id is not set. + """ + + payload = {"url": url, "name": name, "event_types": event_types} + response = await self.async_client.post(f"api/v1/webhooks/projects/{project_id}/", json=payload) + response.raise_for_status() + capture_client_event("client.create_webhook", self, {"sync_type": "async"}) + return response.json() + + @api_error_handler + async def update_webhook( + self, + webhook_id: int, + name: Optional[str] = None, + url: Optional[str] = None, + event_types: Optional[list[str]] = None, + ) -> dict[str, Any]: + """Update a webhook configuration. + + Args: + webhook_id: ID of the webhook to update + name: Optional new name for the webhook + url: Optional new URL for the webhook + event_types: Optional list of event types to trigger the webhook for. + + Returns: + Dictionary containing the updated webhook details. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + """ + + payload = {k: v for k, v in {"name": name, "url": url, "event_types": event_types}.items() if v is not None} + response = await self.async_client.put(f"api/v1/webhooks/{webhook_id}/", json=payload) + response.raise_for_status() + capture_client_event("client.update_webhook", self, {"webhook_id": webhook_id, "sync_type": "async"}) + return response.json() + + @api_error_handler + async def delete_webhook(self, webhook_id: int) -> dict[str, str]: + """Delete a webhook configuration. + + Args: + webhook_id: ID of the webhook to delete + + Returns: + Dictionary containing success message. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + MemoryQuotaExceededError: If memory quota is exceeded. + NetworkError: If network connectivity issues occur. + MemoryNotFoundError: If the memory doesn't exist (for updates/deletes). + """ + + response = await self.async_client.delete(f"api/v1/webhooks/{webhook_id}/") + response.raise_for_status() + capture_client_event("client.delete_webhook", self, {"webhook_id": webhook_id, "sync_type": "async"}) + return response.json() + + @api_error_handler + async def feedback( + self, memory_id: str, feedback: Optional[str] = None, feedback_reason: Optional[str] = None + ) -> dict[str, str]: + valid_feedback_values = {"POSITIVE", "NEGATIVE", "VERY_NEGATIVE"} + + feedback = feedback.upper() if feedback else None + if feedback is not None and feedback not in valid_feedback_values: + raise ValueError(f"feedback must be one of {', '.join(valid_feedback_values)} or None") + + data = {"memory_id": memory_id, "feedback": feedback, "feedback_reason": feedback_reason} + + response = await self.async_client.post("/v1/feedback/", json=data) + response.raise_for_status() + capture_client_event("client.feedback", self, {**data, "sync_type": "async"}) + return response.json() diff --git a/memorylake/mem0/client/project.py b/memorylake/mem0/client/project.py new file mode 100644 index 0000000..9018112 --- /dev/null +++ b/memorylake/mem0/client/project.py @@ -0,0 +1,767 @@ +import logging +from abc import ABC +from typing import Any, ClassVar, Optional + +import httpx +from pydantic import BaseModel, ConfigDict, Field + +from memorylake.mem0.client.utils import api_error_handler +from memorylake.mem0.memory.telemetry import capture_client_event + +# Exception classes are referenced in docstrings only + +logger = logging.getLogger(__name__) + + +class ProjectConfig(BaseModel): + """ + Configuration for project management operations. + """ + + org_id: Optional[str] = Field(default=None, description="Organization ID") + project_id: Optional[str] = Field(default=None, description="Project ID") + user_email: Optional[str] = Field(default=None, description="User email") + + model_config: ClassVar[ConfigDict] = ConfigDict(validate_assignment=True, extra="forbid") + + +class BaseProject(ABC): + """ + Abstract base class for project management operations. + """ + + _client: Any + config: ProjectConfig + + def __init__( + self, + client: Any, + config: Optional[ProjectConfig] = None, + org_id: Optional[str] = None, + project_id: Optional[str] = None, + user_email: Optional[str] = None, + ): + """ + Initialize the project manager. + + Args: + client: HTTP client instance + config: Project manager configuration + org_id: Organization ID + project_id: Project ID + user_email: User email + """ + self._client = client + + # Handle config initialization + if config is not None: + self.config = config + else: + # Create config from parameters + self.config = ProjectConfig(org_id=org_id, project_id=project_id, user_email=user_email) + + @property + def org_id(self) -> Optional[str]: + """Get the organization ID.""" + return self.config.org_id + + @property + def project_id(self) -> Optional[str]: + """Get the project ID.""" + return self.config.project_id + + @property + def user_email(self) -> Optional[str]: + """Get the user email.""" + return self.config.user_email + + def _validate_org_project(self) -> None: + """ + Validate that both org_id and project_id are set. + + Raises: + ValueError: If org_id or project_id are not set. + """ + if not (self.config.org_id and self.config.project_id): + raise ValueError("org_id and project_id must be set to access project operations") + + def _prepare_params(self, kwargs: Optional[dict[str, Any]] = None) -> dict[str, Any]: + """ + Prepare query parameters for API requests. + + Args: + kwargs: Additional keyword arguments. + + Returns: + Dictionary containing prepared parameters. + + Raises: + ValueError: If org_id or project_id validation fails. + """ + if kwargs is None: + kwargs = {} + + # Add org_id and project_id if available + if self.config.org_id and self.config.project_id: + kwargs["org_id"] = self.config.org_id + kwargs["project_id"] = self.config.project_id + elif self.config.org_id or self.config.project_id: + raise ValueError("Please provide both org_id and project_id") + + return {k: v for k, v in kwargs.items() if v is not None} + + def _prepare_org_params(self, kwargs: Optional[dict[str, Any]] = None) -> dict[str, Any]: + """ + Prepare query parameters for organization-level API requests. + + Args: + kwargs: Additional keyword arguments. + + Returns: + Dictionary containing prepared parameters. + + Raises: + ValueError: If org_id is not provided. + """ + if kwargs is None: + kwargs = {} + + # Add org_id if available + if self.config.org_id: + kwargs["org_id"] = self.config.org_id + else: + raise ValueError("org_id must be set for organization-level operations") + + return {k: v for k, v in kwargs.items() if v is not None} + + +class Project(BaseProject): + """ + Synchronous project management operations. + """ + + def __init__( + self, + client: httpx.Client, + config: Optional[ProjectConfig] = None, + org_id: Optional[str] = None, + project_id: Optional[str] = None, + user_email: Optional[str] = None, + ): + """ + Initialize the synchronous project manager. + + Args: + client: HTTP client instance + config: Project manager configuration + org_id: Organization ID + project_id: Project ID + user_email: User email + """ + super().__init__(client, config, org_id, project_id, user_email) + self._validate_org_project() + + @api_error_handler + def get(self, fields: Optional[list[str]] = None) -> dict[str, Any]: + """ + Get project details. + + Args: + fields: List of fields to retrieve + + Returns: + Dictionary containing the requested project fields. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + NetworkError: If network connectivity issues occur. + ValueError: If org_id or project_id are not set. + """ + params = self._prepare_params({"fields": fields}) + response = self._client.get( + f"/api/v1/orgs/organizations/{self.config.org_id}/projects/{self.config.project_id}/", + params=params, + ) + response.raise_for_status() + capture_client_event( + "client.project.get", + self, + {"fields": fields, "sync_type": "sync"}, + ) + return response.json() + + @api_error_handler + def create(self, name: str, description: Optional[str] = None) -> dict[str, Any]: + """ + Create a new project within the organization. + + Args: + name: Name of the project to be created + description: Optional description for the project + + Returns: + Dictionary containing the created project details. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + NetworkError: If network connectivity issues occur. + ValueError: If org_id is not set. + """ + if not self.config.org_id: + raise ValueError("org_id must be set to create a project") + + payload = {"name": name} + if description is not None: + payload["description"] = description + + response = self._client.post( + f"/api/v1/orgs/organizations/{self.config.org_id}/projects/", + json=payload, + ) + response.raise_for_status() + capture_client_event( + "client.project.create", + self, + {"name": name, "description": description, "sync_type": "sync"}, + ) + return response.json() + + @api_error_handler + def update( + self, + custom_instructions: Optional[str] = None, + custom_categories: Optional[list[str]] = None, + retrieval_criteria: Optional[list[dict[str, Any]]] = None, + enable_graph: Optional[bool] = None, + ) -> dict[str, Any]: + """ + Update project settings. + + Args: + custom_instructions: New instructions for the project + custom_categories: New categories for the project + retrieval_criteria: New retrieval criteria for the project + enable_graph: Enable or disable the graph for the project + + Returns: + Dictionary containing the API response. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + NetworkError: If network connectivity issues occur. + ValueError: If org_id or project_id are not set. + """ + if ( + custom_instructions is None + and custom_categories is None + and retrieval_criteria is None + and enable_graph is None + ): + raise ValueError( + "At least one parameter must be provided for update: " + + "custom_instructions, custom_categories, retrieval_criteria, enable_graph" + ) + + payload = self._prepare_params( + { + "custom_instructions": custom_instructions, + "custom_categories": custom_categories, + "retrieval_criteria": retrieval_criteria, + "enable_graph": enable_graph, + } + ) + response = self._client.patch( + f"/api/v1/orgs/organizations/{self.config.org_id}/projects/{self.config.project_id}/", + json=payload, + ) + response.raise_for_status() + capture_client_event( + "client.project.update", + self, + { + "custom_instructions": custom_instructions, + "custom_categories": custom_categories, + "retrieval_criteria": retrieval_criteria, + "enable_graph": enable_graph, + "sync_type": "sync", + }, + ) + return response.json() + + @api_error_handler + def delete(self) -> dict[str, Any]: + """ + Delete the current project and its related data. + + Returns: + Dictionary containing the API response. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + NetworkError: If network connectivity issues occur. + ValueError: If org_id or project_id are not set. + """ + response = self._client.delete( + f"/api/v1/orgs/organizations/{self.config.org_id}/projects/{self.config.project_id}/", + ) + response.raise_for_status() + capture_client_event( + "client.project.delete", + self, + {"sync_type": "sync"}, + ) + return response.json() + + @api_error_handler + def get_members(self) -> dict[str, Any]: + """ + Get all members of the current project. + + Returns: + Dictionary containing the list of project members. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + NetworkError: If network connectivity issues occur. + ValueError: If org_id or project_id are not set. + """ + response = self._client.get( + f"/api/v1/orgs/organizations/{self.config.org_id}/projects/{self.config.project_id}/members/", + ) + response.raise_for_status() + capture_client_event( + "client.project.get_members", + self, + {"sync_type": "sync"}, + ) + return response.json() + + @api_error_handler + def add_member(self, email: str, role: str = "READER") -> dict[str, Any]: + """ + Add a new member to the current project. + + Args: + email: Email address of the user to add + role: Role to assign ("READER" or "OWNER") + + Returns: + Dictionary containing the API response. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + NetworkError: If network connectivity issues occur. + ValueError: If org_id or project_id are not set. + """ + if role not in ["READER", "OWNER"]: + raise ValueError("Role must be either 'READER' or 'OWNER'") + + payload = {"email": email, "role": role} + + response = self._client.post( + f"/api/v1/orgs/organizations/{self.config.org_id}/projects/{self.config.project_id}/members/", + json=payload, + ) + response.raise_for_status() + capture_client_event( + "client.project.add_member", + self, + {"email": email, "role": role, "sync_type": "sync"}, + ) + return response.json() + + @api_error_handler + def update_member(self, email: str, role: str) -> dict[str, Any]: + """ + Update a member's role in the current project. + + Args: + email: Email address of the user to update + role: New role to assign ("READER" or "OWNER") + + Returns: + Dictionary containing the API response. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + NetworkError: If network connectivity issues occur. + ValueError: If org_id or project_id are not set. + """ + if role not in ["READER", "OWNER"]: + raise ValueError("Role must be either 'READER' or 'OWNER'") + + payload = {"email": email, "role": role} + + response = self._client.put( + f"/api/v1/orgs/organizations/{self.config.org_id}/projects/{self.config.project_id}/members/", + json=payload, + ) + response.raise_for_status() + capture_client_event( + "client.project.update_member", + self, + {"email": email, "role": role, "sync_type": "sync"}, + ) + return response.json() + + @api_error_handler + def remove_member(self, email: str) -> dict[str, Any]: + """ + Remove a member from the current project. + + Args: + email: Email address of the user to remove + + Returns: + Dictionary containing the API response. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + NetworkError: If network connectivity issues occur. + ValueError: If org_id or project_id are not set. + """ + params = {"email": email} + + response = self._client.delete( + f"/api/v1/orgs/organizations/{self.config.org_id}/projects/{self.config.project_id}/members/", + params=params, + ) + response.raise_for_status() + capture_client_event( + "client.project.remove_member", + self, + {"email": email, "sync_type": "sync"}, + ) + return response.json() + + +class AsyncProject(BaseProject): + """ + Asynchronous project management operations. + """ + + def __init__( + self, + client: httpx.AsyncClient, + config: Optional[ProjectConfig] = None, + org_id: Optional[str] = None, + project_id: Optional[str] = None, + user_email: Optional[str] = None, + ): + """ + Initialize the asynchronous project manager. + + Args: + client: HTTP client instance + config: Project manager configuration + org_id: Organization ID + project_id: Project ID + user_email: User email + """ + super().__init__(client, config, org_id, project_id, user_email) + self._validate_org_project() + + @api_error_handler + async def get(self, fields: Optional[list[str]] = None) -> dict[str, Any]: + """ + Get project details. + + Args: + fields: List of fields to retrieve + + Returns: + Dictionary containing the requested project fields. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + NetworkError: If network connectivity issues occur. + ValueError: If org_id or project_id are not set. + """ + params = self._prepare_params({"fields": fields}) + response = await self._client.get( + f"/api/v1/orgs/organizations/{self.config.org_id}/projects/{self.config.project_id}/", + params=params, + ) + response.raise_for_status() + capture_client_event( + "client.project.get", + self, + {"fields": fields, "sync_type": "async"}, + ) + return response.json() + + @api_error_handler + async def create(self, name: str, description: Optional[str] = None) -> dict[str, Any]: + """ + Create a new project within the organization. + + Args: + name: Name of the project to be created + description: Optional description for the project + + Returns: + Dictionary containing the created project details. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + NetworkError: If network connectivity issues occur. + ValueError: If org_id is not set. + """ + if not self.config.org_id: + raise ValueError("org_id must be set to create a project") + + payload = {"name": name} + if description is not None: + payload["description"] = description + + response = await self._client.post( + f"/api/v1/orgs/organizations/{self.config.org_id}/projects/", + json=payload, + ) + response.raise_for_status() + capture_client_event( + "client.project.create", + self, + {"name": name, "description": description, "sync_type": "async"}, + ) + return response.json() + + @api_error_handler + async def update( + self, + custom_instructions: Optional[str] = None, + custom_categories: Optional[list[str]] = None, + retrieval_criteria: Optional[list[dict[str, Any]]] = None, + enable_graph: Optional[bool] = None, + ) -> dict[str, Any]: + """ + Update project settings. + + Args: + custom_instructions: New instructions for the project + custom_categories: New categories for the project + retrieval_criteria: New retrieval criteria for the project + enable_graph: Enable or disable the graph for the project + + Returns: + Dictionary containing the API response. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + NetworkError: If network connectivity issues occur. + ValueError: If org_id or project_id are not set. + """ + if ( + custom_instructions is None + and custom_categories is None + and retrieval_criteria is None + and enable_graph is None + ): + raise ValueError( + "At least one parameter must be provided for update: " + + "custom_instructions, custom_categories, retrieval_criteria, enable_graph" + ) + + payload = self._prepare_params( + { + "custom_instructions": custom_instructions, + "custom_categories": custom_categories, + "retrieval_criteria": retrieval_criteria, + "enable_graph": enable_graph, + } + ) + response = await self._client.patch( + f"/api/v1/orgs/organizations/{self.config.org_id}/projects/{self.config.project_id}/", + json=payload, + ) + response.raise_for_status() + capture_client_event( + "client.project.update", + self, + { + "custom_instructions": custom_instructions, + "custom_categories": custom_categories, + "retrieval_criteria": retrieval_criteria, + "enable_graph": enable_graph, + "sync_type": "async", + }, + ) + return response.json() + + @api_error_handler + async def delete(self) -> dict[str, Any]: + """ + Delete the current project and its related data. + + Returns: + Dictionary containing the API response. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + NetworkError: If network connectivity issues occur. + ValueError: If org_id or project_id are not set. + """ + response = await self._client.delete( + f"/api/v1/orgs/organizations/{self.config.org_id}/projects/{self.config.project_id}/", + ) + response.raise_for_status() + capture_client_event( + "client.project.delete", + self, + {"sync_type": "async"}, + ) + return response.json() + + @api_error_handler + async def get_members(self) -> dict[str, Any]: + """ + Get all members of the current project. + + Returns: + Dictionary containing the list of project members. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + NetworkError: If network connectivity issues occur. + ValueError: If org_id or project_id are not set. + """ + response = await self._client.get( + f"/api/v1/orgs/organizations/{self.config.org_id}/projects/{self.config.project_id}/members/", + ) + response.raise_for_status() + capture_client_event( + "client.project.get_members", + self, + {"sync_type": "async"}, + ) + return response.json() + + @api_error_handler + async def add_member(self, email: str, role: str = "READER") -> dict[str, Any]: + """ + Add a new member to the current project. + + Args: + email: Email address of the user to add + role: Role to assign ("READER" or "OWNER") + + Returns: + Dictionary containing the API response. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + NetworkError: If network connectivity issues occur. + ValueError: If org_id or project_id are not set. + """ + if role not in ["READER", "OWNER"]: + raise ValueError("Role must be either 'READER' or 'OWNER'") + + payload = {"email": email, "role": role} + + response = await self._client.post( + f"/api/v1/orgs/organizations/{self.config.org_id}/projects/{self.config.project_id}/members/", + json=payload, + ) + response.raise_for_status() + capture_client_event( + "client.project.add_member", + self, + {"email": email, "role": role, "sync_type": "async"}, + ) + return response.json() + + @api_error_handler + async def update_member(self, email: str, role: str) -> dict[str, Any]: + """ + Update a member's role in the current project. + + Args: + email: Email address of the user to update + role: New role to assign ("READER" or "OWNER") + + Returns: + Dictionary containing the API response. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + NetworkError: If network connectivity issues occur. + ValueError: If org_id or project_id are not set. + """ + if role not in ["READER", "OWNER"]: + raise ValueError("Role must be either 'READER' or 'OWNER'") + + payload = {"email": email, "role": role} + + response = await self._client.put( + f"/api/v1/orgs/organizations/{self.config.org_id}/projects/{self.config.project_id}/members/", + json=payload, + ) + response.raise_for_status() + capture_client_event( + "client.project.update_member", + self, + {"email": email, "role": role, "sync_type": "async"}, + ) + return response.json() + + @api_error_handler + async def remove_member(self, email: str) -> dict[str, Any]: + """ + Remove a member from the current project. + + Args: + email: Email address of the user to remove + + Returns: + Dictionary containing the API response. + + Raises: + ValidationError: If the input data is invalid. + AuthenticationError: If authentication fails. + RateLimitError: If rate limits are exceeded. + NetworkError: If network connectivity issues occur. + ValueError: If org_id or project_id are not set. + """ + params = {"email": email} + + response = await self._client.delete( + f"/api/v1/orgs/organizations/{self.config.org_id}/projects/{self.config.project_id}/members/", + params=params, + ) + response.raise_for_status() + capture_client_event( + "client.project.remove_member", + self, + {"email": email, "sync_type": "async"}, + ) + return response.json() diff --git a/memorylake/mem0/client/utils.py b/memorylake/mem0/client/utils.py new file mode 100644 index 0000000..93f3b94 --- /dev/null +++ b/memorylake/mem0/client/utils.py @@ -0,0 +1,137 @@ +import json +import logging +from functools import wraps +from typing import TYPE_CHECKING, Any, Callable + +import httpx +from typeguard import TypeCheckError as TypeCheckError +from typeguard import check_type as typeguard_check_type + +from memorylake.mem0.exceptions import ( + NetworkError, + create_exception_from_response, +) + +logger = logging.getLogger(__name__) + + +if TYPE_CHECKING: + # When type-checking, `safe_cast` is just an alias to `cast` + from typing import cast as cast + safe_cast = cast +else: + # When not type-checking, `safe_cast` is a function that actually performs runtime type checking + def safe_cast(typ: Any, value: object) -> Any: + """ + Safely cast a value to the given type - if the cast fails, an TypeCheckError is raised. + + This replaces `typing.cast`, which does not perform any runtime checks. + """ + try: + return typeguard_check_type(value, typ) + except TypeCheckError: + # Here, the type checking failed + raise + + +class APIError(Exception): + """Exception raised for errors in the API. + + Deprecated: Use specific exception classes from mem0.exceptions instead. + This class is maintained for backward compatibility. + """ + + pass + + +def api_error_handler(func: Callable[..., Any]) -> Callable[..., Any]: + """Decorator to handle API errors consistently. + + This decorator catches HTTP and request errors and converts them to + appropriate structured exception classes with detailed error information. + + The decorator analyzes HTTP status codes and response content to create + the most specific exception type with helpful error messages, suggestions, + and debug information. + """ + @wraps(func) + def wrapper(*args: Any, **kwargs: Any) -> Any: + try: + return func(*args, **kwargs) + except httpx.HTTPStatusError as e: + logger.error(f"HTTP error occurred: {e}") + + # Extract error details from response + response_text: str = "" + error_details: dict[str, Any] = {} + debug_info: dict[str, Any] = { + "status_code": e.response.status_code, + "url": str(e.request.url), + "method": e.request.method, + } + + try: + response_text = e.response.text + # Try to parse JSON response for additional error details + if e.response.headers.get("content-type", "").startswith("application/json"): + error_data = json.loads(response_text) + if isinstance(error_data, dict): + error_details = safe_cast(dict[str, Any], error_data) + response_text = error_details.get("detail", response_text) + except (json.JSONDecodeError, AttributeError): + # Fallback to plain text response + pass + + # Add rate limit information if available + if e.response.status_code == 429: + retry_after = e.response.headers.get("Retry-After") + if retry_after: + try: + debug_info["retry_after"] = int(retry_after) + except ValueError: + pass + + # Add rate limit headers if available + for header in ["X-RateLimit-Limit", "X-RateLimit-Remaining", "X-RateLimit-Reset"]: + value = e.response.headers.get(header) + if value: + debug_info[header.lower().replace("-", "_")] = value + + # Create specific exception based on status code + exception = create_exception_from_response( + status_code=e.response.status_code, + response_text=response_text, + details=error_details, + debug_info=debug_info, + ) + + raise exception + + except httpx.RequestError as e: + logger.error(f"Request error occurred: {e}") + + # Determine the appropriate exception type based on error type + if isinstance(e, httpx.TimeoutException): + raise NetworkError( + message=f"Request timed out: {str(e)}", + error_code="NET_TIMEOUT", + suggestion="Please check your internet connection and try again", + debug_info={"error_type": "timeout", "original_error": str(e)}, + ) + elif isinstance(e, httpx.ConnectError): + raise NetworkError( + message=f"Connection failed: {str(e)}", + error_code="NET_CONNECT", + suggestion="Please check your internet connection and try again", + debug_info={"error_type": "connection", "original_error": str(e)}, + ) + else: + # Generic network error for other request errors + raise NetworkError( + message=f"Network request failed: {str(e)}", + error_code="NET_GENERIC", + suggestion="Please check your internet connection and try again", + debug_info={"error_type": "request", "original_error": str(e)}, + ) + + return wrapper diff --git a/memorylake/mem0/exceptions.py b/memorylake/mem0/exceptions.py new file mode 100644 index 0000000..e14fa89 --- /dev/null +++ b/memorylake/mem0/exceptions.py @@ -0,0 +1,542 @@ +"""Structured exception classes for Mem0 with error codes, suggestions, and debug information. + +This module provides a comprehensive set of exception classes that replace the generic +APIError with specific, actionable exceptions. Each exception includes error codes, +user-friendly suggestions, and debug information to enable better error handling +and recovery in applications using Mem0. + +Example: + Basic usage: + try: + memory.add(content, user_id=user_id) + except RateLimitError as e: + # Implement exponential backoff + time.sleep(e.debug_info.get('retry_after', 60)) + except MemoryQuotaExceededError as e: + # Trigger quota upgrade flow + logger.error(f"Quota exceeded: {e.error_code}") + except ValidationError as e: + # Return user-friendly error + raise HTTPException(400, detail=e.suggestion) + + Advanced usage with error context: + try: + memory.update(memory_id, content=new_content) + except MemoryNotFoundError as e: + logger.warning(f"Memory {memory_id} not found: {e.message}") + if e.suggestion: + logger.info(f"Suggestion: {e.suggestion}") +""" + +from typing import Any, Optional + +from typing_extensions import override + + +class MemoryError(Exception): + """Base exception for all memory-related errors. + + This is the base class for all Mem0-specific exceptions. It provides a structured + approach to error handling with error codes, contextual details, suggestions for + resolution, and debug information. + + Attributes: + message (str): Human-readable error message. + error_code (str): Unique error identifier for programmatic handling. + details (dict): Additional context about the error. + suggestion (str): User-friendly suggestion for resolving the error. + debug_info (dict): Technical debugging information. + + Example: + raise MemoryError( + message="Memory operation failed", + error_code="MEM_001", + details={"operation": "add", "user_id": "user123"}, + suggestion="Please check your API key and try again", + debug_info={"request_id": "req_456", "timestamp": "2024-01-01T00:00:00Z"} + ) + """ + + def __init__( + self, + message: str, + error_code: str, + details: Optional[dict[str, Any]] = None, + suggestion: Optional[str] = None, + debug_info: Optional[dict[str, Any]] = None, + ): + """Initialize a MemoryError. + + Args: + message: Human-readable error message. + error_code: Unique error identifier. + details: Additional context about the error. + suggestion: User-friendly suggestion for resolving the error. + debug_info: Technical debugging information. + """ + self.message: str = message + self.error_code: str = error_code + self.details: dict[str, Any] = details or {} + self.suggestion: Optional[str] = suggestion + self.debug_info: dict[str, Any] = debug_info or {} + super().__init__(self.message) + + @override + def __repr__(self) -> str: + return ( + f"{self.__class__.__name__}(" + f"message={self.message!r}, " + f"error_code={self.error_code!r}, " + f"details={self.details!r}, " + f"suggestion={self.suggestion!r}, " + f"debug_info={self.debug_info!r})" + ) + + +class AuthenticationError(MemoryError): + """Raised when authentication fails. + + This exception is raised when API key validation fails, tokens are invalid, + or authentication credentials are missing or expired. + + Common scenarios: + - Invalid API key + - Expired authentication token + - Missing authentication headers + - Insufficient permissions + + Example: + raise AuthenticationError( + message="Invalid API key provided", + error_code="AUTH_001", + suggestion="Please check your API key in the Mem0 dashboard" + ) + """ + pass + + +class RateLimitError(MemoryError): + """Raised when rate limits are exceeded. + + This exception is raised when the API rate limit has been exceeded. + It includes information about retry timing and current rate limit status. + + The debug_info typically contains: + - retry_after: Seconds to wait before retrying + - limit: Current rate limit + - remaining: Remaining requests in current window + - reset_time: When the rate limit window resets + + Example: + raise RateLimitError( + message="Rate limit exceeded", + error_code="RATE_001", + suggestion="Please wait before making more requests", + debug_info={"retry_after": 60, "limit": 100, "remaining": 0} + ) + """ + pass + + +class ValidationError(MemoryError): + """Raised when input validation fails. + + This exception is raised when request parameters, memory content, + or configuration values fail validation checks. + + Common scenarios: + - Invalid user_id format + - Missing required fields + - Content too long or too short + - Invalid metadata format + - Malformed filters + + Example: + raise ValidationError( + message="Invalid user_id format", + error_code="VAL_001", + details={"field": "user_id", "value": "123", "expected": "string"}, + suggestion="User ID must be a non-empty string" + ) + """ + pass + + +class MemoryNotFoundError(MemoryError): + """Raised when a memory is not found. + + This exception is raised when attempting to access, update, or delete + a memory that doesn't exist or is not accessible to the current user. + + Example: + raise MemoryNotFoundError( + message="Memory not found", + error_code="MEM_404", + details={"memory_id": "mem_123", "user_id": "user_456"}, + suggestion="Please check the memory ID and ensure it exists" + ) + """ + pass + + +class NetworkError(MemoryError): + """Raised when network connectivity issues occur. + + This exception is raised for network-related problems such as + connection timeouts, DNS resolution failures, or service unavailability. + + Common scenarios: + - Connection timeout + - DNS resolution failure + - Service temporarily unavailable + - Network connectivity issues + + Example: + raise NetworkError( + message="Connection timeout", + error_code="NET_001", + suggestion="Please check your internet connection and try again", + debug_info={"timeout": 30, "endpoint": "api.mem0.ai"} + ) + """ + pass + + +class ConfigurationError(MemoryError): + """Raised when client configuration is invalid. + + This exception is raised when the client is improperly configured, + such as missing required settings or invalid configuration values. + + Common scenarios: + - Missing API key + - Invalid host URL + - Incompatible configuration options + - Missing required environment variables + + Example: + raise ConfigurationError( + message="API key not configured", + error_code="CFG_001", + suggestion="Set MEM0_API_KEY environment variable or pass api_key parameter" + ) + """ + pass + + +class MemoryQuotaExceededError(MemoryError): + """Raised when user's memory quota is exceeded. + + This exception is raised when the user has reached their memory + storage or usage limits. + + The debug_info typically contains: + - current_usage: Current memory usage + - quota_limit: Maximum allowed usage + - usage_type: Type of quota (storage, requests, etc.) + + Example: + raise MemoryQuotaExceededError( + message="Memory quota exceeded", + error_code="QUOTA_001", + suggestion="Please upgrade your plan or delete unused memories", + debug_info={"current_usage": 1000, "quota_limit": 1000, "usage_type": "memories"} + ) + """ + pass + + +class MemoryCorruptionError(MemoryError): + """Raised when memory data is corrupted. + + This exception is raised when stored memory data is found to be + corrupted, malformed, or otherwise unreadable. + + Example: + raise MemoryCorruptionError( + message="Memory data is corrupted", + error_code="CORRUPT_001", + details={"memory_id": "mem_123"}, + suggestion="Please contact support for data recovery assistance" + ) + """ + pass + + +class VectorSearchError(MemoryError): + """Raised when vector search operations fail. + + This exception is raised when vector database operations fail, + such as search queries, embedding generation, or index operations. + + Common scenarios: + - Embedding model unavailable + - Vector index corruption + - Search query timeout + - Incompatible vector dimensions + + Example: + raise VectorSearchError( + message="Vector search failed", + error_code="VEC_001", + details={"query": "find similar memories", "vector_dim": 1536}, + suggestion="Please try a simpler search query" + ) + """ + pass + + +class CacheError(MemoryError): + """Raised when caching operations fail. + + This exception is raised when cache-related operations fail, + such as cache misses, cache invalidation errors, or cache corruption. + + Example: + raise CacheError( + message="Cache operation failed", + error_code="CACHE_001", + details={"operation": "get", "key": "user_memories_123"}, + suggestion="Cache will be refreshed automatically" + ) + """ + pass + + +# OSS-specific exception classes +class VectorStoreError(MemoryError): + """Raised when vector store operations fail. + + This exception is raised when vector store operations fail, + such as embedding storage, similarity search, or vector operations. + + Example: + raise VectorStoreError( + message="Vector store operation failed", + error_code="VECTOR_001", + details={"operation": "search", "collection": "memories"}, + suggestion="Please check your vector store configuration and connection" + ) + """ + + def __init__( + self, + message: str, + error_code: str = "VECTOR_001", + details: Optional[dict[str, Any]] = None, + suggestion: str = "Please check your vector store configuration and connection", + debug_info: Optional[dict[str, Any]] = None, + ): + super().__init__(message, error_code, details, suggestion, debug_info) + + +class GraphStoreError(MemoryError): + """Raised when graph store operations fail. + + This exception is raised when graph store operations fail, + such as relationship creation, entity management, or graph queries. + + Example: + raise GraphStoreError( + message="Graph store operation failed", + error_code="GRAPH_001", + details={"operation": "create_relationship", "entity": "user_123"}, + suggestion="Please check your graph store configuration and connection" + ) + """ + + def __init__( + self, + message: str, + error_code: str = "GRAPH_001", + details: Optional[dict[str, Any]] = None, + suggestion: str = "Please check your graph store configuration and connection", + debug_info: Optional[dict[str, Any]] = None, + ): + super().__init__(message, error_code, details, suggestion, debug_info) + + +class EmbeddingError(MemoryError): + """Raised when embedding operations fail. + + This exception is raised when embedding operations fail, + such as text embedding generation or embedding model errors. + + Example: + raise EmbeddingError( + message="Embedding generation failed", + error_code="EMBED_001", + details={"text_length": 1000, "model": "openai"}, + suggestion="Please check your embedding model configuration" + ) + """ + + def __init__( + self, + message: str, + error_code: str = "EMBED_001", + details: Optional[dict[str, Any]] = None, + suggestion: str = "Please check your embedding model configuration", + debug_info: Optional[dict[str, Any]] = None, + ): + super().__init__(message, error_code, details, suggestion, debug_info) + + +class LLMError(MemoryError): + """Raised when LLM operations fail. + + This exception is raised when LLM operations fail, + such as text generation, completion, or model inference errors. + + Example: + raise LLMError( + message="LLM operation failed", + error_code="LLM_001", + details={"model": "gpt-4", "prompt_length": 500}, + suggestion="Please check your LLM configuration and API key" + ) + """ + + def __init__( + self, + message: str, + error_code: str = "LLM_001", + details: Optional[dict[str, Any]] = None, + suggestion: str = "Please check your LLM configuration and API key", + debug_info: Optional[dict[str, Any]] = None, + ): + super().__init__(message, error_code, details, suggestion, debug_info) + + +class DatabaseError(MemoryError): + """Raised when database operations fail. + + This exception is raised when database operations fail, + such as SQLite operations, connection issues, or data corruption. + + Example: + raise DatabaseError( + message="Database operation failed", + error_code="DB_001", + details={"operation": "insert", "table": "memories"}, + suggestion="Please check your database configuration and connection" + ) + """ + + def __init__( + self, + message: str, + error_code: str = "DB_001", + details: Optional[dict[str, Any]] = None, + suggestion: str = "Please check your database configuration and connection", + debug_info: Optional[dict[str, Any]] = None, + ): + super().__init__(message, error_code, details, suggestion, debug_info) + + +class DependencyError(MemoryError): + """Raised when required dependencies are missing. + + This exception is raised when required dependencies are missing, + such as optional packages for specific providers or features. + + Example: + raise DependencyError( + message="Required dependency missing", + error_code="DEPS_001", + details={"package": "kuzu", "feature": "graph_store"}, + suggestion="Please install the required dependencies: pip install kuzu" + ) + """ + + def __init__( + self, + message: str, + error_code: str = "DEPS_001", + details: Optional[dict[str, Any]] = None, + suggestion: str = "Please install the required dependencies", + debug_info: Optional[dict[str, Any]] = None, + ): + super().__init__(message, error_code, details, suggestion, debug_info) + + +# Mapping of HTTP status codes to specific exception classes +HTTP_STATUS_TO_EXCEPTION = { + 400: ValidationError, + 401: AuthenticationError, + 403: AuthenticationError, + 404: MemoryNotFoundError, + 408: NetworkError, + 409: ValidationError, + 413: MemoryQuotaExceededError, + 422: ValidationError, + 429: RateLimitError, + 500: MemoryError, + 502: NetworkError, + 503: NetworkError, + 504: NetworkError, +} + + +def create_exception_from_response( + status_code: int, + response_text: str, + error_code: Optional[str] = None, + details: Optional[dict[str, Any]] = None, + debug_info: Optional[dict[str, Any]] = None, +) -> MemoryError: + """Create an appropriate exception based on HTTP response. + + This function analyzes the HTTP status code and response to create + the most appropriate exception type with relevant error information. + + Args: + status_code: HTTP status code from the response. + response_text: Response body text. + error_code: Optional specific error code. + details: Additional error context. + debug_info: Debug information. + + Returns: + An instance of the appropriate MemoryError subclass. + + Example: + exception = create_exception_from_response( + status_code=429, + response_text="Rate limit exceeded", + debug_info={"retry_after": 60} + ) + # Returns a RateLimitError instance + """ + exception_class = HTTP_STATUS_TO_EXCEPTION.get(status_code, MemoryError) + + # Generate error code if not provided + if not error_code: + error_code = f"HTTP_{status_code}" + + # Create appropriate suggestion based on status code + suggestions = { + 400: "Please check your request parameters and try again", + 401: "Please check your API key and authentication credentials", + 403: "You don't have permission to perform this operation", + 404: "The requested resource was not found", + 408: "Request timed out. Please try again", + 409: "Resource conflict. Please check your request", + 413: "Request too large. Please reduce the size of your request", + 422: "Invalid request data. Please check your input", + 429: "Rate limit exceeded. Please wait before making more requests", + 500: "Internal server error. Please try again later", + 502: "Service temporarily unavailable. Please try again later", + 503: "Service unavailable. Please try again later", + 504: "Gateway timeout. Please try again later", + } + + suggestion = suggestions.get(status_code, "Please try again later") + + return exception_class( + message=response_text or f"HTTP {status_code} error", + error_code=error_code, + details=details or {}, + suggestion=suggestion, + debug_info=debug_info or {}, + ) diff --git a/memorylake/mem0/extend/main.py b/memorylake/mem0/extend/main.py new file mode 100644 index 0000000..5a05005 --- /dev/null +++ b/memorylake/mem0/extend/main.py @@ -0,0 +1,222 @@ +from __future__ import annotations + +import uuid +from typing import Any, Literal, Optional + +from memorylake.mem0.client.main import AsyncMemoryClient, MemoryClient +from memorylake.mem0.client.utils import api_error_handler +from memorylake.mem0.memory.telemetry import capture_client_event + + +class MemoryLakeClient(MemoryClient): + + def new_reflection( + self, + user_id: str, + target_type: Literal["user", "location"], + target_id: str, + ) -> Reflection: + return Reflection( + user_id=user_id, + target_type=target_type, + target_id=target_id, + memory_client=self, + ) + + @api_error_handler + def end_session( + self, + chat_session_id: str, + timestamp: int, + ) -> dict[str, Any]: + """End a chat session. + + Args: + chat_session_id: The ID of the chat session to end. + timestamp: The timestamp of the session end event. + + Returns: + A dictionary containing the API response. + """ + payload = self._prepare_params( + { + "chat_session_id": chat_session_id, + "timestamp": timestamp, + "event_type": "end", + } + ) + response = self.client.post("/v3/chat_session/event/", json=payload) + response.raise_for_status() + capture_client_event( + "client.end_session", + self, + {"chat_session_id": chat_session_id, "sync_type": "sync"}, + ) + return response.json() + + def prepare_params(self, kwargs: Optional[dict[str, Any]] = None) -> dict[str, Any]: + return self._prepare_params(kwargs) + + +class AsyncMemoryLakeClient(AsyncMemoryClient): + + def new_reflection( + self, + user_id: str, + target_type: Literal["user", "location"], + target_id: str, + ) -> AsyncReflection: + return AsyncReflection( + user_id=user_id, + target_type=target_type, + target_id=target_id, + memory_client=self, + ) + + @api_error_handler + async def end_session( + self, + chat_session_id: str, + timestamp: int, + ) -> dict[str, Any]: + """End a chat session. + + Args: + chat_session_id: The ID of the chat session to end. + timestamp: The timestamp of the session end event. + + Returns: + A dictionary containing the API response. + """ + payload = self._prepare_params( + { + "chat_session_id": chat_session_id, + "timestamp": timestamp, + "event_type": "end", + } + ) + response = await self.async_client.post("/v3/chat_session/event/", json=payload) + response.raise_for_status() + capture_client_event( + "client.end_session", + self, + {"chat_session_id": chat_session_id, "sync_type": "async"}, + ) + return response.json() + + def prepare_params(self, kwargs: Optional[dict[str, Any]] = None) -> dict[str, Any]: + return self._prepare_params(kwargs) + + +class Reflection: + + user_id: str + target_type: Literal["user", "location"] + target_id: str + memory_client: MemoryLakeClient + reflect_id: str + + def __init__( + self, + user_id: str, + target_type: Literal["user", "location"], + target_id: str, + memory_client: MemoryLakeClient, + ): + self.user_id = user_id + self.target_type = target_type + self.target_id = target_id + self.memory_client = memory_client + self.reflect_id = str(uuid.uuid4()) + + @api_error_handler + def recollect(self, **kwargs: Any) -> dict[str, Any]: + kwargs["user_id"] = self.user_id + kwargs["metadata"] = self._prepare_metadata(kwargs.get("metadata") or {}) + payload = self.memory_client.prepare_params(kwargs) + response = self.memory_client.client.post("/v3/memories/recollect/", json=payload) + response.raise_for_status() + capture_client_event( + "client.recollect", + self.memory_client, + {"reflect_id": self.reflect_id, "sync_type": "sync"}, + ) + return response.json() + + def save(self, messages: Any, **kwargs: Any) -> dict[str, Any]: + kwargs["user_id"] = self.user_id + kwargs["metadata"] = self._prepare_metadata(kwargs.get("metadata") or {}, "reflect") + return self.memory_client.add(messages, **kwargs) + + def _prepare_metadata(self, metadata: dict[str, Any], category: Optional[str] = None) -> dict[str, Any]: + user_extension: dict[str, Any] = metadata.get("memorylake_extension") or {} + metadata["memorylake_extension"] = { + **user_extension, + "reflect_id": self.reflect_id, + "reflect_target": { + "target_type": self.target_type, + "target_id": self.target_id, + }, + } + + if category: + metadata["memorylake_extension"]["category"] = category + + return metadata + + +class AsyncReflection: + + user_id: str + target_type: Literal["user", "location"] + target_id: str + memory_client: AsyncMemoryLakeClient + reflect_id: str + + def __init__( + self, + user_id: str, + target_type: Literal["user", "location"], + target_id: str, + memory_client: AsyncMemoryLakeClient, + ): + self.user_id = user_id + self.target_type = target_type + self.target_id = target_id + self.memory_client = memory_client + self.reflect_id = str(uuid.uuid4()) + + @api_error_handler + async def recollect(self, **kwargs: Any) -> dict[str, Any]: + kwargs["user_id"] = self.user_id + kwargs["metadata"] = self._prepare_metadata(kwargs.get("metadata") or {}) + payload = self.memory_client.prepare_params(kwargs) + response = await self.memory_client.async_client.post("/v3/memories/recollect/", json=payload) + response.raise_for_status() + capture_client_event( + "client.recollect", + self.memory_client, + {"reflect_id": self.reflect_id, "sync_type": "async"}, + ) + return response.json() + + async def save(self, messages: Any, **kwargs: Any) -> dict[str, Any]: + kwargs["user_id"] = self.user_id + kwargs["metadata"] = self._prepare_metadata(kwargs.get("metadata") or {}, "reflect") + return await self.memory_client.add(messages, **kwargs) + + def _prepare_metadata(self, metadata: dict[str, Any], category: Optional[str] = None) -> dict[str, Any]: + user_extension: dict[str, Any] = metadata.get("memorylake_extension") or {} + metadata["memorylake_extension"] = { + **user_extension, + "reflect_id": self.reflect_id, + "reflect_target": { + "target_type": self.target_type, + "target_id": self.target_id, + }, + } + + if category: + metadata["memorylake_extension"]["category"] = category + + return metadata diff --git a/memorylake/mem0/memory/setup.py b/memorylake/mem0/memory/setup.py new file mode 100644 index 0000000..45a91e1 --- /dev/null +++ b/memorylake/mem0/memory/setup.py @@ -0,0 +1,57 @@ +import json +import os +import uuid +from typing import Any + +# Set up the directory path +VECTOR_ID = str(uuid.uuid4()) +home_dir = os.path.expanduser("~") +mem0_dir = os.environ.get("MEM0_DIR") or os.path.join(home_dir, ".mem0") +os.makedirs(mem0_dir, exist_ok=True) + + +def setup_config(): + config_path = os.path.join(mem0_dir, "config.json") + if not os.path.exists(config_path): + user_id = str(uuid.uuid4()) + config = {"user_id": user_id} + with open(config_path, "w") as config_file: + json.dump(config, config_file, indent=4) + + +def get_user_id() -> str: + config_path = os.path.join(mem0_dir, "config.json") + if not os.path.exists(config_path): + return "anonymous_user" + + try: + with open(config_path, "r") as config_file: + config = json.load(config_file) + user_id = config.get("user_id") + return user_id + except Exception: + return "anonymous_user" + + +def get_or_create_user_id(vector_store: Any) -> str: + """Store user_id in vector store and return it.""" + user_id = get_user_id() + + # Try to get existing user_id from vector store + try: + existing: Any = vector_store.get(vector_id=user_id) + if existing and hasattr(existing, "payload") and existing.payload and "user_id" in existing.payload: + return str(existing.payload["user_id"]) + except Exception: + pass + + # If we get here, we need to insert the user_id + try: + dims: int = int(getattr(vector_store, "embedding_model_dims", 1536)) + vector_store.insert( + vectors=[[0.1] * dims], payloads=[{"user_id": user_id, "type": "user_identity"}], ids=[user_id] + ) + except Exception: + pass + + return user_id diff --git a/memorylake/mem0/memory/telemetry.py b/memorylake/mem0/memory/telemetry.py new file mode 100644 index 0000000..9a91ac6 --- /dev/null +++ b/memorylake/mem0/memory/telemetry.py @@ -0,0 +1,17 @@ +from typing import Any, Optional + + +def capture_client_event( + event_name: str, + instance: Any, + additional_data: Optional[dict[str, Any]] = None, +) -> None: + """Capture a client event for telemetry. + + This is a stub implementation. Parameters are part of the public API + and are intentionally accepted but not used. + """ + _ = event_name + _ = instance + _ = additional_data + ... diff --git a/pyproject.toml b/pyproject.toml index b6d2176..47c989c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,6 +26,8 @@ dependencies = [ "pydantic>=2.9.2", "pyyaml>=6.0.1", "tomli>=2.0.0; python_version < '3.11'", + "typeguard==4.4.2", + "typing_extensions>=4.0.0", ] [project.urls] @@ -99,7 +101,7 @@ env_files = ["cicd/ci-test.env"] # PytestDeprecationWarning: The configuration option "asyncio_default_fixture_loop_scope" is unset. # The event loop scope for asynchronous fixtures will default to the fixture caching scope. # Future versions of pytest-asyncio will default the loop scope for asynchronous fixtures to function scope. -# Set the default fixture loop scope explicitly in order to avoid unexpected behavior in the future. +# Set the default fixture loop scope explicitly in order to avoid unexpected behavior in the future. # Valid fixture loop scopes are: "function", "class", "module", "package", "session" asyncio_default_fixture_loop_scope = "function" asyncio_default_test_loop_scope = "function"