TomerGutman1/PG_SQL_MCP_SERVER

PostgreSQL MCP Server

Production-ready PostgreSQL server supporting both MCP and A2A protocols for secure, read-only database access.

Overview

Secure PostgreSQL access for LLMs and AI agents with:

• 37 Tools: Comprehensive database introspection, queries, and performance analysis
• Dual Protocol Support: MCP (Claude Desktop) + A2A (Google AI Agents)
• Production Security: RBAC, SQL guardrails, row limiting, audit logging
• Connection Pooling: asyncpg with configurable pool management
• Observability: Prometheus metrics, structured logging, audit trails

Quick Start

Prerequisites

• Docker & Docker Compose
• Or: Python 3.11+ and PostgreSQL 12+

Start with Docker (Recommended)

# Clone repository
git clone <repository-url>
cd PGSQL_FASTMCP

# Start all services
docker-compose up -d

# Check status
docker-compose ps

# View logs
docker-compose logs -f mcp-server

Services Started:

• PostgreSQL (port 5432, internal)
• MCP Server (port 8000)
• A2A Server (port 8001)
• Prometheus (port 9091)

Verify Installation

# Check A2A agent card
curl http://localhost:8001/.well-known/agent-card.json

# Check Prometheus metrics
curl http://localhost:9091/metrics

37 Available Tools

Basic Introspection (6 tools)

• pg_list_databases - List all accessible databases
• pg_list_schemas - List schemas in a database
• pg_list_tables - List tables and views in a schema
• pg_table_info - Complete table metadata (columns, constraints, indexes)
• pg_list_columns - List columns for a table
• pg_list_indexes - List indexes for a table

Advanced Introspection (7 tools)

• pg_list_functions - List functions/stored procedures
• pg_function_info - Detailed function information
• pg_list_triggers - List triggers on a table
• pg_trigger_info - Detailed trigger information
• pg_list_sequences - List sequences
• pg_sequence_info - Current sequence state
• pg_list_extensions - List installed extensions

Relationships & Partitioning (6 tools)

• pg_table_relationships - Foreign key relationships for a table
• pg_schema_relationships - All relationships in a schema
• pg_list_partitions - List table partitions
• pg_partition_info - Partition details
• pg_list_materialized_views - List materialized views
• pg_materialized_view_info - Materialized view details

Performance Analysis (8 tools)

• pg_slow_queries - Analyze slow queries (requires pg_stat_statements)
• pg_active_queries - Currently active queries
• pg_lock_analysis - Lock analysis and blocking queries
• pg_unused_indexes - Find unused indexes
• pg_missing_indexes - Tables needing indexes
• pg_index_usage_stats - Index usage statistics
• pg_vacuum_status - Vacuum status and dead tuples
• pg_bloat_analysis - Table/index bloat estimation

Query & Explain (2 tools)

• pg_query_readonly - Execute read-only SQL queries
• pg_explain_readonly - Get query execution plans

Health Monitoring (2 tools)

• pg_health_check - Database connectivity and health
• pg_pool_status - Connection pool status

Search & Sampling (4 tools)

• pg_search_tables - Search for tables by name pattern
• pg_search_columns - Search for columns by name pattern
• pg_sample_rows - Get sample rows from a table
• pg_table_stats - Table statistics (size, row count)

Full Documentation:

Usage

With Claude Desktop (MCP)

Add to ~/Library/Application Support/Claude/claude_desktop_config.json:

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

Restart Claude Desktop. You can now ask:

• "List all the tables in the test_mcp database"
• "Show me the structure of the users table"
• "SELECT * FROM customers WHERE active = true LIMIT 10"

With A2A Protocol (Google AI Agents)

# Discover capabilities
curl http://localhost:8001/.well-known/agent-card.json

# Send task
curl -X POST http://localhost:8001/a2a \
  -H "X-API-Key: demo-key-12345" \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "tasks/send",
    "id": "1",
    "params": {
      "id": "task-1",
      "message": {
        "role": "user",
        "parts": [
          {"type": "text", "text": "List all tables in test_mcp"}
        ]
      }
    }
  }'

Full Documentation:

Python Client Example

from mcp_client.client import McpClient

# Initialize client
client = McpClient("http://localhost:8000/mcp")
client.initialize()
client.send_initialized()

# List databases
databases = client.tools_call("pg_list_databases", {})

# Query data
result = client.tools_call("pg_query_readonly", {
    "database": "test_mcp",
    "sql": "SELECT * FROM customers WHERE active = true LIMIT 5"
})

# Get table info
info = client.tools_call("pg_table_info", {
    "database": "test_mcp",
    "schema": "demo",
    "table": "customers"
})

Configuration

Main Configuration

Edit config/config.yaml:

# Server settings
server:
  host: "0.0.0.0"
  port: 8000
  log_level: "INFO"

# Database connections
databases:
  test_mcp:
    host: "postgres"      # "localhost" for non-Docker
    port: 5432
    user: "mcp_test"
    password: "${DB_PASSWORD}"  # Use environment variables
    database: "test_mcp"
    pool_min_size: 2
    pool_max_size: 10
    command_timeout: 30   # Query timeout (seconds)
    ssl_mode: "prefer"    # disable/allow/prefer/require/verify-ca/verify-full

