数据检索与分析平台

# 创建虚拟环境
python -m venv mcp-data-platform
source mcp-data-platform/bin/activate  # Linux/Mac
# 或
mcp-data-platform\Scripts\activate  # Windows

# 安装依赖
pip install mcp-python pandas numpy matplotlib plotly fastapi uvicorn sqlalchemy

# server/data_manager.py
import os
import pandas as pd
import json
import sqlite3
from typing import Dict, List, Any, Optional, Union
from abc import ABC, abstractmethod

class DataSource(ABC):
    """数据源抽象基类"""
    
    @abstractmethod
    def read(self) -> pd.DataFrame:
        """读取数据，返回DataFrame"""
        pass
    
    @abstractmethod
    def get_metadata(self) -> Dict[str, Any]:
        """获取数据源元数据"""
        pass

class CsvDataSource(DataSource):
    """CSV数据源"""
    
    def __init__(self, file_path: str, **kwargs):
        self.file_path = file_path
        self.options = kwargs
    
    def read(self) -> pd.DataFrame:
        return pd.read_csv(self.file_path, **self.options)
    
    def get_metadata(self) -> Dict[str, Any]:
        file_stats = os.stat(self.file_path)
        return {
            "type": "csv",
            "path": self.file_path,
            "size_bytes": file_stats.st_size,
            "modified_time": file_stats.st_mtime
        }

class JsonDataSource(DataSource):
    """JSON数据源"""
    
    def __init__(self, file_path: str, record_path: Optional[Union[str, List[str]]] = None):
        self.file_path = file_path
        self.record_path = record_path
    
    def read(self) -> pd.DataFrame:
        with open(self.file_path, 'r') as f:
            data = json.load(f)
        
        if self.record_path:
            return pd.json_normalize(data, record_path=self.record_path)
        else:
            return pd.json_normalize(data)
    
    def get_metadata(self) -> Dict[str, Any]:
        file_stats = os.stat(self.file_path)
        return {
            "type": "json",
            "path": self.file_path,
            "size_bytes": file_stats.st_size,
            "modified_time": file_stats.st_mtime
        }

class SqlDataSource(DataSource):
    """SQL数据源"""
    
    def __init__(self, db_path: str, query: str):
        self.db_path = db_path
        self.query = query
    
    def read(self) -> pd.DataFrame:
        conn = sqlite3.connect(self.db_path)
        df = pd.read_sql_query(self.query, conn)
        conn.close()
        return df
    
    def get_metadata(self) -> Dict[str, Any]:
        return {
            "type": "sql",
            "db_path": self.db_path,
            "query": self.query
        }

# server/data_manager.py (续)
import uuid
import os
from datetime import datetime

class DatasetManager:
    """数据集管理器"""
    
    def __init__(self, storage_dir: str = "data/datasets"):
        self.storage_dir = storage_dir
        os.makedirs(storage_dir, exist_ok=True)
        self.datasets = {}
        self.metadata = {}
    
    def import_dataset(self, name: str, data_source: DataSource) -> str:
        """导入数据集"""
        # 读取数据
        df = data_source.read()
        
        # 生成唯一ID
        dataset_id = str(uuid.uuid4())
        
        # 保存数据集
        self.datasets[dataset_id] = df
        
        # 保存元数据
        self.metadata[dataset_id] = {
            "id": dataset_id,
            "name": name,
            "rows": len(df),
            "columns": list(df.columns),
            "created_at": datetime.now().isoformat(),
            "source": data_source.get_metadata()
        }
        
        # 持久化存储
        df.to_parquet(os.path.join(self.storage_dir, f"{dataset_id}.parquet"))
        
        with open(os.path.join(self.storage_dir, f"{dataset_id}.meta.json"), 'w') as f:
            json.dump(self.metadata[dataset_id], f)
        
        return dataset_id
    
    def get_dataset(self, dataset_id: str) -> Optional[pd.DataFrame]:
        """获取数据集"""
        if dataset_id in self.datasets:
            return self.datasets[dataset_id]
        
        # 尝试从存储加载
        try:
            file_path = os.path.join(self.storage_dir, f"{dataset_id}.parquet")
            if os.path.exists(file_path):
                df = pd.read_parquet(file_path)
                self.datasets[dataset_id] = df
                return df
        except Exception as e:
            print(f"Error loading dataset {dataset_id}: {e}")
        
        return None
    
    def get_metadata(self, dataset_id: str) -> Optional[Dict[str, Any]]:
        """获取数据集元数据"""
        if dataset_id in self.metadata:
            return self.metadata[dataset_id]
        
        # 尝试从存储加载
        try:
            meta_path = os.path.join(self.storage_dir, f"{dataset_id}.meta.json")
            if os.path.exists(meta_path):
                with open(meta_path, 'r') as f:
                    meta = json.load(f)
                self.metadata[dataset_id] = meta
                return meta
        except Exception as e:
            print(f"Error loading metadata for dataset {dataset_id}: {e}")
        
        return None
    
    def list_datasets(self) -> List[Dict[str, Any]]:
        """列出所有数据集"""
        # 确保加载了所有元数据
        for file in os.listdir(self.storage_dir):
            if file.endswith('.meta.json'):
                dataset_id = file.split('.meta.json')[0]
                if dataset_id not in self.metadata:
                    self.get_metadata(dataset_id)
        
        return list(self.metadata.values())
    
    def delete_dataset(self, dataset_id: str) -> bool:
        """删除数据集"""
        if dataset_id not in self.metadata:
            return False
        
        # 删除内存中的数据
        if dataset_id in self.datasets:
            del self.datasets[dataset_id]
        
        if dataset_id in self.metadata:
            del self.metadata[dataset_id]
        
        # 删除文件
        try:
            data_path = os.path.join(self.storage_dir, f"{dataset_id}.parquet")
            meta_path = os.path.join(self.storage_dir, f"{dataset_id}.meta.json")
            
            if os.path.exists(data_path):
                os.remove(data_path)
            
            if os.path.exists(meta_path):
                os.remove(meta_path)
            
            return True
        except Exception as e:
            print(f"Error deleting dataset {dataset_id}: {e}")
            return False

# server/query_engine.py
from typing import Dict, List, Any, Optional, Union
import pandas as pd
import numpy as np
from dataclasses import dataclass

@dataclass
class QueryCondition:
    """查询条件"""
    column: str
    operator: str  # '=', '!=', '>', '<', '>=', '<=', 'contains', 'in', 'not in'
    value: Any

@dataclass
class QueryOrder:
    """排序条件"""
    column: str
    ascending: bool = True

