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 ）。

系统要求

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 ... } }
注意：该示例使用默认端口3000 。如果您的服务器配置为在其他端口上运行（例如用户示例中的3010 ），请相应地调整 URL。

在 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 （ MetadataExtractor类） ：提供检索数据库元数据的高级方法，例如列出数据库/表（ 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 （ execute_sql_query函数） ：提供对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 注册（ tool_initializer.py ）

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

示例（ tool_initializer.py ）：

# 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 注册（ mcp_core.py ）

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

示例（ mcp_core.py ）：

# 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

Integrations