db阅读器mcp
用于安全生产数据查询的只读数据库MCP服务器
英语| 简体中文
db reader mcp是一个模型上下文协议（mcp）服务器，它使AI助手能够安全地查询数据库 100%只读 保证。非常适合数据安全至关重要的生产环境。
特性
- 100%只读：多层保护确保数据不可修改
- 多数据库支持：跨多个类别的15个数据库
| 类别 | 数据库 |
|---|---|
| 关系型 | PostgreSQL、MySQL、MariaDB、SQLite |
| 企业 | Oracle、SQL Server |
| 中文数据库 | openGauss、TiDB、达蒙\* |
| 分析 | ClickHouse |
| NoSQL | Redis、MongoDB、Elasticsearch |
| 时间序列 | Cassandra，InfluxDB |
- HTTP传输：用于远程访问的流式HTTP服务器
- 安全第一：SQL注入保护、表阻塞、敏感列掩码、查询验证
- 演出：查询缓存、并发查询限制、查询超时、查询日志记录缓慢
- 监控：使用连接池统计信息进行健康检查，审计日志记录
- 易于配置：简单的YAML配置，支持环境变量
快速开始
紫外线安装（推荐）
# Install base dependencies
uv pip install -e ".[dev]"
# Install with optional database support
uv pip install -e ".[dev,oracle]" # + Oracle support
uv pip install -e ".[dev,sqlserver]" # + SQL Server support
uv pip install -e ".[dev,clickhouse]" # + ClickHouse support
uv pip install -e ".[dev,opengauss]" # + openGauss support (uses psycopg2)
uv pip install -e ".[dev,tidb]" # + TiDB support (uses pymysql)
uv pip install -e ".[dev,mariadb]" # + MariaDB support (uses pymysql)
uv pip install -e ".[dev,dameng]" # + Dameng support
uv pip install -e ".[dev,redis]" # + Redis support
uv pip install -e ".[dev,mongodb]" # + MongoDB support
uv pip install -e ".[dev,elasticsearch]" # + Elasticsearch support
uv pip install -e ".[dev,cassandra]" # + Cassandra support
uv pip install -e ".[dev,influxdb]" # + InfluxDB support管道安装
pip install -e ".[dev]"
# With optional dependencies
pip install -e ".[dev,redis,mongodb]"配置
- 创建配置文件
config/config.yaml:
databases:
# SQL Database
- name: my_database
type: postgresql
connection:
host: localhost
port: 5432
database: myapp
user: readonly_user
password: ${DB_PASSWORD}
# Redis
- name: cache
type: redis
connection:
host: localhost
port: 6379
db: 0
# Oracle (using service_name)
- name: oracle_db
type: oracle
connection:
host: localhost
port: 1521
service_name: ORCL
user: readonly_user
password: ${ORACLE_PASSWORD}
# ClickHouse
- name: analytics
type: clickhouse
connection:
host: localhost
port: 9000
database: default
user: default
password: ${CLICKHOUSE_PASSWORD}
# protocol: native # or 'http' for port 8123
# SQL Server
- name: sqlserver_db
type: sqlserver
connection:
host: localhost
port: 1433
database: myapp
user: readonly_user
password: ${SQLSERVER_PASSWORD}
# driver: pymssql # or 'pyodbc'
# openGauss (Huawei)
- name: opengauss_db
type: opengauss
connection:
host: localhost
port: 5432
database: myapp
user: readonly_user
password: ${OPENGAUSS_PASSWORD}
# TiDB (PingCAP)
- name: tidb_db
type: tidb
connection:
host: localhost
port: 4000
database: myapp
user: readonly_user
password: ${TIDB_PASSWORD}
# MariaDB
- name: mariadb_db
type: mariadb
connection:
host: localhost
port: 3306
database: myapp
user: readonly_user
password: ${MARIADB_PASSWORD}
# MongoDB
- name: docs
type: mongodb
connection:
host: localhost
port: 27017
database: myapp
# Elasticsearch
- name: search
type: elasticsearch
connection:
host: localhost
port: 9200
# scheme: http # or 'https'
# user: elastic
# password: ${ES_PASSWORD}
# api_key: ${ES_API_KEY} # alternative to user/password
# Security settings
security:
blocked_tables:
- users_password
- api_keys
sensitive_columns:
- table: users
columns:
- name: email
mask_strategy: partial
- name: phone
mask_strategy: partial
# Performance limits
limits:
max_rows_per_query: 1000
query_timeout_seconds: 30
max_concurrent_queries: 10
enable_query_cache: true
cache_ttl_seconds: 300
# Logging
logging:
audit_enabled: true
slow_query_threshold_ms: 1000- 设置环境变量：
export DB_PASSWORD="your_password"
export DB_MCP_CONFIG_PATH="config/config.yaml"与MCP客户端一起使用
启动HTTP服务器
# Start server (default: 0.0.0.0:3001)
python -m db_reader_mcp.server
# Or with custom host/port
export MCP_HOST="0.0.0.0"
export MCP_PORT="3001"
python -m db_reader_mcp.server
# Test health check
curl http://localhost:3001/healthMCP客户端配置
添加到您的MCP客户端配置中（例如Cherry Studio、Cline等）：
Cherry Studio（streamableHttp）：
{
"mcpServers": {
"db-reader": {
"type": "streamableHttp",
"url": "http://localhost:3001/mcp",
"headers": {
"Content-Type": "application/json"
}
}
}
}其他MCP客户端：
{
"mcpServers": {
"db-reader": {
"url": "http://localhost:3001/mcp",
"transport": "http"
}
}
}可用工具
SQL数据库工具
1.查询
执行只读SQL查询：
query(sql="SELECT * FROM users LIMIT 10", database="my_database")2.列表表
列出数据库中的所有表：
list_tables(database="my_database")3.描述表
获取详细的表格结构：
describe_table(table_name="users")4.get_schema
获取完整的数据库架构概述：
get_schema(database="my_database")5.表格状态
获取表统计信息（行数、大小、索引）：
table_stats(table_name="users")6.样本数据
从表中获取示例数据：
sample_data(table_name="users", limit=5)Redis工具
7.redis_command
执行只读Redis命令：
redis_command(command="GET", args=["key"], database="cache")
redis_command(command="HGETALL", args=["hash_key"])
redis_command(command="KEYS", args=["pattern*"])允许的命令：GET、MGET、STRLEN、HGET、HGETALL、HKEYS、HVALS、HLEN、HEXISTS、LRANGE、LLEN、LINDEX、SMEMBERS、SCARD、SISMEMBER、ZRANGE、ZRANGEBYSCORE、ZCARD、ZSCORE、ZRANK、KEYS、EXISTS、TYPE、TTL、PTTL、SCAN、DBSIZE、INFO
MongoDB工具
8.mongo_query
执行只读MongoDB操作：
mongo_query(operation="find", collection="users", query={"active": true}, limit=10)
mongo_query(operation="count_documents", collection="orders", query={"status": "pending"})
mongo_query(operation="aggregate", collection="sales", pipeline=[{"$group": {"_id": "$category", "total": {"$sum": "$amount"}}}])
mongo_query(operation="distinct", collection="products", field="category")
mongo_query(operation="list_collection_names")允许的操作：find、find_one、count_documents、estimated_document_count、distinct、聚合、list_collection_names
Elasticsearch工具
9.es_query
执行只读Elasticsearch操作：
es_query(operation="search", index="logs", query={"match": {"message": "error"}}, limit=10)
es_query(operation="get", index="products", doc_id="123")
es_query(operation="count", index="events", query={"range": {"timestamp": {"gte": "now-1d"}}})
es_query(operation="list_indices")
es_query(operation="get_mapping", index="users")
es_query(operation="search", index="sales", body={"size": 0, "aggs": {"total": {"sum": {"field": "amount"}}}})允许的操作：搜索、获取、计数、list_indices、get_mapping、get_setings、存在
Cassandra工具
10.cassandra-query
执行只读Cassandra CQL查询：
cassandra_query(cql="SELECT * FROM users LIMIT 10", keyspace="my_keyspace")
cassandra_query(cql="SELECT * FROM orders WHERE user_id = 123")
cassandra_query(cql="DESCRIBE TABLE users")11.cassandra_list_key空格
列出可用的Cassandra密钥空间：
cassandra_list_keyspaces(database="my_cassandra")12.cassandra_list_tables
在Cassandra键空间中列出表：
cassandra_list_tables(database="my_cassandra", keyspace="my_keyspace")允许的语句：SELECT、DESCRIBE、SHOW
InfluxDB工具
13.influexdb_query
执行只读InfluxDB Flux查询：
influxdb_query(flux='from(bucket: "metrics") |> range(start: -1h) |> limit(n: 100)')
influxdb_query(flux='from(bucket: "sensors") |> range(start: -24h) |> filter(fn: (r) => r._measurement == "temperature")')14.influexdb_list_buckets
列出可用的InfluxDB存储桶：
influxdb_list_buckets(database="metrics_db")15.influexdb_list_measures
列出InfluxDB存储桶中的度量值：
influxdb_list_measurements(database="metrics_db", bucket="metrics")注： to() 函数被阻止以阻止写入操作
安全
📖 详细安全指南: 安全.md | 中文版
- SQL验证：阻止所有写入操作（INSERT、UPDATE、DELETE、DROP等）
- 表阻塞：配置永远不应访问的表
- 敏感柱遮蔽：自动屏蔽敏感数据（电子邮件、电话等）
- 策略： null, partial, hash:N, truncate:N, first_n:N, last_n:N, random
- 只读用户：使用仅具有SELECT权限的数据库用户
- 查询限制：自动行限制（max_rows_per_query）和超时保护
- NoSQL命令白名单：只允许只读Redis命令和MongoDB操作
监控
健康检查
curl http://localhost:3001/health退货：
{
"service": "db-reader-mcp",
"status": "healthy",
"databases": {
"my_database": {
"status": "healthy",
"type": "postgresql",
"message": "Connection OK",
"pool": {
"size": 5,
"checked_in": 5,
"checked_out": 0,
"overflow": 0
}
}
},
"total": 1,
"healthy_count": 1
}审计日志
所有查询均以以下方式记录：
- 时间戳、数据库、SQL/命令
- 成功/失败状态
- 行数、执行时间
- 查询标记速度慢(
[SLOW]对于超过阈值的查询）
数据库设置
PostgreSQL
CREATE USER mcp_readonly WITH PASSWORD 'secure_password';
GRANT CONNECT ON DATABASE myapp TO mcp_readonly;
GRANT USAGE ON SCHEMA public TO mcp_readonly;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO mcp_readonly;MySQL
CREATE USER 'mcp_readonly'@'%' IDENTIFIED BY 'secure_password';
GRANT SELECT ON myapp.* TO 'mcp_readonly'@'%';甲骨文
CREATE USER mcp_readonly IDENTIFIED BY "secure_password";
GRANT CREATE SESSION TO mcp_readonly;
GRANT SELECT ANY TABLE TO mcp_readonly;
-- Or grant on specific tables:
-- GRANT SELECT ON schema.table_name TO mcp_readonly;ClickHouse
CREATE USER mcp_readonly IDENTIFIED BY 'secure_password';
GRANT SELECT ON *.* TO mcp_readonly;
-- Or grant on specific database:
-- GRANT SELECT ON analytics.* TO mcp_readonly;戴蒙（DM7/DM8）
注： Dameng支持需要额外设置：
- 这
dmPython驱动程序仅支持Linux x86_64/aarch64和Windows x64平台（不支持macOS） - 需要官方的大萌加密模块（pip包中不包含）
- 如需生产使用，请从官方网站安装完整的达梦驱动程序
对于测试/开发：
# Install the dialect (Linux/Windows only)
pip install sqlalchemy-dm-dialect dmPython数据库设置：
CREATE USER mcp_readonly IDENTIFIED BY "secure_password";
GRANT SELECT ANY TABLE TO mcp_readonly;SQL Server
CREATE LOGIN mcp_readonly WITH PASSWORD = 'secure_password';
CREATE USER mcp_readonly FOR LOGIN mcp_readonly;
GRANT SELECT ON SCHEMA::dbo TO mcp_readonly;
-- Or grant on specific tables:
-- GRANT SELECT ON dbo.table_name TO mcp_readonly;openGauss（华为）
CREATE USER mcp_readonly WITH PASSWORD 'secure_password';
GRANT CONNECT ON DATABASE myapp TO mcp_readonly;
GRANT USAGE ON SCHEMA public TO mcp_readonly;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO mcp_readonly;TiDB（PingCAP）
CREATE USER 'mcp_readonly'@'%' IDENTIFIED BY 'secure_password';
GRANT SELECT ON myapp.* TO 'mcp_readonly'@'%';MariaDB
CREATE USER 'mcp_readonly'@'%' IDENTIFIED BY 'secure_password';
GRANT SELECT ON myapp.* TO 'mcp_readonly'@'%';发展
用uv设置开发环境
# Create virtual environment (if not exists)
uv venv
# Activate virtual environment
source .venv/bin/activate # On Windows: .venv\Scripts\activate
# Install dependencies
uv pip install -e ".[dev]"运行测试
# Run all tests
uv run pytest
# Run unit tests only
uv run pytest tests/unit/ -v
# Run integration tests (requires running databases)
uv run pytest tests/integration/real_db/ -v
# Run specific test file
uv run pytest tests/unit/test_validator.py -v
# Run with coverage
uv run pytest --cov=src/db_reader_mcp --cov-report=html使用Podman/Docker进行本地集成测试
使用docker compose启动测试数据库：
# Start all test databases
podman-compose -f docker-compose.test.yml up -d
# Or with Docker
docker-compose -f docker-compose.test.yml up -d
# Wait for services to be ready, then run integration tests
uv run pytest tests/integration/real_db/ -v
# Stop test databases
podman-compose -f docker-compose.test.yml down集成测试包括：
- PostgreSQL：复杂查询、CTE、窗口函数、JSON操作
- MySQL：JOIN、子查询、模式自检
- MongoDB：聚合管道、嵌套查询、方面操作
- 瑞迪斯：所有数据类型（字符串、哈希、列表、集合、排序集合）
- 弹性搜索：布尔查询、聚合、映射
- 甲骨文：复杂查询、层次查询、模式自省
- SQL Server：JOIN、子查询、模式自检
- open高斯：PostgreSQL兼容的查询和操作
- TiDB：MySQL兼容的查询和操作
- ClickHouse：分析查询和聚合
- 达蒙：与Oracle兼容的查询和操作
代码质量
# Linting
uv run ruff check .
# Format code
uv run ruff format .
# Type checking
uv run mypy src/预提交钩子
安装预提交挂钩，以便在每次提交之前自动运行检查：
cp scripts/pre-commit.sh .git/hooks/pre-commit预提交钩子运行：
- 检查 -代码linting
- ruff格式 -格式验证
- 米皮 -类型检查
- 单元测试 -总是跑
- 集成测试 -仅在检测到数据库时运行
如果没有数据库正在运行，则跳过集成测试并发出警告。
在本地运行服务器
# Using uv
uv run python -m db_reader_mcp.server
# Or after activating venv
python -m db_reader_mcp.server项目结构
db-reader-mcp/
├── src/db_reader_mcp/
│ ├── server.py # MCP server entry point
│ ├── config/
│ │ └── loader.py # Configuration loader
│ ├── database/
│ │ ├── base.py # DatabaseAdapter abstract base class
│ │ ├── registry.py # AdapterRegistry (type routing)
│ │ ├── coordinator.py # DatabaseCoordinator (cache, timeout, audit)
│ │ └── adapters/
│ │ ├── redis.py # Redis adapter (read-only)
│ │ ├── mongodb.py # MongoDB adapter (read-only)
│ │ ├── elasticsearch.py # Elasticsearch adapter (read-only)
│ │ └── sql/
│ │ ├── base.py # SQLAdapter base (SQLAlchemy)
│ │ ├── postgresql.py
│ │ ├── mysql.py
│ │ ├── sqlite.py
│ │ ├── oracle.py
│ │ ├── clickhouse.py
│ │ ├── sqlserver.py
│ │ ├── opengauss.py
│ │ ├── tidb.py
│ │ ├── mariadb.py
│ │ └── dameng.py
│ ├── security/
│ │ ├── validator.py # SQL security validator
│ │ └── masker.py # Sensitive data masker
│ ├── logging/
│ │ └── audit.py # Audit logger
│ ├── utils/
│ │ ├── concurrency.py # Concurrent query limiter
│ │ └── timeout.py # Query timeout handler
│ └── tools/
│ ├── query.py # Query execution tool
│ ├── schema.py # Schema inspection tools
│ ├── redis_tool.py # Redis command tool
│ ├── mongo_tool.py # MongoDB query tool
│ ├── es_tool.py # Elasticsearch query tool
│ ├── cassandra_tool.py # Cassandra query tool
│ └── influxdb_tool.py # InfluxDB query tool
├── tests/
│ ├── unit/ # Unit tests
│ └── integration/
│ └── real_db/ # Real database integration tests
├── config/
│ └── config.example.yaml # Configuration example
├── docker-compose.test.yml # Test database containers
└── pyproject.toml # Project configuration许可证
MIT许可证-有关详细信息，请参阅许可证文件
贡献
欢迎投稿！请阅读CONTRIBUTING.md了解指南。
支持
- 问题：https://github.com/puppet4/db-reader-mcp/issues