mem0/mem0/client/main.py

import hashlib
import logging
import os
import uuid
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"})

# One collection for every generation; the operation is a body field.
PROFILE_JOBS_PATH = "/v2/profiles/jobs/"
PROFILE_SETTINGS_PATH = "/v2/profiles/settings/"

# Distinguishes an omitted argument from an explicit ``None`` that clears a field.
_UNSET: Any = object()


def _profile_settings_payload(
    enabled: Optional[bool],
    schema: Any = _UNSET,
    custom_instructions: Any = _UNSET,
) -> Dict[str, Any]:
    """Build the settings body the API accepts.

    ``schema`` and ``custom_instructions`` are per user and nest under
    ``entities``; only ``enabled`` is project-wide. This mirrors what
    ``get_profile_settings`` returns, so the two round-trip.

    Sending them flat is rejected with ``Unsupported settings``, so this shape is
    not cosmetic. ``_UNSET`` leaves a field unchanged; an explicit ``None`` clears it.
    """

    payload: Dict[str, Any] = {}
    if enabled is not None:
        payload["enabled"] = enabled

    entity_settings: Dict[str, Any] = {}
    if schema is not _UNSET:
        entity_settings["schema"] = schema
    if custom_instructions is not _UNSET:
        entity_settings["custom_instructions"] = custom_instructions

    if entity_settings:
        payload["entities"] = {"user": entity_settings}
    return payload


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)


def _sdk_version() -> str:
    """Resolved here rather than imported from the package root, which would cycle."""
    try:
        import importlib.metadata

        return importlib.metadata.version("mem0ai")
    except Exception:
        return "unknown"


def _apply_client_headers(client: Any, api_key: str, user_id: str) -> None:
    """Merge our headers into a caller-supplied client without erasing theirs.

    A wrapper may hand us a client already carrying its own X-Mem0-Source or a
    partial X-Mem0-Client stack. Blanket update() replaced both, which is the
    opposite of the set-once / append-only contract: the outermost layer is the
    one whose identity should survive.
    """
    existing = client.headers
    mine = _client_headers(api_key, user_id)

    outer_stack = existing.get("X-Mem0-Client")
    if outer_stack:
        entries = [part.strip() for part in str(outer_stack).split(",") if part.strip()]
        mine["X-Mem0-Client"] = _bounded_stack(entries, f"mem0-python/{_sdk_version()}")

    for name, value in mine.items():
        if name in ("X-Mem0-Source", "X-Application") and existing.get(name):
            continue
        existing[name] = value


MAX_STACK_ENTRIES = 4
MAX_STACK_CHARS = 200


def _bounded_stack(caller_entries, own: str) -> str:
    """Append our own entry and bound the result, dropping WHOLE entries.

    Two rules, and the second is the one that was wrong. Neither cap cuts
    characters: a blunt slice severs an identifier and leaves a fragment that
    parses as a real client name. And the reserved slot is OURS. Appending first
    and then trimming to four dropped exactly the entry this function exists to
    add, every time a caller already sent four, so the SDK vanished from its own
    stack while the caller's claims all survived.
    """
    kept = []
    budget = MAX_STACK_CHARS - len(own)
    for entry in list(caller_entries)[: MAX_STACK_ENTRIES - 1]:
        cost = len(entry) + len(", ")
        if cost > budget:
            break
        budget -= cost
        kept.append(entry)
    return ", ".join(kept + [own])


def _client_headers(api_key: str, user_id: str) -> Dict[str, str]:
    """Auth plus surface-identity headers.

    X-Mem0-Source and X-Application are SET-ONCE by contract: whichever layer is
    outermost sets them, and nothing below overwrites. A plugin or harness that
    wraps this SDK therefore keeps its own identity — it declares via MEM0_SOURCE
    / MEM0_APPLICATION and the SDK defers.

    X-Mem0-Client is APPEND-ONLY: every layer adds itself, so the platform sees
    the whole stack rather than only whoever spoke last.
    """
    headers = {
        "Authorization": f"Token {api_key}",
        "Mem0-User-ID": user_id,
        "X-Mem0-Client": _client_stack(),
    }
    source = os.getenv("MEM0_SOURCE", "").strip()
    if source:
        headers["X-Mem0-Source"] = source
    application = os.getenv("MEM0_APPLICATION", "").strip()
    if application:
        headers["X-Application"] = application
    return headers