@dataclass
class QueryOptions:
    """查询选项"""
    limit: Optional[int] = None
    offset: Optional[int] = 0
    columns: Optional[List[str]] = None

class QueryBuilder:
    """查询构建器"""
    
    def __init__(self):
        self.conditions = []
        self.orders = []
        self.options = QueryOptions()
    
    def where(self, column: str, operator: str, value: Any) -> 'QueryBuilder':
        """添加条件"""
        self.conditions.append(QueryCondition(column, operator, value))
        return self
    
    def order_by(self, column: str, ascending: bool = True) -> 'QueryBuilder':
        """添加排序"""
        self.orders.append(QueryOrder(column, ascending))
        return self
    
    def limit(self, limit: int) -> 'QueryBuilder':
        """设置返回数量限制"""
        self.options.limit = limit
        return self
    
    def offset(self, offset: int) -> 'QueryBuilder':
        """设置偏移量"""
        self.options.offset = offset
        return self
    
    def select(self, columns: List[str]) -> 'QueryBuilder':
        """选择返回的列"""
        self.options.columns = columns
        return self

class QueryEngine:
    """查询引擎"""
    
    def __init__(self, dataset_manager):
        self.dataset_manager = dataset_manager
    
    def execute_query(self, dataset_id: str, query_builder: QueryBuilder) -> pd.DataFrame:
        """执行查询"""
        # 获取数据集
        df = self.dataset_manager.get_dataset(dataset_id)
        if df is None:
            raise ValueError(f"Dataset {dataset_id} not found")
        
        # 应用查询条件
        df_result = df.copy()
        for condition in query_builder.conditions:
            df_result = self._apply_condition(df_result, condition)
        
        # 应用排序
        if query_builder.orders:
            columns = [order.column for order in query_builder.orders]
            ascending = [order.ascending for order in query_builder.orders]
            df_result = df_result.sort_values(by=columns, ascending=ascending)
        
        # 应用列选择
        if query_builder.options.columns:
            df_result = df_result[query_builder.options.columns]
        
        # 应用分页
        if query_builder.options.offset is not None:
            df_result = df_result.iloc[query_builder.options.offset:]
        
        if query_builder.options.limit is not None:
            df_result = df_result.iloc[:query_builder.options.limit]
        
        return df_result
    
    def _apply_condition(self, df: pd.DataFrame, condition: QueryCondition) -> pd.DataFrame:
        """应用查询条件"""
        column = condition.column
        operator = condition.operator
        value = condition.value
        
        if column not in df.columns:
            raise ValueError(f"Column {column} not found")
        
        if operator == '=':
            return df[df[column] == value]
        elif operator == '!=':
            return df[df[column] != value]
        elif operator == '>':
            return df[df[column] > value]
        elif operator == '<':
            return df[df[column] < value]
        elif operator == '>=':
            return df[df[column] >= value]
        elif operator == '<=':
            return df[df[column] <= value]
        elif operator == 'contains':
            if df[column].dtype == 'object':  # 字符串类型
                return df[df[column].str.contains(value, na=False)]
            else:
                raise ValueError(f"Contains operator not supported for column {column}")
        elif operator == 'in':
            return df[df[column].isin(value)]
        elif operator == 'not in':
            return df[~df[column].isin(value)]
        else:
            raise ValueError(f"Unsupported operator: {operator}")

    def execute_sql(self, dataset_id: str, sql_query: str) -> pd.DataFrame:
        """执行SQL查询"""
        # 获取数据集
        df = self.dataset_manager.get_dataset(dataset_id)
        if df is None:
            raise ValueError(f"Dataset {dataset_id} not found")
        
        # 使用pandas的query功能
        try:
            return df.query(sql_query)
        except Exception as e:
            raise ValueError(f"SQL query error: {e}")

# server/query_engine.py (续)
from typing import Dict, List, Tuple

class QueryResultProcessor:
    """查询结果处理器"""
    
    @staticmethod
    def to_dict(df: pd.DataFrame) -> List[Dict[str, Any]]:
        """转换为字典列表"""
        return df.to_dict(orient='records')
    
    @staticmethod
    def get_summary(df: pd.DataFrame) -> Dict[str, Any]:
        """获取数据摘要"""
        summary = {
            "row_count": len(df),
            "column_count": len(df.columns),
            "columns": {}
        }
        
        for col in df.columns:
            col_type = str(df[col].dtype)
            if np.issubdtype(df[col].dtype, np.number):
                summary["columns"][col] = {
                    "type": col_type,
                    "min": df[col].min(),
                    "max": df[col].max(),
                    "mean": df[col].mean(),
                    "median": df[col].median(),
                    "null_count": df[col].isna().sum()
                }
            else:
                summary["columns"][col] = {
                    "type": col_type,
                    "unique_count": df[col].nunique(),
                    "null_count": df[col].isna().sum()
                }
                # 如果唯一值较少，添加值分布
                if df[col].nunique() <= 10:
                    value_counts = df[col].value_counts().to_dict()
                    summary["columns"][col]["value_distribution"] = value_counts
        
        return summary
    
    @staticmethod
    def get_sample(df: pd.DataFrame, sample_size: int = 5) -> pd.DataFrame:
        """获取样本数据"""
        if len(df) <= sample_size:
            return df
        return df.sample(sample_size)
    
    @staticmethod
    def get_head_tail(df: pd.DataFrame, n: int = 5) -> Tuple[pd.DataFrame, pd.DataFrame]:
        """获取头尾数据"""
        return df.head(n), df.tail(n)

# server/analytics.py
from typing import Dict, List, Any, Optional, Union, Tuple
import pandas as pd
import numpy as np
from scipy import stats

