How to Connect GPT-5 to PostgreSQL with Arcade’s MCP

Large language models need structured data access to provide accurate, data-driven insights. This guide demonstrates how to connect GPT-5 to PostgreSQL databases through Arcade's Model Context Protocol implementation, enabling secure database queries without exposing credentials directly to language models.

Prerequisites

Before implementing database connectivity, ensure you have:

Python 3.8 or higher installed
PostgreSQL database with connection credentials
Arcade API key (free tier available)
Development environment (VS Code, PyCharm, or terminal)
Basic understanding of SQL and async Python patterns

Architecture Overview

The integration follows this data flow:

GPT-5 (via MCP Client) → Arcade MCP Server → Custom Database Tool → PostgreSQL Database

Arcade acts as the intermediary layer that:

Manages secure database credential storage through environment variables
Exposes database operations as MCP tools
Handles authentication and authorization logic
Provides logging and observability for database queries

Installation and Setup

Install Required Packages

# Install Arcade MCP server SDK
pip install arcade-mcp-server

# Install PostgreSQL adapter
pip install psycopg2-binary

# Install Arcade CLI for project scaffolding
uv tool install arcade-mcp

The arcade-mcp-server package provides the FastAPI-like interface for building MCP servers, while psycopg2-binary enables PostgreSQL connections from Python.

Configure Environment Variables

Create a .env file in your project root to store database credentials securely:

# Database connection
POSTGRES_HOST=localhost
POSTGRES_PORT=5432
POSTGRES_DATABASE=your_database
POSTGRES_USER=your_username
POSTGRES_PASSWORD=your_password

# Arcade API configuration
ARCADE_API_KEY=your_arcade_api_key

Critical security note: Never commit the .env file to version control. Add it to .gitignore immediately to prevent credential exposure.

Building the Postgres MCP Server

Create the Server Structure

Use the Arcade CLI to generate a new MCP server project:

# Generate new MCP server
arcade new postgres_server

# Navigate to project
cd postgres_server

This creates a structured project with:

server.py - Main MCP server definition
.env.example - Template for environment variables
Configuration files for MCP clients

Implement Database Connection Management

Create a database.py file to handle PostgreSQL connections:

import os
import psycopg2
from psycopg2 import pool
from contextlib import contextmanager
from typing import List, Dict, Any

class DatabaseManager:
    """Manages PostgreSQL connection pooling and query execution"""

    def __init__(self):
        self.connection_pool = psycopg2.pool.SimpleConnectionPool(
            minconn=1,
            maxconn=10,
            host=os.getenv("POSTGRES_HOST"),
            port=os.getenv("POSTGRES_PORT"),
            database=os.getenv("POSTGRES_DATABASE"),
            user=os.getenv("POSTGRES_USER"),
            password=os.getenv("POSTGRES_PASSWORD")
        )

    @contextmanager
    def get_connection(self):
        """Context manager for database connections"""
        conn = self.connection_pool.getconn()
        try:
            yield conn
        finally:
            self.connection_pool.putconn(conn)

    def execute_query(self, query: str, params: tuple = None) -> List[Dict[str, Any]]:
        """
        Execute SQL query and return results as list of dictionaries

        Args:
            query: SQL query string
            params: Query parameters for safe parameterization

        Returns:
            List of row dictionaries with column names as keys
        """
        with self.get_connection() as conn:
            with conn.cursor() as cursor:
                cursor.execute(query, params)

                if cursor.description:
                    columns = [desc[0] for desc in cursor.description]
                    results = cursor.fetchall()
                    return [dict(zip(columns, row)) for row in results]

                conn.commit()
                return []

    def get_schema_info(self) -> Dict[str, List[Dict[str, str]]]:
        """Retrieve database schema information for all tables"""
        query = """
        SELECT
            table_name,
            column_name,
            data_type,
            is_nullable
        FROM information_schema.columns
        WHERE table_schema = 'public'
        ORDER BY table_name, ordinal_position
        """

        results = self.execute_query(query)

        # Group by table name
        schema = {}
        for row in results:
            table = row['table_name']
            if table not in schema:
                schema[table] = []
            schema[table].append({
                'column': row['column_name'],
                'type': row['data_type'],
                'nullable': row['is_nullable']
            })

        return schema

# Initialize singleton instance
db_manager = DatabaseManager()

This implementation provides connection pooling for efficient resource management and includes schema introspection capabilities that help GPT-5 understand your database structure.

Define Database Query Tools

Create the main server file server.py with MCP tools for database operations:

