feat(cli): add mem0 whoami + mem0 agent-rush subcommands

Mirrors implementation across Node (TypeScript) and Python CLIs.

mem0 whoami:
  Prints the agent's default_user_id from local config — the AGENTRUSH
  leaderboard identifier. Used by players to cross-verify their row.

mem0 agent-rush <add|search>:
  Wraps the platform's /v1/agent-rush/{memories/, memories/search/}
  endpoints. Project routing is implicit (server-resolved); no flags.
  Pretty-prints platform error codes into actionable hints (e.g.
  agentrush_search_first → "Run 3 'mem0 agent-rush search' commands
  before adding.").

Adds X-Mem0-Mode: agent-rush header for telemetry. Reuses existing
agent-mode API key from config; surfaces a graceful error if config
is missing or the key was created outside agent-mode.

Platform PR: mem0ai/platform#2804
This commit is contained in:
Mgeeeek
2026-05-20 03:31:44 +05:30
parent 843ab82905
commit abc57323c2
6 changed files with 326 additions and 0 deletions
+59
View File
@@ -914,6 +914,65 @@ def identify(
run_identify(name)
@app.command(name="whoami", rich_help_panel="Setup")
def whoami_cmd() -> None:
"""Print your AGENTRUSH identifier (default_user_id).
Example:
mem0 whoami
"""
from mem0_cli.commands.whoami_cmd import run_whoami
run_whoami()
# ── AGENTRUSH sub-app ─────────────────────────────────────────────────────
agent_rush_app = typer.Typer(
name="agent-rush",
help="AGENTRUSH game commands",
no_args_is_help=True,
rich_markup_mode="rich",
)
@agent_rush_app.callback(invoke_without_command=True)
def _agent_rush_callback(ctx: typer.Context) -> None:
if ctx.invoked_subcommand:
_fire_telemetry(f"agent-rush.{ctx.invoked_subcommand}")
@agent_rush_app.command(name="add")
def agent_rush_add(
content: str = typer.Argument(..., help="Memory content (50–1000 characters, no URLs)."),
) -> None:
"""Submit a memory to AGENTRUSH.
Example:
mem0 agent-rush add "I enjoy solving constraint-satisfaction problems."
"""
from mem0_cli.commands.agent_rush_cmd import run_agent_rush_add
run_agent_rush_add(content)
@agent_rush_app.command(name="search")
def agent_rush_search(
query: str = typer.Argument(..., help="Search query."),
) -> None:
"""Search AGENTRUSH memories.
Example:
mem0 agent-rush search "constraint satisfaction"
"""
from mem0_cli.commands.agent_rush_cmd import run_agent_rush_search
run_agent_rush_search(query)
app.add_typer(agent_rush_app, name="agent-rush", rich_help_panel="Setup")
# (entity_app registered at module level, below sub-group definitions)
@@ -0,0 +1,86 @@
"""mem0 agent-rush — AGENTRUSH game commands.
Wraps the platform's /v1/agent-rush/{memories/, memories/search/} endpoints.
Hardcoded routing; no flags needed.
"""
from __future__ import annotations
import httpx
import typer
from rich.console import Console
from mem0_cli.branding import BRAND_COLOR, print_error, print_success
from mem0_cli.config import load_config
console = Console()
err_console = Console(stderr=True)
_SOURCE_HEADERS = {
"X-Mem0-Source": "cli",
"X-Mem0-Client-Language": "python",
"X-Mem0-Mode": "agent-rush",
}
_ERROR_HINTS = {
"agentrush_search_first": "Run 3 'mem0 agent-rush search' commands before adding.",
"agentrush_search_quota": "You've used your 3 lifetime searches.",
"agentrush_add_quota": "You've used your 3 lifetime adds.",
"agentrush_not_agent_mode": "Re-run 'mem0 init --agent' to bootstrap an agent-mode key.",
"agentrush_length": "Memory text must be 50-1000 characters.",
"agentrush_no_urls": "URLs are not allowed.",
"agentrush_blocklist": "Content contains a blocked term.",
"agentrush_global_quota": "Event-wide cap reached. Try again later.",
"agentrush_not_provisioned": "AGENTRUSH is not provisioned in this environment.",
}
def _call(path: str, body: dict) -> dict:
config = load_config()
if not config.platform.api_key:
print_error(err_console, "Not initialized. Run `mem0 init --agent` first.")
raise typer.Exit(1)
base_url = (config.platform.base_url or "https://api.mem0.ai").rstrip("/")
try:
with httpx.Client(timeout=30.0) as client:
resp = client.post(
f"{base_url}{path}",
headers={
**_SOURCE_HEADERS,
"Authorization": f"Token {config.platform.api_key}",
"Content-Type": "application/json",
},
json=body,
)
except httpx.HTTPError as exc:
print_error(err_console, f"Network error: {exc}")
raise typer.Exit(1) from exc
try:
data = resp.json()
except Exception:
data = {}
if resp.status_code >= 400:
code = (data.get("error") or {}).get("code", "unknown") if isinstance(data, dict) else "unknown"
print_error(err_console, f"AGENTRUSH error: {code}")
hint = _ERROR_HINTS.get(code)
if hint:
console.print(f" [dim]{hint}[/dim]")
raise typer.Exit(1)
return data
def run_agent_rush_add(content: str) -> None:
result = _call("/v1/agent-rush/memories/", {"content": content})
event_id = result.get("event_id", "?")
print_success(console, f"Memory submitted (event_id: {event_id})")
def run_agent_rush_search(query: str) -> None:
result = _call("/v1/agent-rush/memories/search/", {"query": query})
memories = result.get("results") or result.get("memories") or []
if not memories:
console.print("[dim](no results)[/dim]")
return
for i, m in enumerate(memories[:5], start=1):
text = m.get("memory") if isinstance(m, dict) else str(m)
console.print(f" {i}. {text}")
@@ -0,0 +1,25 @@
"""mem0 whoami — print the active agent's default_user_id (AGENTRUSH identifier)."""
from __future__ import annotations
import typer
from rich.console import Console
from mem0_cli.branding import BRAND_COLOR, print_error, print_info
from mem0_cli.config import load_config
console = Console()
err_console = Console(stderr=True)
def run_whoami() -> None:
config = load_config()
session_id = config.platform.default_user_id if config.platform else None
if not session_id:
print_error(
err_console,
"No default_user_id found. Run `mem0 init --agent` first.",
)
raise typer.Exit(1)
console.print(f"Your AGENTRUSH identifier: [{BRAND_COLOR}]{session_id}[/{BRAND_COLOR}]")
print_info(console, "Find your row at https://mem0.ai/agentrush")