class StatisticalAnalyzer:
    """基础统计分析工具"""
    
    @staticmethod
    def describe(df: pd.DataFrame, include: Optional[List[str]] = None) -> Dict[str, Any]:
        """获取统计描述"""
        if include:
            desc = df[include].describe(include='all')
        else:
            desc = df.describe(include='all')
        return desc.to_dict()
    
    @staticmethod
    def correlation_matrix(df: pd.DataFrame, 
                          method: str = 'pearson',
                          numeric_only: bool = True) -> Dict[str, Dict[str, float]]:
        """计算相关性矩阵"""
        if numeric_only:
            numeric_df = df.select_dtypes(include=np.number)
        else:
            numeric_df = df
            
        corr_matrix = numeric_df.corr(method=method)
        return corr_matrix.to_dict()
    
    @staticmethod
    def detect_outliers(df: pd.DataFrame, 
                       column: str, 
                       method: str = 'zscore',
                       threshold: float = 3.0) -> List[int]:
        """检测异常值"""
        if column not in df.columns:
            raise ValueError(f"Column {column} not found")
            
        values = df[column].dropna()
        
        if method == 'zscore':
            z_scores = np.abs(stats.zscore(values))
            outlier_indices = np.where(z_scores > threshold)[0]
        elif method == 'iqr':
            q1 = values.quantile(0.25)
            q3 = values.quantile(0.75)
            iqr = q3 - q1
            lower_bound = q1 - (threshold * iqr)
            upper_bound = q3 + (threshold * iqr)
            outlier_indices = values[(values < lower_bound) | (values > upper_bound)].index
        else:
            raise ValueError(f"Unsupported outlier detection method: {method}")
            
        return list(outlier_indices)
    
    @staticmethod
    def group_by_analysis(df: pd.DataFrame,
                         group_column: str,
                         agg_columns: List[str],
                         agg_functions: List[str]) -> pd.DataFrame:
        """分组分析"""
        if group_column not in df.columns:
            raise ValueError(f"Group column {group_column} not found")
            
        for col in agg_columns:
            if col not in df.columns:
                raise ValueError(f"Aggregation column {col} not found")
        
        # 构建聚合函数字典
        agg_dict = {}
        for col in agg_columns:
            agg_dict[col] = agg_functions
            
        result = df.groupby(group_column).agg(agg_dict)
        return result

# server/analytics.py (续)
from sklearn.preprocessing import StandardScaler, MinMaxScaler
from sklearn.decomposition import PCA
from sklearn.cluster import KMeans, DBSCAN
from sklearn.metrics import silhouette_score

class AdvancedAnalyzer:
    """高级分析工具"""
    
    @staticmethod
    def run_pca(df: pd.DataFrame, 
               n_components: int = 2,
               columns: Optional[List[str]] = None) -> Tuple[pd.DataFrame, pd.DataFrame]:
        """主成分分析"""
        if columns:
            data = df[columns]
        else:
            # 默认使用所有数值列
            data = df.select_dtypes(include=np.number)
            
        # 标准化数据
        scaler = StandardScaler()
        scaled_data = scaler.fit_transform(data)
        
        # PCA分析
        pca = PCA(n_components=n_components)
        principal_components = pca.fit_transform(scaled_data)
        
        # 创建主成分DataFrame
        pc_columns = [f"PC{i+1}" for i in range(n_components)]
        pc_df = pd.DataFrame(data=principal_components, columns=pc_columns)
        
        # 创建解释方差DataFrame
        explained_variance = pca.explained_variance_ratio_
        loadings = pd.DataFrame(
            pca.components_.T,
            columns=pc_columns,
            index=data.columns
        )
        
        return pc_df, loadings
    
    @staticmethod
    def cluster_analysis(df: pd.DataFrame,
                        algorithm: str = 'kmeans',
                        n_clusters: int = 3,
                        columns: Optional[List[str]] = None) -> Tuple[List[int], Dict[str, Any]]:
        """聚类分析"""
        if columns:
            data = df[columns]
        else:
            # 默认使用所有数值列
            data = df.select_dtypes(include=np.number)
            
        # 标准化数据
        scaler = MinMaxScaler()
        scaled_data = scaler.fit_transform(data)
        
        # 聚类分析
        if algorithm == 'kmeans':
            model = KMeans(n_clusters=n_clusters, random_state=42)
            clusters = model.fit_predict(scaled_data)
            
            # 评估聚类质量
            silhouette_avg = silhouette_score(scaled_data, clusters)
            inertia = model.inertia_
            
            metrics = {
                "silhouette_score": silhouette_avg,
                "inertia": inertia,
                "cluster_centers": model.cluster_centers_.tolist()
            }
            
        elif algorithm == 'dbscan':
            model = DBSCAN(eps=0.5, min_samples=5)
            clusters = model.fit_predict(scaled_data)
            
            # 评估聚类质量
            if len(set(clusters)) > 1:  # 只有在有多个聚类的情况下计算
                silhouette_avg = silhouette_score(scaled_data, clusters)
            else:
                silhouette_avg = 0
                
            n_noise = list(clusters).count(-1)
            
            metrics = {
                "silhouette_score": silhouette_avg,
                "n_clusters": len(set(clusters)) - (1 if -1 in clusters else 0),
                "n_noise": n_noise
            }
            
        else:
            raise ValueError(f"Unsupported clustering algorithm: {algorithm}")
            
        return clusters.tolist(), metrics

# server/visualizer.py
from typing import Dict, List, Any, Optional, Union, Tuple
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
import io
import base64
import json
from dataclasses import dataclass

@dataclass
class VisualizationOptions:
    """可视化选项"""
    title: Optional[str] = None
    x_label: Optional[str] = None
    y_label: Optional[str] = None
    figsize: Tuple[int, int] = (10, 6)
    color: Optional[str] = None
    palette: Optional[str] = None
    font_size: int = 12

