Doris MCP 服务器

Doris MCP（模型控制面板）服务器是一个使用 Python 和 FastAPI 构建的后端服务。它实现了 MCP（模型控制面板）协议，允许客户端通过定义的“工具”与其交互。它主要用于连接 Apache Doris 数据库，并可能利用大型语言模型 (LLM) 执行诸如将自然语言查询转换为 SQL（NL2SQL）、执行查询以及进行元数据管理和分析等任务。

核心功能

MCP协议实现：提供标准MCP接口，支持工具调用、资源管理、以及快捷交互。
多种通讯模式：
- SSE（服务器发送事件） ：通过/sse （初始化）和/mcp/messages （通信）端点提供服务（ src/sse_server.py ）。
- 可流式传输的 HTTP ：通过统一的/mcp端点提供服务，支持请求/响应和流式传输（ src/streamable_server.py ）。
- （可选）Stdio ：可以通过标准输入/输出（ src/stdio_server.py ）进行交互，需要特定的启动配置。
基于工具的接口：核心功能被封装为 MCP 工具，客户端可以根据需要调用。目前可用的主要工具主要侧重于直接与数据库交互：
- SQL 执行（ mcp_doris_exec_query ）
- 数据库和表列表（ mcp_doris_get_db_list ， mcp_doris_get_db_table_list ）
- 元数据检索（ mcp_doris_get_table_schema 、 mcp_doris_get_table_comment 、 mcp_doris_get_table_column_comments 、 mcp_doris_get_table_indexes ）
- 审计日志检索（ mcp_doris_get_recent_audit_logs ）注意：当前工具主要关注直接数据库操作。
数据库交互：提供连接 Apache Doris（或其他兼容数据库）并执行查询的功能（ src/utils/db.py ）。
灵活的配置：通过.env文件配置，支持数据库连接、LLM 提供程序/模型、API 密钥、日志级别等设置。
元数据提取：能够提取数据库元数据信息（ src/utils/schema_extractor.py ）。

Related MCP server: Superset MCP Server

系统要求

Python 3.12+
数据库连接详细信息（例如 Doris 主机、端口、用户、密码、数据库）

快速入门

1.克隆存储库

# Replace with the actual repository URL if different git clone https://github.com/apache/doris-mcp-server.git cd doris-mcp-server

2.安装依赖项

pip install -r requirements.txt

3.配置环境变量

将.env.example文件复制到.env并根据你的环境修改设置：

cp env.example .env

关键环境变量：

数据库连接：
- DB_HOST ：数据库主机名
- DB_PORT ：数据库端口（默认 9030）
- DB_USER ：数据库用户名
- DB_PASSWORD ：数据库密码
- DB_DATABASE ：默认数据库名称
服务器配置：
- SERVER_HOST ：服务器监听的主机地址（默认0.0.0.0 ）
- SERVER_PORT ：服务器监听的端口（默认3000 ）
- ALLOWED_ORIGINS ：CORS 允许的来源（以逗号分隔， *允许所有）
- MCP_ALLOW_CREDENTIALS ：是否允许 CORS 凭据（默认false ）
日志配置：
- LOG_DIR ：日志文件目录（默认./logs ）
- LOG_LEVEL ：日志级别（例如， INFO ， DEBUG ， WARNING ， ERROR ，默认INFO ）
- CONSOLE_LOGGING ：是否将日志输出到控制台（默认false ）

可用的 MCP 工具

下表列出了当前可通过 MCP 客户端调用的主要工具：

