Python CLI Architecture

Copy"""Allow running with `python -m mem0_cli`."""
 
from mem0_cli.app import main
 
main()

Copyapp = typer.Typer(
    name="mem0",
    help=f"◆ Mem0 CLI v{__version__} · Python SDK\n\n   The Memory Layer for AI Agents",
    no_args_is_help=True,
    rich_markup_mode="rich",
    pretty_exceptions_enable=False,
    add_completion=False,
    subcommand_metavar="<command> [options]",
    options_metavar="",
)

Copyconfig_app = typer.Typer(
    name="config",
    help="Manage mem0 configuration.",
    no_args_is_help=True,
    rich_markup_mode="rich",
)
 
entity_app = typer.Typer(
    name="entity",
    help="Manage entities.",
    no_args_is_help=True,
    rich_markup_mode="rich",
)
 
event_app = typer.Typer(
    name="event",
    help="Inspect background processing events.",
    no_args_is_help=True,
    rich_markup_mode="rich",
)

Copydef _resolve_ids(
    config,
    *,
    user_id: str | None = None,
    agent_id: str | None = None,
    app_id: str | None = None,
    run_id: str | None = None,
):
    """Resolve entity IDs: CLI flag > config default > None.
 
    If any explicit ID is provided, only use explicit IDs (don't mix
    in defaults for other entity types which would over-filter).
    If no explicit IDs, fall back to all configured defaults.
    """
    has_explicit = any([user_id, agent_id, app_id, run_id])
    if has_explicit:
        return {
            "user_id": user_id or None,
            "agent_id": agent_id or None,
            "app_id": app_id or None,
            "run_id": run_id or None,
        }
    return {
        "user_id": config.defaults.user_id or None,
        "agent_id": config.defaults.agent_id or None,
        "app_id": config.defaults.app_id or None,
        "run_id": config.defaults.run_id or None,
    }

Copydef _stdin_is_piped() -> bool:
    """Return True only when stdin is an actual pipe or file redirect — not a bare open fd."""
    from mem0_cli.state import is_agent_mode
 
    if is_agent_mode():
        return False
    try:
        mode = os.fstat(sys.stdin.fileno()).st_mode
        return _stat_mod.S_ISFIFO(mode) or _stat_mod.S_ISREG(mode)
    except Exception:
        return False
 
 
def _read_stdin() -> str | None:
    """Read from stdin if it is an actual pipe or file redirect (not a TTY, not agent mode)."""
    if _stdin_is_piped():
        return sys.stdin.read().strip() or None
    return None

Copydef _get_backend_and_config(
    api_key: str | None = None,
    base_url: str | None = None,
):
    """Build and return the Platform backend plus the loaded config.
 
    Validates the API key upfront via ``/v1/ping/`` and caches the
    resolved user email for telemetry.
    """
    global _validated_user_email
 
    from mem0_cli.backend import get_backend
    from mem0_cli.backend.platform import AuthError
    from mem0_cli.config import load_config, save_config
 
    config = load_config()
 
    if api_key:
        config.platform.api_key = api_key
    if base_url:
        config.platform.base_url = base_url
 
    if not config.platform.api_key:
        print_error(
            err_console,
            "No API key configured.",
            hint="Run 'mem0 init' or set MEM0_API_KEY environment variable.",
        )
        raise typer.Exit(1)
 
    backend = get_backend(config)
 
    # Validate the API key upfront with a fast timeout
    try:
        ping_data = backend.ping(timeout=5.0)
        email = ping_data.get("user_email") if isinstance(ping_data, dict) else None
        if email:
            _validated_user_email = email
            if config.platform.user_email != email:
                config.platform.user_email = email
                with contextlib.suppress(Exception):
                    save_config(config)
    except AuthError:
        print_error(
            err_console,
            "Invalid or expired API key.",
            hint="Run 'mem0 init' or set MEM0_API_KEY environment variable.",
        )
        raise typer.Exit(1) from None
    except Exception:
        print_warning(err_console, "Could not validate API key (network issue). Proceeding anyway.")
 
    return backend, config

CopyCONFIG_DIR = Path.home() / ".mem0"
CONFIG_FILE = CONFIG_DIR / "config.json"
 