class MatplotlibVisualizer:
    """基于Matplotlib的可视化工具"""
    
    @staticmethod
    def _prepare_figure(options: VisualizationOptions):
        """准备图形"""
        plt.figure(figsize=options.figsize)
        if options.title:
            plt.title(options.title, fontsize=options.font_size + 2)
        if options.x_label:
            plt.xlabel(options.x_label, fontsize=options.font_size)
        if options.y_label:
            plt.ylabel(options.y_label, fontsize=options.font_size)
        plt.xticks(fontsize=options.font_size - 2)
        plt.yticks(fontsize=options.font_size - 2)
    
    @staticmethod
    def _figure_to_base64():
        """转换图形为base64编码"""
        buffer = io.BytesIO()
        plt.savefig(buffer, format='png', bbox_inches='tight')
        buffer.seek(0)
        image_png = buffer.getvalue()
        buffer.close()
        plt.close()
        return base64.b64encode(image_png).decode('utf-8')
    
    @staticmethod
    def bar_chart(df: pd.DataFrame, 
                 x_column: str, 
                 y_column: str,
                 options: Optional[VisualizationOptions] = None) -> str:
        """生成条形图"""
        if options is None:
            options = VisualizationOptions()
            
        MatplotlibVisualizer._prepare_figure(options)
        
        sns.barplot(x=x_column, y=y_column, data=df, color=options.color, palette=options.palette)
        
        return MatplotlibVisualizer._figure_to_base64()
    
    @staticmethod
    def line_chart(df: pd.DataFrame, 
                  x_column: str, 
                  y_column: str,
                  options: Optional[VisualizationOptions] = None) -> str:
        """生成折线图"""
        if options is None:
            options = VisualizationOptions()
            
        MatplotlibVisualizer._prepare_figure(options)
        
        sns.lineplot(x=x_column, y=y_column, data=df, color=options.color, palette=options.palette)
        
        return MatplotlibVisualizer._figure_to_base64()
    
    @staticmethod
    def scatter_plot(df: pd.DataFrame, 
                    x_column: str, 
                    y_column: str,
                    hue_column: Optional[str] = None,
                    options: Optional[VisualizationOptions] = None) -> str:
        """生成散点图"""
        if options is None:
            options = VisualizationOptions()
            
        MatplotlibVisualizer._prepare_figure(options)
        
        sns.scatterplot(
            x=x_column, 
            y=y_column, 
            hue=hue_column, 
            data=df, 
            palette=options.palette
        )
        
        return MatplotlibVisualizer._figure_to_base64()
    
    @staticmethod
    def histogram(df: pd.DataFrame, 
                 column: str,
                 bins: int = 10,
                 options: Optional[VisualizationOptions] = None) -> str:
        """生成直方图"""
        if options is None:
            options = VisualizationOptions()
            
        MatplotlibVisualizer._prepare_figure(options)
        
        sns.histplot(df[column], bins=bins, color=options.color)
        
        return MatplotlibVisualizer._figure_to_base64()
    
    @staticmethod
    def box_plot(df: pd.DataFrame, 
                x_column: Optional[str] = None,
                y_column: str = None,
                options: Optional[VisualizationOptions] = None) -> str:
        """生成箱线图"""
        if options is None:
            options = VisualizationOptions()
            
        MatplotlibVisualizer._prepare_figure(options)
        
        sns.boxplot(x=x_column, y=y_column, data=df, color=options.color, palette=options.palette)
        
        return MatplotlibVisualizer._figure_to_base64()
    
    @staticmethod
    def heatmap(data: pd.DataFrame,
               options: Optional[VisualizationOptions] = None) -> str:
        """生成热力图"""
        if options is None:
            options = VisualizationOptions()
            
        MatplotlibVisualizer._prepare_figure(options)
        
        sns.heatmap(data, annot=True, cmap=options.palette or 'viridis', fmt='.2f')
        
        return MatplotlibVisualizer._figure_to_base64()

# server/visualizer.py (续)
import plotly.express as px
import plotly.graph_objects as go
from plotly.subplots import make_subplots

class PlotlyVisualizer:
    """基于Plotly的交互式可视化工具"""
    
    @staticmethod
    def bar_chart(df: pd.DataFrame, 
                 x_column: str, 
                 y_column: str,
                 color_column: Optional[str] = None,
                 title: Optional[str] = None) -> Dict[str, Any]:
        """生成交互式条形图"""
        fig = px.bar(
            df, 
            x=x_column, 
            y=y_column, 
            color=color_column,
            title=title
        )
        
        return json.loads(fig.to_json())
    
    @staticmethod
    def line_chart(df: pd.DataFrame, 
                  x_column: str, 
                  y_column: Union[str, List[str]],
                  color_column: Optional[str] = None,
                  title: Optional[str] = None) -> Dict[str, Any]:
        """生成交互式折线图"""
        fig = px.line(
            df, 
            x=x_column, 
            y=y_column, 
            color=color_column,
            title=title
        )
        
        return json.loads(fig.to_json())
    
    @staticmethod
    def scatter_plot(df: pd.DataFrame, 
                    x_column: str, 
                    y_column: str,
                    color_column: Optional[str] = None,
                    size_column: Optional[str] = None,
                    hover_data: Optional[List[str]] = None,
                    title: Optional[str] = None) -> Dict[str, Any]:
        """生成交互式散点图"""
        fig = px.scatter(
            df, 
            x=x_column, 
            y=y_column, 
            color=color_column,
            size=size_column,
            hover_data=hover_data,
            title=title
        )
        
        return json.loads(fig.to_json())
    
    @staticmethod
    def pie_chart(df: pd.DataFrame, 
                 names_column: str, 
                 values_column: str,
                 title: Optional[str] = None) -> Dict[str, Any]:
        """生成交互式饼图"""
        fig = px.pie(
            df, 
            names=names_column, 
            values=values_column,
            title=title
        )
        
        return json.loads(fig.to_json())
    
    @staticmethod
    def heatmap(data: pd.DataFrame,
               title: Optional[str] = None) -> Dict[str, Any]:
        """生成交互式热力图"""
        fig = px.imshow(
            data,
            text_auto=True,
            aspect="auto",
            title=title
        )
        
        return json.loads(fig.to_json())
    
    @staticmethod
    def box_plot(df: pd.DataFrame, 
                x_column: Optional[str] = None,
                y_column: str = None,
                color_column: Optional[str] = None,
                title: Optional[str] = None) -> Dict[str, Any]:
        """生成交互式箱线图"""
        fig = px.box(
            df, 
            x=x_column, 
            y=y_column, 
            color=color_column,
            title=title
        )
        
        return json.loads(fig.to_json())
    
    @staticmethod
    def histogram(df: pd.DataFrame, 
                 x_column: str,
                 color_column: Optional[str] = None,
                 nbins: int = 10,
                 title: Optional[str] = None) -> Dict[str, Any]:
        """生成交互式直方图"""
        fig = px.histogram(
            df, 
            x=x_column, 
            color=color_column,
            nbins=nbins,
            title=title
        )
        
        return json.loads(fig.to_json())

# server/main.py
import os
import asyncio
from typing import Dict, List, Any, Optional, Union

from mcp.server.fastmcp import FastMCP
from mcp.types import ResourceResponse, ResourceRequest, ToolCall, ToolCallResponse

from .data_manager import DatasetManager, CsvDataSource, JsonDataSource, SqlDataSource
from .query_engine import QueryEngine, QueryBuilder
from .analytics import StatisticalAnalyzer, AdvancedAnalyzer
from .visualizer import MatplotlibVisualizer, PlotlyVisualizer

