feat(mem0-plugin): support self-hosted Mem0 server via MEM0_BASE_URL

Adds an optional MEM0_BASE_URL override (env var, falling back to
settings.json's base_url, defaulting to the hosted Platform unchanged)
so the plugin's REST-backed hooks can talk to a self-hosted server/
deployment instead of api.mem0.ai.

A new scripts/_api.py centralizes the hosted-vs-self-hosted protocol
differences (unversioned paths, X-API-Key auth, agent_id instead of
app_id) that every hook previously hardcoded, and all 9 urllib call
sites now route through it.

MCP tools have no self-hosted backend (server/ implements no MCP), so
they continue to talk to the hosted Platform regardless of
MEM0_BASE_URL; this is documented rather than silently left unclear.
Coding-category customization also has no self-hosted SDK equivalent
and now skips cleanly with a log message instead of failing.

Closes MEM-5647

integrations/mem0-plugin/scripts/_api.py

@@ -0,0 +1,135 @@
+"""Translate memory API calls between the hosted Mem0 Platform and a self-hosted server.
+
+The hosted Platform speaks versioned /v1 and /v3 paths, `Authorization: Token`
+auth, and scopes writes/searches with a top-level app_id. The self-hosted OSS
+server (server/) speaks unversioned /memories and /search paths, `X-API-Key`
+auth, and has no app_id concept -- agent_id is the closest first-class
+scoping dimension there. Every helper below branches on resolve_base_url().
+"""
+
+from __future__ import annotations
+
+import json
+import urllib.error
+import urllib.request
+
+from _identity import DEFAULT_BASE_URL, resolve_base_url
+
+FETCH_TIMEOUT = 5
+SELF_HOSTED_COUNT_CAP = 100
+
+
+def is_self_hosted(base_url: str | None = None) -> bool:
+    base_url = base_url if base_url is not None else resolve_base_url()
+    return base_url.rstrip("/") != DEFAULT_BASE_URL
+
+
+def auth_headers(api_key: str) -> dict[str, str]:
+    if is_self_hosted():
+        return {"X-API-Key": api_key}
+    return {"Authorization": f"Token {api_key}"}
+
+
+def project_field() -> str:
+    """Body/filter key for project scoping: agent_id (self-hosted) or app_id (hosted Platform)."""
+    return "agent_id" if is_self_hosted() else "app_id"
+
+
+def add_url() -> str:
+    base_url = resolve_base_url()
+    if is_self_hosted(base_url):
+        return f"{base_url}/memories"
+    return f"{base_url}/v3/memories/add/"
+
+
+def search_url() -> str:
+    base_url = resolve_base_url()
+    if is_self_hosted(base_url):
+        return f"{base_url}/search"
+    return f"{base_url}/v3/memories/search/"
+
+
+def delete_url(memory_id: str) -> str:
+    base_url = resolve_base_url()
+    if is_self_hosted(base_url):
+        return f"{base_url}/memories/{memory_id}"
+    return f"{base_url}/v1/memories/{memory_id}/"
+
+
+def fetch_recent(api_key: str, user_id: str, project_id: str, top_k: int = 1, global_search: bool = False) -> list:
+    """Fetch recent memories for a scope. Returns [] on any error; never raises.
+
+    Hosted Platform lists via POST with an AND/OR filters body. The self-hosted
+    server has no such endpoint -- it only supports GET /memories with flat
+    user_id/agent_id/top_k query params, so global (cross-user) listing there
+    is best-effort and may return nothing if the caller lacks admin rights.
+    """
+    base_url = resolve_base_url()
+    headers = auth_headers(api_key)
+    try:
+        if is_self_hosted(base_url):
+            headers = dict(headers)
+            params = f"top_k={top_k}"
+            if not global_search:
+                params = f"user_id={user_id}&{project_field()}={project_id}&{params}"
+            req = urllib.request.Request(f"{base_url}/memories?{params}", headers=headers, method="GET")
+        else:
+            if global_search:
+                filters = {"OR": [{"user_id": "*"}]}
+            else:
+                filters = {"AND": [{"user_id": user_id}, {project_field(): project_id}]}
+            body = json.dumps({"filters": filters}).encode()
+            headers = dict(headers)
+            headers["Content-Type"] = "application/json"
+            req = urllib.request.Request(
+                f"{base_url}/v3/memories/?page=1&page_size={top_k}", data=body, headers=headers, method="POST"
+            )
+        with urllib.request.urlopen(req, timeout=FETCH_TIMEOUT) as r:
+            data = json.loads(r.read())
+        if isinstance(data, list):
+            return data
+        if isinstance(data, dict):
+            return data.get("results", [])
+        return []
+    except (urllib.error.URLError, OSError, json.JSONDecodeError, ValueError):
+        return []
+
+
+def count_memories(api_key: str, user_id: str, project_id: str, global_search: bool = False) -> str:
+    """Total memory count for a scope, as a string ("?" on any error).
+
+    Hosted Platform's list endpoint reports a true total in its `count` field
+    even with page_size=1. The self-hosted server has no such total, so its
+    count is a best-effort len() over a capped GET (accurate up to SELF_HOSTED_COUNT_CAP).
+    """
+    base_url = resolve_base_url()
+    headers = auth_headers(api_key)
+    try:
+        if is_self_hosted(base_url):
+            params = f"top_k={SELF_HOSTED_COUNT_CAP}"
+            if not global_search:
+                params = f"user_id={user_id}&{project_field()}={project_id}&{params}"
+            req = urllib.request.Request(f"{base_url}/memories?{params}", headers=headers, method="GET")
+            with urllib.request.urlopen(req, timeout=FETCH_TIMEOUT) as r:
+                data = json.loads(r.read())
+            results = data.get("results", []) if isinstance(data, dict) else data
+            return str(len(results)) if isinstance(results, list) else "?"
+
+        if global_search:
+            filters = {"OR": [{"user_id": "*"}]}
+        else:
+            filters = {"AND": [{"user_id": user_id}, {project_field(): project_id}]}
+        body = json.dumps({"filters": filters}).encode()
+        headers = {**headers, "Content-Type": "application/json"}
+        req = urllib.request.Request(
+            f"{base_url}/v3/memories/?page=1&page_size=1", data=body, headers=headers, method="POST"
+        )
+        with urllib.request.urlopen(req, timeout=FETCH_TIMEOUT) as r:
+            data = json.loads(r.read())
+        if isinstance(data, dict) and "count" in data:
+            return str(data["count"])
+        if isinstance(data, list):
+            return str(len(data))
+        return "?"
+    except (urllib.error.URLError, OSError, json.JSONDecodeError, ValueError):
+        return "?"

integrations/mem0-plugin/scripts/_identity.py

@@ -14,6 +14,11 @@ User ID resolution:
 
 Settings resolution:
   ~/.mem0/settings.json (user-editable, falls back to defaults)
+
+Base URL resolution (first non-empty wins):
+  1. MEM0_BASE_URL env var (explicit override, for self-hosted servers)
+  2. "base_url" in ~/.mem0/settings.json
+  3. DEFAULT_BASE_URL (the hosted Mem0 Platform)
 """
 
 from __future__ import annotations
@@ -75,6 +80,9 @@ def resolve_user_id() -> str:
     return os.environ.get("USER") or "default"
 
 
+DEFAULT_BASE_URL = "https://api.mem0.ai"
+
+
 def resolve_config() -> dict:
     """Resolve settings from ~/.mem0/settings.json (primary) with env var overrides."""
     try:
@@ -88,9 +96,21 @@ def resolve_config() -> dict:
             "retention_session_days": 90,
             "confidence_threshold": 0.3,
             "debug": False,
+            "base_url": "",
         }
 
 
+def resolve_base_url() -> str:
+    """Resolve the Mem0 API base URL: MEM0_BASE_URL env var, then settings.json, then the hosted platform."""
+    explicit = os.environ.get("MEM0_BASE_URL", "").strip()
+    if explicit:
+        return explicit.rstrip("/")
+    configured = str(resolve_config().get("base_url", "")).strip()
+    if configured:
+        return configured.rstrip("/")
+    return DEFAULT_BASE_URL
+
+
 try:
     from _project import resolve_branch, resolve_project_id, save_project_mapping
 except ImportError:

integrations/mem0-plugin/scripts/_identity.sh

@@ -83,5 +83,12 @@ else
 fi
 export MEM0_AUTO_SAVE MEM0_AUTO_SEARCH MEM0_SEARCH_LIMIT MEM0_RETENTION_SESSION_DAYS MEM0_CONFIDENCE_THRESHOLD MEM0_GLOBAL_SEARCH MEM0_DEBUG
 
+# Resolve base URL: MEM0_BASE_URL env var > settings.json "base_url" > hosted platform
+if [ -z "${MEM0_BASE_URL:-}" ] && command -v python3 >/dev/null 2>&1; then
+  MEM0_BASE_URL=$(PYTHONPATH="$_SCRIPT_DIR" python3 -c "from _identity import resolve_base_url; print(resolve_base_url())" 2>/dev/null || echo "https://api.mem0.ai")
+fi
+: "${MEM0_BASE_URL:=https://api.mem0.ai}"
+export MEM0_BASE_URL
+
 # Also resolve project context
 . "$_SCRIPT_DIR/_project.sh"

integrations/mem0-plugin/scripts/_search.py

@@ -1,7 +1,8 @@
 """Shared mem0 search API helper.
 
-Wraps POST /v3/memories/search/ into a single function call.
-All pre-fetch hooks use this instead of duplicating urllib boilerplate.
+Wraps the search endpoint (hosted Platform or self-hosted server) into a
+single function call. All pre-fetch hooks use this instead of duplicating
+urllib boilerplate.
 """
 
 from __future__ import annotations
@@ -10,7 +11,8 @@ import json
 import os
 import urllib.request
 
-SEARCH_URL = "https://api.mem0.ai/v3/memories/search/"
+from _api import auth_headers, project_field, search_url
+
 SEARCH_TIMEOUT = 5
 
 
@@ -34,12 +36,8 @@ def should_rerank() -> bool:
 
 def _do_search(api_key: str, payload: dict) -> list[dict]:
     body = json.dumps(payload).encode()
-    req = urllib.request.Request(
-        SEARCH_URL,
-        data=body,
-        headers={"Authorization": f"Token {api_key}", "Content-Type": "application/json"},
-        method="POST",
-    )
+    headers = {**auth_headers(api_key), "Content-Type": "application/json"}
+    req = urllib.request.Request(search_url(), data=body, headers=headers, method="POST")
     with urllib.request.urlopen(req, timeout=SEARCH_TIMEOUT) as r:
         data = json.loads(r.read())
         return data if isinstance(data, list) else data.get("results", [])
@@ -64,7 +62,7 @@ def search_memories(
     if global_search:
         filters: dict = {"OR": [{"user_id": "*"}]}
     else:
-        base_clauses: list[dict] = [{"user_id": user_id}, {"app_id": project_id}]
+        base_clauses: list[dict] = [{"user_id": user_id}, {project_field(): project_id}]
         if metadata_type:
             base_clauses.append({"metadata": {"type": metadata_type}})
         if metadata_filters:

integrations/mem0-plugin/scripts/auto_capture.py

@@ -20,6 +20,7 @@ import urllib.error
 import urllib.request
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from _api import add_url, auth_headers, project_field
 from _identity import resolve_api_key, resolve_user_id
 from _instructions import load_instructions
 from _project import resolve_branch, resolve_project_id
@@ -40,7 +41,6 @@ if os.environ.get("MEM0_DEBUG"):
     except OSError:
         pass
 
-API_URL = "https://api.mem0.ai"
 TAIL_LINES = 200
 MAX_CONTENT_CHARS = 8000
 MIN_CONTENT_CHARS = 100
@@ -123,7 +123,7 @@ def store_exchange(api_key: str, messages: list[dict], user_id: str,
     body = {
         "messages": messages,
         "user_id": user_id,
-        "app_id": project_id,
+        project_field(): project_id,
         "metadata": metadata,
         "infer": True,
     }
@@ -131,15 +131,8 @@ def store_exchange(api_key: str, messages: list[dict], user_id: str,
     body.update(load_instructions())
 
     data = json.dumps(body).encode("utf-8")
-    req = urllib.request.Request(
-        f"{API_URL}/v3/memories/add/",
-        data=data,
-        headers={
-            "Content-Type": "application/json",
-            "Authorization": f"Token {api_key}",
-        },
-        method="POST",
-    )
+    headers = {**auth_headers(api_key), "Content-Type": "application/json"}
+    req = urllib.request.Request(add_url(), data=data, headers=headers, method="POST")
     try:
         with urllib.request.urlopen(req, timeout=15) as resp:
             if resp.status in (200, 201):

integrations/mem0-plugin/scripts/auto_import.py

@@ -21,6 +21,7 @@ import urllib.error
 import urllib.request
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from _api import add_url, auth_headers, delete_url, project_field, search_url
 from _chunking import filter_and_truncate, split_by_headers
 from _identity import resolve_api_key, resolve_user_id
 from _project import resolve_branch, resolve_project_id, save_project_mapping
@@ -41,7 +42,6 @@ if os.environ.get("MEM0_DEBUG"):
     except OSError:
         pass
 
-API_URL = "https://api.mem0.ai"
 MAX_FILE_SIZE = 100_000  # skip files over 100 KB
 TARGET_FILES = ["CLAUDE.md", "AGENTS.md", ".cursorrules", ".windsurfrules", "mem0.md"]
 HASH_STORE = os.path.expanduser("~/.mem0/file_hashes.json")
@@ -127,19 +127,15 @@ def already_imported(api_key: str, user_id: str, project_id: str, filename: str)
         "filters": {
             "AND": [
                 {"user_id": user_id},
-                {"app_id": project_id},
+                {project_field(): project_id},
                 {"metadata": {"source": "auto-import"}},
             ]
         },
         "top_k": 10,
         "threshold": 0.0,
     }).encode()
-    req = urllib.request.Request(
-        f"{API_URL}/v3/memories/search/",
-        data=body,
-        headers={"Content-Type": "application/json", "Authorization": f"Token {api_key}"},
-        method="POST",
-    )
+    headers = {**auth_headers(api_key), "Content-Type": "application/json"}
+    req = urllib.request.Request(search_url(), data=body, headers=headers, method="POST")
     try:
         with urllib.request.urlopen(req, timeout=5) as r:
             data = json.loads(r.read())
@@ -161,19 +157,15 @@ def _delete_stale_chunks(api_key: str, user_id: str, project_id: str, filename:
         "filters": {
             "AND": [
                 {"user_id": user_id},
-                {"app_id": project_id},
+                {project_field(): project_id},
                 {"metadata": {"source": "auto-import"}},
             ]
         },
         "top_k": 20,
         "threshold": 0.0,
     }).encode()
-    req = urllib.request.Request(
-        f"{API_URL}/v3/memories/search/",
-        data=body,
-        headers={"Content-Type": "application/json", "Authorization": f"Token {api_key}"},
-        method="POST",
-    )
+    headers = {**auth_headers(api_key), "Content-Type": "application/json"}
+    req = urllib.request.Request(search_url(), data=body, headers=headers, method="POST")
     ids_to_delete = []
     try:
         with urllib.request.urlopen(req, timeout=10) as r:
@@ -196,8 +188,8 @@ def _delete_stale_chunks(api_key: str, user_id: str, project_id: str, filename:
     for mid in ids_to_delete:
         try:
             del_req = urllib.request.Request(
-                f"{API_URL}/v1/memories/{mid}/",
-                headers={"Authorization": f"Token {api_key}"},
+                delete_url(mid),
+                headers=auth_headers(api_key),
                 method="DELETE",
             )
             with urllib.request.urlopen(del_req, timeout=10):
@@ -227,21 +219,14 @@ def post_memory(api_key: str, content: str, user_id: str, filename: str, project
             }
         ],
         "user_id": user_id,
-        "app_id": project_id,
+        project_field(): project_id,
         "metadata": metadata,
         "infer": False,
     }
 
     data = json.dumps(body).encode("utf-8")
-    req = urllib.request.Request(
-        f"{API_URL}/v3/memories/add/",
-        data=data,
-        headers={
-            "Content-Type": "application/json",
-            "Authorization": f"Token {api_key}",
-        },
-        method="POST",
-    )
+    headers = {**auth_headers(api_key), "Content-Type": "application/json"}
+    req = urllib.request.Request(add_url(), data=data, headers=headers, method="POST")
 
     try:
         with urllib.request.urlopen(req, timeout=15) as resp:

integrations/mem0-plugin/scripts/auto_setup_categories.py

@@ -38,6 +38,7 @@ sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
 # Importing setup_coding_categories also injects the plugin venv's site-packages
 # onto sys.path (its module-level bootstrap), so ``from mem0 import MemoryClient``
 # works even when this script is run with the system python3.
+from _api import is_self_hosted  # noqa: E402
 from _identity import resolve_api_key  # noqa: E402
 from setup_coding_categories import CODING_CATEGORIES, _categories_match  # noqa: E402
 
@@ -189,6 +190,10 @@ def main() -> None:
         log.debug("MEM0_API_KEY not set, skipping coding-categories setup")
         return
 
+    if is_self_hosted():
+        log.debug("Self-hosted server detected; category customization requires the hosted Mem0 Platform, skipping")
+        return
+
     key_fp = apikey_fingerprint(api_key)
     cat_fp = categories_fingerprint()
 

integrations/mem0-plugin/scripts/capture_compact_summary.py

@@ -25,6 +25,7 @@ import urllib.request
 from datetime import date, timedelta
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from _api import add_url, auth_headers, project_field
 from _identity import resolve_api_key, resolve_user_id
 from _instructions import load_instructions
 from _project import resolve_branch, resolve_project_id
@@ -45,7 +46,6 @@ if os.environ.get("MEM0_DEBUG"):
     except OSError:
         pass
 
-API_URL = "https://api.mem0.ai"
 MAX_TAIL_LINES = 2000
 MAX_SUMMARY_CHARS = 50000
 # Compact summaries describe a single session's state -- stale after a quarter.
@@ -111,7 +111,7 @@ def store_summary(api_key: str, summary: str, user_id: str, session_id: str, pro
     body = {
         "messages": [{"role": "assistant", "content": summary}],
         "user_id": user_id,
-        "app_id": project_id,
+        project_field(): project_id,
         "metadata": metadata,
         "infer": True,
         "expiration_date": expires,
@@ -120,15 +120,8 @@ def store_summary(api_key: str, summary: str, user_id: str, session_id: str, pro
     body.update(load_instructions(cwd))
 
     data = json.dumps(body).encode("utf-8")
-    req = urllib.request.Request(
-        f"{API_URL}/v3/memories/add/",
-        data=data,
-        headers={
-            "Content-Type": "application/json",
-            "Authorization": f"Token {api_key}",
-        },
-        method="POST",
-    )
+    headers = {**auth_headers(api_key), "Content-Type": "application/json"}
+    req = urllib.request.Request(add_url(), data=data, headers=headers, method="POST")
     try:
         with urllib.request.urlopen(req, timeout=15) as resp:
             if resp.status in (200, 201):

integrations/mem0-plugin/scripts/capture_session_summary.py

@@ -23,6 +23,7 @@ import urllib.request
 from datetime import date, timedelta
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from _api import add_url, auth_headers, project_field
 from _identity import resolve_api_key, resolve_user_id
 from _instructions import load_instructions
 from _project import resolve_branch, resolve_project_id
@@ -43,7 +44,6 @@ if os.environ.get("MEM0_DEBUG"):
     except OSError:
         pass
 
-API_URL = "https://api.mem0.ai"
 MAX_TAIL_LINES = 3000
 MAX_SUMMARY_CHARS = 50000
 SUMMARY_EXPIRY_DAYS = 90
@@ -184,7 +184,7 @@ def store_summary(
     body = {
         "messages": [{"role": "assistant", "content": summary_prompt}],
         "user_id": user_id,
-        "app_id": project_id,
+        project_field(): project_id,
         "run_id": session_id,
         "metadata": metadata,
         "infer": True,
@@ -194,15 +194,8 @@ def store_summary(
     body.update(load_instructions(cwd))
 
     data = json.dumps(body).encode("utf-8")
-    req = urllib.request.Request(
-        f"{API_URL}/v3/memories/add/",
-        data=data,
-        headers={
-            "Content-Type": "application/json",
-            "Authorization": f"Token {api_key}",
-        },
-        method="POST",
-    )
+    headers = {**auth_headers(api_key), "Content-Type": "application/json"}
+    req = urllib.request.Request(add_url(), data=data, headers=headers, method="POST")
     try:
         with urllib.request.urlopen(req, timeout=15) as resp:
             if resp.status in (200, 201):

integrations/mem0-plugin/scripts/import_competing_tools.py

@@ -25,6 +25,7 @@ import urllib.error
 import urllib.request
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from _api import add_url, auth_headers, project_field
 from _chunking import (
     filter_and_truncate,
     split_by_headers,
@@ -33,7 +34,6 @@ from _chunking import (
 from _identity import resolve_api_key, resolve_user_id
 from _project import resolve_branch, resolve_project_id
 
-API_URL = "https://api.mem0.ai"
 HASH_STORE = os.path.expanduser("~/.mem0/import_hashes.json")
 
 
@@ -77,20 +77,13 @@ def post_memory(api_key: str, content: str, user_id: str, project_id: str, branc
     body = {
         "messages": [{"role": "user", "content": content}],
         "user_id": user_id,
-        "app_id": project_id,
+        project_field(): project_id,
         "metadata": metadata,
         "infer": False,
     }
     data = json.dumps(body).encode("utf-8")
-    req = urllib.request.Request(
-        f"{API_URL}/v3/memories/add/",
-        data=data,
-        headers={
-            "Content-Type": "application/json",
-            "Authorization": f"Token {api_key}",
-        },
-        method="POST",
-    )
+    headers = {**auth_headers(api_key), "Content-Type": "application/json"}
+    req = urllib.request.Request(add_url(), data=data, headers=headers, method="POST")
     try:
         with urllib.request.urlopen(req, timeout=20) as resp:
             return resp.status in (200, 201)

integrations/mem0-plugin/scripts/load_settings.py

@@ -18,6 +18,7 @@ DEFAULTS = {
     "confidence_threshold": 0.3,
     "global_search": False,
     "debug": False,
+    "base_url": "",
 }
 
 

integrations/mem0-plugin/scripts/on_pre_compact.py

@@ -23,6 +23,7 @@ import urllib.request
 from datetime import date, timedelta
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from _api import add_url, auth_headers, project_field
 from _identity import resolve_api_key, resolve_user_id
 from _project import resolve_branch, resolve_project_id
 
@@ -42,7 +43,6 @@ if os.environ.get("MEM0_DEBUG"):
     except OSError:
         pass
 
-API_URL = "https://api.mem0.ai"
 MAX_TAIL_LINES = 500
 MAX_USER_MESSAGES = 30
 MAX_BASH_COMMANDS = 20
@@ -173,22 +173,15 @@ def store_memory(api_key: str, content: str, user_id: str, source: str, session_
             {"role": "user", "content": content}
         ],
         "user_id": user_id,
-        "app_id": project_id,
+        project_field(): project_id,
         "metadata": metadata,
         "expiration_date": expires,
         "infer": True,
     }
 
     data = json.dumps(body).encode("utf-8")
-    req = urllib.request.Request(
-        f"{API_URL}/v3/memories/add/",
-        data=data,
-        headers={
-            "Content-Type": "application/json",
-            "Authorization": f"Token {api_key}",
-        },
-        method="POST",
-    )
+    headers = {**auth_headers(api_key), "Content-Type": "application/json"}
+    req = urllib.request.Request(add_url(), data=data, headers=headers, method="POST")
 
     try:
         with urllib.request.urlopen(req, timeout=15) as resp:

integrations/mem0-plugin/scripts/on_session_start.sh

@@ -71,37 +71,15 @@ fi
 
 MEM0_COUNT="?"
 if command -v python3 >/dev/null 2>&1; then
-  MEM0_COUNT=$(python3 -c "
-import json, os, urllib.request, urllib.error
+  MEM0_COUNT=$(PYTHONPATH="$SCRIPT_DIR" python3 -c "
+import os
+from _api import count_memories
+
 api_key = os.environ.get('MEM0_API_KEY', '')
 user_id = os.environ.get('MEM0_RESOLVED_USER_ID', 'default')
-app_id = os.environ.get('MEM0_PROJECT_ID', '')
+project_id = os.environ.get('MEM0_PROJECT_ID', '')
 global_search = os.environ.get('MEM0_GLOBAL_SEARCH', 'false') == 'true'
-
-def get_count(filters):
-    body = json.dumps({'filters': filters}).encode()
-    req = urllib.request.Request(
-        'https://api.mem0.ai/v3/memories/?page=1&page_size=1',
-        headers={'Authorization': f'Token {api_key}', 'Content-Type': 'application/json'},
-        data=body, method='POST',
-    )
-    with urllib.request.urlopen(req, timeout=5) as r:
-        data = json.loads(r.read())
-        if isinstance(data, dict) and 'count' in data:
-            return data['count']
-        if isinstance(data, list):
-            return len(data)
-    return 0
-
-try:
-    if global_search:
-        filters = {'OR': [{'user_id': '*'}]}
-    else:
-        filters = {'AND': [{'user_id': user_id}, {'app_id': app_id}]}
-    total = get_count(filters)
-    print(total)
-except Exception:
-    print('?')
+print(count_memories(api_key, user_id, project_id, global_search=global_search))
 " 2>/dev/null || echo "?")
 fi
 

integrations/mem0-plugin/scripts/session_timeline.py

@@ -11,51 +11,23 @@ Output: Compact timeline text to stdout (empty if nothing found)
 
 from __future__ import annotations
 
-import json
 import os
 import sys
-import urllib.request
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from _api import fetch_recent
 from _formatting import TYPE_ICONS, format_age
 from _identity import resolve_api_key, resolve_user_id
 from _project import resolve_project_id
 
-API_URL = "https://api.mem0.ai"
 MAX_RECENT = 10
 MAX_SUMMARIES = 3
-FETCH_TIMEOUT = 5
 
 
 def fetch_recent_memories(api_key: str, user_id: str, project_id: str) -> list[dict]:
-    """Fetch the most recent memories for this project via GET list endpoint."""
+    """Fetch the most recent memories for this project."""
     global_search = os.environ.get("MEM0_GLOBAL_SEARCH", "false") == "true"
-
-    if global_search:
-        filters = {"OR": [{"user_id": "*"}]}
-    else:
-        filters = {"AND": [{"user_id": user_id}, {"app_id": project_id}]}
-
-    body = json.dumps({"filters": filters}).encode()
-    req = urllib.request.Request(
-        f"{API_URL}/v3/memories/?page=1&page_size={MAX_RECENT}",
-        data=body,
-        headers={
-            "Authorization": f"Token {api_key}",
-            "Content-Type": "application/json",
-        },
-        method="POST",
-    )
-    try:
-        with urllib.request.urlopen(req, timeout=FETCH_TIMEOUT) as r:
-            result = json.loads(r.read())
-            if isinstance(result, dict) and "results" in result:
-                return result["results"][:MAX_RECENT]
-            if isinstance(result, list):
-                return result[:MAX_RECENT]
-            return []
-    except Exception:
-        return []
+    return fetch_recent(api_key, user_id, project_id, top_k=MAX_RECENT, global_search=global_search)[:MAX_RECENT]
 
 
 def format_timeline(memories: list[dict]) -> str:

integrations/mem0-plugin/scripts/setup_coding_categories.py

@@ -24,6 +24,7 @@ import sys
 
 _script_dir = os.path.dirname(os.path.abspath(__file__))
 sys.path.insert(0, _script_dir)
+from _api import is_self_hosted  # noqa: E402
 from _identity import resolve_api_key  # noqa: E402
 
 _plugin_root = os.environ.get("CLAUDE_PLUGIN_ROOT", os.path.join(_script_dir, ".."))
@@ -179,6 +180,16 @@ def main() -> int:
     if not api_key:
         print("ERROR: MEM0_API_KEY is not set. Export it or configure it via plugin userConfig.", file=sys.stderr)
         return 1
+
+    if is_self_hosted():
+        print(
+            "Custom category taxonomies require the hosted Mem0 Platform (client.project.update\n"
+            "has no self-hosted equivalent). Unset MEM0_BASE_URL to use the hosted platform, or\n"
+            "skip this step on a self-hosted server.",
+            file=sys.stderr,
+        )
+        return 1
+
     os.environ["MEM0_API_KEY"] = api_key
 
     try: