Python Agent Core

Copydef read_hook_input() -> dict:
    try:
        value = json.load(sys.stdin)
        return value if isinstance(value, dict) else {}
    except (json.JSONDecodeError, OSError):
        return {}
 

Copydef first_prompt_memory_output(store: EvidenceStore, hook_input: dict) -> dict:
    """Search once before the agent handles the first prompt in a session."""
    repo, session_id, prompt, is_first_prompt = record_user_prompt(store, hook_input)
    if not is_first_prompt:
        return {}
    try:
        minimum_query_chars = int(os.environ.get("MEM0_CODE_MIN_QUERY_CHARS", "20"))
    except ValueError:
        minimum_query_chars = 20
    if len(prompt.strip()) < max(minimum_query_chars, 1):
        return {}
    result = search_memories(
        store, repo, session_id, bounded(prompt, 6000),
        top_k=5, operation="first-prompt-search", timeout=2,
    )

Copyif not result.memories:
    return {}
context = format_context(
    result.memories,
    "Mem0 found these relevant memories from earlier work in this repository:",
)
telemetry.record(
    "context_injected",
    repo=repo, session_id=session_id, trigger="first-prompt",
    memory_count=len(result.memories), context_chars=len(context),
    prompt_chars=len(prompt),
)
return {
    "hookSpecificOutput": {
        "hookEventName": "UserPromptSubmit",
        "additionalContext": context,
    },
}

Copydef default_record_stop(store: EvidenceStore, hook_input: dict):
    """Record the assistant's response without transcript parsing."""
    session_id = _session_id(hook_input)
    repo = store.repo_for_session(session_id, hook_input.get("cwd"))
    message = redact(hook_input.get("last_assistant_message", "")).strip()
    if message:
        store.record_assistant_response(repo, session_id, message)
    return repo, session_id
 

Copyresponse = payload.get("tool_response")
failed = None
if isinstance(response, dict):
    exit_code = response.get("exit_code")
    if isinstance(exit_code, int) and not isinstance(exit_code, bool):
        failed = exit_code != 0
    elif isinstance(response.get("isError"), bool):
        failed = response["isError"]
    elif isinstance(response.get("success"), bool):
        failed = not response["success"]
if failed:
    payload = {**payload, "error": response}
record_tool(store, payload, failed=failed)

Copydef main() -> int:
    if len(sys.argv) != 2:
        return 2
    handoff_path = Path(sys.argv[1])
    os.environ["MEM0_CODE_HANDOFF_PATH"] = str(handoff_path)
    harness = os.environ.get("MEM0_PLUGIN_HARNESS")
    if harness:
        source_tag = os.environ.get("MEM0_PLUGIN_SOURCE_TAG", "")
        configure_harness(
            harness,
            env_prefix=os.environ.get("MEM0_PLUGIN_ENV_PREFIX", ""),
            data_dir_name=os.environ.get("MEM0_PLUGIN_DATA_DIR_NAME", ""),
            source_tag=source_tag,
        )
        telemetry.init(harness=harness, source_tag=source_tag.upper())
    completed = False

Copytry:
    payload = json.loads(handoff_path.read_text(encoding="utf-8"))
    delay = float(payload.get("delay_seconds") or 0)
    if delay > 0:
        payload.pop("delay_seconds", None)
        temporary = handoff_path.with_suffix(f".{os.getpid()}.tmp")
        try:
            temporary.write_text(json.dumps(payload), encoding="utf-8")
            temporary.replace(handoff_path)
        finally:
            temporary.unlink(missing_ok=True)
        time.sleep(delay)
        if not handoff_path.exists():
            return 0

Copyhook_input = payload.get("hook_input") or {}
reason = str(payload.get("reason") or "checkpoint")
wait_for_inflight = bool(payload.get("wait_for_inflight"))
store = EvidenceStore()
try:
    if wait_for_inflight:
        session_id = str(
            hook_input.get("session_id") or "unknown-session"
        )
        repo = store.repo_for_session(session_id, hook_input.get("cwd"))
        deadline = time.monotonic() + float(
            os.environ.get("MEM0_CODE_EXTRACTION_WAIT_SECONDS", "120")
        )
        while (
            store.has_inflight_flush(repo.identity, session_id)
            and time.monotonic() < deadline
        ):
            touch_handoff_heartbeat()
            time.sleep(0.25)

Copydeadline = time.monotonic() + float(
    os.environ.get("MEM0_CODE_EXTRACTION_WAIT_SECONDS", "120")
)
while (
    store.has_inflight_flush(repo.identity, session_id)
    and time.monotonic() < deadline
):
    touch_handoff_heartbeat()
    time.sleep(0.25)

