From 16c2cfe50edd776dad5a0ac16a8753a77bf16a3c Mon Sep 17 00:00:00 2001 From: kartik-mem0 Date: Thu, 13 Aug 2026 18:29:24 +0530 Subject: [PATCH] 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 --- docs/integrations/claude-code.mdx | 21 ++ integrations/mem0-plugin/README.md | 23 ++ integrations/mem0-plugin/scripts/_api.py | 135 ++++++++ integrations/mem0-plugin/scripts/_identity.py | 20 ++ integrations/mem0-plugin/scripts/_identity.sh | 7 + integrations/mem0-plugin/scripts/_search.py | 18 +- .../mem0-plugin/scripts/auto_capture.py | 15 +- .../mem0-plugin/scripts/auto_import.py | 39 +-- .../scripts/auto_setup_categories.py | 5 + .../scripts/capture_compact_summary.py | 15 +- .../scripts/capture_session_summary.py | 15 +- .../scripts/import_competing_tools.py | 15 +- .../mem0-plugin/scripts/load_settings.py | 1 + .../mem0-plugin/scripts/on_pre_compact.py | 15 +- .../mem0-plugin/scripts/on_session_start.sh | 34 +- .../mem0-plugin/scripts/session_timeline.py | 34 +- .../scripts/setup_coding_categories.py | 11 + .../mem0-plugin/tests/test_self_hosted.py | 307 ++++++++++++++++++ 18 files changed, 579 insertions(+), 151 deletions(-) create mode 100644 integrations/mem0-plugin/scripts/_api.py create mode 100644 integrations/mem0-plugin/tests/test_self_hosted.py diff --git a/docs/integrations/claude-code.mdx b/docs/integrations/claude-code.mdx index cf48915e3..e1acd1e0d 100644 --- a/docs/integrations/claude-code.mdx +++ b/docs/integrations/claude-code.mdx @@ -157,6 +157,27 @@ When installed via the plugin marketplace, Mem0 hooks into Claude Code's lifecyc What you type is stored as yours. What Claude produces (session summaries and compaction summaries) is stored as the assistant's, so its suggestions never become your stated preferences. +## Self-Hosted Mem0 Server (Optional) + +By default the plugin talks to the hosted Mem0 Platform (`https://api.mem0.ai`). To point the REST-backed hooks at your own [self-hosted Mem0 server](/open-source/features/rest-api) instead, set `MEM0_BASE_URL` before starting your session: + +```bash +export MEM0_BASE_URL="http://localhost:8000" # your server/ deployment +export MEM0_API_KEY="" +``` + +Or persist it in `~/.mem0/settings.json`: + +```json +{ "base_url": "http://localhost:8000" } +``` + +`MEM0_BASE_URL` takes precedence over the settings file; leaving both unset keeps the hosted Platform behavior unchanged. The plugin automatically switches the auth header (`X-API-Key` instead of `Authorization: Token`) and the project-scoping field (`agent_id` instead of `app_id`) to match the self-hosted server's REST API. + + + The self-hosted `server/` package doesn't implement MCP, so **MCP tools always connect to the hosted Platform** regardless of `MEM0_BASE_URL`. Only the REST-backed lifecycle hooks (session-start banner, auto-import, auto-capture, summaries, and `/mem0:` skill searches) route to your self-hosted server. Automatic coding-category setup also requires the hosted Platform and skips cleanly when `MEM0_BASE_URL` points elsewhere. + + ## Example Workflow ```text diff --git a/integrations/mem0-plugin/README.md b/integrations/mem0-plugin/README.md index 1481597cc..72c1b3b7e 100644 --- a/integrations/mem0-plugin/README.md +++ b/integrations/mem0-plugin/README.md @@ -48,6 +48,29 @@ Humans setting up Mem0 by hand should continue with Step 1 below. # Should print: m0-your-api-key ``` +## Self-hosted Mem0 server (optional) + +By default the plugin talks to the hosted Mem0 Platform (`https://api.mem0.ai`). To point the REST-backed hooks at your own [self-hosted Mem0 server](https://docs.mem0.ai/open-source/features/rest-api) instead, set `MEM0_BASE_URL` before starting your session: + +```bash +export MEM0_BASE_URL="http://localhost:8000" # your server/ deployment +export MEM0_API_KEY="" +``` + +Or persist it in `~/.mem0/settings.json`: + +```json +{ "base_url": "http://localhost:8000" } +``` + +`MEM0_BASE_URL` (env var) takes precedence over the settings file; leaving both unset keeps the hosted Platform behavior unchanged. + +**What works self-hosted:** every lifecycle hook that reads or writes memories over REST — the session-start banner and recent-activity timeline, auto-import of `CLAUDE.md`/`AGENTS.md`, auto-capture, pre-compact and session summaries, and `/mem0:` skill searches. The plugin automatically switches the auth header (`X-API-Key` instead of `Authorization: Token`) and the project-scoping field (`agent_id` instead of `app_id`, since the self-hosted server has no `app_id` concept) — no other config needed. + +**What doesn't work self-hosted (skips cleanly, no crash):** +- **MCP tools** (`add_memory`, `search_memories`, etc., and the `/mem0:` skills that call them through MCP) — the self-hosted `server/` package doesn't implement MCP, so the MCP server connection always talks to the hosted Platform regardless of `MEM0_BASE_URL`. Use the REST-backed hooks above for self-hosted memory capture and the `/mem0:` search skills instead. +- **Automatic coding-category setup** (`auto_setup_categories.py`, `setup_coding_categories.py`) — `project.update(custom_categories=...)` is a Platform-only SDK call with no self-hosted equivalent. Both scripts detect a self-hosted `MEM0_BASE_URL` and skip with a log message rather than failing. + ## Step 2: Install the plugin Choose one of the options below. All require `MEM0_API_KEY` to be set first (see above). diff --git a/integrations/mem0-plugin/scripts/_api.py b/integrations/mem0-plugin/scripts/_api.py new file mode 100644 index 000000000..841925457 --- /dev/null +++ b/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 "?" diff --git a/integrations/mem0-plugin/scripts/_identity.py b/integrations/mem0-plugin/scripts/_identity.py index d60ccb4a3..f5db35f6d 100644 --- a/integrations/mem0-plugin/scripts/_identity.py +++ b/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: diff --git a/integrations/mem0-plugin/scripts/_identity.sh b/integrations/mem0-plugin/scripts/_identity.sh index 4a25b8afc..eb714c0bb 100644 --- a/integrations/mem0-plugin/scripts/_identity.sh +++ b/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" diff --git a/integrations/mem0-plugin/scripts/_search.py b/integrations/mem0-plugin/scripts/_search.py index 5a19cfd41..157602283 100644 --- a/integrations/mem0-plugin/scripts/_search.py +++ b/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: diff --git a/integrations/mem0-plugin/scripts/auto_capture.py b/integrations/mem0-plugin/scripts/auto_capture.py index b48a928b4..1feef827f 100755 --- a/integrations/mem0-plugin/scripts/auto_capture.py +++ b/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): diff --git a/integrations/mem0-plugin/scripts/auto_import.py b/integrations/mem0-plugin/scripts/auto_import.py index 004772bbd..30410d6fb 100644 --- a/integrations/mem0-plugin/scripts/auto_import.py +++ b/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: diff --git a/integrations/mem0-plugin/scripts/auto_setup_categories.py b/integrations/mem0-plugin/scripts/auto_setup_categories.py index a94ed00a5..05b429fc4 100644 --- a/integrations/mem0-plugin/scripts/auto_setup_categories.py +++ b/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() diff --git a/integrations/mem0-plugin/scripts/capture_compact_summary.py b/integrations/mem0-plugin/scripts/capture_compact_summary.py index d2f7de0ba..9e5c6208d 100644 --- a/integrations/mem0-plugin/scripts/capture_compact_summary.py +++ b/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): diff --git a/integrations/mem0-plugin/scripts/capture_session_summary.py b/integrations/mem0-plugin/scripts/capture_session_summary.py index 6c3fa92fa..13f126792 100644 --- a/integrations/mem0-plugin/scripts/capture_session_summary.py +++ b/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): diff --git a/integrations/mem0-plugin/scripts/import_competing_tools.py b/integrations/mem0-plugin/scripts/import_competing_tools.py index 968bbbcd2..9c59c8fcf 100644 --- a/integrations/mem0-plugin/scripts/import_competing_tools.py +++ b/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) diff --git a/integrations/mem0-plugin/scripts/load_settings.py b/integrations/mem0-plugin/scripts/load_settings.py index 139d180c6..acaa09b9c 100644 --- a/integrations/mem0-plugin/scripts/load_settings.py +++ b/integrations/mem0-plugin/scripts/load_settings.py @@ -18,6 +18,7 @@ DEFAULTS = { "confidence_threshold": 0.3, "global_search": False, "debug": False, + "base_url": "", } diff --git a/integrations/mem0-plugin/scripts/on_pre_compact.py b/integrations/mem0-plugin/scripts/on_pre_compact.py index 882e26481..eb5930690 100755 --- a/integrations/mem0-plugin/scripts/on_pre_compact.py +++ b/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: diff --git a/integrations/mem0-plugin/scripts/on_session_start.sh b/integrations/mem0-plugin/scripts/on_session_start.sh index b2e83926b..fad20e96d 100755 --- a/integrations/mem0-plugin/scripts/on_session_start.sh +++ b/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 diff --git a/integrations/mem0-plugin/scripts/session_timeline.py b/integrations/mem0-plugin/scripts/session_timeline.py index 86e22f7fb..cdfe0241a 100644 --- a/integrations/mem0-plugin/scripts/session_timeline.py +++ b/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: diff --git a/integrations/mem0-plugin/scripts/setup_coding_categories.py b/integrations/mem0-plugin/scripts/setup_coding_categories.py index c3850adba..c8403e415 100644 --- a/integrations/mem0-plugin/scripts/setup_coding_categories.py +++ b/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: diff --git a/integrations/mem0-plugin/tests/test_self_hosted.py b/integrations/mem0-plugin/tests/test_self_hosted.py new file mode 100644 index 000000000..60094e084 --- /dev/null +++ b/integrations/mem0-plugin/tests/test_self_hosted.py @@ -0,0 +1,307 @@ +"""Tests for self-hosted Mem0 server support (MEM0_BASE_URL routing). + +Covers: +1. Default (no override) stays on the hosted Platform -- no behavior change. +2. An explicit override (env var or settings.json) routes to the self-hosted + server: different URLs, X-API-Key auth, agent_id instead of app_id. +3. A malformed/empty override never silently falls back to the hosted + Platform -- it is treated as self-hosted and fails closed instead of + leaking requests to api.mem0.ai. +""" + +from __future__ import annotations + +import json +from unittest.mock import MagicMock, patch + + +# --------------------------------------------------------------------------- # +# resolve_base_url() -- config plumbing # +# --------------------------------------------------------------------------- # +def test_resolve_base_url_defaults_to_hosted_platform(monkeypatch): + from _identity import DEFAULT_BASE_URL, resolve_base_url + + monkeypatch.delenv("MEM0_BASE_URL", raising=False) + assert resolve_base_url() == DEFAULT_BASE_URL == "https://api.mem0.ai" + + +def test_resolve_base_url_env_var_overrides(monkeypatch): + from _identity import resolve_base_url + + monkeypatch.setenv("MEM0_BASE_URL", "http://localhost:8000") + assert resolve_base_url() == "http://localhost:8000" + + +def test_resolve_base_url_strips_trailing_slash(monkeypatch): + from _identity import resolve_base_url + + monkeypatch.setenv("MEM0_BASE_URL", "http://localhost:8000/") + assert resolve_base_url() == "http://localhost:8000" + + +def test_resolve_base_url_falls_back_to_settings_file(monkeypatch, tmp_path): + import load_settings + from _identity import resolve_base_url + + monkeypatch.delenv("MEM0_BASE_URL", raising=False) + settings_path = tmp_path / "settings.json" + settings_path.write_text(json.dumps({"base_url": "http://self-hosted.local:8000"})) + monkeypatch.setattr(load_settings, "SETTINGS_PATH", settings_path) + + assert resolve_base_url() == "http://self-hosted.local:8000" + + +def test_resolve_base_url_env_var_takes_precedence_over_settings(monkeypatch, tmp_path): + import load_settings + from _identity import resolve_base_url + + settings_path = tmp_path / "settings.json" + settings_path.write_text(json.dumps({"base_url": "http://from-settings:8000"})) + monkeypatch.setattr(load_settings, "SETTINGS_PATH", settings_path) + monkeypatch.setenv("MEM0_BASE_URL", "http://from-env:9000") + + assert resolve_base_url() == "http://from-env:9000" + + +def test_resolve_base_url_empty_env_var_falls_through(monkeypatch, tmp_path): + """A blank MEM0_BASE_URL must not be treated as an explicit override.""" + import load_settings + from _identity import DEFAULT_BASE_URL, resolve_base_url + + monkeypatch.setenv("MEM0_BASE_URL", " ") + settings_path = tmp_path / "settings.json" + monkeypatch.setattr(load_settings, "SETTINGS_PATH", settings_path) + + assert resolve_base_url() == DEFAULT_BASE_URL + + +# --------------------------------------------------------------------------- # +# is_self_hosted() # +# --------------------------------------------------------------------------- # +def test_is_self_hosted_false_by_default(monkeypatch): + from _api import is_self_hosted + + monkeypatch.delenv("MEM0_BASE_URL", raising=False) + assert is_self_hosted() is False + + +def test_is_self_hosted_true_when_overridden(monkeypatch): + from _api import is_self_hosted + + monkeypatch.setenv("MEM0_BASE_URL", "http://localhost:8000") + assert is_self_hosted() is True + + +def test_is_self_hosted_malformed_url_treated_as_self_hosted(monkeypatch): + """A garbage override must not be silently coerced back to the hosted URL.""" + from _api import is_self_hosted + + monkeypatch.setenv("MEM0_BASE_URL", "not-a-url") + assert is_self_hosted() is True + + +# --------------------------------------------------------------------------- # +# auth_headers() / project_field() # +# --------------------------------------------------------------------------- # +def test_auth_headers_hosted_uses_token(monkeypatch): + from _api import auth_headers + + monkeypatch.delenv("MEM0_BASE_URL", raising=False) + assert auth_headers("m0-secret") == {"Authorization": "Token m0-secret"} + + +def test_auth_headers_self_hosted_uses_api_key_header(monkeypatch): + from _api import auth_headers + + monkeypatch.setenv("MEM0_BASE_URL", "http://localhost:8000") + assert auth_headers("m0sk-secret") == {"X-API-Key": "m0sk-secret"} + + +def test_project_field_hosted_is_app_id(monkeypatch): + from _api import project_field + + monkeypatch.delenv("MEM0_BASE_URL", raising=False) + assert project_field() == "app_id" + + +def test_project_field_self_hosted_is_agent_id(monkeypatch): + from _api import project_field + + monkeypatch.setenv("MEM0_BASE_URL", "http://localhost:8000") + assert project_field() == "agent_id" + + +# --------------------------------------------------------------------------- # +# URL builders # +# --------------------------------------------------------------------------- # +def test_add_url_hosted(monkeypatch): + from _api import add_url + + monkeypatch.delenv("MEM0_BASE_URL", raising=False) + assert add_url() == "https://api.mem0.ai/v3/memories/add/" + + +def test_add_url_self_hosted(monkeypatch): + from _api import add_url + + monkeypatch.setenv("MEM0_BASE_URL", "http://localhost:8000") + assert add_url() == "http://localhost:8000/memories" + + +def test_search_url_hosted(monkeypatch): + from _api import search_url + + monkeypatch.delenv("MEM0_BASE_URL", raising=False) + assert search_url() == "https://api.mem0.ai/v3/memories/search/" + + +def test_search_url_self_hosted(monkeypatch): + from _api import search_url + + monkeypatch.setenv("MEM0_BASE_URL", "http://localhost:8000") + assert search_url() == "http://localhost:8000/search" + + +def test_delete_url_hosted(monkeypatch): + from _api import delete_url + + monkeypatch.delenv("MEM0_BASE_URL", raising=False) + assert delete_url("mem-123") == "https://api.mem0.ai/v1/memories/mem-123/" + + +def test_delete_url_self_hosted(monkeypatch): + from _api import delete_url + + monkeypatch.setenv("MEM0_BASE_URL", "http://localhost:8000") + assert delete_url("mem-123") == "http://localhost:8000/memories/mem-123" + + +def test_search_url_malformed_override_does_not_leak_to_hosted(monkeypatch): + """A garbage override must build a URL against itself, never api.mem0.ai.""" + from _api import search_url + + monkeypatch.setenv("MEM0_BASE_URL", "not-a-url") + url = search_url() + assert "api.mem0.ai" not in url + assert url == "not-a-url/search" + + +# --------------------------------------------------------------------------- # +# End-to-end: write path (auto_import.post_memory) # +# --------------------------------------------------------------------------- # +def test_auto_import_post_memory_hosted_unchanged(monkeypatch): + """Default (no override) still posts to the hosted Platform with app_id.""" + from auto_import import post_memory + + monkeypatch.delenv("MEM0_BASE_URL", raising=False) + captured = {} + + def mock_urlopen(req, timeout=None): + captured["url"] = req.full_url + captured["headers"] = req.headers + captured.update(json.loads(req.data.decode("utf-8"))) + resp = MagicMock() + resp.status = 200 + resp.__enter__ = lambda s: s + resp.__exit__ = MagicMock(return_value=False) + return resp + + with patch("urllib.request.urlopen", side_effect=mock_urlopen): + result = post_memory("test-key", "content", "user", "CLAUDE.md", "my-project", "main") + + assert result is True + assert captured["url"] == "https://api.mem0.ai/v3/memories/add/" + assert captured["headers"]["Authorization"] == "Token test-key" + assert captured["app_id"] == "my-project" + assert "agent_id" not in captured + + +def test_auto_import_post_memory_self_hosted_routes_correctly(monkeypatch): + """MEM0_BASE_URL override posts to the self-hosted server with agent_id.""" + from auto_import import post_memory + + monkeypatch.setenv("MEM0_BASE_URL", "http://localhost:8000") + captured = {} + + def mock_urlopen(req, timeout=None): + captured["url"] = req.full_url + captured["headers"] = req.headers + captured.update(json.loads(req.data.decode("utf-8"))) + resp = MagicMock() + resp.status = 200 + resp.__enter__ = lambda s: s + resp.__exit__ = MagicMock(return_value=False) + return resp + + with patch("urllib.request.urlopen", side_effect=mock_urlopen): + result = post_memory("m0sk-key", "content", "user", "CLAUDE.md", "my-project", "main") + + assert result is True + assert captured["url"] == "http://localhost:8000/memories" + assert captured["headers"]["X-api-key"] == "m0sk-key" + assert captured["agent_id"] == "my-project" + assert "app_id" not in captured + + +# --------------------------------------------------------------------------- # +# End-to-end: search path (_search.search_memories) # +# --------------------------------------------------------------------------- # +def test_search_memories_self_hosted_routes_correctly(monkeypatch): + from _search import search_memories + + monkeypatch.setenv("MEM0_BASE_URL", "http://localhost:8000") + captured = {} + + def mock_urlopen(req, timeout=None): + captured["url"] = req.full_url + captured["headers"] = req.headers + captured.update(json.loads(req.data.decode("utf-8"))) + resp = MagicMock() + resp.read.return_value = json.dumps({"results": []}).encode() + resp.__enter__ = lambda s: s + resp.__exit__ = MagicMock(return_value=False) + return resp + + with patch("urllib.request.urlopen", side_effect=mock_urlopen): + search_memories("m0sk-key", "user", "proj", "query") + + assert captured["url"] == "http://localhost:8000/search" + assert captured["headers"]["X-api-key"] == "m0sk-key" + filters = captured["filters"] + assert {"agent_id": "proj"} in filters["AND"] + assert {"app_id": "proj"} not in filters["AND"] + + +def test_search_memories_malformed_base_url_fails_closed_not_hosted(monkeypatch): + """A garbage override must fail (return []) rather than leak to api.mem0.ai. + + No urlopen mock here on purpose: urllib rejects the schemeless URL before + any network I/O happens, proving the request never reaches a real host -- + hosted or otherwise. search_memories' broad except turns that into []. + """ + from _search import search_memories + + monkeypatch.setenv("MEM0_BASE_URL", "not-a-url") + + with patch("urllib.request.urlopen") as mock_urlopen: + results = search_memories("key", "user", "proj", "query") + mock_urlopen.assert_not_called() + + assert results == [] + + +# --------------------------------------------------------------------------- # +# Category-taxonomy scripts: no self-hosted equivalent, must skip cleanly # +# --------------------------------------------------------------------------- # +def test_auto_setup_categories_skips_when_self_hosted(monkeypatch): + import auto_setup_categories as asc + + monkeypatch.setattr(asc, "resolve_api_key", lambda: "m0sk-key") + monkeypatch.setenv("MEM0_BASE_URL", "http://localhost:8000") + + calls = [] + monkeypatch.setattr(asc, "make_client", lambda: calls.append("make_client")) + + asc.main() + + assert calls == []