数据库 MCP 服务器

一个模型上下文协议 (MCP) 服务器，提供连接和交互各种数据库系统的工具。

特性

• 多数据库支持: 连接到 SQLite, PostgreSQL, MySQL/MariaDB, 和 SQL Server 数据库
• 统一接口: 跨所有支持的数据库类型的通用数据库操作工具
• 数据库特定扩展: 在需要时，提供数据库特定功能的专用工具
• 模式管理: 创建、修改和删除表和索引
• 查询执行: 执行原始 SQL 查询或使用结构化查询工具
• 事务支持: 开始、提交和回滚事务

安装

前提条件

• Python 3.8 或更高版本
• 必需的 Python 包 (使用 pip 自动安装):
  • SQLAlchemy
  • 各种数据库驱动程序，具体取决于您要使用的数据库：
    • SQLite (Python 自带)
    • PostgreSQL: psycopg2-binary
    • MySQL/MariaDB: mysql-connector-python
    • SQL Server: pyodbc

从源码安装

# 克隆仓库
git clone <repository-url>

# 安装包
pip install -e .

配置

服务器可以使用环境变量、配置文件或在运行时提供连接详细信息进行配置。

环境变量

• DB_CONFIG_PATH: JSON 配置文件的路径
• DB_CONNECTIONS: 以逗号分隔的连接 ID 列表或包含连接详细信息的 JSON 字符串

配置文件格式

{
  "connections": {
    "sqlite_conn": {
      "type": "sqlite",
      "db_path": "/path/to/database.db"
    },
    "postgres_conn": {
      "type": "postgres",
      "host": "localhost",
      "port": 5432,
      "database": "mydatabase",
      "user": "myuser",
      "password": "mypassword"
    }
  }
}

用法

运行服务器

作为 Claude 的 MCP 服务器

# 使用默认设置运行
python -m db_mcp_server

# 指定配置文件
python -m db_mcp_server --config /path/to/config.json

# 设置日志级别
python -m db_mcp_server --log-level DEBUG

作为独立的 Web 服务器 (适用于任何 LLM)

# 作为 Web 服务器运行
python -m db_mcp_server.web_server

# 指定主机和端口
python -m db_mcp_server.web_server --host 0.0.0.0 --port 8000

# 指定配置文件和日志级别
python -m db_mcp_server.web_server --config /path/to/config.json --log-level DEBUG

可用的 MCP 工具

连接管理

• add_connection: 添加新的数据库连接
• test_connection: 测试数据库连接
• list_connections: 列出所有数据库连接
• remove_connection: 删除数据库连接

查询执行

• execute_query: 执行 SQL 查询
• get_records: 从表中获取记录
• insert_record: 将记录插入到表中
• update_record: 更新表中的记录
• delete_record: 从表中删除记录

模式管理

• list_tables: 列出数据库中的所有表
• get_table_schema: 获取表的模式
• create_table: 创建新表
• drop_table: 删除表
• create_index: 在表上创建索引
• drop_index: 删除索引
• alter_table: 更改表结构

事务管理

• begin_transaction: 开始事务
• commit_transaction: 提交事务
• rollback_transaction: 回滚事务

示例

添加连接

{
  "connection_id": "my_sqlite_db",
  "type": "sqlite",
  "db_path": "/path/to/database.db"
}

执行查询

{
  "connection_id": "my_sqlite_db",
  "query": "SELECT * FROM users WHERE age > ?",
  "params": [21]
}

创建表

{
  "connection_id": "my_sqlite_db",
  "table": "users",
  "columns": [
    {
      "name": "id",
      "type": "INTEGER",
      "primary_key": true,
      "nullable": false
    },
    {
      "name": "name",
      "type": "TEXT",
      "nullable": false
    },
    {
      "name": "email",
      "type": "TEXT",
      "nullable": true
    }
  ]
}

插入记录

{
  "connection_id": "my_sqlite_db",
  "table": "users",
  "data": {
    "name": "John Doe",
    "email": "john@example.com"
  }
}

开发

运行测试

# 运行所有测试
python -m unittest discover

# 运行特定的测试文件
python -m unittest tests.test_sqlite

从其他 LLM 连接

当作为独立的 Web 服务器运行时，其他 LLM（如 Llama 3）可以通过 HTTP 连接到数据库 MCP 服务器。 服务器公开以下端点：

端点

• /list_tools - GET 或 POST: 返回所有可用工具的列表，其中包含它们的描述和输入模式
• /call_tool - POST: 执行特定的数据库工具

示例：从另一个 LLM 调用

要将此服务器与另一个 LLM 一起使用，请让 LLM 生成对服务器的 HTTP 请求。 以下是如何为 Llama 3 等 LLM 构建提示的示例：

您可以通过向 http://localhost:8000 上的数据库服务发出 HTTP 请求来与数据库交互。
该服务提供以下端点：

1. 获取可用工具列表：
   向以下地址发出 POST 请求：http://localhost:8000/list_tools
   
2. 执行数据库工具：
   向以下地址发出 POST 请求：http://localhost:8000/call_tool
   带有如下 JSON 正文：
   {
     "name": "tool_name",
     "arguments": {
       "param1": "value1",
       "param2": "value2"
     }
   }

例如，要执行 SQL 查询，您可以发出如下请求：
POST http://localhost:8000/call_tool
Content-Type: application/json

{
  "name": "execute_query",
  "arguments": {
    "connection_id": "my_db",
    "query": "SELECT * FROM users"
  }
}

客户端集成的示例 Python 代码

import requests
import json

# 数据库 MCP 服务器的基本 URL
BASE_URL = "http://localhost:8000"

# 列出可用工具
def list_tools():
    response = requests.post(f"{BASE_URL}/list_tools")
    return response.json()

# 执行数据库工具
def call_tool(tool_name, arguments):
    payload = {
        "name": tool_name,
        "arguments": arguments
    }
    response = requests.post(f"{BASE_URL}/call_tool", json=payload)
    return response.json()

# 示例：列出数据库中的表
def list_tables(connection_id):
    return call_tool("list_tables", {"connection_id": connection_id})

# 示例：执行 SQL 查询
def execute_query(connection_id, query, params=None):
    return call_tool("execute_query", {
        "connection_id": connection_id,
        "query": query,
        "params": params
    })

# 示例：添加新连接
def add_connection(connection_id, db_type, **kwargs):
    args = {"connection_id": connection_id, "type": db_type}
    args.update(kwargs)
    return call_tool("add_connection", args)

许可证

MIT 许可证