#!/usr/bin/env python3
"""PostgreSQL MCP Server for Arcade"""

import sys
import json
from typing import Annotated, List, Dict, Any
from arcade_mcp_server import Context, MCPApp
from database import db_manager

app = MCPApp(
    name="postgres_server",
    version="1.0.0",
    instructions="PostgreSQL database query server for GPT-5 integration"
)

@app.tool(requires_secrets=["POSTGRES_PASSWORD"])
async def query_database(
    context: Context,
    sql_query: Annotated[str, "SQL SELECT query to execute"],
    parameters: Annotated[List[Any], "Query parameters for safe parameterization"] = None
) -> Annotated[str, "Query results in JSON format"]:
    """
    Execute a SELECT query against the PostgreSQL database.

    This tool allows GPT-5 to query database tables with proper parameterization
    to prevent SQL injection attacks. Only SELECT statements are permitted.
    """

    # Validate query is a SELECT statement
    normalized_query = sql_query.strip().upper()
    if not normalized_query.startswith('SELECT'):
        return json.dumps({
            "error": "Only SELECT queries are permitted",
            "query": sql_query
        })

    try:
        await context.log.info(f"Executing query: {sql_query}")

        # Execute query with parameters
        results = db_manager.execute_query(
            sql_query,
            tuple(parameters) if parameters else None
        )

        await context.log.info(f"Query returned {len(results)} rows")

        return json.dumps({
            "success": True,
            "row_count": len(results),
            "data": results
        }, indent=2)

    except Exception as e:
        await context.log.error(f"Query execution failed: {str(e)}")
        return json.dumps({
            "error": str(e),
            "query": sql_query
        })

@app.tool
async def get_database_schema(
    context: Context
) -> Annotated[str, "Database schema information"]:
    """
    Retrieve complete database schema including tables, columns, and data types.

    This tool helps GPT-5 understand your database structure to generate
    appropriate queries without prior knowledge of table names or columns.
    """

    try:
        await context.log.info("Fetching database schema")

        schema = db_manager.get_schema_info()

        return json.dumps({
            "success": True,
            "tables": schema,
            "table_count": len(schema)
        }, indent=2)

    except Exception as e:
        await context.log.error(f"Schema retrieval failed: {str(e)}")
        return json.dumps({"error": str(e)})

@app.tool
async def list_tables(
    context: Context
) -> Annotated[str, "List of database tables"]:
    """
    List all tables in the public schema.

    Useful for GPT-5 to understand what data is available before constructing queries.
    """

    query = """
    SELECT table_name
    FROM information_schema.tables
    WHERE table_schema = 'public'
    ORDER BY table_name
    """

    try:
        await context.log.info("Listing database tables")

        results = db_manager.execute_query(query)
        tables = [row['table_name'] for row in results]

        return json.dumps({
            "success": True,
            "tables": tables,
            "count": len(tables)
        }, indent=2)

    except Exception as e:
        await context.log.error(f"Table listing failed: {str(e)}")
        return json.dumps({"error": str(e)})

@app.tool
async def describe_table(
    context: Context,
    table_name: Annotated[str, "Name of the table to describe"]
) -> Annotated[str, "Table structure and column details"]:
    """
    Get detailed information about a specific table's structure.

    Returns column names, data types, constraints, and sample values to help
    GPT-5 construct accurate queries for that table.
    """

    # Get column information
    column_query = """
    SELECT
        column_name,
        data_type,
        character_maximum_length,
        is_nullable,
        column_default
    FROM information_schema.columns
    WHERE table_schema = 'public' AND table_name = %s
    ORDER BY ordinal_position
    """

    # Get row count
    count_query = f"SELECT COUNT(*) as row_count FROM {table_name}"

    try:
        await context.log.info(f"Describing table: {table_name}")

        columns = db_manager.execute_query(column_query, (table_name,))
        count_result = db_manager.execute_query(count_query)

        if not columns:
            return json.dumps({
                "error": f"Table '{table_name}' not found"
            })

        return json.dumps({
            "success": True,
            "table_name": table_name,
            "row_count": count_result[0]['row_count'],
            "columns": columns
        }, indent=2)

    except Exception as e:
        await context.log.error(f"Table description failed: {str(e)}")
        return json.dumps({"error": str(e)})

if __name__ == "__main__":
    # Support both stdio and HTTP transports
    transport = sys.argv[1] if len(sys.argv) > 1 else "stdio"
    app.run(transport=transport)

