Update app.py

app.py

@@ -1,5 +1,4 @@
-from typing import Dict, List, Union, Optional, Tuple
-from urllib.parse import urlparse
+from typing import List, Tuple
 import requests
 import time
 import json
@@ -98,21 +97,17 @@ def parse_huggingface_url(url: str) -> str:
     
     return json.dumps(result, indent=2)
 
-def parse_huggingface_url_with_summary(url: str, progress=gr.Progress()) -> tuple:
+def parse_huggingface_url_with_summary(url: str) -> tuple:
     """Parse URL and return both markdown summary and JSON."""
-    progress(0, desc="🔍 Starting URL parsing...")
     
     if not url.strip():
         return "# ❌ No URL Provided\n\nPlease enter a URL to parse.", "{}"
     
-    progress(0.3, desc="🔍 Analyzing URL format...")
     json_result = parse_huggingface_url(url)
     
-    progress(0.7, desc="🔍 Generating summary...")
     parsed_info = json.loads(json_result)
     md_summary = format_url_summary(parsed_info)
     
-    progress(1.0, desc="✅ URL parsing complete!")
     return md_summary, json_result
 
 def format_url_summary(parsed_info: dict) -> str:
@@ -145,7 +140,7 @@ def format_url_summary(parsed_info: dict) -> str:
     
     return md
 
-def check_single_server_health(url: str, progress=gr.Progress()) -> tuple:
+def check_single_server_health(url: str) -> tuple:
     """
     Check health of a single MCP server from any URL format.
     
@@ -155,12 +150,10 @@ def check_single_server_health(url: str, progress=gr.Progress()) -> tuple:
     Returns:
         tuple: (markdown_summary, json_data)
     """
-    progress(0, desc="🏥 Starting health check...")
     
     if not url.strip():
         return "# ❌ No URL Provided\n\nPlease enter a URL to check.", "{}"
     
-    progress(0.1, desc="🏥 Parsing URL...")
     parsed_info = json.loads(parse_huggingface_url(url))
     
     if not parsed_info["is_valid"]:
@@ -182,7 +175,6 @@ def check_single_server_health(url: str, progress=gr.Progress()) -> tuple:
     }
     
     # Test 1: Check space URL health
-    progress(0.3, desc="🌐 Checking space accessibility...")
     if parsed_info["space_url"] != "unknown":
         start_time = time.time()
         try:
@@ -206,7 +198,6 @@ def check_single_server_health(url: str, progress=gr.Progress()) -> tuple:
             }
     
     # Test 2: Check MCP endpoint health
-    progress(0.6, desc="🔧 Checking MCP endpoint...")
     start_time = time.time()
     try:
         response = requests.get(parsed_info["mcp_endpoint"], timeout=8, stream=True)
@@ -228,8 +219,6 @@ def check_single_server_health(url: str, progress=gr.Progress()) -> tuple:
             "error": str(e)
         }
     
-    progress(0.9, desc="📊 Analyzing results...")
-    
     # Determine overall status
     space_ok = results["space_health"] is None or results["space_health"]["accessible"]
     mcp_ok = results["mcp_health"]["accessible"]
@@ -246,7 +235,6 @@ def check_single_server_health(url: str, progress=gr.Progress()) -> tuple:
     # Generate markdown summary
     md = format_health_summary(results)
     
-    progress(1.0, desc="✅ Health check complete!")
     return md, json.dumps(results, indent=2)
 
 def format_health_summary(results: dict) -> str:
@@ -333,7 +321,7 @@ def extract_functions_from_source(source_code: str) -> List[Tuple[str, str, List
     
     return functions
 
-def discover_server_tools(url: str, progress=gr.Progress()) -> tuple:
+def discover_server_tools(url: str) -> tuple:
     """
     Discover available MCP tools from a server.
     
@@ -343,12 +331,10 @@ def discover_server_tools(url: str, progress=gr.Progress()) -> tuple:
     Returns:
         tuple: (markdown_summary, json_data)
     """
-    progress(0, desc="🛠️ Starting tools discovery...")
     
     if not url.strip():
         return "# ❌ No URL Provided\n\nPlease enter a URL to discover tools.", "{}"
     
-    progress(0.1, desc="🛠️ Parsing URL...")
     parsed_info = json.loads(parse_huggingface_url(url))
     
     if not parsed_info["is_valid"]:
@@ -364,15 +350,12 @@ def discover_server_tools(url: str, progress=gr.Progress()) -> tuple:
     discovery_methods = []
     
     # Method: Analyze app.py source code
-    progress(0.3, desc="🛠️ Fetching source code...")
     try:
         # Try to get app.py from HF spaces
         if parsed_info["hf_spaces_url"] != "unknown":
             app_url = f"{parsed_info['hf_spaces_url']}/raw/main/app.py"
-            progress(0.5, desc="🛠️ Analyzing app.py...")
             response = requests.get(app_url, timeout=10)
             if response.status_code == 200:
-                progress(0.7, desc="🛠️ Extracting functions...")
                 functions = extract_functions_from_source(response.text)
                 for func_name, docstring, params in functions:
                     tools.append({
@@ -385,8 +368,6 @@ def discover_server_tools(url: str, progress=gr.Progress()) -> tuple:
     except Exception as e:
         discovery_methods.append(f"Failed to analyze app.py: {str(e)}")
     
-    progress(0.9, desc="🛠️ Preparing results...")
-    
     # Prepare result
     result = {
         "original_url": url,
@@ -404,7 +385,6 @@ def discover_server_tools(url: str, progress=gr.Progress()) -> tuple:
     # Generate markdown summary
     md = format_tools_summary(result)
     
-    progress(1.0, desc="✅ Tools discovery complete!")
     return md, json.dumps(result, indent=2)
 
 def format_tools_summary(result: dict) -> str:
@@ -434,7 +414,7 @@ def format_tools_summary(result: dict) -> str:
     
     return md
 
-def monitor_multiple_servers(urls_text: str, progress=gr.Progress()) -> tuple:
+def monitor_multiple_servers(urls_text: str) -> tuple:
     """
     Monitor health and tools of multiple MCP servers simultaneously.
     
@@ -444,7 +424,6 @@ def monitor_multiple_servers(urls_text: str, progress=gr.Progress()) -> tuple:
     Returns:
         tuple: (markdown_summary, json_data)
     """
-    progress(0, desc="📊 Starting multi-server monitoring...")
     
     if not urls_text.strip():
         result = {
@@ -455,27 +434,22 @@ def monitor_multiple_servers(urls_text: str, progress=gr.Progress()) -> tuple:
         md = "# ❌ No URLs Provided\n\nPlease enter URLs to monitor."
         return md, json.dumps(result, indent=2)
     
-    progress(0.1, desc="📊 Parsing URL list...")
     urls = [url.strip() for url in urls_text.strip().split('\n') if url.strip()]
     
     if not urls:
         result = {
             "error": "No valid URLs found",
             "servers": [],
-            "total_servers": 0
-        }
+            "total_servers": 0        }
         md = "# ❌ No Valid URLs\n\nPlease check the URL format."
         return md, json.dumps(result, indent=2)
     
     results = []
     
     for i, url in enumerate(urls, 1):
-        progress_pct = 0.1 + (i / len(urls)) * 0.8  # Reserve 10% for setup, 10% for final processing
-        progress(progress_pct, desc=f"📊 Checking server {i}/{len(urls)}: {url[:50]}...")
+        print(f"🔍 Checking server {i}/{len(urls)}: {url}")
         
         try:
-            print(f"🔍 Checking server {i}/{len(urls)}: {url}")
-            
             _, health_json = check_single_server_health(url)
             health_data = json.loads(health_json)
             
@@ -499,8 +473,6 @@ def monitor_multiple_servers(urls_text: str, progress=gr.Progress()) -> tuple:
                 "combined_status": "error"
             })
     
-    progress(0.95, desc="📊 Generating report...")
-    
     final_result = {
         "servers": results,
         "total_servers": len(urls),
@@ -510,7 +482,6 @@ def monitor_multiple_servers(urls_text: str, progress=gr.Progress()) -> tuple:
     # Generate markdown summary
     md = format_multiple_servers_summary(final_result)
     
-    progress(1.0, desc="✅ Multi-server monitoring complete!")
     return md, json.dumps(final_result, indent=2)
 
 def format_multiple_servers_summary(result: dict) -> str:
@@ -546,7 +517,7 @@ def format_multiple_servers_summary(result: dict) -> str:
     
     return md
 
-def validate_mcp_endpoint(url: str, progress=gr.Progress()) -> tuple:
+def validate_mcp_endpoint(url: str) -> tuple:
     """
     Validate that a URL is a working MCP endpoint by checking its schema.
     
@@ -556,12 +527,10 @@ def validate_mcp_endpoint(url: str, progress=gr.Progress()) -> tuple:
     Returns:
         tuple: (markdown_summary, json_data)
     """
-    progress(0, desc="✅ Starting MCP validation...")
     
     if not url.strip():
         return "# ❌ No URL Provided\n\nPlease enter a URL to validate.", "{}"
     
-    progress(0.1, desc="✅ Parsing URL...")
     parsed_info = json.loads(parse_huggingface_url(url))
     
     validation_result = {
@@ -581,8 +550,6 @@ def validate_mcp_endpoint(url: str, progress=gr.Progress()) -> tuple:
 
     mcp_endpoint = validation_result["mcp_endpoint_url"]
     
-    progress(0.3, desc="✅ Building schema URL...")
-    
     # Construct schema URL from MCP SSE endpoint
     # Example: https://user-space.hf.space/gradio_api/mcp/sse -> https://user-space.hf.space/gradio_api/mcp/schema
     if mcp_endpoint.endswith("/sse"):
@@ -596,8 +563,6 @@ def validate_mcp_endpoint(url: str, progress=gr.Progress()) -> tuple:
 
     print(f"ℹ️ Validating MCP: Original URL='{url}', Endpoint='{mcp_endpoint}', Schema='{mcp_schema_url}'")
 
-    progress(0.5, desc="✅ Testing schema endpoint...")
-    
     # Test MCP schema endpoint
     try:
         headers = {'User-Agent': 'MCP-Validator/1.0'}
@@ -605,8 +570,6 @@ def validate_mcp_endpoint(url: str, progress=gr.Progress()) -> tuple:
         
         validation_result["schema_http_status"] = response.status_code
 
-        progress(0.7, desc="✅ Processing schema response...")
-
         if response.status_code == 200:
             try:
                 schema_data = response.json()
@@ -656,12 +619,9 @@ def validate_mcp_endpoint(url: str, progress=gr.Progress()) -> tuple:
         validation_result["error"] = f"An unexpected error occurred during validation: {str(e_gen)}"
         print(f"❌ Unexpected error during MCP validation for {mcp_schema_url}: {str(e_gen)}")
 
-    progress(0.9, desc="✅ Generating validation report...")
-    
     # Generate markdown summary
     md = format_validation_summary(validation_result)
     
-    progress(1.0, desc="✅ MCP validation complete!")
     return md, json.dumps(validation_result, indent=2)
 
 def format_validation_summary(result: dict) -> str:
@@ -711,40 +671,18 @@ def format_validation_summary(result: dict) -> str:
         
     return md
 
-def scrape_hf_spaces_with_progress(max_pages: int, sort_by: str, progress=gr.Progress()) -> tuple:
-    """Wrapper function for scraping with progress feedback."""
-    progress(0, desc="🕷️ Starting HuggingFace scraping...")
+def scrape_hf_spaces_with_progress(max_pages: int, sort_by: str) -> tuple:
+    """Wrapper function for scraping."""
     
     # Validate sort option
     if sort_by not in SORT_OPTIONS:
         sort_by = "relevance"
     
-    progress(0.1, desc=f"🕷️ Using sort method: {SORT_OPTIONS[sort_by]['label']}")
-    
-    # Update the scraper to accept a progress callback
-    def progress_callback(message):
-        # Extract progress percentage from message if possible
-        if "page" in message.lower():
-            try:
-                # Try to extract current page from message
-                import re
-                match = re.search(r'page (\d+)(/(\d+))?', message.lower())
-                if match:
-                    current = int(match.group(1))
-                    total = int(match.group(3)) if match.group(3) else max_pages
-                    progress_pct = min(0.9, (current / total) * 0.8 + 0.1)  # Reserve 10% for final processing
-                    progress(progress_pct, desc=f"🕷️ {message}")
-                else:
-                    progress(None, desc=f"🕷️ {message}")
-            except:
-                progress(None, desc=f"🕷️ {message}")
-        else:
-            progress(None, desc=f"🕷️ {message}")
-    
-    # Call the scraper with progress callback and sort option
+    # Call the scraper with sort option
+    # The imported scrape_and_monitor_hf_mcp_servers function
+    # will print its own progress to the console.
     md, json_data = scrape_and_monitor_hf_mcp_servers(max_pages, sort_by)
     
-    progress(1.0, desc="✅ HuggingFace scraping complete!")
     return md, json_data
 
 # Default URLs for testing