class DataPlatformServer:
    """数据平台MCP服务器"""
    
    def __init__(self, data_dir: str = "data"):
        # 初始化组件
        self.dataset_manager = DatasetManager(os.path.join(data_dir, "datasets"))
        self.query_engine = QueryEngine(self.dataset_manager)
        
        # 创建MCP服务器
        self.app = FastMCP(title="数据检索与分析平台")
        
        # 注册资源和工具
        self._register_resources()
        self._register_tools()
    
    def _register_resources(self):
        """注册资源"""
        
        # 数据集列表资源
        @self.app.resource("/datasets")
        async def datasets_resource(request: ResourceRequest) -> ResourceResponse:
            datasets = self.dataset_manager.list_datasets()
            return ResourceResponse(content=datasets)
        
        # 数据集详情资源
        @self.app.resource("/datasets/{dataset_id}")
        async def dataset_detail_resource(request: ResourceRequest) -> ResourceResponse:
            dataset_id = request.params.get("dataset_id")
            metadata = self.dataset_manager.get_metadata(dataset_id)
            if metadata is None:
                return ResourceResponse(error=f"Dataset {dataset_id} not found", status=404)
            
            return ResourceResponse(content=metadata)
        
        # 数据集预览资源
        @self.app.resource("/datasets/{dataset_id}/preview")
        async def dataset_preview_resource(request: ResourceRequest) -> ResourceResponse:
            dataset_id = request.params.get("dataset_id")
            df = self.dataset_manager.get_dataset(dataset_id)
            if df is None:
                return ResourceResponse(error=f"Dataset {dataset_id} not found", status=404)
            
            # 获取前10行数据
            preview = df.head(10).to_dict(orient='records')
            return ResourceResponse(content=preview)
        
        # 数据字段信息资源
        @self.app.resource("/datasets/{dataset_id}/schema")
        async def dataset_schema_resource(request: ResourceRequest) -> ResourceResponse:
            dataset_id = request.params.get("dataset_id")
            df = self.dataset_manager.get_dataset(dataset_id)
            if df is None:
                return ResourceResponse(error=f"Dataset {dataset_id} not found", status=404)
            
            schema = []
            for column in df.columns:
                col_type = str(df[column].dtype)
                schema.append({
                    "name": column,
                    "type": col_type,
                    "nullable": df[column].isna().any()
                })
            
            return ResourceResponse(content=schema)
    
    def _register_tools(self):
        """注册工具"""
        
        # 数据导入工具
        @self.app.tool("import_csv_data")
        async def import_csv_data(name: str, file_path: str, **kwargs) -> Dict[str, Any]:
            """
            导入CSV数据文件
            
            参数:
            - name: 数据集名称
            - file_path: CSV文件路径
            - kwargs: 其他CSV读取选项
            
            返回:
            - 导入的数据集ID和基本信息
            """
            try:
                data_source = CsvDataSource(file_path, **kwargs)
                dataset_id = self.dataset_manager.import_dataset(name, data_source)
                return {
                    "dataset_id": dataset_id,
                    "status": "success",
                    "message": f"Successfully imported dataset '{name}'"
                }
            except Exception as e:
                return {
                    "status": "error",
                    "message": str(e)
                }
        
        @self.app.tool("import_json_data")
        async def import_json_data(name: str, file_path: str, record_path: Optional[str] = None) -> Dict[str, Any]:
            """
            导入JSON数据文件
            
            参数:
            - name: 数据集名称
            - file_path: JSON文件路径
            - record_path: JSON数组路径(可选)
            
            返回:
            - 导入的数据集ID和基本信息
            """
            try:
                data_source = JsonDataSource(file_path, record_path)
                dataset_id = self.dataset_manager.import_dataset(name, data_source)
                return {
                    "dataset_id": dataset_id,
                    "status": "success",
                    "message": f"Successfully imported dataset '{name}'"
                }
            except Exception as e:
                return {
                    "status": "error",
                    "message": str(e)
                }
        
        @self.app.tool("import_sql_data")
        async def import_sql_data(name: str, db_path: str, query: str) -> Dict[str, Any]:
            """
            导入SQL查询结果
            
            参数:
            - name: 数据集名称
            - db_path: SQLite数据库路径
            - query: SQL查询语句
            
            返回:
            - 导入的数据集ID和基本信息
            """
            try:
                data_source = SqlDataSource(db_path, query)
                dataset_id = self.dataset_manager.import_dataset(name, data_source)
                return {
                    "dataset_id": dataset_id,
                    "status": "success",
                    "message": f"Successfully imported dataset '{name}'"
                }
            except Exception as e:
                return {
                    "status": "error",
                    "message": str(e)
                }
        
        # 查询工具
        @self.app.tool("query_dataset")
        async def query_dataset(dataset_id: str, conditions: List[Dict[str, Any]], 
                              order_by: Optional[List[Dict[str, Any]]] = None,
                              limit: Optional[int] = None,
                              offset: Optional[int] = 0,
                              columns: Optional[List[str]] = None) -> Dict[str, Any]:
            """
            查询数据集
            
            参数:
            - dataset_id: 数据集ID
            - conditions: 查询条件列表 [{"column": "col1", "operator": "=", "value": 10}, ...]
            - order_by: 排序规则列表 [{"column": "col1", "ascending": true}, ...]
            - limit: 返回结果数量限制
            - offset: 结果偏移量
            - columns: 要返回的列名列表
            
            返回:
            - 查询结果和元数据
            """
            try:
                # 构建查询
                query_builder = QueryBuilder()
                
                # 添加条件
                for condition in conditions:
                    query_builder.where(
                        condition["column"],
                        condition["operator"],
                        condition["value"]
                    )
                
                # 添加排序
                if order_by:
                    for order in order_by:
                        query_builder.order_by(
                            order["column"],
                            order.get("ascending", True)
                        )
                
                # 设置分页和列选择
                if limit is not None:
                    query_builder.limit(limit)
                
                if offset is not None:
                    query_builder.offset(offset)
                    
                if columns:
                    query_builder.select(columns)
                
                # 执行查询
                result_df = self.query_engine.execute_query(dataset_id, query_builder)
                
                # 转换结果
                return {
                    "status": "success",
                    "data": result_df.to_dict(orient='records'),
                    "total_rows": len(result_df),
                    "columns": list(result_df.columns)
                }
                
            except Exception as e:
                return {
                    "status": "error",
                    "message": str(e)
                }
        
        # 数据分析工具
        @self.app.tool("statistical_analysis")
        async def statistical_analysis(dataset_id: str, columns: Optional[List[str]] = None) -> Dict[str, Any]:
            """
            数据统计分析
            
            参数:
            - dataset_id: 数据集ID
            - columns: 要分析的列名列表(可选)
            
            返回:
            - 统计分析结果
            """
            try:
                df = self.dataset_manager.get_dataset(dataset_id)
                if df is None:
                    return {"status": "error", "message": f"Dataset {dataset_id} not found"}
                
                if columns:
                    # 检查列是否存在
                    missing_columns = [col for col in columns if col not in df.columns]
                    if missing_columns:
                        return {
                            "status": "error",
                            "message": f"Columns {missing_columns} not found in dataset"
                        }
                    analysis_df = df[columns]
                else:
                    analysis_df = df
                
                stats = StatisticalAnalyzer.describe(analysis_df)
                
                return {
                    "status": "success",
                    "data": stats
                }
                
            except Exception as e:
                return {
                    "status": "error",
                    "message": str(e)
                }
        
        @self.app.tool("correlation_analysis")
        async def correlation_analysis(dataset_id: str, method: str = 'pearson') -> Dict[str, Any]:
            """
            相关性分析
            
            参数:
            - dataset_id: 数据集ID
            - method: 相关性计算方法 ('pearson', 'kendall', 'spearman')
            
            返回:
            - 相关性矩阵
            """
            try:
                df = self.dataset_manager.get_dataset(dataset_id)
                if df is None:
                    return {"status": "error", "message": f"Dataset {dataset_id} not found"}
                
                corr_matrix = StatisticalAnalyzer.correlation_matrix(df, method=method)
                
                return {
                    "status": "success",
                    "data": corr_matrix
                }
                
            except Exception as e:
                return {
                    "status": "error",
                    "message": str(e)
                }
        
        # 可视化工具
        @self.app.tool("generate_chart")
        async def generate_chart(
            dataset_id: str, 
            chart_type: str,
            x_column: Optional[str] = None,
            y_column: Optional[str] = None,
            color_column: Optional[str] = None,
            title: Optional[str] = None,
            interactive: bool = True
        ) -> Dict[str, Any]:
            """
            生成数据可视化图表
            
            参数:
            - dataset_id: 数据集ID
            - chart_type: 图表类型 ('bar', 'line', 'scatter', 'pie', 'histogram', 'box', 'heatmap')
            - x_column: X轴列名
            - y_column: Y轴列名
            - color_column: 颜色分组列名(可选)
            - title: 图表标题(可选)
            - interactive: 是否生成交互式图表
            
            返回:
            - 图表数据或base64编码的图像
            """
            try:
                df = self.dataset_manager.get_dataset(dataset_id)
                if df is None:
                    return {"status": "error", "message": f"Dataset {dataset_id} not found"}
                
                # 根据图表类型调用相应的方法
                if interactive:
                    visualizer = PlotlyVisualizer
                else:
                    visualizer = MatplotlibVisualizer
                    
                    # 创建可视化选项
                    options = VisualizationOptions(
                        title=title,
                        x_label=x_column,
                        y_label=y_column
                    )
                
                # 调用对应的可视化方法
                if chart_type == 'bar':
                    if x_column is None or y_column is None:
                        return {"status": "error", "message": "x_column and y_column are required for bar chart"}
                    
                    if interactive:
                        chart_data = visualizer.bar_chart(df, x_column, y_column, color_column, title)
                    else:
                        chart_data = visualizer.bar_chart(df, x_column, y_column, options)
                        
                elif chart_type == 'line':
                    if x_column is None or y_column is None:
                        return {"status": "error", "message": "x_column and y_column are required for line chart"}
                    
                    if interactive:
                        chart_data = visualizer.line_chart(df, x_column, y_column, color_column, title)
                    else:
                        chart_data = visualizer.line_chart(df, x_column, y_column, options)
                        
                elif chart_type == 'scatter':
                    if x_column is None or y_column is None:
                        return {"status": "error", "message": "x_column and y_column are required for scatter plot"}
                    
                    if interactive:
                        chart_data = visualizer.scatter_plot(df, x_column, y_column, color_column, title=title)
                    else:
                        chart_data = visualizer.scatter_plot(df, x_column, y_column, hue_column=color_column, options=options)
                        
                elif chart_type == 'histogram':
                    if x_column is None:
                        return {"status": "error", "message": "x_column is required for histogram"}
                    
                    if interactive:
                        chart_data = visualizer.histogram(df, x_column, color_column, title=title)
                    else:
                        chart_data = visualizer.histogram(df, x_column, options=options)
                        
                elif chart_type == 'box':
                    if y_column is None:
                        return {"status": "error", "message": "y_column is required for box plot"}
                    
                    if interactive:
                        chart_data = visualizer.box_plot(df, x_column, y_column, color_column, title)
                    else:
                        chart_data = visualizer.box_plot(df, x_column, y_column, options=options)
                        
                elif chart_type == 'heatmap':
                    if isinstance(df, pd.DataFrame) and df.select_dtypes(include=['number']).shape[1] < 2:
                        return {"status": "error", "message": "Need at least 2 numeric columns for heatmap"}
                    
                    # 获取数值列的相关性矩阵
                    numeric_df = df.select_dtypes(include=['number'])
                    corr_matrix = numeric_df.corr()
                    
                    if interactive:
                        chart_data = visualizer.heatmap(corr_matrix, title)
                    else:
                        chart_data = visualizer.heatmap(corr_matrix, options=options)
                        
                else:
                    return {"status": "error", "message": f"Unsupported chart type: {chart_type}"}
                
                return {
                    "status": "success",
                    "chart_type": chart_type,
                    "interactive": interactive,
                    "data": chart_data
                }
                
            except Exception as e:
                return {
                    "status": "error",
                    "message": str(e)
                }
        
    async def start(self, host: str = "localhost", port: int = 8000):
        """启动服务器"""
        await self.app.start()
        print(f"Data Platform MCP server running at http://{host}:{port}")
    
    async def stop(self):
        """停止服务器"""
        await self.app.stop()
        print("Data Platform MCP server stopped")

