kartik-mem0
16c2cfe50e
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
79 lines
2.3 KiB
Python
79 lines
2.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Fetch recent memories and format a compact timeline for SessionStart.
|
|
|
|
Searches mem0 cloud API for the most recent memories in the project
|
|
and formats them as a compact activity timeline injected below the
|
|
existing SessionStart banner.
|
|
|
|
Input: env vars for identity (MEM0_API_KEY, MEM0_RESOLVED_USER_ID, etc.)
|
|
Output: Compact timeline text to stdout (empty if nothing found)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import sys
|
|
|
|
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
|
|
|
|
MAX_RECENT = 10
|
|
MAX_SUMMARIES = 3
|
|
|
|
|
|
def fetch_recent_memories(api_key: str, user_id: str, project_id: str) -> list[dict]:
|
|
"""Fetch the most recent memories for this project."""
|
|
global_search = os.environ.get("MEM0_GLOBAL_SEARCH", "false") == "true"
|
|
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:
|
|
"""Format memories into a compact recent activity timeline."""
|
|
if not memories:
|
|
return ""
|
|
|
|
lines = ["### Recent Activity", ""]
|
|
|
|
for m in memories:
|
|
mid = m.get("id", "?")[:8]
|
|
text = (m.get("memory", "") or "")[:120].replace("\n", " ").strip()
|
|
meta = m.get("metadata") or {}
|
|
cat = meta.get("type", "unknown")
|
|
icon = TYPE_ICONS.get(cat, "❓")
|
|
age = format_age(m)
|
|
age_str = f" ({age})" if age else ""
|
|
lines.append(f"- {icon} [{cat}]{age_str} {text} [mem0:{mid}]")
|
|
|
|
lines.append("")
|
|
lines.append("Search mem0 for details on any of these, or for past decisions and task learnings relevant to the current task.")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def main():
|
|
api_key = resolve_api_key()
|
|
if not api_key:
|
|
return
|
|
|
|
user_id = resolve_user_id()
|
|
project_id = resolve_project_id(os.environ.get("MEM0_CWD"))
|
|
|
|
memories = fetch_recent_memories(api_key, user_id, project_id)
|
|
if not memories:
|
|
return
|
|
|
|
timeline = format_timeline(memories)
|
|
if timeline:
|
|
print(timeline, end="")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except Exception:
|
|
pass
|
|
sys.exit(0)
|