| | |
| | """ |
| | Mnemo MCP Server |
| | Provides memory capabilities to Claude and other MCP-compatible LLMs |
| | |
| | Usage: |
| | uvx mnemo-memory # Run as MCP server |
| | |
| | Or in Claude config: |
| | { |
| | "mcpServers": { |
| | "mnemo": {"command": "uvx", "args": ["mnemo-memory"]} |
| | } |
| | } |
| | """ |
| |
|
| | import json |
| | import sys |
| | from typing import Any, Dict, List |
| |
|
| | try: |
| | from mcp.server import Server |
| | from mcp.types import Tool, TextContent |
| | HAS_MCP = True |
| | except ImportError: |
| | HAS_MCP = False |
| | print("Warning: MCP not installed. Install with: pip install mcp", file=sys.stderr) |
| |
|
| | from mnemo import Mnemo, should_inject_memory |
| |
|
| | |
| | memory = Mnemo(use_real_embeddings=True) |
| |
|
| | |
| | TOOLS = [ |
| | { |
| | "name": "mnemo_add", |
| | "description": "Store a new memory. Use this to save important information for later retrieval.", |
| | "inputSchema": { |
| | "type": "object", |
| | "properties": { |
| | "content": { |
| | "type": "string", |
| | "description": "The content to remember" |
| | }, |
| | "metadata": { |
| | "type": "object", |
| | "description": "Optional metadata (tags, source, etc.)" |
| | } |
| | }, |
| | "required": ["content"] |
| | } |
| | }, |
| | { |
| | "name": "mnemo_search", |
| | "description": "Search stored memories. Returns relevant memories ranked by similarity.", |
| | "inputSchema": { |
| | "type": "object", |
| | "properties": { |
| | "query": { |
| | "type": "string", |
| | "description": "Search query" |
| | }, |
| | "top_k": { |
| | "type": "integer", |
| | "description": "Number of results (default: 5)", |
| | "default": 5 |
| | } |
| | }, |
| | "required": ["query"] |
| | } |
| | }, |
| | { |
| | "name": "mnemo_should_inject", |
| | "description": "Check if memory should be injected for this query. Uses context-check algorithm with 90% accuracy.", |
| | "inputSchema": { |
| | "type": "object", |
| | "properties": { |
| | "query": { |
| | "type": "string", |
| | "description": "The query to check" |
| | }, |
| | "context": { |
| | "type": "string", |
| | "description": "Optional additional context" |
| | } |
| | }, |
| | "required": ["query"] |
| | } |
| | }, |
| | { |
| | "name": "mnemo_get_context", |
| | "description": "Get formatted memory context for injection into prompts.", |
| | "inputSchema": { |
| | "type": "object", |
| | "properties": { |
| | "query": { |
| | "type": "string", |
| | "description": "Search query for finding relevant context" |
| | }, |
| | "top_k": { |
| | "type": "integer", |
| | "description": "Number of memories to include (default: 3)", |
| | "default": 3 |
| | } |
| | }, |
| | "required": ["query"] |
| | } |
| | }, |
| | { |
| | "name": "mnemo_list", |
| | "description": "List all stored memories.", |
| | "inputSchema": { |
| | "type": "object", |
| | "properties": {} |
| | } |
| | }, |
| | { |
| | "name": "mnemo_delete", |
| | "description": "Delete a specific memory by ID.", |
| | "inputSchema": { |
| | "type": "object", |
| | "properties": { |
| | "memory_id": { |
| | "type": "string", |
| | "description": "ID of the memory to delete" |
| | } |
| | }, |
| | "required": ["memory_id"] |
| | } |
| | }, |
| | { |
| | "name": "mnemo_stats", |
| | "description": "Get memory system statistics.", |
| | "inputSchema": { |
| | "type": "object", |
| | "properties": {} |
| | } |
| | }, |
| | { |
| | "name": "mnemo_clear", |
| | "description": "Clear all stored memories. Use with caution!", |
| | "inputSchema": { |
| | "type": "object", |
| | "properties": { |
| | "confirm": { |
| | "type": "boolean", |
| | "description": "Must be true to confirm deletion" |
| | } |
| | }, |
| | "required": ["confirm"] |
| | } |
| | } |
| | ] |
| |
|
| |
|
| | def handle_tool_call(name: str, arguments: Dict[str, Any]) -> str: |
| | """Handle a tool call and return result""" |
| | |
| | if name == "mnemo_add": |
| | content = arguments.get("content", "") |
| | metadata = arguments.get("metadata", {}) |
| | mem_id = memory.add(content, metadata) |
| | return json.dumps({ |
| | "status": "success", |
| | "memory_id": mem_id, |
| | "message": f"Memory stored successfully" |
| | }) |
| | |
| | elif name == "mnemo_search": |
| | query = arguments.get("query", "") |
| | top_k = arguments.get("top_k", 5) |
| | results = memory.search(query, top_k=top_k) |
| | |
| | return json.dumps({ |
| | "status": "success", |
| | "count": len(results), |
| | "results": [ |
| | { |
| | "id": r.id, |
| | "content": r.content, |
| | "score": round(r.score, 3), |
| | "metadata": r.metadata |
| | } |
| | for r in results |
| | ] |
| | }) |
| | |
| | elif name == "mnemo_should_inject": |
| | query = arguments.get("query", "") |
| | context = arguments.get("context", "") |
| | should, reason = should_inject_memory(query, context) |
| | |
| | return json.dumps({ |
| | "should_inject": should, |
| | "reason": reason, |
| | "recommendation": "Inject memory context" if should else "Skip memory - standalone query" |
| | }) |
| | |
| | elif name == "mnemo_get_context": |
| | query = arguments.get("query", "") |
| | top_k = arguments.get("top_k", 3) |
| | context = memory.get_context(query, top_k=top_k) |
| | |
| | return json.dumps({ |
| | "status": "success", |
| | "context": context if context else None, |
| | "message": "Context retrieved" if context else "No relevant context found" |
| | }) |
| | |
| | elif name == "mnemo_list": |
| | memories = memory.list_all() |
| | return json.dumps({ |
| | "status": "success", |
| | "count": len(memories), |
| | "memories": [ |
| | { |
| | "id": m.id, |
| | "content": m.content[:200] + "..." if len(m.content) > 200 else m.content, |
| | "created_at": m.created_at, |
| | "metadata": m.metadata |
| | } |
| | for m in memories |
| | ] |
| | }) |
| | |
| | elif name == "mnemo_delete": |
| | memory_id = arguments.get("memory_id", "") |
| | success = memory.delete(memory_id) |
| | |
| | return json.dumps({ |
| | "status": "success" if success else "error", |
| | "message": f"Memory {memory_id} deleted" if success else f"Memory {memory_id} not found" |
| | }) |
| | |
| | elif name == "mnemo_stats": |
| | stats = memory.get_stats() |
| | return json.dumps({ |
| | "status": "success", |
| | "stats": stats |
| | }) |
| | |
| | elif name == "mnemo_clear": |
| | if arguments.get("confirm", False): |
| | memory.clear() |
| | return json.dumps({ |
| | "status": "success", |
| | "message": "All memories cleared" |
| | }) |
| | else: |
| | return json.dumps({ |
| | "status": "error", |
| | "message": "Must set confirm=true to clear all memories" |
| | }) |
| | |
| | else: |
| | return json.dumps({ |
| | "status": "error", |
| | "message": f"Unknown tool: {name}" |
| | }) |
| |
|
| |
|
| | def run_mcp_server(): |
| | """Run as MCP server""" |
| | if not HAS_MCP: |
| | print("Error: MCP not installed. Install with: pip install mcp", file=sys.stderr) |
| | sys.exit(1) |
| | |
| | server = Server("mnemo-memory") |
| | |
| | @server.list_tools() |
| | async def list_tools() -> List[Tool]: |
| | return [ |
| | Tool( |
| | name=tool["name"], |
| | description=tool["description"], |
| | inputSchema=tool["inputSchema"] |
| | ) |
| | for tool in TOOLS |
| | ] |
| | |
| | @server.call_tool() |
| | async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]: |
| | result = handle_tool_call(name, arguments) |
| | return [TextContent(type="text", text=result)] |
| | |
| | |
| | import asyncio |
| | asyncio.run(server.run()) |
| |
|
| |
|
| | def run_cli(): |
| | """Simple CLI for testing""" |
| | print("Mnemo Memory System - CLI Mode") |
| | print("Commands: add, search, inject, context, list, stats, clear, quit") |
| | print("-" * 50) |
| | |
| | while True: |
| | try: |
| | cmd = input("\n> ").strip().lower() |
| | |
| | if cmd == "quit": |
| | break |
| | elif cmd == "add": |
| | content = input("Content: ") |
| | result = handle_tool_call("mnemo_add", {"content": content}) |
| | print(result) |
| | elif cmd == "search": |
| | query = input("Query: ") |
| | result = handle_tool_call("mnemo_search", {"query": query}) |
| | print(result) |
| | elif cmd == "inject": |
| | query = input("Query: ") |
| | result = handle_tool_call("mnemo_should_inject", {"query": query}) |
| | print(result) |
| | elif cmd == "context": |
| | query = input("Query: ") |
| | result = handle_tool_call("mnemo_get_context", {"query": query}) |
| | print(result) |
| | elif cmd == "list": |
| | result = handle_tool_call("mnemo_list", {}) |
| | print(result) |
| | elif cmd == "stats": |
| | result = handle_tool_call("mnemo_stats", {}) |
| | print(result) |
| | elif cmd == "clear": |
| | confirm = input("Are you sure? (yes/no): ") |
| | if confirm.lower() == "yes": |
| | result = handle_tool_call("mnemo_clear", {"confirm": True}) |
| | print(result) |
| | else: |
| | print("Unknown command. Use: add, search, inject, context, list, stats, clear, quit") |
| | |
| | except KeyboardInterrupt: |
| | print("\nBye!") |
| | break |
| | except Exception as e: |
| | print(f"Error: {e}") |
| |
|
| |
|
| | if __name__ == "__main__": |
| | if len(sys.argv) > 1 and sys.argv[1] == "--cli": |
| | run_cli() |
| | else: |
| | run_mcp_server() |
| |
|
