hermes-agent/tools/memory_backend_tool.py

"""Memory Backend Tool — manage cross-session memory backends.

Provides store/retrieve/query/evaluate/list actions for the
pluggable memory backend system.
"""

import json
import logging
from typing import Optional

from tools.registry import registry

logger = logging.getLogger(__name__)


def memory_backend(
    action: str,
    user_id: str = "default",
    key: str = None,
    value: str = None,
    query_text: str = None,
    metadata: dict = None,
) -> str:
    """Manage cross-session memory backends.

    Actions:
      store    — store a user preference/pattern
      retrieve — retrieve a specific memory by key
      query    — search memories by text
      list     — list all keys for a user
      delete   — delete a memory entry
      info     — show current backend info
      evaluate — run evaluation framework comparing backends
    """
    from agent.memory import get_memory_backend

    backend = get_memory_backend()

    if action == "info":
        return json.dumps({
            "success": True,
            "backend": backend.backend_name,
            "is_cloud": backend.is_cloud,
            "available": backend.is_available(),
        })

    if action == "store":
        if not key or value is None:
            return json.dumps({"success": False, "error": "key and value are required for 'store'."})
        success = backend.store(user_id, key, value, metadata)
        return json.dumps({"success": success, "key": key})

    if action == "retrieve":
        if not key:
            return json.dumps({"success": False, "error": "key is required for 'retrieve'."})
        entry = backend.retrieve(user_id, key)
        if entry is None:
            return json.dumps({"success": False, "error": f"No memory found for key '{key}'."})
        return json.dumps({
            "success": True,
            "key": entry.key,
            "value": entry.value,
            "metadata": entry.metadata,
            "updated_at": entry.updated_at,
        })

    if action == "query":
        if not query_text:
            return json.dumps({"success": False, "error": "query_text is required for 'query'."})
        results = backend.query(user_id, query_text)
        return json.dumps({
            "success": True,
            "results": [
                {"key": e.key, "value": e.value, "metadata": e.metadata}
                for e in results
            ],
            "count": len(results),
        })

    if action == "list":
        keys = backend.list_keys(user_id)
        return json.dumps({"success": True, "keys": keys, "count": len(keys)})

    if action == "delete":
        if not key:
            return json.dumps({"success": False, "error": "key is required for 'delete'."})
        success = backend.delete(user_id, key)
        return json.dumps({"success": success})

    if action == "evaluate":
        from agent.memory.evaluation import evaluate_backends
        report = evaluate_backends()
        return json.dumps({
            "success": True,
            **report,
        })

    return json.dumps({
        "success": False,
        "error": f"Unknown action '{action}'. Use: store, retrieve, query, list, delete, info, evaluate",
    })


MEMORY_BACKEND_SCHEMA = {
    "name": "memory_backend",
    "description": (
        "Manage cross-session memory backends for user preference persistence. "
        "Pluggable architecture supports local SQLite (default, zero cloud dependency) "
        "and optional Honcho cloud backend (requires HONCHO_API_KEY).\n\n"
        "Actions:\n"
        "  store    — store a user preference/pattern\n"
        "  retrieve — retrieve a specific memory by key\n"
        "  query    — search memories by text\n"
        "  list     — list all keys for a user\n"
        "  delete   — delete a memory entry\n"
        "  info     — show current backend info\n"
        "  evaluate — run evaluation framework comparing backends"
    ),
    "parameters": {
        "type": "object",
        "properties": {
            "action": {
                "type": "string",
                "enum": ["store", "retrieve", "query", "list", "delete", "info", "evaluate"],
                "description": "The action to perform.",
            },
            "user_id": {
                "type": "string",
                "description": "User identifier for memory operations (default: 'default').",
            },
            "key": {
                "type": "string",
                "description": "Memory key for store/retrieve/delete.",
            },
            "value": {
                "type": "string",
                "description": "Value to store.",
            },
            "query_text": {
                "type": "string",
                "description": "Search text for query action.",
            },
            "metadata": {
                "type": "object",
                "description": "Optional metadata dict for store.",
            },
        },
        "required": ["action"],
    },
}

registry.register(
    name="memory_backend",
    toolset="skills",
    schema=MEMORY_BACKEND_SCHEMA,
    handler=lambda args, **kw: memory_backend(
        action=args.get("action", ""),
        user_id=args.get("user_id", "default"),
        key=args.get("key"),
        value=args.get("value"),
        query_text=args.get("query_text"),
        metadata=args.get("metadata"),
    ),
    emoji="🧠",
)