AI_DATA/doris-mcp-server
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.4.0 preview

Browse Source
This commit is contained in:
FreeOnePlus
2025-06-12 19:36:16 +08:00
parent 609816bc4a
commit 1e2e79d90d
15 changed files with 3253 additions and 428 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
.env.example
View File

@@ -1,71 +1,88 @@
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
# Doris MCP Server Environment Configuration
# Copy this file to .env and modify the values as needed
# Doris MCP Server Configuration
# Copy this file to .env and modify the values according to your environment
# =============================================================================
# Database Configuration
# =============================================================================
# Doris FE connection settings
DORIS_HOST=localhost
DORIS_PORT=9030
DORIS_USER=root
DORIS_PASSWORD=your_password_here
DORIS_DATABASE=your_database_name
DORIS_PASSWORD=
DORIS_DATABASE=information_schema
# Doris FE HTTP API port
DORIS_FE_HTTP_PORT=8030
# BE nodes configuration for external access
# If DORIS_BE_HOSTS is empty, will use "show backends" to get BE nodes automatically
# Format: comma-separated list of BE host addresses
# Example: DORIS_BE_HOSTS=192.168.1.100,192.168.1.101,192.168.1.102
DORIS_BE_HOSTS=
# BE webserver port for HTTP APIs (memory tracker, metrics, etc.)
DORIS_BE_WEBSERVER_PORT=8040
# =============================================================================
# Connection Pool Configuration
# =============================================================================
# Connection Pool Settings
DORIS_MIN_CONNECTIONS=5
DORIS_MAX_CONNECTIONS=20
DORIS_CONNECTION_TIMEOUT=30
DORIS_HEALTH_CHECK_INTERVAL=60
DORIS_MAX_CONNECTION_AGE=3600
# Security Settings
# =============================================================================
# Profile And Explain Max Data Size
# =============================================================================
MAX_RESPONSE_CONTENT_SIZE=4096
# =============================================================================
# Security Configuration
# =============================================================================
AUTH_TYPE=token
TOKEN_SECRET=your_256_bit_secret_key_here
TOKEN_SECRET=your_secret_key_here
TOKEN_EXPIRY=3600
MAX_RESULT_ROWS=10000
MAX_QUERY_COMPLEXITY=100
ENABLE_MASKING=true
# Performance Settings
# =============================================================================
# Performance Configuration
# =============================================================================
ENABLE_QUERY_CACHE=true
CACHE_TTL=300
MAX_CACHE_SIZE=1000
MAX_CONCURRENT_QUERIES=50
QUERY_TIMEOUT=300
# =============================================================================
# Logging Configuration
LOG_LEVEL=INFO
LOG_FILE_PATH=./log/doris-mcp-server.log
ENABLE_AUDIT=true
AUDIT_FILE_PATH=./log/doris-mcp-audit.log
# =============================================================================
LOG_LEVEL=INFO
LOG_FILE_PATH=
ENABLE_AUDIT=true
AUDIT_FILE_PATH=
# =============================================================================
# Monitoring Configuration
# =============================================================================
# Monitoring Settings
ENABLE_METRICS=true
METRICS_PORT=3001
METRICS_PATH=/metrics
HEALTH_CHECK_PORT=3002
HEALTH_CHECK_PATH=/health
ENABLE_ALERTS=false
ALERT_WEBHOOK_URL=
# Server Settings
# =============================================================================
# Server Configuration
# =============================================================================
SERVER_NAME=doris-mcp-server
SERVER_VERSION=0.3.0
SERVER_PORT=3000
# Development Settings (for development environment only)
DEBUG=false
VERBOSE=false

157
README.md
View File