# Security
security:
  secret_key: "${SECRET_KEY}"
  rbac_config_path: "config/rbac.yaml"
  allowed_operations_path: "config/allowed_operations.yaml"
  enable_audit_log: true

# A2A Protocol
a2a:
  enabled: true
  port: 8001
  api_keys:
    - "${A2A_API_KEY_1}"
    - "${A2A_API_KEY_2}"

RBAC (Access Control)

Edit config/rbac.yaml:

clients:
  default:
    databases:
      - "*"                # All databases
    operations:
      - "SELECT"           # Read-only
    row_limit: 1000        # Max rows per query

  analyst-team:
    databases:
      - "analytics"
      - "reporting"
    schemas:
      - "public"
      - "mart"
    row_limit: 10000

Full Documentation:

Security Features

1. SQL Guardrails

• Blocks: DROP, DELETE, UPDATE, INSERT, TRUNCATE, ALTER, CREATE
• Allows: SELECT, WITH (CTEs)
• SQL injection pattern detection

2. RBAC

• Database-level permissions
• Schema-level filtering
• Table-level restrictions
• Row limiting per client

3. Resource Limits

• Query timeouts (configurable per database)
• Row limits (configurable per client)
• Connection pool limits

4. Audit Logging

• All queries logged with client ID, duration, row count
• Structured JSON logs (via structlog)
• Success/failure tracking

5. Connection Security

• SSL/TLS support (6 modes: disable to verify-full)
• Connection pooling with asyncpg
• Read-only database users recommended

Full Documentation:

Monitoring

Prometheus Metrics

Access metrics at http://localhost:9091/metrics:

# Query counts
pgsql_mcp_queries_total

# Query duration
pgsql_mcp_query_duration_seconds

# Errors
pgsql_mcp_query_errors_total

# Pool status
pgsql_mcp_pool_size
pgsql_mcp_active_connections

# RBAC rejections
pgsql_mcp_rbac_rejections_total

Structured Logs

# View all logs
docker logs -f mcp-server

# Filter for queries
docker logs mcp-server | grep "event.*query"

# Filter for errors
docker logs mcp-server | grep "level.*error"

Log Format (JSON):

{
  "timestamp": "2025-11-24T15:30:45.123Z",
  "level": "info",
  "event": "query_success",
  "client_id": "default",
  "database": "test_mcp",
  "duration_seconds": 0.045,
  "row_count": 25
}

Development

Run Without Docker

# Create virtual environment
python3 -m venv venv
source venv/bin/activate

# Install dependencies
pip install -r requirements.txt
pip install -e .

# Start MCP server
python -m pgsql_mcp.server

# Start A2A server (separate terminal)
python -m pgsql_mcp.a2a_server

Run Tests

# Unit tests
pytest tests/unit/

# Integration tests
pytest tests/integration/

# All tests
pytest

Troubleshooting

Connection Refused

Problem: Cannot connect to PostgreSQL

Solutions:

• Docker: Use host: "postgres" in config
• Local: Use host: "localhost" in config
• Check: docker-compose ps postgres

Permission Denied

Problem: RBAC blocks query

Solutions:

• Check config/rbac.yaml
• Verify database is in allowed list
• Ensure operations include "SELECT"

Query Timeout

Problem: Query takes too long

Solutions:

• Increase command_timeout in config
• Add indexes to tables
• Use pg_explain_readonly to analyze query

Full Documentation:

Project Structure

PGSQL_FASTMCP/
├── config/              # Configuration files
├── src/pgsql_mcp/       # Source code
│   ├── server.py        # MCP server
│   ├── a2a_server.py    # A2A server
│   ├── core/            # Pool management, config
│   ├── security/        # RBAC, SQL guardrails
│   ├── tools/           # 37 database tools
│   ├── observability/   # Metrics, logging, audit
│   └── utils/           # Helpers
├── tests/               # Test suite
├── examples/            # Usage examples
├── docs/                # Detailed documentation
├── docker-compose.yml   # Docker orchestration
└── README.md            # This file

Full Documentation:

Documentation

• - All 37 tools with examples
• - System design and data flow
• - Complete configuration guide
• - Security features and best practices
• - A2A protocol support
• - Common issues and solutions
• - Test database schema

Production Deployment Checklist

• Use environment variables for all secrets
• Set ssl_mode: "verify-full" for database connections
• Create read-only database users
• Configure RBAC with least privilege
• Set restrictive row limits (1000-10000)
• Generate strong API keys for A2A
• Enable audit logging
• Set up Prometheus alerts
• Use reverse proxy with HTTPS
• Implement rate limiting
• Monitor for RBAC rejections and errors

Full Guide:

License

[Specify your license]

Contributing

[Add contribution guidelines]

Support

For issues and questions:

• GitHub Issues: [Link to issues]
• Documentation:

Built with:

• FastMCP - MCP framework
• asyncpg - PostgreSQL driver
• Prometheus - Metrics
• structlog - Logging
• Docker - Containerization