YDB MCP

YDB MCP

---

适用于 YDB 的 Model Context Protocol 服务。它允许任何支持 MCP 的 大模型与 YDB 数据库交互。此集成支持由 AI 驱动的数据库操作，以及通过自然语言与 YDB 实例交互。

用法

通过 uvx

uvx 是 uv run tool 的别名，可让你在不显式安装的情况下运行各种 Python 应用。以下是使用 uvx 配置 YDB MCP 的示例。

示例：使用匿名身份验证

{
  "mcpServers": {
    "ydb": {
      "command": "uvx",
      "args": [
        "ydb-mcp",
        "--ydb-endpoint", "grpc://localhost:2136",
        "--ydb-database", "/local"
      ]
    }
  }
}

通过 pipx

pipx 可让你运行各种 PyPI 应用，而无需显式安装每个应用。不过，必须先安装。以下是使用 pipx 配置 YDB MCP 的示例。

示例：使用匿名身份验证

{
  "mcpServers": {
    "ydb": {
      "command": "pipx",
      "args": [
        "run", "ydb-mcp",
        "--ydb-endpoint", "grpc://localhost:2136",
        "--ydb-database", "/local"
      ]
    }
  }
}

通过 pip

YDB MCP 可通过 pip 安装，pip 是 Python 的包安装程序。该软件包可在 PyPI 上获取，并包含所有必需的依赖项。

pip install ydb-mcp

要开始使用 YDB MCP，你需要配置 MCP 客户端，使其与 YDB 实例通信。以下是示例配置文件，你可以根据自己的设置进行自定义，然后将其放入 MCP 客户端的设置中。可能还需要将 Python 解释器路径调整为已安装 ydb-mcp 软件包的正确虚拟环境路径。

示例：使用匿名身份验证

{
  "mcpServers": {
    "ydb": {
      "command": "python3",
      "args": [
        "-m", "ydb_mcp",
        "--ydb-endpoint", "grpc://localhost:2136",
        "--ydb-database", "/local"
      ]
    }
  }
}

身份验证

无论使用哪种方式（uvx、pipx 或 pip），你都可以为 YDB 安装配置身份验证。为此，请传入专用的命令行参数。

使用登录名／密码身份验证

要使用登录名／密码身份验证，请指定 --ydb-auth-mode、--ydb-login 和 --ydb-password 参数：

{
  "mcpServers": {
    "ydb": {
      "command": "uvx",
      "args": [
        "ydb-mcp",
        "--ydb-endpoint", "grpc://localhost:2136",
        "--ydb-database", "/local",
        "--ydb-auth-mode", "login-password",
        "--ydb-login", "<your-username>",
        "--ydb-password", "<your-password>"
      ]
    }
  }
}

使用访问令牌身份验证

要使用访问令牌身份验证，请指定 --ydb-auth-mode 和 --ydb-access-token 参数：

{
  "mcpServers": {
    "ydb": {
      "command": "uvx",
      "args": [
        "ydb-mcp",
        "--ydb-endpoint", "grpc://localhost:2136",
        "--ydb-database", "/local",
        "--ydb-auth-mode", "access-token",
        "--ydb-access-token", "qwerty123"
      ]
    }
  }
}

使用服务账号身份验证

服务账号身份验证需要 yandexcloud 软件包，默认情况下不会安装。请确保在运行 YDB MCP 的环境中提供该软件包：

• uvx：使用 --with yandexcloud 临时添加（放在 ydb-mcp 之前传入）。
• pipx：使用额外的软件包安装 YDB MCP：先运行 pipx install ydb-mcp，再运行 pipx inject ydb-mcp yandexcloud。
• pip：使用 pip install ydb-mcp yandexcloud 将其与 YDB MCP 一起安装。

要使用服务账号身份验证，请指定 --ydb-auth-mode 和 --ydb-sa-key-file 参数：

{
  "mcpServers": {
    "ydb": {
      "command": "uvx",
      "args": [
        "--with", "yandexcloud",
        "ydb-mcp",
        "--ydb-endpoint", "grpc://localhost:2136",
        "--ydb-database", "/local",
        "--ydb-auth-mode", "service-account",
        "--ydb-sa-key-file", "~/sa_key.json"
      ]
    }
  }
}

