SelfMemory

"""SelfMemory MCP Server Implements an MCP (Model Context Protocol) server that provides memory operations for SelfMemory using simple Bearer token authentication. Features: - Simple Bearer token authentication with SelfMemory API keys - Per-request client creation for proper user isolation - Graceful error handling when core server is unavailable - Tools: add_memory and search_memories - Streamable HTTP transport for production deployment """ import logging import os import sys from pathlib import Path from typing import Any from dotenv import load_dotenv from mcp.server.fastmcp import Context, FastMCP from selfmemory import SelfMemoryClient # Ensure project root is in sys.path (two levels up from this file) PROJECT_ROOT = Path(__file__).resolve().parent.parent if str(PROJECT_ROOT) not in sys.path: sys.path.append(str(PROJECT_ROOT)) # Add selfmemory-mcp to path for telemetry imports sys.path.insert(0, str(Path(__file__).parent)) load_dotenv() # Load environment variables from .env # Import telemetry after adding to path from telemetry import init_logging, init_telemetry # noqa: E402 # Initialize logging based on environment (console for dev, file for prod) init_logging() logger = logging.getLogger(__name__) # Initialize OpenTelemetry if enabled (optional) init_telemetry() # Configuration CORE_SERVER_HOST = os.getenv("SELFMEMORY_API_HOST", "http://localhost:8081") MCP_SERVER_PORT = int(os.getenv("MCP_SERVER_PORT", "5055")) MCP_SERVER_HOST = os.getenv("MCP_SERVER_HOST", "0.0.0.0") # Initialize MCP server without OAuth (simple Bearer token approach) mcp = FastMCP( name="SelfMemory", instructions="Memory management server for SelfMemory - store and search personal memories with metadata", stateless_http=True, json_response=True, port=MCP_SERVER_PORT, host=MCP_SERVER_HOST, ) logger.info(f"SelfMemory MCP Server initialized - Core server: {CORE_SERVER_HOST}") def _extract_memory_contents(search_result: dict[str, Any]) -> list[str]: """Extract only content strings from search results for LLM consumption. Args: search_result: Full search result dictionary from client.search() Returns: List of memory content strings, empty list if no results """ if "results" not in search_result or not search_result["results"]: return [] return [memory.get("content", "") for memory in search_result["results"]] def _generate_memory_confirmation(content: str) -> str: """Generate a personalized confirmation message for stored memory. Args: content: The memory content that was stored Returns: Personalized confirmation message string """ return "I learnt more about you with this!" def validate_and_get_client(ctx: Context) -> SelfMemoryClient: """ Validate request and create authenticated SelfMemoryClient. Supports both dashboard session auth and direct API key auth. Args: ctx: FastMCP Context containing request information Returns: SelfMemoryClient: Client authenticated with the user's token Raises: ValueError: If authentication fails """ try: # Extract headers from the HTTP request request = ctx.request_context.request auth_header = request.headers.get("Authorization") if not auth_header or not auth_header.startswith("Bearer "): raise ValueError("No valid authorization header found") token = auth_header.replace("Bearer ", "") # Create and validate client - this will raise ValueError if token is invalid client = SelfMemoryClient(api_key=token, host=CORE_SERVER_HOST) logger.info( f"✅ MCP: API key authenticated for user: {client.user_info.get('user_id', 'unknown')}" ) return client except AttributeError as e: logger.error(f"Context structure error: {e}") raise ValueError("Request context not available") from e except ValueError: # Re-raise ValueError as-is (these are our custom auth errors) raise except Exception as e: logger.error(f"Authentication error: {e}") raise ValueError("Authentication failed") from e @mcp.tool() async def add_memory( content: str, ctx: Context, tags: str = "", people: str = "", category: str = "" ) -> str: """ Store new memories with metadata. Args: content: The memory content to store tags: Optional comma-separated tags (e.g., "work,meeting,important") people: Optional comma-separated people mentioned (e.g., "Alice,Bob") category: Optional topic category (e.g., "work", "personal", "learning") Returns: Personalized confirmation message string Examples: - add_memory("Had a great meeting about the new project", tags="work,meeting", people="Sarah,Mike") -> "I learnt more about you with this!" - add_memory("Learned about Python decorators today", category="learning") -> "I learnt more about you with this!" - add_memory("Birthday party this weekend", tags="personal,social", people="Emma") -> "I learnt more about you with this!" """ try: logger.info(f"Adding memory: {content[:50]}...") # Validate token and get authenticated client client = validate_and_get_client(ctx) # Format data in the correct selfmemory format that the core server expects memory_data = { "messages": [{"role": "user", "content": content}], "metadata": { "tags": tags, "people_mentioned": people, "topic_category": category, }, } # Use the client's underlying httpx client to send the correct format response = client.client.post("/api/memories", json=memory_data) response.raise_for_status() # Close the client connection client.close() logger.info("Memory added successfully") # Generate personalized confirmation message return _generate_memory_confirmation(content) except ValueError as e: error_msg = f"Authentication error: {str(e)}" logger.error(error_msg) return f"Authentication failed: {str(e)}" except Exception as e: error_msg = f"Failed to add memory: {str(e)}" logger.error(error_msg) return f"Failed to store memory: {str(e)}" @mcp.tool() async def search_memories( query: str, ctx: Context, limit: int = 10, tags: list[str] | None = None, people: list[str] | None = None, category: str | None = None, threshold: float | None = None, ) -> list[str]: """ Search memories using semantic search with optional filters. Args: query: The search query (e.g., "meeting notes", "python learning", "weekend plans") limit: Maximum number of results to return (default: 10, max: 50) tags: Optional list of tags to filter by (e.g., ["work", "important"]) people: Optional list of people to filter by (e.g., ["Alice", "Bob"]) category: Optional category filter (e.g., "work", "personal") threshold: Optional minimum similarity score (0.0 to 1.0) Returns: List of memory content strings for LLM consumption Examples: - search_memories("project meeting") -> ["Had a meeting about the new project", ...] - search_memories("Python", tags=["learning"], limit=5) -> ["Learned Python decorators", ...] - search_memories("birthday", people=["Emma"], category="personal") -> ["Emma's birthday party", ...] """ try: logger.info(f"Searching memories: '{query}'") # Validate limit if limit > 50: limit = 50 elif limit < 1: limit = 1 # Validate token and get authenticated client client = validate_and_get_client(ctx) # Use SelfMemoryClient properly (no circular dependency) result = client.search( query=query, limit=limit, tags=tags, people_mentioned=people, topic_category=category, threshold=threshold, ) # Close the client connection client.close() results_count = len(result.get("results", [])) logger.info(f"Search completed: {results_count} results found") # Extract only content strings for LLM consumption return _extract_memory_contents(result) except ValueError as e: error_msg = f"Authentication error: {str(e)}" logger.error(error_msg) return [] except Exception as e: error_msg = f"Search failed: {str(e)}" logger.error(error_msg) return [] def main(): """Main entry point for the SelfMemory MCP server.""" logger.info("=" * 60) logger.info("🚀 Starting SelfMemory MCP Server") logger.info("=" * 60) logger.info(f"📡 Core Server: {CORE_SERVER_HOST}") logger.info(f"🌐 MCP Server: http://{MCP_SERVER_HOST}:{MCP_SERVER_PORT}") logger.info("🔒 Authentication: Bearer Token") logger.info("🛠️ Tools: add_memory, search_memories") logger.info("=" * 60) try: # Run server with streamable HTTP transport mcp.run(transport="streamable-http") except KeyboardInterrupt: logger.info("Server stopped by user") except Exception as e: logger.error(f"Server error: {e}") raise if __name__ == "__main__": main()