The tool definitions use @app.tool decorator to expose functions as MCP tools. The requires_secrets parameter ensures database credentials are available without exposing them to the language model.

Connecting to GPT-5 via MCP Clients

Claude Desktop Configuration

Claude Desktop provides native MCP support. Configure it to connect to your Postgres MCP server:

# Configure Claude Desktop to use your server
arcade configure claude

This command automatically updates Claude Desktop's configuration file (typically ~/Library/Application Support/Claude/claude_desktop_config.json on macOS) to include your server.

Alternatively, manually edit the configuration:

{
  "mcpServers": {
    "postgres": {
      "command": "python",
      "args": ["/path/to/your/server.py"],
      "env": {
        "POSTGRES_HOST": "localhost",
        "POSTGRES_PORT": "5432",
        "POSTGRES_DATABASE": "your_database",
        "POSTGRES_USER": "your_username",
        "POSTGRES_PASSWORD": "your_password"
      }
    }
  }
}

Restart Claude Desktop after configuration changes. You should see a hammer icon indicating MCP servers are connected.

Visual Studio Code Configuration

For VS Code with Copilot or Cursor:

# Configure VS Code MCP connection
arcade configure vscode --entrypoint server.py

Create .vscode/mcp.json in your project root:

{
  "servers": {
    "postgres": {
      "command": "python",
      "args": ["server.py"],
      "env": {
        "POSTGRES_HOST": "localhost",
        "POSTGRES_PORT": "5432",
        "POSTGRES_DATABASE": "your_database",
        "POSTGRES_USER": "your_username",
        "POSTGRES_PASSWORD": "your_password"
      }
    }
  }
}

HTTP Transport for Web Applications

For production web applications, serve the MCP server over HTTP:

# Start HTTP server on port 8000
python server.py http

Access the API documentation at http://127.0.0.1:8000/docs to verify server operation and test tools manually.

Configure your web application to connect via HTTP transport:

from arcadepy import Arcade

client = Arcade(api_key=os.getenv("ARCADE_API_KEY"))

# List available tools from your Postgres server
tools = await client.tools.list(toolkit="postgres", limit=30)

# Execute a database query tool
result = await client.tools.execute(
    tool_name="postgres.query_database",
    input={
        "sql_query": "SELECT * FROM users WHERE created_at > %s",
        "parameters": ["2024-01-01"]
    },
    user_id="unique_user_identifier"
)

Usage Examples with GPT-5

Once configured, interact with your database through natural language:

Example 1: Schema Exploration

User prompt to GPT-5:

What tables are available in my database?

GPT-5 will call the list_tables tool and respond with the complete list of tables.

Example 2: Data Analysis

User prompt:

Show me the top 10 customers by total purchase amount

GPT-5 will:

Call get_database_schema to understand table structure
Identify relevant tables (customers, orders, etc.)
Construct appropriate SQL query
Call query_database with the SQL
Present results in a readable format

Example 3: Complex Queries

User prompt:

What's the average order value by customer segment for Q1 2024?

GPT-5 generates and executes:

SELECT
    c.segment,
    AVG(o.total_amount) as avg_order_value,
    COUNT(*) as order_count
FROM customers c
JOIN orders o ON c.id = o.customer_id
WHERE o.created_at BETWEEN '2024-01-01' AND '2024-03-31'
GROUP BY c.segment
ORDER BY avg_order_value DESC

Security Best Practices

Query Validation and Restrictions

Implement strict query validation to prevent unauthorized database operations:

def validate_query(sql_query: str) -> tuple[bool, str]:
    """
    Validate SQL query meets security requirements

    Returns: (is_valid, error_message)
    """
    normalized = sql_query.strip().upper()

    # Only allow SELECT statements
    if not normalized.startswith('SELECT'):
        return False, "Only SELECT queries are permitted"

    # Block dangerous keywords
    forbidden_keywords = ['DROP', 'DELETE', 'UPDATE', 'INSERT', 'ALTER', 'CREATE', 'TRUNCATE']
    for keyword in forbidden_keywords:
        if keyword in normalized:
            return False, f"Query contains forbidden keyword: {keyword}"

    # Prevent multiple statements
    if ';' in sql_query[:-1]:  # Allow trailing semicolon
        return False, "Multiple statements not allowed"

    return True, ""

Database User Permissions

Create a read-only database user specifically for GPT-5 queries:

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

-- Grant connect permission
GRANT CONNECT ON DATABASE your_database TO gpt5_readonly;

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

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

-- Ensure future tables are also accessible
ALTER DEFAULT PRIVILEGES IN SCHEMA public
GRANT SELECT ON TABLES TO gpt5_readonly;

