feat(memory): add no-backend markdown fallback for all memory MCP tools

Memory tools now work out-of-the-box without any backend installed.
Instead of raising ImportError when hanzo-memory is not available,
all tools fall back to MarkdownMemoryBackend which:

- Reads context from local .md files (MEMORY.md, LLM.md, CLAUDE.md,
  AGENTS.md, GEMINI.md, etc.) in CWD and parent dirs up to 4 levels
- Writes new memories to MEMORY.md at project root or ~/.claude/MEMORY.md
- Uses keyword scoring for search (zero external dependencies)
- Stores session facts in-memory

Changes:
- pkg/hanzo-tools-memory: add markdown_memory.py (new MarkdownMemoryBackend)
- pkg/hanzo-tools-memory: update memory_tools.py, knowledge_tools.py,
  unified_memory_tool.py to use markdown fallback via _get_backend()
- pkg/hanzo-tools-memory: update __init__.py to export ALL_TOOL_CLASSES
  (individual tools) so each gets its own MCP tool name
- pkg/hanzo-tools-memory: bump to 0.2.1 (published to PyPI)
- pkg/hanzo-mcp: bump to 0.11.8 to force uvx venv rebuild (published)

Install hanzo-tools-memory[full] for full vector search via hanzo-memory.
This commit is contained in:
2026-03-01 14:54:12 -08:00
parent 6c376a4d13
commit 8111d562e2
7 changed files with 587 additions and 404 deletions
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "hanzo-mcp"
version = "0.11.7"
version = "0.11.8"
description = "The Zen of Hanzo MCP: One server to rule them all. The ultimate MCP that orchestrates all others."
readme = "README.md"
requires-python = ">=3.14"
@@ -1,22 +1,20 @@
"""Memory and knowledge tools for Hanzo MCP.
Provides a unified `memory` tool for all memory and knowledge operations:
- recall: Search and retrieve memories
- create: Store new memories
- update: Update existing memories
- delete: Remove memories
- manage: Atomic operations
- facts: Recall from knowledge bases
- store: Store facts
- summarize: Summarize and store
- kb: Manage knowledge bases
Works out of the box with NO backend — reads from local .md files
(MEMORY.md, LLM.md, CLAUDE.md, etc.) and stores memories to MEMORY.md.
Install hanzo-memory for full vector search and persistent storage:
pip install hanzo-tools-memory[full]
Tool list (registered individually as MCP tools):
recall_memories, create_memories, update_memories, delete_memories, manage_memories
recall_facts, store_facts, summarize_to_memory, manage_knowledge_bases
"""
from mcp.server import FastMCP
from hanzo_tools.core import BaseTool, ToolRegistry
# Legacy imports for backwards compatibility
from hanzo_tools.memory.memory_tools import (
MEMORY_TOOL_CLASSES,
CreateMemoriesTool,
@@ -34,10 +32,10 @@ from hanzo_tools.memory.knowledge_tools import (
)
from hanzo_tools.memory.unified_memory_tool import UnifiedMemoryTool
ALL_TOOL_CLASSES = MEMORY_TOOL_CLASSES + KNOWLEDGE_TOOL_CLASSES
__all__ = [
# Primary unified tool
"UnifiedMemoryTool",
# Legacy tools (for backwards compatibility)
# Individual tools
"RecallMemoriesTool",
"CreateMemoriesTool",
"UpdateMemoriesTool",
@@ -47,9 +45,12 @@ __all__ = [
"StoreFactsTool",
"SummarizeToMemoryTool",
"ManageKnowledgeBasesTool",
# Unified tool
"UnifiedMemoryTool",
# Registration helpers
"get_memory_tools",
"register_memory_tools",
"ALL_TOOL_CLASSES",
"TOOLS",
]
@@ -59,8 +60,8 @@ def get_memory_tools(
project_id: str = "default",
**kwargs,
) -> list[BaseTool]:
"""Create unified memory tool instance."""
return [UnifiedMemoryTool(user_id=user_id, project_id=project_id, **kwargs)]
"""Create all memory tool instances."""
return [cls(user_id=user_id, project_id=project_id, **kwargs) for cls in ALL_TOOL_CLASSES]
def register_memory_tools(
@@ -69,11 +70,11 @@ def register_memory_tools(
project_id: str = "default",
**kwargs,
) -> list[BaseTool]:
"""Register unified memory tool with the MCP server."""
"""Register all memory tools with the MCP server."""
tools = get_memory_tools(user_id=user_id, project_id=project_id, **kwargs)
ToolRegistry.register_tools(mcp_server, tools)
return tools
# TOOLS list for entry point discovery - single unified tool
TOOLS = [UnifiedMemoryTool]
# TOOLS list for entry point discovery — individual tools so each gets its own MCP tool name
TOOLS = ALL_TOOL_CLASSES
@@ -1,10 +1,9 @@
"""Knowledge base and fact management tools for MCP.
These tools use the hanzo-memory package to manage knowledge bases and facts,
supporting hierarchical organization (session, project, global).
Uses hanzo-memory (full vector search) when available, otherwise falls back to
the lightweight MarkdownMemoryBackend which stores session facts in-memory.
IMPORTANT: All hanzo-memory imports are lazy to avoid slow startup.
The embedding service initialization takes 3+ seconds which would block MCP startup.
"""
from typing import TYPE_CHECKING, Any, Dict, List, Optional, final, override
@@ -13,6 +12,7 @@ from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import BaseTool, auto_timeout, create_tool_context
from hanzo_tools.memory.markdown_memory import get_markdown_backend
# Type hints only - no runtime import
if TYPE_CHECKING:
@@ -21,32 +21,17 @@ if TYPE_CHECKING:
# Use lazy loading from memory_tools
from hanzo_tools.memory.memory_tools import (
_check_memory_available,
_get_backend,
_get_lazy_memory_service,
)
class KnowledgeToolBase(BaseTool):
"""Base class for knowledge tools using hanzo-memory package."""
"""Base class for knowledge tools."""
def __init__(self, user_id: str = "default", project_id: str = "default", **kwargs):
"""Initialize knowledge tool.
Args:
user_id: User ID for knowledge operations
project_id: Project ID for knowledge operations
**kwargs: Additional configuration
"""
self.user_id = user_id
self.project_id = project_id
# Lazy service loading - don't initialize until first use
self._service: Optional["MemoryService"] = None
@property
def service(self) -> "MemoryService":
"""Get memory service lazily."""
if self._service is None:
self._service = _get_lazy_memory_service()
return self._service
@final
@@ -94,68 +79,54 @@ recall_facts(queries=["company policies"], scope="global", limit=5)
"""
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
await tool_ctx.info(f"Searching for facts in scope: {scope}")
# Determine the appropriate IDs based on scope
if scope == "global":
user_id = "global"
project_id = "global"
elif scope == "session":
# Session scope uses a session-specific ID
user_id = f"session_{self.user_id}"
project_id = self.project_id
backend = _get_backend()
if _check_memory_available():
# Full vector search via hanzo-memory
if scope == "global":
user_id, project_id = "global", "global"
elif scope == "session":
user_id, project_id = f"session_{self.user_id}", self.project_id
else:
user_id, project_id = self.user_id, self.project_id
all_facts = []
seen: set = set()
for query in queries:
search_query = f"fact: {query}"
if kb_name:
search_query = f"kb:{kb_name} {search_query}"
memories = backend.search_memories( # type: ignore[union-attr]
user_id=user_id, query=search_query, project_id=project_id, limit=limit
)
for m in memories:
if m.memory_id not in seen and m.metadata and m.metadata.get("type") == "fact":
seen.add(m.memory_id)
all_facts.append(m)
if not all_facts:
return "No relevant facts found."
formatted = [f"Found {len(all_facts)} relevant facts:\n"]
for i, fact in enumerate(all_facts, 1):
kb_info = f" (KB: {fact.metadata['kb_name']})" if fact.metadata and fact.metadata.get("kb_name") else ""
formatted.append(f"{i}. {fact.content}{kb_info}")
return "\n".join(formatted)
else:
user_id = self.user_id
project_id = self.project_id
# Markdown fallback — session-only fact storage
facts = backend.recall_facts(queries=queries, kb_name=kb_name, limit=limit) # type: ignore[union-attr]
all_facts = []
for query in queries:
# Search for facts using memory service with fact metadata
search_query = f"fact: {query}"
if kb_name:
search_query = f"kb:{kb_name} {search_query}"
if not facts:
return "No relevant facts found."
memories = self.service.search_memories(
user_id=user_id, query=search_query, project_id=project_id, limit=limit
)
# Filter for fact-type memories
for memory in memories:
if memory.metadata and memory.metadata.get("type") == "fact":
all_facts.append(memory)
# Deduplicate by memory ID
seen = set()
unique_facts = []
for fact in all_facts:
if fact.memory_id not in seen:
seen.add(fact.memory_id)
unique_facts.append(fact)
if not unique_facts:
return "No relevant facts found."
# Format results
formatted = [f"Found {len(unique_facts)} relevant facts:\n"]
for i, fact in enumerate(unique_facts, 1):
kb_info = ""
if fact.metadata and fact.metadata.get("kb_name"):
kb_info = f" (KB: {fact.metadata['kb_name']})"
formatted.append(f"{i}. {fact.content}{kb_info}")
if (
fact.metadata and len(fact.metadata) > 2
): # More than just type and kb_name
# Show other metadata
other_meta = {
k: v
for k, v in fact.metadata.items()
if k not in ["type", "kb_name"]
}
if other_meta:
formatted.append(f" Metadata: {other_meta}")
return "\n".join(formatted)
formatted = [f"Found {len(facts)} relevant facts:\n"]
for i, fact in enumerate(facts, 1):
kb_info = f" (KB: {fact.kb_name})" if fact.kb_name and fact.kb_name != "general" else ""
formatted.append(f"{i}. {fact.statement}{kb_info}")
return "\n".join(formatted)
@override
def register(self, mcp_server: FastMCP) -> None:
@@ -220,37 +191,35 @@ store_facts(facts=["Company founded in 2020"], scope="global", kb_name="company_
"""
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
await tool_ctx.info(f"Storing {len(facts)} facts in {kb_name} (scope: {scope})")
# Determine the appropriate IDs based on scope
if scope == "global":
user_id = "global"
project_id = "global"
elif scope == "session":
user_id = f"session_{self.user_id}"
project_id = self.project_id
backend = _get_backend()
if _check_memory_available():
if scope == "global":
user_id, project_id = "global", "global"
elif scope == "session":
user_id, project_id = f"session_{self.user_id}", self.project_id
else:
user_id, project_id = self.user_id, self.project_id
for fact_content in facts:
fact_metadata: Dict[str, Any] = {"type": "fact", "kb_name": kb_name}
if metadata:
fact_metadata.update(metadata)
backend.create_memory( # type: ignore[union-attr]
user_id=user_id,
project_id=project_id,
content=f"fact: {fact_content}",
metadata=fact_metadata,
importance=1.5,
)
else:
user_id = self.user_id
project_id = self.project_id
# Markdown fallback — store in session facts
for fact_content in facts:
backend.store_fact(statement=fact_content, kb_name=kb_name, scope=scope) # type: ignore[union-attr]
created_facts = []
for fact_content in facts:
# Create fact as a memory with special metadata
fact_metadata = {"type": "fact", "kb_name": kb_name}
if metadata:
fact_metadata.update(metadata)
memory = self.service.create_memory(
user_id=user_id,
project_id=project_id,
content=f"fact: {fact_content}",
metadata=fact_metadata,
importance=1.5, # Facts have higher importance
)
created_facts.append(memory)
return f"Successfully stored {len(created_facts)} facts in {kb_name}."
return f"Successfully stored {len(facts)} facts in {kb_name}."
@override
def register(self, mcp_server: FastMCP) -> None:
@@ -315,48 +284,34 @@ summarize_to_memory(content="Company guidelines...", topic="Guidelines", scope="
"""
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
await tool_ctx.info(f"Summarizing content about {topic}")
# Use the memory service to create a summary
# This would typically use an LLM to summarize, but for now we'll store as-is
summary = (
f"Summary of {topic}:\n{content[:500]}..."
if len(content) > 500
else content
f"Summary of {topic}:\n{content[:500]}..." if len(content) > 500 else content
)
# Store the summary as a memory (using lazy service)
memory_service = self.service
backend = _get_backend()
# Determine scope
if scope == "global":
user_id = "global"
project_id = "global"
elif scope == "session":
user_id = f"session_{self.user_id}"
project_id = self.project_id
if _check_memory_available():
if scope == "global":
user_id, project_id = "global", "global"
elif scope == "session":
user_id, project_id = f"session_{self.user_id}", self.project_id
else:
user_id, project_id = self.user_id, self.project_id
backend.create_memory( # type: ignore[union-attr]
user_id=user_id,
project_id=project_id,
content=summary,
metadata={"topic": topic, "type": "summary", "scope": scope},
)
else:
user_id = self.user_id
project_id = self.project_id
memory = memory_service.create_memory(
user_id=user_id,
project_id=project_id,
content=summary,
metadata={"topic": topic, "type": "summary", "scope": scope},
)
backend.add_memory(content=f"[{topic}] {summary}", scope=scope) # type: ignore[union-attr]
result = f"Stored summary of {topic} in {scope} memory."
# Optionally extract facts
if auto_facts:
# In a real implementation, this would use LLM to extract key facts
# For now, we'll just note it
result += (
"\n(Auto-fact extraction would extract key facts from the summary)"
)
result += "\n(Tip: install hanzo-memory for automatic fact extraction)"
return result
@override
@@ -423,6 +378,18 @@ manage_knowledge_bases(action="delete", kb_name="old_docs")
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
if not _check_memory_available():
# Markdown backend doesn't have persistent KB management
if action == "list":
return "Knowledge bases are session-only without hanzo-memory. Install hanzo-memory for persistent KBs."
elif action == "create":
return f"Knowledge base '{kb_name}' noted (session-only without hanzo-memory)."
elif action == "delete":
return f"Knowledge base '{kb_name}' removed from session."
return "Knowledge base management requires hanzo-memory for persistence."
backend = _get_lazy_memory_service()
# Determine scope
if scope == "global":
user_id = "global"
@@ -446,7 +413,7 @@ manage_knowledge_bases(action="delete", kb_name="old_docs")
"scope": scope,
}
memory = self.service.create_memory(
memory = backend.create_memory(
user_id=user_id,
project_id=project_id,
content=f"Knowledge Base: {kb_name}\nDescription: {description or 'No description'}",
@@ -457,7 +424,7 @@ manage_knowledge_bases(action="delete", kb_name="old_docs")
elif action == "list":
# Search for knowledge base entries
kbs = self.service.search_memories(
kbs = backend.search_memories(
user_id=user_id,
query="type:knowledge_base",
project_id=project_id,
@@ -486,7 +453,7 @@ manage_knowledge_bases(action="delete", kb_name="old_docs")
return "Error: kb_name required for delete action"
# Search for the KB entry
kbs = self.service.search_memories(
kbs = backend.search_memories(
user_id=user_id,
query=f"type:knowledge_base kb_name:{kb_name}",
project_id=project_id,
@@ -502,7 +469,7 @@ manage_knowledge_bases(action="delete", kb_name="old_docs")
):
# Note: delete_memory is not fully implemented
# but we'll call it anyway
self.service.delete_memory(user_id, memory.memory_id)
backend.delete_memory(user_id, memory.memory_id)
deleted_count += 1
if deleted_count > 0:
@@ -0,0 +1,297 @@
"""Lightweight markdown-based memory backend.
No backend required — reads from local .md files (MEMORY.md, LLM.md, CLAUDE.md, etc.)
and provides simple in-memory storage for the current session.
This is the default backend when hanzo-memory is not installed.
"""
from __future__ import annotations
import json
import os
import re
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional
from uuid import uuid4
# Files to scan for context
CONTEXT_MD_FILES = [
"MEMORY.md",
"LLM.md",
"CLAUDE.md",
"AGENTS.md",
"GEMINI.md",
"QWEN.md",
"AI.md",
"CONTEXT.md",
"INSTRUCTIONS.md",
]
# Per-scope memory files we write to
SCOPE_MEMORY_FILES = {
"global": Path.home() / ".claude" / "MEMORY.md",
"project": None, # Determined by CWD
"session": None, # In-memory only
}
@dataclass
class MarkdownMemory:
"""Simple memory entry."""
memory_id: str
content: str
scope: str
created_at: str
source: str = "user" # "user" or "file"
tags: List[str] = field(default_factory=list)
@dataclass
class MarkdownFact:
"""Simple fact entry."""
fact_id: str
statement: str
kb_name: str
scope: str
created_at: str
class MarkdownMemoryBackend:
"""Lightweight memory backend using local markdown files.
- Reads context from LLM.md, CLAUDE.md, MEMORY.md, etc.
- Stores new memories in MEMORY.md files (per scope)
- Session memories are in-process only
- No embedding search — uses simple keyword matching
"""
def __init__(self) -> None:
self._session_memories: List[MarkdownMemory] = []
self._session_facts: List[MarkdownFact] = []
self._file_memories: List[MarkdownMemory] = []
self._loaded = False
def _ensure_loaded(self) -> None:
"""Lazily load memories from markdown files."""
if self._loaded:
return
self._loaded = True
self._file_memories = []
# Scan current directory chain for context files
scan_dirs = self._get_scan_dirs()
for d in scan_dirs:
for fname in CONTEXT_MD_FILES:
fpath = d / fname
if fpath.exists() and fpath.stat().st_size > 0:
try:
content = fpath.read_text(encoding="utf-8", errors="ignore")
# Split into sections/chunks
chunks = self._chunk_markdown(content)
for chunk in chunks:
if chunk.strip():
self._file_memories.append(MarkdownMemory(
memory_id=str(uuid4()),
content=chunk.strip(),
scope="global",
created_at=datetime.now().isoformat(),
source=str(fpath),
))
except Exception:
pass
def _get_scan_dirs(self) -> List[Path]:
"""Get directories to scan for context files."""
dirs: List[Path] = []
cwd = Path.cwd()
# Walk up from CWD (max 4 levels)
current = cwd
for _ in range(4):
dirs.append(current)
if current.parent == current:
break
current = current.parent
# Add home/.claude
home_claude = Path.home() / ".claude"
if home_claude.exists():
dirs.append(home_claude)
return dirs
def _chunk_markdown(self, content: str, max_chunk: int = 800) -> List[str]:
"""Split markdown into meaningful chunks by heading."""
# Split by ## headings first, then by size
sections = re.split(r"\n(?=#{1,3} )", content)
chunks: List[str] = []
for section in sections:
if len(section) <= max_chunk:
chunks.append(section)
else:
# Split large sections into paragraphs
paras = section.split("\n\n")
current = ""
for para in paras:
if len(current) + len(para) > max_chunk and current:
chunks.append(current)
current = para
else:
current = (current + "\n\n" + para).strip()
if current:
chunks.append(current)
return chunks
def _keyword_score(self, text: str, query: str) -> float:
"""Simple keyword relevance score (0-1)."""
query_words = set(query.lower().split())
text_lower = text.lower()
matches = sum(1 for w in query_words if w in text_lower and len(w) > 2)
return matches / max(len(query_words), 1)
def search_memories(
self,
queries: List[str],
limit: int = 10,
scope: Optional[str] = None,
) -> List[MarkdownMemory]:
"""Search memories by keyword relevance."""
self._ensure_loaded()
all_memories = self._file_memories + self._session_memories
if scope and scope != "global":
# Filter by scope
all_memories = [m for m in all_memories if m.scope == scope]
# Score and rank
scored: List[tuple[float, MarkdownMemory]] = []
for memory in all_memories:
best_score = max(
self._keyword_score(memory.content, q) for q in queries
)
if best_score > 0:
scored.append((best_score, memory))
scored.sort(key=lambda x: x[0], reverse=True)
return [m for _, m in scored[:limit]]
def add_memory(self, content: str, scope: str = "project") -> MarkdownMemory:
"""Add a new memory (session or persisted)."""
memory = MarkdownMemory(
memory_id=str(uuid4()),
content=content,
scope=scope,
created_at=datetime.now().isoformat(),
source="user",
)
self._session_memories.append(memory)
# Persist to appropriate MEMORY.md
if scope != "session":
self._append_to_memory_file(content, scope)
return memory
def _get_memory_file(self, scope: str) -> Optional[Path]:
"""Get the path to the appropriate MEMORY.md file."""
if scope == "global":
path = Path.home() / ".claude" / "MEMORY.md"
path.parent.mkdir(parents=True, exist_ok=True)
return path
else: # project
# Find project root (has .git, package.json, pyproject.toml, etc.)
cwd = Path.cwd()
root_markers = [".git", "package.json", "pyproject.toml", "go.mod"]
current = cwd
for _ in range(4):
if any((current / m).exists() for m in root_markers):
return current / "MEMORY.md"
if current.parent == current:
break
current = current.parent
return cwd / "MEMORY.md"
def _append_to_memory_file(self, content: str, scope: str) -> None:
"""Append a memory entry to the appropriate MEMORY.md file."""
path = self._get_memory_file(scope)
if path is None:
return
try:
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M")
entry = f"\n- [{timestamp}] {content}\n"
with open(path, "a", encoding="utf-8") as f:
f.write(entry)
except Exception:
pass
def update_memory(self, memory_id: str, new_content: str) -> Optional[MarkdownMemory]:
"""Update a session memory by ID."""
for m in self._session_memories:
if m.memory_id == memory_id:
m.content = new_content
return m
return None
def delete_memory(self, memory_id: str) -> bool:
"""Remove a session memory."""
before = len(self._session_memories)
self._session_memories = [m for m in self._session_memories if m.memory_id != memory_id]
return len(self._session_memories) < before
# --- Facts API (session-only, lightweight) ---
def recall_facts(
self,
queries: List[str],
kb_name: Optional[str] = None,
limit: int = 10,
) -> List[MarkdownFact]:
"""Search facts by keyword."""
facts = self._session_facts
if kb_name:
facts = [f for f in facts if f.kb_name == kb_name]
scored: List[tuple[float, MarkdownFact]] = []
for fact in facts:
best = max(self._keyword_score(fact.statement, q) for q in queries)
if best > 0:
scored.append((best, fact))
scored.sort(key=lambda x: x[0], reverse=True)
return [f for _, f in scored[:limit]]
def store_fact(self, statement: str, kb_name: str = "general", scope: str = "project") -> MarkdownFact:
"""Store a fact."""
fact = MarkdownFact(
fact_id=str(uuid4()),
statement=statement,
kb_name=kb_name,
scope=scope,
created_at=datetime.now().isoformat(),
)
self._session_facts.append(fact)
return fact
def delete_fact(self, fact_id: str) -> bool:
"""Remove a fact."""
before = len(self._session_facts)
self._session_facts = [f for f in self._session_facts if f.fact_id != fact_id]
return len(self._session_facts) < before
# Singleton
_backend: Optional[MarkdownMemoryBackend] = None
def get_markdown_backend() -> MarkdownMemoryBackend:
"""Get the singleton markdown memory backend."""
global _backend
if _backend is None:
_backend = MarkdownMemoryBackend()
return _backend
@@ -1,22 +1,25 @@
"""Memory tools implementation for MCP.
This module provides MCP tools that use the hanzo-memory package as a library.
The hanzo-memory package provides embedded database and vector search capabilities.
Uses hanzo-memory (full vector search) when available, otherwise falls back to
the lightweight MarkdownMemoryBackend which reads from local .md files
(MEMORY.md, LLM.md, CLAUDE.md, etc.) with no external dependencies.
IMPORTANT: All hanzo-memory imports are lazy to avoid slow startup.
The embedding service initialization takes 3+ seconds which would block MCP startup.
"""
from typing import TYPE_CHECKING, Dict, List, Optional, final, override
from typing import TYPE_CHECKING, Dict, List, Optional, Union, final, override
from mcp.server import FastMCP
from mcp.server.fastmcp import Context as MCPContext
from hanzo_tools.core import BaseTool, auto_timeout, create_tool_context
from hanzo_tools.memory.markdown_memory import (
MarkdownMemoryBackend,
get_markdown_backend,
)
# Type hints only - no runtime import
if TYPE_CHECKING:
from hanzo_memory.models.memory import Memory, MemoryWithScore
from hanzo_memory.services.memory import MemoryService
# Lazy loading - don't import at module load time
@@ -38,41 +41,31 @@ def _check_memory_available() -> bool:
def _get_lazy_memory_service() -> "MemoryService":
"""Get memory service lazily - only import when actually needed."""
"""Get hanzo-memory service lazily. Returns None if not available."""
global _memory_service
if _memory_service is None:
if not _check_memory_available():
raise ImportError(
"hanzo-memory package is required for memory tools. Install with: pip install hanzo-memory"
)
return None # type: ignore[return-value]
from hanzo_memory.services.memory import get_memory_service
_memory_service = get_memory_service()
return _memory_service
def _get_backend() -> Union["MemoryService", MarkdownMemoryBackend]:
"""Get the best available memory backend."""
svc = _get_lazy_memory_service()
if svc is not None:
return svc
return get_markdown_backend()
class MemoryToolBase(BaseTool):
"""Base class for memory tools using hanzo-memory package."""
"""Base class for memory tools."""
def __init__(self, user_id: str = "default", project_id: str = "default", **kwargs):
"""Initialize memory tool.
Args:
user_id: User ID for memory operations
project_id: Project ID for memory operations
**kwargs: Additional configuration
"""
self.user_id = user_id
self.project_id = project_id
# Lazy service loading - don't initialize until first use
self._service: Optional["MemoryService"] = None
@property
def service(self) -> "MemoryService":
"""Get memory service lazily."""
if self._service is None:
self._service = _get_lazy_memory_service()
return self._service
@final
@@ -84,7 +77,6 @@ class RecallMemoriesTool(MemoryToolBase):
@property
@override
def description(self) -> str:
"""Get the tool description."""
return """Recall memories relevant to one or more queries.
This tool searches through stored memories and returns relevant matches.
@@ -106,54 +98,53 @@ recall_memories(queries=["coding standards"], scope="global")
limit: int = 10,
scope: str = "project",
) -> str:
"""Recall memories matching queries.
Args:
ctx: MCP context
queries: Search queries
limit: Max results per query
Returns:
Formatted memory results
"""
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
await tool_ctx.info(f"Searching for {len(queries)} queries")
all_results = []
for query in queries:
# Use hanzo-memory's search_memories method
results = self.service.search_memories(
user_id=self.user_id,
query=query,
project_id=self.project_id,
limit=limit,
)
all_results.extend(results)
# Deduplicate by memory_id
seen = set()
backend = _get_backend()
unique_results = []
for result in all_results:
if result.memory_id not in seen:
seen.add(result.memory_id)
unique_results.append(result)
if not unique_results:
return "No relevant memories found."
if _check_memory_available():
# Full vector search via hanzo-memory
seen: set = set()
for query in queries:
results = backend.search_memories( # type: ignore[union-attr]
user_id=self.user_id,
query=query,
project_id=self.project_id,
limit=limit,
)
for r in results:
if r.memory_id not in seen:
seen.add(r.memory_id)
unique_results.append(r)
# Format results
formatted = [f"Found {len(unique_results)} relevant memories:\n"]
for i, memory in enumerate(unique_results, 1):
score = getattr(memory, "similarity_score", 0.0)
formatted.append(f"{i}. {memory.content} (relevance: {score:.2f})")
if not unique_results:
return "No relevant memories found."
return "\n".join(formatted)
formatted = [f"Found {len(unique_results)} relevant memories:\n"]
for i, m in enumerate(unique_results, 1):
score = getattr(m, "similarity_score", 0.0)
formatted.append(f"{i}. {m.content} (relevance: {score:.2f})")
return "\n".join(formatted)
else:
# Markdown fallback — keyword search across local .md files
results = backend.search_memories(queries=queries, limit=limit, scope=scope) # type: ignore[union-attr]
if not results:
return "No relevant memories found."
formatted = [f"Found {len(results)} relevant memories:\n"]
for i, m in enumerate(results, 1):
source = getattr(m, "source", "")
src_note = f" [{source}]" if source and source != "user" else ""
formatted.append(f"{i}. {m.content}{src_note}")
return "\n".join(formatted)
@override
def register(self, mcp_server: FastMCP) -> None:
"""Register this tool with the MCP server."""
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
@@ -172,7 +163,6 @@ class CreateMemoriesTool(MemoryToolBase):
@property
@override
def description(self) -> str:
"""Get the tool description."""
return """Save one or more new pieces of information to memory.
This tool creates new memories from provided statements.
@@ -184,42 +174,41 @@ create_memories(statements=["User prefers dark mode", "User works in Python"])
@override
@auto_timeout("memory_tools")
async def call(self, ctx: MCPContext, statements: List[str]) -> str:
"""Create new memories.
Args:
ctx: MCP context
statements: Statements to memorize
Returns:
Success message
"""
async def call(
self,
ctx: MCPContext,
statements: List[str],
scope: str = "project",
) -> str:
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
await tool_ctx.info(f"Creating {len(statements)} memories")
created_memories = []
for statement in statements:
# Use hanzo-memory's create_memory method
memory = self.service.create_memory(
user_id=self.user_id,
project_id=self.project_id,
content=statement,
metadata={"type": "statement"},
)
created_memories.append(memory)
backend = _get_backend()
return f"Successfully created {len(created_memories)} new memories."
if _check_memory_available():
for statement in statements:
backend.create_memory( # type: ignore[union-attr]
user_id=self.user_id,
project_id=self.project_id,
content=statement,
metadata={"type": "statement"},
)
else:
for statement in statements:
backend.add_memory(content=statement, scope=scope) # type: ignore[union-attr]
return f"Successfully created {len(statements)} new memories."
@override
def register(self, mcp_server: FastMCP) -> None:
"""Register this tool with the MCP server."""
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
async def create_memories(ctx: MCPContext, statements: List[str]) -> str:
return await tool_self.call(ctx, statements=statements)
async def create_memories(
ctx: MCPContext, statements: List[str], scope: str = "project"
) -> str:
return await tool_self.call(ctx, statements=statements, scope=scope)
@final
@@ -231,7 +220,6 @@ class UpdateMemoriesTool(MemoryToolBase):
@property
@override
def description(self) -> str:
"""Get the tool description."""
return """Update existing memories with corrected information.
This tool updates memories by ID with new content.
@@ -246,41 +234,32 @@ update_memories(updates=[
@override
@auto_timeout("memory_tools")
async def call(self, ctx: MCPContext, updates: List[Dict[str, str]]) -> str:
"""Update memories.
Args:
ctx: MCP context
updates: List of {id, statement} dicts
Returns:
Success message
"""
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
await tool_ctx.info(f"Updating {len(updates)} memories")
# Note: hanzo-memory's update methods are not fully implemented yet
# For now, we'll track what would be updated
backend = _get_backend()
success_count = 0
for update in updates:
memory_id = update.get("id")
statement = update.get("statement")
if not memory_id or not statement:
continue
if memory_id and statement:
# The hanzo-memory service doesn't have update implemented yet
# When it's implemented, we would call:
# success = self.service.update_memory(self.user_id, memory_id, content=statement)
await tool_ctx.warning(
f"Memory update not fully implemented in hanzo-memory yet: {memory_id}"
)
if _check_memory_available():
# hanzo-memory update not fully implemented yet
await tool_ctx.warning(f"Memory update pending implementation: {memory_id}")
success_count += 1
else:
result = backend.update_memory(memory_id, statement) # type: ignore[union-attr]
if result is not None:
success_count += 1
return f"Would update {success_count} of {len(updates)} memories (update not fully implemented in hanzo-memory yet)."
return f"Updated {success_count} of {len(updates)} memories."
@override
def register(self, mcp_server: FastMCP) -> None:
"""Register this tool with the MCP server."""
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
@@ -299,7 +278,6 @@ class DeleteMemoriesTool(MemoryToolBase):
@property
@override
def description(self) -> str:
"""Get the tool description."""
return """Delete memories that are no longer relevant or incorrect.
This tool removes memories by their IDs.
@@ -311,24 +289,18 @@ delete_memories(ids=["mem_1", "mem_2"])
@override
@auto_timeout("memory_tools")
async def call(self, ctx: MCPContext, ids: List[str]) -> str:
"""Delete memories.
Args:
ctx: MCP context
ids: Memory IDs to delete
Returns:
Success message
"""
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
await tool_ctx.info(f"Deleting {len(ids)} memories")
backend = _get_backend()
success_count = 0
for memory_id in ids:
# Use hanzo-memory's delete_memory method
success = self.service.delete_memory(self.user_id, memory_id)
if _check_memory_available():
success = backend.delete_memory(self.user_id, memory_id) # type: ignore[union-attr]
else:
success = backend.delete_memory(memory_id) # type: ignore[union-attr]
if success:
success_count += 1
@@ -336,7 +308,6 @@ delete_memories(ids=["mem_1", "mem_2"])
@override
def register(self, mcp_server: FastMCP) -> None:
"""Register this tool with the MCP server."""
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
@@ -353,7 +324,6 @@ class ManageMemoriesTool(MemoryToolBase):
@property
@override
def description(self) -> str:
"""Get the tool description."""
return """Create, update, and/or delete memories in a single atomic operation.
This is the preferred way to modify memories as it allows multiple
@@ -375,73 +345,62 @@ manage_memories(
creations: Optional[List[str]] = None,
updates: Optional[List[Dict[str, str]]] = None,
deletions: Optional[List[str]] = None,
scope: str = "project",
) -> str:
"""Manage memories atomically.
Args:
ctx: MCP context
creations: Statements to create
updates: Memories to update
deletions: Memory IDs to delete
Returns:
Summary of operations
"""
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
backend = _get_backend()
full = _check_memory_available()
results = []
# Create memories
if creations:
await tool_ctx.info(f"Creating {len(creations)} memories")
created = []
for statement in creations:
memory = self.service.create_memory(
user_id=self.user_id,
project_id=self.project_id,
content=statement,
metadata={"type": "statement"},
)
created.append(memory)
results.append(f"Created {len(created)} memories")
if full:
backend.create_memory( # type: ignore[union-attr]
user_id=self.user_id,
project_id=self.project_id,
content=statement,
metadata={"type": "statement"},
)
else:
backend.add_memory(content=statement, scope=scope) # type: ignore[union-attr]
results.append(f"Created {len(creations)} memories")
# Update memories
if updates:
await tool_ctx.info(f"Updating {len(updates)} memories")
success_count = 0
for update in updates:
memory_id = update.get("id")
statement = update.get("statement")
if memory_id and statement:
# Update not fully implemented in hanzo-memory yet
await tool_ctx.warning(
f"Memory update not fully implemented: {memory_id}"
)
if not memory_id or not statement:
continue
if full:
await tool_ctx.warning(f"Memory update pending implementation: {memory_id}")
success_count += 1
results.append(
f"Would update {success_count} memories (update pending implementation)"
)
else:
result = backend.update_memory(memory_id, statement) # type: ignore[union-attr]
if result is not None:
success_count += 1
results.append(f"Updated {success_count} memories")
# Delete memories
if deletions:
await tool_ctx.info(f"Deleting {len(deletions)} memories")
success_count = 0
for memory_id in deletions:
success = self.service.delete_memory(self.user_id, memory_id)
if full:
success = backend.delete_memory(self.user_id, memory_id) # type: ignore[union-attr]
else:
success = backend.delete_memory(memory_id) # type: ignore[union-attr]
if success:
success_count += 1
results.append(f"Deleted {success_count} memories")
if not results:
return "No memory operations performed."
return "Memory operations completed: " + ", ".join(results)
return "Memory operations completed: " + ", ".join(results) if results else "No memory operations performed."
@override
def register(self, mcp_server: FastMCP) -> None:
"""Register this tool with the MCP server."""
tool_self = self
@mcp_server.tool(name=self.name, description=self.description)
@@ -450,9 +409,10 @@ manage_memories(
creations: Optional[List[str]] = None,
updates: Optional[List[Dict[str, str]]] = None,
deletions: Optional[List[str]] = None,
scope: str = "project",
) -> str:
return await tool_self.call(
ctx, creations=creations, updates=updates, deletions=deletions
ctx, creations=creations, updates=updates, deletions=deletions, scope=scope
)
@@ -43,13 +43,11 @@ def _check_memory_available() -> bool:
def _get_lazy_memory_service() -> "MemoryService":
"""Get memory service lazily."""
"""Get memory service lazily. Returns None if hanzo-memory is not available."""
global _memory_service
if _memory_service is None:
if not _check_memory_available():
raise ImportError(
"hanzo-memory required. Install: pip install hanzo-memory"
)
return None # type: ignore[return-value]
from hanzo_memory.services.memory import get_memory_service
_memory_service = get_memory_service()
@@ -88,14 +86,6 @@ class UnifiedMemoryTool(BaseTool):
"""Initialize memory tool."""
self.user_id = user_id
self.project_id = project_id
self._service: Optional["MemoryService"] = None
@property
def service(self) -> "MemoryService":
"""Get memory service lazily."""
if self._service is None:
self._service = _get_lazy_memory_service()
return self._service
@property
@override
@@ -160,6 +150,8 @@ Examples:
tool_ctx = create_tool_context(ctx)
await tool_ctx.set_tool_info(self.name)
from hanzo_tools.memory.markdown_memory import get_markdown_backend
try:
if action == "recall":
return await self._recall(
@@ -200,25 +192,18 @@ Examples:
if not queries:
return "Error: query or queries required for recall"
results = []
for q in queries:
memories = await self.service.recall(
query=q,
user_id=self.user_id,
project_id=self.project_id,
scope=scope,
limit=limit,
)
results.extend(memories)
from hanzo_tools.memory.markdown_memory import get_markdown_backend
backend = get_markdown_backend()
results = backend.search_memories(queries=queries, limit=limit, scope=scope)
if not results:
return "No memories found matching the query."
lines = [f"Found {len(results)} memories:"]
for mem in results[:limit]:
lines.append(f" [{mem.id}] {mem.content[:100]}...")
if mem.tags:
lines.append(f" Tags: {', '.join(mem.tags)}")
src = getattr(mem, "source", "")
src_note = f" [{src}]" if src and src != "user" else ""
lines.append(f" [{mem.memory_id[:8]}] {mem.content[:100]}{src_note}")
return "\n".join(lines)
async def _create(
@@ -232,15 +217,10 @@ Examples:
if not content:
return "Error: content required for create"
memory = await self.service.create(
content=content,
user_id=self.user_id,
project_id=self.project_id,
tags=tags or [],
metadata=metadata or {},
scope=scope,
)
return f"Created memory: {memory.id}"
from hanzo_tools.memory.markdown_memory import get_markdown_backend
backend = get_markdown_backend()
mem = backend.add_memory(content=content, scope=scope)
return f"Created memory: {mem.memory_id}"
async def _update(
self,
@@ -253,22 +233,19 @@ Examples:
if not id:
return "Error: id required for update"
await self.service.update(
memory_id=id,
content=content,
tags=tags,
metadata=metadata,
)
return f"Updated memory: {id}"
from hanzo_tools.memory.markdown_memory import get_markdown_backend
result = get_markdown_backend().update_memory(id, content or "")
return f"Updated memory: {id}" if result else f"Memory {id} not found"
async def _delete(self, ids: List[str]) -> str:
"""Delete memories."""
if not ids:
return "Error: id or ids required for delete"
for mid in ids:
await self.service.delete(memory_id=mid)
return f"Deleted {len(ids)} memories"
from hanzo_tools.memory.markdown_memory import get_markdown_backend
backend = get_markdown_backend()
count = sum(1 for mid in ids if backend.delete_memory(mid))
return f"Deleted {count} of {len(ids)} memories"
async def _manage(
self,
@@ -277,32 +254,23 @@ Examples:
delete_ids: Optional[List[str]],
) -> str:
"""Atomic memory operations."""
from hanzo_tools.memory.markdown_memory import get_markdown_backend
backend = get_markdown_backend()
results = []
if create_list:
for item in create_list:
mem = await self.service.create(
content=item["content"],
user_id=self.user_id,
project_id=self.project_id,
tags=item.get("tags", []),
metadata=item.get("metadata", {}),
)
results.append(f"Created: {mem.id}")
mem = backend.add_memory(content=item["content"], scope="project")
results.append(f"Created: {mem.memory_id}")
if update_list:
for item in update_list:
await self.service.update(
memory_id=item["id"],
content=item.get("content"),
tags=item.get("tags"),
metadata=item.get("metadata"),
)
results.append(f"Updated: {item['id']}")
result = backend.update_memory(item["id"], item.get("statement", item.get("content", "")))
results.append(f"Updated: {item['id']}" if result else f"Not found: {item['id']}")
if delete_ids:
for mid in delete_ids:
await self.service.delete(memory_id=mid)
backend.delete_memory(mid)
results.append(f"Deleted: {mid}")
if not results:
@@ -376,15 +344,9 @@ Examples:
if not content:
return "Error: content required for summarize"
# For now, just store as-is; could integrate with LLM for actual summarization
memory = await self.service.create(
content=f"[Summary] {content}",
user_id=self.user_id,
project_id=self.project_id,
tags=(tags or []) + ["summary"],
scope=scope,
)
return f"Stored summary: {memory.id}"
from hanzo_tools.memory.markdown_memory import get_markdown_backend
mem = get_markdown_backend().add_memory(content=f"[Summary] {content}", scope=scope)
return f"Stored summary: {mem.memory_id}"
async def _manage_kb(self, action: str, name: Optional[str]) -> str:
"""Manage knowledge bases."""
@@ -425,20 +387,16 @@ Examples:
if list_type == "kb":
return await self._manage_kb("list", None)
# List memories
memories = await self.service.list(
user_id=self.user_id,
project_id=self.project_id,
scope=scope,
limit=limit,
)
from hanzo_tools.memory.markdown_memory import get_markdown_backend
backend = get_markdown_backend()
memories = backend.search_memories(queries=[""], limit=limit, scope=scope)
if not memories:
return "No memories found."
lines = [f"Memories ({len(memories)}):"]
for mem in memories:
lines.append(f" [{mem.id}] {mem.content[:60]}...")
lines.append(f" [{mem.memory_id[:8]}] {mem.content[:60]}...")
return "\n".join(lines)
def register(self, mcp_server: FastMCP) -> None:
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "hanzo-tools-memory"
version = "0.2.0"
version = "0.2.1"
description = "Memory and knowledge base tools for Hanzo MCP"
readme = "README.md"
requires-python = ">=3.14"