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)。
  • （可选）标准输入输出：可以通过标准输入输出进行交互 (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. 克隆仓库

bash

如果不同，请替换为实际的仓库 URL

git clone https://github.com/apache/doris-mcp-server.git
cd doris-mcp-server

2. 安装依赖

bash
pip install -r requirements.txt

3. 配置环境变量

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

bash
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 模式，请执行以下命令：

bash
./start_server.sh

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

服务端点：

• SSE 初始化: http://<host>:<port>/sse
• SSE 通信: http://<host>:<port>/mcp/messages (POST)
• Streamable 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）来调用服务器上的工具。

主要交互流程：

1. 客户端初始化：连接到 /sse (SSE) 或向 /mcp 发送 initialize 方法调用 (Streamable)。

2. （可选）发现工具：客户端可以调用 mcp/listTools 或 mcp/listOfferings 来获取支持的工具列表、其描述和参数模式。

3. 调用工具：客户端发送 tool_call 消息/请求，指定 tool_name 和 arguments。

   • 示例：获取表结构
     • tool_name: mcp_doris_get_table_schema (或特定模式的名称)* arguments: 包含 random_string, table_name, db_name。

4. 处理响应:

   • 非流式: 客户端接收包含 result 或 error 的响应。
   • 流式: 客户端先接收一系列 tools/progress 通知，然后是包含 result 或 error 的最终响应。

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

使用 Cursor 连接

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

Stdio 模式

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

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

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

1. 配置 Cursor: 在你的 Cursor MCP 配置中添加如下条目：

   json
   {
   "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"
   }
   },
   // ... 其他服务器配置 ...
   }
   }

2. 要点:

   • 将 /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 会将这些传递给服务器进程。在这种模式下，不需要 .env 文件。

SSE 模式

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

1. 配置 .env: 确保你的数据库凭据和其他必要设置（如 SERVER_PORT 如果不使用默认的 3000）在项目目录中的 .env 文件中正确配置。

2. 启动服务器: 在项目的根目录下从终端运行服务器：
   bash
   ./start_server.sh

   该脚本通常读取 .env 文件并以 SSE 模式启动 FastAPI 服务器（检查脚本和 sse_server.py / main.py 获取具体信息）。注意服务器监听的主机和端口（默认为 0.0.0.0:3000）。

3. 配置 Cursor: 在你的 Cursor MCP 配置中添加如下条目，指向正在运行的服务器的 SSE 端点：

   json
   {
   "mcpServers": {
   "doris-sse": {
   "url": "http://127.0.0.1:3000/sse" // 如果你的服务器运行在其他地方，请调整主机/端口
   },
   // ... 其他服务器配置 ...
   }
   }

   注意：示例使用默认端口 3000。如果您的服务器配置为在不同端口上运行（例如用户示例中的 3010），请相应地调整 URL。

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

目录结构

doris-mcp-server/
├── doris_mcp_server/ # MCP 服务器的源代码
│ ├── main.py # 主入口点，FastAPI 应用程序定义
│ ├── mcp_core.py # 核心 MCP 工具注册和 Stdio 处理
│ ├── sse_server.py # SSE 服务器实现
│ ├── streamable_server.py # 可流式 HTTP 服务器实现
│ ├── config.py # 配置加载
│ ├── tools/ # MCP 工具定义
│ │ ├── mcp_doris_tools.py # 主要 Doris 相关的 MCP 工具
│ │ ├── tool_initializer.py # 工具注册助手（由 mcp_core.py 使用）
│ │ └── init.py
│ ├── utils/ # 实用类和辅助函数
│ │ ├── db.py # 数据库连接和操作
│ │ ├── logger.py # 日志配置
│ │ ├── schema_extractor.py # Doris 元数据/架构提取逻辑
│ │ ├── sql_executor_tools.py # SQL 执行帮助（可能是遗留的）
│ │ └── init.py
│ └── init.py
├── logs/ # 日志文件目录（如果启用了文件日志记录）
├── README.md # 此文件
├── .env.example # 示例环境变量文件
├── requirements.txt # Python 依赖项，用于 pip
├── pyproject.toml # 项目元数据和构建系统配置（PEP 518）
├── uv.lock # 用于 uv 包管理器的锁文件（pip 的替代品）
├── start_server.sh # 启动服务器的脚本
└── restart_server.sh # 重启服务器的脚本## 开发新工具