# 启动服务器的入口点
async def main():
    server = DataPlatformServer()
    try:
        await server.start()
        # 保持服务器运行
        while True:
            await asyncio.sleep(1)
    except KeyboardInterrupt:
        await server.stop()

if __name__ == "__main__":
    asyncio.run(main())

# client/cli.py
import asyncio
import argparse
import json
import pandas as pd
from typing import Dict, List, Any, Optional
from mcp.client import Client
from tabulate import tabulate
import matplotlib.pyplot as plt
import io
import base64
from PIL import Image

class DataPlatformCLI:
    """数据平台命令行客户端"""
    
    def __init__(self, server_url: str = "http://localhost:8000"):
        self.server_url = server_url
        self.client = None
    
    async def connect(self):
        """连接到服务器"""
        self.client = Client()
        await self.client.connect(self.server_url)
        print(f"Connected to server at {self.server_url}")
    
    async def disconnect(self):
        """断开连接"""
        if self.client:
            await self.client.disconnect()
            print("Disconnected from server")
    
    async def list_datasets(self):
        """列出所有数据集"""
        response = await self.client.get_resource("/datasets")
        datasets = response.content
        
        if not datasets:
            print("No datasets found")
            return
        
        # 打印数据集表格
        table_data = []
        for ds in datasets:
            table_data.append([
                ds["id"],
                ds["name"],
                ds.get("rows", "N/A"),
                len(ds.get("columns", [])),
                ds.get("created_at", "N/A")
            ])
        
        headers = ["ID", "Name", "Rows", "Columns", "Created At"]
        print(tabulate(table_data, headers=headers, tablefmt="grid"))
    
    async def import_csv(self, name: str, file_path: str, **kwargs):
        """导入CSV文件"""
        options = {}
        for k, v in kwargs.items():
            options[k] = v
            
        result = await self.client.call_tool("import_csv_data", name=name, file_path=file_path, **options)
        
        if result.get("status") == "success":
            print(f"Successfully imported dataset '{name}' with ID: {result.get('dataset_id')}")
        else:
            print(f"Error importing dataset: {result.get('message')}")
    
    async def preview_dataset(self, dataset_id: str, rows: int = 10):
        """预览数据集"""
        response = await self.client.get_resource(f"/datasets/{dataset_id}/preview")
        
        if not response.content:
            print(f"Dataset {dataset_id} not found or is empty")
            return
        
        # 转换为DataFrame并显示
        df = pd.DataFrame(response.content)
        print(f"Dataset {dataset_id} preview:")
        print(tabulate(df.head(rows), headers=df.columns, tablefmt="grid"))
    
    async def analyze_dataset(self, dataset_id: str, columns: Optional[List[str]] = None):
        """分析数据集"""
        result = await self.client.call_tool("statistical_analysis", dataset_id=dataset_id, columns=columns)
        
        if result.get("status") == "error":
            print(f"Error: {result.get('message')}")
            return
            
        stats = result.get("data", {})
        
        # 打印统计信息
        for column, values in stats.items():
            print(f"\n=== {column} ===")
            for stat, value in values.items():
                print(f"{stat}: {value}")
    
    async def create_chart(self, dataset_id: str, chart_type: str, **kwargs):
        """创建图表"""
        result = await self.client.call_tool(
            "generate_chart", 
            dataset_id=dataset_id, 
            chart_type=chart_type,
            **kwargs
        )
        
        if result.get("status") == "error":
            print(f"Error: {result.get('message')}")
            return
        
        # 显示图表
        if result.get("interactive"):
            print("Interactive chart data (use with web client):")
            print(json.dumps(result.get("data"), indent=2)[:200] + "...")
        else:
            # 显示静态图表
            base64_data = result.get("data")
            img_data = base64.b64decode(base64_data)
            img = Image.open(io.BytesIO(img_data))
            img.show()
            print(f"Created {chart_type} chart for dataset {dataset_id}")
    
    async def query_data(self, dataset_id: str, conditions: List[Dict[str, Any]], **kwargs):
        """查询数据"""
        result = await self.client.call_tool(
            "query_dataset", 
            dataset_id=dataset_id, 
            conditions=conditions,
            **kwargs
        )
        
        if result.get("status") == "error":
            print(f"Error: {result.get('message')}")
            return
        
        data = result.get("data", [])
        if not data:
            print("No data found matching the query")
            return
        
        # 显示结果
        df = pd.DataFrame(data)
        print(f"Query results ({len(data)} rows):")
        print(tabulate(df.head(20), headers=df.columns, tablefmt="grid"))
        
        if len(data) > 20:
            print(f"... showing only 20 of {len(data)} rows")

