AI导哈

PostgreSQL只读查询服务器

PostgreSQL MCP服务器提供对PostgreSQL数据库的只读访问，支持执行SELECT查询、检查数据库架构和跟踪查询历史记录。

  • by jplwunder
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 jplwunder/postgres-mcp-server

使用说明

PostgreSQL MCP服务器

一种模型上下文协议（MCP）服务器，提供对PostgreSQL数据库的只读访问。执行SELECT查询，检查数据库模式，并使用内置的安全功能跟踪查询历史。

特性

  • 只读查询执行：使用自动只读事务强制执行SELECT查询
  • 多种输出格式：JSON、CSV或Markdown表格格式的结果
  • 架构检查：列出表、描述表结构、查看索引和探索模式
  • 查询历史：使用执行时间和元数据跟踪最近执行的查询
  • 连接池：具有可配置池大小的高效连接管理
  • 安全：查询验证、SQL注入防止和只读事务保证
  • 数据库统计：查看数据库大小、表计数和连接信息

安装

  1. 克隆此存储库：
git clone <repository-url>
cd postgres-mcp
  1. 使用安装依赖项 uv:
uv sync

配置

环境变量

创建一个 .env 项目根目录中的文件（使用 .env.example 作为模板）：

# Required: PostgreSQL Connection Parameters
POSTGRES_HOST=localhost
POSTGRES_PORT=5432
POSTGRES_DATABASE=myapp
POSTGRES_USER=readonly_user
POSTGRES_PASSWORD=secure_password

# Optional: Connection Pool Configuration
POSTGRES_POOL_MIN_SIZE=2          # Default: 2
POSTGRES_POOL_MAX_SIZE=10         # Default: 10
POSTGRES_COMMAND_TIMEOUT=60       # Default: 60 seconds
POSTGRES_CONNECTION_TIMEOUT=10    # Default: 10 seconds

# Optional: Query History Configuration
QUERY_HISTORY_SIZE=100            # Default: 100

# Optional: Logging Configuration
LOG_LEVEL=INFO                    # Default: INFO (DEBUG, INFO, WARNING, ERROR)

数据库用户设置

为了安全起见，创建一个专用的只读PostgreSQL用户：

-- Create read-only user
CREATE USER readonly_user WITH PASSWORD 'secure_password';

-- Grant connect permission
GRANT CONNECT ON DATABASE myapp TO readonly_user;

-- Grant schema usage
GRANT USAGE ON SCHEMA public TO readonly_user;

-- Grant select on all tables
GRANT SELECT ON ALL TABLES IN SCHEMA public TO readonly_user;

-- Grant select on future tables
ALTER DEFAULT PRIVILEGES IN SCHEMA public
    GRANT SELECT ON TABLES TO readonly_user;

用法

运行服务器

uv run python main.py

服务器通过stdio通信，可以与Claude Desktop等MCP客户端集成。

与Claude Desktop集成

将此配置添加到您的Claude Desktop配置文件中：

MacOS: ~/Library/Application Support/Claude/claude_desktop_config.json 视窗: %APPDATA%\Claude\claude_desktop_config.json

{
  "mcpServers": {
    "postgres": {
      "command": "python",
      "args": ["/path/to/postgres-mcp/main.py"],
      "env": {
        "POSTGRES_HOST": "localhost",
        "POSTGRES_PORT": "5432",
        "POSTGRES_DATABASE": "myapp",
        "POSTGRES_USER": "readonly_user",
        "POSTGRES_PASSWORD": "secure_password"
      }
    }
  }
}

或者，如果使用 uv:

{
  "mcpServers": {
    "postgres": {
      "command": "uv",
      "args": [
        "run",
        "--directory",
        "/path-to-postgres-mcp/postgres-mcp",
        "python",
        "main.py"
      ]
    }
}

可用工具

1.查询_数据库

对具有格式化输出的数据库执行SELECT查询。

输入：

{
  "query": "SELECT * FROM users LIMIT 10",
  "format": "json",
  "timeout": 30
}

参数：

  • query （必需）：要执行的SQL SELECT查询
  • format （可选）：输出格式- json （默认）， csv，或 markdown
  • timeout （可选）：查询超时（秒）（最大300）

输出：

{
  "rows": [...],
  "row_count": 10,
  "columns": ["id", "name", "email"],
  "execution_time_ms": 45.32,
  "format": "json",
  "formatted_output": "..."
}

2.列表表

列出数据库中包含元数据的所有表。

输入：

{
  "schema": "public"
}

参数：

  • schema （可选）：按架构名称筛选表

输出：

{
  "tables": [
    {
      "schema": "public",
      "name": "users",
      "row_count_estimate": 1500,
      "size": "128 KB"
    }
  ]
}

3.描述表

获取详细的表结构，包括列、类型、索引和约束。

输入：

{
  "table_name": "users",
  "schema": "public"
}

参数：

  • table_name （必填）：表格名称
  • schema （可选）：架构名称（默认值： public)