TLS 连接

使用 grpcs:// 端点通过 TLS 连接。如果集群证书由私有 CA 签发，请通过 --ydb-root-certificates 传入包含根证书的 PEM 文件路径；否则将使用系统信任存储：

{
  "mcpServers": {
    "ydb": {
      "command": "uvx",
      "args": [
        "ydb-mcp",
        "--ydb-endpoint", "grpcs://localhost:2135",
        "--ydb-database", "/local",
        "--ydb-root-certificates", "~/ydb_ca.pem"
      ]
    }
  }
}

编写查询

默认情况下，服务以只读模式运行查询。只读查询使用 YDB 原生的快照只读事务，因此 YDB 本身会拒绝数据和架构写入，不依赖对 SQL 文本进行匹配。此模式同时适用于内置查询工具和调用 YDBMCPServer.execute() 的自定义工具。

如需明确允许写入查询，请设置 --ydb-allow-write：

{
  "mcpServers": {
    "ydb": {
      "command": "uvx",
      "args": [
        "ydb-mcp",
        "--ydb-endpoint", "grpc://localhost:2136",
        "--ydb-database", "/local",
        "--ydb-allow-write"
      ]
    }
  }
}

也可以通过 YDB_ALLOW_WRITE=true 提供相同设置。仅在需要写入时启用此选项；只要服务不需要修改数据，就应优先使用只读 YDB 凭据作为额外的权限边界。

可用工具

YDB MCP 提供以下用于与 YDB 数据库交互的工具：

• ydb_query：对 YDB 数据库运行 SQL 查询

  • 参数：
    • sql：要执行的 SQL 查询字符串

• ydb_query_with_params：使用 JSON 参数运行参数化 SQL 查询

  • 参数：
    • sql：包含参数占位符的 SQL 查询字符串
    • params：包含参数值的 JSON 字符串

• ydb_explain_query：解释 SQL 查询（返回执行计划）

  • 参数：
    • sql：要解释的 SQL 查询字符串

• ydb_explain_query_with_params：解释参数化 SQL 查询

  • 参数：
    • sql：包含参数占位符的 SQL 查询字符串
    • params：包含参数值的 JSON 字符串

• ydb_list_directory：列出 YDB 中的目录内容

  • 参数：
    • path：要列出的 YDB 目录路径

• ydb_describe_path：获取 YDB 路径的详细信息（表、目录等）

  • 参数：
    • path：要描述的 YDB 路径

• ydb_status：获取 YDB 连接的当前状态

构建自定义 MCP 服务

YDBMCPServer 设计为可供子类继承。你可以在已建立的 YDB 连接上添加自己的工具，也可以选择禁用内置通用工具，只公开应用所需的查询。

为什么要构建自定义服务？

• 安全性 — 将大模型限制为一组固定的只读查询，而不是开放任意 SQL 执行。
• 领域针对性 — 为模型提供符合业务逻辑的工具，而非原始数据库操作。
• 简洁性 — 工具越少，模型越不容易产生歧义。

可用方法

在子类中重写或调用以下方法：

方法	描述
await self.execute(sql, params=None)	运行 SQL 查询。返回 list[dict]，每个 dict 都包含 "columns" 和 "rows"。
await self.explain(sql, params=None)	以 dict 形式返回查询执行计划。
await self.list_directory(path)	列出 YDB 目录。返回包含 "path" 和 "items" 的 dict。
await self.describe_path(path)	描述 YDB 路径（表架构、目录等）。返回一个 dict。

params 参数是普通的 dict。不带 $ 前缀的键会自动添加此前缀。如需指定明确的 YDB 类型，请使用 (value, "TypeName") 元组，例如 {"id": (42, "Int64")}。

控制通用工具

使用 generic_tools 类属性控制要注册的内置工具：

值	效果
set(YDBGenericTool)	所有内置工具（默认）
set()	不使用内置工具 — 仅使用你自己的工具
{YDBGenericTool.QUERY, YDBGenericTool.STATUS}	仅使用列出的工具