工具名称	描述	参数	地位
`mcp_doris_get_db_list`	获取服务器上所有数据库名称的列表。	`random_string` （字符串，必需）	✅ 活跃
`mcp_doris_get_db_table_list`	获取指定数据库中所有表名的列表。	`random_string` （字符串，必需）， `db_name` （字符串，可选，默认为当前数据库）	✅ 活跃
`mcp_doris_get_table_schema`	获取指定表的详细结构。	`random_string` （字符串，必需）、 `table_name` （字符串，必需）、 `db_name` （字符串，可选）	✅ 活跃
`mcp_doris_get_table_comment`	获取指定表的注释。	`random_string` （字符串，必需）、 `table_name` （字符串，必需）、 `db_name` （字符串，可选）	✅ 活跃
`mcp_doris_get_table_column_comments`	获取指定表中所有列的注释。	`random_string` （字符串，必需）、 `table_name` （字符串，必需）、 `db_name` （字符串，可选）	✅ 活跃
`mcp_doris_get_table_indexes`	获取指定表的索引信息。	`random_string` （字符串，必需）、 `table_name` （字符串，必需）、 `db_name` （字符串，可选）	✅ 活跃
`mcp_doris_exec_query`	执行SQL查询并返回结果命令。	`random_string` （字符串，必需）、 `sql` （字符串，必需）、 `db_name` （字符串，可选）、 `max_rows` （整数，可选，默认 100）、 `timeout` （整数，可选，默认 30）	✅ 活跃
`mcp_doris_get_recent_audit_logs`	获取最近一段时间的审计日志记录。	`random_string` （字符串，必需）、 `days` （整数，可选，默认 7）、 `limit` （整数，可选，默认 100）	✅ 活跃

**注意：**所有工具都需要一个random_string参数作为调用标识符，通常由 MCP 客户端自动处理。“可选”和“必需”指的是工具的内部逻辑；客户端可能需要根据其实现为所有参数提供值。此处列出的工具名称是基本名称；客户端可能会根据连接模式看到它们的前缀（例如mcp_doris_stdio3_get_db_list ）。

4.运行服务

如果使用SSE模式，执行以下命令：

./start_server.sh

此命令启动 FastAPI 应用程序，默认提供 SSE 和 Streamable HTTP MCP 服务。

服务端点：

SSE 初始化： http://<host>:<port>/sse
SSE 通信： http://<host>:<port>/mcp/messages (POST)
可流式传输的 HTTP ： http://<host>:<port>/mcp （支持 GET、POST、DELETE、OPTIONS）
健康检查： http://<host>:<port>/health
（潜在）状态检查： http://<host>:<port>/status （确认是否在main.py中实现）

用法

与 Doris MCP 服务器交互需要MCP 客户端。客户端连接到服务器的 SSE 或 Streamable HTTP 端点，并根据 MCP 规范发送请求（例如tool_call ）来调用服务器的工具。

主要交互流程：

客户端初始化：连接到/sse (SSE) 或向/mcp (Streamable) 发送initialize方法调用。
（可选）发现工具：客户端可以调用mcp/listTools或mcp/listOfferings来获取支持的工具列表、它们的描述和参数模式。
调用工具：客户端发送tool_call消息/请求，指定tool_name和arguments 。
- 示例：获取表架构
  - tool_name ： mcp_doris_get_table_schema （或特定模式的名称）
  - arguments ：包括random_string 、 table_name 、 db_name 。
处理响应：
- 非流式：客户端收到包含result或error响应。
- 流式：客户端接收一系列tools/progress通知，然后是包含result或error的最终响应。

具体工具名称和参数应从src/tools/代码中引用或通过 MCP 发现机制获取。

使用光标连接

您可以使用 Stdio 或 SSE 模式将 Cursor 连接到此 MCP 服务器。

标准输入输出模式

Stdio 模式允许 Cursor 直接管理服务器进程。配置在 Cursor 的 MCP 服务器设置文件（通常是~/.cursor/mcp.json或类似文件）中完成。

如果使用stdio模式，请执行以下命令下载并构建环境依赖包，但请注意需要将项目路径改为正确的路径地址：

uv --project /your/path/doris-mcp-server run doris-mcp