Copydef touch_handoff_heartbeat() -> None:
    """Mark the worker's handoff file alive so recovery does not relaunch it."""
    path = os.environ.get("MEM0_CODE_HANDOFF_PATH", "")
    if not path:
        return
    try:
        os.utime(path)
    except OSError:
        pass

Copy    # Hooks capture the conversation before handoff; the worker only flushes it.
    result = checkpoint_session(store, hook_input, reason)
    print(json.dumps(result, sort_keys=True), flush=True)
    completed = result.get("status") in {
        "semantic-succeeded",
        "explicitly-stored",
        "nothing-to-flush",
    }
finally:
    store.close()
return 0

Copyfinally:
    telemetry.flush()
    if completed:
        try:
            handoff_path.unlink()
        except OSError:
            pass
    elif handoff_path.suffix == ".running":
        try:
            handoff_path.replace(handoff_path.with_suffix(".json"))
        except OSError:
            pass
 

Copydef handle_request(message: Any) -> dict[str, Any] | None:
    if not isinstance(message, dict):
        return None
    request_id = message.get("id")
    method = message.get("method")
 
    if method == "notifications/initialized":
        return None
    if method == "initialize":
        requested = (message.get("params") or {}).get("protocolVersion")
        return {
            "jsonrpc": "2.0",
            "id": request_id,
            "result": {
                "protocolVersion": requested or PROTOCOL_VERSION,
                "capabilities": {"tools": {"listChanged": False}},
                "serverInfo": {"name": "mem0", "version": PLUGIN_VERSION},
            },
        }
    if method == "ping":
        return {"jsonrpc": "2.0", "id": request_id, "result": {}}
    if method == "tools/list":
        return {
            "jsonrpc": "2.0",
            "id": request_id,
            "result": {
                "tools": [
                    {
                        "name": TOOL_NAME,
                        "description": TOOL_DESCRIPTION,
                        "inputSchema": TOOL_SCHEMA,
                        "annotations": {
                            "readOnlyHint": True,
                            "idempotentHint": True,
                            "openWorldHint": True,
                        },
                    }
                ]
            },
        }

Copydef call_search_memories(arguments: Any, cwd: str | None = None) -> str:
    query, top_k, category, scope, run_id = _validate_arguments(arguments)
    repo = resolve_repo(cwd or os.environ.get("CLAUDE_PROJECT_DIR") or os.getcwd())
    result = search_memories(
        None,
        repo,
        None,
        query,
        top_k=top_k,
        category=category,
        scope=scope,
        run_id=run_id,
        operation="mcp-search",
    )
    return format_search_result(result)
 
 
def _workspace_cwd(params: dict[str, Any]) -> str | None:
    meta = params.get("_meta")
    if not isinstance(meta, dict):
        return None
    metadata = meta.get("x-codex-turn-metadata")
    if not isinstance(metadata, dict):
        return None
    workspaces = metadata.get("workspaces") or {}
    if isinstance(workspaces, dict):
        return next((path for path in workspaces if isinstance(path, str) and path), None)
    return None

Copydef _print_status(value: dict) -> None:
    last = value.get("last_operation") or {}
    print(f"Mem0: {'paused' if value['paused'] else 'active'}")
    print(f"Repository: {value['repo_id']}")
    print(f"Local data: {value['data_dir']}")
    print(f"API key: {'configured' if value['api_key_configured'] else 'missing'}")
    print(
        "Saved on this computer: "
        f"{value['events']} session details, {value['flushes']} memory updates"
    )
    print(
        f"Used in this repository: {value['retrievals']} memories returned, "
        f"{value['subagent_runs']} subagent runs"
    )
    if last:
        item_label = ""
        if last["operation"] in {"flush", "flush-retry"}:
            item_label = f", {last['item_count']} memories"
        operation = (
            "memory update"
            if last["operation"] in {"flush", "flush-retry"}
            else last["operation"].replace("-", " ")
        )
        print(
            f"Last {operation}: "
            f"{'succeeded' if last['success'] else 'failed'} "
            f"({last['duration_ms']:.1f} ms{item_label})"
        )
    subagent = value.get("last_subagent") or {}
    if subagent:
        state = "finished" if subagent.get("stopped_at") else "started"
        print(
            "Last subagent: "
            f"{state}, received {subagent['context_chars']} characters of memory, "
            f"agent {subagent['agent_id']}"
        )