Use this restricted user in your environment variables to limit potential damage from query mistakes or security vulnerabilities.

Query Logging and Monitoring

Implement comprehensive logging for audit trails:

@app.tool(requires_secrets=["POSTGRES_PASSWORD"])
async def query_database(
    context: Context,
    sql_query: Annotated[str, "SQL SELECT query to execute"],
    parameters: Annotated[List[Any], "Query parameters"] = None
) -> Annotated[str, "Query results"]:

    # Log query with timestamp and context
    await context.log.info(f"Query initiated by user")
    await context.log.info(f"SQL: {sql_query}")
    await context.log.info(f"Parameters: {parameters}")

    try:
        results = db_manager.execute_query(sql_query, tuple(parameters) if parameters else None)

        # Log success metrics
        await context.log.info(f"Query successful - {len(results)} rows returned")

        return json.dumps({"success": True, "data": results})

    except Exception as e:
        # Log error details
        await context.log.error(f"Query failed: {str(e)}")
        await context.log.error(f"Query: {sql_query}")

        return json.dumps({"error": str(e)})

Access logs through Arcade Dashboard for monitoring and debugging.

Performance Optimization

Query Result Limiting

Prevent resource exhaustion by limiting result set sizes:

def add_limit_to_query(sql_query: str, max_rows: int = 1000) -> str:
    """Add LIMIT clause if not present"""
    normalized = sql_query.strip().upper()

    if 'LIMIT' not in normalized:
        # Remove trailing semicolon if present
        query = sql_query.rstrip(';')
        return f"{query} LIMIT {max_rows};"

    return sql_query

Connection Pooling

The DatabaseManager class implements connection pooling using psycopg2.pool.SimpleConnectionPool. Configure pool size based on expected concurrency:

self.connection_pool = psycopg2.pool.SimpleConnectionPool(
    minconn=2,  # Minimum idle connections
    maxconn=20,  # Maximum connections (adjust based on load)
    host=os.getenv("POSTGRES_HOST"),
    # ... other parameters
)

Query Timeout Configuration

Set query timeouts to prevent long-running queries from blocking resources:

def execute_query_with_timeout(
    self,
    query: str,
    params: tuple = None,
    timeout_seconds: int = 30
) -> List[Dict[str, Any]]:
    """Execute query with timeout"""

    with self.get_connection() as conn:
        with conn.cursor() as cursor:
            # Set statement timeout
            cursor.execute(f"SET statement_timeout = {timeout_seconds * 1000}")

            cursor.execute(query, params)

            if cursor.description:
                columns = [desc[0] for desc in cursor.description]
                results = cursor.fetchall()
                return [dict(zip(columns, row)) for row in results]

            return []

Deployment Options

Local Development

For development and testing:

# Start server with stdio transport
python server.py

# Or with HTTP transport for web testing
python server.py http

Cloud Deployment with Arcade

Deploy your MCP server to Arcade's hosted infrastructure:

# Login to Arcade Cloud
arcade login

# Deploy your server
arcade deploy

# View deployment status
arcade dashboard

Arcade Deploy handles:

Automatic scaling based on load
SSL/TLS certificate management
Environment variable management
Logging and monitoring integration

Self-Hosted Production

For on-premises deployments, use Docker for consistent environments:

FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application
COPY server.py database.py ./

# Set environment
ENV PYTHONUNBUFFERED=1

# Run server
CMD ["python", "server.py", "http"]

Deploy with Docker Compose:

version: '3.8'

services:
  postgres-mcp:
    build: .
    ports:
      - "8000:8000"
    environment:
      - POSTGRES_HOST=${POSTGRES_HOST}
      - POSTGRES_PORT=${POSTGRES_PORT}
      - POSTGRES_DATABASE=${POSTGRES_DATABASE}
      - POSTGRES_USER=${POSTGRES_USER}
      - POSTGRES_PASSWORD=${POSTGRES_PASSWORD}
      - ARCADE_API_KEY=${ARCADE_API_KEY}
    restart: unless-stopped

Troubleshooting

Connection Issues

Problem: "Could not connect to database"

Solutions:

Verify database credentials in .env file
Check PostgreSQL is running: pg_isready -h localhost
Ensure firewall allows connections on PostgreSQL port (default 5432)
Verify user has CONNECT privilege: GRANT CONNECT ON DATABASE your_db TO your_user;

Tool Not Appearing in GPT-5

Problem: Database tools don't appear in MCP client

