内容
# 数据库 MCP 服务器
一个模型上下文协议（Model Context Protocol，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`
### 从源代码安装
```bash
# 克隆仓库
git clone <repository-url>
# 安装包
pip install -e .
```
## 配置
服务器可以通过环境变量、配置文件或在运行时提供连接详细信息进行配置。
### 环境变量
- `DB_CONFIG_PATH`: JSON 配置文件的路径
- `DB_CONNECTIONS`: 连接 ID 的逗号分隔列表或包含连接详细信息的 JSON 字符串
### 配置文件格式
```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 服务器
```bash
# 使用默认设置运行
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）
```bash
# 作为 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`: 回滚事务
## 示例
### 添加连接
```json
{
"connection_id": "my_sqlite_db",
"type": "sqlite",
"db_path": "/path/to/database.db"
}
```
### 执行查询
```json
{
"connection_id": "my_sqlite_db",
"query": "SELECT * FROM users WHERE age > ?",
"params": [21]
}
```
### 创建表
```json
{
"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
}
]
}
```
### 插入记录
```json
{
"connection_id": "my_sqlite_db",
"table": "users",
"data": {
"name": "John Doe",
"email": "john@example.com"
}
}
```
## 开发
### 运行测试
```bash
# 运行所有测试
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 请求到服务器。以下是如何为 LLM（如 Llama 3）构建提示的示例：
```
您可以通过向 http://localhost:8000 的数据库服务发出 HTTP 请求与数据库进行交互。
该服务提供以下端点：
1. 获取可用工具列表：
向 http://localhost:8000/list_tools 发出 POST 请求
2. 执行数据库工具：
向 http://localhost:8000/call_tool 发出 POST 请求
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 代码
```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 许可证](LICENSE)
MCP 市场