# 命令行接口
async def main():
    parser = argparse.ArgumentParser(description="Data Platform CLI")
    parser.add_argument("--server", default="http://localhost:8000", help="MCP server URL")
    
    subparsers = parser.add_subparsers(dest="command", help="Command")
    
    # 列出数据集命令
    list_parser = subparsers.add_parser("list", help="List all datasets")
    
    # 导入CSV命令
    import_parser = subparsers.add_parser("import-csv", help="Import CSV file")
    import_parser.add_argument("name", help="Dataset name")
    import_parser.add_argument("file", help="CSV file path")
    
    # 预览数据集命令
    preview_parser = subparsers.add_parser("preview", help="Preview dataset")
    preview_parser.add_argument("id", help="Dataset ID")
    preview_parser.add_argument("--rows", type=int, default=10, help="Number of rows to show")
    
    # 分析数据集命令
    analyze_parser = subparsers.add_parser("analyze", help="Analyze dataset")
    analyze_parser.add_argument("id", help="Dataset ID")
    analyze_parser.add_argument("--columns", nargs="+", help="Columns to analyze")
    
    # 创建图表命令
    chart_parser = subparsers.add_parser("chart", help="Create chart")
    chart_parser.add_argument("id", help="Dataset ID")
    chart_parser.add_argument("type", choices=["bar", "line", "scatter", "histogram", "box", "heatmap"], help="Chart type")
    chart_parser.add_argument("--x", help="X-axis column")
    chart_parser.add_argument("--y", help="Y-axis column")
    chart_parser.add_argument("--color", help="Color column")
    chart_parser.add_argument("--title", help="Chart title")
    chart_parser.add_argument("--static", action="store_true", help="Generate static chart")
    
    args = parser.parse_args()
    
    # 创建并连接客户端
    cli = DataPlatformCLI(args.server)
    
    try:
        await cli.connect()
        
        if args.command == "list":
            await cli.list_datasets()
        elif args.command == "import-csv":
            await cli.import_csv(args.name, args.file)
        elif args.command == "preview":
            await cli.preview_dataset(args.id, args.rows)
        elif args.command == "analyze":
            await cli.analyze_dataset(args.id, args.columns)
        elif args.command == "chart":
            kwargs = {
                "x_column": args.x,
                "y_column": args.y,
                "color_column": args.color,
                "title": args.title,
                "interactive": not args.static
            }
            await cli.create_chart(args.id, args.type, **kwargs)
        else:
            print("Please specify a command. Use --help for more information.")
    
    finally:
        await cli.disconnect()

if __name__ == "__main__":
    asyncio.run(main())

# client/app.py
import asyncio
import uvicorn
from fastapi import FastAPI, Request, Form, UploadFile, File
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
import json
import os
import pandas as pd
from mcp.client import Client
from typing import Dict, List, Any, Optional

