diff --git a/.github/workflows/sync-upstream.yml b/.github/workflows/sync-upstream.yml
new file mode 100644
index 0000000000..71aca772b0
--- /dev/null
+++ b/.github/workflows/sync-upstream.yml
@@ -0,0 +1,80 @@
+name: Sync Upstream
+
+on:
+  workflow_dispatch:
+  schedule:
+    - cron: "17 * * * *"
+
+permissions:
+  contents: write
+  pull-requests: write
+
+jobs:
+  sync:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout fork branch
+        uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: Add upstream remote
+        run: |
+          git remote add upstream https://github.com/mem0ai/mem0.git
+          git fetch upstream main
+
+      - name: Check for upstream changes
+        id: changes
+        run: |
+          if [ "$(git rev-list --count origin/main..upstream/main)" -eq 0 ]; then
+            echo "changed=false" >> "$GITHUB_OUTPUT"
+          else
+            echo "changed=true" >> "$GITHUB_OUTPUT"
+          fi
+
+      - name: Stop when upstream is unchanged
+        if: steps.changes.outputs.changed != 'true'
+        run: echo "Upstream main is already in sync."
+
+      - name: Create sync branch
+        if: steps.changes.outputs.changed == 'true'
+        run: |
+          git checkout -B sync/upstream origin/main
+
+      - name: Merge upstream into sync branch
+        if: steps.changes.outputs.changed == 'true'
+        run: |
+          git config user.name "github-actions[bot]"
+          git config user.email "github-actions[bot]@users.noreply.github.com"
+          git merge --no-edit upstream/main
+
+      - name: Push sync branch
+        if: steps.changes.outputs.changed == 'true'
+        run: |
+          git push --force-with-lease origin sync/upstream
+
+      - name: Create or update PR
+        if: steps.changes.outputs.changed == 'true'
+        id: pr
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: |
+          PR_NUMBER="$(gh pr list --head sync/upstream --base main --state open --json number --jq '.[0].number')"
+          if [ -n "$PR_NUMBER" ]; then
+            echo "PR already exists: #$PR_NUMBER"
+          else
+            gh pr create \
+              --base main \
+              --head sync/upstream \
+              --title "chore: sync upstream mem0" \
+              --body "Automated sync from mem0ai/mem0 main into this fork."
+            PR_NUMBER="$(gh pr list --head sync/upstream --base main --state open --json number --jq '.[0].number')"
+          fi
+          echo "pr_number=$PR_NUMBER" >> "$GITHUB_OUTPUT"
+
+      - name: Enable auto-merge
+        if: steps.changes.outputs.changed == 'true'
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: |
+          gh pr merge "${{ steps.pr.outputs.pr_number }}" --auto --squash
diff --git a/integrations/mem0-plugin/scripts/_api.py b/integrations/mem0-plugin/scripts/_api.py
new file mode 100644
index 0000000000..8538451b9d
--- /dev/null
+++ b/integrations/mem0-plugin/scripts/_api.py
@@ -0,0 +1,134 @@
+"""Mem0 REST compatibility helpers for plugin hooks.
+
+Cloud remains the default. Set ``MEM0_API_MODE=self_hosted`` and
+``MEM0_API_URL`` to target the self-hosted FastAPI server.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import urllib.parse
+import urllib.request
+from typing import Any
+
+CLOUD_API_URL = "https://api.mem0.ai"
+
+
+def api_mode() -> str:
+    raw = os.environ.get("MEM0_API_MODE", "cloud").strip().lower().replace("-", "_")
+    return "self_hosted" if raw in {"self_hosted", "selfhosted", "self"} else "cloud"
+
+
+def api_base_url() -> str:
+    return os.environ.get("MEM0_API_URL", CLOUD_API_URL).rstrip("/")
+
+
+def auth_headers(api_key: str) -> dict[str, str]:
+    if api_mode() == "self_hosted":
+        header = os.environ.get("MEM0_AUTH_HEADER", "X-API-Key")
+        return {header: api_key}
+    return {"Authorization": f"Token {api_key}"}
+
+
+def _request_json(
+    api_key: str,
+    path: str,
+    *,
+    method: str = "GET",
+    body: dict[str, Any] | None = None,
+    timeout: int = 15,
+) -> tuple[int, Any]:
+    data = json.dumps(body).encode("utf-8") if body is not None else None
+    headers = {"Content-Type": "application/json", **auth_headers(api_key)}
+    req = urllib.request.Request(f"{api_base_url()}{path}", data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=timeout) as resp:
+        raw = resp.read()
+        if not raw:
+            return resp.status, {}
+        if not isinstance(raw, (bytes, bytearray, str)):
+            return resp.status, {}
+        return resp.status, json.loads(raw)
+
+
+def _self_hosted_body(body: dict[str, Any]) -> dict[str, Any]:
+    mapped = dict(body)
+    app_id = mapped.pop("app_id", None)
+    if app_id:
+        metadata = mapped.setdefault("metadata", {}) or {}
+        metadata["project"] = app_id
+        mapped["metadata"] = metadata
+    return mapped
+
+
+def _self_hosted_filters(value: Any) -> Any:
+    if not isinstance(value, dict):
+        if isinstance(value, list):
+            return [_self_hosted_filters(item) for item in value]
+        return value
+
+    # Flatten AND clauses if present to support flat backend filters
+    flat: dict[str, Any] = {}
+    if "AND" in value and isinstance(value["AND"], list):
+        for item in value["AND"]:
+            if isinstance(item, dict):
+                for k, v in item.items():
+                    flat[k] = v
+        # Also include any other top-level keys that aren't logical operators
+        for k, v in value.items():
+            if k not in ("AND", "OR", "NOT"):
+                flat[k] = v
+    else:
+        flat = value
+
+    mapped: dict[str, Any] = {}
+    for key, item in flat.items():
+        mapped["project" if key == "app_id" else key] = _self_hosted_filters(item)
+    return mapped
+
+
+def add_memory(api_key: str, body: dict[str, Any], timeout: int = 15) -> tuple[int, Any]:
+    path = "/memories" if api_mode() == "self_hosted" else "/v3/memories/add/"
+    if api_mode() == "self_hosted":
+        body = _self_hosted_body(body)
+    return _request_json(api_key, path, method="POST", body=body, timeout=timeout)
+
+
+def search_memories(api_key: str, body: dict[str, Any], timeout: int = 5) -> tuple[int, Any]:
+    path = "/search" if api_mode() == "self_hosted" else "/v3/memories/search/"
+    if api_mode() == "self_hosted" and "limit" in body and "top_k" not in body:
+        body = {**body, "top_k": body["limit"]}
+    if api_mode() == "self_hosted" and "filters" in body:
+        body = {**body, "filters": _self_hosted_filters(body["filters"])}
+    return _request_json(api_key, path, method="POST", body=body, timeout=timeout)
+
+
+def list_memories(api_key: str, body: dict[str, Any], timeout: int = 5) -> tuple[int, Any]:
+    if api_mode() == "self_hosted":
+        filters = _self_hosted_filters(body.get("filters") or {})
+        params: dict[str, Any] = {}
+        for key in ("user_id", "agent_id", "run_id"):
+            if key in filters and isinstance(filters[key], str) and filters[key] != "*":
+                params[key] = filters[key]
+        if "page_size" in body:
+            params["top_k"] = body["page_size"]
+        elif "top_k" in body:
+            params["top_k"] = body["top_k"]
+        query = f"?{urllib.parse.urlencode(params)}" if params else ""
+        return _request_json(api_key, f"/memories{query}", method="GET", timeout=timeout)
+
+    page = body.get("page", 1)
+    page_size = body.get("page_size") or body.get("top_k") or 10
+    return _request_json(
+        api_key,
+        f"/v3/memories/?page={page}&page_size={page_size}",
+        method="POST",
+        body={"filters": body.get("filters", {})},
+        timeout=timeout,
+    )
+
+
+def delete_memory(api_key: str, memory_id: str, timeout: int = 10) -> tuple[int, Any]:
+    if api_mode() == "self_hosted":
+        return _request_json(api_key, f"/memories/{memory_id}", method="DELETE", timeout=timeout)
+    return _request_json(api_key, f"/v1/memories/{memory_id}/", method="DELETE", timeout=timeout)
diff --git a/integrations/mem0-plugin/scripts/_identity.sh b/integrations/mem0-plugin/scripts/_identity.sh
index 4a25b8afcc..766d3e0bfb 100644
--- a/integrations/mem0-plugin/scripts/_identity.sh
+++ b/integrations/mem0-plugin/scripts/_identity.sh
@@ -12,6 +12,16 @@
 
 _SCRIPT_DIR="$( cd "$(dirname "${BASH_SOURCE[0]:-$0}")" && pwd )"
 
+# Define a fallback function for python3 if only python is available (common on Windows)
+if ! command -v python3 >/dev/null 2>&1; then
+  if command -v python >/dev/null 2>&1; then
+    python3() {
+      python "$@"
+    }
+    export -f python3 2>/dev/null || true
+  fi
+fi
+
 # Resolve API key: env var > userConfig > shell profile extraction
 if [ -z "${MEM0_API_KEY:-}" ] && [ -n "${CLAUDE_PLUGIN_OPTION_API_KEY:-}" ]; then
   MEM0_API_KEY="$CLAUDE_PLUGIN_OPTION_API_KEY"
@@ -56,6 +66,17 @@ _mem0_resolve_identity() {
 MEM0_RESOLVED_USER_ID="$(_mem0_resolve_identity)"
 export MEM0_RESOLVED_USER_ID
 
+_mem0_resolve_agent() {
+  if [ -n "${MEM0_AGENT_ID:-}" ]; then
+    printf '%s' "$MEM0_AGENT_ID"
+    return
+  fi
+  printf '%s' "codex"
+}
+
+MEM0_RESOLVED_AGENT_ID="$(_mem0_resolve_agent)"
+export MEM0_RESOLVED_AGENT_ID
+
 _MEM0_IDENTITY_ANNOTATION=""
 if [ -n "${MEM0_USER_ID:-}" ] && [ "$MEM0_USER_ID" != "${USER:-default}" ]; then
   _MEM0_IDENTITY_ANNOTATION=" (override; default: ${USER:-default})"
diff --git a/integrations/mem0-plugin/scripts/_search.py b/integrations/mem0-plugin/scripts/_search.py
index 5a19cfd41f..668302239a 100644
--- a/integrations/mem0-plugin/scripts/_search.py
+++ b/integrations/mem0-plugin/scripts/_search.py
@@ -6,11 +6,10 @@
 
 from __future__ import annotations
 
-import json
 import os
-import urllib.request
 
-SEARCH_URL = "https://api.mem0.ai/v3/memories/search/"
+from _api import search_memories as api_search_memories
+
 SEARCH_TIMEOUT = 5
 
 
@@ -33,16 +32,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",
-    )
-    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", [])
+    _status, data = api_search_memories(api_key, payload, timeout=SEARCH_TIMEOUT)
+    return data if isinstance(data, list) else data.get("results", [])
 
 
 def search_memories(
diff --git a/integrations/mem0-plugin/scripts/auto_capture.py b/integrations/mem0-plugin/scripts/auto_capture.py
index 356c02803b..255c43869b 100755
--- a/integrations/mem0-plugin/scripts/auto_capture.py
+++ b/integrations/mem0-plugin/scripts/auto_capture.py
@@ -17,9 +17,9 @@
 import os
 import sys
 import urllib.error
-import urllib.request
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from _api import add_memory
 from _identity import resolve_api_key, resolve_user_id
 from _project import resolve_branch, resolve_project_id
 
@@ -39,7 +39,6 @@
     except OSError:
         pass
 
-API_URL = "https://api.mem0.ai"
 TAIL_LINES = 200
 MAX_CONTENT_CHARS = 8000
 MIN_CONTENT_CHARS = 100
@@ -127,25 +126,14 @@ def store_exchange(api_key: str, messages: list[dict], user_id: str,
         "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",
-    )
     try:
-        with urllib.request.urlopen(req, timeout=15) as resp:
-            if resp.status in (200, 201):
-                result = json.loads(resp.read())
-                log.info("Auto-captured: event_id=%s status=%s",
-                         result.get("event_id", "?"), result.get("status", "?"))
-                return True
-            log.warning("API returned status %d", resp.status)
-            return False
+        status, result = add_memory(api_key, body, timeout=15)
+        if status in (200, 201):
+            log.info("Auto-captured: event_id=%s status=%s",
+                     result.get("event_id", "?"), result.get("status", "?"))
+            return True
+        log.warning("API returned status %d", status)
+        return False
     except urllib.error.URLError as e:
         log.warning("API call failed: %s", e)
         return False
diff --git a/integrations/mem0-plugin/scripts/auto_import.py b/integrations/mem0-plugin/scripts/auto_import.py
index 004772bbd7..5c52f371b9 100644
--- a/integrations/mem0-plugin/scripts/auto_import.py
+++ b/integrations/mem0-plugin/scripts/auto_import.py
@@ -18,9 +18,9 @@
 import os
 import sys
 import urllib.error
-import urllib.request
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from _api import add_memory, delete_memory, search_memories
 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 +41,6 @@
     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")
@@ -122,7 +121,7 @@ def save_hashes(hashes: dict[str, str]) -> None:
 
 
 def already_imported(api_key: str, user_id: str, project_id: str, filename: str) -> bool:
-    body = json.dumps({
+    body = {
         "query": filename,
         "filters": {
             "AND": [
@@ -133,30 +132,23 @@ def already_imported(api_key: str, user_id: str, project_id: str, filename: str)
         },
         "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",
-    )
+    }
     try:
-        with urllib.request.urlopen(req, timeout=5) as r:
-            data = json.loads(r.read())
-            results = data if isinstance(data, list) else data.get("results", [])
-            for result in results:
-                meta = result.get("metadata", {}) if isinstance(result, dict) else {}
-                file_field = meta.get("file", "")
-                if file_field == filename or file_field.startswith(f"{filename}["):
-                    return True
-            return False
+        _status, data = search_memories(api_key, body, timeout=5)
+        results = data if isinstance(data, list) else data.get("results", [])
+        for result in results:
+            meta = result.get("metadata", {}) if isinstance(result, dict) else {}
+            file_field = meta.get("file", "")
+            if file_field == filename or file_field.startswith(f"{filename}["):
+                return True
+        return False
     except Exception:
         return False
 
 
 def _delete_stale_chunks(api_key: str, user_id: str, project_id: str, filename: str) -> int:
     """Find and delete existing chunks for a file before re-import. Returns count deleted."""
-    body = json.dumps({
+    body = {
         "query": filename,
         "filters": {
             "AND": [
@@ -167,27 +159,20 @@ def _delete_stale_chunks(api_key: str, user_id: str, project_id: str, filename:
         },
         "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",
-    )
+    }
     ids_to_delete = []
     try:
-        with urllib.request.urlopen(req, timeout=10) as r:
-            data = json.loads(r.read())
-            results = data if isinstance(data, list) else data.get("results", [])
-            for result in results:
-                if not isinstance(result, dict):
-                    continue
-                meta = result.get("metadata", {})
-                file_field = meta.get("file", "")
-                if file_field == filename or file_field.startswith(f"{filename}["):
-                    mid = result.get("id")
-                    if mid:
-                        ids_to_delete.append(mid)
+        _status, data = search_memories(api_key, body, timeout=10)
+        results = data if isinstance(data, list) else data.get("results", [])
+        for result in results:
+            if not isinstance(result, dict):
+                continue
+            meta = result.get("metadata", {})
+            file_field = meta.get("file", "")
+            if file_field == filename or file_field.startswith(f"{filename}["):
+                mid = result.get("id")
+                if mid:
+                    ids_to_delete.append(mid)
     except Exception as e:
         log.warning("Failed to search for stale chunks of %s: %s", filename, e)
         return 0
@@ -195,12 +180,8 @@ def _delete_stale_chunks(api_key: str, user_id: str, project_id: str, filename:
     deleted = 0
     for mid in ids_to_delete:
         try:
-            del_req = urllib.request.Request(
-                f"{API_URL}/v1/memories/{mid}/",
-                headers={"Authorization": f"Token {api_key}"},
-                method="DELETE",
-            )
-            with urllib.request.urlopen(del_req, timeout=10):
+            status, _data = delete_memory(api_key, mid, timeout=10)
+            if status in (200, 204):
                 deleted += 1
         except Exception as e:
             log.warning("Failed to delete stale chunk %s: %s", mid, e)
@@ -232,24 +213,13 @@ def post_memory(api_key: str, content: str, user_id: str, filename: str, project
         "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",
-    )
-
     try:
-        with urllib.request.urlopen(req, timeout=15) as resp:
-            if resp.status in (200, 201):
-                log.info("Imported %s (project=%s)", filename, project_id)
-                return True
-            log.warning("API returned status %d for %s", resp.status, filename)
-            return False
+        status, _result = add_memory(api_key, body, timeout=15)
+        if status in (200, 201):
+            log.info("Imported %s (project=%s)", filename, project_id)
+            return True
+        log.warning("API returned status %d for %s", status, filename)
+        return False
     except urllib.error.URLError as e:
         log.warning("API call failed for %s: %s", filename, e)
         return False
diff --git a/integrations/mem0-plugin/scripts/capture_compact_summary.py b/integrations/mem0-plugin/scripts/capture_compact_summary.py
index d6e70b3567..5ce7c2ad2b 100644
--- a/integrations/mem0-plugin/scripts/capture_compact_summary.py
+++ b/integrations/mem0-plugin/scripts/capture_compact_summary.py
@@ -21,10 +21,10 @@
 import os
 import sys
 import urllib.error
-import urllib.request
 from datetime import date, timedelta
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from _api import add_memory
 from _identity import resolve_api_key, resolve_user_id
 from _project import resolve_branch, resolve_project_id
 
@@ -44,7 +44,6 @@
     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.
@@ -113,23 +112,13 @@ def store_summary(api_key: str, summary: str, user_id: str, session_id: str, pro
         "expiration_date": expires,
     }
 
-    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",
-    )
     try:
-        with urllib.request.urlopen(req, timeout=15) as resp:
-            if resp.status in (200, 201):
-                log.info("Compact summary stored")
-                return True
-            log.warning("API returned status %d", resp.status)
-            return False
+        status, _result = add_memory(api_key, body, timeout=15)
+        if status in (200, 201):
+            log.info("Compact summary stored")
+            return True
+        log.warning("API returned status %d", status)
+        return False
     except urllib.error.URLError as e:
         log.warning("API call failed: %s", e)
         return False
diff --git a/integrations/mem0-plugin/scripts/capture_session_summary.py b/integrations/mem0-plugin/scripts/capture_session_summary.py
index 441ed9a6e7..dfd1d373a1 100644
--- a/integrations/mem0-plugin/scripts/capture_session_summary.py
+++ b/integrations/mem0-plugin/scripts/capture_session_summary.py
@@ -19,10 +19,10 @@
 import re
 import sys
 import urllib.error
-import urllib.request
 from datetime import date, timedelta
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from _api import add_memory
 from _identity import resolve_api_key, resolve_user_id
 from _project import resolve_branch, resolve_project_id
 
@@ -42,7 +42,6 @@
     except OSError:
         pass
 
-API_URL = "https://api.mem0.ai"
 MAX_TAIL_LINES = 3000
 MAX_SUMMARY_CHARS = 50000
 SUMMARY_EXPIRY_DAYS = 90
@@ -185,23 +184,13 @@ def store_summary(
         "expiration_date": expires,
     }
 
-    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",
-    )
     try:
-        with urllib.request.urlopen(req, timeout=15) as resp:
-            if resp.status in (200, 201):
-                log.info("Session summary stored")
-                return True
-            log.warning("API returned status %d", resp.status)
-            return False
+        status, _result = add_memory(api_key, body, timeout=15)
+        if status in (200, 201):
+            log.info("Session summary stored")
+            return True
+        log.warning("API returned status %d", status)
+        return False
     except urllib.error.URLError as e:
         log.warning("API call failed: %s", e)
         return False
diff --git a/integrations/mem0-plugin/scripts/enforce_metadata_defaults.sh b/integrations/mem0-plugin/scripts/enforce_metadata_defaults.sh
index e24b636486..0dd40985d9 100755
--- a/integrations/mem0-plugin/scripts/enforce_metadata_defaults.sh
+++ b/integrations/mem0-plugin/scripts/enforce_metadata_defaults.sh
@@ -44,6 +44,7 @@ _PATCH_OUT="/tmp/mem0_enforce_$$"
 trap 'rm -f "$_PATCH_OUT"' EXIT
 _MEM0_TOOL_INPUT="$TOOL_INPUT" \
 _MEM0_USER_ID="${MEM0_RESOLVED_USER_ID:-}" \
+_MEM0_AGENT_ID="${MEM0_RESOLVED_AGENT_ID:-}" \
 _MEM0_APP_ID="${MEM0_PROJECT_ID:-}" \
 _MEM0_GLOBAL_SEARCH="${MEM0_GLOBAL_SEARCH:-false}" \
 _MEM0_HANDLER="$HANDLER" \
@@ -58,13 +59,14 @@ except Exception:
 
 handler = os.environ.get("_MEM0_HANDLER", "")
 resolved_uid = os.environ.get("_MEM0_USER_ID", "")
+resolved_agent_id = os.environ.get("_MEM0_AGENT_ID", "")
 resolved_aid = os.environ.get("_MEM0_APP_ID", "")
 global_search = os.environ.get("_MEM0_GLOBAL_SEARCH", "false") == "true"
 changed = False
 
 
-def inject_top_level_identity(inp, uid, aid):
-    """Inject user_id/app_id as top-level params (for add_memory, delete_all)."""
+def inject_top_level_identity(inp, uid, aid, agent_id):
+    """Inject user_id/app_id/agent_id as top-level params (for add_memory, delete_all)."""
     changed = False
     if uid and not inp.get("user_id"):
         inp["user_id"] = uid
@@ -72,6 +74,9 @@ def inject_top_level_identity(inp, uid, aid):
     if aid and not inp.get("app_id"):
         inp["app_id"] = aid
         changed = True
+    if agent_id and not inp.get("agent_id"):
+        inp["agent_id"] = agent_id
+        changed = True
     return changed
 
 
@@ -136,7 +141,7 @@ def inject_filter_identity(inp, uid, aid):
 
 
 if handler == "add_memory":
-    changed = inject_top_level_identity(inp, resolved_uid, resolved_aid)
+    changed = inject_top_level_identity(inp, resolved_uid, resolved_aid, resolved_agent_id)
 
     meta = inp.get("metadata") or {}
 
@@ -186,7 +191,7 @@ elif handler in ("search_memories", "get_memories"):
         changed = inject_filter_identity(inp, resolved_uid, resolved_aid)
 
 elif handler == "delete_all":
-    changed = inject_top_level_identity(inp, resolved_uid, resolved_aid)
+    changed = inject_top_level_identity(inp, resolved_uid, resolved_aid, resolved_agent_id)
 
 if changed:
     print(json.dumps(inp))
diff --git a/integrations/mem0-plugin/scripts/import_competing_tools.py b/integrations/mem0-plugin/scripts/import_competing_tools.py
index 968bbbcd22..e5538ed147 100644
--- a/integrations/mem0-plugin/scripts/import_competing_tools.py
+++ b/integrations/mem0-plugin/scripts/import_competing_tools.py
@@ -22,9 +22,9 @@
 import os
 import sys
 import urllib.error
-import urllib.request
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from _api import add_memory
 from _chunking import (
     filter_and_truncate,
     split_by_headers,
@@ -33,7 +33,6 @@
 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")
 
 
@@ -81,19 +80,9 @@ def post_memory(api_key: str, content: str, user_id: str, project_id: str, branc
         "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",
-    )
     try:
-        with urllib.request.urlopen(req, timeout=20) as resp:
-            return resp.status in (200, 201)
+        status, _result = add_memory(api_key, body, timeout=20)
+        return status in (200, 201)
     except urllib.error.URLError as e:
         print(f"  [warn] API call failed: {e}", file=sys.stderr)
         return False
diff --git a/integrations/mem0-plugin/scripts/on_pre_compact.py b/integrations/mem0-plugin/scripts/on_pre_compact.py
index 882e264811..e7b78b3ad6 100755
--- a/integrations/mem0-plugin/scripts/on_pre_compact.py
+++ b/integrations/mem0-plugin/scripts/on_pre_compact.py
@@ -19,10 +19,10 @@
 import os
 import sys
 import urllib.error
-import urllib.request
 from datetime import date, timedelta
 
 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from _api import add_memory
 from _identity import resolve_api_key, resolve_user_id
 from _project import resolve_branch, resolve_project_id
 
@@ -42,7 +42,6 @@
     except OSError:
         pass
 
-API_URL = "https://api.mem0.ai"
 MAX_TAIL_LINES = 500
 MAX_USER_MESSAGES = 30
 MAX_BASH_COMMANDS = 20
@@ -179,24 +178,13 @@ def store_memory(api_key: str, content: str, user_id: str, source: str, session_
         "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",
-    )
-
     try:
-        with urllib.request.urlopen(req, timeout=15) as resp:
-            if resp.status in (200, 201):
-                log.info("Session state stored successfully")
-                return True
-            log.warning("API returned status %d", resp.status)
-            return False
+        status, _result = add_memory(api_key, body, timeout=15)
+        if status in (200, 201):
+            log.info("Session state stored successfully")
+            return True
+        log.warning("API returned status %d", status)
+        return False
     except urllib.error.URLError as e:
         log.warning("API call failed: %s", e)
         return False
diff --git a/integrations/mem0-plugin/scripts/on_session_start.sh b/integrations/mem0-plugin/scripts/on_session_start.sh
index 9f89e6745e..18cc278c33 100755
--- a/integrations/mem0-plugin/scripts/on_session_start.sh
+++ b/integrations/mem0-plugin/scripts/on_session_start.sh
@@ -71,26 +71,22 @@ fi
 
 MEM0_COUNT="?"
 if command -v python3 >/dev/null 2>&1; then
-  MEM0_COUNT=$(python3 -c "
+  MEM0_COUNT=$(PYTHONPATH="$SCRIPT_DIR" python3 -c "
 import json, os, urllib.request, urllib.error
+from _api import list_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', '')
 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)
+    _status, data = list_memories(api_key, {'filters': filters, 'page_size': 1}, timeout=5)
+    if isinstance(data, dict) and 'count' in data:
+        return data['count']
+    if isinstance(data, dict) and 'results' in data:
+        return len(data['results'])
+    if isinstance(data, list):
+        return len(data)
     return 0
 
 try:
diff --git a/integrations/mem0-plugin/scripts/session_stats.py b/integrations/mem0-plugin/scripts/session_stats.py
index 357fe4dd60..a29c828ab7 100644
--- a/integrations/mem0-plugin/scripts/session_stats.py
+++ b/integrations/mem0-plugin/scripts/session_stats.py
@@ -16,9 +16,10 @@
 import json
 import os
 import sys
+import tempfile
 from datetime import datetime
 
-STATS_FILE = f"/tmp/mem0_session_stats_{os.environ.get('USER', 'default')}.json"
+STATS_FILE = os.path.join(tempfile.gettempdir(), f"mem0_session_stats_{os.environ.get('USER', 'default')}.json")
 
 
 def _load() -> dict:
@@ -46,14 +47,16 @@ def _save(stats: dict) -> None:
 
 
 def init() -> None:
-    _save({
-        "adds": 0,
-        "searches": 0,
-        "categories": [],
-        "category_counts": {},
-        "recent_ids": [],
-        "started": datetime.now().isoformat(),
-    })
+    _save(
+        {
+            "adds": 0,
+            "searches": 0,
+            "categories": [],
+            "category_counts": {},
+            "recent_ids": [],
+            "started": datetime.now().isoformat(),
+        }
+    )
 
 
 def record_add(category: str = "", memory_id: str = "") -> None:
diff --git a/integrations/mem0-plugin/scripts/session_timeline.py b/integrations/mem0-plugin/scripts/session_timeline.py
index 86e22f7fbf..03b4fc7a5c 100644
--- a/integrations/mem0-plugin/scripts/session_timeline.py
+++ b/integrations/mem0-plugin/scripts/session_timeline.py
@@ -11,17 +11,15 @@
 
 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 list_memories
 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
@@ -36,24 +34,13 @@ def fetch_recent_memories(api_key: str, user_id: str, project_id: str) -> list[d
     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 []
+        _status, result = list_memories(api_key, {"filters": filters, "page_size": MAX_RECENT}, timeout=FETCH_TIMEOUT)
+        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 []
 
diff --git a/integrations/mem0-plugin/tests/test_capture_session_summary.py b/integrations/mem0-plugin/tests/test_capture_session_summary.py
index 7ce1f14589..da64c9f522 100644
--- a/integrations/mem0-plugin/tests/test_capture_session_summary.py
+++ b/integrations/mem0-plugin/tests/test_capture_session_summary.py
@@ -21,6 +21,9 @@ def __enter__(self):
     def __exit__(self, *_):
         return False
 
+    def read(self):
+        return b"{}"
+
 
 def _capture_request_body(monkeypatch):
     """Patch urlopen so store_summary posts nowhere; capture the request body."""
diff --git a/integrations/mem0-plugin/tests/test_self_hosted_api.py b/integrations/mem0-plugin/tests/test_self_hosted_api.py
new file mode 100644
index 0000000000..64f9904c12
--- /dev/null
+++ b/integrations/mem0-plugin/tests/test_self_hosted_api.py
@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+import json
+from unittest.mock import MagicMock, patch
+
+
+def test_self_hosted_add_memory_maps_app_id_to_agent_id(monkeypatch):
+    from _api import add_memory
+
+    monkeypatch.setenv("MEM0_API_MODE", "self_hosted")
+    monkeypatch.setenv("MEM0_API_URL", "https://mem0.example.test")
+
+    captured = {}
+
+    def mock_urlopen(req, timeout=None):
+        captured["url"] = req.full_url
+        captured["headers"] = dict(req.header_items())
+        captured["body"] = json.loads(req.data.decode("utf-8"))
+        resp = MagicMock()
+        resp.status = 200
+        resp.read.return_value = b'{"results":[]}'
+        resp.__enter__ = lambda s: s
+        resp.__exit__ = MagicMock(return_value=False)
+        return resp
+
+    with patch("urllib.request.urlopen", side_effect=mock_urlopen):
+        add_memory("m0sk-test", {"messages": [], "user_id": "demo-user", "app_id": "mem0"})
+
+    assert captured["url"] == "https://mem0.example.test/memories"
+    assert captured["headers"]["X-api-key"] == "m0sk-test"
+    assert captured["body"]["metadata"]["project"] == "mem0"
+    assert "app_id" not in captured["body"]
+    assert "agent_id" not in captured["body"]
+
+
+def test_self_hosted_search_maps_app_id_filter(monkeypatch):
+    from _api import search_memories
+
+    monkeypatch.setenv("MEM0_API_MODE", "self_hosted")
+    monkeypatch.setenv("MEM0_API_URL", "https://mem0.example.test")
+
+    captured = {}
+
+    def mock_urlopen(req, timeout=None):
+        captured["body"] = json.loads(req.data.decode("utf-8"))
+        resp = MagicMock()
+        resp.status = 200
+        resp.read.return_value = b'{"results":[]}'
+        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-test",
+            {"query": "decisions", "filters": {"AND": [{"user_id": "demo-user"}, {"app_id": "mem0"}]}},
+        )
+
+    assert captured["body"]["filters"]["project"] == "mem0"
+    assert captured["body"]["filters"]["user_id"] == "demo-user"
+    assert "app_id" not in captured["body"]["filters"]
+    assert "agent_id" not in captured["body"]["filters"]
+    assert "AND" not in captured["body"]["filters"]
diff --git a/server/mcp/Dockerfile b/server/mcp/Dockerfile
new file mode 100644
index 0000000000..e8824dedb8
--- /dev/null
+++ b/server/mcp/Dockerfile
@@ -0,0 +1,14 @@
+FROM python:3.12-slim
+
+ENV PYTHONUNBUFFERED=1
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY mem0_mcp_bridge ./mem0_mcp_bridge
+
+ENV HOST=0.0.0.0
+ENV PORT=8765
+
+CMD ["python", "-m", "mem0_mcp_bridge.server"]
diff --git a/server/mcp/README.md b/server/mcp/README.md
new file mode 100644
index 0000000000..50c7e027af
--- /dev/null
+++ b/server/mcp/README.md
@@ -0,0 +1,71 @@
+# Mem0 Self-Hosted MCP Bridge
+
+This sidecar exposes MCP tools for the self-hosted Mem0 REST server. It keeps the hosted MCP tool signatures while forwarding requests to the self-hosted REST API.
+
+## Tools
+
+- `add_memory` (stores preferences/facts, accepts `project` and `task`/`run_id`)
+- `search_memories` (searches facts, supports filtering by `project`)
+- `get_memories` (lists user memories)
+- `get_memory` (gets a single memory)
+- `update_memory` (updates a memory)
+- `delete_memory` (deletes a memory)
+- `delete_all_memories` (deletes memories for a user/run scope)
+- `list_entities` (lists users, agents, and runs holding memories)
+- `delete_entities` (removes user, agent, or run entities)
+
+## Run With Docker Compose
+
+From the repository root:
+
+```bash
+docker compose up -d --build
+```
+
+Expose `127.0.0.1:8765` through your reverse proxy or Tailscale HTTPS endpoint, then configure MCP clients with the `/mcp` Streamable HTTP URL.
+
+## Dynamic Client Scoping & Authentication
+
+The MCP bridge dynamically routes requests to the backend using the client's credentials and parameters sent via headers:
+
+- **Authentication**: Forwarded to the backend as `X-API-Key`.
+- **User Identity (`X-User-Id`)**: Associates memories to the developer.
+- **Source Agent (`X-Source-Agent`)**: Automatically records which agent wrote the memory (stored in `metadata.source`).
+
+### Client Configuration
+
+#### 1. Codex (JSON format)
+Configure the MCP server in your `.codex-mcp.json` or equivalent configuration:
+
+```json
+{
+  "mcpServers": {
+    "mem0": {
+      "type": "http",
+      "url": "https://<your-vps-url>:8765/mcp",
+      "bearer_token_env_var": "MEM0_API_KEY",
+      "http_headers": {
+        "X-User-Id": "seth",
+        "X-Source-Agent": "codex"
+      }
+    }
+  }
+}
+```
+*Make sure to export `MEM0_API_KEY` locally in the shell running Codex.*
+
+#### 2. Other Agents (TOML format)
+Configure using TOML:
+
+```toml
+[mcp_servers.mem0]
+url = "https://<your-vps-url>:8765/mcp"
+bearer_token_env_var = "MEM0_API_KEY"
+http_headers = { "X-User-Id" = "seth", "X-Source-Agent" = "codex" }
+```
+
+## Memory Scoping
+
+The self-hosted REST server supports `user_id`, `agent_id`, and `run_id`.
+- **Project Scope**: Use the `project` tool parameter. It is automatically stored inside the memory's `metadata` and is used for scoping search queries.
+- **Task Scope**: Use the `run_id` (or `task`) tool parameter to isolate memories to specific active debug sessions or tickets.
diff --git a/server/mcp/mem0_mcp_bridge/__init__.py b/server/mcp/mem0_mcp_bridge/__init__.py
new file mode 100644
index 0000000000..2c05762b5e
--- /dev/null
+++ b/server/mcp/mem0_mcp_bridge/__init__.py
@@ -0,0 +1 @@
+"""MCP bridge for the Mem0 self-hosted REST server."""
diff --git a/server/mcp/mem0_mcp_bridge/client.py b/server/mcp/mem0_mcp_bridge/client.py
new file mode 100644
index 0000000000..52ff76c49c
--- /dev/null
+++ b/server/mcp/mem0_mcp_bridge/client.py
@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+import os
+from typing import Any
+
+import httpx
+
+
+class Mem0SelfHostedClient:
+    def __init__(self, api_url: str, api_key: str = "", timeout: float = 15.0) -> None:
+        self.api_url = api_url.rstrip("/")
+        self.api_key = api_key
+        self.timeout = timeout
+
+    @classmethod
+    def from_env(cls) -> "Mem0SelfHostedClient":
+        api_url = os.environ.get("MEM0_API_URL", "http://mem0:8000")
+        api_key = os.environ.get("MEM0_API_KEY") or os.environ.get("ADMIN_API_KEY", "")
+        return cls(api_url=api_url, api_key=api_key)
+
+    def with_api_key(self, api_key: str) -> "Mem0SelfHostedClient":
+        """Return a new client instance using a different API key."""
+        return Mem0SelfHostedClient(api_url=self.api_url, api_key=api_key, timeout=self.timeout)
+
+    @property
+    def headers(self) -> dict[str, str]:
+        headers = {"Content-Type": "application/json"}
+        if self.api_key:
+            headers["X-API-Key"] = self.api_key
+        return headers
+
+    def request(
+        self,
+        method: str,
+        path: str,
+        *,
+        json_body: dict[str, Any] | None = None,
+        params: dict[str, Any] | None = None,
+    ) -> Any:
+        with httpx.Client(timeout=self.timeout) as client:
+            response = client.request(
+                method,
+                f"{self.api_url}{path}",
+                headers=self.headers,
+                json=json_body,
+                params={k: v for k, v in (params or {}).items() if v is not None},
+            )
+            response.raise_for_status()
+            if not response.content:
+                return {}
+            return response.json()
+
+
+def build_filters(
+    *,
+    user_id: str | None = None,
+    agent_id: str | None = None,
+    run_id: str | None = None,
+    project: str | None = None,
+    extra: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """Build a flat filters dict that the self-hosted backend expects.
+
+    Entity IDs (user_id, agent_id, run_id) go at the top level.
+    The optional ``project`` key is also placed at the top level so
+    the vector store can filter on it as a metadata field.
+    Any ``extra`` filters are merged in as well.
+    """
+    filters: dict[str, Any] = {}
+    if user_id:
+        filters["user_id"] = user_id
+    if agent_id:
+        filters["agent_id"] = agent_id
+    if run_id:
+        filters["run_id"] = run_id
+    if project:
+        filters["project"] = project
+    if extra:
+        filters.update(extra)
+    return filters
diff --git a/server/mcp/mem0_mcp_bridge/server.py b/server/mcp/mem0_mcp_bridge/server.py
new file mode 100644
index 0000000000..288b1290e8
--- /dev/null
+++ b/server/mcp/mem0_mcp_bridge/server.py
@@ -0,0 +1,302 @@
+from __future__ import annotations
+
+import contextvars
+import json
+import logging
+import os
+from typing import Annotated, Any
+
+from mcp.server.fastmcp import FastMCP
+from mcp.server.transport_security import TransportSecuritySettings
+from pydantic import Field
+from starlette.types import ASGIApp, Receive, Scope, Send
+
+from .client import Mem0SelfHostedClient, build_filters
+
+logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s | %(message)s")
+logger = logging.getLogger("mem0_mcp_bridge")
+
+# ---------------------------------------------------------------------------
+# Defaults from env (used as fallback when headers are absent)
+# ---------------------------------------------------------------------------
+DEFAULT_USER_ID = os.environ.get("MEM0_DEFAULT_USER_ID", "mem0-mcp")
+DEFAULT_AGENT_ID = os.environ.get("MEM0_DEFAULT_AGENT_ID", "")
+
+# ---------------------------------------------------------------------------
+# Context variables – populated per-request by ASGI middleware
+# ---------------------------------------------------------------------------
+_request_api_key: contextvars.ContextVar[str] = contextvars.ContextVar("_request_api_key", default="")
+_request_user_id: contextvars.ContextVar[str] = contextvars.ContextVar("_request_user_id", default="")
+_request_source_agent: contextvars.ContextVar[str] = contextvars.ContextVar("_request_source_agent", default="")
+
+
+class _HeaderMiddleware:
+    """ASGI middleware that extracts per-client identity from HTTP headers.
+
+    Headers read:
+      - ``Authorization: Bearer <key>``  → forwarded to the backend as API key
+      - ``X-User-Id``                    → scopes memories to this user
+      - ``X-Source-Agent``               → auto-tagged on writes (provenance)
+    """
+
+    def __init__(self, app: ASGIApp) -> None:
+        self.app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] in ("http", "websocket"):
+            headers = dict(scope.get("headers", []))
+            # Extract Bearer token or raw API key
+            auth = headers.get(b"authorization", b"").decode().strip()
+            if auth:
+                if auth.lower().startswith("bearer "):
+                    _request_api_key.set(auth[7:].strip())
+                else:
+                    _request_api_key.set(auth)
+            # Extract identity headers
+            user_id = headers.get(b"x-user-id", b"").decode().strip()
+            if user_id:
+                _request_user_id.set(user_id)
+            source = headers.get(b"x-source-agent", b"").decode().strip()
+            if source:
+                _request_source_agent.set(source)
+        await self.app(scope, receive, send)
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _json_call(func, *args, **kwargs) -> str:
+    try:
+        return json.dumps(func(*args, **kwargs), ensure_ascii=False)
+    except Exception as exc:
+        logger.exception("Mem0 MCP bridge call failed")
+        return json.dumps({"error": type(exc).__name__, "detail": str(exc)}, ensure_ascii=False)
+
+
+_base_client = Mem0SelfHostedClient.from_env()
+
+
+def _client() -> Mem0SelfHostedClient:
+    """Return a client using the per-request Bearer token if present,
+    otherwise fall back to the env-configured API key."""
+    api_key = _request_api_key.get()
+    if api_key:
+        return _base_client.with_api_key(api_key)
+    return _base_client
+
+
+def _effective_user_id(user_id: str | None) -> str:
+    """Resolve the user_id: explicit arg > header > env default."""
+    return user_id or _request_user_id.get() or DEFAULT_USER_ID
+
+
+def _effective_agent_id(agent_id: str | None) -> str | None:
+    """Resolve the agent_id: explicit arg > header > env default."""
+    return agent_id or _request_source_agent.get() or DEFAULT_AGENT_ID or None
+
+
+def _source_metadata(metadata: dict[str, Any] | None) -> dict[str, Any]:
+    """Inject ``source`` into metadata from the X-Source-Agent header."""
+    source = _request_source_agent.get()
+    if not source:
+        return metadata or {}
+    result = dict(metadata) if metadata else {}
+    result.setdefault("source", source)
+    return result
+
+
+# ---------------------------------------------------------------------------
+# MCP Server
+# ---------------------------------------------------------------------------
+
+server = FastMCP(
+    "mem0",
+    host=os.environ.get("HOST", "0.0.0.0"),
+    port=int(os.environ.get("PORT", "8765")),
+    transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False),
+)
+
+
+# ---------------------------------------------------------------------------
+# Tools
+# ---------------------------------------------------------------------------
+
+
+@server.tool(description="Store a new preference, fact, or conversation snippet.")
+def add_memory(
+    text: Annotated[str | None, Field(default=None, description="Plain sentence summarizing what to store.")] = None,
+    messages: Annotated[list[dict[str, str]] | None, Field(default=None, description="Role/content messages.")] = None,
+    user_id: Annotated[str | None, Field(default=None, description="User scope (default: from config header).")] = None,
+    agent_id: Annotated[
+        str | None, Field(default=None, description="Agent scope (default: from config header).")
+    ] = None,
+    run_id: Annotated[str | None, Field(default=None, description="Run/session scope.")] = None,
+    project: Annotated[
+        str | None, Field(default=None, description="Project scope. Stored in metadata for filtering.")
+    ] = None,
+    metadata: Annotated[dict[str, Any] | None, Field(default=None, description="Metadata JSON.")] = None,
+    infer: Annotated[bool, Field(default=True, description="Whether Mem0 should extract facts.")] = True,
+) -> str:
+    if not messages:
+        if not text:
+            return json.dumps({"error": "messages_missing", "detail": "Provide text or messages."})
+        messages = [{"role": "user", "content": text}]
+
+    effective_metadata = _source_metadata(metadata)
+    if project:
+        effective_metadata["project"] = project
+
+    payload: dict[str, Any] = {
+        "messages": messages,
+        "user_id": _effective_user_id(user_id),
+        "agent_id": _effective_agent_id(agent_id),
+        "run_id": run_id,
+        "metadata": effective_metadata or None,
+        "infer": infer,
+    }
+    payload = {k: v for k, v in payload.items() if v is not None}
+    return _json_call(_client().request, "POST", "/memories", json_body=payload)
+
+
+@server.tool(description="Run a semantic search over existing memories.")
+def search_memories(
+    query: Annotated[str, Field(description="Natural language search query.")],
+    user_id: Annotated[str | None, Field(default=None, description="User scope (default: from config header).")] = None,
+    agent_id: Annotated[str | None, Field(default=None, description="Agent scope.")] = None,
+    run_id: Annotated[str | None, Field(default=None, description="Run scope.")] = None,
+    project: Annotated[str | None, Field(default=None, description="Filter by project.")] = None,
+    filters: Annotated[dict[str, Any] | None, Field(default=None, description="Additional structured filters.")] = None,
+    top_k: Annotated[int | None, Field(default=None, description="Maximum results.")] = None,
+    threshold: Annotated[float | None, Field(default=None, description="Minimum similarity score.")] = None,
+) -> str:
+    search_filters = build_filters(
+        user_id=_effective_user_id(user_id),
+        agent_id=agent_id,
+        run_id=run_id,
+        project=project,
+        extra=filters,
+    )
+    payload: dict[str, Any] = {
+        "query": query,
+        "filters": search_filters,
+        "top_k": top_k,
+        "threshold": threshold,
+    }
+    payload = {k: v for k, v in payload.items() if v is not None}
+    return _json_call(_client().request, "POST", "/search", json_body=payload)
+
+
+@server.tool(description="List memories using structured filters.")
+def get_memories(
+    user_id: Annotated[str | None, Field(default=None, description="User scope (default: from config header).")] = None,
+    agent_id: Annotated[str | None, Field(default=None, description="Agent scope.")] = None,
+    run_id: Annotated[str | None, Field(default=None, description="Run scope.")] = None,
+    project: Annotated[str | None, Field(default=None, description="Filter by project.")] = None,
+    top_k: Annotated[int | None, Field(default=None, description="Maximum memories to list.")] = None,
+) -> str:
+    params: dict[str, Any] = {
+        "user_id": _effective_user_id(user_id),
+        "top_k": top_k,
+    }
+    if agent_id:
+        params["agent_id"] = agent_id
+    if run_id:
+        params["run_id"] = run_id
+    # Note: project filtering on GET /memories is not supported by the backend
+    # query params; agent should use search_memories with project filter instead.
+    return _json_call(_client().request, "GET", "/memories", params=params)
+
+
+@server.tool(description="Fetch a single memory by ID.")
+def get_memory(memory_id: Annotated[str, Field(description="Memory ID.")]) -> str:
+    return _json_call(_client().request, "GET", f"/memories/{memory_id}")
+
+
+@server.tool(description="Overwrite an existing memory's text.")
+def update_memory(
+    memory_id: Annotated[str, Field(description="Memory ID.")],
+    text: Annotated[str, Field(description="Replacement memory text.")],
+    metadata: Annotated[
+        dict[str, Any] | None, Field(default=None, description="Optional replacement metadata.")
+    ] = None,
+) -> str:
+    payload: dict[str, Any] = {"text": text}
+    if metadata is not None:
+        payload["metadata"] = metadata
+    return _json_call(_client().request, "PUT", f"/memories/{memory_id}", json_body=payload)
+
+
+@server.tool(description="Delete one memory by ID.")
+def delete_memory(memory_id: Annotated[str, Field(description="Memory ID.")]) -> str:
+    return _json_call(_client().request, "DELETE", f"/memories/{memory_id}")
+
+
+@server.tool(description="Delete every memory in a user/agent/run scope.")
+def delete_all_memories(
+    user_id: Annotated[str | None, Field(default=None, description="User scope (default: from config header).")] = None,
+    agent_id: Annotated[str | None, Field(default=None, description="Agent scope.")] = None,
+    run_id: Annotated[str | None, Field(default=None, description="Run scope.")] = None,
+) -> str:
+    params: dict[str, Any] = {
+        "user_id": _effective_user_id(user_id),
+    }
+    if agent_id:
+        params["agent_id"] = agent_id
+    if run_id:
+        params["run_id"] = run_id
+    return _json_call(_client().request, "DELETE", "/memories", params=params)
+
+
+@server.tool(description="List users, agents, and runs currently holding memories.")
+def list_entities() -> str:
+    return _json_call(_client().request, "GET", "/entities")
+
+
+@server.tool(description="Delete a user/agent/run entity and its memories.")
+def delete_entities(
+    user_id: Annotated[str | None, Field(default=None, description="User entity to delete.")] = None,
+    agent_id: Annotated[str | None, Field(default=None, description="Agent entity to delete.")] = None,
+    run_id: Annotated[str | None, Field(default=None, description="Run entity to delete.")] = None,
+) -> str:
+    scopes = [(k[:-3], v) for k, v in {"user_id": user_id, "agent_id": agent_id, "run_id": run_id}.items() if v]
+    if len(scopes) != 1:
+        return json.dumps({"error": "scope_invalid", "detail": "Provide exactly one of user_id, agent_id, or run_id."})
+    entity_type, entity_id = scopes[0]
+    return _json_call(_client().request, "DELETE", f"/entities/{entity_type}/{entity_id}")
+
+
+# ---------------------------------------------------------------------------
+# Prompt
+# ---------------------------------------------------------------------------
+
+
+@server.prompt()
+def memory_assistant() -> str:
+    return (
+        "You have access to Mem0 tools for long-term memory. "
+        "Memories are scoped by user_id (the human), agent_id (the AI agent), "
+        "and optionally by project (stored in metadata) and run_id (task/session). "
+        "Use the 'project' parameter to organize memories by project/workspace. "
+        "Store durable facts, preferences, project decisions, and task learnings."
+    )
+
+
+# ---------------------------------------------------------------------------
+# Entrypoint
+# ---------------------------------------------------------------------------
+
+
+def main() -> None:
+    logger.info("Starting Mem0 self-hosted MCP bridge at %s:%s", server.settings.host, server.settings.port)
+    # Wrap the ASGI app with header-extraction middleware
+    original_app = server.streamable_http_app()
+    wrapped_app = _HeaderMiddleware(original_app)
+    # Replace the method so the server uses our wrapped app
+    server.streamable_http_app = lambda: wrapped_app  # type: ignore[method-assign]
+    server.run(transport="streamable-http")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/server/mcp/requirements.txt b/server/mcp/requirements.txt
new file mode 100644
index 0000000000..1752a94307
--- /dev/null
+++ b/server/mcp/requirements.txt
@@ -0,0 +1,3 @@
+mcp[cli]>=1.25.4
+httpx>=0.28.0
+pydantic>=2.7.3
diff --git a/tests/test_mem0_mcp_bridge.py b/tests/test_mem0_mcp_bridge.py
new file mode 100644
index 0000000000..d2269ca42c
--- /dev/null
+++ b/tests/test_mem0_mcp_bridge.py
@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+import os
+import sys
+
+ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+MCP_DIR = os.path.join(ROOT, "server", "mcp")
+if MCP_DIR not in sys.path:
+    sys.path.insert(0, MCP_DIR)
+
+from mem0_mcp_bridge.client import build_filters  # noqa: E402
+
+
+def test_build_filters_basic():
+    filters = build_filters(
+        user_id="demo-user",
+        agent_id="demo-agent",
+        run_id="demo-run",
+        project="demo-project",
+    )
+
+    assert filters["user_id"] == "demo-user"
+    assert filters["agent_id"] == "demo-agent"
+    assert filters["run_id"] == "demo-run"
+    assert filters["project"] == "demo-project"
+
+
+def test_build_filters_with_extra():
+    extra = {"metadata": {"type": "decision"}}
+    filters = build_filters(
+        user_id="demo-user",
+        project="demo-project",
+        extra=extra,
+    )
+
+    assert filters["user_id"] == "demo-user"
+    assert filters["project"] == "demo-project"
+    assert filters["metadata"] == {"type": "decision"}