def _client_stack() -> str:
    """This SDK appended to any stack an outer layer already declared."""
    existing = os.getenv("MEM0_CLIENT_STACK", "").strip()
    entries = [part.strip() for part in existing.split(",") if part.strip()] if existing else []
    return _bounded_stack(entries, f"mem0-python/{_sdk_version()}")


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)
            _apply_client_headers(self.client, self.api_key, self.user_id)
        else:
            self.client = httpx.Client(
                base_url=self.host,
                headers=_client_headers(self.api_key, 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)

        query_params = {key: params.pop(key) for key in ("page", "page_size") if key in params}
        if query_params:
            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_profile(self, entity_id: str) -> Dict[str, Any]:
        """Get the memory profile for a single user.

        Branch on ``status``, not on an empty ``profile``: generation is
        asynchronous, so a known user without a profile yet is a normal response.

        Args:
            entity_id: The user's id, as you supplied it on ``add`` (e.g. "alice").

        Returns:
            Dict with ``profile``, ``status``, ``entity_type``, ``entity_id``,
            ``updated_at`` and ``generation_count``. ``status`` is one of
            "succeeded", "pending", "failed", "not_enabled" or "insufficient_data".

        Raises:
            AuthenticationError: If authentication fails.
            NotFoundError: If no such user exists in the project.
        """

        response = self.client.get(f"/v2/entities/user/{_encode_path_segment(entity_id)}/profile/")
        response.raise_for_status()
        capture_client_event("client.get_profile", self, {"sync_type": "sync"})
        return response.json()

    @api_error_handler
    def generate_profile(self, entity_id: str, idempotency_key: Optional[str] = None) -> Dict[str, Any]:
        """Generate or refresh the profile for a single user, now.

        Profiles are otherwise built once a user crosses an internal message
        threshold, so a new user has none for its first few memories. Returns as
        soon as the work is queued: poll :meth:`get_profile` and branch on ``status``.

        Args:
            entity_id: The user's id, as you supplied it on ``add``.
            idempotency_key: Optional key that makes the create idempotent. Reuse
                the same value to safely retry a lost request without starting
                (and being billed for) a second job. A fresh key is generated when
                omitted.

        Returns:
            Dict containing ``job_id``, ``status``, ``status_url``, ``operation``,
            ``entity_type``, ``entity_count_reserved``, ``event_id`` and
            ``replayed``. Poll :meth:`get_profile` and branch on ``status``.

        Raises:
            ValidationError: If profiles are not enabled and configured for the
                project.
            NotFoundError: If no such user exists in the project.
        """

        response = self.client.post(
            PROFILE_JOBS_PATH,
            json={"operation": "trigger", "entity_type": "user", "entity_id": entity_id},
            headers={"Idempotency-Key": idempotency_key or uuid.uuid4().hex},
        )
        response.raise_for_status()
        capture_client_event("client.generate_profile", self, {"sync_type": "sync"})
        return response.json()

    @api_error_handler
    def get_profile_settings(self) -> Dict[str, Any]:
        """Get the profile settings for the current project.

        Returns:
            Dict with ``enabled`` and ``capabilities`` at the top level, and
            ``entities`` holding each entity type's ``schema`` and
            ``custom_instructions``.
        """

        response = self.client.get(PROFILE_SETTINGS_PATH)
        response.raise_for_status()
        capture_client_event("client.get_profile_settings", self, {"sync_type": "sync"})
        return response.json()

    @api_error_handler
    def update_profile_settings(
        self,
        enabled: Optional[bool] = None,
        schema: Any = _UNSET,
        custom_instructions: Any = _UNSET,
    ) -> Dict[str, Any]:
        """Update the profile settings for the current project.

        Only the arguments you pass are written.

        Args:
            enabled: Turn profile generation on or off. Project-wide.
            schema: JSON Schema for the profile. Every property needs a
                ``description``. Pass ``None`` to clear it; omit to leave it
                unchanged. Applies to user profiles.
            custom_instructions: Extra guidance for the extraction step. Pass
                ``None`` to clear it; omit to leave it unchanged. Applies to user
                profiles.

        Returns:
            Dict with the settings as stored after the update, in the same
            shape :meth:`get_profile_settings` returns.

        Raises:
            ValidationError: If the schema is not a valid profile schema.
        """

        payload = _profile_settings_payload(enabled, schema, custom_instructions)
        response = self.client.post(PROFILE_SETTINGS_PATH, json=payload)
        response.raise_for_status()
        capture_client_event(
            "client.update_profile_settings",
            self,
            {"keys": list(payload.keys()), "sync_type": "sync"},
        )
        return response.json()

    @api_error_handler
    def sample_profiles(self, limit: Optional[int] = None, idempotency_key: Optional[str] = None) -> Dict[str, Any]:
        """Generate profiles for a few real users, to check a schema.

        Real generations against real memories, and the results are kept. The
        profiles are written to those users and count toward usage.

        Args:
            limit: How many users to sample, 1-10. Defaults to the server value.
            idempotency_key: Optional key that makes the create idempotent. Reuse
                the same value to safely retry without starting a second sample
                run. A fresh key is generated when omitted.

        Returns:
            Dict containing ``job_id``, ``status``, ``status_url``, ``sampled``
            and the ``entity_ids`` that were picked. Poll
            :meth:`get_profile_job` with ``status_url`` until the status is terminal.

        Raises:
            ValidationError: If profiles are not enabled and configured.
            RateLimitError: If a sample run was already started very recently.
        """

        payload = self._prepare_params({"limit": limit})
        payload["operation"] = "sample"
        payload["entity_type"] = "user"
        response = self.client.post(
            PROFILE_JOBS_PATH,
            json=payload,
            headers={"Idempotency-Key": idempotency_key or uuid.uuid4().hex},
        )
        response.raise_for_status()
        capture_client_event("client.sample_profiles", self, {"sync_type": "sync"})
        return response.json()

    @api_error_handler
    def get_profile_job(self, job_id_or_status_url: str) -> Dict[str, Any]:
        """Read one generation job.

        Accepts the ``status_url`` from a create call, or a bare job id. Prefer
        passing ``status_url`` so a route change needs no client update.

        Returns:
            Dict whose ``job`` key holds the job: ``status``, ``total``,
            ``completed``, ``succeeded``, ``failed`` and ``skipped``. ``total`` is
            null until ``enumeration_complete``.
        """

        path = job_id_or_status_url
        if not path.startswith("/"):
            path = f"{PROFILE_JOBS_PATH}{path}/"
        response = self.client.get(path)
        response.raise_for_status()
        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,
        memory_depth: Optional[str] = None,
        usecase_setting: Optional[str] = None,
        multilingual: Optional[bool] = None,
        agent_custom_instructions: Optional[str] = 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.
            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.
            agent_custom_instructions: New extraction instructions for agent-scoped memories.

        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,
                    "memory_depth": memory_depth,
                    "usecase_setting": usecase_setting,
                    "multilingual": multilingual,
                    "agent_custom_instructions": agent_custom_instructions,
                }.items()
                if v is not None
            },
        }

        if not kwargs:
            raise ValueError(
                "Currently we only support updating custom_instructions or "
                "custom_categories, 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)
            _apply_client_headers(self.async_client, self.api_key, self.user_id)
        else:
            self.async_client = httpx.AsyncClient(
                base_url=self.host,
                headers=_client_headers(self.api_key, 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=_client_headers(self.api_key, 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)

        query_params = {key: params.pop(key) for key in ("page", "page_size") if key in params}
        if query_params:
            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_profile(self, entity_id: str) -> Dict[str, Any]:
        """Get the memory profile for a single user.

        Branch on ``status``, not on an empty ``profile``: generation is
        asynchronous, so a known user without a profile yet is a normal response.

        Args:
            entity_id: The user's id, as you supplied it on ``add`` (e.g. "alice").

        Returns:
            Dict with ``profile``, ``status``, ``entity_type``, ``entity_id``,
            ``updated_at`` and ``generation_count``. ``status`` is one of
            "succeeded", "pending", "failed", "not_enabled" or "insufficient_data".

        Raises:
            AuthenticationError: If authentication fails.
            NotFoundError: If no such user exists in the project.
        """

        response = await self.async_client.get(f"/v2/entities/user/{_encode_path_segment(entity_id)}/profile/")
        response.raise_for_status()
        capture_client_event("client.get_profile", self, {"sync_type": "async"})
        return response.json()

    @api_error_handler
    async def generate_profile(self, entity_id: str, idempotency_key: Optional[str] = None) -> Dict[str, Any]:
        """Generate or refresh the profile for a single user, now.

        Profiles are otherwise built once a user crosses an internal message
        threshold, so a new user has none for its first few memories. Returns as
        soon as the work is queued: poll :meth:`get_profile` and branch on ``status``.

        Args:
            entity_id: The user's id, as you supplied it on ``add``.
            idempotency_key: Optional key that makes the create idempotent. Reuse
                the same value to safely retry a lost request without starting
                (and being billed for) a second job. A fresh key is generated when
                omitted.

        Returns:
            Dict containing ``job_id``, ``status``, ``status_url``, ``operation``,
            ``entity_type``, ``entity_count_reserved``, ``event_id`` and
            ``replayed``. Poll :meth:`get_profile` and branch on ``status``.

        Raises:
            ValidationError: If profiles are not enabled and configured for the
                project.
            NotFoundError: If no such user exists in the project.
        """

        response = await self.async_client.post(
            PROFILE_JOBS_PATH,
            json={"operation": "trigger", "entity_type": "user", "entity_id": entity_id},
            headers={"Idempotency-Key": idempotency_key or uuid.uuid4().hex},
        )
        response.raise_for_status()
        capture_client_event("client.generate_profile", self, {"sync_type": "async"})
        return response.json()

    @api_error_handler
    async def get_profile_settings(self) -> Dict[str, Any]:
        """Get the profile settings for the current project.

        Returns:
            Dict with ``enabled`` and ``capabilities`` at the top level, and
            ``entities`` holding each entity type's ``schema`` and
            ``custom_instructions``.
        """

        response = await self.async_client.get(PROFILE_SETTINGS_PATH)
        response.raise_for_status()
        capture_client_event("client.get_profile_settings", self, {"sync_type": "async"})
        return response.json()

    @api_error_handler
    async def update_profile_settings(
        self,
        enabled: Optional[bool] = None,
        schema: Any = _UNSET,
        custom_instructions: Any = _UNSET,
    ) -> Dict[str, Any]:
        """Update the profile settings for the current project.

        Only the arguments you pass are written.

        Args:
            enabled: Turn profile generation on or off. Project-wide.
            schema: JSON Schema for the profile. Every property needs a
                ``description``. Pass ``None`` to clear it; omit to leave it
                unchanged. Applies to user profiles.
            custom_instructions: Extra guidance for the extraction step. Pass
                ``None`` to clear it; omit to leave it unchanged. Applies to user
                profiles.

        Returns:
            Dict with the settings as stored after the update, in the same
            shape :meth:`get_profile_settings` returns.

        Raises:
            ValidationError: If the schema is not a valid profile schema.
        """

        payload = _profile_settings_payload(enabled, schema, custom_instructions)
        response = await self.async_client.post(PROFILE_SETTINGS_PATH, json=payload)
        response.raise_for_status()
        capture_client_event(
            "client.update_profile_settings",
            self,
            {"keys": list(payload.keys()), "sync_type": "async"},
        )
        return response.json()

    @api_error_handler
    async def sample_profiles(
        self, limit: Optional[int] = None, idempotency_key: Optional[str] = None
    ) -> Dict[str, Any]:
        """Generate profiles for a few real users, to check a schema.

        Real generations against real memories, and the results are kept. The
        profiles are written to those users and count toward usage.

        Args:
            limit: How many users to sample, 1-10. Defaults to the server value.
            idempotency_key: Optional key that makes the create idempotent. Reuse
                the same value to safely retry without starting a second sample
                run. A fresh key is generated when omitted.

        Returns:
            Dict containing ``job_id``, ``status``, ``status_url``, ``sampled``
            and the ``entity_ids`` that were picked. Poll
            :meth:`get_profile_job` with ``status_url`` until the status is terminal.

        Raises:
            ValidationError: If profiles are not enabled and configured.
            RateLimitError: If a sample run was already started very recently.
        """

        payload = self._prepare_params({"limit": limit})
        payload["operation"] = "sample"
        payload["entity_type"] = "user"
        response = await self.async_client.post(
            PROFILE_JOBS_PATH,
            json=payload,
            headers={"Idempotency-Key": idempotency_key or uuid.uuid4().hex},
        )
        response.raise_for_status()
        capture_client_event("client.sample_profiles", self, {"sync_type": "async"})
        return response.json()

    @api_error_handler
    async def get_profile_job(self, job_id_or_status_url: str) -> Dict[str, Any]:
        """Read one generation job.

        Accepts the ``status_url`` from a create call, or a bare job id. Prefer
        passing ``status_url`` so a route change needs no client update.

        Returns:
            Dict whose ``job`` key holds the job: ``status``, ``total``,
            ``completed``, ``succeeded``, ``failed`` and ``skipped``. ``total`` is
            null until ``enumeration_complete``.
        """

        path = job_id_or_status_url
        if not path.startswith("/"):
            path = f"{PROFILE_JOBS_PATH}{path}/"
        response = await self.async_client.get(path)
        response.raise_for_status()
        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,
        memory_depth: Optional[str] = None,
        usecase_setting: Optional[str] = None,
        multilingual: Optional[bool] = None,
        agent_custom_instructions: Optional[str] = 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.
            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.
            agent_custom_instructions: New extraction instructions for agent-scoped memories.

        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,
                    "memory_depth": memory_depth,
                    "usecase_setting": usecase_setting,
                    "multilingual": multilingual,
                    "agent_custom_instructions": agent_custom_instructions,
                }.items()
                if v is not None
            },
        }

        if not kwargs:
            raise ValueError(
                "Currently we only support updating custom_instructions or "
                "custom_categories, 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()