**配置 Cursor：**将如下条目添加到您的 Cursor MCP 配置中：
{ "mcpServers": { "doris-stdio": { "command": "uv", "args": ["--project", "/path/to/your/doris-mcp-server", "run", "doris-mcp"], "env": { "DB_HOST": "127.0.0.1", "DB_PORT": "9030", "DB_USER": "root", "DB_PASSWORD": "your_db_password", "DB_DATABASE": "your_default_db" } }, // ... other server configurations ... } }
要点：
- 将/path/to/your/doris-mcp替换为系统上项目根目录的实际绝对路径。 --project参数对于uv找到pyproject.toml并运行正确的命令至关重要。
- 该command设置为uv （假设你使用uv进行包管理，如uv.lock所示）。 args包括--project 、路径、 run和mcp-doris （它应该与你在pyproject.toml中定义的脚本相对应）。
- 数据库连接详细信息（ DB_HOST 、 DB_PORT 、 DB_USER 、 DB_PASSWORD 、 DB_DATABASE ）直接在配置文件的env块中设置。Cursor 会将这些信息传递给服务器进程。通过 Cursor 配置时，此模式无需.env文件。

SSE模式

SSE模式要求你先独立运行MCP服务器，然后告诉Cursor如何连接它。

**配置.env ：**确保您的数据库凭据和任何其他必要的设置（如果不使用默认值 3000，则为SERVER_PORT ）在项目目录中的.env文件中正确配置。
**启动服务器：**从项目根目录中的终端运行服务器：
./start_server.sh
此脚本通常会读取.env文件并以 SSE 模式启动 FastAPI 服务器（具体细节请查看脚本和sse_server.py / main.py ）。请记录服务器监听的主机和端口（默认值为0.0.0.0:3000 ）。
**配置 Cursor：**将如下条目添加到您的 Cursor MCP 配置中，指向正在运行的服务器的 SSE 端点：
{ "mcpServers": { "doris-sse": { "url": "http://127.0.0.1:3000/sse" // Adjust host/port if your server runs elsewhere }, // ... other server configurations ... } }
注意：该示例使用默认端口

在 Cursor 中配置任一模式后，您应该能够选择服务器（例如， doris-stdio或doris-sse ）并使用其工具。

目录结构

doris-mcp-server/ ├── doris_mcp_server/ # Source code for the MCP server │ ├── main.py # Main entry point, FastAPI app definition │ ├── mcp_core.py # Core MCP tool registration and Stdio handling │ ├── sse_server.py # SSE server implementation │ ├── streamable_server.py # Streamable HTTP server implementation │ ├── config.py # Configuration loading │ ├── tools/ # MCP tool definitions │ │ ├── mcp_doris_tools.py # Main Doris-related MCP tools │ │ ├── tool_initializer.py # Tool registration helper (used by mcp_core.py) │ │ └── __init__.py │ ├── utils/ # Utility classes and helper functions │ │ ├── db.py # Database connection and operations │ │ ├── logger.py # Logging configuration │ │ ├── schema_extractor.py # Doris metadata/schema extraction logic │ │ ├── sql_executor_tools.py # SQL execution helper (might be legacy) │ │ └── __init__.py │ └── __init__.py ├── logs/ # Log file directory (if file logging enabled) ├── README.md # This file ├── .env.example # Example environment variable file ├── requirements.txt # Python dependencies for pip ├── pyproject.toml # Project metadata and build system configuration (PEP 518) ├── uv.lock # Lock file for 'uv' package manager (alternative to pip) ├── start_server.sh # Script to start the server └── restart_server.sh # Script to restart the server

开发新工具

本节概述了在考虑当前项目结构的情况下向 Doris MCP 服务器添加新 MCP 工具的过程。

1. 利用实用模块

在从头编写新的数据库交互逻辑之前，请检查现有的实用程序模块：

doris_mcp_server/utils/db.py ：提供获取数据库连接（ get_db_connection ）和执行原始查询（ execute_query 、 execute_query_df ）的基本功能。
doris_mcp_server/utils/schema_extractor.py ：提供检索数据库元数据的高级方法，例如列出数据库/表（ get_all_databases 、 get_database_tables ）、获取表模式/注释/索引（ get_table_schema 、 get_table_comment 、 get_column_comments 、 get_table_indexes ）以及访问审计日志（ get_recent_audit_logs ）。它包含缓存机制。
doris_mcp_server/utils/sql_executor_tools.py ：提供对db.execute_query的包装，包含安全检查（可选，由ENABLE_SQL_SECURITY_CHECK环境变量控制）、为 SELECT 查询添加自动LIMIT 、处理结果序列化（日期、小数），并将输出格式化为标准的 MCP 成功/错误结构。建议使用它来执行用户提供或生成的 SQL。

您可以导入并组合这些模块的功能来构建新的工具。

2. 实现工具逻辑

在doris_mcp_server/tools/mcp_doris_tools.py文件中，将新工具的核心逻辑实现为async函数。这样可以使主要工具的实现保持集中化。确保函数返回的数据格式能够轻松封装到标准的 MCP 响应结构中（请参阅同一文件中的_format_response ）。

**示例：**让我们创建一个简单的工具get_server_time 。

# In doris_mcp_server/tools/mcp_doris_tools.py import datetime # ... other imports ... from doris_mcp_server.tools.mcp_doris_tools import _format_response # Reuse formatter # ... existing tools ... async def mcp_doris_get_server_time() -> Dict[str, Any]: """Gets the current server time.""" logger.info(f"MCP Tool Call: mcp_doris_get_server_time") try: current_time = datetime.datetime.now().isoformat() # Use the existing formatter for consistency return _format_response(success=True, result={"server_time": current_time}) except Exception as e: logger.error(f"MCP tool execution failed mcp_doris_get_server_time: {str(e)}", exc_info=True) return _format_response(success=False, error=str(e), message="Error getting server time")

3. 注册工具（双重注册）

由于 SSE/Streamable 和 Stdio 模式的单独处理，您需要在两个地方注册该工具：

A.SSE/Streamable 注册（

从mcp_doris_tools.py导入新的工具功能。
在register_mcp_tools函数内部，添加一个用@mcp.tool()装饰的新包装函数。
包装函数应该调用你的核心工具函数。
在装饰器中定义工具名称并提供详细描述（如有参数，也请包含）。即使您的包装器未明确使用random_string参数描述，也请务必包含该参数描述以确保客户端兼容性。

示例（

# In doris_mcp_server/tools/tool_initializer.py # ... other imports ... from doris_mcp_server.tools.mcp_doris_tools import ( # ... existing tool imports ... mcp_doris_get_server_time # <-- Import the new tool ) async def register_mcp_tools(mcp): # ... existing tool registrations ... # Register Tool: Get Server Time @mcp.tool("get_server_time", description="""[Function Description]: Get the current time of the MCP server.\n [Parameter Content]:\n - random_string (string) [Required] - Unique identifier for the tool call\n""") async def get_server_time_tool() -> Dict[str, Any]: """Wrapper: Get server time""" # Note: No parameters needed for the core function call here return await mcp_doris_get_server_time() # ... logging registration count ...

B. Stdio 注册（

与 SSE 类似，添加一个用@stdio_mcp.tool()装饰的新包装函数。
**重要提示：**在包装函数中导入核心工具函数（ mcp_doris_get_server_time ）（此文件中使用的延迟导入模式）。
包装器调用核心工具函数。包装器本身可能需要async def具体取决于FastMCP在 Stdio 模式下处理工具的方式，即使底层函数很简单（如当前文件结构所示）。请确保调用方式匹配（例如，如果调用异步函数，请使用await ）。

示例（

# In doris_mcp_server/mcp_core.py # ... other imports and setup ... # ... existing Stdio tool registrations ... # Register Tool: Get Server Time (for Stdio) @stdio_mcp.tool("get_server_time", description="""[Function Description]: Get the current time of the MCP server.\n [Parameter Content]:\n - random_string (string) [Required] - Unique identifier for the tool call\n""") async def get_server_time_tool_stdio() -> Dict[str, Any]: # Using a slightly different wrapper name for clarity if needed """Wrapper: Get server time (Stdio)""" from doris_mcp_server.tools.mcp_doris_tools import mcp_doris_get_server_time # <-- Delayed import # Assuming the Stdio runner handles async wrappers correctly return await mcp_doris_get_server_time() # --- Register Tools --- (Or wherever the registrations are finalized)

4.重启并测试

在两个文件中实现并注册该工具后，重新启动 MCP 服务器（通过./start_server.sh启动 SSE 模式，并确保 Cursor 使用的 Stdio 命令在必要时进行更新），并在两种连接模式下使用 MCP 客户端（如 Cursor）测试新工具。

贡献

欢迎通过问题或拉取请求做出贡献。

执照

本项目遵循 Apache 2.0 许可证。详情请参阅 LICENSE 文件（如有）。

Doris MCP Server