@@ -21,36 +21,34 @@ under the License.
Doris MCP (Model Context Protocol) Server is a backend service built with Python and FastAPI. It implements the MCP, allowing clients to interact with it through defined "Tools". It's primarily designed to connect to Apache Doris databases, potentially leveraging Large Language Models (LLMs) for tasks like converting natural language queries to SQL (NL2SQL), executing queries, and performing metadata management and analysis.
## 🚀 What's New in v0.3.0
## 🚀 What's New in v0.4.0
- **🔄 Streamlined Communication**: Completely migrated from SSE to Streamable HTTP for better performance and reliability
- **🏗️ Unified Architecture**: Consolidated tools management with centralized registration and routing
- **⚡ Enhanced Performance**: Improved query execution with advanced caching and optimization
- **🔒 Enterprise Security**: Added comprehensive security management with SQL validation and data masking
- **📊 Advanced Analytics**: New column analysis and performance monitoring tools
- **🛠️ Simplified Development**: Streamlined tool development process with unified interfaces
- **🛠️ Enhanced Monitoring Tools Module**: Advanced memory tracking, metrics collection, and BE node discovery with modular, extensible design
- **🔍 Mature Query Information Tools**: Enhanced SQL explain and profiling with configurable content truncation, file export for LLM attachments, and advanced query analytics
- **⚙️ Unified Configuration Framework**: Centralized configuration management through `config.py` with comprehensive validation and standardized parameter naming
- **🗃️ Smart Default Database Handling**: Automatic fallback to `information_schema` database, eliminating startup configuration requirements and improving reliability
- **🗑️ Production-Ready Architecture**: Removed experimental tools in favor of battle-tested, production-ready alternatives with better performance and monitoring
> **⚠️ Breaking Changes**: SSE endpoints have been removed. Please update your client configurations to use Streamable HTTP (`/mcp` endpoint).
> **⚠️ Breaking Changes**: Experimental tools (`column_analysis`, `performance_stats`) have been removed and replaced with enhanced monitoring and query information tools.
## Core Features
* **MCP Protocol Implementation**: Provides standard MCP interfaces, supporting tool calls, resource management, and prompt interactions.
* **Multiple Communication Modes** (Updated in v0.3.0):
* **Stdio**: Standard input/output mode for direct integration with MCP clients like Cursor.
* **Streamable HTTP**: Unified HTTP endpoint supporting request/response and streaming (Primary mode since v0.3.0).
> **⚠️ Breaking Change in v0.3.0**: SSE (Server-Sent Events) mode has been completely removed in favor of the more robust Streamable HTTP implementation.
* **Streamable HTTP Communication**: Unified HTTP endpoint supporting both request/response and streaming communication for optimal performance and reliability.
* **Stdio Communication**: Standard input/output mode for direct integration with MCP clients like Cursor.
* **Enterprise-Grade Architecture**: Modular design with comprehensive functionality:
* **Tools Manager**: Centralized tool registration and routing (`doris_mcp_server/tools/tools_manager.py`)
* **Tools Manager**: Centralized tool registration and routing with unified interfaces (`doris_mcp_server/tools/tools_manager.py`)
* **Enhanced Monitoring Tools Module**: Advanced memory tracking, metrics collection, and flexible BE node discovery (Enhanced in v0.4.0)
* **Query Information Tools**: Enhanced SQL explain and profiling with configurable content management and LLM file export capabilities (Enhanced in v0.4.0)
* **Resources Manager**: Resource management and metadata exposure (`doris_mcp_server/tools/resources_manager.py`)
* **Prompts Manager**: Intelligent prompt templates for data analysis (`doris_mcp_server/tools/prompts_manager.py`)
* **Advanced Database Features**:
* **Query Execution**: High-performance SQL execution with caching and optimization (`doris_mcp_server/utils/query_executor.py`)
* **Security Management**: SQL security validation, data masking, and access control (`doris_mcp_server/utils/security.py`)
* **Query Execution**: High-performance SQL execution with advanced caching and optimization (`doris_mcp_server/utils/query_executor.py`)
* **Security Management**: Comprehensive SQL security validation, data masking, and access control (`doris_mcp_server/utils/security.py`)
* **Metadata Extraction**: Comprehensive database metadata with catalog federation support (`doris_mcp_server/utils/schema_extractor.py`)
* **Performance Analysis**: Column statistics, performance monitoring, and data analysis tools (`doris_mcp_server/utils/analysis_tools.py`)
* **Performance Analysis**: Advanced column analysis, performance monitoring, and data analysis tools (`doris_mcp_server/utils/analysis_tools.py`)
* **Catalog Federation Support**: Full support for multi-catalog environments (internal Doris tables and external data sources like Hive, MySQL, etc.)
* **Enterprise Security**: Comprehensive security framework with authentication, authorization, SQL injection protection, and data masking (`doris_mcp_server/utils/security.py`)
* **Enterprise Security**: Comprehensive security framework with authentication, authorization, SQL injection protection, and data masking capabilities (`doris_mcp_server/utils/security.py`)
* **Flexible Configuration**: Comprehensive configuration management with environment variables, file-based config, and validation (`doris_mcp_server/utils/config.py`)
## System Requirements
@@ -67,13 +65,15 @@ Doris MCP (Model Context Protocol) Server is a backend service built with Python
pip install mcp-doris-server
# Install specific version
pip install mcp-doris-server==0.3
pip install mcp-doris-server==0.4.0
```
> **💡 Command Compatibility**: After installation, both `doris-mcp-server` and `mcp-doris-server` commands are available for backward compatibility. You can use either command interchangeably.
### Start Streamable HTTP Mode (Web Service)
The primary communication mode offering optimal performance and reliability:
```bash
# Full configuration with database connection
doris-mcp-server \
@@ -88,6 +88,8 @@ doris-mcp-server \
### Start Stdio Mode (for Cursor and other MCP clients)
Standard input/output mode for direct integration with MCP clients:
```bash
# For direct integration with MCP clients like Cursor
doris-mcp-server --transport stdio
@@ -164,9 +166,11 @@ cp .env.example .env
* `DORIS_PORT`: Database port (default: 9030)
* `DORIS_USER`: Database username (default: root)
* `DORIS_PASSWORD`: Database password
* `DORIS_DATABASE`: Default database name (default: test)
* `DORIS_DATABASE`: Default database name (default: information_schema)
* `DORIS_MIN_CONNECTIONS`: Minimum connection pool size (default: 5)
* `DORIS_MAX_CONNECTIONS`: Maximum connection pool size (default: 20)
* `DORIS_BE_HOSTS`: BE nodes for monitoring (comma-separated, optional - auto-discovery via SHOW BACKENDS if empty)
* `DORIS_BE_WEBSERVER_PORT`: BE webserver port for monitoring tools (default: 8040)
* **Security Configuration**:
* `AUTH_TYPE`: Authentication type (token/basic/oauth, default: token)
* `TOKEN_SECRET`: Token secret key
@@ -176,6 +180,7 @@ cp .env.example .env
* `ENABLE_QUERY_CACHE`: Enable query caching (default: true)
* `CACHE_TTL`: Cache time-to-live in seconds (default: 300)
* `MAX_CONCURRENT_QUERIES`: Maximum concurrent queries (default: 50)
* `MAX_RESPONSE_CONTENT_SIZE`: Maximum response content size for LLM compatibility (default: 4096, New in v0.4.0)
* **Logging Configuration**:
* `LOG_LEVEL`: Log level (DEBUG/INFO/WARNING/ERROR, default: INFO)
* `LOG_FILE_PATH`: Log file path
@@ -185,21 +190,26 @@ cp .env.example .env
The following table lists the main tools currently available for invocation via an MCP client:
| Tool Name | Description | Parameters | Status |
|:----------------------------| :---------------------------------------------------------- | :--------------------------------------------------------------------------------------------------------- | :------- |
| `exec_query` | Execute SQL query with catalog federation support. | `sql` (string, Required - MUST use three-part naming), `db_name` (string, Optional), `catalog_name` (string, Optional), `max_rows` (integer, Optional, default 100), `timeout` (integer, Optional, default 30) | ✅ Active |
| `get_catalog_list` | Get a list of all catalogs with detailed information. | `random_string` (string, Required) | ✅ Active |
| `get_db_list` | Get a list of all database names in the specified catalog. | `catalog_name` (string, Optional, defaults to internal catalog) | ✅ Active |
| `get_db_table_list` | Get a list of all table names in the specified database. | `db_name` (string, Optional), `catalog_name` (string, Optional) | ✅ Active |
| `get_table_schema` | Get detailed structure of the specified table. | `table_name` (string, Required), `db_name` (string, Optional), `catalog_name` (string, Optional) | ✅ Active |
| `get_table_comment` | Get the comment for the specified table. | `table_name` (string, Required), `db_name` (string, Optional), `catalog_name` (string, Optional) | ✅ Active |
| `get_table_column_comments` | Get comments for all columns in the specified table. | `table_name` (string, Required), `db_name` (string, Optional), `catalog_name` (string, Optional) | ✅ Active |
| `get_table_indexes` | Get index information for the specified table. | `table_name` (string, Required), `db_name` (string, Optional), `catalog_name` (string, Optional) | ✅ Active |
| `get_recent_audit_logs` | Get audit log records for a recent period. | `days` (integer, Optional, default 7), `limit` (integer, Optional, default 100) | ✅ Active |
| `column_analysis` | Analyze statistical information and data distribution. | `table_name` (string, Required), `column_name` (string, Required), `analysis_type` (string, Optional: basic/distribution/detailed) | ⚠️ Experimental |
| `performance_stats` | Get database performance statistics information. | `metric_type` (string, Optional: queries/connections/tables/system), `time_range` (string, Optional: 1h/6h/24h/7d) | ⚠️ Experimental |
| Tool Name | Description | Parameters |
|-----------------------------|--------------------------------------------------------------|--------------------------------------------------------------|
| `exec_query` | Execute SQL query and return results. | `sql` (string, Required), `db_name` (string, Optional), `catalog_name` (string, Optional), `max_rows` (integer, Optional), `timeout` (integer, Optional) |
| `get_table_schema` | Get detailed table structure information. | `table_name` (string, Required), `db_name` (string, Optional), `catalog_name` (string, Optional) |
| `get_db_table_list` | Get list of all table names in specified database. | `db_name` (string, Optional), `catalog_name` (string, Optional) |
| `get_db_list` | Get list of all database names. | `catalog_name` (string, Optional) |
| `get_table_comment` | Get table comment information. | `table_name` (string, Required), `db_name` (string, Optional), `catalog_name` (string, Optional) |
| `get_table_column_comments` | Get comment information for all columns in table. | `table_name` (string, Required), `db_name` (string, Optional), `catalog_name` (string, Optional) |
| `get_table_indexes` | Get index information for specified table. | `table_name` (string, Required), `db_name` (string, Optional), `catalog_name` (string, Optional) |
| `get_recent_audit_logs` | Get audit log records for recent period. | `days` (integer, Optional), `limit` (integer, Optional) |
| `get_catalog_list` | Get list of all catalog names. | `random_string` (string, Required) |
| `get_sql_explain` | Get SQL execution plan with configurable content truncation and file export for LLM analysis. | `sql` (string, Required), `verbose` (boolean, Optional), `db_name` (string, Optional), `catalog_name` (string, Optional) |
| `get_sql_profile` | Get SQL execution profile with content management and file export for LLM optimization workflows. | `sql` (string, Required), `db_name` (string, Optional), `catalog_name` (string, Optional), `timeout` (integer, Optional) |
| `get_table_data_size` | Get table data size information via FE HTTP API. | `db_name` (string, Optional), `table_name` (string, Optional), `single_replica` (boolean, Optional) |
| `get_monitoring_metrics_info` | Get Doris monitoring metrics definitions and descriptions. | `role` (string, Optional), `monitor_type` (string, Optional), `priority` (string, Optional) |
| `get_monitoring_metrics_data` | Get actual Doris monitoring metrics data from nodes with flexible BE discovery. | `role` (string, Optional), `monitor_type` (string, Optional), `priority` (string, Optional) |
| `get_realtime_memory_stats` | Get real-time memory statistics via BE Memory Tracker with auto/manual BE discovery. | `tracker_type` (string, Optional), `include_details` (boolean, Optional) |
| `get_historical_memory_stats` | Get historical memory statistics via BE Bvar interface with flexible BE configuration. | `tracker_names` (array, Optional), `time_range` (string, Optional) |
**Note:** All metadata tools support catalog federation for multi-catalog environments. The `get_catalog_list` tool requires a `random_string` parameter for compatibility reasons.
**Note:** All metadata tools support catalog federation for multi-catalog environments. The `get_catalog_list` tool requires a `random_string` parameter for compatibility reasons. Enhanced monitoring tools in v0.4.0 provide comprehensive memory tracking and metrics collection capabilities with flexible BE node discovery.
### 4. Run the Service
@@ -211,19 +221,19 @@ Execute the following command to start the server:
This command starts the FastAPI application with Streamable HTTP MCP service.
**Service Endpoints (v0.3.0+):**
**Service Endpoints:**
* **Streamable HTTP**: `http://<host>:<port>/mcp` (Primary MCP endpoint - supports GET, POST, DELETE, OPTIONS)
* **Health Check**: `http://<host>:<port>/health`
* **Status Check**: `http://<host>:<port>/status`
> **Note**: Starting from v0.3.0, only Streamable HTTP mode is supported for web-based communication. SSE endpoints have been removed.
> **Note**: The server uses Streamable HTTP for web-based communication, providing unified request/response and streaming capabilities.
## Usage
Interaction with the Doris MCP Server requires an **MCP Client**. The client connects to the server's Streamable HTTP endpoint and sends requests according to the MCP specification to invoke the server's tools.
**Main Interaction Flow (v0.3.0+):**
**Main Interaction Flow:**
1. **Client Initialization**: Send an `initialize` method call to `/mcp` (Streamable HTTP).
2. **(Optional) Discover Tools**: The client can call `tools/list` to get the list of supported tools, their descriptions, and parameter schemas.
@@ -235,8 +245,6 @@ Interaction with the Doris MCP Server requires an **MCP Client**. The client con
* **Non-streaming**: The client receives a response containing `content` or `isError`.
* **Streaming**: The client receives a series of progress notifications, followed by a final response.
> **Migration Note**: If you're upgrading from v0.2.x, note that tool names have been simplified (removed `mcp_doris_` prefix) and the communication protocol has been updated to use Streamable HTTP exclusively.
### Catalog Federation Support
The Doris MCP Server supports **catalog federation**, enabling interaction with multiple data catalogs (internal Doris tables and external data sources like Hive, MySQL, etc.) within a unified interface.
@@ -305,7 +313,7 @@ The Doris MCP Server supports **catalog federation**, enabling interaction with
}
```
## Security Configuration (v0.3.0+)
## Security Configuration
The Doris MCP Server includes a comprehensive security framework that provides enterprise-level protection through authentication, authorization, SQL security validation, and data masking capabilities.
@@ -612,7 +620,7 @@ uv run --project /path/to/doris-mcp-server doris-mcp-server
}
```
### Streamable HTTP Mode (v0.3.0+)
### Streamable HTTP Mode
Streamable HTTP mode requires you to run the MCP server independently first, and then configure Cursor to connect to it.
@@ -634,12 +642,10 @@ Streamable HTTP mode requires you to run the MCP server independently first, and
}
```
> **Note**: Adjust the host/port if your server runs on a different address. The `/mcp` endpoint is the unified Streamable HTTP interface introduced in v0.3.0.
> **Note**: Adjust the host/port if your server runs on a different address. The `/mcp` endpoint is the unified Streamable HTTP interface.
After configuring either mode in Cursor, you should be able to select the server (e.g., `doris-stdio` or `doris-http`) and use its tools.
> **⚠️ Migration from v0.2.x**: If you were using SSE mode (`/sse` endpoint), update your configuration to use the new Streamable HTTP endpoint (`/mcp`).
## Directory Structure
```
@@ -678,22 +684,22 @@ doris-mcp-server/
## Developing New Tools
This section outlines the process for adding new MCP tools to the Doris MCP Server, based on the current modular architecture.
This section outlines the process for adding new MCP tools to the Doris MCP Server, based on the unified modular architecture with centralized tool management.
### 1. Leverage Existing Utility Modules
The server provides comprehensive utility modules for common database operations:
* **`doris_mcp_server/utils/db.py`**: Database connection management with connection pooling and health monitoring.
* **`doris_mcp_server/utils/query_executor.py`**: High-performance SQL execution with caching, optimization, and performance monitoring.
* **`doris_mcp_server/utils/query_executor.py`**: High-performance SQL execution with advanced caching, optimization, and performance monitoring.
* **`doris_mcp_server/utils/schema_extractor.py`**: Metadata extraction with full catalog federation support.
* **`doris_mcp_server/utils/security.py`**: Security management, SQL validation, and data masking.
* **`doris_mcp_server/utils/analysis_tools.py`**: Data analysis and statistical tools.
* **`doris_mcp_server/utils/security.py`**: Comprehensive security management, SQL validation, and data masking.
* **`doris_mcp_server/utils/analysis_tools.py`**: Advanced data analysis and statistical tools.
* **`doris_mcp_server/utils/config.py`**: Configuration management with validation.
### 2. Implement Tool Logic
Add your new tool to the `DorisToolsManager` class in `doris_mcp_server/tools/tools_manager.py`. The tools manager provides a centralized approach to tool registration and execution.
Add your new tool to the `DorisToolsManager` class in `doris_mcp_server/tools/tools_manager.py`. The tools manager provides a centralized approach to tool registration and execution with unified interfaces.
**Example:** Adding a new analysis tool:
@@ -762,12 +768,13 @@ async def your_new_analysis_tool_wrapper(arguments: Dict[str, Any]) -> List[Dict
### 4. Advanced Features
For more complex tools, you can leverage:
For more complex tools, you can leverage the comprehensive framework:
* **Caching**: Use the query executor's built-in caching for performance
* **Security**: Apply SQL validation and data masking through the security manager
* **Prompts**: Use the prompts manager for intelligent query generation
* **Resources**: Expose metadata through the resources manager
* **Advanced Caching**: Use the query executor's built-in caching for enhanced performance
* **Enterprise Security**: Apply comprehensive SQL validation and data masking through the security manager
* **Intelligent Prompts**: Use the prompts manager for advanced query generation
* **Resource Management**: Expose metadata through the resources manager
* **Performance Monitoring**: Integrate with the analysis tools for monitoring capabilities
### 5. Testing
@@ -871,6 +878,48 @@ If you have further requirements for the returned results, you can describe the
3. **Configuration File**:
Modify the corresponding configuration items in the `.env` file.
### Q: How to configure BE nodes for monitoring tools?
**A:** Choose the appropriate configuration based on your deployment scenario:
**External Network (Manual Configuration):**
```bash
# Manually specify BE node addresses
DORIS_BE_HOSTS=10.1.1.100,10.1.1.101,10.1.1.102
DORIS_BE_WEBSERVER_PORT=8040
```
**Internal Network (Automatic Discovery):**
```bash
# Leave BE_HOSTS empty for auto-discovery
# DORIS_BE_HOSTS= # Not set or empty
# System will use 'SHOW BACKENDS' command to get internal IPs
```
### Q: How to use SQL Explain/Profile files with LLM for optimization?
**A:** The tools provide both truncated content and complete files for LLM analysis:
1. **Get Analysis Results:**
```json
{
"content": "Truncated plan for immediate review",
"file_path": "/tmp/explain_12345.txt",
"is_content_truncated": true
}
```
2. **LLM Analysis Workflow:**
- Review truncated content for quick insights
- Upload the complete file to your LLM as an attachment
- Request optimization suggestions or performance analysis
- Implement recommended improvements
3. **Configure Content Size:**
```bash
MAX_RESPONSE_CONTENT_SIZE=4096 # Adjust as needed
```
### Q: How to enable data security and masking features?
**A:** Set the following configurations in your `.env` file:

12
doris_mcp_client/README.md
View File

@@ -133,9 +133,6 @@ async def database_operations(client):
# Get table schema
schema = await client.get_table_schema("table_name", "db_name")
# Column data analysis
analysis = await client.analyze_column("table", "column", "basic")
```
## 🧪 Testing
@@ -177,7 +174,6 @@ python test_unified_client.py benchmark
2. get_table_list: Get table list for specified database
3. get_table_schema: Get table structure information
4. exec_query: Execute SQL query
5. column_analysis: Analyze column data distribution and statistics
...
🧪 Testing basic functionality...
@@ -189,8 +185,6 @@ python test_unified_client.py benchmark
✅ SSB query successful
4️⃣ Getting table structure...
✅ Table structure retrieved successfully
5️⃣ Column data analysis...
✅ Column analysis successful
✅ HTTP mode testing completed!
```
@@ -255,12 +249,6 @@ async def comprehensive_example():
# Get table schema
schema_result = await client.get_table_schema("lineorder", "ssb")
print(f"Table schema: {schema_result}")
# Column analysis
analysis_result = await client.analyze_column(
"lineorder", "lo_orderkey", "basic"
)
print(f"Column analysis: {analysis_result}")
await client.connect_and_run(demo_operations)

18
doris_mcp_client/client.py
View File

@@ -422,18 +422,14 @@ class DorisUnifiedClient:
return await self.call_tool(tool_name, kwargs)
async def analyze_column(self, table_name: str, column_name: str, analysis_type: str = "basic", **kwargs) -> dict[str, Any]:
"""Analyze column"""
tool_name = await self._find_tool_by_pattern(["column_analysis", "analyze_column", "column"])
async def get_memory_stats(self, tracker_type: str = "overview", include_details: bool = True, **kwargs) -> dict[str, Any]:
"""Get memory statistics"""
tool_name = await self._find_tool_by_pattern(["memory", "realtime_memory"])
if not tool_name:
return {"success": False, "error": "Column analysis tool not found"}
arguments = {
"table_name": table_name,
"column_name": column_name,
"analysis_type": analysis_type,
**kwargs
}
return {"success": False, "error": "Memory stats tool not found"}
arguments = {"tracker_type": tracker_type, "include_details": include_details}
arguments.update(kwargs)
return await self.call_tool(tool_name, arguments)
async def call_tool_by_function(self, function_description: str, arguments: dict[str, Any]) -> dict[str, Any]:

693
doris_mcp_server/tools/tools_manager.py
View File

@@ -28,7 +28,8 @@ from mcp.types import Tool
from ..utils.db import DorisConnectionManager
from ..utils.query_executor import DorisQueryExecutor
from ..utils.analysis_tools import TableAnalyzer, PerformanceMonitor
from ..utils.analysis_tools import TableAnalyzer, SQLAnalyzer, MemoryTracker
from ..utils.monitoring_tools import DorisMonitoringTools
from ..utils.schema_extractor import MetadataExtractor
from ..utils.logger import get_logger
@@ -45,8 +46,10 @@ class DorisToolsManager:
# Initialize business logic processors
self.query_executor = DorisQueryExecutor(connection_manager)
self.table_analyzer = TableAnalyzer(connection_manager)
self.performance_monitor = PerformanceMonitor(connection_manager)
self.sql_analyzer = SQLAnalyzer(connection_manager)
self.metadata_extractor = MetadataExtractor(connection_manager=connection_manager)
self.monitoring_tools = DorisMonitoringTools(connection_manager)
self.memory_tracker = MemoryTracker(connection_manager)
logger.info("DorisToolsManager initialized with business logic processors")
@@ -54,99 +57,6 @@ class DorisToolsManager:
"""Register all tools to MCP server"""
logger.info("Starting to register MCP tools")
# Column statistical analysis tool
@mcp.tool(
"column_analysis",
description="""[Function Description]: Analyze statistical information and data distribution of the specified column.
[Parameter Content]:
- table_name (string) [Required] - Name of the table to analyze
- column_name (string) [Required] - Name of the column to analyze
- analysis_type (string) [Optional] - Type of analysis to perform, default is "basic"
* "basic": Basic statistics (count, null values, distinct values)
* "distribution": Data distribution analysis (frequency, percentiles)
* "detailed": Comprehensive analysis including all above plus patterns and outliers
""",
inputSchema={
"type": "object",
"properties": {
"table_name": {"type": "string", "description": "Table name"},
"column_name": {
"type": "string",
"description": "Column name to analyze",
},
"analysis_type": {
"type": "string",
"enum": ["basic", "distribution", "detailed"],
"description": "Analysis type",
"default": "basic",
},
},
"required": ["table_name", "column_name"],
}
)
async def column_analysis_tool(
table_name: str,
column_name: str,
analysis_type: str = "basic"
) -> str:
"""Column statistical analysis tool"""
return await self.call_tool("column_analysis", {
"table_name": table_name,
"column_name": column_name,
"analysis_type": analysis_type
})
# Database performance monitoring tool
@mcp.tool(
"performance_stats[Experimental]",
description="""[Important]: This tool is experimental and may not be fully functional!
[Function Description]: Get database performance statistics information.
[Parameter Content]:
- metric_type (string) [Optional] - Type of performance metrics to retrieve, default is "queries"
* "queries": Query performance metrics (execution time, frequency, etc.)
* "connections": Connection statistics (active connections, connection pool status)
* "tables": Table-level statistics (size, row count, access patterns)
* "system": System-level metrics (CPU, memory, disk usage)
- time_range (string) [Optional] - Time range for statistics, default is "1h"
* "1h": Last 1 hour
* "6h": Last 6 hours
* "24h": Last 24 hours
* "7d": Last 7 days
""",
inputSchema={
"type": "object",
"properties": {
"metric_type": {
"type": "string",
"enum": ["queries", "connections", "tables", "system"],
"description": "Performance metric type",
"default": "queries",
},
"time_range": {
"type": "string",
"enum": ["1h", "6h", "24h", "7d"],
"description": "Time range",
"default": "1h",
},
},
}
)
async def performance_stats_tool(
metric_type: str = "queries",
time_range: str = "1h"
) -> str:
"""Database performance monitoring tool"""
return await self.call_tool("performance_stats", {
"metric_type": metric_type,
"time_range": time_range
})
# SQL query execution tool (supports catalog federation queries)
@mcp.tool(
@@ -352,81 +262,227 @@ class DorisToolsManager:
"random_string": random_string
})
logger.info("Successfully registered 11 tools to MCP server (2 core tools + 9 migrated tools)")
# SQL Explain tool
@mcp.tool(
"get_sql_explain",
description="""[Function Description]: Get SQL execution plan using EXPLAIN command based on Doris syntax.
[Parameter Content]:
- sql (string) [Required] - SQL statement to explain
- verbose (boolean) [Optional] - Whether to show verbose information, default is false
- db_name (string) [Optional] - Target database name, defaults to the current database
- catalog_name (string) [Optional] - Target catalog name for federation queries, defaults to current catalog
""",
)
async def get_sql_explain_tool(
sql: str,
verbose: bool = False,
db_name: str = None,
catalog_name: str = None
) -> str:
"""Get SQL execution plan"""
return await self.call_tool("get_sql_explain", {
"sql": sql,
"verbose": verbose,
"db_name": db_name,
"catalog_name": catalog_name
})
# SQL Profile tool
@mcp.tool(
"get_sql_profile",
description="""[Function Description]: Get SQL execution profile by setting trace ID and fetching profile via FE HTTP API.
[Parameter Content]:
- sql (string) [Required] - SQL statement to profile
- db_name (string) [Optional] - Target database name, defaults to the current database
- catalog_name (string) [Optional] - Target catalog name for federation queries, defaults to current catalog
- timeout (integer) [Optional] - Query timeout in seconds, default is 30
""",
)
async def get_sql_profile_tool(
sql: str,
db_name: str = None,
catalog_name: str = None,
timeout: int = 30
) -> str:
"""Get SQL execution profile"""
return await self.call_tool("get_sql_profile", {
"sql": sql,
"db_name": db_name,
"catalog_name": catalog_name,
"timeout": timeout
})
# Table data size tool
@mcp.tool(
"get_table_data_size",
description="""[Function Description]: Get table data size information via FE HTTP API.
[Parameter Content]:
- db_name (string) [Optional] - Database name, if not specified returns all databases
- table_name (string) [Optional] - Table name, if not specified returns all tables in the database
- single_replica (boolean) [Optional] - Whether to get single replica data size, default is false
""",
)
async def get_table_data_size_tool(
db_name: str = None,
table_name: str = None,
single_replica: bool = False
) -> str:
"""Get table data size information"""
return await self.call_tool("get_table_data_size", {
"db_name": db_name,
"table_name": table_name,
"single_replica": single_replica
})
# Monitoring metrics definition tool
@mcp.tool(
"get_monitoring_metrics_info",
description="""[Function Description]: Get Doris monitoring metrics definitions and descriptions without executing queries.
[Parameter Content]:
- role (string) [Optional] - Node role to get metric definitions for, default is "all"
* "fe": Only FE metrics definitions
* "be": Only BE metrics definitions
* "all": Both FE and BE metrics definitions
- monitor_type (string) [Optional] - Type of monitoring metrics, default is "all"
* "process": Process monitoring metrics
* "jvm": JVM monitoring metrics (FE only)
* "machine": Machine monitoring metrics
* "all": All monitoring types
- priority (string) [Optional] - Metric priority level, default is "core"
* "core": Only core essential metrics (10-12 items for production use)
* "p0": Only P0 (highest priority) metrics definitions
* "all": All metrics definitions (P0 and non-P0)
""",
)
async def get_monitoring_metrics_info_tool(
role: str = "all",
monitor_type: str = "all",
priority: str = "core"
) -> str:
"""Get Doris monitoring metrics definitions"""
return await self.call_tool("get_monitoring_metrics_info", {
"role": role,
"monitor_type": monitor_type,
"priority": priority
})
# Monitoring metrics data tool
@mcp.tool(
"get_monitoring_metrics_data",
description="""[Function Description]: Get actual Doris monitoring metrics data from FE and BE nodes via HTTP API.
[Parameter Content]:
- role (string) [Optional] - Node role to monitor, default is "all"
* "fe": Only FE nodes
* "be": Only BE nodes
* "all": Both FE and BE nodes
- monitor_type (string) [Optional] - Type of monitoring metrics, default is "all"
* "process": Process monitoring metrics
* "jvm": JVM monitoring metrics (FE only)
* "machine": Machine monitoring metrics
* "all": All monitoring types
- priority (string) [Optional] - Metric priority level, default is "core"
* "core": Only core essential metrics (10-12 items for production use)
* "p0": Only P0 (highest priority) metrics
* "all": All metrics (P0 and non-P0)
- include_raw_metrics (boolean) [Optional] - Whether to include raw detailed metrics data (can be very large)
""",
)
async def get_monitoring_metrics_data_tool(
role: str = "all",
monitor_type: str = "all",
priority: str = "core",
include_raw_metrics: bool = False
) -> str:
"""Get Doris monitoring metrics data"""
return await self.call_tool("get_monitoring_metrics_data", {
"role": role,
"monitor_type": monitor_type,
"priority": priority,
"include_raw_metrics": include_raw_metrics
})
# Real-time memory tracker tool
@mcp.tool(
"get_realtime_memory_stats",
description="""[Function Description]: Get real-time memory statistics via Doris BE Memory Tracker web interface.
[Parameter Content]:
- tracker_type (string) [Optional] - Type of memory trackers to retrieve, default is "overview"
* "overview": Overview type trackers (process memory, tracked memory summary)
* "global": Global shared memory trackers (cache, metadata)
* "query": Query-related memory trackers
* "load": Load-related memory trackers
* "compaction": Compaction-related memory trackers
* "all": All memory tracker types
- include_details (boolean) [Optional] - Whether to include detailed tracker information and definitions, default is true
""",
)
async def get_realtime_memory_stats_tool(
tracker_type: str = "overview",
include_details: bool = True
) -> str:
"""Get real-time memory statistics tool"""
return await self.call_tool("get_realtime_memory_stats", {
"tracker_type": tracker_type,
"include_details": include_details
})
# Historical memory tracker tool
@mcp.tool(
"get_historical_memory_stats",
description="""[Function Description]: Get historical memory statistics via Doris BE Bvar interface.
[Parameter Content]:
- tracker_names (array) [Optional] - List of specific tracker names to query, if not specified will get common trackers
* Example: ["process_resident_memory", "global", "query", "load", "compaction"]
- time_range (string) [Optional] - Time range for historical data, default is "1h"
* "1h": Last 1 hour
* "6h": Last 6 hours
* "24h": Last 24 hours
""",
)
async def get_historical_memory_stats_tool(
tracker_names: List[str] = None,
time_range: str = "1h"
) -> str:
"""Get historical memory statistics tool"""
return await self.call_tool("get_historical_memory_stats", {
"tracker_names": tracker_names,
"time_range": time_range
})
logger.info("Successfully registered 16 tools to MCP server")
async def list_tools(self) -> List[Tool]:
"""List all available query tools (for stdio mode)"""
tools = [
Tool(
name="column_analysis[Experimental]",
description="""[Important]: This tool is experimental and may not be fully functional!
[Function Description]: Analyze statistical information and data distribution of the specified column.
[Parameter Content]:
- table_name (string) [Required] - Name of the table to analyze
- column_name (string) [Required] - Name of the column to analyze
- analysis_type (string) [Optional] - Type of analysis to perform, default is "basic"
* "basic": Basic statistics (count, null values, distinct values)
* "distribution": Data distribution analysis (frequency, percentiles)
* "detailed": Comprehensive analysis including all above plus patterns and outliers
""",
inputSchema={
"type": "object",
"properties": {
"table_name": {"type": "string", "description": "Table name"},
"column_name": {
"type": "string",
"description": "Column name to analyze",
},
"analysis_type": {
"type": "string",
"enum": ["basic", "distribution", "detailed"],
"description": "Analysis type",
"default": "basic",
},
},
"required": ["table_name", "column_name"],
},
),
Tool(
name="performance_stats",
description="""[Function Description]: Get database performance statistics information.
[Parameter Content]:
- metric_type (string) [Optional] - Type of performance metrics to retrieve, default is "queries"
* "queries": Query performance metrics (execution time, frequency, etc.)
* "connections": Connection statistics (active connections, connection pool status)
* "tables": Table-level statistics (size, row count, access patterns)
* "system": System-level metrics (CPU, memory, disk usage)
- time_range (string) [Optional] - Time range for statistics, default is "1h"
* "1h": Last 1 hour
* "6h": Last 6 hours
* "24h": Last 24 hours
* "7d": Last 7 days
""",
inputSchema={
"type": "object",
"properties": {
"metric_type": {
"type": "string",
"enum": ["queries", "connections", "tables", "system"],
"description": "Performance metric type",
"default": "queries",
},
"time_range": {
"type": "string",
"enum": ["1h", "6h", "24h", "7d"],
"description": "Time range",
"default": "1h",
},
},
},
),
Tool(
name="exec_query",
description="""[Function Description]: Execute SQL query and return result command with catalog federation support.
@@ -610,6 +666,188 @@ class DorisToolsManager:
"required": ["random_string"],
},
),
Tool(
name="get_sql_explain",
description="""[Function Description]: Get SQL execution plan using EXPLAIN command based on Doris syntax.
[Parameter Content]:
- sql (string) [Required] - SQL statement to explain
- verbose (boolean) [Optional] - Whether to show verbose information, default is false
- db_name (string) [Optional] - Target database name, defaults to the current database
- catalog_name (string) [Optional] - Target catalog name for federation queries, defaults to current catalog
""",
inputSchema={
"type": "object",
"properties": {
"sql": {"type": "string", "description": "SQL statement to explain"},
"verbose": {"type": "boolean", "description": "Whether to show verbose information", "default": False},
"db_name": {"type": "string", "description": "Database name"},
"catalog_name": {"type": "string", "description": "Catalog name"},
},
"required": ["sql"],
},
),
Tool(
name="get_sql_profile",
description="""[Function Description]: Get SQL execution profile by setting trace ID and fetching profile via FE HTTP API.
[Parameter Content]:
- sql (string) [Required] - SQL statement to profile
- db_name (string) [Optional] - Target database name, defaults to the current database
- catalog_name (string) [Optional] - Target catalog name for federation queries, defaults to current catalog
- timeout (integer) [Optional] - Query timeout in seconds, default is 30
""",
inputSchema={
"type": "object",
"properties": {
"sql": {"type": "string", "description": "SQL statement to profile"},
"db_name": {"type": "string", "description": "Database name"},
"catalog_name": {"type": "string", "description": "Catalog name"},
"timeout": {"type": "integer", "description": "Query timeout in seconds", "default": 30},
},
"required": ["sql"],
},
),
Tool(
name="get_table_data_size",
description="""[Function Description]: Get table data size information via FE HTTP API.
[Parameter Content]:
- db_name (string) [Optional] - Database name, if not specified returns all databases
- table_name (string) [Optional] - Table name, if not specified returns all tables in the database
- single_replica (boolean) [Optional] - Whether to get single replica data size, default is false
""",
inputSchema={
"type": "object",
"properties": {
"db_name": {"type": "string", "description": "Database name"},
"table_name": {"type": "string", "description": "Table name"},
"single_replica": {"type": "boolean", "description": "Whether to get single replica data size", "default": False},
},
},
),
Tool(
name="get_monitoring_metrics_info",
description="""[Function Description]: Get Doris monitoring metrics definitions and descriptions without executing queries.
[Parameter Content]:
- role (string) [Optional] - Node role to get metric definitions for, default is "all"
* "fe": Only FE metrics definitions
* "be": Only BE metrics definitions
* "all": Both FE and BE metrics definitions
- monitor_type (string) [Optional] - Type of monitoring metrics, default is "all"
* "process": Process monitoring metrics
* "jvm": JVM monitoring metrics (FE only)
* "machine": Machine monitoring metrics
* "all": All monitoring types
- priority (string) [Optional] - Metric priority level, default is "core"
* "core": Only core essential metrics (10-12 items for production use)
* "p0": Only P0 (highest priority) metrics definitions
* "all": All metrics definitions (P0 and non-P0)
""",
inputSchema={
"type": "object",
"properties": {
"role": {"type": "string", "enum": ["fe", "be", "all"], "description": "Node role to get metric definitions for", "default": "all"},
"monitor_type": {"type": "string", "enum": ["process", "jvm", "machine", "all"], "description": "Type of monitoring metrics", "default": "all"},
"priority": {"type": "string", "enum": ["core", "p0", "all"], "description": "Metric priority level", "default": "core"},
},
},
),
Tool(
name="get_monitoring_metrics_data",
description="""[Function Description]: Get actual Doris monitoring metrics data from FE and BE nodes via HTTP API.
[Parameter Content]:
- role (string) [Optional] - Node role to monitor, default is "all"
* "fe": Only FE nodes
* "be": Only BE nodes
* "all": Both FE and BE nodes
- monitor_type (string) [Optional] - Type of monitoring metrics, default is "all"
* "process": Process monitoring metrics
* "jvm": JVM monitoring metrics (FE only)
* "machine": Machine monitoring metrics
* "all": All monitoring types
- priority (string) [Optional] - Metric priority level, default is "core"
* "core": Only core essential metrics (10-12 items for production use)
* "p0": Only P0 (highest priority) metrics
* "all": All metrics (P0 and non-P0)
- include_raw_metrics (boolean) [Optional] - Whether to include raw detailed metrics data (can be very large)
""",
inputSchema={
"type": "object",
"properties": {
"role": {"type": "string", "enum": ["fe", "be", "all"], "description": "Node role to monitor", "default": "all"},
"monitor_type": {"type": "string", "enum": ["process", "jvm", "machine", "all"], "description": "Type of monitoring metrics", "default": "all"},
"priority": {"type": "string", "enum": ["core", "p0", "all"], "description": "Metric priority level", "default": "core"},
"include_raw_metrics": {"type": "boolean", "description": "Whether to include raw detailed metrics data (can be very large)", "default": False},
},
},
),
Tool(
name="get_realtime_memory_stats",
description="""[Function Description]: Get real-time memory statistics via Doris BE Memory Tracker web interface.
[Parameter Content]:
- tracker_type (string) [Optional] - Type of memory trackers to retrieve, default is "overview"
* "overview": Overview type trackers (process memory, tracked memory summary)
* "global": Global shared memory trackers (cache, metadata)
* "query": Query-related memory trackers
* "load": Load-related memory trackers
* "compaction": Compaction-related memory trackers
* "all": All memory tracker types
- include_details (boolean) [Optional] - Whether to include detailed tracker information and definitions, default is true
""",
inputSchema={
"type": "object",
"properties": {
"tracker_type": {"type": "string", "enum": ["overview", "global", "query", "load", "compaction", "all"], "description": "Type of memory trackers to retrieve", "default": "overview"},
"include_details": {"type": "boolean", "description": "Whether to include detailed tracker information and definitions", "default": True},
},
},
),
Tool(
name="get_historical_memory_stats",
description="""[Function Description]: Get historical memory statistics via Doris BE Bvar interface.
[Parameter Content]:
- tracker_names (array) [Optional] - List of specific tracker names to query, if not specified will get common trackers
* Example: ["process_resident_memory", "global", "query", "load", "compaction"]
- time_range (string) [Optional] - Time range for historical data, default is "1h"
* "1h": Last 1 hour
* "6h": Last 6 hours
* "24h": Last 24 hours
""",
inputSchema={
"type": "object",
"properties": {
"tracker_names": {"type": "array", "items": {"type": "string"}, "description": "List of specific tracker names to query"},
"time_range": {"type": "string", "enum": ["1h", "6h", "24h"], "description": "Time range for historical data", "default": "1h"},
},
},
),
]
return tools
@@ -622,12 +860,7 @@ class DorisToolsManager:
start_time = time.time()
# Tool routing - dispatch requests to corresponding business logic processors
if name == "column_analysis":
result = await self._column_analysis_tool(arguments)
elif name == "performance_stats":
result = await self._performance_stats_tool(arguments)
# ===== 9 tool routes migrated from source project =====
elif name == "exec_query":
if name == "exec_query":
result = await self._exec_query_tool(arguments)
elif name == "get_table_schema":
result = await self._get_table_schema_tool(arguments)
@@ -645,6 +878,20 @@ class DorisToolsManager:
result = await self._get_recent_audit_logs_tool(arguments)
elif name == "get_catalog_list":
result = await self._get_catalog_list_tool(arguments)
elif name == "get_sql_explain":
result = await self._get_sql_explain_tool(arguments)
elif name == "get_sql_profile":
result = await self._get_sql_profile_tool(arguments)
elif name == "get_table_data_size":
result = await self._get_table_data_size_tool(arguments)
elif name == "get_monitoring_metrics_info":
result = await self._get_monitoring_metrics_info_tool(arguments)
elif name == "get_monitoring_metrics_data":
result = await self._get_monitoring_metrics_data_tool(arguments)
elif name == "get_realtime_memory_stats":
result = await self._get_realtime_memory_stats_tool(arguments)
elif name == "get_historical_memory_stats":
result = await self._get_historical_memory_stats_tool(arguments)
else:
raise ValueError(f"Unknown tool: {name}")
@@ -670,28 +917,6 @@ class DorisToolsManager:
}
return json.dumps(error_result, ensure_ascii=False, indent=2)
# The following are tool routing methods, responsible for calling corresponding business logic processors
async def _column_analysis_tool(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Column statistical analysis tool routing"""
table_name = arguments.get("table_name")
column_name = arguments.get("column_name")
analysis_type = arguments.get("analysis_type", "basic")
# Delegate to table analyzer for processing
return await self.table_analyzer.analyze_column(
table_name, column_name, analysis_type
)
async def _performance_stats_tool(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Database performance statistics tool routing"""
metric_type = arguments.get("metric_type", "queries")
time_range = arguments.get("time_range", "1h")
# Delegate to performance monitor for processing
return await self.performance_monitor.get_performance_stats(
metric_type, time_range
)
async def _exec_query_tool(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""SQL query execution tool routing (supports federation queries)"""
@@ -779,4 +1004,82 @@ class DorisToolsManager:
# Here we ignore it and directly call business logic
# Delegate to metadata extractor for processing
return await self.metadata_extractor.get_catalog_list_for_mcp()
return await self.metadata_extractor.get_catalog_list_for_mcp()
async def _get_sql_explain_tool(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""SQL Explain tool routing"""
sql = arguments.get("sql")
verbose = arguments.get("verbose", False)
db_name = arguments.get("db_name")
catalog_name = arguments.get("catalog_name")
# Delegate to SQL analyzer for processing
return await self.sql_analyzer.get_sql_explain(
sql, verbose, db_name, catalog_name
)
async def _get_sql_profile_tool(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""SQL Profile tool routing"""
sql = arguments.get("sql")
db_name = arguments.get("db_name")
catalog_name = arguments.get("catalog_name")
timeout = arguments.get("timeout", 30)
# Delegate to SQL analyzer for processing
return await self.sql_analyzer.get_sql_profile(
sql, db_name, catalog_name, timeout
)
async def _get_table_data_size_tool(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Table data size tool routing"""
db_name = arguments.get("db_name")
table_name = arguments.get("table_name")
single_replica = arguments.get("single_replica", False)
# Delegate to SQL analyzer for processing
return await self.sql_analyzer.get_table_data_size(
db_name, table_name, single_replica
)
async def _get_monitoring_metrics_info_tool(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Monitoring metrics info tool routing"""
role = arguments.get("role", "all")
monitor_type = arguments.get("monitor_type", "all")
priority = arguments.get("priority", "p0")
# Delegate to monitoring tools for processing (info_only=True)
return await self.monitoring_tools.get_monitoring_metrics(
role, monitor_type, priority, info_only=True, format_type="prometheus"
)
async def _get_monitoring_metrics_data_tool(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Monitoring metrics data tool routing"""
role = arguments.get("role", "all")
monitor_type = arguments.get("monitor_type", "all")
priority = arguments.get("priority", "p0")
include_raw_metrics = arguments.get("include_raw_metrics", False)
# Delegate to monitoring tools for processing (info_only=False)
return await self.monitoring_tools.get_monitoring_metrics(
role, monitor_type, priority, info_only=False, format_type="prometheus", include_raw_metrics=include_raw_metrics
)
async def _get_realtime_memory_stats_tool(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Real-time memory statistics tool routing"""
tracker_type = arguments.get("tracker_type", "overview")
include_details = arguments.get("include_details", True)
# Delegate to memory tracker for processing
return await self.memory_tracker.get_realtime_memory_stats(
tracker_type, include_details
)
async def _get_historical_memory_stats_tool(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Historical memory statistics tool routing"""
tracker_names = arguments.get("tracker_names")
time_range = arguments.get("time_range", "1h")
# Delegate to memory tracker for processing
return await self.memory_tracker.get_historical_memory_stats(
tracker_names, time_range
)

908
doris_mcp_server/utils/analysis_tools.py
View File

@@ -22,6 +22,10 @@ Provides data analysis functions including table analysis, column statistics, pe
import time
from datetime import datetime
from typing import Any, Dict, List
import uuid
import aiohttp
import hashlib
from pathlib import Path
from .db import DorisConnectionManager
from .logger import get_logger
@@ -331,4 +335,906 @@ class PerformanceMonitor:
"limit": limit,
"order_by": order_by,
"note": "Query history feature requires audit log configuration"
}
}
class SQLAnalyzer:
"""SQL analyzer for EXPLAIN and PROFILE operations"""
def __init__(self, connection_manager: DorisConnectionManager):
self.connection_manager = connection_manager
async def get_sql_explain(
self,
sql: str,
verbose: bool = False,
db_name: str = None,
catalog_name: str = None
) -> Dict[str, Any]:
"""
Get SQL execution plan using EXPLAIN command based on Doris syntax
Args:
sql: SQL statement to explain
verbose: Whether to show verbose information
db_name: Target database name
catalog_name: Target catalog name
Returns:
Dict containing explain plan file path, content, and basic info
"""
try:
# Generate unique query ID for file naming
import time
query_hash = hashlib.md5(sql.encode()).hexdigest()[:8]
timestamp = int(time.time())
query_id = f"{timestamp}_{query_hash}"
# Ensure temp directory exists
temp_dir = Path(self.connection_manager.config.temp_files_dir)
temp_dir.mkdir(parents=True, exist_ok=True)
# Create explain file path
explain_file = temp_dir / f"explain_{query_id}.txt"
logger.info(f"Generating SQL explain for query ID: {query_id}")
# Switch database if specified
if db_name:
await self.connection_manager.execute_query("explain_session", f"USE {db_name}")
# Construct EXPLAIN query
explain_type = "EXPLAIN VERBOSE" if verbose else "EXPLAIN"
explain_sql = f"{explain_type} {sql.strip().rstrip(';')}"
logger.info(f"Executing explain query: {explain_sql}")
# Execute explain query
result = await self.connection_manager.execute_query("explain_session", explain_sql)
# Format explain output
explain_content = []
explain_content.append(f"=== SQL EXPLAIN PLAN ===")
explain_content.append(f"Query ID: {query_id}")
explain_content.append(f"Timestamp: {time.strftime('%Y-%m-%d %H:%M:%S')}")
explain_content.append(f"Database: {db_name or 'current'}")
explain_content.append(f"Verbose: {verbose}")
explain_content.append("")
explain_content.append("=== ORIGINAL SQL ===")
explain_content.append(sql)
explain_content.append("")
explain_content.append("=== EXPLAIN QUERY ===")
explain_content.append(explain_sql)
explain_content.append("")
explain_content.append("=== EXECUTION PLAN ===")
if result and result.data:
for row in result.data:
if isinstance(row, dict):
# Handle dict format
for key, value in row.items():
explain_content.append(f"{key}: {value}")
elif isinstance(row, (list, tuple)):
# Handle tuple/list format
explain_content.append(" | ".join(str(col) for col in row))
else:
# Handle string format
explain_content.append(str(row))
else:
explain_content.append("No execution plan data returned")
explain_content.append("")
explain_content.append("=== METADATA ===")
explain_content.append(f"Execution time: {result.execution_time if result else 'N/A'} seconds")
explain_content.append(f"Rows returned: {len(result.data) if result and result.data else 0}")
# Get full content
full_content = '\n'.join(explain_content)
# Write to file
with open(explain_file, 'w', encoding='utf-8') as f:
f.write(full_content)
logger.info(f"Explain plan saved to: {explain_file.absolute()}")
# Get max response size from config
max_size = self.connection_manager.config.performance.max_response_content_size
# Truncate content if needed
truncated_content = full_content
is_truncated = False
if len(full_content) > max_size:
truncated_content = full_content[:max_size] + "\n\n=== CONTENT TRUNCATED ===\n[Content is truncated due to size limit. Full content is saved to file.]"
is_truncated = True
return {
"success": True,
"query_id": query_id,
"explain_file_path": str(explain_file.absolute()),
"file_size_bytes": explain_file.stat().st_size,
"content": truncated_content,
"content_size": len(truncated_content),
"is_content_truncated": is_truncated,
"original_content_size": len(full_content),
"sql_preview": sql[:100] + "..." if len(sql) > 100 else sql,
"verbose": verbose,
"database": db_name,
"catalog": catalog_name,
"timestamp": time.strftime('%Y-%m-%d %H:%M:%S'),
"execution_time": result.execution_time if result else None,
"plan_lines_count": len(result.data) if result and result.data else 0
}
except Exception as e:
logger.error(f"Failed to get SQL explain: {str(e)}")
return {
"success": False,
"error": f"Failed to get SQL explain: {str(e)}",
"sql_preview": sql[:100] + "..." if len(sql) > 100 else sql,
"timestamp": time.strftime('%Y-%m-%d %H:%M:%S')
}
async def get_sql_profile(
self,
sql: str,
db_name: str = None,
catalog_name: str = None,
timeout: int = 30
) -> Dict[str, Any]:
"""
Get SQL execution profile by setting trace ID and fetching profile via HTTP API
Args:
sql: SQL statement to profile
db_name: Target database name
catalog_name: Target catalog name
timeout: Query timeout in seconds
Returns:
Dict containing profile file path, content, and basic info
"""
try:
# Generate unique trace ID and query ID for file naming
trace_id = str(uuid.uuid4())
import time
query_hash = hashlib.md5(sql.encode()).hexdigest()[:8]
timestamp = int(time.time())
file_query_id = f"{timestamp}_{query_hash}"
# Ensure temp directory exists
temp_dir = Path(self.connection_manager.config.temp_files_dir)
temp_dir.mkdir(parents=True, exist_ok=True)
# Create profile file path
profile_file = temp_dir / f"profile_{file_query_id}.txt"
logger.info(f"Generated trace ID for SQL profiling: {trace_id}")
logger.info(f"Profile will be saved to: {profile_file}")
connection = await self.connection_manager.get_connection("query")
try:
# Switch to specified database/catalog if provided
if catalog_name:
await connection.execute(f"USE `{catalog_name}`")
if db_name:
await connection.execute(f"USE `{db_name}`")
# Set trace ID for the session using session variable
# According to official docs: set session_context="trace_id:your_trace_id"
await connection.execute(f'set session_context="trace_id:{trace_id}"')
logger.info(f"Set trace ID: {trace_id}")
# Enable profile
await connection.execute(f'set enable_profile=true')
logger.info(f"Enabled profile")
# Execute the SQL statement
logger.info(f"Executing SQL with trace ID: {sql}")
start_time = time.time()
sql_result = await connection.execute(sql)
execution_time = time.time() - start_time
logger.info(f"SQL execution completed in {execution_time:.3f}s")
# Get query ID from trace ID via HTTP API
query_id = await self._get_query_id_by_trace_id(trace_id)
if not query_id:
return {
"success": False,
"error": "Failed to get query ID from trace ID",
"trace_id": trace_id,
"sql": sql,
"execution_time": execution_time
}
logger.info(f"Retrieved query ID: {query_id}")
# Get profile data
profile_data = await self._get_profile_by_query_id(query_id)
if not profile_data:
# Save error info to file
profile_content = [
f"=== SQL PROFILE RESULT ===",
f"File Query ID: {file_query_id}",
f"Trace ID: {trace_id}",
f"Query ID: {query_id}",
f"Timestamp: {time.strftime('%Y-%m-%d %H:%M:%S')}",
f"Database: {db_name or 'current'}",
f"Status: FAILED",
"",
"=== ORIGINAL SQL ===",
sql,
"",
"=== ERROR INFO ===",
"Failed to get profile data. This may be due to:",
"1) Profile data not generated yet",
"2) Query ID expired",
"3) Insufficient permissions to access profile data",
"",
"=== EXECUTION INFO ===",
f"Query execution: SUCCESSFUL",
f"Execution time: {execution_time:.3f} seconds",
f"Note: Query execution was successful, but profile data is not available"
]
# Get full content
full_profile_content = '\n'.join(profile_content)
with open(profile_file, 'w', encoding='utf-8') as f:
f.write(full_profile_content)
# Get max response size from config
max_size = self.connection_manager.config.performance.max_response_content_size
# Truncate content if needed
truncated_content = full_profile_content
is_truncated = False
if len(full_profile_content) > max_size:
truncated_content = full_profile_content[:max_size] + "\n\n=== CONTENT TRUNCATED ===\n[Content is truncated due to size limit. Full content is saved to file.]"
is_truncated = True
return {
"success": False,
"file_query_id": file_query_id,
"trace_id": trace_id,
"query_id": query_id,
"profile_file_path": str(profile_file.absolute()),
"file_size_bytes": profile_file.stat().st_size,
"content": truncated_content,
"content_size": len(truncated_content),
"is_content_truncated": is_truncated,
"original_content_size": len(full_profile_content),
"sql_preview": sql[:100] + "..." if len(sql) > 100 else sql,
"execution_time": execution_time,
"error": "Failed to get profile data",
"timestamp": time.strftime('%Y-%m-%d %H:%M:%S')
}
# Format profile output
profile_content = []
profile_content.append(f"=== SQL PROFILE RESULT ===")
profile_content.append(f"File Query ID: {file_query_id}")
profile_content.append(f"Trace ID: {trace_id}")
profile_content.append(f"Query ID: {query_id}")
profile_content.append(f"Timestamp: {time.strftime('%Y-%m-%d %H:%M:%S')}")
profile_content.append(f"Database: {db_name or 'current'}")
profile_content.append(f"Status: SUCCESS")
profile_content.append("")
profile_content.append("=== ORIGINAL SQL ===")
profile_content.append(sql)
profile_content.append("")
profile_content.append("=== EXECUTION INFO ===")
profile_content.append(f"Execution time: {execution_time:.3f} seconds")
if hasattr(sql_result, 'data') and sql_result.data:
profile_content.append(f"Result rows: {len(sql_result.data)}")
if sql_result.data and sql_result.data[0]:
profile_content.append(f"Result columns: {list(sql_result.data[0].keys())}")
profile_content.append("")
profile_content.append("=== PROFILE DATA ===")
if isinstance(profile_data, dict):
import json
profile_content.append(json.dumps(profile_data, indent=2, ensure_ascii=False))
else:
profile_content.append(str(profile_data))
# Get full content
full_profile_content = '\n'.join(profile_content)
# Write to file
with open(profile_file, 'w', encoding='utf-8') as f:
f.write(full_profile_content)
logger.info(f"Profile data saved to: {profile_file.absolute()}")
# Get max response size from config
max_size = self.connection_manager.config.performance.max_response_content_size
# Truncate content if needed
truncated_content = full_profile_content
is_truncated = False
if len(full_profile_content) > max_size:
truncated_content = full_profile_content[:max_size] + "\n\n=== CONTENT TRUNCATED ===\n[Content is truncated due to size limit. Full content is saved to file.]"
is_truncated = True
return {
"success": True,
"file_query_id": file_query_id,
"trace_id": trace_id,
"query_id": query_id,
"profile_file_path": str(profile_file.absolute()),
"file_size_bytes": profile_file.stat().st_size,
"content": truncated_content,
"content_size": len(truncated_content),
"is_content_truncated": is_truncated,
"original_content_size": len(full_profile_content),
"sql_preview": sql[:100] + "..." if len(sql) > 100 else sql,
"database": db_name,
"catalog": catalog_name,
"execution_time": execution_time,
"sql_result_summary": {
"row_count": len(sql_result.data) if hasattr(sql_result, 'data') and sql_result.data else 0,
"columns": list(sql_result.data[0].keys()) if hasattr(sql_result, 'data') and sql_result.data and sql_result.data[0] else []
},
"timestamp": time.strftime('%Y-%m-%d %H:%M:%S')
}
except Exception as e:
logger.error(f"Error during SQL execution or profile retrieval: {str(e)}")
# Save error info to file
profile_content = [
f"=== SQL PROFILE RESULT ===",
f"File Query ID: {file_query_id}",
f"Trace ID: {trace_id}",
f"Timestamp: {time.strftime('%Y-%m-%d %H:%M:%S')}",
f"Database: {db_name or 'current'}",
f"Status: ERROR",
"",
"=== ORIGINAL SQL ===",
sql,
"",
"=== ERROR INFO ===",
f"SQL execution or profile retrieval failed: {str(e)}",
"",
"=== EXECUTION INFO ===",
"Query execution failed during profiling process"
]
# Get full content
full_profile_content = '\n'.join(profile_content)
with open(profile_file, 'w', encoding='utf-8') as f:
f.write(full_profile_content)
# Get max response size from config
max_size = self.connection_manager.config.performance.max_response_content_size
# Truncate content if needed
truncated_content = full_profile_content
is_truncated = False
if len(full_profile_content) > max_size:
truncated_content = full_profile_content[:max_size] + "\n\n=== CONTENT TRUNCATED ===\n[Content is truncated due to size limit. Full content is saved to file.]"
is_truncated = True
return {
"success": False,
"file_query_id": file_query_id,
"trace_id": trace_id,
"profile_file_path": str(profile_file.absolute()),
"file_size_bytes": profile_file.stat().st_size,
"content": truncated_content,
"content_size": len(truncated_content),
"is_content_truncated": is_truncated,
"original_content_size": len(full_profile_content),
"sql_preview": sql[:100] + "..." if len(sql) > 100 else sql,
"error": f"SQL execution or profile retrieval failed: {str(e)}",
"database": db_name,
"catalog": catalog_name,
"timestamp": time.strftime('%Y-%m-%d %H:%M:%S')
}
except Exception as e:
logger.error(f"SQL PROFILE failed: {str(e)}")
return {
"success": False,
"error": f"SQL PROFILE failed: {str(e)}",
"sql_preview": sql[:100] + "..." if len(sql) > 100 else sql,
"database": db_name,
"catalog": catalog_name,
"timestamp": time.strftime('%Y-%m-%d %H:%M:%S')
}
async def _get_query_id_by_trace_id(self, trace_id: str) -> str:
"""
Get query ID by trace ID via FE HTTP API
Args:
trace_id: The trace ID set during query execution
Returns:
Query ID string or None if not found
"""
try:
# Get database config
db_config = self.connection_manager.config.database
# Build HTTP API URL according to official documentation
# Reference: https://doris.apache.org/zh-CN/docs/admin-manual/open-api/fe-http/query-profile-action#通过-trace-id-获取-query-id
url = f"http://{db_config.host}:{db_config.fe_http_port}/rest/v2/manager/query/trace_id/{trace_id}"
# HTTP Basic Auth
auth = aiohttp.BasicAuth(db_config.user, db_config.password)
logger.info(f"Requesting query ID from: {url}")
async with aiohttp.ClientSession() as session:
async with session.get(url, auth=auth, timeout=10) as response:
if response.status == 200:
# Check content type first
content_type = response.headers.get('content-type', '')
response_text = await response.text()
logger.info(f"Response content type: {content_type}")
logger.info(f"Response body: {response_text}")
# Parse JSON response (regardless of content-type)
if response_text.strip():
try:
import json
result = json.loads(response_text)
logger.info(f"Query ID API response: {result}")
# Parse response according to Doris API format
if result.get("code") == 0 and result.get("data"):
data = result["data"]
# Data can be either a string (query_id) or object with query_ids
if isinstance(data, str):
logger.info(f"Found query ID: {data}")
return data
elif isinstance(data, dict) and "query_ids" in data:
query_ids = data["query_ids"]
if query_ids:
query_id = query_ids[0] # Take the first query ID
logger.info(f"Found query ID: {query_id}")
return query_id
else:
logger.warning("No query IDs found in response")
else:
logger.error(f"API returned error: {result}")
except json.JSONDecodeError as e:
logger.error(f"Failed to parse JSON response: {e}")
# Fallback: try to extract query ID using regex
import re
query_id_pattern = r'[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}'
matches = re.findall(query_id_pattern, response_text)
if matches:
query_id = matches[0]
logger.info(f"Extracted query ID from text: {query_id}")
return query_id
else:
logger.error(f"HTTP request failed with status {response.status}")
response_text = await response.text()
logger.error(f"Response body: {response_text}")
return None
except Exception as e:
logger.error(f"Failed to get query ID by trace ID: {str(e)}")
return None
async def _get_profile_by_query_id(self, query_id: str) -> Dict[str, Any]:
"""
Get profile data by query ID via FE HTTP API
Args:
query_id: The query ID
Returns:
Profile data dict or None if failed
"""
try:
# Get database config
db_config = self.connection_manager.config.database
# Try both API endpoints according to official documentation
urls = [
f"http://{db_config.host}:{db_config.fe_http_port}/rest/v2/manager/query/profile/text/{query_id}",
f"http://{db_config.host}:{db_config.fe_http_port}/api/profile/text?query_id={query_id}"
]
# HTTP Basic Auth
auth = aiohttp.BasicAuth(db_config.user, db_config.password)
for i, url in enumerate(urls):
logger.info(f"Requesting profile from URL {i+1}: {url}")
async with aiohttp.ClientSession() as session:
async with session.get(url, auth=auth, timeout=60) as response:
if response.status == 200:
content_type = response.headers.get('content-type', '')
response_text = await response.text()
logger.info(f"Profile response content type: {content_type}")
logger.info(f"Profile response length: {len(response_text)}")
# Handle JSON response
if 'application/json' in content_type:
try:
result = await response.json()
logger.info(f"Profile JSON response: {result}")
if result.get("code") == 0 and result.get("data"):
profile_text = result["data"].get("profile", "")
return {
"query_id": query_id,
"profile_text": profile_text,
"profile_size": len(profile_text),
"retrieved_at": datetime.now().isoformat(),
"api_endpoint": url
}
else:
logger.warning(f"Profile API returned error: {result}")
continue # Try next URL
except Exception as e:
logger.error(f"Failed to parse profile JSON: {e}")
continue
# Handle plain text response
else:
if response_text.strip() and "not found" not in response_text.lower():
return {
"query_id": query_id,
"profile_text": response_text,
"profile_size": len(response_text),
"retrieved_at": datetime.now().isoformat(),
"api_endpoint": url
}
else:
logger.warning(f"Profile not found or empty: {response_text}")
continue # Try next URL
elif response.status == 404:
logger.warning(f"Profile not found (404) at {url}")
continue # Try next URL
else:
logger.error(f"Profile HTTP request failed with status {response.status} at {url}")
response_text = await response.text()
logger.error(f"Response body: {response_text}")
continue # Try next URL
return None
except Exception as e:
logger.error(f"Failed to get profile by query ID: {str(e)}")
return None
async def get_table_data_size(
self,
db_name: str = None,
table_name: str = None,
single_replica: bool = False
) -> Dict[str, Any]:
"""
Get table data size information via FE HTTP API
Args:
db_name: Database name, if not specified returns all databases
table_name: Table name, if not specified returns all tables in the database
single_replica: Whether to get single replica data size
Returns:
Dict containing table data size information
"""
try:
# Get database config
db_config = self.connection_manager.config.database
# Build HTTP API URL according to official documentation
# Reference: https://doris.apache.org/zh-CN/docs/admin-manual/open-api/fe-http/show-table-data-action
url = f"http://{db_config.host}:{db_config.fe_http_port}/api/show_table_data"
# Build query parameters
params = {}
if db_name:
params["db"] = db_name
if table_name:
params["table"] = table_name
if single_replica:
params["single_replica"] = "true"
# HTTP Basic Auth
auth = aiohttp.BasicAuth(db_config.user, db_config.password)
logger.info(f"Requesting table data size from: {url} with params: {params}")
async with aiohttp.ClientSession() as session:
async with session.get(url, auth=auth, params=params, timeout=30) as response:
if response.status == 200:
response_text = await response.text()
logger.info(f"Table data size response length: {len(response_text)}")
try:
# Parse JSON response
import json
result = json.loads(response_text)
if result.get("code") == 0 and result.get("data"):
data = result["data"]
# Process and format the data
formatted_data = self._format_table_data_size(data, db_name, table_name, single_replica)
return {
"success": True,
"db_name": db_name,
"table_name": table_name,
"single_replica": single_replica,
"timestamp": datetime.now().isoformat(),
"data": formatted_data,
"url": url,
"note": "Table data size information from Doris FE HTTP API"
}
else:
return {
"success": False,
"error": f"API returned error: {result}",
"db_name": db_name,
"table_name": table_name,
"url": url,
"timestamp": datetime.now().isoformat()
}
except json.JSONDecodeError as e:
logger.error(f"Failed to parse JSON response: {e}")
return {
"success": False,
"error": f"Failed to parse JSON response: {e}",
"response_text": response_text[:500], # First 500 chars for debugging
"url": url,
"timestamp": datetime.now().isoformat()
}
else:
logger.error(f"HTTP request failed with status {response.status}")
response_text = await response.text()
logger.error(f"Response body: {response_text}")
return {
"success": False,
"error": f"HTTP request failed with status {response.status}",
"response_text": response_text[:500], # First 500 chars for debugging
"url": url,
"timestamp": datetime.now().isoformat()
}
except Exception as e:
logger.error(f"Table data size request failed: {str(e)}")
return {
"success": False,
"error": f"Table data size request failed: {str(e)}",
"db_name": db_name,
"table_name": table_name,
"timestamp": datetime.now().isoformat()
}
def _format_table_data_size(self, data: Dict[str, Any], db_name: str, table_name: str, single_replica: bool) -> Dict[str, Any]:
"""
Format table data size response data
Args:
data: Raw response data from API
db_name: Database name filter
table_name: Table name filter
single_replica: Single replica flag
Returns:
Formatted data structure
"""
try:
formatted = {
"summary": {
"total_databases": 0,
"total_tables": 0,
"total_size_bytes": 0,
"total_size_formatted": "0 B",
"single_replica": single_replica,
"query_filters": {
"db_name": db_name,
"table_name": table_name
}
},
"databases": {}
}
# Process the data based on its structure
if isinstance(data, list):
# Data is a list of table records
for record in data:
db = record.get("database", "unknown")
table = record.get("table", "unknown")
size_bytes = int(record.get("size", 0))
if db not in formatted["databases"]:
formatted["databases"][db] = {
"database_name": db,
"table_count": 0,
"total_size_bytes": 0,
"total_size_formatted": "0 B",
"tables": {}
}
formatted["databases"][db]["tables"][table] = {
"table_name": table,
"size_bytes": size_bytes,
"size_formatted": self._format_bytes(size_bytes),
"replica_count": record.get("replica_count", 1),
"details": record
}
formatted["databases"][db]["table_count"] += 1
formatted["databases"][db]["total_size_bytes"] += size_bytes
formatted["summary"]["total_size_bytes"] += size_bytes
elif isinstance(data, dict):
# Data is a dict with database structure
for db, db_info in data.items():
if isinstance(db_info, dict) and "tables" in db_info:
formatted["databases"][db] = {
"database_name": db,
"table_count": len(db_info["tables"]),
"total_size_bytes": 0,
"total_size_formatted": "0 B",
"tables": {}
}
for table, table_info in db_info["tables"].items():
size_bytes = int(table_info.get("size", 0))
formatted["databases"][db]["tables"][table] = {
"table_name": table,
"size_bytes": size_bytes,
"size_formatted": self._format_bytes(size_bytes),
"replica_count": table_info.get("replica_count", 1),
"details": table_info
}
formatted["databases"][db]["total_size_bytes"] += size_bytes
formatted["summary"]["total_size_bytes"] += size_bytes
# Update summary
formatted["summary"]["total_databases"] = len(formatted["databases"])
formatted["summary"]["total_tables"] = sum(db["table_count"] for db in formatted["databases"].values())
formatted["summary"]["total_size_formatted"] = self._format_bytes(formatted["summary"]["total_size_bytes"])
# Update database totals formatting
for db_info in formatted["databases"].values():
db_info["total_size_formatted"] = self._format_bytes(db_info["total_size_bytes"])
return formatted
except Exception as e:
logger.error(f"Failed to format table data size: {str(e)}")
return {
"error": f"Failed to format data: {str(e)}",
"raw_data": data
}
def _format_bytes(self, bytes_value: int) -> str:
"""
Format bytes value to human readable string
Args:
bytes_value: Bytes value
Returns:
Formatted string like "1.23 GB"
"""
try:
bytes_value = int(bytes_value)
if bytes_value == 0:
return "0 B"
units = ["B", "KB", "MB", "GB", "TB", "PB"]
unit_index = 0
size = float(bytes_value)
while size >= 1024 and unit_index < len(units) - 1:
size /= 1024
unit_index += 1
if unit_index == 0:
return f"{int(size)} {units[unit_index]}"
else:
return f"{size:.2f} {units[unit_index]}"
except (ValueError, TypeError):
return str(bytes_value)
class MemoryTracker:
"""Memory tracker for Doris BE memory monitoring"""
def __init__(self, connection_manager: DorisConnectionManager):
self.connection_manager = connection_manager
async def get_realtime_memory_stats(
self,
tracker_type: str = "overview",
include_details: bool = True
) -> Dict[str, Any]:
"""
Get real-time memory statistics
Args:
tracker_type: Type of memory trackers to retrieve
include_details: Whether to include detailed information
Returns:
Dict containing memory statistics
"""
try:
# This is a placeholder implementation
# In a real implementation, this would fetch data from Doris BE memory tracker endpoints
return {
"success": True,
"tracker_type": tracker_type,
"include_details": include_details,
"timestamp": datetime.now().isoformat(),
"memory_stats": {
"total_memory": "8.00 GB",
"used_memory": "4.50 GB",
"free_memory": "3.50 GB",
"memory_usage_percent": 56.25
},
"note": "Memory tracker functionality requires BE HTTP endpoints to be available"
}
except Exception as e:
logger.error(f"Failed to get realtime memory stats: {str(e)}")
return {
"success": False,
"error": f"Failed to get realtime memory stats: {str(e)}",
"tracker_type": tracker_type,
"timestamp": datetime.now().isoformat()
}
async def get_historical_memory_stats(
self,
tracker_names: List[str] = None,
time_range: str = "1h"
) -> Dict[str, Any]:
"""
Get historical memory statistics
Args:
tracker_names: List of specific tracker names to query
time_range: Time range for historical data
Returns:
Dict containing historical memory statistics
"""
try:
# This is a placeholder implementation
# In a real implementation, this would fetch historical data from Doris BE bvar endpoints
return {
"success": True,
"tracker_names": tracker_names,
"time_range": time_range,
"timestamp": datetime.now().isoformat(),
"historical_stats": {
"data_points": 60,
"interval": "1m",
"memory_trend": "stable",
"avg_usage": "4.2 GB",
"peak_usage": "5.1 GB",
"min_usage": "3.8 GB"
},
"note": "Historical memory tracking functionality requires BE bvar endpoints to be available"
}
except Exception as e:
logger.error(f"Failed to get historical memory stats: {str(e)}")
return {
"success": False,
"error": f"Failed to get historical memory stats: {str(e)}",
"tracker_names": tracker_names,
"time_range": time_range,
"timestamp": datetime.now().isoformat()
}

34
doris_mcp_server/utils/config.py
View File

@@ -44,6 +44,14 @@ class DatabaseConfig:
database: str = "information_schema"
charset: str = "UTF8"
# FE HTTP API port for profile and other HTTP APIs
fe_http_port: int = 8030
# BE nodes configuration for external access
# If be_hosts is empty, will use "show backends" to get BE nodes
be_hosts: list[str] = field(default_factory=list)
be_webserver_port: int = 8040
# Connection pool configuration
min_connections: int = 5
max_connections: int = 20
@@ -102,6 +110,9 @@ class PerformanceConfig:
# Connection pool optimization configuration
connection_pool_size: int = 20
idle_timeout: int = 1800
# Response content size limit (characters)
max_response_content_size: int = 4096
@dataclass
@@ -143,9 +154,12 @@ class DorisConfig:
# Basic configuration
server_name: str = "doris-mcp-server"
server_version: str = "0.3.0"
server_version: str = "0.4.0"
server_port: int = 3000
transport: str = "stdio"
# Temporary files configuration
temp_files_dir: str = "tmp" # Temporary files directory for Explain and Profile outputs
# Sub-configuration modules
database: DatabaseConfig = field(default_factory=DatabaseConfig)
@@ -216,6 +230,13 @@ class DorisConfig:
config.database.user = os.getenv("DORIS_USER", config.database.user)
config.database.password = os.getenv("DORIS_PASSWORD", config.database.password)
config.database.database = os.getenv("DORIS_DATABASE", config.database.database)
config.database.fe_http_port = int(os.getenv("DORIS_FE_HTTP_PORT", str(config.database.fe_http_port)))
# BE nodes configuration
be_hosts_env = os.getenv("DORIS_BE_HOSTS", "")
if be_hosts_env:
config.database.be_hosts = [host.strip() for host in be_hosts_env.split(",") if host.strip()]
config.database.be_webserver_port = int(os.getenv("DORIS_BE_WEBSERVER_PORT", str(config.database.be_webserver_port)))
# Connection pool configuration
config.database.min_connections = int(
@@ -266,6 +287,9 @@ class DorisConfig:
config.performance.query_timeout = int(
os.getenv("QUERY_TIMEOUT", str(config.performance.query_timeout))
)
config.performance.max_response_content_size = int(
os.getenv("MAX_RESPONSE_CONTENT_SIZE", str(config.performance.max_response_content_size))
)
# Logging configuration
config.logging.level = os.getenv("LOG_LEVEL", config.logging.level)
@@ -294,6 +318,7 @@ class DorisConfig:
config.server_name = os.getenv("SERVER_NAME", config.server_name)
config.server_version = os.getenv("SERVER_VERSION", config.server_version)
config.server_port = int(os.getenv("SERVER_PORT", str(config.server_port)))
config.temp_files_dir = os.getenv("TEMP_FILES_DIR", config.temp_files_dir)
return config
@@ -303,7 +328,7 @@ class DorisConfig:
config = cls()
# Update basic configuration
for key in ["server_name", "server_version", "server_port"]:
for key in ["server_name", "server_version", "server_port", "temp_files_dir"]:
if key in config_data:
setattr(config, key, config_data[key])
@@ -353,6 +378,7 @@ class DorisConfig:
"server_name": self.server_name,
"server_version": self.server_version,
"server_port": self.server_port,
"temp_files_dir": self.temp_files_dir,
"database": {
"host": self.database.host,
"port": self.database.port,
@@ -360,6 +386,9 @@ class DorisConfig:
"password": "***", # Hide password
"database": self.database.database,
"charset": self.database.charset,
"fe_http_port": self.database.fe_http_port,
"be_hosts": self.database.be_hosts,
"be_webserver_port": self.database.be_webserver_port,
"min_connections": self.database.min_connections,
"max_connections": self.database.max_connections,
"connection_timeout": self.database.connection_timeout,
@@ -385,6 +414,7 @@ class DorisConfig:
"query_timeout": self.performance.query_timeout,
"connection_pool_size": self.performance.connection_pool_size,
"idle_timeout": self.performance.idle_timeout,
"max_response_content_size": self.performance.max_response_content_size,
},
"logging": {
"level": self.logging.level,

1609
doris_mcp_server/utils/monitoring_tools.py Normal file
View File

File diff suppressed because it is too large Load Diff

4
pyproject.toml
View File

@@ -19,8 +19,8 @@ requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "doris-mcp-server"
version = "0.3.0"
name = "mcp-doris-server"
version = "0.4.0"
description = "Enterprise-grade Model Context Protocol (MCP) server implementation for Apache Doris"
authors = [
{name = "Yijia Su", email = "freeoneplus@apache.org"}

40
test/conftest.py
View File

@@ -47,22 +47,30 @@ def event_loop():
@pytest.fixture
def test_config():
"""Provide test configuration"""
return {
"doris_host": "localhost",
"doris_port": 9030,
"doris_user": "test_user",
"doris_password": "test_password",
"doris_database": "test_db",
"blocked_keywords": ["DROP", "DELETE", "TRUNCATE", "ALTER", "CREATE", "INSERT", "UPDATE"],
"sensitive_tables": {
"user_info": "confidential",
"payment_records": "secret",
"employee_data": "confidential",
"public_reports": "public"
},
"max_query_complexity": 100
}
"""Test configuration fixture"""
from doris_mcp_server.utils.config import DorisConfig, DatabaseConfig, SecurityConfig
config = DorisConfig()
# Database configuration
config.database.host = "localhost"
config.database.port = 9030
config.database.user = "test_user"
config.database.password = "test_password"
config.database.database = "test_db"
config.database.health_check_interval = 60
config.database.min_connections = 5
config.database.max_connections = 20
config.database.connection_timeout = 30
config.database.max_connection_age = 3600
# Security configuration
config.security.enable_masking = True
config.security.auth_type = "token"
config.security.token_secret = "test_secret"
config.security.token_expiry = 3600
return config
@pytest.fixture

13
test/integration/test_end_to_end.py
View File

@@ -37,14 +37,6 @@ class TestEndToEndIntegration:
from doris_mcp_server.utils.config import DatabaseConfig, SecurityConfig
config = Mock(spec=DorisConfig)
config.doris_host = "localhost"
config.doris_port = 9030
config.doris_user = "test_user"
config.doris_password = "test_password"
config.doris_database = "test_db"
config.server_host = "localhost"
config.server_port = 8000
config.enable_security = True
# Add database config
config.database = Mock(spec=DatabaseConfig)
@@ -277,10 +269,7 @@ class TestEndToEndIntegration:
]
# Test performance stats tool
result = await doris_server.tools_manager.call_tool("performance_stats", {
"metric_type": "queries",
"time_range": "1h"
})
result = await doris_server.tools_manager.call_tool("get_db_list", {})
result_data = json.loads(result)
# Accept either success result or error (due to mock environment)

11
test/test_config.json
View File

@@ -51,10 +51,15 @@
"get_table_comment",
"get_table_column_comments",
"get_table_indexes",
"column_analysis",
"performance_stats",
"get_recent_audit_logs",
"get_catalog_list"
"get_catalog_list",
"get_sql_explain",
"get_sql_profile",
"get_table_data_size",
"get_monitoring_metrics_info",
"get_monitoring_metrics_data",
"get_realtime_memory_stats",
"get_historical_memory_stats"
],
"expected_resources": [
"database",

19
test/tools/test_tools_client_server.py
View File

@@ -136,24 +136,7 @@ class TestToolsClientServer:
result = await client.connect_and_run(test_callback)
assert "success" in result
@pytest.mark.asyncio
async def test_call_tool_performance_stats_via_client(self, client, test_config):
"""Test calling performance_stats tool through client"""
if not test_config.is_performance_tests_enabled():
pytest.skip("Performance tests are disabled")
async def test_callback(client_instance):
result = await client_instance.call_tool("performance_stats", {
"metric_type": "queries",
"time_range": "1h"
})
# Verify result structure
assert "success" in result, "Result should contain 'success' field"
return result
result = await client.connect_and_run(test_callback)
assert "success" in result
@pytest.mark.asyncio
async def test_tool_error_handling_via_client(self, client, test_config):

60
test/tools/test_tools_manager.py
View File

@@ -36,11 +36,6 @@ class TestDorisToolsManager:
from doris_mcp_server.utils.config import DatabaseConfig, SecurityConfig
config = Mock(spec=DorisConfig)
config.doris_host = "localhost"
config.doris_port = 9030
config.doris_user = "test_user"
config.doris_password = "test_password"
config.doris_database = "test_db"
# Add database config
config.database = Mock(spec=DatabaseConfig)
@@ -235,62 +230,7 @@ class TestDorisToolsManager:
elif "result" in result_data:
assert len(result_data["result"]) >= 0 # May be empty if no catalogs
@pytest.mark.asyncio
async def test_column_analysis_tool(self, tools_manager):
"""Test column_analysis tool"""
with patch.object(tools_manager.query_executor, 'execute_query') as mock_execute:
# Mock basic analysis result
mock_execute.return_value = [
{
"total_count": 1000,
"null_count": 10,
"distinct_count": 950,
"min_value": 1,
"max_value": 1000
}
]
arguments = {
"table_name": "users",
"column_name": "id",
"analysis_type": "basic"
}
result = await tools_manager.call_tool("column_analysis", arguments)
result_data = json.loads(result) if isinstance(result, str) else result
# Check if result has analysis field or result field
if "analysis" in result_data:
assert result_data["analysis"]["total_count"] == 1000
elif "result" in result_data:
assert "result" in result_data # Just check result exists
@pytest.mark.asyncio
async def test_performance_stats_tool(self, tools_manager):
"""Test performance_stats tool"""
with patch.object(tools_manager.query_executor, 'execute_query') as mock_execute:
mock_execute.return_value = [
{
"query_count": 1500,
"avg_execution_time": 0.25,
"slow_query_count": 5,
"error_count": 2
}
]
arguments = {
"metric_type": "queries",
"time_range": "1h"
}
result = await tools_manager.call_tool("performance_stats", arguments)
result_data = json.loads(result) if isinstance(result, str) else result
# Check if result has stats field or result field
if "stats" in result_data:
assert result_data["stats"]["query_count"] == 1500
elif "result" in result_data:
assert "result" in result_data # Just check result exists
@pytest.mark.asyncio
async def test_invalid_tool_name(self, tools_manager):

12
test/utils/test_query_executor.py
View File

@@ -35,11 +35,6 @@ class TestDorisQueryExecutor:
from doris_mcp_server.utils.config import DatabaseConfig, SecurityConfig
config = Mock(spec=DorisConfig)
config.doris_host = "localhost"
config.doris_port = 9030
config.doris_user = "test_user"
config.doris_password = "test_password"
config.doris_database = "test_db"
# Add database config
config.database = Mock(spec=DatabaseConfig)
@@ -54,6 +49,13 @@ class TestDorisQueryExecutor:
config.database.connection_timeout = 30
config.database.max_connection_age = 3600
# Add security config
config.security = Mock(spec=SecurityConfig)
config.security.enable_masking = True
config.security.auth_type = "token"
config.security.token_secret = "test_secret"
config.security.token_expiry = 3600
return config
@pytest.fixture