Solutions:

Restart MCP client (Claude Desktop, VS Code, etc.)
Verify server configuration in client config file
Check server logs for startup errors: python server.py --debug
Confirm server process is running: ps aux | grep server.py

Query Execution Failures

Problem: Queries return errors or empty results

Solutions:

Check query syntax in PostgreSQL directly first
Verify table and column names match database schema
Review server logs for detailed error messages
Ensure database user has SELECT privileges on target tables

Performance Issues

Problem: Slow query execution

Solutions:

Add database indexes on frequently queried columns
Implement query result caching for repeated queries
Increase connection pool size if concurrent queries are high
Set appropriate query timeouts to prevent resource exhaustion

Advanced Features

Multi-Database Support

Extend the server to support multiple database connections:

class MultiDatabaseManager:
    def __init__(self):
        self.databases = {
            'production': self._create_pool(
                os.getenv("PROD_DB_HOST"),
                os.getenv("PROD_DB_NAME")
            ),
            'analytics': self._create_pool(
                os.getenv("ANALYTICS_DB_HOST"),
                os.getenv("ANALYTICS_DB_NAME")
            )
        }

    def _create_pool(self, host, database):
        return psycopg2.pool.SimpleConnectionPool(
            minconn=1, maxconn=10,
            host=host, database=database,
            user=os.getenv("POSTGRES_USER"),
            password=os.getenv("POSTGRES_PASSWORD")
        )

    def execute_query(self, database_name: str, query: str, params=None):
        # Use specified database connection pool
        pass

Query Result Caching

Implement caching for frequently accessed data:

from functools import lru_cache
import hashlib

def get_query_hash(query: str, params: tuple) -> str:
    """Generate cache key from query and parameters"""
    content = f"{query}:{params}"
    return hashlib.md5(content.encode()).hexdigest()

@lru_cache(maxsize=100)
def cached_query(query_hash: str, query: str, params: tuple):
    """Execute and cache query results"""
    return db_manager.execute_query(query, params)

Query Analytics

Track query patterns for optimization:

query_metrics = {
    'total_queries': 0,
    'queries_by_table': {},
    'average_execution_time': 0,
    'error_count': 0
}

async def track_query_metrics(table_name: str, execution_time: float, success: bool):
    """Record query analytics"""
    query_metrics['total_queries'] += 1

    if table_name not in query_metrics['queries_by_table']:
        query_metrics['queries_by_table'][table_name] = 0
    query_metrics['queries_by_table'][table_name] += 1

    if not success:
        query_metrics['error_count'] += 1

Conclusion

Connecting GPT-5 to PostgreSQL through Arcade's MCP implementation provides secure, scalable database access for AI applications. The architecture separates credential management from the language model, implements query validation and logging, and supports both development and production deployments.

Key implementation points:

Use arcade-mcp-server SDK for standardized tool definitions
Implement read-only database users for security
Validate all queries before execution to prevent SQL injection
Log all database operations for audit compliance
Configure connection pooling for performance at scale
Deploy via Arcade Cloud or self-hosted infrastructure based on requirements

For additional resources:

This implementation provides a production-ready foundation for database-backed AI applications, supporting use cases from business intelligence queries to automated data analysis and reporting.

How to Query Postgres from GPT-5 via Arcade (MCP)

Prerequisites

Architecture Overview

Installation and Setup

Install Required Packages

Configure Environment Variables

Building the Postgres MCP Server

Create the Server Structure

Implement Database Connection Management

Define Database Query Tools

Connecting to GPT-5 via MCP Clients

Claude Desktop Configuration

Visual Studio Code Configuration

HTTP Transport for Web Applications

Usage Examples with GPT-5

Example 1: Schema Exploration

Example 2: Data Analysis

Example 3: Complex Queries

Security Best Practices

Query Validation and Restrictions

Database User Permissions

Query Logging and Monitoring

Performance Optimization

Query Result Limiting

Connection Pooling

Query Timeout Configuration

Deployment Options

Local Development

Cloud Deployment with Arcade

Self-Hosted Production

Troubleshooting

Connection Issues

Tool Not Appearing in GPT-5

Query Execution Failures

Performance Issues

Advanced Features

Multi-Database Support

Query Result Caching

Query Analytics

Conclusion

RECENT ARTICLES

How to Connect GPT-5 to Slack with Arcade (MCP)

How to Build a GPT-5 Gmail Agent with Arcade (MCP)

How to Call Custom Tools from Python Agent via Arcade

Get early access to Arcade, and start building now.