mem0ai--mem0/mem0/client/main.py

import hashlib
import logging
import os
import warnings
from typing import Any, Dict, List, Optional
from urllib.parse import quote

import httpx
import requests

from mem0.client.project import AsyncProject, Project
from mem0.client.types import (
    AddMemoryOptions,
    DeleteAllMemoryOptions,
    GetAllMemoryOptions,
    ProjectUpdateOptions,
    SearchMemoryOptions,
    UpdateMemoryOptions,
)
from mem0.client.utils import api_error_handler

# Exception classes are referenced in docstrings only
from mem0.memory.setup import get_user_id, is_aliased, mark_aliased, read_anon_ids, setup_config
from mem0.memory.telemetry import capture_client_event, client_telemetry

logger = logging.getLogger(__name__)

warnings.filterwarnings("default", category=DeprecationWarning)

# Setup user config
setup_config()

# Entity parameters that must be passed via filters, not top-level
ENTITY_PARAMS = frozenset({"user_id", "agent_id", "app_id", "run_id"})


def _validate_and_trim_search_query(query: str) -> str:
    if not isinstance(query, str):
        raise ValueError("Invalid query: must be a non-empty string.")
    trimmed = query.strip()
    if not trimmed:
        raise ValueError("Invalid query: cannot be empty or whitespace-only.")
    return trimmed


def _encode_path_segment(value: Any) -> str:
    return quote(str(value), safe="")


