Mem0 Strands

Copyfrom importlib.metadata import PackageNotFoundError, version
 
from mem0_strands.client import Mem0ServiceClient
from mem0_strands.store import Mem0MemoryStore
 
__all__ = [
    "Mem0MemoryStore",
    "Mem0ServiceClient",
]
 
try:
    __version__ = version("mem0-strands")
except PackageNotFoundError:  # pragma: no cover - only when running from a source tree
    __version__ = "0.0.0+unknown"

Copy@property
def client(self) -> Mem0ServiceClient:
    """The Mem0 service client, constructed lazily on first use.
 
    Note: constructing the underlying SDK client can block (the platform client
    validates the API key over HTTP; the OSS client builds embedders / vector
    stores), so first use is deferred and always happens inside a worker thread
    via :func:`asyncio.to_thread`, never on the event loop.
    """
    client_injected = self._client is not None
    if self._client is None:
        self._client = Mem0ServiceClient(api_key=self._api_key, host=self._host, config=self._config)
    if not self._recorded_init:
        self._recorded_init = True
        telemetry.record(
            "store.init",
            self._client,
            scopes=sorted(self.scope),
            writable=self.writable,
            extraction_enabled=bool(self.extraction),
            has_default_metadata=bool(self.metadata),
            max_search_results=self.max_search_results,
            host_overridden=bool(self._host),
            client_injected=client_injected,
        )
    return self._client

CopyDEFAULT_MAX_SEARCH_RESULTS = 5

Copyasync def search(self, query: str, options: SearchOptions | None = None) -> list[MemoryEntry]:
    """Search Mem0 for entries matching ``query``, ordered by relevance."""
    top_k = options.get("max_search_results") if options is not None else None
    if top_k is None:
        top_k = self.max_search_results
    if top_k is None:
        top_k = DEFAULT_MAX_SEARCH_RESULTS
 
    # ``self.client`` is resolved inside the thread so lazy construction (a
    # blocking call) does not run on the event loop.
    started = time.perf_counter()
    try:
        memories = await asyncio.to_thread(lambda: self.client.search_memories(query, self.scope, top_k))
    except Exception as exc:
        self._record("store.search", started, False, top_k=top_k, error_kind=telemetry.error_kind(exc))
        raise
    self._record("store.search", started, True, top_k=top_k, result_count=len(memories))
    return [self._to_entry(memory) for memory in memories]
 

Copydef __init__(
    self,
    api_key: str | None = None,
    host: str | None = None,
    config: dict[str, Any] | None = None,
    client: Any | None = None,
) -> None:
    """Initialize the Mem0 client.
 
    Args:
        api_key: Mem0 platform API key. Falls back to ``$MEM0_API_KEY``.
        host: Mem0 platform base URL. Defaults to the SDK default
            (``https://api.mem0.ai``).
        config: A Mem0 OSS config dict; when given, a self-hosted
            :class:`mem0.Memory` is built instead of the platform client.
        client: A pre-built Mem0 client to use directly (platform or OSS).
 
    Raises:
        ValueError: If ``client`` is an async Mem0 client (its coroutines
            would never be awaited off the worker thread).
    """
    if client is not None:
        if _is_async_client(client):
            raise ValueError(
                "Async Mem0 clients are not supported. Pass a synchronous "
                "mem0.MemoryClient (or a mem0.Memory / config): the store runs the "
                "SDK in a worker thread, so an async client's coroutines would "
                "never be awaited and every write would silently no-op."
            )
        self.mem0 = client
        self.is_platform = _is_platform_client(client)
        return
 
    if config is not None:
        try:
            from mem0 import Memory
        except ImportError as err:  # pragma: no cover - exercised via install docs
            raise ImportError(
                "The mem0ai package is required. Install it with: pip install 'mem0-strands'"
            ) from err
        self.mem0 = Memory.from_config(config)
        self.is_platform = False
        return
 
    try:
        from mem0 import MemoryClient
    except ImportError as err:  # pragma: no cover - exercised via install docs
        raise ImportError("The mem0ai package is required. Install it with: pip install 'mem0-strands'") from err
    api_key = api_key or os.environ.get("MEM0_API_KEY")
    # MemoryClient(host=None) would override the SDK default with None, so only
    # pass host when the caller actually set one.
    self.mem0 = MemoryClient(api_key=api_key, host=host) if host else MemoryClient(api_key=api_key)
    self.is_platform = True
 

Copydef _check_scope(self, scope: dict[str, str]) -> None:
    """Reject scope the selected backend cannot honor.
 
    ``app_id`` exists only on the platform; the OSS ``Memory`` API has no
    ``app_id`` parameter, so we fail loudly here rather than let it surface as
    a ``TypeError`` on ``add`` or silently miss on ``search``.
    """
    if not self.is_platform and "app_id" in scope:
        raise ValueError(
            "app_id is a Mem0 platform-only scope. The OSS backend supports "
            "user_id, agent_id, and run_id; drop app_id or use the platform client."
        )
 

Copy# Tags platform writes so Mem0's backend attributes the memory to this integration
# in telemetry (recognized values live in the backend's KNOWN_EVENT_SOURCES
# allowlist; unknown ones bucket into "OTHERS"). Platform only.
_SOURCE = "STRANDS"
 

Copydef _write_extras(self) -> dict[str, str]:
    """Extra kwargs attached to platform writes: the telemetry ``source`` tag."""
    return {"source": _SOURCE} if self.is_platform else {}

Copydef _extract_results(response: Any) -> list[dict[str, Any]]:
    """Normalize a Mem0 search response to a list of memory dicts.
 
    Mem0 returns ``{"results": [...]}`` (v1.1) or, on older paths, a bare list.
    """
    if isinstance(response, dict):
        results = response.get("results", [])
        return list(results) if isinstance(results, list) else []
    if isinstance(response, list):
        return response
    return []

Copy``Mem0MemoryStore`` implements both write sinks, which is what sets it apart from a
vector-DB-style store:
 
- :meth:`add` writes a single distilled fact verbatim (``infer=False``). This is
  the sink for the ``add_memory`` tool and for a client-side extractor.
- :meth:`add_messages` renders raw conversation turns to text and hands them to
  Mem0 for **server-side extraction** (``infer=True``). Because this sink exists,
  enabling ``extraction`` routes messages straight to Mem0's own extraction
  pipeline -- no extra client-side model call, and Mem0's de-duplication applies.
 

Copydef record(event: str, client: Mem0ServiceClient, **properties: Any) -> None:
    """Send one strands.* usage event. Never raises."""
    try:
        client_telemetry.capture_event(
            f"strands.{event}",
            {
                "source": SOURCE,
                "language": "python",
                "strands_store_version": _VERSION,
                "backend": "platform" if client.is_platform else "oss",
                **properties,
            },
            _distinct_id(client),
        )

CopyOpt out with MEM0_TELEMETRY=false.

Copyexcept Exception:
    pass