麦普 postgres

Name: 麦普 postgres
Author: gldc

作者：gldc15 星标更新：2025-08-29

安装使用来源：spider==github.com

项目介绍

PostgreSQL MCP 服务器

这是一个使用 Model Context Protocol (MCP) Python SDK 实现的 PostgreSQL MCP 服务器。MCP 是一个开放协议，它使大型语言模型应用程序与外部数据源之间的无缝集成成为可能。此服务器允许 AI 代理通过标准化接口与 PostgreSQL 数据库进行交互。

特性

列出数据库模式
列出模式中的表
描述表结构
列出表约束和关系
获取外键信息
执行 SQL 查询
带有 JSON/markdown 输出的类型化工具
可选的表资源和指导提示

快速开始

# 不带数据库连接运行服务器（适用于 Glama 或检查）
python postgres_server.py

# 使用实时数据库 —— 选择一种方法：
export POSTGRES_CONNECTION_STRING="postgresql://user:pass@host:5432/db"
python postgres_server.py

# …或…
python postgres_server.py --conn "postgresql://user:pass@host:5432/db"

# 或使用 Docker（构建一次，然后运行）：
# docker build -t mcp-postgres . && docker run -p 8000:8000 mcp-postgres

安装

通过 Smithery 安装

要通过 Smithery 自动安装 PostgreSQL MCP 服务器到 Claude Desktop：

npx -y @smithery/cli install @gldc/mcp-postgres --client claude

手动安装

克隆此仓库：

git clone <repository-url>
cd mcp-postgres

创建并激活虚拟环境（推荐）：

python -m venv venv
source venv/bin/activate  # 在 Windows 上使用：venv\Scripts\activate

安装依赖项：

pip install -r requirements.txt

使用

启动 MCP 服务器。

# 不带连接字符串（服务器启动，DB 支持的工具会返回友好的错误）
python postgres_server.py

# 或者通过环境变量设置连接字符串：
export POSTGRES_CONNECTION_STRING="postgresql://username:password@host:port/database"
python postgres_server.py

# 或者通过 --conn 标志传递：
python postgres_server.py --conn "postgresql://username:password@host:port/database"

# 可选：通过 HTTP 运输方式运行
# 流式 HTTP（推荐用于流式工具输出）
python postgres_server.py --transport streamable-http --host 0.0.0.0 --port 8000

# SSE 运输方式（服务器发送事件）挂载在 /sse 和 /messages/
python postgres_server.py --transport sse --host 0.0.0.0 --port 8000 --mount /mcp

服务器提供以下工具：

query: 对数据库执行 SQL 查询
list_schemas: 列出所有可用的模式
list_tables: 列出特定模式中的所有表
describe_table: 获取关于表结构的详细信息
get_foreign_keys: 获取表的外键关系
find_relationships: 发现表的显式和隐式关系
db_identity: 显示当前的 db/user/host/port、search_path 和版本

类型化（推荐）：

run_query(input): 使用类型化的输入执行（sql, parameters, row_limit, format: 'markdown'|'json'）。
run_query_json(input): 执行并返回可序列化为 JSON 的行。
list_schemas_json(input): 使用过滤器列出模式（include_system, include_temp, require_usage, row_limit）。
list_schemas_json_page(input): 使用过滤器和 name_like 模式的分页列表。
list_tables_json(input): 使用过滤器列出模式内的表（名称模式，大小写敏感性，表类型，行限制）。
list_tables_json_page(input): 使用过滤器的分页表列表。

示例：

// run_query (markdown)
{
  "sql": "SELECT * FROM information_schema.tables WHERE table_schema = %s",
  "parameters": ["public"],
  "row_limit": 50,
  "format": "markdown"
}

// run_query_json
{
  "sql": "SELECT now() as ts",
  "row_limit": 1
}

检查当前连接身份：

// db_identity (无输入)
{}

使用过滤器列出模式（JSON）：

{
  "include_system": false,
  "include_temp": false,
  "require_usage": true,
  "row_limit": 10000
}

带有模式过滤器的分页列表：

{
  "include_system": false,
  "include_temp": false,
  "require_usage": true,
  "page_size": 200,
  "cursor": null,
  "name_like": "sales_*",
  "case_sensitive": false
}

响应形状：

{
  "items": [ { "schema_name": "sales_eu", "owner": "...", "is_system": false, "is_temporary": false, "has_usage": true } ],
  "next_cursor": "...base64..." // 当没有更多页面时为 null
}

使用过滤器列出表（JSON）：

{
  "db_schema": "public",
  "name_like": "orders_*",
  "case_sensitive": false,
  "table_types": ["BASE TABLE", "VIEW"],
  "row_limit": 1000
}

分页表列表：

{
  "db_schema": "public",
  "page_size": 200,
  "cursor": null,
  "name_like": "orders_%"
}

资源（如果客户端支持）：

table://{schema}/{table} 用于读取表行。提供备用工具：
- list_table_resources(schema) → table://... URI
- read_table_resource(schema, table, row_limit) → 行 JSON

提示（当支持时注册；也作为工具暴露）：

write_safe_select / prompt_write_safe_select_tool
explain_plan_tips / prompt_explain_plan_tips_tool

使用 Docker 运行

构建镜像：

docker build -t mcp-postgres .

不带数据库连接运行容器（服务器仍然可检查）：

docker run -p 8000:8000 mcp-postgres

通过提供 POSTGRES_CONNECTION_STRING 连接到实时 PostgreSQL 数据库：

docker run \
  -e POSTGRES_CONNECTION_STRING="postgresql://username:password@host:5432/database" \
  -p 8000:8000 \
  mcp-postgres

如果省略环境变量，服务器正常启动，并且所有数据库支持的工具会返回友好的“连接字符串未设置”消息，直到您提供该变量。

使用 mcp.json 配置

要将此服务器与兼容 MCP 的工具（如 Cursor）集成，请将其添加到您的 ~/.cursor/mcp.json 中：

{
  "servers": {
    "postgres": {
      "command": "/path/to/venv/bin/python",
      "args": [
        "/path/to/postgres_server.py"
      ],
      "env": {
        "POSTGRES_CONNECTION_STRING": "postgresql://username:password@host:5432/database?ssl=true"
      }
    }
  }
}

运输环境变量

MCP_TRANSPORT=stdio|sse|streamable-http（默认：stdio）
MCP_HOST=0.0.0.0 和 MCP_PORT=8000 用于 SSE/HTTP 运输方式
MCP_SSE_MOUNT=/mcp 可选的 SSE 挂载路径

如果省略 POSTGRES_CONNECTION_STRING，服务器仍然启动并且完全可检查；数据库支持的工具会简单地返回一个有用的错误，直到提供该变量。

替换：

/path/to/venv 为您虚拟环境的路径
/path/to/postgres_server.py 为服务器脚本的绝对路径

HTTP 客户端集成

使用 Streamable HTTP 运行服务器：

python postgres_server.py --transport streamable-http --host 0.0.0.0 --port 8000
# 或使用 Docker
docker run -p 8000:8000 mcp-postgres \
  python postgres_server.py --transport streamable-http --host 0.0.0.0 --port 8000

基本可达性检查（期望非 200 状态码，因为 MCP 需要握手）：

curl -i http://localhost:8000/mcp
# 404/405/422 表示服务器可达；客户端必须使用 MCP 协议。

指向 Streamable HTTP 终点的示例 MCP 客户端配置（概念性）：

{
  "servers": {
    "postgres": {
      "transport": "streamable-http",
      "url": "http://localhost:8000/mcp"
    }
  }
}

对于 SSE 而不是 Streamable HTTP：

python postgres_server.py --transport sse --host 0.0.0.0 --port 8000 --mount /mcp
curl -N http://localhost:8000/sse  # 连接到 SSE 终点

Python MCP 客户端示例（Streamable HTTP）

import asyncio
from mcp.client import streamable_http
from mcp.client.session import ClientSession


async def main():
    url = "http://localhost:8000/mcp"
    async with streamable_http.streamablehttp_client(url) as (read, write, _get_session_id):
        session = ClientSession(read, write)
        init = await session.initialize()
        print("协议版本:", init.protocolVersion)

        # 列出工具
        tools = await session.list_tools()
        print("工具:", [t.name for t in tools.tools])

        # 调用类型化工具：run_query_json
        result = await session.call_tool(
            "run_query_json",
            {"input": {"sql": "SELECT 1 AS n", "row_limit": 1}},
        )
        # 如果提供了结构化内容，则优先使用；否则使用文本内容
        if result.structuredContent is not None:
            print("结构化内容:", result.structuredContent)
        else:
            print("文本块:", [getattr(b, "text", None) for b in result.content])


if __name__ == "__main__":
    asyncio.run(main())

安全性

永远不要在代码中暴露敏感的数据库凭据
使用环境变量或安全配置文件来存储数据库连接字符串
考虑使用连接池以更好地管理资源
实现适当的访问控制和用户认证

环境选项

POSTGRES_READONLY=true 仅允许 SELECT/CTE/EXPLAIN/SHOW/VALUES
POSTGRES_STATEMENT_TIMEOUT_MS=15000 限制语句运行时间

贡献

欢迎贡献！请随时提交拉取请求。

开发与测试

创建 venv 并安装运行时依赖项：pip install -r requirements.txt
（可选）安装测试依赖项：pip install -r dev-requirements.txt
运行测试：pytest -q

许可证

MIT 许可证

在此授权任何人免费获取本软件及其相关文档文件（以下简称“软件”），在不受限制的情况下使用、复制、修改、合并、发布、分发、再许可和/或出售软件的副本，并允许向其提供软件的人这样做，但需遵守以下条件：

软件按“原样”提供，不附带任何明示或暗示的担保，包括但不限于适销性、特定用途适用性和非侵权性的担保。在任何情况下，作者或版权持有人都不对因软件或其使用或其他交易而产生的任何索赔、损害或其他责任承担任何责任，无论是合同行为、侵权行为还是其他行为引起的。