def _maybe_alias_anon_to_email(user_email):
    """Fire $identify per prior anon ID so PostHog merges them into email.

    Idempotent via telemetry.aliased_pairs: only writes markers when
    telemetry is actually enabled, so disabling/re-enabling MEM0_TELEMETRY still works.
    Best-effort: never raises.
    """
    if client_telemetry.posthog is None:
        return
    if not user_email or "@" not in user_email:
        return
    try:
        anon_ids = read_anon_ids()
        seen = set()
        for anon_id in (anon_ids.get("oss"), anon_ids.get("cli")):
            if not anon_id or anon_id == user_email or anon_id in seen:
                continue
            seen.add(anon_id)
            if is_aliased(anon_id, user_email):
                continue
            if client_telemetry.capture_identify(anon_id, user_email):
                mark_aliased(anon_id, user_email)
    except Exception as e:
        logger.debug("Failed to alias anon telemetry to %r: %s", user_email, e)


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.
        user_id (str): Unique identifier for the user.
    """

    def __init__(
        self,
        api_key: Optional[str] = None,
        host: 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".
            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 = None
        self.project_id = None
        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,
        )

        _maybe_alias_anon_to_email(self.user_email)
        capture_client_event("client.init", self, {"sync_type": "sync"})

    def _validate_api_key(self):
        """Validate the API key by making a test request."""
        try:
            params = self._prepare_params()
            response = self.client.get("/v1/ping/", params=params)
            response.raise_for_status()

            data = response.json()

            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, options: Optional[AddMemoryOptions] = None, **kwargs) -> 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.
            options: Typed options for the add operation (AddMemoryOptions).
            **kwargs: Additional parameters such as user_id, agent_id, app_id,
                      metadata, filters.

        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).
        """
        kwargs = {**(options.model_dump(exclude_unset=True) if options else {}), **kwargs}
        # 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)
        payload = self._prepare_payload(messages, kwargs)
        response = self.client.post("/v3/memories/add/", 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/{_encode_path_segment(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, options: Optional[GetAllMemoryOptions] = None, **kwargs) -> Dict[str, Any]:
        """Retrieve all memories, with optional filtering.

        Args:
            options: Typed options for the get_all operation (GetAllMemoryOptions).
            **kwargs: Optional parameters for filtering (filters, page, page_size).

        Returns:
            A paginated dict: {"count": int, "next": str | None, "previous": str | None, "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).
        """
        # Reject top-level entity params - must use filters instead
        invalid_keys = ENTITY_PARAMS & set(kwargs.keys())
        if invalid_keys:
            raise ValueError(
                f"Top-level entity parameters {invalid_keys} are not supported in get_all(). "
                f"Use filters={{'user_id': '...'}} instead."
            )

        kwargs = {**(options.model_dump(exclude_unset=True) if options else {}), **kwargs}
        params = self._prepare_params(kwargs)

        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("/v3/memories/", json=params, params=query_params)
        else:
            response = self.client.post("/v3/memories/", json=params)
        response.raise_for_status()
        if "metadata" in kwargs:
            del kwargs["metadata"]
        capture_client_event(
            "client.get_all",
            self,
            {
                "keys": list(kwargs.keys()),
                "sync_type": "sync",
            },
        )
        return response.json()

    @api_error_handler
    def search(self, query: str, options: Optional[SearchMemoryOptions] = None, **kwargs) -> Dict[str, Any]:
        """Search memories based on a query.

        Args:
            query: The search query string.
            options: Typed options for the search operation (SearchMemoryOptions).
            **kwargs: Additional parameters such as filters, top_k, rerank.

        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).
        """
        # Reject top-level entity params - must use filters instead
        invalid_keys = ENTITY_PARAMS & set(kwargs.keys())
        if invalid_keys:
            raise ValueError(
                f"Top-level entity parameters {invalid_keys} are not supported in search(). "
                f"Use filters={{'user_id': '...'}} instead."
            )

        kwargs = {**(options.model_dump(exclude_unset=True) if options else {}), **kwargs}
        params = self._prepare_params(kwargs)
        query = _validate_and_trim_search_query(query)
        payload = {"query": query, **params}

        response = self.client.post("/v3/memories/search/", json=payload)
        response.raise_for_status()
        if "metadata" in kwargs:
            del kwargs["metadata"]
        capture_client_event(
            "client.search",
            self,
            {
                "keys": list(kwargs.keys()),
                "sync_type": "sync",
            },
        )
        return response.json()

    @api_error_handler
    def update(
        self,
        memory_id: str,
        options: Optional[UpdateMemoryOptions] = None,
        **kwargs,
    ) -> Dict[str, Any]:
        """Update a memory by ID.

        Args:
            memory_id: The ID of the memory to update.
            options: Typed options (UpdateMemoryOptions) with text, metadata,
                     timestamp, and/or expiration_date fields.
            **kwargs: Alternatively pass text, metadata, timestamp, or
                      expiration_date as keyword args.

        Returns:
            Dict[str, Any]: The response from the server.

        Raises:
            ValueError: If none of text, metadata, timestamp, or expiration_date are provided.

        Example:
            >>> client.update("mem_123", UpdateMemoryOptions(text="Updated text"))
            >>> client.update("mem_123", text="Updated text")
            >>> client.update("mem_123", expiration_date=None)
        """
        payload = {**(options.model_dump(exclude_unset=True) if options else {}), **kwargs}
        payload = {k: v for k, v in payload.items() if v is not None or k == "expiration_date"}

        if not payload:
            raise ValueError("At least one of text, metadata, timestamp, or expiration_date must be provided for update.")

        capture_client_event("client.update", self, {"memory_id": memory_id, "sync_type": "sync"})
        params = self._prepare_params()
        response = self.client.put(f"/v1/memories/{_encode_path_segment(memory_id)}/", json=payload, params=params)
        response.raise_for_status()
        return response.json()

    @api_error_handler
    def delete(self, memory_id: str, delete_linked: bool = False) -> Dict[str, Any]:
        """Delete a specific memory by ID.

        Args:
            memory_id: The ID of the memory to delete.
            delete_linked: When True, also delete the older memories this one
                superseded (the v3 ``linked_memory_ids`` chain), transitively.
                This is the delete-side counterpart of ``latest_only`` — it
                stops a superseded memory from resurfacing after you delete the
                current one. Defaults to False (only the given memory is deleted).

        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({"delete_linked": delete_linked or None})
        response = self.client.delete(f"/v1/memories/{_encode_path_segment(memory_id)}/", params=params)
        response.raise_for_status()
        capture_client_event(
            "client.delete", self, {"memory_id": memory_id, "delete_linked": delete_linked, "sync_type": "sync"}
        )
        return response.json()

    @api_error_handler
    def delete_all(self, options: Optional[DeleteAllMemoryOptions] = None, **kwargs) -> Dict[str, str]:
        """Delete all memories, with optional filtering.

        Args:
            options: Typed options for the delete_all operation (DeleteAllMemoryOptions).
            **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).
        """
        kwargs = {**(options.model_dump(exclude_unset=True) if options else {}), **kwargs}
        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/{_encode_path_segment(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/{_encode_path_segment(entity['type'])}/{_encode_path_segment(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) -> 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) -> 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,
        options: Optional[ProjectUpdateOptions] = None,
        custom_instructions: Optional[str] = None,
        custom_categories: Optional[List[str]] = None,
        retrieval_criteria: Optional[List[Dict[str, Any]]] = None,
        memory_depth: Optional[str] = None,
        usecase_setting: Optional[str] = None,
        multilingual: Optional[bool] = None,
    ) -> Dict[str, Any]:
        """Update the project settings.

        Args:
            options: Typed options for the update operation (ProjectUpdateOptions).
            custom_instructions: New instructions for the project.
            custom_categories: New categories for the project.
            retrieval_criteria: New retrieval criteria for the project.
            memory_depth: Memory depth for the project.
            usecase_setting: Usecase setting for the project.
            multilingual: Whether to use the input language for memory storage and retrieval.

        Returns:
            Dictionary containing the API response.

        Raises:
            ValueError: If org_id or project_id are not set, or no update fields provided.
        """
        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")

        kwargs = {
            **(options.model_dump(exclude_unset=True) if options else {}),
            **{
                k: v
                for k, v in {
                    "custom_instructions": custom_instructions,
                    "custom_categories": custom_categories,
                    "retrieval_criteria": retrieval_criteria,
                    "memory_depth": memory_depth,
                    "usecase_setting": usecase_setting,
                    "multilingual": multilingual,
                }.items()
                if v is not None
            },
        }

        if not kwargs:
            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(kwargs)
        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,
            {**kwargs, "sync_type": "sync"},
        )
        return response.json()

    def chat(self):
        """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 = {}
        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.
        """

        if kwargs is None:
            kwargs = {}

        # org_id and project_id are resolved from API key — not injected into params

        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.
    """

    def __init__(
        self,
        api_key: Optional[str] = None,
        host: 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".
            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 = None
        self.project_id = None
        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,
        )

        _maybe_alias_anon_to_email(self.user_email)
        capture_client_event("client.init", self, {"sync_type": "async"})

    def _validate_api_key(self):
        """Validate the API key by making a test request."""
        try:
            params = self._prepare_params()
            response = requests.get(
                f"{self.host}/v1/ping/",
                headers={
                    "Authorization": f"Token {self.api_key}",
                    "Mem0-User-ID": self.user_id,
                },
                params=params,
            )
            response.raise_for_status()

            data = response.json()

            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 requests.exceptions.HTTPError 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 = {}
        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.
        """

        if kwargs is None:
            kwargs = {}

        # org_id and project_id are resolved from API key — not injected into params

        return {k: v for k, v in kwargs.items() if v is not None}

    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.async_client.aclose()

    @api_error_handler
    async def add(self, messages, options: Optional[AddMemoryOptions] = None, **kwargs) -> 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.
            options: Typed options for the add operation (AddMemoryOptions).
            **kwargs: Additional parameters such as user_id, agent_id, app_id,
                      metadata, filters.

        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).
        """
        kwargs = {**(options.model_dump(exclude_unset=True) if options else {}), **kwargs}
        # 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)
        payload = self._prepare_payload(messages, kwargs)
        response = await self.async_client.post("/v3/memories/add/", 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/{_encode_path_segment(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, options: Optional[GetAllMemoryOptions] = None, **kwargs) -> Dict[str, Any]:
        """Retrieve all memories, with optional filtering.

        Args:
            options: Typed options for the get_all operation (GetAllMemoryOptions).
            **kwargs: Optional parameters for filtering (filters, page, page_size).

        Returns:
            A paginated dict: {"count": int, "next": str | None, "previous": str | None, "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).
        """
        # Reject top-level entity params - must use filters instead
        invalid_keys = ENTITY_PARAMS & set(kwargs.keys())
        if invalid_keys:
            raise ValueError(
                f"Top-level entity parameters {invalid_keys} are not supported in get_all(). "
                f"Use filters={{'user_id': '...'}} instead."
            )

        kwargs = {**(options.model_dump(exclude_unset=True) if options else {}), **kwargs}
        params = self._prepare_params(kwargs)

        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("/v3/memories/", json=params, params=query_params)
        else:
            response = await self.async_client.post("/v3/memories/", json=params)
        response.raise_for_status()
        if "metadata" in kwargs:
            del kwargs["metadata"]
        capture_client_event(
            "client.get_all",
            self,
            {
                "keys": list(kwargs.keys()),
                "sync_type": "async",
            },
        )
        return response.json()

    @api_error_handler
    async def search(self, query: str, options: Optional[SearchMemoryOptions] = None, **kwargs) -> Dict[str, Any]:
        """Search memories based on a query.

        Args:
            query: The search query string.
            options: Typed options for the search operation (SearchMemoryOptions).
            **kwargs: Additional parameters such as filters, top_k, rerank.

        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).
        """
        # Reject top-level entity params - must use filters instead
        invalid_keys = ENTITY_PARAMS & set(kwargs.keys())
        if invalid_keys:
            raise ValueError(
                f"Top-level entity parameters {invalid_keys} are not supported in search(). "
                f"Use filters={{'user_id': '...'}} instead."
            )

        kwargs = {**(options.model_dump(exclude_unset=True) if options else {}), **kwargs}
        params = self._prepare_params(kwargs)
        query = _validate_and_trim_search_query(query)
        payload = {"query": query, **params}

        response = await self.async_client.post("/v3/memories/search/", json=payload)
        response.raise_for_status()
        if "metadata" in kwargs:
            del kwargs["metadata"]
        capture_client_event(
            "client.search",
            self,
            {
                "keys": list(kwargs.keys()),
                "sync_type": "async",
            },
        )
        return response.json()

    @api_error_handler
    async def update(
        self,
        memory_id: str,
        options: Optional[UpdateMemoryOptions] = None,
        **kwargs,
    ) -> Dict[str, Any]:
        """Update a memory by ID asynchronously.

        Args:
            memory_id: The ID of the memory to update.
            options: Typed options (UpdateMemoryOptions) with text, metadata,
                     timestamp, and/or expiration_date fields.
            **kwargs: Alternatively pass text, metadata, timestamp, or
                      expiration_date as keyword args.

        Returns:
            Dict[str, Any]: The response from the server.

        Raises:
            ValueError: If none of text, metadata, timestamp, or expiration_date are provided.

        Example:
            >>> await client.update("mem_123", UpdateMemoryOptions(text="Updated text"))
            >>> await client.update("mem_123", text="Updated text")
            >>> await client.update("mem_123", expiration_date=None)
        """
        payload = {**(options.model_dump(exclude_unset=True) if options else {}), **kwargs}
        payload = {k: v for k, v in payload.items() if v is not None or k == "expiration_date"}

        if not payload:
            raise ValueError("At least one of text, metadata, timestamp, or expiration_date must be provided for update.")

        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/{_encode_path_segment(memory_id)}/", json=payload, params=params
        )
        response.raise_for_status()
        return response.json()

    @api_error_handler
    async def delete(self, memory_id: str, delete_linked: bool = False) -> Dict[str, Any]:
        """Delete a specific memory by ID.

        Args:
            memory_id: The ID of the memory to delete.
            delete_linked: When True, also delete the older memories this one
                superseded (the v3 ``linked_memory_ids`` chain), transitively.
                This is the delete-side counterpart of ``latest_only`` — it
                stops a superseded memory from resurfacing after you delete the
                current one. Defaults to False (only the given memory is deleted).

        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({"delete_linked": delete_linked or None})
        response = await self.async_client.delete(f"/v1/memories/{_encode_path_segment(memory_id)}/", params=params)
        response.raise_for_status()
        capture_client_event(
            "client.delete", self, {"memory_id": memory_id, "delete_linked": delete_linked, "sync_type": "async"}
        )
        return response.json()

    @api_error_handler
    async def delete_all(self, options: Optional[DeleteAllMemoryOptions] = None, **kwargs) -> Dict[str, str]:
        """Delete all memories, with optional filtering.

        Args:
            options: Typed options for the delete_all operation (DeleteAllMemoryOptions).
            **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).
        """
        kwargs = {**(options.model_dump(exclude_unset=True) if options else {}), **kwargs}
        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/{_encode_path_segment(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/{_encode_path_segment(entity['type'])}/{_encode_path_segment(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) -> 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) -> 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,
        options: Optional[ProjectUpdateOptions] = None,
        custom_instructions: Optional[str] = None,
        custom_categories: Optional[List[str]] = None,
        retrieval_criteria: Optional[List[Dict[str, Any]]] = None,
        memory_depth: Optional[str] = None,
        usecase_setting: Optional[str] = None,
        multilingual: Optional[bool] = None,
    ) -> Dict[str, Any]:
        """Update the project settings.

        Args:
            options: Typed options for the update operation (ProjectUpdateOptions).
            custom_instructions: New instructions for the project.
            custom_categories: New categories for the project.
            retrieval_criteria: New retrieval criteria for the project.
            memory_depth: Memory depth for the project.
            usecase_setting: Usecase setting for the project.
            multilingual: Whether to use the input language for memory storage and retrieval.

        Returns:
            Dictionary containing the API response.

        Raises:
            ValueError: If org_id or project_id are not set, or no update fields provided.
        """
        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")

        kwargs = {
            **(options.model_dump(exclude_unset=True) if options else {}),
            **{
                k: v
                for k, v in {
                    "custom_instructions": custom_instructions,
                    "custom_categories": custom_categories,
                    "retrieval_criteria": retrieval_criteria,
                    "memory_depth": memory_depth,
                    "usecase_setting": usecase_setting,
                    "multilingual": multilingual,
                }.items()
                if v is not None
            },
        }

        if not kwargs:
            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(kwargs)
        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,
            {**kwargs, "sync_type": "async"},
        )
        return response.json()

    async def chat(self):
        """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()