# pg-mcp

**Repository Path**: liliangzi/pg-mcp

## Basic Information

- **Project Name**: pg-mcp
- **Description**: 一个安全、高性能的 PostgreSQL MCP Server，使 AI 助手能够通过自然语言与 PostgreSQL 数据库进行安全交互。
- **Primary Language**: Python
- **License**: Not specified
- **Default Branch**: main
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 1
- **Forks**: 0
- **Created**: 2026-02-02
- **Last Updated**: 2026-05-21

## Categories & Tags

**Categories**: Uncategorized

**Tags**: MCP, aicoding, Python, claude, speckit

## README

# pg-mcp - PostgreSQL MCP Server

[![Python 3.11+](https://img.shields.io/badge/python-3.11+-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

一个安全、高性能的 PostgreSQL MCP Server，使 AI 助手能够通过自然语言与 PostgreSQL 数据库进行安全交互。

## 功能特性

- 🔒 **安全优先**: 多层 SQL 验证，只读事务，危险函数黑名单，EXPLAIN 限制
- 🚀 **高性能**: 全链路异步，连接池管理，并发控制
- 🤖 **AI 驱动**: 支持 OpenAI GPT-4 / DeepSeek 等 LLM，自然语言转 SQL
- 📊 **可观测性**: Prometheus 指标，结构化日志，敏感数据脱敏
- 🛡️ **弹性设计**: LLM 熔断器，超时控制，健康检查
- 🔍 **结果验证**: LLM 驱动的结果置信度评估
- 🗄️ **多数据库**: 支持多数据库配置，每库独立安全策略

---

## 目录

- [快速开始](#快速开始)
- [主要功能](#主要功能)
- [技术架构](#技术架构)
- [工程结构](#工程结构)
- [配置说明](#配置说明)
- [开发指南](#开发指南)
- [相关文档](#相关文档)

---

## 快速开始

### 1. 安装依赖

```bash
# 使用 uv（推荐）
uv sync

# 或使用 bun
bun install
```

### 2. 配置环境变量

复制 `.env.example` 到项目根目录并重命名为 `.env`，然后填写配置：

```bash
cp .env.example .env
```

**最小配置示例**：

```bash
# 数据库配置
PG_MCP_DATABASES__0__NAME=mydb
PG_MCP_DATABASES__0__HOST=localhost
PG_MCP_DATABASES__0__PORT=5432
PG_MCP_DATABASES__0__USER=postgres
PG_MCP_DATABASES__0__PASSWORD=your_password
PG_MCP_DATABASES__0__DATABASE=mydb

# LLM 配置（支持 OpenAI / DeepSeek 等兼容 API）
PG_MCP_OPENAI__API_KEY=sk-your-api-key
PG_MCP_OPENAI__MODEL=gpt-4o
# 如使用 DeepSeek，添加 base_url：
# PG_MCP_OPENAI__BASE_URL=https://api.deepseek.com/v1
```

### 3. 运行 MCP Server

```bash
# 使用 uvx 直接运行（推荐）
uvx --from /path/to/pg-mcp pg-mcp

# 或使用 Python 模块
python -m pg_mcp

# 或使用 Makefile
make dev
```

### 4. 在 Claude Code 中配置

编辑 `~/.claude.json`，添加 MCP Server 配置：

```json
{
  "mcpServers": {
    "pg-mcp": {
      "command": "uvx",
      "args": ["--from", "/path/to/pg-mcp", "pg-mcp"],
      "env": {
        "PG_MCP_DATABASES__0__NAME": "mydb",
        "PG_MCP_DATABASES__0__HOST": "localhost",
        "PG_MCP_DATABASES__0__PORT": "5432",
        "PG_MCP_DATABASES__0__USER": "postgres",
        "PG_MCP_DATABASES__0__PASSWORD": "your_password",
        "PG_MCP_DATABASES__0__DATABASE": "mydb",
        "PG_MCP_OPENAI__API_KEY": "sk-your-api-key",
        "PG_MCP_OPENAI__MODEL": "gpt-4o"
      }
    }
  }
}
```

重启 Claude Code，即可使用自然语言查询数据库！

---

## 主要功能

### MCP Server 功能

pg-mcp 提供以下 MCP Tools：

| Tool | 描述 |
|------|------|
| `query` | 执行自然语言查询，返回 SQL 或查询结果 |
| `refresh_schema` | 手动刷新数据库 Schema 缓存 |

**使用示例**：

```
# 在 Claude Code 中直接提问
查询最近 7 天的订单总金额

统计每个用户的订单数量，按数量降序排列

找出销售额最高的 10 个产品
```

pg-mcp 会自动：
1. 将自然语言转换为 SQL
2. 验证 SQL 安全性（只允许 SELECT，禁止危险函数）
3. 在只读事务中执行查询
4. 返回结果并评估置信度

### pg-data Skill（Claude Code 专用）

项目还提供了 `pg-data` Skill，位于 `.claude/skills/pg-data/`，可直接在 Claude Code 中使用 psql 查询测试数据库。

**支持的测试数据库**：
- `pg_mcp_small` - 博客系统（用户、文章、分类、标签）
- `pg_mcp_medium` - 电商系统（用户、商品、订单、支付、评价）
- `pg_mcp_large` - 企业 SaaS 系统（多租户、CRM、财务、人力资源）

**使用方式**：在 Claude Code 中调用 `/pg-data` skill 或直接描述查询需求。

---

## 技术架构

### 架构总览

```
┌─────────────────────────────────────────────────────────────┐
│                        MCP Client                            │
│                (Claude Code / Claude Desktop)                │
└─────────────────────────────────────────────────────────────┘
                            │
                            ▼ MCP Protocol (stdio)
┌─────────────────────────────────────────────────────────────┐
│                   FastMCP Server Layer                       │
│  ┌─────────────────────────────────────────────────────┐    │
│  │              Query Orchestrator                      │    │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐          │    │
│  │  │ Schema   │  │   SQL    │  │   SQL    │          │    │
│  │  │  Cache   │  │Generator │  │Validator │          │    │
│  │  └──────────┘  └──────────┘  └──────────┘          │    │
│  │  ┌──────────┐  ┌──────────┐  ┌──────────┐          │    │
│  │  │   SQL    │  │  Result  │  │ Circuit  │          │    │
│  │  │ Executor │  │Validator │  │ Breaker  │          │    │
│  │  └──────────┘  └──────────┘  └──────────┘          │    │
│  └─────────────────────────────────────────────────────┘    │
└─────────────────────────────────────────────────────────────┘
                            │
                ┌───────────┴───────────┐
                ▼                       ▼
        ┌──────────────┐        ┌──────────────┐
        │  PostgreSQL  │        │   LLM API    │
        │   Database   │        │ (OpenAI等)   │
        └──────────────┘        └──────────────┘
```

### 多数据库隔离架构

每个数据库拥有独立的组件实例，确保完全隔离：

```
┌─────────────────────────────────────────────────────────────┐
│                        MCP Server                            │
└─────────────────────────────────────────────────────────────┘
                              │
              ┌───────────────┼───────────────┐
              ▼               ▼               ▼
    ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐
    │ Pool: db1       │ │ Pool: db2       │ │ Pool: db3       │
    │ Executor: db1   │ │ Executor: db2   │ │ Executor: db3   │
    │ Validator: db1  │ │ Validator: db2  │ │ Validator: db3  │
    │ Schema: db1     │ │ Schema: db2     │ │ Schema: db3     │
    └─────────────────┘ └─────────────────┘ └─────────────────┘
```

### 核心技术栈

| 组件 | 技术 | 用途 |
|------|------|------|
| MCP 框架 | FastMCP | MCP 协议服务器实现 |
| 数据库驱动 | asyncpg | 高性能异步 PostgreSQL 连接 |
| SQL 解析 | SQLGlot | SQL AST 解析和安全验证 |
| 数据验证 | Pydantic v2 | 配置管理和数据模型 |
| LLM 客户端 | OpenAI SDK | 自然语言转 SQL |

### 安全多层防御

```
┌─────────────────────────────────────────────────────────────┐
│ Layer 1: 输入清洗 (InputSanitizer)                          │
│  - Prompt Injection 检测                                    │
│  - 敏感信息过滤                                             │
├─────────────────────────────────────────────────────────────┤
│ Layer 2: SQL 验证 (SQLValidator + SQLGlot)                  │
│  - 只允许 SELECT/WITH...SELECT                              │
│  - 禁止 EXPLAIN、DML、DDL                                   │
│  - 危险函数黑名单 (pg_sleep, lo_import 等)                  │
│  - 表/列黑名单检查                                          │
│  - SQL 长度和嵌套深度限制                                   │
├─────────────────────────────────────────────────────────────┤
│ Layer 3: 执行层安全 (SQLExecutor)                           │
│  - 只读事务 (BEGIN READ ONLY)                               │
│  - 语句超时 (statement_timeout)                             │
│  - 结果行数和大小限制                                       │
│  - search_path 限制                                         │
│  - 可选只读角色切换                                         │
├─────────────────────────────────────────────────────────────┤
│ Layer 4: 数据脱敏 (DataMasker)                              │
│  - 敏感字段自动过滤                                         │
│  - 日志脱敏                                                 │
└─────────────────────────────────────────────────────────────┘
```

---

## 工程结构

```
pg-mcp/
├── src/pg_mcp/                    # 源代码
│   ├── __init__.py
│   ├── __main__.py                # 入口点
│   ├── server.py                  # FastMCP Server 定义
│   ├── health_check.py            # 健康检查
│   │
│   ├── config/                    # 配置管理
│   │   └── settings.py            # Pydantic Settings
│   │
│   ├── models/                    # 数据模型
│   │   ├── schema.py              # Schema 模型
│   │   ├── query.py               # 查询请求/响应模型
│   │   └── errors.py              # 错误模型
│   │
│   ├── services/                  # 核心服务
│   │   ├── orchestrator.py        # 查询协调器
│   │   ├── sql_generator.py       # SQL 生成 (LLM)
│   │   ├── sql_validator.py       # SQL 验证 (SQLGlot)
│   │   ├── sql_executor.py        # SQL 执行 (asyncpg)
│   │   ├── result_validator.py    # 结果验证 (LLM)
│   │   ├── data_masker.py         # 敏感数据脱敏
│   │   └── input_sanitizer.py     # 输入清洗
│   │
│   ├── cache/                     # 缓存管理
│   │   └── schema_cache.py        # Schema 缓存
│   │
│   ├── db/                        # 数据库层
│   │   ├── pool.py                # 连接池管理
│   │   └── introspection.py       # Schema 内省
│   │
│   ├── resilience/                # 弹性组件
│   │   ├── circuit_breaker.py     # LLM 熔断器
│   │   └── health_checker.py      # 健康检查器
│   │
│   ├── observability/             # 可观测性
│   │   ├── metrics.py             # Prometheus 指标
│   │   └── logging.py             # 结构化日志
│   │
│   └── prompts/                   # LLM Prompt 模板
│       ├── sql_generation.py
│       └── result_validation.py
│
├── tests/                         # 测试
│   ├── conftest.py                # pytest fixtures
│   ├── unit/                      # 单元测试
│   ├── integration/               # 集成测试
│   └── e2e/                       # 端到端测试
│
├── fixtures/                      # 测试数据库 SQL
│   ├── pg_mcp_small.sql
│   ├── pg_mcp_medium.sql
│   └── pg_mcp_large.sql
│
├── specs/                         # 设计文档
│   ├── 0001-pg-mcp-prd.md         # 产品需求文档
│   ├── 0002-pg-mcp-design.md      # 技术设计文档
│   └── ...
│
├── .claude/                       # Claude Code 配置
│   ├── CLAUDE.md                  # 项目开发规范
│   └── skills/                    # Skills
│       └── pg-data/               # PostgreSQL 查询 Skill
│
├── .env.example                   # 环境变量示例
├── pyproject.toml                 # 项目配置
├── Makefile                       # 常用命令
└── README.md
```

---

## 配置说明

### 环境变量配置

所有配置通过环境变量设置，支持 `.env` 文件。完整配置参考 `.env.example`。

#### 数据库配置

```bash
# 单数据库
PG_MCP_DATABASES__0__NAME=mydb
PG_MCP_DATABASES__0__HOST=localhost
PG_MCP_DATABASES__0__PORT=5432
PG_MCP_DATABASES__0__USER=postgres
PG_MCP_DATABASES__0__PASSWORD=your_password
PG_MCP_DATABASES__0__DATABASE=mydb
PG_MCP_DATABASES__0__MIN_POOL_SIZE=5
PG_MCP_DATABASES__0__MAX_POOL_SIZE=20

# 多数据库（索引递增）
PG_MCP_DATABASES__1__NAME=another_db
PG_MCP_DATABASES__1__HOST=localhost
# ...
```

#### LLM 配置

```bash
PG_MCP_OPENAI__API_KEY=sk-your-api-key
PG_MCP_OPENAI__MODEL=gpt-4o
PG_MCP_OPENAI__TEMPERATURE=0.0
PG_MCP_OPENAI__MAX_TOKENS=2000
PG_MCP_OPENAI__TIMEOUT_SECONDS=30.0

# 使用 DeepSeek 或其他兼容 API
PG_MCP_OPENAI__BASE_URL=https://api.deepseek.com/v1
PG_MCP_OPENAI__MODEL=deepseek-chat
```

#### 安全配置

```bash
# 查询限制
PG_MCP_SECURITY__QUERY_TIMEOUT_SECONDS=30.0
PG_MCP_SECURITY__MAX_RESULT_ROWS=1000
PG_MCP_SECURITY__MAX_CONCURRENT_QUERIES=10

# 访问控制
PG_MCP_SECURITY__BLOCKED_TABLES=secret_table,audit_logs
PG_MCP_SECURITY__BLOCKED_COLUMNS=password,api_key,token
PG_MCP_SECURITY__SAFE_SEARCH_PATH=public
PG_MCP_SECURITY__READONLY_ROLE=readonly_user

# SQL 限制
PG_MCP_SECURITY__MAX_SQL_LENGTH=10000
PG_MCP_SECURITY__MAX_NESTING_DEPTH=10
```

#### 数据库级别安全配置（可选）

```bash
# 为特定数据库设置独立的安全策略
PG_MCP_DATABASES__0__SECURITY__BLOCKED_TABLES=sensitive_data
PG_MCP_DATABASES__0__SECURITY__READONLY_ROLE=db_readonly
PG_MCP_DATABASES__0__SECURITY__QUERY_TIMEOUT_SECONDS=10.0
```

---

## 开发指南

### 安装开发依赖

```bash
# 使用 uv
uv sync --all-extras

# 或使用 pip
pip install -e ".[dev]"
```

### 常用命令

```bash
# 运行开发服务
make dev

# 运行测试
make test                    # 所有测试
make test-unit               # 单元测试
make test-integration        # 集成测试（需要数据库）

# 代码质量
make lint                    # ruff 检查
make format                  # 格式化代码
make typecheck               # mypy 类型检查

# 测试数据库
make db-setup                # 创建测试数据库
make db-teardown             # 删除测试数据库
```

### 运行测试

```bash
# 运行所有单元测试
pytest tests/unit

# 运行集成测试（需要 PostgreSQL）
pytest tests/integration -m integration

# 运行 E2E 测试（需要 PostgreSQL 和 LLM API）
pytest tests/e2e -m e2e

# 生成覆盖率报告
pytest --cov=src/pg_mcp --cov-report=html
```

### 测试数据库

项目提供三个测试数据库，用于开发和测试：

```bash
# 创建测试数据库
make db-setup

# 数据库信息
# - pg_mcp_small: 博客系统，少量数据
# - pg_mcp_medium: 电商系统，中等数据量
# - pg_mcp_large: 企业系统，大量数据和复杂 schema
```

---

## 相关文档

- [产品需求文档](specs/0001-pg-mcp-prd.md)
- [技术设计文档](specs/0002-pg-mcp-design.md)
- [实现计划](specs/0004-pg-mcp-impl-plan.md)
- [测试计划](specs/0007-pg-mcp-test-plan.md)
- [安全分析报告](specs/0010-pg-mcp-security-multidb-review.md)
- [弹性模块分析](specs/0009-pg-mcp-resilience-observability-review.md)

---

## 许可证

MIT License - 详见 [LICENSE](LICENSE) 文件。

## 致谢

- [FastMCP](https://github.com/jlowin/fastmcp) - MCP 框架
- [asyncpg](https://github.com/MagicStack/asyncpg) - PostgreSQL 驱动
- [SQLGlot](https://github.com/tobymao/sqlglot) - SQL 解析器
- [OpenAI](https://openai.com/) - GPT 模型