DEFAULT_BASE_URL = "https://api.mem0.ai"
CONFIG_VERSION = 1

Copydef ensure_config_dir() -> Path:
    """Create ~/.mem0 directory with secure permissions if it doesn't exist."""
    CONFIG_DIR.mkdir(parents=True, exist_ok=True)
    os.chmod(CONFIG_DIR, stat.S_IRWXU)  # 0700
    return CONFIG_DIR

Copywith open(CONFIG_FILE, "w") as f:
    json.dump(data, f, indent=2)
 
os.chmod(CONFIG_FILE, stat.S_IRUSR | stat.S_IWUSR)  # 0600
 

Copy@dataclass
class PlatformConfig:
    api_key: str = ""
    base_url: str = DEFAULT_BASE_URL
    user_email: str = ""
    # Agent Mode (unclaimed-shadow signup)
    agent_mode: bool = False  # True while the key is an unclaimed agent-mode key
    created_via: str = ""  # "agent_mode" | "email" | "api_key" | "existing_key"
    agent_caller: str = (
        ""  # canonical agent name when created_via == "agent_mode" (e.g. "claude-code")
    )
    claimed_at: str = ""  # ISO timestamp once the agent has been claimed by a human
    default_user_id: str = ""  # `user_<slug>` returned by bootstrap; used as auto-default
 
 
@dataclass
class DefaultsConfig:
    user_id: str = ""
    agent_id: str = ""
    app_id: str = ""
    run_id: str = ""
 
 
@dataclass
class TelemetryConfig:
    anonymous_id: str = ""
 
 
@dataclass
class AgentRushConfig:
    # ISO timestamp the human acknowledged the "memories are public" warning.
    # Empty until first interactive `mem0 agent-rush add`.
    acknowledged_at: str = ""
 
 
@dataclass
class Mem0Config:
    version: int = CONFIG_VERSION
    defaults: DefaultsConfig = field(default_factory=DefaultsConfig)
    platform: PlatformConfig = field(default_factory=PlatformConfig)
    telemetry: TelemetryConfig = field(default_factory=TelemetryConfig)
    agent_rush: AgentRushConfig = field(default_factory=AgentRushConfig)
 
 
SHORT_KEY_ALIASES: dict[str, str] = {
    "api_key": "platform.api_key",
    "base_url": "platform.base_url",
    "user_email": "platform.user_email",
    "user_id": "defaults.user_id",
    "agent_id": "defaults.agent_id",
    "app_id": "defaults.app_id",
    "run_id": "defaults.run_id",
}

Copydef load_config() -> Mem0Config:
    """Load config from file, applying env var overrides."""
    config = Mem0Config()
 
    if CONFIG_FILE.exists():
        with open(CONFIG_FILE) as f:
            data = json.load(f)
 
        config.version = data.get("version", CONFIG_VERSION)
 
        plat = data.get("platform", {})
        config.platform.api_key = plat.get("api_key", "")
        config.platform.base_url = plat.get("base_url", DEFAULT_BASE_URL)
        config.platform.user_email = plat.get("user_email", "")
        config.platform.agent_mode = bool(plat.get("agent_mode", False))
        config.platform.created_via = plat.get("created_via", "")
        config.platform.agent_caller = plat.get("agent_caller", "")
        config.platform.claimed_at = plat.get("claimed_at", "")
        config.platform.default_user_id = plat.get("default_user_id", "")
 
        defaults = data.get("defaults", {})
        config.defaults.user_id = defaults.get("user_id", "")
        config.defaults.agent_id = defaults.get("agent_id", "")
        config.defaults.app_id = defaults.get("app_id", "")
        config.defaults.run_id = defaults.get("run_id", "")
        telemetry = data.get("telemetry", {})
        config.telemetry.anonymous_id = telemetry.get("anonymous_id", "")
 
        agent_rush = data.get("agent_rush", {})
        config.agent_rush.acknowledged_at = agent_rush.get("acknowledged_at", "")
 
    # Environment variable overrides
    env_key = os.environ.get("MEM0_API_KEY")
    if env_key:
        config.platform.api_key = env_key
 
    env_base = os.environ.get("MEM0_BASE_URL")
    if env_base:
        config.platform.base_url = env_base
 
    env_user_id = os.environ.get("MEM0_USER_ID")
    if env_user_id:
        config.defaults.user_id = env_user_id
 
    env_agent_id = os.environ.get("MEM0_AGENT_ID")
    if env_agent_id:
        config.defaults.agent_id = env_agent_id
 
    env_app_id = os.environ.get("MEM0_APP_ID")
    if env_app_id:
        config.defaults.app_id = env_app_id
 
    env_run_id = os.environ.get("MEM0_RUN_ID")
    if env_run_id:
        config.defaults.run_id = env_run_id
 
    return config