本节概述了在Doris MCP Server中添加新的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。

python

在 doris_mcp_server/tools/mcp_doris_tools.py 中

import datetime

... 其他导入 ...

from doris_mcp_server.tools.mcp_doris_tools import _format_response # 重用格式化程序

... 现有工具 ...

async def mcp_doris_get_server_time() -> Dict[str, Any]:
"""获取当前服务器时间。"""
logger.info(f"MCP 工具调用: mcp_doris_get_server_time")
try:
current_time = datetime.datetime.now().isoformat()
# 使用现有的格式化程序以保持一致性
return _format_response(success=True, result={"server_time": current_time})
except Exception as e:
logger.error(f"MCP 工具执行失败 mcp_doris_get_server_time: {str(e)}", exc_info=True)
return _format_response(success=False, error=str(e), message="获取服务器时间时出错")

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

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

A. SSE/Streamable 注册 (tool_initializer.py)

• 从mcp_doris_tools.py导入您的新工具函数。
• 在register_mcp_tools函数内部，添加一个新的装饰有@mcp.tool()的包装函数。
• 包装函数应调用您的核心工具函数。
• 在装饰器中定义工具名称并提供详细的描述（如果有的话包括参数）。即使您的包装器不显式使用它，也请记得包含强制性的random_string参数描述以保证客户端兼容性。

示例 (tool_initializer.py):

python

在 doris_mcp_server/tools/tool_initializer.py 中

... 其他导入 ...

from doris_mcp_server.tools.mcp_doris_tools import (
# ... 现有工具导入 ...
mcp_doris_get_server_time # <-- 导入新工具
)

async def register_mcp_tools(mcp):
# ... 现有工具注册 ...

# 注册工具: 获取服务器时间
@mcp.tool("get_server_time", description="""[函数描述]: 获取MCP服务器的当前时间。

[参数内容]：

• random_string (字符串) [必需] - 工具调用的唯一标识符
  """)
  async def get_server_time_tool() -> Dict[str, Any]:
  """包装器: 获取服务器时间"""
  # 注意：这里不需要为核心函数调用传递参数
  return await mcp_doris_get_server_time()

  ... 注册计数日志 ...B. 标准输入输出注册 (mcp_core.py)

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

示例 (mcp_core.py)：

python

在 doris_mcp_server/mcp_core.py 中

... 其他导入和设置 ...

... 现有的标准输入输出工具注册 ...

注册工具：获取服务器时间（针对标准输入输出）

@stdio_mcp.tool("get_server_time", description="""[函数描述]：获取MCP服务器的当前时间。

[参数内容]：

• random_string (字符串) [必需] - 工具调用的唯一标识符
  """)
  async def get_server_time_tool_stdio() -> Dict[str, Any]: # 如果需要，可以使用略有不同的包装器名称以提高清晰度
  """包装器：获取服务器时间（标准输入输出）"""
  from doris_mcp_server.tools.mcp_doris_tools import mcp_doris_get_server_time # <-- 延迟导入

  假设标准输入输出运行程序正确处理了异步包装器

  return await mcp_doris_get_server_time()

--- 注册工具 --- （或在任何最终确定注册的地方）

4. 重启并测试

在两个文件中实现并注册工具后，重启MCP服务器（通过./start_server.sh启动SSE模式，并根据需要更新Cursor使用的标准输入输出命令），然后使用您的MCP客户端（如Cursor）在两种连接模式下测试新工具。

贡献

欢迎通过Issues或Pull Requests贡献代码。

许可证

本项目采用Apache 2.0许可证。详情请参阅LICENSE文件（如果存在的话）。