数据库 MCP 服务器

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

特征

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

Related MCP server: SQLite MCP Server

安装

先决条件

Python 3.8 或更高版本
所需的 Python 包（使用 pip 自动安装）：
- SQLAlchemy
- 各种数据库驱动程序，取决于您要使用的数据库：
  - SQLite（包含在 Python 中）
  - PostgreSQL： psycopg2-binary
  - MySQL/MariaDB： mysql-connector-python
  - SQL 服务器： pyodbc

从源代码安装

# Clone the repository git clone <repository-url> # 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 生成发送到该服务器的 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)

执照

MIT 许可证

Database MCP Server