PostgreSQL MCP Server Interface

This component implements the Model Context Protocol (MCP) framework, leveraging the MCP Python SDK to provide an abstraction layer for AI agents to communicate with PostgreSQL data stores. It ensures a uniform interface for database operations.

Core Capabilities

• Schema inventory listing
• Table enumeration within defined schemas
• Detailed output on table blueprints (columns, types)
• Retrieval of referencing and referenced key information
• Execution engine for direct SQL statements
• Tools emit strongly typed results in JSON or Markdown formats
• Support for optional resource pointers (e.g., table data access) and guidance prompts.

Initiation Guide

bash

Launch server in inspection mode (no active DB connection)

python postgres_server.py

Connect to a live instance by setting the environment variable:

export POSTGRES_CONNECTION_STRING="postgresql://user:pass@host:5432/db" python postgres_server.py

Alternatively, supply connection details directly via flag:

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

Docker deployment example (build first):

docker build -t mcp-postgres .

docker run -p 8000:8000 mcp-postgres

Installation Procedures

Installation via Smithery

Automated setup for Claude Desktop integration using Smithery:

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

Manual Setup

1. Clone the source repository: bash git clone cd mcp-postgres

2. Establish and activate a dedicated Python environment: bash python -m venv venv source venv/bin/activate # Use venv\Scripts\activate on Windows

3. Install required Python packages: bash pip install -r requirements.txt

Operational Usage

1. Start the service process.

bash # Server starts but DB tools will fail gracefully without a connection string python postgres_server.py

# Set connection via environment variable: export POSTGRES_CONNECTION_STRING="postgresql://username:password@host:port/database" python postgres_server.py

# Pass connection string via argument: python postgres_server.py --conn "postgresql://username:password@host:port/database"

# Configure specific network transport (recommended for streaming): python postgres_server.py --transport streamable-http --host 0.0.0.0 --port 8000

# Configure SSE transport (mounted at /mcp): python postgres_server.py --transport sse --host 0.0.0.0 --port 8000 --mount /mcp

1. The available MCP tools are:

2. query: Executes specified SQL.

3. list_schemas: Enumerates schemas.
4. list_tables: Retrieves tables for a given schema.
5. describe_table: Fetches structural metadata for a table.
6. get_foreign_keys: Retrieves FK definitions.
7. find_relationships: Discovers explicit and inferred data links.
8. db_identity: Reports connection metadata (host, user, version).

Typed Tool Endpoints (Preferred):

• run_query(input): Executes SQL with structured parameters (sql, parameters, row_limit, format: 'markdown'|'json').
• run_query_json(input): Executes and guarantees JSON array output.
• list_schemas_json(input): Schema listing with filtering options (include_system, row_limit).
• list_schemas_json_page(input): Paginated schema retrieval with fuzzy matching (name_like).
• list_tables_json(input): Filtered listing of tables by schema, type, and pattern.
• list_tables_json_page(input): Cursor-based pagination for tables.

Input Examples (JSON format):

// Executing a query returning Markdown { "sql": "SELECT count(*) FROM user_sessions WHERE login_date > %s", "parameters": ["2024-01-01"], "row_limit": 1, "format": "markdown" }

// Simple JSON query { "sql": "SELECT current_timestamp() as time", "row_limit": 1 }

Connection Inspection:

// db_identity call {}

**Schema Listing Example (Filtered): **

{ "include_system": false, "require_usage": true, "row_limit": 500 }

Paginated Listing with Pattern Matching:

{ "require_usage": false, "page_size": 100, "cursor": null, "name_like": "data_staging_%", "case_sensitive": true }

Expected Paginated Response Structure:

{ "items": [ { "schema_name": "data_staging_01", "owner": "db_admin", ... } ], "next_cursor": "encoded_token_or_null" }

**Table Listing Example (JSON): **

{ "db_schema": "public", "name_like": "audit_log%", "table_types": ["BASE TABLE"], "row_limit": 50 }

Resource Pointers: Clients supporting MCP resources can access table data via URIs like table://{schema}/{table}. Fallback tools are provided for clients that do not support direct URI resolution: - list_table_resources(schema) → Generates URIs. - read_table_resource(schema, table, row_limit) → Fetches row data as JSON.

Registered Prompts (Also exposed as tools): - write_safe_select - explain_plan_tips

Docker Operation

Building the execution image:

bash docker build -t mcp-postgres .

Running without DB access (for protocol testing):

bash docker run -p 8000:8000 mcp-postgres

Running with secured connection environment variable:

bash docker run \ -e POSTGRES_CONNECTION_STRING="postgresql://user:pass@host:5432/db" \ -p 8000:8000 \ mcp-postgres

Note: If POSTGRES_CONNECTION_STRING is absent, the server initializes, but all DB interactions will result in an informative error message.

Configuration via mcp.json (Client Integration)

To register this service with an MCP client (e.g., Cursor):

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

Transport Configuration Environment Variables

• MCP_TRANSPORT: Specifies communication protocol (stdio, sse, or streamable-http). Default is stdio.
• MCP_HOST (0.0.0.0) and MCP_PORT (8000) control network binding.
• MCP_SSE_MOUNT: Optional path prefix for SSE endpoints (e.g., /mcp).

Network Endpoint Verification (HTTP)

Ensure the server is listening using Streamable HTTP:

bash python postgres_server.py --transport streamable-http --host 0.0.0.0 --port 8000

Verify reachability (expecting an MCP handshake error, not HTTP success):

curl -i http://localhost:8000/mcp

For SSE, connect to the base or mounted path:

bash curl -N http://localhost:8000/sse

Python MCP Client Sketch (Streamable HTTP)

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

async def main(): endpoint_url = "http://localhost:8000/mcp" async with streamable_http.streamablehttp_client(endpoint_url) as (reader, writer, get_session_id): client_session = ClientSession(reader, writer) await client_session.initialize()

    # Discover available tools
    tool_list = await client_session.list_tools()
    print(f"Available Tools: {[t.name for t in tool_list.tools]}")

    # Execute a typed tool call
    execution_result = await client_session.call_tool(
        "run_query_json",
        {"input": {"sql": "SELECT 42 AS answer", "row_limit": 1}},
    )

    # Output processing logic
    if execution_result.structuredContent is not None:
        print("Structured Data Received:", execution_result.structuredContent)
    else:
        print("Textual Output Blocks:", execution_result.content)

if name == "main": asyncio.run(main())

Security Posture

• Credential Handling: Connection strings must be managed via secure environment variables, never hardcoded.
• Resource Management: Connection pooling is advised for high-throughput environments.
• Access Control: Database users should operate under the principle of least privilege.

Operational Configuration Overrides

• POSTGRES_READONLY=true: Restricts executable operations primarily to read-only commands (SELECT, EXPLAIN, etc.).
• POSTGRES_STATEMENT_TIMEOUT_MS: Sets a maximum execution duration for any single query (e.g., 15000ms).

Development & Testing

To prepare the environment for contribution: - Install core dependencies: pip install -r requirements.txt - Install testing dependencies: pip install -r dev-requirements.txt (if applicable) - Execute unit tests: pytest -q

Related Resources

• MCP Protocol Specification
• Official MCP Python SDK

Legal Notice

Licensed under the MIT License. Copyright (c) 2025 gldc. Redistribution subject to license terms.