YDBGenericTool 是字符串枚举 — 可用值：QUERY、QUERY_WITH_PARAMS、EXPLAIN、EXPLAIN_WITH_PARAMS、STATUS、LIST_DIRECTORY、DESCRIBE_PATH。

示例

# my_server.py
from ydb_mcp import YDBMCPServer, YDBGenericTool, serialize_ydb_response


class OrdersServer(YDBMCPServer):
    """Minimal read-only MCP server for the orders service."""

    generic_tools = {YDBGenericTool.STATUS}  # keep status check for diagnostics

    def __init__(self, **kwargs):
        super().__init__(**kwargs)

        @self.tool()
        async def get_order(order_id: str) -> str:
            """Fetch a single order by ID."""
            rows = await self.execute(
                "SELECT * FROM orders WHERE id = $id",
                {"id": order_id},
            )
            return serialize_ydb_response(rows)

        @self.tool()
        async def list_recent_orders(limit: int = 10) -> str:
            """Return the most recent orders."""
            rows = await self.execute(
                "SELECT * FROM orders ORDER BY created_at DESC LIMIT $limit",
                {"limit": limit},
            )
            return serialize_ydb_response(rows)


if __name__ == "__main__":
    OrdersServer(
        endpoint="grpc://localhost:2136",
        database="/local",
    ).run()

直接运行：

python my_server.py

或者在客户端配置中将其配置为 MCP 服务：

{
  "mcpServers": {
    "orders": {
      "command": "python",
      "args": ["my_server.py"]
    }
  }
}

开发

该项目使用 Make 作为主要开发工具，为常见开发任务提供一致的操作接口。

可用的 Make 命令

项目包含一个全面的 Makefile，其中提供了各种开发任务命令。每条命令都旨在简化开发工作流并确保代码质量：

• make all：依次运行 clean、lint 和 test（默认目标）
• make clean：移除所有构建产物和临时文件
• make test：使用 pytest 运行所有测试
  • 可通过环境变量配置：
    • LOG_LEVEL（默认值：WARNING）- 控制测试输出详细程度（DEBUG、INFO、WARNING、ERROR）
• make unit-tests：仅运行单元测试，并输出详细信息
  • 可通过环境变量配置：
    • LOG_LEVEL（默认值：WARNING）- 控制测试输出详细程度（DEBUG、INFO、WARNING、ERROR）
• make integration-tests：仅运行集成测试，并输出详细信息
  • 可通过环境变量配置：
    • YDB_ENDPOINT（默认值：grpc://localhost:2136）
    • YDB_DATABASE（默认值：/local）
    • MCP_HOST（默认值：127.0.0.1）
    • MCP_PORT（默认值：8989）
    • LOG_LEVEL（默认值：WARNING）- 控制测试输出详细程度（DEBUG、INFO、WARNING、ERROR）
• make run-server：启动 YDB MCP 服务
  • 可通过环境变量配置：
    • YDB_ENDPOINT（默认值：grpc://localhost:2136）
    • YDB_DATABASE（默认值：/local）
  • 可使用 ARGS="your args" 传递其他参数
• make lint：运行所有代码检查（flake8、mypy、black、isort）
• make format：使用 black 和 isort 格式化代码
• make install：以开发模式安装软件包
• make dev：以开发模式安装软件包及所有开发依赖项

测试输出详细程度控制

默认情况下，测试以最低输出级别（WARNING）运行，以保持输出简洁。你可以使用 LOG_LEVEL 环境变量控制测试输出的详细程度：

# Run all tests with debug output
make test LOG_LEVEL=DEBUG

# Run integration tests with info output
make integration-tests LOG_LEVEL=INFO

# Run unit tests with warning output (default)
make unit-tests LOG_LEVEL=WARNING

可用的日志级别：

• DEBUG：显示所有调试消息，适用于查看详细的测试流程
• INFO：显示信息级别及以上的消息
• WARNING：仅显示警告和错误（默认）
• ERROR：仅显示错误消息