数据库MCP服务器
一种模型上下文协议（MCP）服务器，提供连接到各种数据库系统并与之交互的工具。
特性
- 多数据库支持：连接到SQLite、PostgreSQL、MySQL/MariaDB和SQL Server数据库
- 统一接口：用于跨所有支持的数据库类型进行数据库操作的通用工具
- 数据库特定扩展：必要时，针对数据库特定功能的特定工具
- 模式管理：创建、更改和删除表和索引
- 查询执行：执行原始SQL查询或使用结构化查询工具
- 事务支持：开始、提交和回滚事务
安装
先决条件
- Python 3.8或更高版本
- 所需的Python包（使用pip自动安装）：
- SQL炼金术 - 各种数据库驱动程序，具体取决于您要使用的数据库： - SQLite（包含在Python中） - PostgreSQL： psycopg2-binary - MySQL/MariaDB： mysql-connector-python - SQL服务器： pyodbc
从源安装
# Clone the repository
git clone
# Install the package
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服务器
# Run with default settings
python -m db_mcp_server
# Specify a configuration file
python -m db_mcp_server --config /path/to/config.json
# Set logging level
python -m db_mcp_server --log-level DEBUG作为独立Web服务器（适用于任何LLM）
# Run as a web server
python -m db_mcp_server.web_server
# Specify host and port
python -m db_mcp_server.web_server --host 0.0.0.0 --port 8000
# Specify configuration file and logging level
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"
}
}发展
运行测试
# Run all tests
python -m unittest discover
# Run specific test file
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构建提示的示例：
You can interact with a database by making HTTP requests to a database service at http://localhost:8000.
The service provides the following endpoints:
1. To get a list of available tools:
Make a POST request to: http://localhost:8000/list_tools
2. To execute a database tool:
Make a POST request to: http://localhost:8000/call_tool
with a JSON body like:
{
"name": "tool_name",
"arguments": {
"param1": "value1",
"param2": "value2"
}
}
For example, to execute a SQL query, you would make a request like:
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
# Base URL of the database MCP server
BASE_URL = "http://localhost:8000"
# List available tools
def list_tools():
response = requests.post(f"{BASE_URL}/list_tools")
return response.json()
# Execute a database tool
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()
# Example: List tables in a database
def list_tables(connection_id):
return call_tool("list_tables", {"connection_id": connection_id})
# Example: Execute a SQL query
def execute_query(connection_id, query, params=None):
return call_tool("execute_query", {
"connection_id": connection_id,
"query": query,
"params": params
})
# Example: Add a new connection
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)