#!/usr/bin/env -S uv run --script
# -*- coding: utf-8 -*-
# /// script
# dependencies = [
# "mcp>=0.3.0",
# "sqlite-vec>=0.1.6",
# "sentence-transformers>=2.2.2"
# ]
# requires-python = ">=3.8"
# ///
"""
Vector Memory MCP Server - Main Entry Point
===========================================
A secure, vector-based memory server using sqlite-vec for semantic search.
Stores and retrieves coding memories, experiences, and knowledge using
384-dimensional embeddings generated by sentence-transformers.
Usage:
python main.py --working-dir /path/to/project
Memory files stored in: {working_dir}/memory/vector_memory.db
"""
import sys
from pathlib import Path
from typing import Dict, Any
# Add src to path for imports
sys.path.insert(0, str(Path(__file__).parent / "src"))
from mcp.server.fastmcp import FastMCP
# Import our modules
from src.models import Config
from src.security import validate_working_dir, SecurityError
from src.memory_store import VectorMemoryStore
def get_working_dir() -> Path:
"""Get working directory from command line arguments"""
if "--working-dir" in sys.argv:
idx = sys.argv.index("--working-dir")
if idx + 1 < len(sys.argv):
return validate_working_dir(sys.argv[idx + 1])
# Default to current directory
return validate_working_dir(".")
def get_memory_limit() -> int:
"""Get memory limit from command line arguments"""
if "--memory-limit" in sys.argv:
idx = sys.argv.index("--memory-limit")
if idx + 1 < len(sys.argv):
try:
limit = int(sys.argv[idx + 1])
if limit < 1000:
print(f"Warning: memory-limit {limit} is too low, using minimum 1000", file=sys.stderr)
return 1000
if limit > 10_000_000:
print(f"Warning: memory-limit {limit} is too high, using maximum 10,000,000", file=sys.stderr)
return 10_000_000
return limit
except ValueError:
print(f"Warning: invalid memory-limit value, using default {Config.MAX_TOTAL_MEMORIES}", file=sys.stderr)
return Config.MAX_TOTAL_MEMORIES
# Default from config
return Config.MAX_TOTAL_MEMORIES
def create_server() -> FastMCP:
"""Create and configure the MCP server"""
# Initialize global memory store
try:
memory_dir = get_working_dir()
memory_limit = get_memory_limit()
db_path = memory_dir / Config.DB_NAME
memory_store = VectorMemoryStore(db_path, memory_limit=memory_limit)
print(f"Memory database initialized: {db_path}", file=sys.stderr)
print(f"Memory limit: {memory_limit:,} entries", file=sys.stderr)
except Exception as e:
print(f"Failed to initialize memory store: {e}", file=sys.stderr)
sys.exit(1)
# Create FastMCP server
mcp = FastMCP(Config.SERVER_NAME)
# ===============================================================================
# MCP TOOLS IMPLEMENTATION
# ===============================================================================
@mcp.tool()
def store_memory(
content: str,
category: str = "other",
tags: list[str] = None
) -> dict[str, Any]:
"""
Store coding memory with vector embedding for semantic search.
Args:
content: Memory content (max 10K chars)
category: code-solution, bug-fix, architecture, learning, tool-usage, debugging, performance, security, other
tags: Tags for organization (max 10)
"""
try:
if tags is None:
tags = []
result = memory_store.store_memory(content, category, tags)
return result
except SecurityError as e:
return {
"success": False,
"error": "Security validation failed",
"message": str(e)
}
except Exception as e:
return {
"success": False,
"error": "Storage failed",
"message": str(e)
}
@mcp.tool()
def search_memories(
query: str,
limit: int = 10,
category: str = None,
offset: int = 0,
tags: list[str] = None
) -> dict[str, Any]:
"""
Search memories using semantic similarity (vector search).
Args:
query: Search query
limit: Max results (1-50, default 10)
category: Optional category filter
offset: Starting position for results (pagination, 0-based index, default 0)
tags: Optional list of tags to filter by (matches memories containing ANY of the specified tags)
"""
try:
search_results, total = memory_store.search_memories(query, limit, category, offset, tags)
if not search_results:
return {
"success": True,
"results": [],
"total": total,
"count": 0,
"message": "No matching memories found. Try different keywords or broader terms."
}
# Convert SearchResult objects to dictionaries
results = [result.to_dict() for result in search_results]
return {
"success": True,
"query": query,
"results": results,
"total": total,
"count": len(results),
"message": f"Show {len(results)} of {total} total memories matching filters"
}
except SecurityError as e:
return {
"success": False,
"error": "Security validation failed",
"message": str(e)
}
except Exception as e:
return {
"success": False,
"error": "Search failed",
"message": str(e)
}
@mcp.tool()
def list_recent_memories(limit: int = 10) -> dict[str, Any]:
"""
List recent memories in chronological order.
Args:
limit: Max results (1-50, default 10)
"""
try:
limit = min(max(1, limit), Config.MAX_MEMORIES_PER_SEARCH)
memories = memory_store.get_recent_memories(limit)
# Convert MemoryEntry objects to dictionaries
memory_dicts = [memory.to_dict() for memory in memories]
return {
"success": True,
"memories": memory_dicts,
"count": len(memory_dicts),
"message": f"Retrieved {len(memory_dicts)} recent memories"
}
except Exception as e:
return {
"success": False,
"error": "Failed to get recent memories",
"message": str(e)
}
@mcp.tool()
def get_memory_stats() -> dict[str, Any]:
"""Get database statistics (total memories, categories, usage, health)."""
try:
stats = memory_store.get_stats()
result = stats.to_dict()
result["success"] = True
return result
except Exception as e:
return {
"success": False,
"error": "Failed to get statistics",
"message": str(e)
}
@mcp.tool()
def clear_old_memories(
days_old: int = 30,
max_to_keep: int = 1000
) -> dict[str, Any]:
"""
Clear old memories to free space (keeps frequently accessed).
Args:
days_old: Min age in days (default 30)
max_to_keep: Max total memories (default 1000)
"""
try:
if days_old < 1:
return {
"success": False,
"error": "Invalid parameter",
"message": "days_old must be at least 1"
}
result = memory_store.clear_old_memories(days_old, max_to_keep)
return result
except SecurityError as e:
return {
"success": False,
"error": "Security validation failed",
"message": str(e)
}
except Exception as e:
return {
"success": False,
"error": "Cleanup failed",
"message": str(e)
}
@mcp.tool()
def get_by_memory_id(memory_id: int) -> dict[str, Any]:
"""
Get specific memory by ID.
Args:
memory_id: Memory ID to retrieve
"""
try:
if not isinstance(memory_id, int) or memory_id < 1:
return {
"success": False,
"error": "Invalid parameter",
"message": "memory_id must be a positive integer"
}
memory = memory_store.get_memory_by_id(memory_id)
if memory is None:
return {
"success": False,
"error": "Not found",
"message": f"Memory with ID {memory_id} not found"
}
return {
"success": True,
"memory": memory.to_dict(),
"message": "Memory retrieved successfully"
}
except Exception as e:
return {
"success": False,
"error": "Retrieval failed",
"message": str(e)
}
@mcp.tool()
def delete_by_memory_id(memory_id: int) -> dict[str, Any]:
"""
Delete memory by ID (permanent, cannot be undone).
Args:
memory_id: Memory ID to delete
"""
try:
if not isinstance(memory_id, int) or memory_id < 1:
return {
"success": False,
"error": "Invalid parameter",
"message": "memory_id must be a positive integer"
}
deleted = memory_store.delete_memory(memory_id)
if not deleted:
return {
"success": False,
"error": "Not found",
"message": f"Memory with ID {memory_id} not found"
}
return {
"success": True,
"memory_id": memory_id,
"message": "Memory deleted successfully from both metadata and vector tables"
}
except Exception as e:
return {
"success": False,
"error": "Deletion failed",
"message": str(e)
}
@mcp.tool()
def get_unique_tags() -> dict[str, Any]:
"""Get all unique tags from memory database."""
try:
tags = memory_store.get_unique_tags()
return {
"success": True,
"tags": tags,
"count": len(tags),
"message": f"Retrieved {len(tags)} unique tags"
}
except Exception as e:
return {
"success": False,
"error": "Failed to retrieve tags",
"message": str(e)
}
return mcp
def main():
"""Main entry point"""
print(f"Starting {Config.SERVER_NAME} v{Config.SERVER_VERSION}", file=sys.stderr)
try:
# Get working directory and config
memory_dir = get_working_dir()
memory_limit = get_memory_limit()
db_path = memory_dir / Config.DB_NAME
print(f"Working directory: {memory_dir.parent}", file=sys.stderr)
print(f"Memory database: {db_path}", file=sys.stderr)
print(f"Memory limit: {memory_limit:,} entries", file=sys.stderr)
print(f"Embedding model: {Config.EMBEDDING_MODEL}", file=sys.stderr)
print("=" * 50, file=sys.stderr)
# Create and run server
server = create_server()
print("Server ready for connections...", file=sys.stderr)
server.run()
except KeyboardInterrupt:
print("\nServer stopped by user", file=sys.stderr)
except Exception as e:
print(f"Server failed to start: {e}", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()
