"""Tool metadata definitions for the MCP server. All 30 Tool(...) descriptor objects and the get_tools_meta() function. """ from mcp.types import Tool def get_tools_meta() -> list[Tool]: """Return the list of Tool metadata objects for MCP server registration. Each tool gets a mandatory ``_timeout`` parameter so the LLM must specify how many seconds to allow for each tool invocation (0 = no limit). """ raw_tools = [ Tool( name="load_data", description=( "加载匹配glob模式的CSV文件。自动检测每个文件的BOM，验证schema一致性，" "纵向合并，并将结果作为惰性数据集存储在会话存储中。" ), inputSchema={ "type": "object", "properties": { "csv_glob": { "type": "string", "description": "Glob pattern for CSV files (e.g. 'data/*.csv')", }, "config_path": { "type": "string", "description": "Optional path to a YAML config file", }, "encoding": { "type": "string", "description": "Fallback encoding when no BOM is found", "default": "utf-8", }, "delimiter": { "type": "string", "description": "Column delimiter", "default": ",", }, "schema_strict": { "type": "boolean", "description": "When True, raise on column mismatch; when False, merge leniently with nulls", "default": False, }, "recursive": { "type": "boolean", "description": "When True, use recursive glob for **/*.csv patterns", "default": False, }, }, "required": ["csv_glob"], }, ), Tool( name="profile_data", description=( "计算数据集的每列统计信息和数值相关矩阵。结果自动截断至约16 KB以" "节省LLM上下文窗口空间。" ), inputSchema={ "type": "object", "properties": { "dataset_id": { "type": "string", "description": "Dataset identifier from a previous load_data call", }, "sample_size": { "type": "integer", "description": "Maximum rows to profile (sampled if larger)", "default": 1000, }, "columns": { "type": "array", "items": {"type": "string"}, "description": "Columns to profile (default: all)", }, }, "required": ["dataset_id"], }, ), Tool( name="filter_data", description=( "对数据集应用列级过滤器，将结果存储为新的派生数据集。支持eq/neq/gt/gte/" "lt/lte/contains/not_contains/in/not_in/is_null/is_not_null运算符。" ), inputSchema={ "type": "object", "properties": { "dataset_id": { "type": "string", "description": "Source dataset identifier", }, "filters": { "type": "array", "items": { "type": "object", "properties": { "column": {"type": "string"}, "op": { "type": "string", "enum": [ "eq", "neq", "gt", "gte", "lt", "lte", "contains", "not_contains", "in", "not_in", "is_null", "is_not_null", ], }, "value": {"description": "Operand value"}, }, "required": ["column", "op"], }, "description": "List of filter expressions", }, "logic": { "type": "string", "enum": ["and", "or"], "description": "Combine filters with AND or OR logic", "default": "and", }, }, "required": ["dataset_id", "filters"], }, ), Tool( name="preprocess_data", description=( "预处理选定列：标准化（StandardScaler）、独热编码、缺失值填充和列删除。" ), inputSchema={ "type": "object", "properties": { "dataset_id": { "type": "string", "description": "Source dataset identifier", }, "columns": { "type": "array", "items": {"type": "string"}, "description": "Columns to preprocess", }, "config": { "type": "object", "properties": { "standardize": { "type": "boolean", "description": "Apply StandardScaler to numeric columns", "default": False, }, "onehot": { "type": "boolean", "description": "One-hot encode categorical columns", "default": False, }, "fillna": { "type": "string", "enum": ["mean", "median", "zero", "drop"], "description": "Missing value strategy", }, "drop": { "type": "array", "items": {"type": "string"}, "description": "Columns to drop entirely", }, }, }, }, "required": ["dataset_id", "columns"], }, ), Tool( name="run_clustering", description=( "对选定列执行聚类。使用HDBSCAN（默认）或KMeans。特征在聚类前进行" "标准化缩放。返回聚类标签和质量指标。" ), inputSchema={ "type": "object", "properties": { "dataset_id": { "type": "string", "description": "Dataset identifier", }, "cluster_columns": { "type": "array", "items": {"type": "string"}, "description": "Columns to use as clustering features", }, "algorithm": { "type": "string", "enum": ["agglomerative", "hdbscan", "kmeans"], "description": "Clustering algorithm (default: agglomerative/Ward)", "default": "agglomerative", }, "params": { "type": "object", "description": ( "Algorithm parameters. Agglomerative: n_clusters (5), metric ('euclidean'), " "linkage ('ward'). HDBSCAN: min_cluster_size (5), " "min_samples (None), metric ('euclidean'). " "KMeans: n_clusters (3), random_state (42)." ), }, "random_state": { "type": "integer", "description": "Random seed for reproducibility", "default": 42, }, }, "required": ["dataset_id", "cluster_columns"], }, ), Tool( name="evaluate_clustering", description=( "评估数据集的聚类结果。当前支持轮廓系数（silhouette）、Davies-Bouldin指数、" "Calinski-Harabasz指数、噪声比（noise_ratio）和簇大小（cluster_sizes）指标。" ), inputSchema={ "type": "object", "properties": { "cluster_result_id": { "type": "string", "description": "Result ID from run_clustering", }, "dataset_id": { "type": "string", "description": "Original dataset for metric computation", }, "metrics": { "type": "array", "items": {"type": "string"}, "description": ( "Metrics to compute. Options: silhouette, " "davies_bouldin, calinski_harabasz, noise_ratio, " "cluster_sizes (default: all)" ), }, }, "required": ["cluster_result_id", "dataset_id"], }, ), Tool( name="extract_features", description=( "提取每个聚类的区分特征并自动保存到Django ORM（ClusterFeature模型）。" "使用z-score或ANOVA对每个聚类的特征重要性进行排序。" ), inputSchema={ "type": "object", "properties": { "dataset_id": { "type": "string", "description": "Dataset containing the feature columns", }, "cluster_result_id": { "type": "string", "description": "Result ID from run_clustering", }, "top_k": { "type": "integer", "description": "Top distinguishing features per cluster", "default": 10, }, "method": { "type": "string", "enum": ["zscore", "anova"], "description": "Method to rank feature importance", "default": "zscore", }, "save_to_db": { "type": "boolean", "description": "Save features to Django ORM if True", "default": True, }, }, "required": ["dataset_id", "cluster_result_id"], }, ), Tool( name="export_results", description=( "将数据集或聚类结果以CSV或Parquet格式导出到磁盘。数据集物化后写入；" "聚类结果序列化为JSON。" ), inputSchema={ "type": "object", "properties": { "result_id": { "type": "string", "description": "Dataset ID or cluster result ID to export", }, "output_path": { "type": "string", "description": "Output file path or directory", }, "format": { "type": "string", "enum": ["csv", "parquet", "json"], "description": "Output format", "default": "csv", }, "overwrite": { "type": "boolean", "description": "Overwrite existing file", "default": False, }, }, "required": ["result_id", "output_path"], }, ), Tool( name="list_datasets", description="列出会话存储中的所有活跃数据集和结果。", inputSchema={ "type": "object", "properties": {}, }, ), Tool( name="drop_dataset", description=( "从会话存储中移除数据集（或结果）并通过垃圾回收释放内存。" ), inputSchema={ "type": "object", "properties": { "dataset_id": { "type": "string", "description": "Dataset or result identifier to remove", }, }, "required": ["dataset_id"], }, ), Tool( name="clone_dataset", description=( "浅拷贝数据集。由于LazyFrame是查询计划，不会复制实际数据。适用于分支分析。" ), inputSchema={ "type": "object", "properties": { "dataset_id": { "type": "string", "description": "Source dataset identifier", }, }, "required": ["dataset_id"], }, ), Tool( name="filter_and_cluster", description=( "一步完成：先应用列级过滤器，再对过滤结果执行聚类。支持eq/neq/gt/gte/lt/lte/" "contains/not_contains/in/not_in/is_null/is_not_null运算符和HDBSCAN/KMeans聚类。" "自动选择数值列作为聚类特征。将过滤后的数据集和聚类结果都存储在会话中。" ), inputSchema={ "type": "object", "properties": { "dataset_id": { "type": "string", "description": "Source dataset identifier", }, "filters": { "type": "array", "items": { "type": "object", "properties": { "column": {"type": "string"}, "op": { "type": "string", "enum": [ "eq", "neq", "gt", "gte", "lt", "lte", "contains", "not_contains", "in", "not_in", "is_null", "is_not_null", ], }, "value": {"description": "Operand value"}, }, "required": ["column", "op"], }, "description": "List of filter expressions", }, "logic": { "type": "string", "enum": ["and", "or"], "description": "Combine filters with AND or OR logic", "default": "and", }, "algorithm": { "type": "string", "enum": ["agglomerative", "hdbscan", "kmeans"], "description": "Clustering algorithm (default: agglomerative/Ward)", "default": "agglomerative", }, "params": { "type": "object", "description": ( "Algorithm parameters. Agglomerative: n_clusters (5), metric ('euclidean'), " "linkage ('ward'). HDBSCAN: min_cluster_size (5), " "min_samples (None), metric ('euclidean'). " "KMeans: n_clusters (3), random_state (42)." ), }, }, "required": ["dataset_id", "filters"], }, ), Tool( name="build_entity_profiles", description=( "按实体列（源IP、SNI等）分组原始流数据。自动检测实体列（如 src_ip, " "dst_ip, sni, host, mac, domain），按列分组并计算每个实体的聚合特征：" "流数量、唯一端口数、TLS版本分布、字节总量、唯一目标数。" "生成的数据集每个实体一行。结果存储在会话中，可供后续 compute_scores 使用。" ), inputSchema={ "type": "object", "properties": { "dataset_id": { "type": "string", "description": "Dataset ID from load_data or profile_data", }, "entity_column": { "type": "string", "description": "Specific entity column to group by (e.g. 'src_ip')", }, "entity_columns": { "type": "array", "items": {"type": "string"}, "description": "Multiple entity columns to group by", }, "auto_detect": { "type": "boolean", "description": "Auto-detect entity columns (IP, SNI, host, MAC, domain)", "default": True, }, }, "required": ["dataset_id"], }, ), Tool( name="compute_scores", description=( "在实体数据上计算自适应代理/异常/威胁评分。从数据自动学习" "特征权重，计算加权代理评分（proxy_score）、基于IQR的风险等级" "（normal/watch/suspicious/critical）和威胁评分（threat_score）。" "返回评分摘要与阈值。" ), inputSchema={ "type": "object", "properties": { "dataset_id": { "type": "string", "description": "Entity dataset ID from filter_and_cluster or a scored dataset", }, }, "required": ["dataset_id"], }, ), Tool( name="detect_anomalies", description=( "对已评分的实体数据运行孤立森林（Isolation Forest）异常检测。识别偏离" "正常模式的实体。分块处理以应对大数据集。返回异常实体ID列表。" ), inputSchema={ "type": "object", "properties": { "dataset_id": { "type": "string", "description": "Scored entity dataset ID (after compute_scores)", }, "contamination": { "type": "number", "description": "Expected proportion of anomalies (0.0-0.5)", "default": 0.05, }, }, "required": ["dataset_id"], }, ), Tool( name="visualize_anomalies", description=( "为高风险实体生成UMAP-2D嵌入和聚类可视化数据。按风险等级和聚类着色。" "返回用于前端渲染的散点图JSON数据。" ), inputSchema={ "type": "object", "properties": { "dataset_id": { "type": "string", "description": "Scored entity dataset ID (after compute_scores)", }, }, "required": ["dataset_id"], }, ), # ── Edge-case / Diagnostic tools (unexpected situations) ──────── Tool( name="validate_data", description=( "诊断工具：检查数据质量问题——文件间schema一致性、缺失值比例、列类型冲突、" "编码问题和空列。当流水线步骤因schema错误失败，或怀疑CSV数据损坏/格式异常时使用。" ), inputSchema={ "type": "object", "properties": { "dataset_id": {"type": "string", "description": "Dataset ID from load_data"}, "fix_issues": {"type": "boolean", "description": "Auto-fix schema inconsistencies", "default": False}, }, "required": ["dataset_id"], }, ), Tool( name="explore_distributions", description=( "诊断工具：计算每列分布统计（唯一值计数、空值比例、最小/最大/均值/标准差、" "前10高频值、直方图分箱）。当聚类产生意外结果、实体检测失败，或需要在选择" "分析策略前了解数据时使用。" ), inputSchema={ "type": "object", "properties": { "dataset_id": {"type": "string", "description": "Dataset ID from load_data or entity data"}, "columns": {"type": "array", "items": {"type": "string"}, "description": "Specific columns to analyze (default: auto-detect numeric)"}, "max_sample": {"type": "integer", "description": "Max rows to sample", "default": 10000}, }, "required": ["dataset_id"], }, ), Tool( name="find_outliers", description=( "诊断工具：使用IQR方法检测数值列中的统计异常值。返回任何数值列超过中位数3倍IQR的行。" "当怀疑数据质量问题或极端值偏离分析时使用。" ), inputSchema={ "type": "object", "properties": { "dataset_id": {"type": "string", "description": "Dataset ID from load_data"}, "columns": {"type": "array", "items": {"type": "string"}, "description": "Columns to check (default: all numeric)"}, "threshold": {"type": "number", "description": "IQR multiplier (default 3.0)", "default": 3.0}, }, "required": ["dataset_id"], }, ), Tool( name="diagnose_clustering", description=( "诊断工具：当聚类产生太少/太多聚类、全部为噪声或轮廓系数低时——诊断原因。" "检查：特征方差、相关矩阵、特征降维分析、数据稀疏性，并建议参数调整" "（min_cluster_size、算法选择、特征选择）。" ), inputSchema={ "type": "object", "properties": { "dataset_id": {"type": "string", "description": "Entity dataset ID"}, "cluster_result_id": {"type": "string", "description": "Cluster result ID from previous run_clustering"}, }, "required": ["dataset_id", "cluster_result_id"], }, ), Tool( name="compare_datasets", description=( "诊断工具：比较两个数据集（如处理前/后，或两次不同运行）。报告schema差异、" "行数变化、列值分布偏移。用于验证处理步骤未损坏数据或比较不同运行的结果。" ), inputSchema={ "type": "object", "properties": { "dataset_id_a": {"type": "string", "description": "First dataset ID"}, "dataset_id_b": {"type": "string", "description": "Second dataset ID"}, }, "required": ["dataset_id_a", "dataset_id_b"], }, ), Tool( name="export_debug_sample", description=( "诊断工具：导出原始数据样本为JSON用于外部调试。当流水线深处发生错误且需要" "检查实际数据值时使用。返回前N行所有列。" ), inputSchema={ "type": "object", "properties": { "dataset_id": {"type": "string", "description": "Dataset ID"}, "max_rows": {"type": "integer", "description": "Rows to export", "default": 50}, }, "required": ["dataset_id"], }, ), Tool( name="repair_schema", description=( "诊断工具：通过对齐列名（不区分大小写合并、下划线/连字符规范化）并用空值填充" "缺失列，尝试修复文件间的schema不匹配。当load_data因CSV文件具有不同列集而" "失败时使用。" ), inputSchema={ "type": "object", "properties": { "dataset_id": {"type": "string", "description": "Failed dataset ID from load_data"}, }, "required": ["dataset_id"], }, ), # ── Read-only analysis tools (LLM deep-dive, no side effects) ──── Tool( name="analyze_patterns", description=( "分析工具：检测流量模式——top源/目标IP、端口分布、TLS版本分布、协议混合。" "返回汇总统计。只读，不修改数据。用于了解高层流量结构。" ), inputSchema={ "type": "object", "properties": { "dataset_id": {"type": "string", "description": "Dataset ID from load_data"}, "top_n": {"type": "integer", "description": "Number of top items per category", "default": 10}, }, "required": ["dataset_id"], }, ), Tool( name="analyze_temporal", description=( "分析工具：分析随时间变化的流量——小时/日流量计数、繁忙时段、空闲时段。" "需要时间戳列。只读。用于了解时间相关模式和周期性。" ), inputSchema={ "type": "object", "properties": { "dataset_id": {"type": "string", "description": "Dataset ID from load_data"}, "timestamp_column": {"type": "string", "description": "Column name for timestamps (auto-detected if omitted)"}, }, "required": ["dataset_id"], }, ), Tool( name="analyze_fft", description=( "FFT频谱分析——从时间戳列提取周期模式和频率特征。适用于检测时间周期性" "（如每日/每周流量模式）。返回频率、幅度、top周期、频谱质心、周期性评分。" "只读，无副作用。" ), inputSchema={ "type": "object", "properties": { "dataset_id": { "type": "string", "description": "Dataset ID from load_data", }, "timestamp_column": { "type": "string", "description": "Column name containing timestamps or numeric time values", }, "top_n": { "type": "integer", "description": "Number of top frequency peaks to return", "default": 5, }, }, "required": ["dataset_id", "timestamp_column"], }, ), Tool( name="analyze_tls_health", description=( "分析工具：TLS安全态势评估——检查过时的TLS版本（1.0/1.1）、弱加密套件、" "证书问题、SNI异常。只读。用于评估流量的TLS安全性。" ), inputSchema={ "type": "object", "properties": { "dataset_id": {"type": "string", "description": "Dataset ID from load_data"}, }, "required": ["dataset_id"], }, ), Tool( name="analyze_geo_distribution", description=( "分析工具：流量地理分布——top源/目标国家、异常地点对（如不可能的行程）、" "ISP多样性。只读。需要国家或经纬度列。用于检测地理异常。" ), inputSchema={ "type": "object", "properties": { "dataset_id": {"type": "string", "description": "Dataset ID from load_data"}, }, "required": ["dataset_id"], }, ), Tool( name="analyze_entity_detail", description=( "分析工具：深入分析单个实体——其所有流、使用的TLS版本、目标多样性、时间模式、" "异常评分。只读。在compute_scores后调查特定可疑实体时使用。" ), inputSchema={ "type": "object", "properties": { "dataset_id": {"type": "string", "description": "Entity dataset ID (after filter_and_cluster or compute_scores)"}, "entity_value": {"type": "string", "description": "Entity value to investigate (e.g. an IP address or SNI)"}, }, "required": ["dataset_id", "entity_value"], }, ), Tool( name="compute_distance_matrix", description=( "LLM驱动工具：对数据集的每一行执行用户编写的Python函数，返回距离/相似度评分摘要。" "LLM提供函数体，签名格式为`def distance_fn(row: dict) -> float:`。执行器将其包装，" "在受限命名空间（仅math、numpy）中逐行运行，返回得分的最小/最大/均值/标准差以及" "评分行样本。用于现有工具无法表达的一次性度量、过滤或评分任务。" ), inputSchema={ "type": "object", "properties": { "dataset_id": { "type": "string", "description": "Source dataset identifier", }, "python_function": { "type": "string", "description": ( "Body of a Python function with signature `def distance_fn(row: dict) -> float:`. " "Example: \"return abs(row.get('col1', 0)) + row.get('col2', 0)\". " "Available namespace: math, numpy (as np). No dangerous modules." ), }, "columns": { "type": "array", "items": {"type": "string"}, "description": "Columns to pass to the function (default: auto-detect numeric columns)", }, "max_sample": { "type": "integer", "description": "Maximum rows to evaluate (default 10000 for performance)", "default": 10000, }, }, "required": ["dataset_id", "python_function"], }, ), ] # end of raw_tools # Inject mandatory _timeout parameter into every tool result = [] for tool in raw_tools: schema = dict(tool.inputSchema) props = dict(schema.get("properties", {})) req = list(schema.get("required", [])) # Add _timeout as the first required property props["_timeout"] = { "type": "integer", "description": "工具调用超时秒数（必填，LLM决定每次调用的最长等待时间；0表示不限制）", "default": 120, } req.insert(0, "_timeout") schema["properties"] = props schema["required"] = req result.append(Tool( name=tool.name, description=tool.description, inputSchema=schema, )) return result