pg-mcp: PostgreSQL MCP Server

基于自然语言的 PostgreSQL 查询 MCP 服务。接入 DeepSeek 大模型，自动将自然语言转化为 SQL 并执行。

快速开始

1. 安装

pip install -e .

2. 配置

复制 .env.example 为 .env，填写以下必填项：

PG_MCP_DEEPSEEK_API_KEY=sk-your-key
PG_MCP_DATABASES='[{"name":"mydb","dsn":"postgresql://user:pass@localhost:5432/mydb","default":true}]'

或单库模式：

PG_MCP_DEEPSEEK_API_KEY=sk-your-key
PG_MCP_DATABASE_URL=postgresql://user:pass@localhost:5432/mydb

3. 启动

pg-mcp

或

python -m pg_mcp.server

MCP Tool 列表

Tool	功能
query_database	自然语言 → SQL / 查询结果（核心）
explain_sql	自然语言解释 SQL 功能
refresh_schema	手动刷新数据库 Schema 缓存
list_databases	列出所有可访问数据库

MCP Client 配置

在 mcp.json 中添加：

{
  "mcpServers": {
    "pg-mcp": {
      "command": "python",
      "args": ["-m", "pg_mcp.server"],
      "env": {
        "PG_MCP_DEEPSEEK_API_KEY": "${DEEPSEEK_API_KEY}",
        "PG_MCP_DATABASES": "[{\"name\":\"mydb\",\"dsn\":\"postgresql://user:pass@localhost:5432/mydb\",\"default\":true}]"
      }
    }
  }
}

安全

三重防线：

1. sqlglot AST 分析 — 自动拦截 DDL/DML/DCL 及危险表引用
2. 正则兜底 — 处理解析器无法识别的边界情况
3. 数据库只读用户 — 从数据库层面杜绝写入

Fail-Closed 原则：任何 SQL 解析失败、异常 → 一律拒绝执行，绝不默认放行。

开发

# 安装开发依赖
pip install -e ".[dev]"

# 运行测试
pytest tests/ -v
pytest tests/test_sql_validator.py -v   # 安全校验（最重要）

# 跳过集成测试
pytest -m "not integration" -v

技术栈

• FastMCP 3.4.x — MCP Server 框架
• asyncpg — PostgreSQL 异步驱动
• sqlglot — SQL 解析与安全校验
• DeepSeek (openai SDK) — LLM 生成 SQL
• pydantic-settings — 配置管理
• structlog — 结构化日志