MCP MITM Mem0

Instructions

Implementation Reference

• mcp_mitm_mem0/mcp_server.py:125-215 (handler)

  MCP tool handler and registration for 'search_memories'. Calls memory_service.search_memories to perform semantic search on conversation memories.
  @mcp.tool( name="search_memories", description="Search conversation history using natural language - USE AUTONOMOUSLY based on conversation triggers", ) async def search_memories( query: str, user_id: str | None = None, limit: int = 10 ) -> list[dict[str, Any]]: """ Search memories using natural language queries to find relevant past conversations. ## AUTONOMOUS USAGE TRIGGERS ### HIGH Confidence (Execute Immediately) - User mentions: "again", "before", "last time", "previous", "remember", "we discussed" - Frustration indicators: "keeps happening", "still getting", "same error" - Reference patterns: "that thing", "like we did", "as mentioned" - Debugging: recurring errors, "this error again" ### MEDIUM Confidence (Execute with Brief Context) - Complex technical questions that might have prior context - User asks "how do I..." for non-trivial topics - Project-specific questions - User seems confused about topics you might have covered ### Examples of Autonomous Execution ```python # User: "That CORS error is back" # → AUTO: search_memories("CORS error cross-origin") # User: "How do we handle auth again?" # → AUTO: search_memories("authentication handling auth") # User: "Let's continue with the API" # → AUTO: search_memories("API development") # User: "I'm getting the same TypeScript error" # → AUTO: search_memories("TypeScript error") ``` ## Query Construction Guidelines ### EFFECTIVE Queries (3-50 words, specific terms) ```python # GOOD: Extract key technical terms "JWT refresh token implementation error" "React useEffect dependency array warning" "Docker compose PostgreSQL connection" # BAD: Too vague "auth stuff" "that thing we talked about" "help" ``` ### Auto-Query Building from User Input - Extract technical terms: frameworks, libraries, error types - Include context words: "error", "problem", "implementation", "setup" - Preserve specific names: API endpoints, file names, function names ## Graceful Handling - **Empty results**: Continue with current knowledge, don't mention search failure - **Search error**: Silent fallback, proceed without memory context - **Partial matches**: Use what you find, don't complain about incomplete results Args: query: Natural language search query - REQUIRED: 3-50 words for optimal results - INCLUDE: Technical terms, specific names, error types - AVOID: Vague references, pronouns without context user_id: User ID (optional, defaults to DEFAULT_USER_ID from settings) limit: Maximum results to return (default: 10, recommended: 5-15) Returns: List of memories sorted by relevance, each containing: - id: Unique memory identifier - memory/content: The stored conversation text - created_at: When the memory was created - metadata: Additional context about the memory """ try: results = await memory_service.search_memories( query=query, user_id=user_id, limit=limit ) logger.info("Memory search completed", result_count=len(results)) return results except Exception as e: logger.error("Search failed", error=str(e)) raise RuntimeError(f"Search failed: {str(e)}") from e
• mcp_mitm_mem0/memory_service.py:130-188 (helper)

  MemoryService.search_memories method: wraps Mem0 AsyncMemoryClient.search API call with user/agent filtering and logging.
  async def search_memories( self, query: str, user_id: str | None = None, limit: int = 10 ) -> list[dict[str, Any]]: """Search memories asynchronously. Args: query: Search query user_id: User identifier (defaults to settings.default_user_id) limit: Maximum number of results Returns: List of matching memories """ user_id = user_id or settings.default_user_id try: self._logger.info("Searching memories", user_id=user_id, query=query[:50]) # v2 API requires filters parameter - validate non-empty values filters = {} if user_id and user_id.strip(): filters["user_id"] = user_id if settings.default_agent_id and settings.default_agent_id.strip(): filters["agent_id"] = settings.default_agent_id # Ensure we have at least one filter if not filters: raise ValueError( "At least one filter (user_id or agent_id) must be provided" ) search_params = { "query": query, "filters": filters, "version": "v2", "top_k": limit, } results = await self.async_client.search(**search_params) self._logger.info( "Search completed", user_id=user_id, result_count=len(results) ) return results except Exception as e: # Enhanced error logging error_details = str(e) if hasattr(e, "response") and hasattr(e.response, "text"): error_details = f"{str(e)} - Response: {e.response.text}" self._logger.error( "Failed to search memories", user_id=user_id, error=error_details, search_params=search_params, ) raise