import json import os import subprocess import sys import gradio as gr import modal import requests # --- Backend Management --- def check_modal_backend_running(): """Check if the Modal backend is already running.""" try: result = subprocess.run( ["modal", "app", "list", "--json"], capture_output=True, text=True, timeout=30, ) if result.returncode == 0: apps = json.loads(result.stdout) for app in apps: if app.get("name") == "mlops-backend" and app.get("state") == "ready": return True return False except Exception as e: print(f"Error checking Modal backend: {e}") return False def deploy_modal_backend(): """Deploy the Modal backend if not running.""" if check_modal_backend_running(): print("✅ Modal backend is already running") return True print("🚀 Deploying Modal backend...") try: result = subprocess.run( ["modal", "deploy", "modal_backend.py"], capture_output=True, text=True, timeout=300, ) if result.returncode == 0: print("✅ Modal backend deployed successfully") return True print(f"❌ Failed to deploy Modal backend: {result.stderr}") return False except Exception as e: print(f"❌ Error deploying Modal backend: {e}") return False # Initialize Modal function references f_analyze = modal.Function.from_name("mlops-backend", "analyze_data") f_train = modal.Function.from_name("mlops-backend", "train_model") f_check = modal.Function.from_name("mlops-backend", "check_model") # --- Core Logic --- def get_file_content(file_input) -> str: """ Retrieves content from a file path, URL, or Gradio file object. Handles all input types seamlessly for both web and MCP interfaces. Args: file_input: Can be: - String path/URL to file - Gradio file object with .name attribute - File object with path Returns: File content as string """ if not file_input: return "Error: No file provided." # Handle Gradio file objects if hasattr(file_input, 'name'): file_path = file_input.name print(f"DEBUG: Processing Gradio file: {file_path}") elif hasattr(file_input, 'path'): file_path = file_input.path print(f"DEBUG: Processing file with path: {file_path}") else: file_path = str(file_input) print(f"DEBUG: Processing file path/URL: {file_path}") # Handle URLs if file_path.startswith("http://") or file_path.startswith("https://"): # Try to extract local path from Gradio file URL to avoid self-download loops if "/file=" in file_path: local_path = file_path.split("/file=")[1] # URL decode the path import urllib.parse local_path = urllib.parse.unquote(local_path) print(f"DEBUG: Extracted local_path: {local_path}") if os.path.exists(local_path): print("DEBUG: Local file exists. Reading directly.") with open(local_path, "r") as f: return f.read() else: print(f"DEBUG: Local file does not exist at {local_path}") # Fallback: Download from URL print(f"Downloading file from: {file_path}") try: response = requests.get(file_path, timeout=30) response.raise_for_status() return response.text except Exception as e: return f"Error downloading file: {str(e)}" # Handle local file paths if os.path.exists(file_path): with open(file_path, "r") as f: return f.read() return f"Error: File not found at {file_path}" # --- Tool Definitions (MCP & Logic) --- def analyze_data_tool(file_path: str) -> str: """ 📊 Analyze CSV Dataset Analyzes your CSV dataset and provides comprehensive statistical metadata including: - Dataset shape (rows, columns) - Column data types - Missing values analysis - Statistical summaries for numerical columns Args: file_path: URL or local path to your CSV file Returns: JSON formatted analysis report with dataset statistics """ content = get_file_content(file_path) if content.startswith("Error"): return content try: result = f_analyze.remote(content) return json.dumps(result, indent=2) except Exception as e: return f"Error in analysis: {str(e)}" def train_model_tool( file_path: str, target_column: str, task_type: str = "classification" ) -> str: """ 🚈 Train Machine Learning Model Trains a production-ready machine learning model on your CSV dataset. Args: file_path: URL or local path to your CSV file target_column: Name of the column you want to predict task_type: Type of machine learning task: - 'classification': Predict categories (default) - 'regression': Predict numerical values - 'time_series': Forecast time-based data Returns: JSON response containing: - Training status message - Unique model ID for deployment - Performance metrics (accuracy, F1 score, etc.) """ content = get_file_content(file_path) if content.startswith("Error"): return content try: result = f_train.remote(content, target_column, task_type) metrics_str = ", ".join([f"{k}: {v:.4f}" for k, v in result["metrics"].items()]) return json.dumps( { "message": result["message"], "model_id": result["model_id"], "metrics": metrics_str, }, indent=2, ) except Exception as e: return f"Error in training: {str(e)}" def deploy_model_tool(model_id: str) -> str: """ 🚀 Deploy Model to Production API Deploys your trained model to a live production API endpoint. Args: model_id: Unique identifier of the trained model (returned by training tool) Returns: Deployment information including: - Live API endpoint URL - Python code examples for making predictions - cURL command examples for API testing """ try: check = f_check.remote(model_id) if not check["exists"]: return f"Error: Model {model_id} not found." except Exception as e: return f"Error checking model: {str(e)}" api_url = "https://abidali899--mlops-backend-predict-api.modal.run" usage_code = f""" import requests url = "{api_url}" payload = {{ "model_id": "{model_id}", "data": {{ "col1": "val1" }} }} response = requests.post(url, json=payload) print(response.json())""" curl_code = f""" curl -X POST {api_url} \\ -H "Content-Type: application/json" \\ -d '{{ "model_id": "{model_id}", "data": {{ "col1": "val1" }} }}'""" return f"Model {model_id} is live!\n\nEndpoint: {api_url}\n\n### Usage (Python):\n```python\n{usage_code}\n```\n\n### Usage (cURL):\n```bash\n{curl_code}\n```" def auto_deploy_tool( file_path: str, target_column: str, task_type: str = "classification" ) -> str: """ ⚡ Auto Deploy - Complete ML Pipeline One-click solution to analyze your data, train a model, and deploy to production. This end-to-end pipeline automatically: 1. Performs comprehensive dataset analysis and insights 2. Trains an optimized machine learning model 3. Deploys the model to a live production API 4. Provides complete deployment report with usage examples Args: file_path: URL or local path to your CSV file target_column: Name of the column you want to predict task_type: Type of machine learning task: - 'classification': Predict categories (default) - 'regression': Predict numerical values - 'time_series': Forecast time-based data Returns: Comprehensive deployment report including: - Detailed dataset analysis and insights - Data quality assessment and recommendations - Model performance metrics and evaluation - Live API endpoint and usage examples """ content = get_file_content(file_path) if content.startswith("Error"): return content try: # 1. Comprehensive Analysis analysis = f_analyze.remote(content) # Extract basic analysis information shape = analysis.get('shape', 'Unknown') columns = analysis.get('columns', []) # Build simple analysis section analysis_section = f"""### 📊 Dataset Analysis - **Dataset Shape**: {shape} - **Total Columns**: {len(columns)} - **Target Column**: `{target_column}` - **Task Type**: {task_type.title()} """ # 2. Model Training train_result = f_train.remote(content, target_column, task_type) model_id = train_result["model_id"] # Build performance metrics section training_section = "### ⭐ Performance Metrics\n\n" metrics = train_result.get("metrics", {}) for metric_name, metric_value in metrics.items(): if isinstance(metric_value, (int, float)): if metric_name.lower() in ['accuracy', 'precision', 'recall', 'f1']: training_section += f"- **{metric_name.title()}**: {metric_value:.4f} ({metric_value*100:.2f}%)\n" else: training_section += f"- **{metric_name.title()}**: {metric_value:.4f}\n" # 3. Deployment Information deploy_info = deploy_model_tool(model_id) # 4. Final comprehensive report return f"""## 📇 Auto-Deployment Complete Report {analysis_section} {training_section} ### 🌐 Deployment Information {deploy_info} --- *📈 Auto-generated by Auto-Deployer MCP Server*""" except Exception as e: return f"❌ Error in auto-deployment: {str(e)}" # --- UI Construction --- with gr.Blocks() as demo: gr.Markdown( """ # 🚈 Auto-Deployer MCP Server >From CSV to Deployed ML API in 30 seconds. Connect your AI agent to this server to: * 👀 **Analyze** datasets * 🚈 **Train** models on serverless CPUs (Modal) * 🚀 **Deploy** production-ready APIs """ ) # Manual Interface (Web) # Core tools work with both file uploads and URLs seamlessly with gr.Tab("Analyze"): an_file = gr.File(label="CSV File", file_types=[".csv"]) an_btn = gr.Button("Analyze Data") an_out = gr.JSON(label="Output") an_btn.click( fn=lambda: gr.Button(value="Processing...", interactive=False), outputs=an_btn ).then( fn=analyze_data_tool, inputs=[an_file], outputs=an_out ).then( fn=lambda: gr.Button(value="Analyze Data", interactive=True), outputs=an_btn ) gr.Examples( examples=[[os.path.join("upload_files", "heart.csv")]], inputs=[an_file], label="Example Dataset" ) with gr.Tab("Train"): t_file = gr.File(label="CSV File", file_types=[".csv"]) t_col = gr.Textbox(label="Target Column") t_type = gr.Dropdown( ["classification", "regression", "time_series"], label="Task Type", value="classification", ) t_btn = gr.Button("Train") t_out = gr.JSON(label="Output") t_btn.click( fn=lambda: gr.Button(value="Processing...", interactive=False), outputs=t_btn ).then( fn=train_model_tool, inputs=[t_file, t_col, t_type], outputs=t_out ).then( fn=lambda: gr.Button(value="Train", interactive=True), outputs=t_btn ) gr.Examples( examples=[[os.path.join("upload_files", "heart.csv"), "HeartDisease", "classification"]], inputs=[t_file, t_col, t_type], label="Example Training Config" ) with gr.Tab("Deploy"): d_id = gr.Textbox(label="Model ID") d_btn = gr.Button("Deploy") d_out = gr.Markdown(label="Output") d_btn.click( fn=lambda: gr.Button(value="Processing...", interactive=False), outputs=d_btn ).then( fn=deploy_model_tool, inputs=[d_id], outputs=d_out ).then( fn=lambda: gr.Button(value="Deploy", interactive=True), outputs=d_btn ) gr.Examples( examples=[["model_1764524701"]], inputs=[d_id], label="Example Model ID" ) with gr.Tab("Auto Deploy"): a_file = gr.File(label="CSV File", file_types=[".csv"]) a_col = gr.Textbox(label="Target Column") a_type = gr.Dropdown( ["classification", "regression", "time_series"], label="Task Type", value="classification", ) a_btn = gr.Button("Auto Deploy") a_out = gr.Markdown(label="Output") a_btn.click( fn=lambda: gr.Button(value="Processing...", interactive=False), outputs=a_btn ).then( fn=auto_deploy_tool, inputs=[a_file, a_col, a_type], outputs=a_out ).then( fn=lambda: gr.Button(value="Auto Deploy", interactive=True), outputs=a_btn ) gr.Examples( examples=[[os.path.join("upload_files", "housing.csv"), "price", "regression"]], inputs=[a_file, a_col, a_type], label="Example Auto-Deploy Config" ) # MCP tools are automatically generated from the visible interface above # The core tools handle both file uploads and URL inputs seamlessly if __name__ == "__main__": print("🔍 Checking Modal backend status...") if deploy_modal_backend(): print("🎉 Backend is ready. Starting MCP server...") demo.launch(mcp_server=True) else: print("❌ Failed to deploy backend.") sys.exit(1)