Neha-Rudraraju/MCP

1"""2app.py — Data Analyst Duo MCP (no OpenAI) Gradio Space3Shows preview table, stats, corr, plus full JSON histories, with rule-based interpretation.4"""5 6import os7import uuid8import logging9import datetime10 11import pandas as pd12import numpy as np13import gradio as gr14 15# ——— Logging ——————————————————————————————————————————————16logging.basicConfig(17    level=logging.INFO,18    format="%(asctime)s %(levelname)s:%(name)s: %(message)s"19)20logger = logging.getLogger("DataAnalystDuo")21 22# ——— MCP Core ——————————————————————————————————————————————23class MCPMessage:24    def __init__(self, sender, message_type, content):25        self.id = str(uuid.uuid4())26        self.sender = sender27        self.message_type = message_type28        self.content = content29        self.timestamp = datetime.datetime.now().isoformat()30 31    def to_dict(self):32        return {33            "id": self.id,34            "sender": self.sender,35            "message_type": self.message_type,36            "content": self.content,37            "timestamp": self.timestamp,38        }39 40class MCPTool:41    def __init__(self, name, description, func):42        self.name = name43        self.description = description44        self.func = func45 46    def execute(self, params):47        return self.func(params)48 49class MCPAgent:50    def __init__(self, name, description):51        self.name = name52        self.description = description53        self.tools = {}54        self.peers = {}55        self.queue = []56        self.history = []57 58    def register_tool(self, tool):59        self.tools[tool.name] = tool60 61    def connect(self, peer):62        self.peers[peer.name] = peer63 64    def send_message(self, to, mtype, content):65        if to not in self.peers:66            raise ValueError(f"Peer {to} not found")67        msg = MCPMessage(self.name, mtype, content)68        self.history.append({"type": "sent", "message": msg.to_dict()})69        self.peers[to].receive(msg)70        logger.info(f"{self.name} → {to}: {mtype}")71        return msg.to_dict()72 73    def receive(self, msg):74        self.queue.append(msg)75        self.history.append({"type": "received", "message": msg.to_dict()})76        logger.info(f"{self.name} received {msg.message_type} from {msg.sender}")77 78    def process(self):79        while self.queue:80            msg = self.queue.pop(0)81            self.handle_message(msg)82 83    def handle_message(self, message):84        raise NotImplementedError85 86    def get_history(self):87        return self.history88 89# ——— ComputeAgent ——————————————————————————————————————————————90class ComputeAgent(MCPAgent):91    def __init__(self):92        super().__init__("ComputeAgent", "Loads & computes data")93        self.df = None94        self.register_tool(MCPTool("load_dataset", "Load CSV", self._load))95        self.register_tool(MCPTool("compute_statistics", "Stats", self._stats))96        self.register_tool(MCPTool("compute_correlation", "Corr", self._corr))97 98    def _load(self, params):99        url = params.get("url", "").strip()100        if not url:101            url = "https://raw.githubusercontent.com/mwaskom/seaborn-data/master/diamonds.csv"102        try:103            self.df = pd.read_csv(url)104            return {105                "status": "success",106                "rows": self.df.shape[0],107                "columns": list(self.df.columns),108                "preview": self.df.head(5).to_dict(orient="records")109            }110        except Exception as e:111            logger.exception("Load failed")112            return {"status": "error", "message": str(e)}113 114    def _stats(self, params):115        if self.df is None:116            return {"status": "error", "message": "No data loaded"}117        cols = self.df.select_dtypes(include=[np.number]).columns118        stats = self.df[cols].describe().to_dict()119        return {"status": "success", "statistics": stats}120 121    def _corr(self, params):122        if self.df is None:123            return {"status": "error", "message": "No data loaded"}124        cols = self.df.select_dtypes(include=[np.number]).columns125        corr = self.df[cols].corr().to_dict()126        return {"status": "success", "correlation_matrix": corr}127 128    def handle_message(self, m):129        if m.message_type == "request_data_load":130            res = self._load(m.content)131            self.send_message(m.sender, "data_load_result", res)132        elif m.message_type == "request_statistics":133            res = self._stats(m.content)134            self.send_message(m.sender, "statistics_result", res)135        elif m.message_type == "request_correlation":136            res = self._corr(m.content)137            self.send_message(m.sender, "correlation_result", res)138 139# ——— InterpretAgent ——————————————————————————————————————————————140class InterpretAgent(MCPAgent):141    def __init__(self):142        super().__init__("InterpretAgent", "Generates insights from stats & corr")143        self.data_info = None144        self.stats = None145        self.corr = None146        self.register_tool(MCPTool("interpret_statistics", "Rule-based stats insights", self._int_stats))147        self.register_tool(MCPTool("interpret_correlation", "Rule-based corr insights", self._int_corr))148 149    def _int_stats(self, params):150        stats = self.stats.get("statistics", {})151        insights = []152        # Pick top 3 columns by range (max-min)153        ranges = {col: vals.get("max", 0) - vals.get("min", 0) for col, vals in stats.items()}154        top3 = sorted(ranges, key=ranges.get, reverse=True)[:3]155        for col in top3:156            vals = stats[col]157            insights.append(f"{col}: mean={vals['mean']:.2f}, range=[{vals['min']:.2f}, {vals['max']:.2f}]")158        return {"status": "success", "insights": insights, "summary": "Top 3 columns by range"}159 160    def _int_corr(self, params):161        cm = self.corr.get("correlation_matrix", {})162        pairs = []163        for c1, row in cm.items():164            for c2, val in row.items():165                if c1 != c2:166                    pairs.append((c1, c2, val))167        # sort by absolute correlation168        top3 = sorted(pairs, key=lambda x: abs(x[2]), reverse=True)[:3]169        insights = [f"{c1} vs {c2}: corr={corr:.2f}" for c1, c2, corr in top3]170        return {"status": "success", "insights": insights, "summary": "Top 3 correlated pairs"}171 172    def handle_message(self, m):173        if m.message_type == "data_load_result":174            self.data_info = m.content175            self.send_message(m.sender, "ack", {"status": "loaded"})176        elif m.message_type == "statistics_result":177            self.stats = m.content178            res = self.tools["interpret_statistics"].execute({})179            self.send_message(m.sender, "statistics_interpretation", res)180        elif m.message_type == "correlation_result":181            self.corr = m.content182            res = self.tools["interpret_correlation"].execute({})183            self.send_message(m.sender, "correlation_interpretation", res)184        elif m.message_type == "request_report":185            # assemble a simple markdown report186            report_md = "## Analysis Report\n"187            report_md += "### Stats Insights\n- " + "\n- ".join(self.tools["interpret_statistics"].execute({})["insights"]) + "\n"188            report_md += "### Corr Insights\n- " + "\n- ".join(self.tools["interpret_correlation"].execute({})["insights"]) + "\n"189            self.send_message(m.sender, "report_result", {"status": "success", "report_md": report_md})190 191# ——— Orchestration —————————————————————————————————————————————192class DataAnalystDuo:193    def __init__(self):194        self.C = ComputeAgent()195        self.I = InterpretAgent()196        self.C.connect(self.I)197        self.I.connect(self.C)198 199    def run(self, url):200        self.I.send_message("ComputeAgent", "request_data_load", {"url": url})201        self.C.process(); self.I.process()202        self.I.send_message("ComputeAgent", "request_statistics", {})203        self.C.process(); self.I.process()204        self.I.send_message("ComputeAgent", "request_correlation", {})205        self.C.process(); self.I.process()206        self.C.send_message("InterpretAgent", "request_report", {})207        self.I.process(); self.C.process()208 209        hist_c = self.C.get_history()210        hist_i = self.I.get_history()211        load = next(m['message']['content'] for m in hist_c if m['message']['message_type']=='data_load_result')212        stats = next(m['message']['content'] for m in hist_c if m['message']['message_type']=='statistics_result')213        corr = next(m['message']['content'] for m in hist_c if m['message']['message_type']=='correlation_result')214        preview_df = pd.DataFrame(load.get('preview', []))215        # extract latest report216        report = next(m['message']['content'] for m in hist_i if m['message']['message_type']=='report_result')217        return preview_df, stats, corr, hist_c, hist_i, report['report_md']218 219# ——— Gradio app —————————————————————————————————————————————220def run_analysis(url: str):221    return DataAnalystDuo().run(url)222 223demo = gr.Interface(224    fn=run_analysis,225    inputs=[gr.Textbox(label="CSV URL", placeholder="https://...")],226    outputs=[227        gr.Dataframe(label="Preview (first 5 rows)"),228        gr.JSON(label="Statistics"),229        gr.JSON(label="Correlation Matrix"),230        gr.JSON(label="Compute History"),231        gr.JSON(label="Interpret History"),232        gr.Markdown(label="Analysis Report")233    ],234    title="Data Analyst Duo",235    description="Paste any CSV URL (e.g. diamonds.csv) to see data + stats + insights + report"236)237 238if __name__ == "__main__":239    demo.launch(240        server_name="0.0.0.0",241        server_port=int(os.environ.get("PORT", 7860)),242        share=True243    )244