app = FastAPI(title="Data Platform Web Client")

# 配置静态文件和模板
app.mount("/static", StaticFiles(directory="client/static"), name="static")
templates = Jinja2Templates(directory="client/templates")

# 全局MCP客户端
mcp_client = None

async def get_mcp_client():
    """获取或创建MCP客户端"""
    global mcp_client
    if mcp_client is None or not mcp_client.is_connected():
        mcp_client = Client()
        await mcp_client.connect("http://localhost:8000")
    return mcp_client

@app.on_event("startup")
async def startup_event():
    """应用启动事件"""
    # 确保数据目录存在
    os.makedirs("client/uploads", exist_ok=True)

@app.on_event("shutdown")
async def shutdown_event():
    """应用关闭事件"""
    global mcp_client
    if mcp_client and mcp_client.is_connected():
        await mcp_client.disconnect()

@app.get("/", response_class=HTMLResponse)
async def index(request: Request):
    """首页"""
    client = await get_mcp_client()
    response = await client.get_resource("/datasets")
    datasets = response.content or []
    
    return templates.TemplateResponse(
        "index.html", 
        {"request": request, "datasets": datasets}
    )

@app.get("/datasets", response_class=JSONResponse)
async def list_datasets():
    """获取数据集列表"""
    client = await get_mcp_client()
    response = await client.get_resource("/datasets")
    return {"datasets": response.content or []}

@app.get("/datasets/{dataset_id}", response_class=HTMLResponse)
async def view_dataset(request: Request, dataset_id: str):
    """查看数据集详情"""
    client = await get_mcp_client()
    
    # 获取数据集元数据
    meta_response = await client.get_resource(f"/datasets/{dataset_id}")
    metadata = meta_response.content
    
    if not metadata:
        return templates.TemplateResponse(
            "error.html", 
            {"request": request, "message": f"Dataset {dataset_id} not found"}
        )
    
    # 获取数据集预览
    preview_response = await client.get_resource(f"/datasets/{dataset_id}/preview")
    preview = preview_response.content or []
    
    # 获取数据集架构
    schema_response = await client.get_resource(f"/datasets/{dataset_id}/schema")
    schema = schema_response.content or []
    
    return templates.TemplateResponse(
        "dataset_detail.html", 
        {
            "request": request, 
            "dataset": metadata,
            "preview": preview,
            "schema": schema
        }
    )

@app.post("/upload-csv", response_class=JSONResponse)
async def upload_csv(name: str = Form(...), file: UploadFile = File(...)):
    """上传CSV文件"""
    # 保存上传的文件
    file_path = f"client/uploads/{file.filename}"
    with open(file_path, "wb") as f:
        f.write(await file.read())
    
    # 导入数据集
    client = await get_mcp_client()
    result = await client.call_tool("import_csv_data", name=name, file_path=file_path)
    
    return result

@app.get("/analyze/{dataset_id}", response_class=HTMLResponse)
async def analyze_form(request: Request, dataset_id: str):
    """分析数据表单"""
    client = await get_mcp_client()
    
    # 获取数据集架构
    schema_response = await client.get_resource(f"/datasets/{dataset_id}/schema")
    schema = schema_response.content or []
    
    return templates.TemplateResponse(
        "analyze.html", 
        {"request": request, "dataset_id": dataset_id, "schema": schema}
    )

@app.post("/analyze/{dataset_id}", response_class=JSONResponse)
async def analyze_dataset(dataset_id: str, columns: List[str] = Form(None)):
    """分析数据集"""
    client = await get_mcp_client()
    result = await client.call_tool("statistical_analysis", dataset_id=dataset_id, columns=columns)
    return result

@app.get("/visualize/{dataset_id}", response_class=HTMLResponse)
async def visualize_form(request: Request, dataset_id: str):
    """可视化表单"""
    client = await get_mcp_client()
    
    # 获取数据集架构
    schema_response = await client.get_resource(f"/datasets/{dataset_id}/schema")
    schema = schema_response.content or []
    
    return templates.TemplateResponse(
        "visualize.html", 
        {"request": request, "dataset_id": dataset_id, "schema": schema}
    )

@app.post("/visualize/{dataset_id}", response_class=JSONResponse)
async def create_visualization(
    dataset_id: str, 
    chart_type: str = Form(...),
    x_column: Optional[str] = Form(None),
    y_column: Optional[str] = Form(None),
    color_column: Optional[str] = Form(None),
    title: Optional[str] = Form(None)
):
    """创建可视化"""
    client = await get_mcp_client()
    result = await client.call_tool(
        "generate_chart", 
        dataset_id=dataset_id, 
        chart_type=chart_type,
        x_column=x_column,
        y_column=y_column,
        color_column=color_column,
        title=title,
        interactive=True
    )
    return result

@app.get("/query/{dataset_id}", response_class=HTMLResponse)
async def query_form(request: Request, dataset_id: str):
    """查询表单"""
    client = await get_mcp_client()
    
    # 获取数据集架构
    schema_response = await client.get_resource(f"/datasets/{dataset_id}/schema")
    schema = schema_response.content or []
    
    return templates.TemplateResponse(
        "query.html", 
        {"request": request, "dataset_id": dataset_id, "schema": schema}
    )

@app.post("/query/{dataset_id}", response_class=JSONResponse)
async def query_dataset(dataset_id: str, query_json: str = Form(...)):
    """查询数据集"""
    query_data = json.loads(query_json)
    
    conditions = query_data.get("conditions", [])
    order_by = query_data.get("order_by", [])
    limit = query_data.get("limit")
    offset = query_data.get("offset", 0)
    columns = query_data.get("columns")
    
    client = await get_mcp_client()
    result = await client.call_tool(
        "query_dataset", 
        dataset_id=dataset_id, 
        conditions=conditions,
        order_by=order_by,
        limit=limit,
        offset=offset,
        columns=columns
    )
    return result

# 启动应用服务器
def start_server():
    """启动Web服务器"""
    uvicorn.run("client.app:app", host="localhost", port=8080, reload=True)

if __name__ == "__main__":
    start_server()

# 从项目根目录启动服务器
python -m server.main

# 列出数据集
python -m client.cli list

# 导入CSV文件
python -m client.cli import-csv "销售数据" data/samples/sales.csv

# 预览数据集
python -m client.cli preview <dataset_id>

# 分析数据集
python -m client.cli analyze <dataset_id> --columns price quantity

# 创建图表
python -m client.cli chart <dataset_id> bar --x "category" --y "sales" --title "销售分类图表"

# 启动Web客户端
python -m client.app