Tune Formats & Configs

app.py

@@ -1,7 +1,8 @@
 import gradio as gr
 from typing import List, Dict, Any, Tuple
 
-from toolset_semantics import search_suitable_tools, search_suitable_spaces
+from toolset_semantics import search_suitable_tools, search_suitable_spaces, initialize_and_upload_to_vector_db
+from hfspaces_tracking import update_database   
 
 def suitable_MCP_servers(task: str) -> List[Dict[str, Any]]:
     """
@@ -58,4 +59,6 @@ mcps_interface = gr.TabbedInterface(
 
 # Launch the interface and MCP server
 if __name__ == "__main__":
+    update_database()
+    initialize_and_upload_to_vector_db()
     mcps_interface.launch(mcp_server=True)

hfspaces_tracking.py

@@ -217,11 +217,14 @@ def save_to_database(spaces):
     finally:
         conn.close()
 
-if __name__ == "__main__":
+def update_database():
     create_database()
     print("Starting fetching process...")
 
     spaces_data = fetch_spaces()
     print(spaces_data)
     save_to_database(spaces_data)
-    print("Process complete! Data saved to database")
+    print("Process complete! Data saved to database")
+
+if __name__ == "__main__":
+    update_database()

sql/select_spaces.sql

@@ -2,7 +2,7 @@ SELECT
     s.space_id,
     s.title,
     s.description,
-    s.url,
+    s.url as space_url,
     s.tags,
     e.endpoint_url as schema_url,
     COUNT(t.tool_name) as tool_count

toolset_semantics.py

@@ -10,8 +10,8 @@ load_dotenv()
 from sql.sql_utils import load_sql_query
 DB_PATH = '/data/huggingface_spaces.db' if os.path.exists('/data') else 'huggingface_spaces.db'
 PINECONE_API_KEY = os.getenv("PINECONE_API_KEY")
-TOOLS_INDEX_NAME = "mcp-tools"
-SPACES_INDEX_NAME = "mcp-spaces"
+TOOLS_INDEX_NAME = "vix-mcp-tools"
+SPACES_INDEX_NAME = "vix-mcp-spaces"
 SQL_SELECT_TOOLS = "sql/select_tools.sql"
 SQL_SELECT_SPACES = "sql/select_spaces.sql"
 
@@ -102,10 +102,10 @@ def upsert_spaces_to_pinecone(pc: Pinecone, spaces: List[Dict[str, Any]]):
         profile = prepare_space_profile(space)
         record = {
             "_id": space['space_id'],
-            "profile": profile,
-            "title": space['title'],
-            "url": space['url'],
-            "tool_count": space['tool_count'],
+            "profile": profile if profile else "",
+            "title": space['title'] if space['title'] else "",
+            "url": space['schema_url'] if space['schema_url'] else "",
+            "tool_count": space['tool_count'] if space['tool_count'] else 0,
             "tags": space['tags'] if space['tags'] else '[]'
         }
         records.append(record)
@@ -116,7 +116,7 @@ def upsert_spaces_to_pinecone(pc: Pinecone, spaces: List[Dict[str, Any]]):
         time.sleep(1)
     print(f"Uploaded {len(spaces)} spaces")
 
-def search_spaces(pc: Pinecone, query: str, top_k: int = 5) -> List[Dict[str, Any]]:
+def search_spaces(pc: Pinecone, query: str, top_k: int = 5, score_threshold: float = 0.0) -> List[Dict[str, Any]]:
     """Search for relevant MCP spaces based on their description and tools"""
     index = pc.Index(SPACES_INDEX_NAME)
     
@@ -132,16 +132,20 @@ def search_spaces(pc: Pinecone, query: str, top_k: int = 5) -> List[Dict[str, An
     
     spaces_list = []
     for hit in results['result']['hits']:
-        fields = hit.get('fields', {})
-        space = {
-            "title": fields.get("title"),
-            "url": fields.get("url"),
-            "tool_count": fields.get("tool_count"),
-            "tags": fields.get("tags"),
-            "score": hit.get("score", 0)
-        }
-        spaces_list.append(space)
+        score = hit.get("_score", 0)
+        if score > score_threshold:
+            fields = hit.get('fields', {})
+            space = {
+                "title": fields.get("title"),
+                "url": fields.get("url"),
+                "tool_count": fields.get("tool_count"),
+                "tags": fields.get("tags"),
+                "score": score
+            }
+            spaces_list.append(space)
     
+    # Sort by score in descending order
+    spaces_list.sort(key=lambda x: x["score"], reverse=True)
     return spaces_list
 
 def load_tools_from_db() -> List[Dict[str, Any]]:
@@ -176,7 +180,7 @@ def upsert_tools_to_pinecone(pc: Pinecone, tools: List[Dict[str, Any]]):
         time.sleep(1)
     print(f"Uploaded {len(tools)} tools")
 
-def search_tools(pc: Pinecone, query: str, top_k: int = 5) -> List[Dict[str, Any]]:
+def search_tools(pc: Pinecone, query: str, top_k: int = 5, score_threshold: float = 0.0) -> List[Dict[str, Any]]:
     """Search for relevant tools based on description"""
     index = pc.Index(TOOLS_INDEX_NAME)
     
@@ -190,34 +194,37 @@ def search_tools(pc: Pinecone, query: str, top_k: int = 5) -> List[Dict[str, Any
         }
     )
     
-    # Convert to schema endpoint format
     tools_list = []
     for hit in results['result']['hits']:
-        fields = hit.get('fields', {})
-        tool = {
-            "name": fields.get("tool_name"),
-            "description": fields.get("description"),            
-            "inputSchema": json.loads(fields.get("input_schema", "{}")),
-            "server_url": fields.get("server_url")
-        }
-        tools_list.append(tool)
+        score = hit.get("_score", 0)
+        if score > score_threshold:
+            fields = hit.get('fields', {})
+            tool = {
+                "name": fields.get("tool_name"),
+                "description": fields.get("description"),
+                "inputSchema": json.loads(fields.get("input_schema", "{}")),
+                "server_url": fields.get("server_url"),
+                "score": score
+            }
+            tools_list.append(tool)
     
+    # Sort by score in descending order
+    tools_list.sort(key=lambda x: x["score"], reverse=True)
     return tools_list
 
-
 def search_suitable_tools(query: str)->List[Dict[str, Any]]:
     """Search for suitable tools based on query"""
     pc = Pinecone(api_key=PINECONE_API_KEY)
-    tools = search_tools(pc, query)
+    tools = search_tools(pc, query, top_k=13, score_threshold=0.25)
     return tools
 
 def search_suitable_spaces(query: str)->List[Dict[str, Any]]:
     """Search for suitable spaces based on query"""
     pc = Pinecone(api_key=PINECONE_API_KEY)
-    spaces = search_spaces(pc, query, top_k=3)
+    spaces = search_spaces(pc, query, top_k=3, score_threshold=0.1)
     return spaces
 
-def initialize_and_upload():
+def initialize_and_upload_to_vector_db():
     """Initialize Pinecone and upload all tools and spaces"""
     pc = Pinecone(api_key=PINECONE_API_KEY)
     create_tools_index(pc)
@@ -237,7 +244,7 @@ def initialize_and_upload():
     return pc
 
 if __name__ == "__main__":
-    pc = initialize_and_upload()
+    pc = initialize_and_upload_to_vector_db()
     
     # Interactive search loop