输出：

{
  "schema": "public",
  "table": "users",
  "columns": [
    {
      "name": "id",
      "type": "integer",
      "nullable": false,
      "default": "nextval('users_id_seq')",
      "primary_key": true
    }
  ],
  "indexes": [...],
  "foreign_keys": [...]
}

4.列表模式

列出数据库中的所有架构。

输入：

{}

输出：

{
  "schemas": ["public", "auth", "analytics"]
}

5.获取表索引

获取特定表的所有索引。

输入：

{
  "table_name": "users",
  "schema": "public"
}

参数：

  • table_name （必填）：表格名称
  • schema （可选）：架构名称（默认值： public)

输出：

{
  "indexes": [
    {
      "name": "users_pkey",
      "type": "btree",
      "columns": ["id"],
      "unique": true,
      "primary": true
    }
  ]
}

6.get_query_history

使用执行元数据检索最近的查询历史记录。

输入：

{
  "limit": 20
}

参数：

  • limit （可选）：返回的最大查询数（默认值：20，最大值：100）

输出：

{
  "queries": [
    {
      "query": "SELECT * FROM users",
      "timestamp": "2025-12-04T10:30:00Z",
      "execution_time_ms": 45.32,
      "row_count": 10,
      "format": "json",
      "success": true,
      "error": null
    }
  ]
}

7.get_database_stats

获取总体数据库统计信息和元数据。

输入：

{}

输出：

{
  "database_name": "myapp",
  "size": "45 MB",
  "table_count": 12,
  "connection_count": 5,
  "version": "PostgreSQL 15.3"
}

安全功能

只读执行

所有查询都在只读事务中执行：

async with conn.transaction(readonly=True):
    result = await conn.fetch(query)

查询验证

查询在执行前经过验证，以防止：

  • INSERT、UPDATE、DELETE操作
  • DROP、CREATE、ALTER操作
  • TRUNCATE、GRANT、REVOKE运营
  • 其他写入/管理操作

SQL注入防护

  • 表和模式标识符的输入净化
  • 参数化查询（如适用）
  • 基于正则表达式的标识符验证

建筑

组件

  • config.py：环境配置和验证
  • database.py：连接池管理和只读查询执行
  • validators.py：查询验证和清理
  • formatters.py：结果格式（JSON、CSV、Markdown）
  • history.py：线程安全查询历史跟踪
  • tools.py：MCP工具实现
  • server.py：MCP服务器设置和生命周期管理
  • types.py：用于类型安全的Pydantic模型

连接池

  • 最小尺寸：2个热连接
  • 最大尺寸：10个并发连接
  • 超时：60秒命令超时，10秒连接超时
  • 空闲寿命：自动清理非活动连接

故障排除

连接错误

错误：“数据库身份验证失败”

  • 验证 POSTGRES_USER 和 POSTGRES_PASSWORD 是正确的
  • 检查PostgreSQL中是否存在该用户
  • 确保用户具有CONNECT权限

错误：“找不到数据库'myapp'”

  • 验证 POSTGRES_DATABASE 匹配现有数据库
  • 检查数据库名称拼写

错误：“连接被拒绝”

  • 验证PostgreSQL是否在指定的主机和端口上运行
  • 检查防火墙设置
  • 验证 POSTGRES_HOST 和 POSTGRES_PORT 是正确的

查询错误

错误：“查询包含禁用关键字：INSERT”

  • 此服务器只允许SELECT查询
  • 使用不同的工具进行写入操作

错误：“表不存在”

  • 验证表名和架构是否正确
  • 使用 list_tables 查看可用表格
  • 检查用户是否对表具有SELECT权限

错误：“查询执行超时”

  • 查询花费的时间超过了指定的超时时间
  • 优化查询或增加超时参数
  • 检查大型表上是否缺少索引

权限错误

错误：“表X的权限被拒绝”

  • 数据库用户缺少SELECT权限
  • 授予适当的权限（请参阅数据库用户设置）

发展

运行测试

# Add your test commands here
pytest

项目结构

postgres-mcp/
--- .env                    # Configuration (not in git)
--- .env.example            # Configuration template
--- .gitignore
--- README.md
--- pyproject.toml
--- main.py                 # Entry point
--- src/
    --- postgres_mcp/
        --- __init__.py
        --- config.py       # Configuration
        --- database.py     # Connection pool
        --- formatters.py   # Output formatting
        --- history.py      # Query history
        --- server.py       # MCP server
        --- tools.py        # MCP tools
        --- types.py        # Pydantic models
        --- validators.py   # Query validation

许可证

[在此处添加您的许可证]

贡献

[在此处添加贡献指南]

相关 MCP