Copydef get_nested_value(config: Mem0Config, dotted_key: str) -> Any:
    """Get a config value by dotted path, e.g. 'platform.api_key' or short form 'api_key'."""
    dotted_key = SHORT_KEY_ALIASES.get(dotted_key, dotted_key)
    parts = dotted_key.split(".")
    obj: Any = config
    for part in parts:
        if hasattr(obj, part):
            obj = getattr(obj, part)
        else:
            return None
    return obj
 
 
def set_nested_value(config: Mem0Config, dotted_key: str, value: str) -> bool:
    """Set a config value by dotted path. Returns True on success."""
    dotted_key = SHORT_KEY_ALIASES.get(dotted_key, dotted_key)
    parts = dotted_key.split(".")
    obj: Any = config
    for part in parts[:-1]:
        if hasattr(obj, part):
            obj = getattr(obj, part)
        else:
            return False
 
    final_key = parts[-1]
    if not hasattr(obj, final_key):
        return False
 
    current = getattr(obj, final_key)
    # Type coercion
    if isinstance(current, bool):
        value = value.lower() in ("true", "1", "yes")  # type: ignore[assignment]
    elif isinstance(current, int):
        try:
            value = int(value)  # type: ignore[assignment]
        except ValueError:
            return False
 
    setattr(obj, final_key, value)
    return True

