Added files

app.py

@@ -0,0 +1,1344 @@
+"""
+Data Analyst Duo MCP Implementation
+
+This script implements a multi-agent system using the Model Context Protocol (MCP).
+It features two agents that collaborate on data analysis tasks:
+- ComputeAgent: Responsible for data loading, cleaning, and computation
+- InterpretAgent: Responsible for interpreting results and visualizing data
+
+The application includes a Gradio interface for interaction.
+"""
+
+import os
+import sys
+import json
+import time
+import datetime
+import gradio as gr
+import pandas as pd
+import numpy as np
+import matplotlib.pyplot as plt
+import seaborn as sns
+from typing import Dict, List, Any, Optional, Union, Tuple
+import requests
+from io import StringIO
+import logging
+import uuid
+import anthropic
+import openai
+from dotenv import load_dotenv
+
+# Load environment variables
+load_dotenv()
+
+# Configure logging
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+logger = logging.getLogger(__name__)
+
+# ============== MCP Protocol Implementation ==============
+
+class MCPMessage:
+    """Base class for MCP messages that agents exchange"""
+
+    def __init__(self, sender: str, message_type: str, content: Any):
+        self.id = str(uuid.uuid4())
+        self.sender = sender
+        self.message_type = message_type
+        self.content = content
+        self.timestamp = datetime.datetime.now().isoformat()
+
+    def to_dict(self) -> Dict:
+        return {
+            "id": self.id,
+            "sender": self.sender,
+            "message_type": self.message_type,
+            "content": self.content,
+            "timestamp": self.timestamp
+        }
+
+    @staticmethod
+    def from_dict(data: Dict) -> 'MCPMessage':
+        msg = MCPMessage(
+            sender=data["sender"],
+            message_type=data["message_type"],
+            content=data["content"]
+        )
+        # Restore ID and timestamp if present
+        if "id" in data:
+            msg.id = data["id"]
+        if "timestamp" in data:
+            msg.timestamp = data["timestamp"]
+        return msg
+
+
+class MCPTool:
+    """Defines a tool that can be used by agents through the MCP protocol"""
+
+    def __init__(self, name: str, description: str, function):
+        self.name = name
+        self.description = description
+        self.function = function
+
+    def to_dict(self) -> Dict:
+        return {
+            "name": self.name,
+            "description": self.description
+        }
+
+    def execute(self, params: Dict) -> Any:
+        return self.function(params)
+
+
+class MCPAgent:
+    """Base agent class implementing MCP protocol"""
+
+    def __init__(self, name: str, description: str, llm_model: Optional[str] = None, api_key: Optional[str] = None):
+        self.name = name
+        self.description = description
+        self.tools: Dict[str, MCPTool] = {}
+        self.message_queue: List[MCPMessage] = []
+        self.peers: Dict[str, 'MCPAgent'] = {}
+        self.message_history: List[Dict] = []
+        self.llm_model = llm_model
+        self.api_key = api_key
+        self.llm_logs = []
+
+    def register_tool(self, tool: MCPTool):
+        """Register a tool that this agent can use"""
+        self.tools[tool.name] = tool
+
+    def list_tools(self) -> List[Dict]:
+        """List all tools available to this agent"""
+        return [tool.to_dict() for tool in self.tools.values()]
+
+    def call_tool(self, tool_name: str, params: Dict) -> Any:
+        """Call a tool by name with parameters"""
+        if tool_name not in self.tools:
+            raise ValueError(f"Tool {tool_name} not found")
+        return self.tools[tool_name].execute(params)
+
+    def connect(self, peer: 'MCPAgent'):
+        """Connect to another agent as a peer"""
+        self.peers[peer.name] = peer
+
+    def send_message(self, receiver: str, message_type: str, content: Any) -> Dict:
+        """Send a message to a peer agent"""
+        if receiver not in self.peers:
+            raise ValueError(f"Peer {receiver} not found")
+
+        message = MCPMessage(self.name, message_type, content)
+        message_dict = message.to_dict()
+
+        # Save to message history
+        self.message_history.append({
+            "type": "sent",
+            "message": message_dict
+        })
+
+        # Send to receiver
+        self.peers[receiver].receive_message(message)
+        logger.info(f"Agent {self.name} sent {message_type} to {receiver}")
+        return message_dict
+
+    def receive_message(self, message: MCPMessage):
+        """Receive a message from a peer agent"""
+        self.message_queue.append(message)
+
+        # Save to message history
+        self.message_history.append({
+            "type": "received",
+            "message": message.to_dict()
+        })
+
+        logger.info(f"Agent {self.name} received {message.message_type} from {message.sender}")
+
+    def process_messages(self) -> List[Dict]:
+        """Process all messages in the queue"""
+        processed = []
+        while self.message_queue:
+            message = self.message_queue.pop(0)
+            response = self.handle_message(message)
+            processed.append(response)
+        return processed
+
+    def handle_message(self, message: MCPMessage) -> Dict:
+        """Handle a message - to be implemented by subclasses"""
+        raise NotImplementedError("Subclasses must implement handle_message")
+
+    def log_llm_interaction(self, prompt: str, response: str):
+        """Log LLM interactions for transparency"""
+        log_entry = {
+            "timestamp": datetime.datetime.now().isoformat(),
+            "prompt": prompt,
+            "response": response
+        }
+        self.llm_logs.append(log_entry)
+        return log_entry
+
+    def get_message_history(self) -> List[Dict]:
+        """Get the agent's message history"""
+        return self.message_history
+
+    def get_llm_logs(self) -> List[Dict]:
+        """Get the agent's LLM interaction logs"""
+        return self.llm_logs
+
+
+# ============== Compute Agent Implementation ==============
+
+class ComputeAgent(MCPAgent):
+    """Agent responsible for data loading, cleaning, and computation"""
+
+    def __init__(self, name: str = "ComputeAgent", llm_model: Optional[str] = None, api_key: Optional[str] = None):
+        super().__init__(name, "Agent responsible for data loading, cleaning and computation", llm_model, api_key)
+        self.dataframe = None
+        self.current_task = None
+
+        # Register tools
+        self.register_tool(MCPTool(
+            "load_dataset",
+            "Load a dataset from Kaggle or URL",
+            self._load_dataset
+        ))
+
+        self.register_tool(MCPTool(
+            "clean_data",
+            "Clean the loaded dataset by handling missing values, duplicates, etc.",
+            self._clean_data
+        ))
+
+        self.register_tool(MCPTool(
+            "compute_statistics",
+            "Compute basic statistics on the dataset",
+            self._compute_statistics
+        ))
+
+        self.register_tool(MCPTool(
+            "compute_correlation",
+            "Compute correlation between columns",
+            self._compute_correlation
+        ))
+
+        self.register_tool(MCPTool(
+            "filter_data",
+            "Filter data based on conditions",
+            self._filter_data
+        ))
+
+        self.register_tool(MCPTool(
+            "compute_aggregation",
+            "Compute aggregation (sum, mean, etc.) grouped by a column",
+            self._compute_aggregation
+        ))
+
+    def _load_dataset(self, params: Dict) -> Dict:
+        """Load a dataset from Kaggle or URL"""
+        dataset_url = params.get("url")
+
+        try:
+            # Check if it's the default cereals dataset
+            if dataset_url == "default" or dataset_url.lower() == "cereals":
+                dataset_url = "https://raw.githubusercontent.com/datasciencedojo/datasets/master/cereal.csv"
+
+            # Check if it's a Kaggle URL and extract the dataset path
+            elif "kaggle.com/datasets" in dataset_url:
+                # For simplicity, we use direct download links
+                # In real implementation, you would use the Kaggle API
+                prompt = f"""
+                I have a Kaggle dataset URL: {dataset_url}
+                Find the direct download link or alternative source for this dataset if possible.
+                If not, suggest a suitable replacement dataset that's freely available.
+                """
+
+                if self.llm_model and self.llm_model.startswith("claude"):
+                    client = anthropic.Anthropic(api_key=self.api_key)
+                    response = client.messages.create(
+                        model="claude-3-sonnet-20240229",
+                        max_tokens=1000,
+                        messages=[{"role": "user", "content": prompt}]
+                    )
+                    result = response.content[0].text
+                elif self.llm_model and self.llm_model.startswith("gpt"):
+                    client = openai.OpenAI(api_key=self.api_key)
+                    response = client.chat.completions.create(
+                        model="gpt-4o",
+                        messages=[{"role": "user", "content": prompt}]
+                    )
+                    result = response.choices[0].message.content
+                else:
+                    result = "For non-default datasets, please provide a direct download link."
+
+                self.log_llm_interaction(prompt, result)
+
+                # Extract URL from the response
+                lines = result.split('\n')
+                for line in lines:
+                    if line.startswith("http") and (".csv" in line or ".xlsx" in line):
+                        dataset_url = line.strip()
+                        break
+                else:
+                    # If no URL found, use default cereals dataset
+                    dataset_url = "https://raw.githubusercontent.com/datasciencedojo/datasets/master/cereal.csv"
+
+            # Load the dataset
+            response = requests.get(dataset_url)
+            content = response.content.decode('utf-8')
+            self.dataframe = pd.read_csv(StringIO(content))
+
+            # Basic info about the dataset
+            info = {
+                "status": "success",
+                "rows": len(self.dataframe),
+                "columns": list(self.dataframe.columns),
+                "preview": self.dataframe.head(5).to_dict(orient="records"),
+                "dtypes": {col: str(dtype) for col, dtype in self.dataframe.dtypes.items()}
+            }
+
+            return info
+
+        except Exception as e:
+            return {"status": "error", "message": str(e)}
+
+    def _clean_data(self, params: Dict) -> Dict:
+        """Clean the loaded dataset"""
+        if self.dataframe is None:
+            return {"status": "error", "message": "No dataset loaded"}
+
+        try:
+            original_shape = self.dataframe.shape
+
+            # Handle missing values based on strategy
+            missing_strategy = params.get("missing_strategy", "drop")
+            if missing_strategy == "drop":
+                self.dataframe = self.dataframe.dropna()
+            elif missing_strategy == "mean":
+                self.dataframe = self.dataframe.fillna(self.dataframe.mean(numeric_only=True))
+            elif missing_strategy == "median":
+                self.dataframe = self.dataframe.fillna(self.dataframe.median(numeric_only=True))
+            elif missing_strategy == "mode":
+                # Fill categorical with mode, numerics separately
+                for column in self.dataframe.columns:
+                    if pd.api.types.is_numeric_dtype(self.dataframe[column]):
+                        self.dataframe[column] = self.dataframe[column].fillna(self.dataframe[column].mean())
+                    else:
+                        self.dataframe[column] = self.dataframe[column].fillna(self.dataframe[column].mode()[0])
+
+            # Remove duplicates if specified
+            if params.get("remove_duplicates", True):
+                self.dataframe = self.dataframe.drop_duplicates()
+
+            # Convert datatypes if specified
+            if "convert_dtypes" in params:
+                for col, dtype in params["convert_dtypes"].items():
+                    self.dataframe[col] = self.dataframe[col].astype(dtype)
+
+            new_shape = self.dataframe.shape
+
+            return {
+                "status": "success",
+                "original_shape": original_shape,
+                "new_shape": new_shape,
+                "missing_values_remaining": self.dataframe.isna().sum().to_dict(),
+                "duplicate_rows_removed": original_shape[0] - new_shape[0]
+            }
+
+        except Exception as e:
+            return {"status": "error", "message": str(e)}
+
+    def _compute_statistics(self, params: Dict) -> Dict:
+        """Compute basic statistics on the dataset"""
+        if self.dataframe is None:
+            return {"status": "error", "message": "No dataset loaded"}
+
+        try:
+            # Get columns to compute stats for
+            columns = params.get("columns", list(self.dataframe.select_dtypes(include=[np.number]).columns))
+
+            # Compute different statistics based on parameters
+            stats = {}
+
+            # Basic descriptive statistics
+            if params.get("descriptive", True):
+                stats["descriptive"] = self.dataframe[columns].describe().to_dict()
+
+            # Central tendency
+            if params.get("central_tendency", False):
+                stats["mean"] = self.dataframe[columns].mean().to_dict()
+                stats["median"] = self.dataframe[columns].median().to_dict()
+                # Mode is more complex as it can return multiple values
+                mode_results = {}
+                for col in columns:
+                    if pd.api.types.is_numeric_dtype(self.dataframe[col]):
+                        mode_vals = self.dataframe[col].mode().tolist()
+                        mode_results[col] = mode_vals
+                stats["mode"] = mode_results
+
+            # Dispersion
+            if params.get("dispersion", False):
+                stats["variance"] = self.dataframe[columns].var().to_dict()
+                stats["std_dev"] = self.dataframe[columns].std().to_dict()
+                stats["range"] = {col: self.dataframe[col].max() - self.dataframe[col].min() for col in columns}
+                stats["iqr"] = {col: self.dataframe[col].quantile(0.75) - self.dataframe[col].quantile(0.25) for col in columns}
+
+            # Shape
+            if params.get("shape", False):
+                stats["skewness"] = self.dataframe[columns].skew().to_dict()
+                stats["kurtosis"] = self.dataframe[columns].kurtosis().to_dict()
+
+            return {
+                "status": "success",
+                "statistics": stats
+            }
+
+        except Exception as e:
+            return {"status": "error", "message": str(e)}
+
+    def _compute_correlation(self, params: Dict) -> Dict:
+        """Compute correlation between columns"""
+        if self.dataframe is None:
+            return {"status": "error", "message": "No dataset loaded"}
+
+        try:
+            # Get columns to compute correlation for
+            columns = params.get("columns", list(self.dataframe.select_dtypes(include=[np.number]).columns))
+            method = params.get("method", "pearson")  # pearson, kendall, spearman
+
+            corr_matrix = self.dataframe[columns].corr(method=method).to_dict()
+
+            # Find highest correlated pairs
+            corr_df = self.dataframe[columns].corr(method=method).unstack()
+            corr_df = corr_df[corr_df < 1.0]  # Remove self-correlation
+            highest_corr = corr_df.sort_values(ascending=False)[:10].to_dict()
+
+            return {
+                "status": "success",
+                "correlation_matrix": corr_matrix,
+                "highest_correlations": highest_corr
+            }
+
+        except Exception as e:
+            return {"status": "error", "message": str(e)}
+
+    def _filter_data(self, params: Dict) -> Dict:
+        """Filter data based on conditions"""
+        if self.dataframe is None:
+            return {"status": "error", "message": "No dataset loaded"}
+
+        try:
+            # Apply filters
+            filtered_df = self.dataframe.copy()
+            filters = params.get("filters", [])
+
+            for filter_item in filters:
+                column = filter_item["column"]
+                operator = filter_item["operator"]
+                value = filter_item["value"]
+
+                if operator == "==":
+                    filtered_df = filtered_df[filtered_df[column] == value]
+                elif operator == "!=":
+                    filtered_df = filtered_df[filtered_df[column] != value]
+                elif operator == ">":
+                    filtered_df = filtered_df[filtered_df[column] > value]
+                elif operator == "<":
+                    filtered_df = filtered_df[filtered_df[column] < value]
+                elif operator == ">=":
+                    filtered_df = filtered_df[filtered_df[column] >= value]
+                elif operator == "<=":
+                    filtered_df = filtered_df[filtered_df[column] <= value]
+                elif operator == "in":
+                    filtered_df = filtered_df[filtered_df[column].isin(value)]
+                elif operator == "not in":
+                    filtered_df = filtered_df[~filtered_df[column].isin(value)]
+
+            # Store the filtered dataframe temporarily for use in subsequent operations
+            self.filtered_df = filtered_df
+
+            return {
+                "status": "success",
+                "original_rows": len(self.dataframe),
+                "filtered_rows": len(filtered_df),
+                "preview": filtered_df.head(5).to_dict(orient="records")
+            }
+
+        except Exception as e:
+            return {"status": "error", "message": str(e)}
+
+    def _compute_aggregation(self, params: Dict) -> Dict:
+        """Compute aggregation grouped by a column"""
+        if self.dataframe is None:
+            return {"status": "error", "message": "No dataset loaded"}
+
+        try:
+            # Get params
+            groupby_cols = params.get("groupby", [])
+            agg_cols = params.get("columns", [])
+            agg_funcs = params.get("functions", ["mean"])
+
+            # Use filtered dataframe if available, otherwise use original
+            df_to_use = getattr(self, "filtered_df", self.dataframe)
+
+            # Prepare aggregation dict
+            agg_dict = {col: agg_funcs for col in agg_cols}
+
+            # Compute aggregation
+            result = df_to_use.groupby(groupby_cols).agg(agg_dict).reset_index()
+
+            return {
+                "status": "success",
+                "result": result.to_dict(orient="records")
+            }
+
+        except Exception as e:
+            return {"status": "error", "message": str(e)}
+
+    def handle_message(self, message: MCPMessage) -> Dict:
+        """Handle incoming messages from other agents"""
+        if message.message_type == "request_data_load":
+            result = self._load_dataset(message.content)
+            return self.send_message(message.sender, "data_load_result", result)
+
+        elif message.message_type == "request_data_cleaning":
+            result = self._clean_data(message.content)
+            return self.send_message(message.sender, "data_cleaning_result", result)
+
+        elif message.message_type == "request_statistics":
+            result = self._compute_statistics(message.content)
+            return self.send_message(message.sender, "statistics_result", result)
+
+        elif message.message_type == "request_correlation":
+            result = self._compute_correlation(message.content)
+            return self.send_message(message.sender, "correlation_result", result)
+
+        elif message.message_type == "request_filter":
+            result = self._filter_data(message.content)
+            return self.send_message(message.sender, "filter_result", result)
+
+        elif message.message_type == "request_aggregation":
+            result = self._compute_aggregation(message.content)
+            return self.send_message(message.sender, "aggregation_result", result)
+
+        else:
+            return {"status": "error", "message": f"Unknown message type: {message.message_type}"}
+
+
+# ============== Interpret Agent Implementation ==============
+
+class InterpretAgent(MCPAgent):
+    """Agent responsible for interpreting results and visualizing data"""
+
+    def __init__(self, name: str = "InterpretAgent", llm_model: Optional[str] = None, api_key: Optional[str] = None):
+        super().__init__(name, "Agent responsible for interpreting results and visualizing data", llm_model, api_key)
+        self.dataset_info = None
+        self.statistics = None
+        self.correlation_data = None
+        self.filter_results = None
+        self.aggregation_results = None
+        self.visualization_results = {}
+
+        # Register tools
+        self.register_tool(MCPTool(
+            "interpret_statistics",
+            "Interpret statistical results and provide insights",
+            self._interpret_statistics
+        ))
+
+        self.register_tool(MCPTool(
+            "interpret_correlation",
+            "Interpret correlation results and provide insights",
+            self._interpret_correlation
+        ))
+
+        self.register_tool(MCPTool(
+            "create_visualization",
+            "Create a visualization based on data",
+            self._create_visualization
+        ))
+
+        self.register_tool(MCPTool(
+            "generate_report",
+            "Generate a report with key findings",
+            self._generate_report
+        ))
+
+    def _interpret_statistics(self, params: Dict) -> Dict:
+        """Interpret statistical results and provide insights"""
+        if not self.statistics:
+            return {"status": "error", "message": "No statistics data available"}
+
+        try:
+            # If we have LLM access, use it for more advanced interpretation
+            if self.llm_model:
+                prompt = f"""
+                As a data analyst, interpret these statistics and provide insights:
+                {json.dumps(self.statistics, indent=2)}
+
+                Provide:
+                1. 5 key insights about the data
+                2. Any potential anomalies or interesting observations
+                3. Any patterns or trends visible in the descriptive statistics
+                """
+
+                if self.llm_model.startswith("claude"):
+                    client = anthropic.Anthropic(api_key=self.api_key)
+                    response = client.messages.create(
+                        model="claude-3-sonnet-20240229",
+                        max_tokens=1000,
+                        messages=[{"role": "user", "content": prompt}]
+                    )
+                    result = response.content[0].text
+                elif self.llm_model.startswith("gpt"):
+                    client = openai.OpenAI(api_key=self.api_key)
+                    response = client.chat.completions.create(
+                        model="gpt-4o",
+                        messages=[{"role": "user", "content": prompt}]
+                    )
+                    result = response.choices[0].message.content
+
+                self.log_llm_interaction(prompt, result)
+
+                return {
+                    "status": "success",
+                    "insights": result.split('\n'),
+                    "summary": "Statistical analysis complete with LLM-generated insights."
+                }
+
+            # Fallback to rule-based insights if no LLM available
+            insights = []
+            stats = self.statistics.get("statistics", {})
+
+            # Analyze descriptive statistics
+            if "descriptive" in stats:
+                desc_stats = stats["descriptive"]
+
+                # Look at each numerical column
+                for col in desc_stats:
+                    col_stats = desc_stats[col]
+
+                    # Check for outliers using IQR method
+                    q1 = col_stats.get("25%", 0)
+                    q3 = col_stats.get("75%", 0)
+                    iqr = q3 - q1
+                    lower_bound = q1 - 1.5 * iqr
+                    upper_bound = q3 + 1.5 * iqr
+
+                    if col_stats.get("min", 0) < lower_bound or col_stats.get("max", 0) > upper_bound:
+                        insights.append(f"Column '{col}' may contain outliers.")
+
+                    # Check for skewness
+                    mean = col_stats.get("mean", 0)
+                    median = col_stats.get("50%", 0)
+                    if abs(mean - median) > 0.1 * mean:
+                        skew_direction = "right" if mean > median else "left"
+                        insights.append(f"Column '{col}' appears to be skewed to the {skew_direction}.")
+
+                    # Check for variability
+                    std = col_stats.get("std", 0)
+                    mean = col_stats.get("mean", 0)
+                    cv = std / mean if mean != 0 else 0
+                    if cv > 1:
+                        insights.append(f"Column '{col}' shows high variability (CV > 1).")
+
+            return {
+                "status": "success",
+                "insights": insights,
+                "summary": "Statistical analysis reveals potential patterns and anomalies in the data."
+            }
+
+        except Exception as e:
+            return {"status": "error", "message": str(e)}
+
+    def _interpret_correlation(self, params: Dict) -> Dict:
+        """Interpret correlation results and provide insights"""
+        if not self.correlation_data:
+            return {"status": "error", "message": "No correlation data available"}
+
+        try:
+            # If we have LLM access, use it for more advanced interpretation
+            if self.llm_model:
+                prompt = f"""
+                As a data analyst, interpret this correlation data and provide insights:
+                {json.dumps(self.correlation_data, indent=2)}
+
+                Provide:
+                1. The 5 most significant correlations found and what they might indicate
+                2. Any interesting patterns of correlation in the dataset
+                3. Suggestions for variables that might have causal relationships
+                """
+
+                if self.llm_model.startswith("claude"):
+                    client = anthropic.Anthropic(api_key=self.api_key)
+                    response = client.messages.create(
+                        model="claude-3-sonnet-20240229",
+                        max_tokens=1000,
+                        messages=[{"role": "user", "content": prompt}]
+                    )
+                    result = response.content[0].text
+                elif self.llm_model.startswith("gpt"):
+                    client = openai.OpenAI(api_key=self.api_key)
+                    response = client.chat.completions.create(
+                        model="gpt-4o",
+                        messages=[{"role": "user", "content": prompt}]
+                    )
+                    result = response.choices[0].message.content
+
+                self.log_llm_interaction(prompt, result)
+
+                return {
+                    "status": "success",
+                    "insights": result.split('\n'),
+                    "summary": "Correlation analysis complete with LLM-generated insights."
+                }
+
+            # Fallback to rule-based insights if no LLM available
+            insights = []
+            corr_matrix = self.correlation_data.get("correlation_matrix", {})
+            highest_corr = self.correlation_data.get("highest_correlations", {})
+
+            # Find strong positive correlations
+            strong_pos_corr = [(k, v) for k, v in highest_corr.items() if v > 0.7]
+            if strong_pos_corr:
+                for (col1, col2), value in strong_pos_corr[:3]:
+                    insights.append(f"Strong positive correlation ({value:.2f}) between '{col1}' and '{col2}'.")
+
+            # Find strong negative correlations
+            strong_neg_corr = [(k, v) for k, v in highest_corr.items() if v < -0.7]
+            if strong_neg_corr:
+                for (col1, col2), value in strong_neg_corr[:3]:
+                    insights.append(f"Strong negative correlation ({value:.2f}) between '{col1}' and '{col2}'.")
+
+            # Identify potential multicollinearity
+            multi_corr = [(k, v) for k, v in highest_corr.items() if abs(v) > 0.9]
+            if multi_corr:
+                insights.append("Potential multicollinearity detected between some features.")
+
+            return {
+                "status": "success",
+                "insights": insights,
+                "summary": "Correlation analysis reveals interesting relationships between variables."
+            }
+
+        except Exception as e:
+            return {"status": "error", "message": str(e)}
+
+    def _create_visualization(self, params: Dict) -> Dict:
+        """Create a visualization based on data"""
+        try:
+            viz_type = params.get("type", "histogram")
+            title = params.get("title", "Data Visualization")
+            x_column = params.get("x", None)
+            y_column = params.get("y", None)
+
+            # Generate a unique ID for this visualization
+            viz_id = str(uuid.uuid4())
+
+            # Create the visualization and save it to a file
+            plt.figure(figsize=(10, 6))
+
+            if not hasattr(self, "compute_agent") or not hasattr(self.compute_agent, "dataframe"):
+                return {"status": "error", "message": "No data available for visualization"}
+
+            df = self.compute_agent.dataframe
+
+            if viz_type == "histogram":
+                if x_column:
+                    sns.histplot(df[x_column], kde=True)
+                    plt.xlabel(x_column)
+                    plt.ylabel("Frequency")
+                else:
+                    return {"status": "error", "message": "Column name required for histogram"}
+
+            elif viz_type == "scatter":
+                if x_column and y_column:
+                    sns.scatterplot(x=df[x_column], y=df[y_column])
+                    plt.xlabel(x_column)
+                    plt.ylabel(y_column)
+                else:
+                    return {"status": "error", "message": "X and Y column names required for scatter plot"}
+
+            elif viz_type == "bar":
+                if x_column and y_column:
+                    sns.barplot(x=df[x_column], y=df[y_column])
+                    plt.xlabel(x_column)
+                    plt.ylabel(y_column)
+                else:
+                    return {"status": "error", "message": "X and Y column names required for bar chart"}
+
+            elif viz_type == "boxplot":
+                if x_column:
+                    sns.boxplot(y=df[x_column])
+                    plt.ylabel(x_column)
+                elif x_column and y_column:
+                    sns.boxplot(x=df[x_column], y=df[y_column])
+                    plt.xlabel(x_column)
+                    plt.ylabel(y_column)
+                else:
+                    return {"status": "error", "message": "At least one column name required for boxplot"}
+
+            elif viz_type == "heatmap":
+                if params.get("columns"):
+                    corr = df[params["columns"]].corr()
+                    sns.heatmap(corr, annot=True, cmap="coolwarm")
+                else:
+                    corr = df.select_dtypes(include=[np.number]).corr()
+                    sns.heatmap(corr, annot=True, cmap="coolwarm")
+
+            plt.title(title)
+            plt.tight_layout()
+
+            # Save the visualization
+            viz_filename = f"viz_{viz_id}.png"
+            plt.savefig(viz_filename)
+            plt.close()
+
+            # Store visualization details
+            viz_details = {
+                "id": viz_id,
+                "type": viz_type,
+                "title": title,
+                "filename": viz_filename,
+                "x_column": x_column,
+                "y_column": y_column
+            }
+
+            self.visualization_results[viz_id] = viz_details
+
+            return {
+                "status": "success",
+                "visualization": viz_details
+            }
+
+        except Exception as e:
+            return {"status": "error", "message": str(e)}
+
+    def _generate_report(self, params: Dict) -> Dict:
+        """Generate a report with key findings"""
+        try:
+            # If LLM available, use it for advanced report generation
+            if self.llm_model:
+                # Gather all the data we have
+                report_data = {
+                    "dataset_info": self.dataset_info,
+                    "statistics": self.statistics,
+                    "correlation_data": self.correlation_data,
+                    "filter_results": self.filter_results,
+                    "aggregation_results": self.aggregation_results
+                }
+
+                prompt = f"""
+                As a data analyst, generate a comprehensive report based on the following analysis data:
+                {json.dumps(report_data, indent=2)}
+
+                The report should include:
+                1. Dataset Overview
+                2. Key Findings from Statistical Analysis
+                3. Correlation Analysis Highlights
+                4. Filtered Data Analysis (if applicable)
+                5. Aggregation Insights (if applicable)
+                6. Conclusions and Recommendations
+
+                Format the report in a professional style with clear sections.
+                """
+
+                if self.llm_model.startswith("claude"):
+                    client = anthropic.Anthropic(api_key=self.api_key)
+                    response = client.messages.create(
+                        model="claude-3-sonnet-20240229",
+                        max_tokens=2000,
+                        messages=[{"role": "user", "content": prompt}]
+                    )
+                    result = response.content[0].text
+                elif self.llm_model.startswith("gpt"):
+                    client = openai.OpenAI(api_key=self.api_key)
+                    response = client.chat.completions.create(
+                        model="gpt-4o",
+                        messages=[{"role": "user", "content": prompt}]
+                    )
+                    result = response.choices[0].message.content
+
+                self.log_llm_interaction(prompt, result)
+
+                return {
+                    "status": "success",
+                    "report": {
+                        "title": params.get("report_title", "Data Analysis Report"),
+                        "content": result
+                    }
+                }
+
+            # Fallback to template-based report if no LLM available
+            # Gather all the insights and results
+            report_sections = []
+
+            # Dataset overview
+            if self.dataset_info:
+                report_sections.append({
+                    "title": "Dataset Overview",
+                    "content": f"The dataset contains {self.dataset_info.get('rows', 0)} rows and {len(self.dataset_info.get('columns', []))} columns."
+                })
+
+            # Statistical insights
+            if self.statistics:
+                # Interpret statistics if not already done
+                if not hasattr(self, 'stat_insights'):
+                    self.stat_insights = self._interpret_statistics({}).get('insights', [])
+
+                report_sections.append({
+                    "title": "Statistical Analysis",
+                    "content": "Key findings from statistical analysis:",
+                    "insights": self.stat_insights
+                })
+
+            # Correlation insights
+            if self.correlation_data:
+                # Interpret correlations if not already done
+                if not hasattr(self, 'corr_insights'):
+                    self.corr_insights = self._interpret_correlation({}).get('insights', [])
+
+                report_sections.append({
+                    "title": "Correlation Analysis",
+                    "content": "Key findings from correlation analysis:",
+                    "insights": self.corr_insights
+                })
+
+            # Filter results
+            if self.filter_results:
+                report_sections.append({
+                    "title": "Filtered Data Analysis",
+                    "content": f"The filtered dataset contains {self.filter_results.get('filtered_rows', 0)} rows, down from {self.filter_results.get('original_rows', 0)} rows."
+                })
+
+            # Aggregation results
+            if self.aggregation_results:
+                report_sections.append({
+                    "title": "Aggregation Analysis",
+                    "content": "Key insights from aggregated data:",
+                    "data": self.aggregation_results.get('result', [])
+                })
+
+            # Conclusions
+            report_sections.append({
+                "title": "Conclusions",
+                "content": "Based on the analysis, several patterns and relationships have been identified in the data."
+            })
+
+            return {
+                "status": "success",
+                "report": {
+                    "title": params.get("report_title", "Data Analysis Report"),
+                    "sections": report_sections
+                }
+            }
+
+        except Exception as e:
+            return {"status": "error", "message": str(e)}
+
+    def handle_message(self, message: MCPMessage) -> Dict:
+        """Handle incoming messages from other agents"""
+        if message.message_type == "data_load_result":
+            self.dataset_info = message.content
+            return self.send_message(message.sender, "acknowledge", {"status": "received", "message": "Dataset info received"})
+
+        elif message.message_type == "data_cleaning_result":
+            return self.send_message(message.sender, "acknowledge", {"status": "received", "message": "Data cleaning result received"})
+
+        elif message.message_type == "statistics_result":
+            self.statistics = message.content
+            insights = self._interpret_statistics({})
+            return self.send_message(message.sender, "statistics_interpretation", insights)
+
+        elif message.message_type == "correlation_result":
+            self.correlation_data = message.content
+            insights = self._interpret_correlation({})
+            return self.send_message(message.sender, "correlation_interpretation", insights)
+
+        elif message.message_type == "filter_result":
+            self.filter_results = message.content
+            return self.send_message(message.sender, "acknowledge", {"status": "received", "message": "Filter result received"})
+
+        elif message.message_type == "aggregation_result":
+            self.aggregation_results = message.content
+            return self.send_message(message.sender, "acknowledge", {"status": "received", "message": "Aggregation result received"})
+
+        elif message.message_type == "request_report":
+            report = self._generate_report(message.content)
+            return self.send_message(message.sender, "report_result", report)
+
+        elif message.message_type == "request_visualization":
+            visualization = self._create_visualization(message.content)
+            return self.send_message(message.sender, "visualization_result", visualization)
+
+        else:
+            return {"status": "error", "message": f"Unknown message type: {message.message_type}"}
+
+    def set_compute_agent(self, compute_agent):
+        """Set reference to compute agent for access to dataframe"""
+        self.compute_agent = compute_agent
+
+
+# ============== Main Analysis Workflow ==============
+
+class DataAnalystDuo:
+    """Main class for the Data Analyst Duo MCP implementation"""
+
+    def __init__(self, llm_model: Optional[str] = None, api_key: Optional[str] = None):
+        self.compute_agent = ComputeAgent(llm_model=llm_model, api_key=api_key)
+        self.interpret_agent = InterpretAgent(llm_model=llm_model, api_key=api_key)
+
+        # Connect the agents as peers
+        self.compute_agent.connect(self.interpret_agent)
+        self.interpret_agent.connect(self.compute_agent)
+
+        # Set reference to compute agent inside interpret agent
+        self.interpret_agent.set_compute_agent(self.compute_agent)
+
+        # Logs to store message flow and intermediate results
+        self.logs = []
+
+    def log_step(self, step_name: str, details: Dict):
+        """Log a step in the analysis workflow"""
+        log_entry = {
+            "timestamp": datetime.datetime.now().isoformat(),
+            "step": step_name,
+            "details": details
+        }
+        self.logs.append(log_entry)
+        return log_entry
+
+    def run_analysis(self, dataset_url: str, analysis_params: Dict = None) -> Dict:
+        """Run the complete analysis workflow"""
+        if analysis_params is None:
+            analysis_params = {}
+
+        results = {}
+
+        # 1. Load dataset
+        self.log_step("Initiating dataset loading", {"url": dataset_url})
+        self.interpret_agent.send_message("ComputeAgent", "request_data_load", {"url": dataset_url})
+        self.compute_agent.process_messages()
+        self.interpret_agent.process_messages()
+
+        # 2. Clean data
+        clean_params = analysis_params.get("clean_params", {"missing_strategy": "mean", "remove_duplicates": True})
+        self.log_step("Initiating data cleaning", clean_params)
+        self.interpret_agent.send_message("ComputeAgent", "request_data_cleaning", clean_params)
+        self.compute_agent.process_messages()
+        self.interpret_agent.process_messages()
+
+        # 3. Compute statistics
+        stats_params = analysis_params.get("stats_params", {"descriptive": True, "central_tendency": True, "dispersion": True})
+        self.log_step("Initiating statistical analysis", stats_params)
+        self.interpret_agent.send_message("ComputeAgent", "request_statistics", stats_params)
+        self.compute_agent.process_messages()
+        self.interpret_agent.process_messages()
+
+        # 4. Compute correlation
+        corr_params = analysis_params.get("corr_params", {"method": "pearson"})
+        self.log_step("Initiating correlation analysis", corr_params)
+        self.interpret_agent.send_message("ComputeAgent", "request_correlation", corr_params)
+        self.compute_agent.process_messages()
+        self.interpret_agent.process_messages()
+
+        # 5. Filter data if requested
+        if "filter_params" in analysis_params:
+            self.log_step("Initiating data filtering", analysis_params["filter_params"])
+            self.interpret_agent.send_message("ComputeAgent", "request_filter", analysis_params["filter_params"])
+            self.compute_agent.process_messages()
+            self.interpret_agent.process_messages()
+
+        # 6. Compute aggregation if requested
+        if "agg_params" in analysis_params:
+            self.log_step("Initiating data aggregation", analysis_params["agg_params"])
+            self.interpret_agent.send_message("ComputeAgent", "request_aggregation", analysis_params["agg_params"])
+            self.compute_agent.process_messages()
+            self.interpret_agent.process_messages()
+
+        # 7. Create visualizations if requested
+        if "viz_params" in analysis_params:
+            for viz_param in analysis_params["viz_params"]:
+                self.log_step("Initiating visualization creation", viz_param)
+                self.compute_agent.send_message("InterpretAgent", "request_visualization", viz_param)
+                self.interpret_agent.process_messages()
+                self.compute_agent.process_messages()
+
+        # 8. Generate final report
+        report_params = analysis_params.get("report_params", {"report_title": "Data Analysis Report"})
+        self.log_step("Generating final report", report_params)
+        self.compute_agent.send_message("InterpretAgent", "request_report", report_params)
+        self.interpret_agent.process_messages()
+        self.compute_agent.process_messages()
+
+        # Collect results
+        results["dataset_info"] = self.interpret_agent.dataset_info
+        results["statistics"] = self.interpret_agent.statistics
+        results["correlation_data"] = self.interpret_agent.correlation_data
+        results["filter_results"] = self.interpret_agent.filter_results
+        results["aggregation_results"] = self.interpret_agent.aggregation_results
+        results["visualizations"] = self.interpret_agent.visualization_results
+        results["compute_agent_messages"] = self.compute_agent.get_message_history()
+        results["interpret_agent_messages"] = self.interpret_agent.get_message_history()
+        results["compute_agent_llm_logs"] = self.compute_agent.get_llm_logs()
+        results["interpret_agent_llm_logs"] = self.interpret_agent.get_llm_logs()
+        results["workflow_logs"] = self.logs
+
+        return results
+
+
+# ============== Gradio Interface ==============
+
+def format_json(json_data):
+    """Format JSON data for display"""
+    if isinstance(json_data, dict) or isinstance(json_data, list):
+        return json.dumps(json_data, indent=2)
+    return str(json_data)
+
+def run_data_analysis(dataset_url, llm_provider, api_key, missing_strategy, create_visualizations, high_fiber_filter):
+    """Run the data analysis workflow and return results"""
+    try:
+        # Validate inputs
+        if not dataset_url:
+            dataset_url = "default"  # Use default cereals dataset
+
+        if llm_provider != "none" and not api_key:
+            return {
+                'mcp_messages': "Error: API key is required for LLM integration",
+                'llm_logs': "",
+                'visualizations': "",
+                'final_report': ""
+            }
+
+        # Initialize the analyst duo
+        llm_model = None
+        if llm_provider == "claude":
+            llm_model = "claude"
+        elif llm_provider == "gpt":
+            llm_model = "gpt"
+            if not api_key:
+                api_key = os.environ.get("OPENAI_API_KEY", "")
+
+        # Create the data analyst duo
+        duo = DataAnalystDuo(llm_model=llm_model, api_key=api_key)
+
+        # Prepare analysis parameters
+        analysis_params = {
+            "clean_params": {
+                "missing_strategy": missing_strategy,
+                "remove_duplicates": True
+            },
+            "stats_params": {
+                "descriptive": True,
+                "central_tendency": True,
+                "dispersion": True
+            },
+            "corr_params": {
+                "method": "pearson"
+            }
+        }
+
+        # Add filter for high fiber if requested
+        if high_fiber_filter:
+            analysis_params["filter_params"] = {
+                "filters": [
+                    {"column": "fiber", "operator": ">", "value": 5}
+                ]
+            }
+
+            # Add aggregation by manufacturer
+            analysis_params["agg_params"] = {
+                "groupby": ["mfr"],
+                "columns": ["calories", "protein", "fat", "fiber", "sugars"],
+                "functions": ["mean", "min", "max"]
+            }
+
+        # Add visualizations if requested
+        if create_visualizations:
+            analysis_params["viz_params"] = [
+                {
+                    "type": "scatter",
+                    "title": "Calories vs Sugar Content",
+                    "x": "calories",
+                    "y": "sugars"
+                },
+                {
+                    "type": "histogram",
+                    "title": "Distribution of Fiber Content",
+                    "x": "fiber"
+                },
+                {
+                    "type": "heatmap",
+                    "title": "Correlation Matrix",
+                    "columns": ["calories", "protein", "fat", "fiber", "sugars", "rating"]
+                }
+            ]
+
+        # Run the analysis
+        results = duo.run_analysis(dataset_url, analysis_params)
+
+        # Extract MCP messages for display
+        compute_messages = results["compute_agent_messages"]
+        interpret_messages = results["interpret_agent_messages"]
+
+        # Extract LLM logs
+        compute_llm_logs = results["compute_agent_llm_logs"]
+        interpret_llm_logs = results["interpret_agent_llm_logs"]
+
+        # Format messages for display
+        formatted_messages = []
+
+        # Combine and sort messages by timestamp
+        all_messages = []
+        for msg in compute_messages:
+            msg_copy = msg.copy()
+            msg_copy["agent"] = "ComputeAgent"
+            all_messages.append(msg_copy)
+
+        for msg in interpret_messages:
+            msg_copy = msg.copy()
+            msg_copy["agent"] = "InterpretAgent"
+            all_messages.append(msg_copy)
+
+        # Sort by timestamp
+        all_messages.sort(key=lambda x: x["message"]["timestamp"])
+
+        # Format for display
+        for msg in all_messages:
+            agent = msg["agent"]
+            direction = msg["type"]
+            message = msg["message"]
+
+            formatted_msg = f"[{message['timestamp']}] {agent} {direction.upper()} - Type: {message['message_type']}\n"
+            formatted_msg += format_json(message['content'])
+            formatted_msg += "\n\n" + "-"*80 + "\n\n"
+            formatted_messages.append(formatted_msg)
+
+        # Format LLM logs
+        formatted_llm_logs = []
+
+        for log in compute_llm_logs + interpret_llm_logs:
+            formatted_log = f"[{log['timestamp']}]\n"
+            formatted_log += "PROMPT:\n" + log['prompt'] + "\n\n"
+            formatted_log += "RESPONSE:\n" + log['response'] + "\n\n"
+            formatted_log += "-"*80 + "\n\n"
+            formatted_llm_logs.append(formatted_log)
+
+        # Prepare visualization display
+        viz_html = ""
+        if create_visualizations and "visualizations" in results and results["visualizations"]:
+            viz_html = "<div style='display: flex; flex-wrap: wrap;'>"
+            for viz_id, viz_data in results["visualizations"].items():
+                viz_html += f"<div style='margin: 10px;'>"
+                viz_html += f"<h3>{viz_data['title']}</h3>"
+                viz_html += f"<img src='file={viz_data['filename']}' width='400' />"
+                viz_html += "</div>"
+            viz_html += "</div>"
+
+        # Get the final report
+        report_html = "<h2>No report generated</h2>"
+        if "report_result" in [msg["message"]["message_type"] for msg in compute_messages if msg["type"] == "received"]:
+            # Find the report message
+            for msg in compute_messages:
+                if msg["type"] == "received" and msg["message"]["message_type"] == "report_result":
+                    report_content = msg["message"]["content"]["report"]
+                    if "content" in report_content:
+                        # LLM-generated report
+                        report_html = f"<h2>{report_content['title']}</h2>"
+                        report_html += f"<div>{report_content['content'].replace('\n', '<br/>')}</div>"
+                    elif "sections" in report_content:
+                        # Template-based report
+                        report_html = f"<h2>{report_content['title']}</h2>"
+                        for section in report_content["sections"]:
+                            report_html += f"<h3>{section['title']}</h3>"
+                            report_html += f"<p>{section['content']}</p>"
+                            if "insights" in section:
+                                report_html += "<ul>"
+                                for insight in section["insights"]:
+                                    report_html += f"<li>{insight}</li>"
+                                report_html += "</ul>"
+                            if "data" in section:
+                                report_html += "<pre>" + format_json(section["data"]) + "</pre>"
+
+        # Return all results
+        return {
+            'mcp_messages': "\n".join(formatted_messages),
+            'llm_logs': "\n".join(formatted_llm_logs),
+            'visualizations': viz_html,
+            'final_report': report_html
+        }
+
+    except Exception as e:
+        import traceback
+        return {
+            'mcp_messages': f"Error: {str(e)}\n\n{traceback.format_exc()}",
+            'llm_logs': "",
+            'visualizations': "",
+            'final_report': ""
+        }
+
+# Define the Gradio interface
+def create_interface():
+    with gr.Blocks(title="Data Analyst Duo - MCP Implementation") as app:
+        gr.Markdown("""
+        # Data Analyst Duo - Model Context Protocol (MCP) Implementation
+
+        This application demonstrates a multi-agent system using the Model Context Protocol (MCP).
+        It consists of two agents:
+
+        1. **ComputeAgent**: Responsible for data loading, cleaning, and computation
+        2. **InterpretAgent**: Responsible for interpreting results and visualizing data
+
+        The agents communicate directly using standardized MCP messages, showcasing agent-to-agent communication.
+        """)
+
+        with gr.Row():
+            with gr.Column():
+                dataset_url = gr.Textbox(label="Dataset URL (leave empty for default cereals dataset)", placeholder="Enter dataset URL or leave empty for default")
+
+                with gr.Row():
+                    llm_provider = gr.Radio(["none", "claude", "gpt"], label="LLM Provider (Optional)", value="none")
+                    api_key = gr.Textbox(label="API Key (if using LLM)", placeholder="Enter API key if using Claude or GPT")
+
+                with gr.Row():
+                    missing_strategy = gr.Dropdown(["drop", "mean", "median", "mode"], label="Missing Values Strategy", value="mean")
+                    create_visualizations = gr.Checkbox(label="Create Visualizations", value=True)
+                    high_fiber_filter = gr.Checkbox(label="Filter for High Fiber & Aggregate by Manufacturer", value=True)
+
+                run_button = gr.Button("Run Data Analysis")
+
+        with gr.Row():
+            with gr.Tab("MCP Messages"):
+                mcp_messages = gr.Textbox(label="MCP Message Flow", lines=20)
+            with gr.Tab("LLM Logs"):
+                llm_logs = gr.Textbox(label="LLM Interaction Logs", lines=20)
+
+        with gr.Row():
+            with gr.Tab("Visualizations"):
+                visualizations = gr.HTML(label="Data Visualizations")
+            with gr.Tab("Final Report"):
+                final_report = gr.HTML(label="Analysis Report")
+
+        # Connect the button to the analysis function
+        run_button.click(
+            fn=run_data_analysis,
+            inputs=[dataset_url, llm_provider, api_key, missing_strategy, create_visualizations, high_fiber_filter],
+            outputs=[mcp_messages, llm_logs, visualizations, final_report]
+        )
+
+        gr.Markdown("""
+        ## How This Demonstrates MCP
+
+        This application shows the Model Context Protocol in action:
+
+        1. **Standardized Message Structure**: All communication between agents follows a consistent format
+        2. **Tool Registration**: Agents register their capabilities as tools with descriptions
+        3. **Direct Peer Communication**: Agents communicate directly with structured messages
+        4. **Asynchronous Processing**: Each agent processes messages independently
+
+        The message flow display shows the exact JSON messages exchanged between agents, demonstrating the protocol in action.
+        """)
+
+    return app
+
+# Create and launch the interface
+if __name__ == "__main__":
+    app = create_interface()
+    app.launch()

requirements.txt

@@ -0,0 +1,9 @@
+gradio==4.13.0
+pandas==2.1.1
+numpy==1.26.0
+matplotlib==3.8.0
+seaborn==0.13.0
+anthropic==0.8.1
+openai==1.1.1
+python-dotenv==1.0.0
+requests==2.31.0