Copyclass Backend(ABC):
    """Abstract interface for mem0 backends."""
 
    @abstractmethod
    def add(
        self,
        content: str | None = None,
        messages: list[dict] | None = None,
        *,
        user_id: str | None = None,
        agent_id: str | None = None,
        app_id: str | None = None,
        run_id: str | None = None,
        metadata: dict | None = None,
        immutable: bool = False,
        infer: bool = True,
        expires: str | None = None,
        custom_instructions: str | None = None,
        agent_custom_instructions: str | None = None,
        custom_categories: list[dict] | None = None,
        structured_data_schema: dict | None = None,
        timestamp: int | None = None,
    ) -> dict: ...
 
    @abstractmethod
    def search(
        self,
        query: str,
        *,
        user_id: str | None = None,
        agent_id: str | None = None,
        app_id: str | None = None,
        run_id: str | None = None,
        top_k: int = 10,
        threshold: float = 0.3,
        rerank: bool = False,
        keyword: bool = False,
        filters: dict | None = None,
        fields: list[str] | None = None,
        show_expired: bool = False,
        reference_date: str | None = None,
        latest_only: bool = False,
    ) -> list[dict]: ...
 
    @abstractmethod
    def get(self, memory_id: str) -> dict: ...
 
    @abstractmethod
    def list_memories(
        self,
        *,
        user_id: str | None = None,
        agent_id: str | None = None,
        app_id: str | None = None,
        run_id: str | None = None,
        page: int = 1,
        page_size: int = 100,
        category: str | None = None,
        after: str | None = None,
        before: str | None = None,
        show_expired: bool = False,
        latest_only: bool = False,
    ) -> list[dict]: ...
 
    @abstractmethod
    def update(
        self,
        memory_id: str,
        content: str | None = None,
        metadata: dict | None = None,
        *,
        expiration_date: str | None = None,
        timestamp: int | None = None,
    ) -> dict: ...
 
    @abstractmethod
    def delete(
        self,
        memory_id: str | None = None,
        *,

Copyclass PlatformBackend(Backend):
    """Backend that talks to the mem0 Platform API."""
 
    def __init__(self, config: PlatformConfig) -> None:
        self.config = config
        self.base_url = config.base_url.rstrip("/")
        self._client = httpx.Client(
            base_url=self.base_url,
            headers={
                "Authorization": f"Token {config.api_key}",
                "Content-Type": "application/json",
                "X-Mem0-Source": "CLI",
                "X-Mem0-Client": f"mem0-cli-python/{__version__}",
                "X-Mem0-Client-Language": "python",
                "X-Mem0-Client-Version": __version__,
            },
            timeout=30.0,
        )

Copydef _encode_path_segment(value: Any) -> str:
    return quote(str(value), safe="")

Copydef get_backend(config: Mem0Config) -> Backend:
    """Return the Platform backend."""
    from mem0_cli.backend.platform import PlatformBackend
 

Copyif top_k < 1:
    print_error(err_console, "--top-k must be >= 1.")
    raise typer.Exit(1)
if not (0.0 <= threshold <= 1.0):
    print_error(err_console, "--threshold must be between 0.0 and 1.0.")
    raise typer.Exit(1)
 

Copyresults = backend.list_memories(
    user_id=user_id,
    agent_id=agent_id,
    app_id=app_id,
    run_id=run_id,
    page=page,
    page_size=page_size,
    category=category,
    after=after,
    before=before,
    show_expired=show_expired,
    latest_only=latest_only,
)
…
def sanitize_agent_data(command: str, data: Any) -> Any:
    """Project API response data to minimal relevant fields for agent consumption."""
 
    def pick(obj: dict, keys: list) -> dict:
        return {k: obj[k] for k in keys if k in obj}
 
    if data is None:
        return data
 
    if command == "add":
        items = data if isinstance(data, list) else [data]
        result = []
        for item in items:
            if item.get("status") == "PENDING":
                result.append(pick(item, ["status", "event_id"]))
            else:
                result.append(pick(item, ["id", "memory", "event"]))
        return result
 
    if command == "search":
        return [
            pick(r, ["id", "memory", "score", "created_at", "categories", "expiration_date"])
            for r in data
        ]
 
    if command == "list":
        return [
            pick(r, ["id", "memory", "created_at", "categories", "expiration_date"]) for r in data
        ]
 

Copyresults = backend.search(
    query,
    user_id=user_id,
    agent_id=agent_id,
    app_id=app_id,
    run_id=run_id,
    top_k=top_k,
    threshold=threshold,
    rerank=rerank,
    keyword=keyword,
    filters=filters,
    fields=field_list,
    show_expired=show_expired,
    reference_date=reference_date,
    latest_only=latest_only,
)
…
def format_agent_envelope(
    console: Console,
    *,
    command: str,
    data: Any,
    duration_ms: int | None = None,
    scope: dict | None = None,
    count: int | None = None,
) -> None:
    """Output structured JSON envelope for agent/programmatic use (--json/--agent mode)."""
    envelope: dict[str, Any] = {
        "status": "success",
        "command": command,
    }
    if duration_ms is not None:
        envelope["duration_ms"] = duration_ms
    if scope:
        filtered = {k: v for k, v in scope.items() if v}
        if filtered:
            envelope["scope"] = filtered
    if count is not None:
        envelope["count"] = count
    envelope["data"] = sanitize_agent_data(command, data)
 
    # Surface the unclaimed-Agent-Mode notice (if any) in the envelope so an
    # agent reading the JSON output sees it without inspecting HTTP headers.
    from mem0_cli.state import take_notice
 
    notice = take_notice()
    if notice:
        envelope["mem0_notice"] = notice
 
    console.print_json(json.dumps(envelope, default=str))
…
def sanitize_agent_data(command: str, data: Any) -> Any:
    """Project API response data to minimal relevant fields for agent consumption."""
 
    def pick(obj: dict, keys: list) -> dict:
        return {k: obj[k] for k in keys if k in obj}
 
    if data is None:
        return data
 
    if command == "add":
        items = data if isinstance(data, list) else [data]
        result = []
        for item in items:
            if item.get("status") == "PENDING":
                result.append(pick(item, ["status", "event_id"]))
            else:
                result.append(pick(item, ["id", "memory", "event"]))
        return result
 
    if command == "search":
        return [
            pick(r, ["id", "memory", "score", "created_at", "categories", "expiration_date"])
            for r in data
        ]

Copy_STATUS_STYLE = {
    "SUCCEEDED": f"[{SUCCESS_COLOR}]SUCCEEDED[/]",
    "PENDING": f"[{ACCENT_COLOR}]PENDING[/]",
    "FAILED": f"[{ERROR_COLOR}]FAILED[/]",
    "PROCESSING": f"[{WARNING_COLOR}]PROCESSING[/]",
}

Copywith timed_status(err_console, "Fetching event...") as _ts:
    try:
        ev = backend.get_event(event_id)
…
panel = Panel(
    content,
    title=f"[{BRAND_COLOR}]Event Status[/]",
    title_align="left",
    border_style=BRAND_COLOR,
    padding=(1, 1),
)
console.print()
console.print(panel)

CopyBRAND_COLOR = "#8b5cf6"  # Purple
ACCENT_COLOR = "#a78bfa"
SUCCESS_COLOR = "#22c55e"
ERROR_COLOR = "#ef4444"
WARNING_COLOR = "#f59e0b"
DIM_COLOR = "#6b7280"

Copydef _sym(fancy: str, plain: str) -> str:
    """Return *fancy* when stdout is a TTY with colour, else *plain*."""
    if not sys.stdout.isatty() or os.environ.get("NO_COLOR") is not None:
        return plain
    return fancy

Copyfrom mem0_cli.state import is_agent_mode
 
if is_agent_mode():
    return
logo_text = Text(LOGO, style=f"bold {BRAND_COLOR}")
tagline = Text(f"  {TAGLINE}\n", style=f"{ACCENT_COLOR}")
 
content = Text()
content.append_text(logo_text)
content.append_text(tagline)
 
panel = Panel(
    content,
    border_style=BRAND_COLOR,
    padding=(0, 2),
    subtitle=f"[{DIM_COLOR}]Python SDK · v{_get_version()}[/]",
    subtitle_align="right",
)
console.print(panel)

Copydef cmd_entities_list(backend: Backend, entity_type: str, *, output: str) -> None:
    """List entities of a given type."""
    from mem0_cli.state import is_agent_mode, set_current_command
 
    set_current_command("entity list")
    if is_agent_mode():
        output = "agent"
    valid_types = {"users", "agents", "apps", "runs"}
    if entity_type not in valid_types:
        print_error(
            err_console, f"Invalid entity type: {entity_type}. Use: {', '.join(valid_types)}"
        )
        raise typer.Exit(1)
 
    _start = _time.perf_counter()
    with timed_status(err_console, f"Fetching {entity_type}...") as _ts:
        try:
            results = backend.entities(entity_type)
        except Exception as e:
            print_error(err_console, str(e), hint="This feature may require the mem0 Platform.")
            raise typer.Exit(1) from None
    _elapsed = _time.perf_counter() - _start
 
    if output == "agent":
        format_agent_envelope(
            console,
            command="entity list",
            data=results,
            count=len(results),
            duration_ms=int(_elapsed * 1000),
        )
        return
…
def format_agent_envelope(
    console: Console,
    *,
    command: str,
    data: Any,
    duration_ms: int | None = None,
    scope: dict | None = None,
    count: int | None = None,
) -> None:
    """Output structured JSON envelope for agent/programmatic use (--json/--agent mode)."""
    envelope: dict[str, Any] = {
        "status": "success",
        "command": command,
    }
    if duration_ms is not None:
        envelope["duration_ms"] = duration_ms
    if scope:
        filtered = {k: v for k, v in scope.items() if v}
        if filtered:
            envelope["scope"] = filtered
    if count is not None:
        envelope["count"] = count
    envelope["data"] = sanitize_agent_data(command, data)
 
    # Surface the unclaimed-Agent-Mode notice (if any) in the envelope so an
    # agent reading the JSON output sees it without inspecting HTTP headers.
    from mem0_cli.state import take_notice
 
    notice = take_notice()
    if notice:
        envelope["mem0_notice"] = notice
 
    console.print_json(json.dumps(envelope, default=str))
…
def sanitize_agent_data(command: str, data: Any) -> Any:
    """Project API response data to minimal relevant fields for agent consumption."""
 
    def pick(obj: dict, keys: list) -> dict:
        return {k: obj[k] for k in keys if k in obj}
 
    if data is None:
        return data
 
    if command == "add":
        items = data if isinstance(data, list) else [data]
        result = []
        for item in items: